C++ – concurrent queue in C++

cmultithreadingpthreads

I am trying to design a queue which could be simultaneously accessed by multiple read/write threads. I prefer using 2 mutexes, one apiece for read and write. Doing write is simple enough, lock the write mutex, append data, unlock and you are done.

The issue is with read. If there's no data in the in queue I'd like my thread to wait till data is available. 1 obvious way to do this is to acquire read mutex and keep polling the queue every N clock cycles but this is obviously not the best approach. Which brings me to condition variables. Does anybody have any good resource which discusses blocking queue implementation in C++ with condition variables (preferably pthreads based)?

Specifically, I see the following issues:

Once a write is done, the writer thread would do a pthread_cond_signal that data exists but how does it know that some reader thread is waiting? It is illegal to call pthread_cond_signal unless there is a pthread_cond_wait.
Is it okay to call pthread_cond_broadcast instead of pthread_cond_signal? Perhaps that might circumvent the issue with pthread_cond_wait. Also this seems more logical as multiple reader threads is definitely a real possibility.
It also appears that the reader and writer threads must be locked with the same mutex to make use of the condition variable. If this is true then we have a severe problem.

Best Answer

I have an implementation using the same mutex from http://danborn.net/code/ but like I mentioned earlier since this uses a condition variable it also uses 1 mutex.

And here's the boost version, again with single mutex: http://www.justsoftwaresolutions.co.uk/threading/implementing-a-thread-safe-queue-using-condition-variables.html

Related Solutions

Recursive Lock (Mutex) vs Non-Recursive Lock (Mutex)

The difference between a recursive and non-recursive mutex has to do with ownership. In the case of a recursive mutex, the kernel has to keep track of the thread who actually obtained the mutex the first time around so that it can detect the difference between recursion vs. a different thread that should block instead. As another answer pointed out, there is a question of the additional overhead of this both in terms of memory to store this context and also the cycles required for maintaining it.

However, there are other considerations at play here too.

Because the recursive mutex has a sense of ownership, the thread that grabs the mutex must be the same thread that releases the mutex. In the case of non-recursive mutexes, there is no sense of ownership and any thread can usually release the mutex no matter which thread originally took the mutex. In many cases, this type of "mutex" is really more of a semaphore action, where you are not necessarily using the mutex as an exclusion device but use it as synchronization or signaling device between two or more threads.

Another property that comes with a sense of ownership in a mutex is the ability to support priority inheritance. Because the kernel can track the thread owning the mutex and also the identity of all the blocker(s), in a priority threaded system it becomes possible to escalate the priority of the thread that currently owns the mutex to the priority of the highest priority thread that is currently blocking on the mutex. This inheritance prevents the problem of priority inversion that can occur in such cases. (Note that not all systems support priority inheritance on such mutexes, but it is another feature that becomes possible via the notion of ownership).

If you refer to classic VxWorks RTOS kernel, they define three mechanisms:

mutex - supports recursion, and optionally priority inheritance. This mechanism is commonly used to protect critical sections of data in a coherent manner.
binary semaphore - no recursion, no inheritance, simple exclusion, taker and giver does not have to be same thread, broadcast release available. This mechanism can be used to protect critical sections, but is also particularly useful for coherent signalling or synchronization between threads.
counting semaphore - no recursion or inheritance, acts as a coherent resource counter from any desired initial count, threads only block where net count against the resource is zero.

Again, this varies somewhat by platform - especially what they call these things, but this should be representative of the concepts and various mechanisms at play.

C++ – Calling pthread_cond_signal without locking mutex

If you do not lock the mutex in the codepath that changes the condition and signals, you can lose wakeups. Consider this pair of processes:

Process A:

pthread_mutex_lock(&mutex);
while (condition == FALSE)
    pthread_cond_wait(&cond, &mutex);
pthread_mutex_unlock(&mutex);

Process B (incorrect):

condition = TRUE;
pthread_cond_signal(&cond);

Then consider this possible interleaving of instructions, where condition starts out as FALSE:

Process A                             Process B

pthread_mutex_lock(&mutex);
while (condition == FALSE)

                                      condition = TRUE;
                                      pthread_cond_signal(&cond);

pthread_cond_wait(&cond, &mutex);

The condition is now TRUE, but Process A is stuck waiting on the condition variable - it missed the wakeup signal. If we alter Process B to lock the mutex:

Process B (correct):

pthread_mutex_lock(&mutex);
condition = TRUE;
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);

...then the above cannot occur; the wakeup will never be missed.

(Note that you can actually move the pthread_cond_signal() itself after the pthread_mutex_unlock(), but this can result in less optimal scheduling of threads, and you've necessarily locked the mutex already in this code path due to changing the condition itself).

Best Answer

Related Solutions

Recursive Lock (Mutex) vs Non-Recursive Lock (Mutex)

C++ – Calling pthread_cond_signal without locking mutex

Related Topic