ProducerConsumer

#include <stdio.h>  /* printf, NULL */
#include <windows.h>    /* Win API */
#include <queue>    /* std::queue */
#include <stdlib.h>     /* srand, rand */
#include <time.h>       /* time */

/* Slightly modified Request class from the task to track request number */
class Request
{
static int idToGiveNext;
public:
    int id;
    Request():
        id(idToGiveNext++)
        {/*empty body*/};
};

int Request::idToGiveNext = 0;


/* Base class to protect some object with WinAPI CRITICAL_SECTION */
class InterthreadObject
{
protected:
    CRITICAL_SECTION mCriticalSection;
public:
    InterthreadObject()
    {
        InitializeCriticalSection(&mCriticalSection);
    }

virtual ~InterthreadObject()
    {
        DeleteCriticalSection(&mCriticalSection);
    }
};


/* Just a boolean to be checked protected by CRITICAL_SECTION */
class StopperCondition: public InterthreadObject
{
private:
    bool mShouldStop;
public:
    StopperCondition():
        mShouldStop(false)
        {/*empty*/};

    void set()
    {
        EnterCriticalSection(&mCriticalSection);
        {
            mShouldStop = true;
        }
        LeaveCriticalSection(&mCriticalSection);
    }

    bool check()
    {
        bool val;
        EnterCriticalSection(&mCriticalSection);
        {
            val = mShouldStop;
        }
        LeaveCriticalSection(&mCriticalSection);
        return val;
    }
};

/* A Stopper class from the task, just a wrapper over a previous class.
    I did so to implement GetRequest and ProcessRequest exactly in the way that specification required*/
class Stopper
{
public:
    StopperCondition* const mC;
    Stopper(StopperCondition* c):
        mC(c)
        {/*empty*/}
    Stopper(const Stopper &objToCopy):
        mC(objToCopy.mC)
        {/*empty*/}
    void set()
    {
        mC->set();
    }
    bool check()
    {
        return mC->check();
    }
};

/* Function from the task specification*/
Request* GetRequest(Stopper stopSignal) throw()
{
    const int MAX_WAIT_MS = 100;
    int waitMs = rand() % MAX_WAIT_MS + 1;
    Sleep(waitMs); // emulate long-time operations

    return stopSignal.check() ? NULL : (new Request());
};


/* Function from the task specification*/
void ProcessRequest(Request* request, Stopper stopSignal) throw()
{
    if(stopSignal.check())
        return;

    /* some processig there */
    const int MAX_WAIT_MS = 1000;
    int waitMs = rand() % MAX_WAIT_MS + 1;
    Sleep(waitMs); // emulate long-time operations
};


/* std::queue to hold incoming requests protected by CRITICAL_SECTION*/
class InterthreadRequestQueue : public InterthreadObject
{
private:
    std::queue<Request*> mQueue;
public:

    void push(Request* &val)
    {
        EnterCriticalSection(&mCriticalSection);
        {
            mQueue.push(val);
        }
        LeaveCriticalSection(&mCriticalSection);
    }

    Request* pop()
    {
        Request* val = NULL;
        EnterCriticalSection(&mCriticalSection);
        {
            if(!mQueue.empty())
            {
                val = mQueue.front();
                mQueue.pop();
            }
        }
        LeaveCriticalSection(&mCriticalSection);
        return val;
    }
};


/* All interprocess objects in one place. We can easily pass this object to CreateProcess(...) now. */
class SharedObjects
{
private:
    StopperCondition stopCondition;
public:
    InterthreadRequestQueue waitingRequests;
    InterthreadRequestQueue readyRequests;
    Stopper stopper;

    SharedObjects():
        stopper(&stopCondition)
        {/*empty*/};
};


int threadId = 0;
DWORD WINAPI consumerThread( LPVOID lpParam )
{
    int id = threadId++;
    SharedObjects* s = (SharedObjects*)lpParam;
    while(!s->stopper.check())
    {
        Request* r = s->waitingRequests.pop();
        if(!r)
            continue;

        printf("Request %d processed by thread %d\n", r->id, id);
        ProcessRequest(r, s->stopper);
        printf("Request %d is ready\n", r->id);
        s->readyRequests.push(r);
    }

    return 0;
};

DWORD WINAPI producerThread( LPVOID lpParam )
{
    int id = threadId++;
    SharedObjects* s = (SharedObjects*)lpParam;
    while(!s->stopper.check())
    {
        Request* r = GetRequest(s->stopper);
        if(!r)
            continue;

        printf("Request %d is waiting\n", r->id);
        s->waitingRequests.push(r);
    }

    return 0;
};

int main()
{
    const int CONSUMER_THREAD_COUNT = 2;
    const int TOTAL_THREAD_COUNT = CONSUMER_THREAD_COUNT + 1;  //consumers + producer
    const int SECONDS_TO_WORK = 30;

    srand(time(NULL));

    SharedObjects s;
    HANDLE hProducerThread;
    HANDLE hSomeConsumerThread;

    HANDLE* workerThreads;
    workerThreads = new HANDLE[TOTAL_THREAD_COUNT];

    hProducerThread = CreateThread( NULL, 0, producerThread, &s, 0, NULL);
    if (NULL == hProducerThread)
    {
        printf("CreateThread failed (%d)\n", GetLastError());
        return 1;
    }
    workerThreads[0] = hProducerThread;

    for(int i = 1; i <= CONSUMER_THREAD_COUNT; i++)
    {
        hSomeConsumerThread = CreateThread( NULL, 0, consumerThread, &s, 0, NULL);
        if (NULL == hProducerThread)
        {
            printf("CreateThread failed (%d)\n", GetLastError());
            return 1;
        }
        workerThreads[i] = hSomeConsumerThread;
    }


    HANDLE hTimer = NULL;
    LARGE_INTEGER liDueTime;

    liDueTime.QuadPart = -10000000LL * SECONDS_TO_WORK; // LL is in 100ns intervals, negative = relative to call moment

    // Create an unnamed waitable timer.
    hTimer = CreateWaitableTimer(NULL, TRUE, NULL);
    if (NULL == hTimer)
    {
        printf("CreateWaitableTimer failed (%d)\n", GetLastError());
        return 1;
    }

    printf("Waiting for %d seconds...\n", SECONDS_TO_WORK);

    if (!SetWaitableTimer(hTimer, &liDueTime, 0, NULL, NULL, 0))
    {
        printf("SetWaitableTimer failed (%d)\n", GetLastError());
        return 2;
    }

    if (WaitForSingleObject(hTimer, INFINITE) != WAIT_OBJECT_0)
        printf("WaitForSingleObject failed (%d)\n", GetLastError());
    else
    {
        printf("Timer was signaled.\n");
        s.stopper.set();
    }

    WaitForMultipleObjects( TOTAL_THREAD_COUNT,
        workerThreads, TRUE, INFINITE);

    printf("Since All threads executed"
           " lets close their handles \n");

    for(int i = 0; i < TOTAL_THREAD_COUNT; i++)
    {
        CloseHandle(workerThreads[i]);
    }
    delete[] workerThreads;

    Request* r;
    while(r = s.readyRequests.pop(), r)
    {
        int id = r->id;
        delete r;
        printf("Request %d deleted from readyRequests queue\n", id);
    }

    while(r = s.waitingRequests.pop(), r)
    {
        int id = r->id;
        delete r;
        printf("Request %d deleted from waitingRequests queue\n", id);
    }

    printf("All done. Press enter to exit...\n");
    getc(stdin);

    return 0;
}