Blocking Queue in C++ using semaphores

Question

This is mostly an exercise for me to try to understand the differences between semaphores and locks. It's a long and rambling because it took me quite a few tries to understand the concepts. Please bear with me. I am hoping you can either confirm that the lesson I learned was correct or pointing out my misunderstanding. Please jump to the last code section if you just would like to see my final code.

I read about this blog: https://vorbrodt.blog/2019/02/03/blocking-queue/ and it really confused me. If we were going to serialize access to the elements, what's the point of the semaphore? I originally thought a semaphore is basically a counter protected by a lock, so I was having trouble understand the differences. I decided to implement it myself without using a semaphore. Here is my first (incorrect) attempt of implementing a blocking queue with one producer and one consumer:

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

template <typename T>
class OneToOneBlockingQueue {
private:
  unsigned int m_maxSize;
  std::queue <T> m_data;
  std::mutex m_mutex;
  std::condition_variable m_readCond;
  std::condition_variable m_writeCond;
public:
  OneToOneBlockingQueue(unsigned int size): m_maxSize(size) {
  }

  void push(T value) {
    std::unique_lock <std::mutex> myLock(m_mutex);
    m_writeCond.wait(myLock, [this]() { return m_data.size() < m_maxSize; });
    m_data.push(value);
    m_readCond.notify_one();
  }

  void pop(T& value) {
    std::unique_lock <std::mutex> myLock(m_mutex);
    m_readCond.wait(myLock, [this]() { return m_data.size() > 0; });

    value = m_data.front();
    m_data.pop();
    m_writeCond.notify_one();
  }
};

class Producer {
public:
  Producer(OneToOneBlockingQueue <int>& bq, int id):m_bq(bq), m_id(id) {
  }
  
  void operator()() {
    for (int i = 0; i < 10; i++) {
      m_bq.push(i);
    }
  }
private:
  OneToOneBlockingQueue<int> &m_bq;
  int m_id;
};

class Consumer {
public:
  Consumer(OneToOneBlockingQueue <int>& bq, int id):m_bq(bq), m_id(id) {
  }

  void operator()() {
    std::cout << "Reading from queue: ";
    for (int i = 0; i < 10; i++) {
      int value;
      m_bq.pop(value);
      std::cout << value << " ";
    }

    std::cout << std::endl;
  }
private:
  OneToOneBlockingQueue <int> &m_bq;
  int m_id;
};

int main() {
  OneToOneBlockingQueue <int>bq(2);

  std::thread producerThread (Producer(bq, 0));
  std::thread consumerThread (Consumer(bq, 0));

  producerThread.join();
  consumerThread.join(); 
} 

While it worked, I then realized it was not correct since the producer and the consumer can't read and write at the same time. Assuming the consumer is very slow, the producer would be locked out until the consumer finished reading even though the queue is not full. The only critical section is the counter, not the data itself. However, using std::queue, I couldn't separate the two. Maybe that is why the other author used an looping array instead?

Here is my second attempt:

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <thread>

template <typename T>
class OneToOneBlockingQueue {
private:
  unsigned int m_maxSize;
  T *m_data;
  unsigned int m_size;
  std::mutex m_mutex;
  std::condition_variable m_readCond;
  std::condition_variable m_writeCond;
  unsigned int m_readLoc;
  unsigned int m_writeLoc;
public:
  OneToOneBlockingQueue(unsigned int size): m_maxSize(size), m_size(0), m_data(new T[size]), m_readLoc(0), m_writeLoc(0) {
  }

  void push(T value) {
    std::unique_lock <std::mutex> myLock(m_mutex);
    m_writeCond.wait(myLock, [this]() { return m_size < m_maxSize; });
    myLock.unlock();

    m_data[m_writeLoc++] = value;
    if (m_writeLoc == m_maxSize) {
      m_writeLoc = 0;
    }

    myLock.lock();
    m_size++;
    m_readCond.notify_one();
  }

  void pop(T& value) {
    std::unique_lock <std::mutex> myLock(m_mutex);
    m_readCond.wait(myLock, [this]() { return m_size > 0; });
    myLock.unlock();

    value = m_data[m_readLoc++];
    if (m_readLoc == m_maxSize) {
      m_readLoc = 0;
    }

    myLock.lock();
    m_size--;
    m_writeCond.notify_one();
  }
};

class Producer {
public:
  Producer(OneToOneBlockingQueue <int>& bq, int id):m_bq(bq), m_id(id) {
  }
  
  void operator()() {
    for (int i = 0; i < 10; i++) {
      m_bq.push(i);
    }
  }
private:
  OneToOneBlockingQueue<int> &m_bq;
  int m_id;
};

class Consumer {
public:
  Consumer(OneToOneBlockingQueue <int>& bq, int id):m_bq(bq), m_id(id) {
  }

  void operator()() {
    std::cout << "Reading from queue: ";
    for (int i = 0; i < 10; i++) {
      int value;
      m_bq.pop(value);
      std::cout << value << " ";
    }
    std::cout << std::endl;
  }
private:
  OneToOneBlockingQueue <int> &m_bq;
  int m_id;
};

int main() {
  OneToOneBlockingQueue <int>bq(2);

  std::thread producerThread (Producer(bq, 0));
  std::thread consumerThread (Consumer(bq, 0));

  producerThread.join();
  consumerThread.join();
}

I think the differences between semaphore and lock is that the semaphore by itself does not protect the elements, only the use count. The producer and consumer must inherently work on different elements for it to work. Is that correct?

Here is the code after abstracting the counter into a semaphore class.

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <thread>

class Semaphore {
private:
  unsigned int m_counter;
  std::mutex m_mutex;
  std::condition_variable m_cond;
public:
  Semaphore(unsigned int counter):m_counter(counter) {
  }

  void P() {
    std::unique_lock <std::mutex> myLock(m_mutex);
    m_cond.wait(myLock, [this]() { return m_counter > 0; });
    m_counter--;
  }

  void V() {
    std::lock_guard <std::mutex> myLock(m_mutex);
    m_counter++;
    m_cond.notify_one();
  }
};

template <typename T>
class OneToOneBlockingQueue {
private:
  unsigned int m_maxSize;
  T *m_data;
  Semaphore m_filledSlots;
  Semaphore m_emptySlots;
  unsigned int m_readLoc;
  unsigned int m_writeLoc;
public:
  OneToOneBlockingQueue(unsigned int size): m_maxSize(size), m_data(new T[size]), m_filledSlots(0), m_emptySlots(size), m_readLoc(0), m_writeLoc(0) {
  }

  void push(T value) {
    m_emptySlots.P();

    m_data[m_writeLoc++] = value;
    if (m_writeLoc == m_maxSize) {
      m_writeLoc = 0;
    }

    m_filledSlots.V();
  }

  void pop(T& value) {
    m_filledSlots.P();

    value = m_data[m_readLoc++];
    if (m_readLoc == m_maxSize) {
      m_readLoc = 0;
    }

    m_emptySlots.V();
  }
};

class Producer {
public:
  Producer(OneToOneBlockingQueue <int>& bq, int id):m_bq(bq), m_id(id) {
  }
  
  void operator()() {
    for (int i = 0; i < 10; i++) {
      m_bq.push(i);
    }
  }
private:
  OneToOneBlockingQueue<int> &m_bq;
  int m_id;
};

class Consumer {
public:
  Consumer(OneToOneBlockingQueue <int>& bq, int id):m_bq(bq), m_id(id) {
  }

  void operator()() {
    std::cout << "Reading from queue: ";
    for (int i = 0; i < 10; i++) {
      int value;
      m_bq.pop(value);
      std::cout << value << " ";
    }
    std::cout << std::endl;
  }
private:
  OneToOneBlockingQueue <int> &m_bq;
  int m_id;
};

int main() {
  OneToOneBlockingQueue <int>bq(2);

  std::thread producerThread (Producer(bq, 0));
  std::thread consumerThread (Consumer(bq, 0));

  producerThread.join();
  consumerThread.join();  
}

And finally, to allow multiple consumer, we only need to worry about producers and consumers separately. Semaphores do not work between consumers (or producers) since it does not provide exclusive access to individual elements. So I created a producerMutex and a consumerMutex. The reason the original blog post confused me was because he was using a single mutex, which made me think the semaphore was unnecessary. Here is my final code:

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <vector>
#include <queue>
#include <unistd.h>

class Semaphore {
private:
  unsigned int m_counter;
  std::mutex m_mutex;
  std::condition_variable m_cond;
public:
  Semaphore(unsigned int counter):m_counter(counter) {
  }

  void P() {
    std::unique_lock <std::mutex> myLock(m_mutex);
    m_cond.wait(myLock, [this]() { return m_counter > 0; });
    m_counter--;
  }

  void V() {
    std::lock_guard <std::mutex> myLock(m_mutex);
    m_counter++;
    m_cond.notify_one();
  }
};

template <typename T>
class ManyToManyBlockingQueue {
private:
  unsigned int m_maxSize;
  T *m_data;
  Semaphore m_filledSlots;
  Semaphore m_emptySlots;
  unsigned int m_readLoc;
  unsigned int m_writeLoc;
  std::mutex m_consumerMutex;
  std::mutex m_producerMutex;
public:
  ManyToManyBlockingQueue(unsigned int size): m_maxSize(size), m_data(new T[size]), m_filledSlots(0), m_emptySlots(size), m_readLoc(0), m_writeLoc(0) {
  }

  void push(T value) {
    m_emptySlots.P();

    std::unique_lock <std::mutex> producerLock(m_producerMutex);
    m_data[m_writeLoc++] = value;
    if (m_writeLoc == m_maxSize) {
      m_writeLoc = 0;
    }
    producerLock.unlock();

    m_filledSlots.V();
  }

  void pop(T& value) {
    m_filledSlots.P();

    std::unique_lock <std::mutex>consumerLock(m_consumerMutex);
    value = m_data[m_readLoc++];
    if (m_readLoc == m_maxSize) {
      m_readLoc = 0;
    }
    consumerLock.unlock();

    m_emptySlots.V();
  }
};

class Producer {
public:
  Producer(ManyToManyBlockingQueue <int>& bq, int id):m_bq(bq), m_id(id) {
  }
  
  void operator()() {
    for (int i = 0; i < 10; i++) {
      m_bq.push(m_id*10+i);
    }
  }
private:
  ManyToManyBlockingQueue<int> &m_bq;
  int m_id;
};

class Consumer {
public:
  Consumer(ManyToManyBlockingQueue <int>& bq, int id, std::queue <int>&output):m_bq(bq), m_id(id), m_output(output) {
  }

  void operator()() {
    for (int i = 0; i < 50; i++) {
      int value;
      m_bq.pop(value);
      m_output.push(value);
    }
  }
private:
  ManyToManyBlockingQueue <int> &m_bq;
  int m_id;
  std::queue<int> &m_output;
};

int main() {
  ManyToManyBlockingQueue <int>bq(10);

  std::vector <std::thread> producerThreads;
  std::vector <std::thread> consumerThreads;
  std::vector <std::queue<int>> outputs;

  for (int i = 0; i < 10; i++) {
    producerThreads.emplace_back(Producer(bq,i));
  }

  for (int i = 0; i < 2; i++) {
    outputs.emplace_back(std::queue<int>());
  }
  
  for (int i = 0; i < 2; i++) {
    consumerThreads.emplace_back(Consumer(bq,i,outputs[i]));
  }
  
  for (std::vector <std::thread>::iterator it = producerThreads.begin();
       it != producerThreads.end();
       it++) {
    it->join();
  }

  for (std::vector <std::thread>::iterator it = consumerThreads.begin();
       it != consumerThreads.end();
       it++) {
    it->join(); 
  }

  for (std::vector <std::queue<int>>::iterator it = outputs.begin();
       it != outputs.end();
       it++) {
    std::cout << "Number of elements: " << it->size() << " Data: ";
    while(!it->empty()) {
      std::cout << it->front() << " ";
      it->pop();
    }
    std::cout << std::endl;
  }
}

Am I doing this correctly?

A couple of other issues I have with this code. The pop() function bugs me. I would really like it to return the value so that the caller can use it directly instead of having to have a temp variable. However, I can't access it after I have V() the other semaphore or a producer might overwrite it. Holding the lock longer would decrease parallelism. Is this the right way to do it or there's a better way?

The other thing is that I was new to references in C++ as I mostly used pointers before. Originally, I allocated the output queue as I was creating the thread and I was surprised that I wasn't getting any data from the first consumer. After a lot of debugging, I finally realized that the vector moved in order to grow in size. Therefore, it seems that passing a movable object by reference is dangerous. What's the best way to solve that?

Another issue is how best to allow producer to signal end of data. Would a "done" counter protected by another mutex be the right way?

Another issue is what if one partner does not respond for a while. I can't really free the queue since there is no guarantee that the partner wouldn't come back later and write into bad memory. What's the best way to handle it and abort the operation?

Sorry again about the long post. Thanks for your inputs.

p.s. I understand semaphores can behave quite differently depends on the implementation (e.g. interrupt), this is not meant to be a production code, just to understand the concept.

Answer 1

There's too much state

Each queue has four mutexes, four counters and two condition variables. That is way too much. You could do this with just a single mutex and condition variable.

In your push() function, you first have to hold a mutex at least once to check if there are empty slots (if not, you have to wait for a condition variable to be signalled, which means multiple calls the mutex lock and unlock functions), then you have to hold a mutex to update the write location, and then hold the mutex to increment the filled slots semaphore. Locking and unlocking a mutex, despite being quite optimized, is still not free.

Another issue is the duplication of information of the state of the queue. There's m_filledSlots, m_emptySlots (which should be the inverse), and the same information is also present in the difference between the read and write locations. And you have to keep everything updated.

If you just take one lock, check the read and write pointers to see how many free slots there are in the queue, wait for the condition variable if necessary, then update the read or write pointer, signal the variable if necessary, and then unlock, you have spent much less cycles than with this approach with semaphores.

Making pop() return the value

You can just write:

T pop() {
    ...
    T value = m_data[m_readLoc++];
    ...
    return value;
}

Even though it looks like there is a temporary variable that would require an extra copy, the compiler can perform return value optimization here, which is mandatory in C++17, and which most compilers have been doing already for much longer.

Pointers moving when containers grow

Indeed, a std::vector will move its contents in memory if it grows. However, there are other container classes that you can use that do guarantee that elements already in the container will keep their address, even if it needs to allocate more memory. Amongst them are std::list and std::deque. There are also container adapter classes such as std::queue that by default use a std::deque for storage, and thus inherit its properties.

Signalling that production ended

There are two common ways to do this. First is to add a flag variable to your blocking queue class that signals that the producers finished. This flag is set, and then the condition variable that the consumers listen for is broadcast to. Consumers should check this flag each time they want to dequeue an item. If it's set, they can terminate.

The other way is to have some way to enqueue an item that signals that no more data will be coming. If your queue contains pointers to objects, enqueueing a nullptr might suffice. Again, the condition variable should be broadcast, and a consumer should not actually pop this item from the queue, so that other consumers also get a chance to see it. Alternatively, you have to enqueue as many of these special items as there are consumer threads.

Timeouts

Another issue is what if one partner does not respond for a while. I can't really free the queue since there is no guarantee that the partner wouldn't come back later and write into bad memory. What's the best way to handle it and abort the operation?

I'm not sure what you mean by "partner". Is it a consumer or a producer thread? In any case, you can only delete the queue if no threads are left that could read or write from it. You could kill threads that don't respond in time, but it is very hard to do this in a safe way. The best way is to ensure these threads never take too much time to produce or consume an item to begin with.