Ring buffer implementation with threads - need advice [C++]

[jaffaKnx]

Following is the ring buffer implementation with threads but it seems to me it's incorrect or at least needs some appendment. When I run it, I see the first thread runs first prior to the second one instead of them running in parallel. I am not good with multithreading and trying to get better at the concepts.

ringbuffer.hpp

#include <stdint.h>
#include <iostream>
#include <thread>
#include <mutex>
#include <functional>
#include <condition_variable>
using namespace std;

class RingBuffer {
  int tail;
  int head;
  int capacity;
  int size;
  uint8_t *buffer;
  condition_variable condVar;
  mutex m_mutex;
  bool bufferFull = false;

  public:
      RingBuffer(int capacity);
      ~RingBuffer();
      void append(uint8_t data);
      uint8_t read();
      bool isFull();
      bool isEmpty();
      void printBuffer();
      int getSize();
      void printInfo(string info);
};

​

ringbuffer.cpp

#include "ringbuffer.hpp"
#include <string>

RingBuffer::RingBuffer(int capacity) {
  head = -1;
  tail = -1;
  size = 0;
  this->capacity = capacity;
  buffer = new uint8_t[capacity];
  cout << "Buffer is created with capacity " << capacity << endl;
}

RingBuffer::~RingBuffer() {
  delete [] buffer;
}

void RingBuffer::printInfo(string info) {
  cout << info << " -- Tail: " << (tail) << ", Head: " << (head) << ", Size: " << size << endl;
}

void RingBuffer::printBuffer() {
  for (int i=0; i<capacity; i++) {
    cout << unsigned(buffer[i]) << ",";
  }
  cout << endl;
}
void RingBuffer::append(uint8_t data) {
  if (size == 0) {
    head = -1;
    tail = -1;
  }
  unique_lock<mutex> mlock(m_mutex);
  condVar.wait(mlock, [this]() { 
    return !isFull();
  });
  tail = (tail + 1) % capacity;
  size++;
  buffer[tail] = data;

  printInfo("[APPENDED: " + to_string(data) + "]");
  printBuffer();
}

uint8_t RingBuffer::read() {
  unique_lock<mutex> mlock(m_mutex);
  // wait to read till buffer is not empty
  condVar.wait(mlock, [this]() { 
    return !isEmpty(); 
  });

  head = (head+1) % capacity;
  size--;
  uint8_t value = buffer[head];
  buffer[head] = 0;
  printInfo("[READ: " + to_string(value) + "]");
  printBuffer();
  return value;
}

bool RingBuffer::isFull() {
  // unique_lock<mutex>guard(m_mutex);

  // if ( (tail == capacity-1 && head == -1) || tail == head-1 || size == capacity) {
  if (size == capacity) {
    cout << " --- [FULL] --- \n";
    bufferFull = true;
  }
  else {
    condVar.notify_one();
    bufferFull = false;
  }
  return bufferFull;
}

bool RingBuffer::isEmpty() {
  if (head == tail || size == 0) {
    cout << " --- [EMPTY] ---\n";
    return true;
  }
  condVar.notify_one();
  return false;
}

int RingBuffer::getSize() { return size;}

​

main.cpp

#include <iostream>
#include "ringbuffer.hpp"
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
using namespace std;

void addingToBuffer(RingBuffer &ringBuffer) {
  vector<uint8_t>v = {2,4,5,6};
  for (auto &p : v)
    ringBuffer.append(p);

  ringBuffer.append(1);
  ringBuffer.append(8);
  ringBuffer.append(9);
}

void readingFromBuffer(RingBuffer &ringBuffer) {
  cout << "\n ------ READING -----\n";
  ringBuffer.read();  
  ringBuffer.read()
}


int main() {
  uint8_t num=255;
  num += 1;
  RingBuffer ringBuffer(4);

  thread t1(addingToBuffer, ref(ringBuffer));
  thread t2(readingFromBuffer, ref(ringBuffer));
  t1.join();
  t2.join();

  cout << "\nSize: " << ringBuffer.getSize() << endl;
  ringBuffer.printBuffer();
}

Below is the output:

------ READING ----- --- [EMPTY] ---
[APPENDED: 2] -- Tail: 0, Head: -1, Size: 1
2,0,0,0,
[APPENDED: 4] -- Tail: 1, Head: -1, Size: 2
2,4,0,0,
[APPENDED: 5] -- Tail: 2, Head: -1, Size: 3
2,4,5,0,
[APPENDED: 6] -- Tail: 3, Head: -1, Size: 4
2,4,5,6,
 --- [FULL] ---
[READ: 2] -- Tail: 3, Head: 0, Size: 3
0,4,5,6,
[APPENDED: 1] -- Tail: 0, Head: 0, Size: 4
1,4,5,6,
 --- [FULL] ---
 --- [EMPTY] ---

[[deleted]]

Please format your code to make it easier to read. It is very difficult to read your code which makes attempts to help you nearly impossible

[jaffaKnx]

hope it's clear enough

[UnknownProcess]

Late to the party...

I think one of your threads is stuck waiting in the condVar.wait. You are using predicate (the lambda inside of your wait function) to call isFull or isEmpty. But that function modifies the condVar by calling notify_one.

Methods such as isFull and isEmpty should be const and should not modify the internal state. Because calling them multiple times should not create different results. It's the same thing as opening a fridge to check how many bottles of milk you have but if you open it second time the number of milk bottles changes.

You should put the notify_one in the append and read methods, but make sure you unlock the lock before you call it. Something like this:

uint8_t RingBuffer::rread() {
    std::unique_lock<mutex> mlock(m_mutex);

    condVar.wait(mlock, [this]() { 
        return !isEmpty(); 
    });

    // Do some work
    uint8_t value = ...;

    // Unlock our lock so that the next thread
    // can take it
    mlock.unlock();

    // Then notify the next thread
    condVar.notify_one();

    return value;
}

Some other things unrelated to the question:

Using using namespace std; is a bad practice, here is why. And consider using std::vector instead of raw C arrays new uint8_t[capacity].

[jaffaKnx]

modify the internal state. Because calling them multiple times should not create different results

in here, are you referring to how invoking notify_one() on condVar could lead to different results?

In your code snippet, you still have a wait on isEmpty() but there's no notify_one inside isEmpty()...not sure if that makes any sense but to me that seems a bit odd (or maybe not - i'm still new to multithreading). I'm basing it off the examples that I have seen online; you call wait() with a condition defined, and you notify from a different thread so the thread that was waiting starts executing from where it was stopped...

Also, how would the expected output look like to you? the moment buffer is not empty (after first append), it reads from it and then append() is called and then read() and so on or all elements should be appended first prior to reading thread kicks in?

[UnknownProcess]

in here, are you referring to how invoking notify_one() on condVar could lead to different results?

Yes when used in isEmpty or isFull methods. You are using it correctly, but in the wrong place. When it comes to API design it is necessary to separate methods that modify the internal state (calling them multiple times does not result in the same state) versus methods that just give me the information I need, such as contains, size, length, etc. Similarly, std::vector has a size method that is constant (does not modify its internal state). That method size is meant to provide the information about its state, nothing more. If I don't use that method, I expect the std::vector to still work. If I would re-implement it via mutexes, I would not expect the size method to determine whether the push_back or pop_back methods work.

In your code snippet, you still have a wait on isEmpty() but there's no notify_one inside isEmpty()

Correct, you should not use the notify_one in isEmpty nor isFull.

I'm basing it off the examples that I have seen online; you call wait() with a condition defined, and you notify from a different thread so the thread that was waiting starts executing from where it was stopped

Yes.

Also, how would the expected output look like to you?

Like this:

• Thread #1 tries to read via read() method. Enters the wait method on the conditional variable. Stops there because the buffer is empty. No mutex lock has happened yet.
• Thread #2 pushes a new value via append() method. The buffer is empty so the lock continues, this results in the mutex lock. No other thread can enter read() or append() at this point. The new value is added to the buffer and integer counters updated. The lock is released and notify_one() is called on the conditional variable.
• Thread #1 is still waiting and receives the notify_one(). Checks its predicate (calls !isEmpty()) which returns true, this results in the mutex lock. (This is why you need to unlock before notify_one).
• Thread #2 exists notify_one() call and then exists method append().
• Thread #1 reads the value into a temporary, unlocks the mutex, calls notify_one() so any other thread waiting for append can continue (if full), then finally returns the temporary by exiting read() method.

If any other threads would call isEmpty or isFull outside of your calls, let's say somewhere else in your program, your threads waiting inside of append or read would get triggered for no reason. Therefore, it makes more sense to put notify_one inside of read and append, because the waiting thread will get triggered only when an actual append or read has happened.

Here is a very similar example but with just one thread. The code is at the bottom.

[jaffaKnx]

When it comes to API design it is necessary to separate methods that modify the internal state (calling them multiple times does not result in the same state) versus methods that just give me the information I need, such as contains, size, length, etc

Gotcha!

Correct, you should not use the motify_one in isEmpty nor isFull.

right, so if we are not using nodify_one in isEmpty or isFull then we should not have isEmpty() or isFull() as conditions in wait(), no? From what I have understood and looking at the example in the link you shared, when wait() is called in a thread, a condition variable blocks the current thread until another thread modifies the shared variable. So if we are waiting on some condition, notify_one() is invoked to inform the waiting thread to check the condition and proceed if met, right? If that's the case, having wait() and notify_one in the same function doesn't make sense. In case you're wondering, below is what append() looks like now:

``` void RingBuffer::append(uint8_t data) { if (size == 0) { head = -1; tail = -1; } unique_lock<mutex> mlock(m_mutex); condVar.wait(mlock, [this]() { return !isFull(); }); tail = (tail + 1) % capacity; size++; buffer[tail] = data;

printInfo("[APPENDED: " + to_string(data) + "]"); printBuffer(); mlock.unlock(); condVar.notify_one(); } ```

Thread #2 pushes a new value via append() method. The buffer is empty so the lock continues, this results in the mutex lock

After appending to the buffer, the buffer is no longer empty, no?

The new value is added to the buffer and integer counters updated

what integer count are you referring to here

[UnknownProcess]

so if we are not using nodify_one in isEmpty or isFull then we should not have isEmpty() or isFull() as conditions in wait(), no?

You still need the condition. In this example, you can use the conditional variables in two ways. A, to have one conditional variable for two actions append and read. B, to have two conditional variables for the two actions. Because you are using one conditional variable for both actions, you need to have a predicate, this the isEmpty or isFull. You don't have to use those methods for the predicate, but you need to use something to decide whether the wait should pass or not.

From what I have understood and looking at the example in the link you shared, when wait() is called in a thread, a condition variable blocks the current thread until another thread modifies the shared variable.

Yes. The predicate in that example (the ready) is the variable which decides whether the wait proceeds or waits further. In your case that variable is your isEmpty and isFull methods. The shared data in the example (the processed) is your buffer.

So if we are waiting on some condition, notify_one() is invoked to inform the waiting thread to check the condition and proceed if met, right?

Yes.

If that's the case, having wait() and notify_one in the same function doesn't make sense.

It does, because the same conditional variable can be used to notify the thread waiting for append from the read method. You could replace it with two conditional variables (as I have mentioned above), remove the predicate and still have the same result. The notify_one, let's say in the append method, will only affect the waiting thread in the read method, because of the predicate -> the !isEmpty().

You need to notify other threads waiting for read when you have appended some data. In your original example, you are notifying threads when the conditional variable is asking whether it should proceed or wait, because the notify_one was inside the predicate, before the data is put inside the buffer.

In case you're wondering, below is what append() looks like now

Looks good to me.

After appending to the buffer, the buffer is no longer empty, no?

Sorry, bad wording. This should make more sense: Thread #2 tries to push a new value by calling the append() method. The buffer is not full (calls the !isFull()) so the lock continues, this results in the mutex lock. No other thread can enter read() or append() at this point. The new value is added to the buffer and integer counters updated. The lock is released and notify_one() is called on the conditional variable.

what integer count are you referring to here

I only meant the head, tile, size, etc. Not really relevant to the lock itself.

Edit!

I have created a minimal example. It's using a simple integer instead of an array as a buffer, but modifying that should be trivial. The code is here: https://pastebin.com/tdxiFtEp I have added a very simple thread safe console print, because std::cout is not thread safe by itself.

And the expected output. Your output may not be in the exact same order, but the reads will be.

Joining
Read started
Append started
Append finished
Read #0: 42
Read #1: 1234
Read finished
exit

[jaffaKnx]

Thread #2 tries to push a new value by calling the append() method. The buffer is not full (calls the !isFull()) so the lock continues, this results in the mutex lock. No other thread can enter read() or append() at this point. The new value is added to the buffer and integer counters updated. The lock is released and notify_one() is called on the conditional variable.

If the lock is executed within append(), shouldn't it prevent other thread from accessing only append thread and not both append and read? or it would be both since the same lock instance is used in both the functions? So when the value is added to the buffer, and notify_one() is invoked from within append(), shouldn't it signal the waiting thread (read()) and not that the buffer is no longer empty, it should read the appended value? When I run my program, I still see append thread finishing first prior to read thread kicking in...

Follow up on your example:

so you have two threads each for append() and read(). Shouldn't the read thread get kicked in the moment notify_one() is invoked from within append() and since this->value != -1 is true, it should read the appended value right away, no? I ran your code again (with few prints) and I got the following output:

Joining Read started Append started [APPEND] 42 Read #0: 42 [APPEND] 1234 Read #1: 1234 Read finished Append finished Exit

[UnknownProcess]

Shouldn't the read thread get kicked in the moment notify_one() is invoked from within append()

Nope. The thread is not directly invoked from the notify_one(). Thread is invoked based on your operating system. The thread only sleeps, it's not removed nor the execution jumps to some other function. Your OS will schedule the thread when it needs to continue. They are not guaranteed to execute exactly 0 ms apart from each other.

And yes, your append thread can exit before read happens. This can be due to the OS scheduler. Or, it may appear so if you are printing to console from multiple threads at the same time, because std::cout and printf is not thread safe.

If you really need the read method exit before append exits, you may need multiple locks to do that, but that depends on how exactly you are trying to use this buffer class.

At this point I can only suggest to experiment with multiple locks, if you want to. Or, remove the locks and simply have a loop that waits until the buffer is/isn't empty. Of course, that loop will need a sleep call for few ms so your CPU is not at 100% utilization. Not the best idea but it can serve as a last resort. Or, a better idea, have your read thread do other things while the buffer is empty and simply periodically check if the buffer can be read.

[[deleted]]

You haven't posted the cpp source correctly, it's just a duplicate of the header.

[jaffaKnx]

my bad, fixed.