Single producer multiple consumers C++

I am trying to implement a program that consists of a producer thread adding objects to a std::vector and multiple consumer threads removing objects from the same vector until it's empty. I am using a condition_variable to let the consumers know that new objects have been produced. Problem is that in last iteration (n items left in storage where n is number of consumer threads), consumer threads get stuck waiting on a conditional variable, even though that condition should not be met (storage is not empty -> at least that's what I figured with some debug logs).

#include <chrono>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

#define CONSUMER_COUNT 4
#define STORAGE_SIZE CONSUMER_COUNT * 10000

class Foo {
private:
  int _id;

public:
  Foo(int id) : _id(id) {}
  int getId() const { return _id; }
};

std::vector<Foo> storage;
std::mutex storageMutex;
std::condition_variable storageCV;

void Producer(int limit) {
  for (int i = 0; i < limit; ++i) {
    std::lock_guard<std::mutex> lg{storageMutex};
    storage.emplace_back(Foo(i));
    storageCV.notify_one();
  }
  storageCV.notify_all();
}

void Consumer(int id) {
  while (true) {
    std::unique_lock<std::mutex> ul{storageMutex};
    storageCV.wait(ul, []() { return !storage.empty(); });
    if (storage.empty())
      return;
    storage.pop_back();
  }
}

int main(int argc, char *argv[]) {
  std::vector<std::thread> consumers;
  consumers.reserve(CONSUMER_COUNT);

  auto producer = std::thread(Producer, STORAGE_SIZE);

  for (int i = 0; i < CONSUMER_COUNT; ++i) {
    consumers.emplace_back(std::thread(Consumer, i));
  }

  producer.join();
  for (auto &consumer : consumers)
    consumer.join();

  storageCV.notify_all();

  std::cout << "[MAIN] Done!" << std::endl;
  std::cout << "Storage is left with " << storage.size() << " items!"
            << std::endl;

  return 0;
}

I have tried adding a simple boolean flag that producer will toggle once it's done with adding all of the items, but then I am not sure (logically) how should I set the condition in consumer threads. Simply adding that check on top of current one is not enough because then a thread might stop running even though there are still some items in the storage.

As you've discovered, the problem is that consumers get stuck. The problem is here:

storageCV.wait(ul, []() { return !storage.empty(); });
if (storage.empty())
    return;

Note that std::condition::variable::wait(lock, condition) is just a convenience function, and this code is equivalent to:

while (storage.empty())
    storageCV.wait();
if (storage.empty())
    return;

It's easy to see that the conditional return is pointless, because when we reach it, the condition is always false.
The condition would need to be inside the loop to have any effect.

These adjustments are necessary:

1. We have to modify this loop so that it's possible for a consumer to not wait infinitely.

2. The producer thread has to wake up any currently waiting consumers once all the elements have been processed.

3. It's possible that a consumer thread is currently:

   • right before while (storage.empty()), or
   • right before storageCV.wait()

   ... so it is not going to be woken up by notify_all(), because it's not waiting yet, but is about to. We have to repeatedly .notify_all() to wake up every thread, even those that are not waiting yet.

Changes to Producer

// We create an atomic counter to keep track of how many
// consumers remain.
// When this hits zero, the producer can stop notifying the
// remaining consumers.
std::atomic_int consumer_stop_counter;

// The producer needs to receive the number of consumers from the main thread.
// It wouldn't be safe to communicate this via an atomic counter which
// is modified by the consumers, because it's possible that the producer
// is done pushing all of its items to storage before any consumers
// "register themselves".
// In such a scenario, the producer would think that there are no
// consumers that need to be notified anymore and all works has been completed.
void Producer(int limit, int consumers) {
  for (int i = 0; i < limit; ++i) {
    std::lock_guard<std::mutex> lg{storageMutex};
    storage.emplace_back(Foo(i));
    storageCV.notify_one();
  }

  consumer_stop_counter = 0;
  while (consumer_stop_counter < consumers) {
    storageCV.notify_all();

    // optional: mitigate effects of busy wait
    std::this_thread::yield();
  }
}

Changes to Consumer

void Consumer(int id) {
  while (true) {
    std::unique_lock<std::mutex> ul{storageMutex};
    // Transform cv waiting into a regular while-loop.
    while (storage.empty()) {
      storageCV.wait(ul);

      // If we got woken up and the storage is empty, that means that we exit.
      // We hold the lock after calling .wait(), so it is safe to access the storage.
      if (storage.empty()) {
        consumer_stop_counter++;
        return;
      }
    }

    storage.pop_back();
  }
}

</details>