Concurrent Queue w Condition Variable

• The queue we have so far is safe

  • pop() throws an exception when the queue is empty
  • push() throws an exception when the queue is full

• A more useful approach

  • pop() waits until there is some data on the queue
  • push() waits until the queue is no longer full

• We can do this with a condition variable

  • The thread that calls pop() calls wait() on the condition variable
  • The thread that calls push() notifies the condition variable

• We add a predicate to the wait() call

  • If the queue is empty, we continue waiting
  • If it is not empty, then it is safe to continue and pop from the queue
  • Avoids spurious and lost wakeups

concurrent_queue.h

#pragma once

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

using namespace std::chrono_literals;

template <class T> class concurrent_queue {
  std::mutex mut;
  std::queue<T> que;
  size_t max{50};
  std::condition_variable cv;

public:
  concurrent_queue() = default;
  concurrent_queue(size_t max) : max(max) {};

  concurrent_queue(concurrent_queue &&) = delete;
  concurrent_queue(const concurrent_queue &) = delete;
  concurrent_queue &operator=(concurrent_queue &&) = delete;
  concurrent_queue &operator=(const concurrent_queue &) = delete;

  void push(T value) {
    std::unique_lock<std::mutex> uniq_lck(mut);

    while (que.size() > max) {
      uniq_lck.unlock();
      std::this_thread::sleep_for(50ms);
      uniq_lck.lock();
    }

    que.push(value);
    cv.notify_one();
  }

  void pop(T &value) {
    std::unique_lock<std::mutex> uniq_lck(mut);

    cv.wait(uniq_lck, [this] { return !que.empty(); });

    value = que.front();
    que.pop();
  }
};

queue_main.cpp

#include "concurrent_queue.h"
#include <future>
#include <iostream>
#include <string>
#include <thread>

// Shared queue object
concurrent_queue<std::string> conc_que;

// Waiting thread
void reader() {
  using namespace std::chrono_literals;
  std::this_thread::sleep_for(2s); // Pretend to be busy
  std::string sdata;

  // Pop some elements from the queue
  std::cout << "Reader calling pop...\n";
  for (int i = 0; i < 60; ++i) {
    conc_que.pop(sdata);
    std::cout << "Reader received data: " << sdata << '\n';
  }
}

// Modifying thread
void writer() {
  // Push the data onto the queue
  std::this_thread::sleep_for(2s);
  std::cout << "Writer calling push...\n";

  for (int i = 0; i < 60; ++i) {
    std::string sdata = "Item " + std::to_string(i);
    conc_que.push(sdata);
  }
  std::cout << "Writer returned from push...\n";
}

int main() {
  auto read_fut = std::async(std::launch::async, reader);
  auto write_fut = std::async(std::launch::async, writer);

  // Wait for them to complete
  read_fut.wait();
  write_fut.wait();
}

Conclusion

• This is a simple concurrent queue

• It employs "coarse-grained" locking

  • Only one thread can access the queue at any one time
  • In effect, the program becomes single threaded

• Adding the condition variable improves this slightly

  • If the queue is empty and a thread is trying to pop()
  • Other threads can run, until the queue is no longer empty

• A lock free solution would be more efficient