Threading problems

Hello,

Just wondering if anyone could advise me on this problem I have. I have a main thread which creates an instance of one of my classes, call it class A. The main thread then creates another thread and uses a method from class A as the defined function to be executed on the thread.

After which, the main thread starts polling for events to be signaled from the second thread (the one from class A). Now when I get an event signaled, this means that there is a data structure ready for me to read from the main thread. The data structure is a public member of class A. The problem I am fearful of, is that the main thread could try reading the data structure and the second thread is also trying to refill the data structure at the same time.

I could have two versions of the same structure within class a, one public and one private, and the main thread just reads the public one. Nevertheless, the same problem could occur when
class A tries to do a memcpy from the private to public member of the structure, and my main thread tries to read the public structure whilst the 2nd thread is doing the memcpy.

When should I use EnterCriticalSection? Should I use it within class a when it is doing the memcpy or should I do it within my main thread when I want to read the data structure.

Any advise on this would be helpful. Thanks.

The critical section should be as short as possible. So wrapp only the pieces of code in between a critcal section which are manipulating "shared" data

mfg JJ

It all depends on the design. Implement the critical section where it's needed.

Kuphryn

Ok, thanks for the advice. If an error occurs within my worker thread, what should the return value be? Should I return -1 to denote that there was an error for my main thread?

Thanks.

Quote:

---

Originally Posted by subdene

When should I use EnterCriticalSection? Should I use it within class a when it is doing the memcpy or should I do it within my main thread when I want to read the data structure.

---

Both.

If you read from a data structure that can be written to by another thread, then you must thread guard it. If you write to a data structure that can be read by a different thread, again you must thread guard it.

I wanted to submit some code to demonstrate a few concepts, including the "on topic" ones.
For those not familiar with the Win32 API, any unfamiliar functions can be looked up on MSDN (and forgive the Win32 specific code on the c++ board).

The Op's initial post describes a classic producer/consumer scenario that is typical in a multi-threaded application.
The following code shows how the std::deque container can be used as the point of access for multiple producer and consumer threads. It also demonstrates simple encapsulation of synchronization objects to make MT programming a little easier.

I am using a Thread class that I posted previously, but I thought it would be more appropriate since the isn't the Windows board.

Question comments welcome. Feel free to ask if anything needs elabaration.

Code:

---

#include <iostream>
#include <sstream>
#include <string>
#include <deque>
#include <exception>
using namespace std;

// code from this thread.
#include "thread.h"

#include <windows.h>

//-----------------------------------------------------------------------------
// C++ wrapper object for CRITICAL_SECTION
class CriticalSection
{
    CRITICAL_SECTION m_cs;
public:
    CriticalSection() {InitializeCriticalSection(&m_cs);}
    ~CriticalSection() {DeleteCriticalSection(&m_cs);}
    operator CRITICAL_SECTION*() {return &m_cs;}
    void Enter() {EnterCriticalSection(&m_cs);}
    void Leave() {LeaveCriticalSection(&m_cs);}
};//CriticalSection

//-----------------------------------------------------------------------------
// Automatic Enter() of a CriticalSection in the contructor, and Leave() in the
// destructor
class AutoCSLock
{
    CriticalSection *m_cs;
public:
    explicit AutoCSLock(CriticalSection *cs) : m_cs(cs) {m_cs->Enter();}
    ~AutoCSLock() {m_cs->Leave();}
};//CSLock

//-----------------------------------------------------------------------------
// Thread safe queue object for producer consumer threads
template <class T>
class MTqueue
{
    deque<T> m_q;
    CriticalSection m_cs; // for thread safe access to m_q
    HANDLE m_ev; // event to signal consumers to come and get it

public:
    MTqueue() {m_ev = CreateEvent(NULL, TRUE, FALSE, NULL);}
    ~MTqueue() {CloseHandle(m_ev);}

    HANDLE consumer_event() {return m_ev;}

    void en_que(const T &val)
    {
        AutoCSLock lock(&m_cs);
        m_q.push_back(val);
        SetEvent(m_ev); // something to consume
    }//en_que

    // this shuold only be called by the owner of consumer_event()
    void de_que(T &val)
    {
        AutoCSLock lock(&m_cs);
        typename deque<T>::size_type sz = m_q.size();
        if (sz == 0)
        {
            cerr << "MTqueue::de_que() on empyt queue!!" << endl;
            throw bad_exception();
        }//if
        val = m_q.back();
        m_q.pop_back();
        if (sz == 1)
            ResetEvent(m_ev); // m_q is empty now
    }//de_que
};//MTqueue

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

DWORD producer_thread(Thread *t); // implemented below

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

int main()
{
    // a message queue
    MTqueue<string> msg_q;

    // a thread object to run our producer_thread() function
    Thread t(&producer_thread);

    cout << "Starting producer thread...\n" << endl;
    t.SetThreadData(&msg_q);
    t.Start();

    // now we can start consuming messages
    int n = 0;
    HANDLE consume_ev = msg_q.consumer_event();
    string msg;

    while (1)
    {
        if (WaitForSingleObject(consume_ev, INFINITE) != WAIT_OBJECT_0)
            break;
        n++;

        msg_q.de_que(msg);
       
        cout << msg << endl;

        if (n == 10)
            break;
    }//while

    cout << "\nStopping producer thread" << endl;
    t.Stop(INFINITE);

    return 0;
}//main

//-----------------------------------------------------------------------------
// NOTE: remove the "WINAPI" calling convention from ThreadProc_t typedef used
//      by the Thread class - it serves no purpose.

DWORD producer_thread(Thread *t)
{
    MTqueue<string> *q = (MTqueue<string>*)t->GetThreadData();

    // wait a second on the exit event, then produce something
    int n = 1;
    ostringstream next_msg;
    HANDLE exit_ev = t->GetExitEvent();
    while (1)
    {
        // if this doesn't timeout, then exit
        if (WaitForSingleObject(exit_ev, 1000) != WAIT_TIMEOUT)
            break;

        next_msg << "producing message #" << n++;

        q->en_que(next_msg.str());

        next_msg.seekp(0); // seek to beginning of stream
    }//while

    cout << "Producer thread exiting" << endl;
    return 0;
}//producer_thread

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

---

gg