10

I have a C++11 application with a high-priority thread that's producing data, and a low-priority thread that's consuming it (in my case, writing it to disk). I'd like to make sure the high-priority producer thread is never blocked, i.e. it uses only lock-free algorithms.

With a lock-free queue, I can push data to the queue from the producer thread, and poll it from the consumer thread, thus meeting my goals above. I'd like to modify my program so that the consumer thread blocks when inactive instead of polling.

It seems like the C++11 condition variable might be useful to block the consumer thread. Can anyone show me an example of how to use it, while avoiding the possibility that the consumer sleeps with data still in the queue? More specifically, I want to make sure that the consumer is always woken up some finite time after the producer pushes the last item into the queue. It's also important that the producer remains non-blocking.

nonagon
  • 3,271
  • 1
  • 29
  • 42
  • See a usage example of condition variable in http://codereview.stackexchange.com/questions/39199/multi-producer-consumer-queue-without-boost-in-c11 – GabiMe Jan 28 '14 at 19:38
  • It's impossible to do this completely lock-free with a condition variable, because to wait or signal on one you have to have acquired a lock. In practice that doesn't make a real difference though (most locks will have a short busy waiting strategy which should be enough to avoid any kernel calls or otherwise since the critical section for the reader is incredibly short) – Voo Jan 28 '14 at 19:41
  • 1
    @Voo It's not necessary to acquire a lock to signal a condition variable. – Casey Jan 28 '14 at 19:49
  • @Casey Where do you get that information from? [This](http://en.cppreference.com/w/cpp/thread/condition_variable/wait) clearly states `Calling this function if lock.mutex() is not locked by the current thread is undefined behavior`. – Voo Jan 28 '14 at 19:52
  • 2
    @Voo *Waiting* on a condition variable requires a lock, *signaling* it does not. – Casey Jan 28 '14 at 19:57
  • @Casey Ah right you are. – Voo Jan 28 '14 at 19:59
  • 1
    @Casey have you seen documentation stating that signaling a condition variable is non-blocking? I assume so as well, but I've never been able to find explicit confirmation anywhere. – nonagon Jan 28 '14 at 20:23
  • @nonagon No I haven't - I believe it's only implicit. – Casey Jan 28 '14 at 20:50
  • Functions which can block a thread are explicitly stated to do so, and `notify_one()` and `notify_all()` are not stated to do so. – Jonathan Wakely Jan 28 '14 at 23:55

3 Answers3

2

It seems like the C++11 condition variable might be useful to block the consumer thread. Can anyone show me an example of how to use it, while avoiding the possibility that the consumer sleeps with data still in the queue?

To use a condition variable you need a mutex and a condition. In your case the condition will be "there is data available in the queue". Since the producer will be using lock-free updates to produce work, the consumer has to use the same form of synchronisation to consume the work, so the mutex will not actually be used for synchronisation and is only needed by the consumer thread because there's no other way to wait on a condition variable.

// these variables are members or otherwise shared between threads
std::mutex m_mutex;
std::condition_variable m_cv;
lockfree_queue m_data;

// ...

// in producer thread:
while (true)
{
  // add work to queue
  m_data.push(x);
  m_cv.notify_one();
}

// in consumer thread:
while (true)
{
  std::unique_lock<std::mutex> lock(m_mutex);
  m_cv.wait(lock, []{ return !m_data.empty(); });
  // remove data from queue and process it
  auto x = m_data.pop();
}

The condition variable will only block in the wait call if the queue is empty before the wait. The condition variable might wake up spuriously, or because it was notified by the producer, but in either case will only return from the wait call (rather than sleeping again) if the queue is non-empty. That's guaranteed by using the condition_variable::wait overload that takes a predicate, because the condition variable always re-checks the predicate for you.

Since the mutex is only used by the consumer thread it could in fact be local to that thread (as long as you only have one consumer, with more than one they all need to share the same mutex to wait on the same condvar).

Jonathan Wakely
  • 166,810
  • 27
  • 341
  • 521
  • If the producer calls notify_one() just before the consumer calls wait(), won't the consumer end up sleeping with data in the queue (and thus sleeping forever if that data is the last data ever produced)? – nonagon Jan 29 '14 at 04:22
  • No, because the call to `wait` will check its predicate before sleeping, so if `m_data_available` is already true it won't sleep. However in the real code you probably don't want to use a simple `atomic` to indicate data is available, you would want to check `m_queue.size()` or `m_available_count` otherwise the consumer doesn't know if it should set `m_data_available=false` after consuming a single piece of data, as there could be more available. – Jonathan Wakely Jan 29 '14 at 08:32
  • I've edited the answer to replace the `atomic` with operations on a lock-free queue object, to address the issue described in the comment above, and clarified that the CV won't block if there's already data in the queue – Jonathan Wakely Jan 29 '14 at 10:19
  • I believe that form of wait() does not atomically check the predicate and enter the waiting state. So if another thread produces data in between the predicate check and the wait, it can end up waiting forever with data in the queue. – nonagon Jan 29 '14 at 15:57
  • Ah yes, that's true - locking the mutex prevents that if the result of the predicate will only be changed by a thread that holds the mutex, but in your case you don't want the producer to use the mutex as it could block. So much for that idea – Jonathan Wakely Jan 29 '14 at 16:32
  • I don't see how you can actually use signal() without acquiring the lock without running into this race condition. Waiting on the lock needs a kernel call (at least in some situations) and you can't do that atomically with checking the condition. And the kernel can't execute arbitrary user code either. That's probably also the reason why most other languages forbid signaling on a condition variable without holding the lock.. – Voo Jan 29 '14 at 18:02
  • @Voo, signalling doesn't need the lock, only changing whatever state the predicate tests needs the lock – Jonathan Wakely Jan 29 '14 at 19:04
  • @Jonathan True, it's just the usual idiom to lock the whole method and not release the lock before signaling - at least that's what I've seen usually. Shouldn't have a performance cost anyhow except in the rare occasions where the notify is the last action of the thread's timeslice. – Voo Jan 29 '14 at 19:09
  • @JonathanWakely In the example what is lockfree_queue? Can we just use a standard queue for this? – Nubcake Mar 16 '21 at 17:46
  • @Nubcake it's some unspecified lock-free queue type. No you can't use `queue` because it's not thread-safe. – Jonathan Wakely Mar 17 '21 at 18:20
0

One solution I found to this in the past was using Windows events (http://msdn.microsoft.com/en-us/library/windows/desktop/ms682396(v=vs.85).aspx). In this case the event remains signaled until it wakes up a waiting thread, and if no threads are waiting it remains signaled. So the producer simply needs to signal the event after pushing data to the queue. Then we are guaranteed that the consumer will wake up some finite time after this.

I wasn't able to find a way to implement this using the standard library though (at least not without blocking the producer thread).

nonagon
  • 3,271
  • 1
  • 29
  • 42
0

I think semaphores could be used to solve this problem safely:

// in producer thread:
while (true)
{
  m_data.push();
  m_semaphore.release();
}

// in consumer thread:
while (true)
{
  m_semaphore.wait();
  m_data.pop();
}

Unfortunately I don't think C++11 includes a semaphore? I have also not been able to confirm that releasing a semaphore is a non-blocking operation. Certainly implementations with mutexes e.g. C++0x has no semaphores? How to synchronize threads? will not allow for a non-blocking producer thread.

Community
  • 1
  • 1
nonagon
  • 3,271
  • 1
  • 29
  • 42