Producer and Consumer Problem using Windows Threads

Above is class diagram for a simple producer/consumer problem

#include <iostream>
#include "MQ.h"
#include "Thread.h"
#include "Runnable.h"
#include "ProductionTask.h"
#include "ConsumptionTask.h"

//File: Main.cpp

using namespace examples;

int main()
{
    try
    {
        MQ q(10);
        ProductionTask producerTask(q);
        ConsumptionTask consumerTask(q);
        Thread t[2] = { producerTask, consumerTask };

        //start the producer and consumer threads
        t[0].start();
        t[1].start();

        //wait 50000 ms before terminating the threads
        t[0].join(50000);
        t[1].join(50000);
        std::cout << std::endl
            << "Threads timed out!!" << std::endl;

    }catch(std::exception& e)
    {
        std::cerr << e.what() << std::endl;
    }catch(...)
    {
        std::cerr << "Unknown Exception" << std::endl;
    }
    return 0;
}

Key Features

1. Threads are Tasks are decoupled. A thread can execute any task provided it implements the Runnable interface.
2. Windows threads are encapsulated using the technique demonstrated in the earlier blog post Encapsulating Windows Threads in C++ Objects
3. Two Tasks ProductionTask and ConsumptionTask share the common MQ instance q and try to execute add and remove operations on q concurrently. Synchronization is achieved here by means of Lock implemented using Windows CRITICAL_SECTION
4. The threads are allowed to run for 50 seconds before they are terminated. This is not a graceful termination and in real world applications may lead to unexpected results

#ifndef _Runnable_H_
#define _Runnable_H_

//File: Runnable.h

#include <iostream>
#include "NonCopyable.h"

namespace examples
{
    class Runnable
    {
    public:
        virtual ~Runnable(){}
        virtual void run() = 0;
    };

}

#endif //_Runnable_H_

#ifndef _ConsumptionTask_H_
#define _ConsumptionTask_H_

//File: ConsumptionTask.h

#include "Runnable.h"
#include "MQ.h"

namespace examples
{
    class ConsumptionTask : public Runnable
    {
    public:
        ConsumptionTask(MQ &q);
        ~ConsumptionTask();
        virtual void run();

    private:
        MQ&  m_queue;
    };

}

#endif //_ConsumptionTask_H_

#include "ConsumptionTask.h"

//File: ConsumptionTask.cpp

using namespace examples;

ConsumptionTask::ConsumptionTask(MQ &q) : m_queue(q)
{}

ConsumptionTask::~ConsumptionTask()
{}

void ConsumptionTask::run()
{
    while(true)
    {
        m_queue.remove();
        ::Sleep(550);
    }
}

#ifndef _ProductionTask_H_
#define _ProductionTask_H_

//File: ProductionTask.h

#include "Runnable.h"
#include "MQ.h"

namespace examples
{
    class ProductionTask : public Runnable
    {
    public:
        ProductionTask(MQ &q);
        ~ProductionTask();
        virtual void run();

    private:
        MQ&  m_queue;
    };

}

#endif //_ProductionTask_H_

#include "ProductionTask.h"

//File: ProductionTask.cpp

using namespace examples;

ProductionTask::ProductionTask(MQ &q) : m_queue(q)
{}

ProductionTask::~ProductionTask()
{}

void ProductionTask::run()
{
    while(true)
    {
        m_queue.add("Object");
        ::Sleep(500);
    }
}

Read more ...

Encapsulating Windows Threads in C++ Objects

Threads in Windows can be created using the following three routines

HANDLE CreateThread(
LPSECURITY_ATTRIBUTES lpsa,
DWORD cbStack,
LPTHREAD_START_ROUTINE lpStartAddr,
LPVOID lpvThreadParam,
DWORD fdwCreate,
LPDWORD lpIDThread
);

uintptr_t _beginthread( 
void( *start_address )( void * ),
unsigned stack_size,
void *arglist 
);

uintptr_t _beginthreadex( 
void *security,
unsigned stack_size,
unsigned ( *start_address )( void * ),
void *arglist,
unsigned initflag,
unsigned *thrdaddr 
);
//start_address: is the starting address of a routine that begins the execution of a new thread.

Key Points

1. The above routines expects a pointer to the application defined function which would be the entry point or starting address of the thread.
2. A static member function does not have access to the this pointer of the class whereas the member functions have an implicit parameter which points to the object (the this pointer)
3. A static member function is same as an ordinary C function
4. You should not pass pointer to a member function to a system call that starts a thread. A member function is meaning less without an object to invoke it

The threads in windows can be encapsulated using a static member function which acts as the entry point of the thread and passing the this pointer as the function argument. The ThreadImpl class shown below uses static unsigned __stdcall dispatch(void *); method as the thread start address.

#ifndef _Thread_H_
#define _Thread_H_

//File: Thread.h

#include "NonCopyable.h"

namespace examples
{
    class Thread : public NonCopyable
    {
    public:
        Thread();
        Thread(const char *);
        ~Thread();

        bool join();
        bool join(size_t ms);
        void start();
        void setName(const std::string&);
        const char* getName() const;

    private:
        class ThreadImpl *m_impl;
    };

}
#endif //_Thread_H_

#ifndef _ThreadImpl_H_
#define _ThreadImpl_H_

//File: ThreadImpl.h

#include <windows.h>
#include <process.h>
#include <iostream>
#include <string>
#include "NonCopyable.h"

namespace examples
{
    class ThreadImpl : public NonCopyable
    {
    public:
        ThreadImpl();
        ThreadImpl(const std::string&);
        ~ThreadImpl();

        bool join() const;
        bool join(size_t) const;
        void start() const;
        void setName(const std::string&);
        const char* getName() const;

    private:
        // thread entry point
        static unsigned __stdcall dispatch(void *);
        void run();
        bool spawn();

        HANDLE       m_hthread;
        unsigned     m_thrdid;
        std::string m_thrName;
    };

}
#endif //_ThreadImpl_H_

Read more ...

Coding simplified with Guard Classes

With the introduction of exceptions, writing code has become increasingly complex. Any resource acquired has to be released when no longer in use. If not handled properly will lead to handle leaks, memory leaks and in some cases deadlocks which are very difficult to debug.
Guard classes takes advantage of the fact that the destructor is always called for an object on the stack regardless of how you exit from the function scope.

Key Idea

unsigned __stdcall func(void *args)
{
static Lock lock;
Guard guard(lock);
// operations
return 0;
}

#ifndef _Guard_H_
#define _Guard_H_

#include "Lock.h"

namespace examples
{
    class Guard
    {
    public:
        Guard(Lock &lock) : m_lock(lock)
        {
            m_lock.acquire();
        }

        ~Guard()
        {
            m_lock.release();
        }

    private:
        Guard();
        Guard(const Guard&);
        Guard& operator=(const Guard&);

        Lock &m_lock;
    };

}

#endif //_Guard_H_

The program demonstrates the use of Guard class to acquire() and release() the lock

Read more ...

Debug the Code : Thread synchronization

The program is supposed to print multiples of 10 in a new line in increasing order. Although, the program appears to be correct, it is not. It is giving me duplicate values in the output. Locate the problem?

#include <windows.h>
#include <process.h>
#include <iostream>
#include <assert.h>
#include "Lock.h"

using namespace examples;

static bool alive = true;
static int current = 0;

unsigned __stdcall put(void *args)
{
    Lock lock;
    while(alive)
    {
        lock.acquire();
        current = current + 10;
        ::Sleep(500);
        std::cout << current << std::endl;
        lock.release();
    }
    return 0;
}

int main()
{
    // create threads
    unsigned t1;
    HANDLE h1 = (HANDLE) ::_beginthreadex(0, 0, &put, 0, CREATE_SUSPENDED, &t1);
    assert(h1 != 0);

    unsigned t2;
    HANDLE h2 = (HANDLE) ::_beginthreadex(0, 0, &put, 0, CREATE_SUSPENDED, &t2);
    assert(h2 != 0);

    // start threads
    ::ResumeThread(h1);
    ::ResumeThread(h2);

    ::Sleep(10000);
    alive = false;

    ::WaitForSingleObjectEx(h1, INFINITE, false);
    ::WaitForSingleObjectEx(h2, INFINITE, false);

    ::CloseHandle(h1);
    ::CloseHandle(h2);

    return 0;
}

Read more ...

Thread Synchronization in Windows using Critical Section

In multi threaded environments, where more than one thread operates on the shared data (global variables, collections) the results can be unpredictable due to race conditions. Critical Section is a mechanism that ensures that only one thread executes a particular piece of the code. A thread, once it enters a critical section should not be interrupted. Critical Sections have the advantage of not being kernel objects and are maintained in user space. This usually, but not always, provides performance improvements. The CRITICAL_SECTION data type is basically a structure whose fields are used only internally to Windows

Windows provide four functions for using critical sections. To use these functions, you must define an object of type CRITICAL_SECTION.

Key Points

1. Define a CRITICAL_SECTION object cs
2. Initialize a Critical section using InitializeCriticalSection (&cs);
3. After the critical section object has been initialized, a thread enters the critical section by calling EnterCriticalSection(&cs); No two threads can own the critical section at the same time
4. A thread leaves the critical section by calling LeaveCriticalSection (&cs);
5. When the critical section object is no longer needed, it can be deleted by the program by calling DeleteCriticalSection (&cs);

#ifndef _Lock_H_
#define _Lock_H_

#include <windows.h>

/**
*@description: A simple Lock implementation using windows critical section object
*/

namespace examples
{
    class Lock
    {
    public:
        Lock()
        {
            ::InitializeCriticalSection(&m_cs);
        }

        ~Lock()
        {
            ::DeleteCriticalSection(&m_cs);
        }

        void acquire()
        {
            ::EnterCriticalSection(&m_cs);
        }

        void release()
        {
            ::LeaveCriticalSection(&m_cs);
        }

    private:
        Lock(const Lock&);
        Lock& operator=(const Lock&);

        CRITICAL_SECTION m_cs;
    };

}

#endif //_Lock_H_

Read more ...