2 threads vs 1 queue

[cpplinux]

Hello,

I'm looking for a small and clear example in c++ of 2 threads that are using the same queue.
The queue is the one of the c++ STL library.
Program 1 pushs requests in the queue and program 2 get these requests and process them.

Do I need to lock the queue when one of the threads is using it? or is it possible without lock?
What does happen when the queue is empty?

Thanks,

CL

[JVene]

I'm not 100% certain if you are asking about two worker threads servicing requests in a queue, or simply one thread pushing onto the queue and the worker thread reading from the queue, but I'm perhaps 90% sure you mean the latter.

My confusion may be related to the fact that a thread pool queue system would have multiple worker threads feeding off of a single queue of requests, dishing out jobs to the most available thread.

In all cases you must lock the queue container momentarily in order to manage it or read from it. Whether pushing onto the queue or reading (and therefore removing) entries from the queue, the queue is being manipulated in a 'read/write' fashion. Without a lock, an opposing thread might traverse links that become invalid while in process due to the other thread's alteration of the links in the queue's linked list (if that's it's internal storage method).

Some queues are circular, and can be managed using a vector, typically with fixed sizes. Such queues obviate the problem contention produces where the container is temporarily invalid during deletion or insertion, but the pointers (or markers) that identify the read/write positions of the queue's circular vector container might be 'out of sync' temporarily, and so it's difficult to propose a reliable lock free container.

The lock should be held for a short a time as possible.

When the queue is empty, the worker thread should wait on an event (likely an auto reset event). All submissions to the queue should signal this event, so that the waiting worker thread is 'awakened' when a submission is presented.

[cpplinux]

I'm not 100% certain if you are asking about two worker threads servicing requests in a queue, or simply one thread pushing onto the queue and the worker thread reading from the queue, but I'm perhaps 90% sure you mean the latter.


Yes, it's the latter ----> one thread pushing onto the queue and the worker thread reading from the queue


My confusion may be related to the fact that a thread pool queue system would have multiple worker threads feeding off of a single queue of requests, dishing out jobs to the most available thread.

I don't understand exactly what you mean. In my case, it's 1 thread pushing and 1 worker thread


Without a lock, an opposing thread might traverse links that become invalid while in process due to the other thread's alteration of the links in the queue's linked list (if that's it's internal storage method).


Is the best way to do it so:

1- lock the queue 2- write 3- lock the queue 4- read ?



Some queues are circular, and can be managed using a vector, typically with fixed sizes.


It's not clear for me


I found this code in an other forum but I don't know if it's what I'm looking for.
Can someone explain waht this code do?


class Queue
{
public:
void PushQueue(T t)
{
while (!test_and_set(&m_queueLocked, false, true));

// ... access queue ...

m_queueLocked = false;
}

void PopQueue(T& t)
{
while (!test_and_set(&m_queueLocked, false, true));

// ... access queue ...

m_queueLocked = false;
}

static bool test_and_set(bool volatile* x, bool expected, bool new_value)
{
// ACHTUNG: all dies müsste atomar, also unteilbar erfolgen:
if (*x == expected)
{
*x = new_value;
return true;
}
else
return false;
}

public:
bool volatile m_queueLocked;
};



An other question:

Which library would you use for implementing the threads?


Thanks,

CL

[cpplinux]

Hello,

I've found the following libraries. Does anyone know which one would be the best for my issue (2 threads vs 1 queue)?

1- Posix threads
2- Thread library of the wxWidget library
3- ACE library
4- Boots library

Thanks,

CL

[JVene]

I'm not that familiar with boosts threads, or the ACE library, but the others support little more than a thread launcher and a few synchronization objects.

So far as I've investigated (which hasn't been much in the last few years), I've not found any libraries that popularize the notion of a thread queue.

I built my own years ago, and I've never wanted for anything more. You simply need to pair those few pieces together that gets your task done, namely the auto reset event, a critical section or two, and a queue container.

However, for a generalized solution, the queue container implies an object representing pointer to member functions and a function call (at least in my view). I suppose one could operate it similar to message traffic, essentially create a virtual function that 'switches' on some value to select among operations.

[JohnW@Wessex]

It is possible to devise a queue that can be written to by one thread and read from another without locking.

Create a queue class with a fixed queue length (i.e. not dynamically resizable) with each queue member having an 'in use' flag. Create read and write functions that check the flag before accessing the member.

'Read' checks the next 'out' member to see if it is 'in use' (queue not empty). If so then return the value and clear the 'in use' flag.

'Write' checks to see if the next 'in' member has 'in use' clear (queue not full). If so then write the value and set the 'in use' flag.

With just one write and one read thread this should be thread safe.

[Arjay]

This is pretty simple if you create some lightweight synchronization wrappers using RAII.

Check out the LogSnd project in the article, Win32 Thread Synchronization, Part 2: Helper Classes (http://www.codeguru.com/cpp/w-p/win32/tutorials/article.php/c9825/). The LogSnd project shows how to share a 'logging' object between two processes using a memory mapped file. There are two projects, LogSnd and LogRcv. The LogSnd project first buffers the log objects by putting them into a queue in one thread. A second thread then reads the items off the queue and transfers them to the LogRcv project using the MMF.

The part that may interest you is the queue shared between threads which uses a critical section (wrapped in a RAII type class helper) and an event to signal the secondary thread.

The InsertLogItem method is called from the primary thread. The GetNextLogItemFromQueue and PopLogItemFromQueue methods are called from a secondary thread. See the sample code in the article for the complete source.


//
// Method to retrieve the next item from the queue. This
// approach allows locking the queue for the minimum amount
// of time.
//
LPLOGITEM GetNextLogItemFromQueue()
{
CAutoLockT< CLogItemQueue > lock(&m_LogItemQueue);

if( WAIT_OBJECT_0 == WaitForSingleObject( m_hShutdownEvent, 0 ) )
{
return NULL;
}

if(!m_LogItemQueue.empty())
{
return &(m_LogItemQueue.front());
}

return NULL;
}

//
// Pops an item from the queue
//
void PopLogItemFromQueue()
{
CAutoLockT< CLogItemQueue > lock(&m_LogItemQueue);

if(TRUE != m_LogItemQueue.empty())
{
m_LogItemQueue.pop();
}
}

//
// Inserts an log item into the queue
//
void InsertLogItem( LPLOGITEM pLI )
{
CAutoLockT< CLogItemQueue > lock(&m_LogItemQueue);
m_LogItemQueue.push( *pLI );
SetEvent( m_hLogItemQueueEvent );
}

[Hermit]

It is possible to devise a queue that can be written to by one thread and read from another without locking.

Create a queue class with a fixed queue length (i.e. not dynamically resizable) with each queue member having an 'in use' flag. Create read and write functions that check the flag before accessing the member.

'Read' checks the next 'out' member to see if it is 'in use' (queue not empty). If so then return the value and clear the 'in use' flag.

'Write' checks to see if the next 'in' member has 'in use' clear (queue not full). If so then write the value and set the 'in use' flag.

With just one write and one read thread this should be thread safe.
Not thread safe, since a compiler can reorder or remove operations "with no noticeable effect" such that the two threads do not interact in the way you outline. It's possible that the volatile keyword could remedy this, not sure (I'm very interested to know though). Plus, although I can discern an algorithm that theoretically has no race conditions (I think) from what you described, I don't think your description fully addresses the possible race conditions. An advancing read or write pointer would need to set the next "in use" flag before unsetting the current one, and only after this full advance could it read or write anything.

There's also the problem that one "in use" flag per data member can be an awful lot of overhead for large buffers of small data. It's possible to have one flag per N data, but that has side-effects regarding when the buffer is considered full or empty.

[JohnW@Wessex]

There must be a way of ensuring execution order otherwise it would be impossible to create a hardware driver. Registers often need to be initialised in a particular order. If the compiler were to mess about with the order then the hardware may not work. Maybe volatile is the key? Or maybe pointer access will imply "noticeable effect".

[JohnW@Wessex]

Found this on a web search

The content of a volatile variable is unstable (can change by means unknown to the compiler); all writes to volatile data are observable, so they must be executed religiously; and all operations on volatile data are executed in the sequence in which they appear in the source code.

So declaring the fifo elements as volatile may be enough?