I've seen the following type of a dispatch queue implementation several times where the thread pushing a new element to the queue calls notify_one only when the queue was empty before pushing the element. This condition would reduce unnecessary notify_one calls because q.size() != 0 before pushing a new element means there are only active threads (assuming there are multiple consumer threads).

#include <queue>
#include <condition_variable>
#include <mutex>

using Item = int;
std::queue<Item> q;
std::condition_variable cv;
std::mutex m;

void process(Item i){}

void pop() {
  while (true) {
    std::unique_lock<std::mutex> lock(m);
    // This thread releases the lock and start waiting.
    // When notified, the thread start trying to re-acquire the lock and exit wait().
    // During the attempt (and thus before `pop()`), can another `push()` call acquire the lock? 
    cv.wait(lock, [&]{return !q.empty();});
    auto item = q.front();
    q.pop();
    process(item);
  }
}

void push(Item i) {
  std::lock_guard<std::mutex> lock(m);
  q.push(i);
  if (q.size() == 1) cv.notify_one();
}

int main() { /* ... */ }

However, is the following scenario possible ? Suppose that all the consumer threads are waiting.

1. the pushing thread acquires the lock and push a new element and calls notify_one because the queue was empty.
2. a notified thread tries to re-acquire the lock (and exit wait())
3. the pushing thread acquires the lock again and pushes another element before a notified thread re-acquires the lock.

In this case, no notify_one call wouldn't occur after 3. and there would be only one active thread when the queue isn't empty.

CodePudding user response：

For the operation after wake up

According to wake's doc

1. Atomically unlocks lock, blocks the current executing thread, and adds it to the list of threads waiting on *this. The thread will be unblocked when notify_all() or notify_one() is executed. It may also be unblocked spuriously. When unblocked, regardless of the reason, lock is reacquired and wait exits.

Means your code is atomic between #1 to #2 when it's woken up. Don't need to worry about the synchronization since your compiler should handle it.

void pop() {
  while (true) {
    std::unique_lock<std::mutex> lock(m);

    // When sleep, cv release the lock
    cv.wait(lock, [&]{return !q.empty();});   // <----#1
    // When wake up, it acquires the lock
    auto item = q.front();
    q.pop();
    process(item);                            // <----#2
  }
}

For spurious wakeup

The second overload with predicate would handle spurious wakeup for you. template< class Predicate > void wait( std::unique_lock<std::mutex>& lock, Predicate stop_waiting ); (2) (since C 11)

2. Equivalent to

while (!stop_waiting()) {
    wait(lock);
}

This overload may be used to ignore spurious awakenings while waiting for a specific condition to become true.

Note that lock must be acquired before entering this method, and it is reacquired after wait(lock) exits, which means that lock can be used to guard access to stop_waiting().

That is the condition_variable overload with predicate would save you from spurious wake up. The handmade while loop for checking spurious wake up is not needed.