Re: How to use CreateProcess for multiple processes?

This is a multi-part message in MIME format.
--------------000107080702010503020302
Content-Type: text/plain; charset=ISO-8859-1; format=flowed
Content-Transfer-Encoding: 7bit

Tommy wrote:


Ami, I wrote this very quickly to illustrate the bucket/worker threads
method. Attached is:

     TestMaster.Cpp
     TestSlave.cpp

This is a simulator of your master/slave process model. TestSlave use
random sleep to simulate different work times. TestMaster uses the
bucket and worker threads idea It works very nice. Adjust
FillBucket() and the SlaveThread() and you pretty much set.

Note: I wrote this very quickly, and its proof of concept. I'm sure
there are part that can be done more cleanly or C++ correct.

   - If you use the CBucket class in threads to add items, you need
     to add the critical section in the Add() member. It just
     syncs for the Fetch().

   - For this need, you probably might consider a reader/writer
     method instead of critical section. The CS is solid though.

I even added some time/process time stats to get a feel for the total
residence time of the threads. On my XP dual core box, with a
simulated random slave time been 1 sec to 10 secs, the average thread
time was about 50-52 seconds and a total master time of 54 seconds.

What I would do, is use the testslave to simulate different return
code errors which is stored in the global TThreadData block and maybe
have the master log the session error for that run.

Have fun :-)

---

--------------000107080702010503020302
Content-Type: text/plain;
 name="testmaster.cpp"
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="testmaster.cpp"

#include <stdio.h>
#include <afx.h>
#include <afxtempl.h>
#include <conio.h>

//------------------------------------------------------

const DWORD MAX_JOBS = 100;
const DWORD MAX_WORKERS = 10;
const char *SLAVE_EXE = "testslave.exe";

typedef struct _tagTSlaveData {
   char szUser[256];
   char szPwd[256];
   char szHost[256];
} TSlaveData;

typedef struct _tagTThreadData {
   DWORD index;
   DWORD dwStartTime;
   DWORD dwEndTime;
   DWORD dwExitCode;
   TSlaveData sd;
} TThreadData;

class CBucket : public CList< TSlaveData, TSlaveData>
{
public:
    CBucket()
    {
       InitializeCriticalSection(&cs);
    }
    void Add( const TSlaveData &o )
        { AddHead( o ); }

    void Add(const char *user, const char *pwd)
        {
          TSlaveData td = {0};
          strcpy(td.szUser,user);
          strcpy(td.szPwd,pwd);
          AddHead( td );
        }

    // Pop top element off stack
    BOOL Fetch(TSlaveData &o)
        {
          EnterCriticalSection(&cs);
          BOOL res = !IsEmpty();
          if (res) o = RemoveTail();
          LeaveCriticalSection(&cs);
          return res;
        }
private:
   CRITICAL_SECTION cs;
} Bucket;

TThreadData ThreadData[MAX_WORKERS] = {0};

//----------------------------------------------------------------
// Slave Work
//----------------------------------------------------------------

BOOL SlaveWork(TThreadData *data)
{
    STARTUPINFO si;
    ZeroMemory(&si, sizeof(si));
    PROCESS_INFORMATION pi;
    ZeroMemory(&pi, sizeof(pi));

    CString sCmd;
    sCmd.Format("%s %d /user:%s /pwd:%s /host:%s",
                SLAVE_EXE,
                data->index+1,
                data->sd.szUser,
                data->sd.szPwd,
                data->sd.szHost);

    if (!CreateProcess(NULL, (LPSTR &)sCmd,
                       NULL, NULL, FALSE, 0, NULL, NULL, &si, &pi)) {
        return FALSE;
    }
    CloseHandle(pi.hThread);
    if (WAIT_OBJECT_0== WaitForSingleObject(pi.hProcess, INFINITE)) {
       GetExitCodeProcess(pi.hProcess, &data->dwExitCode);
    }
    CloseHandle(pi.hProcess);
    return TRUE;
}

void WINAPI SlaveThread(TThreadData *data)
{
    data->dwStartTime = GetTickCount();
    TSlaveData sd;
    while (Bucket.Fetch(sd)) {
        data->sd = sd;
        SlaveWork(data);
    }
    data->dwEndTime = GetTickCount();
    return;
}

void DoThreads()
{
    ZeroMemory(&ThreadData,sizeof(ThreadData));

    HANDLE hThreads[MAX_WORKERS] = {0};
    DWORD tid;
    DWORD i;
    for(i=0;i < MAX_WORKERS; i++){
        ThreadData[i].index = i;
        hThreads[i] = CreateThread(
                      NULL,
                      0,
                      (LPTHREAD_START_ROUTINE) SlaveThread,
                      (void *)&ThreadData[i],
                      0,
                      &tid);
    }

    DWORD dwMasterTime = GetTickCount();
    DWORD nRemaining = 0;
    while (WaitForMultipleObjects(MAX_WORKERS, hThreads, TRUE, 1000) == WAIT_TIMEOUT) {
       int n = Bucket.GetSize();
       if (n != nRemaining) {
          nRemaining = n;
          printf("- Remaining: %d\n",nRemaining);
       }
       if (_kbhit() && _getch() == 27) {
          break;
       }
    }
    dwMasterTime = GetTickCount() - dwMasterTime;
    _cprintf("* Done\n");

    // show some thread times

    DWORD dwTime = 0;
    for (i = 0; i < MAX_WORKERS; i++) {
       TThreadData dt = ThreadData[i];
       dwTime += dt.dwEndTime-dt.dwStartTime;
       printf("%-3d | Time: %-6d\n",
                  i,dt.dwEndTime-dt.dwStartTime);
    }
    printf("---------------------------------------\n");
    printf("Total Slave Time : %d\n",dwTime);
    printf("Total Master Time : %d\n",dwMasterTime);

}

void FillBucket()
{
    for (int i = 0; i < MAX_JOBS; i++)
    {
         Bucket.Add("user","password");
    }
}

//----------------------------------------------------------------
// Main Thread
//----------------------------------------------------------------

int main(char argc, char *argv[])
{

    FillBucket();

    DoThreads();

    return 0;
}

--------------000107080702010503020302
Content-Type: text/plain;
 name="testslave.cpp"
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="testslave.cpp"

// File: G:\wc7beta\testslave.cpp

#include <stdio.h>
#include <windows.h>

//------------------------------------------------------
#define getrandom( min, max ) (( rand() % (int)((( max ) + 1 ) - ( min ))) + ( min ))
void rtlRANDOMIZE() { srand( (unsigned)GetTickCount() ); }

int main(char argc, char *argv[])
{
      rtlRANDOMIZE();
      int ret = argv[1]?atoi(argv[1]): 0;
      int msecs = getrandom(1000,10000);
      printf("%5d | Thread: %5d | sleep: %5d\n",ret, GetCurrentThreadId(), msecs);
      Sleep(msecs);
      return ret;
}

--------------000107080702010503020302--