Customized lock-free RingBuffer

Question

I have implemented my own lock-free ringbuffer for multiple producers multiple consumers using vector. Can you guys help me review to see if there are any problems and ways to improve?

Explanation: Each producer runs on its own thread and will use Push() to enter its item into the RingBuffer.

Each consumer runs on its own thread and has its own unique number. When consumer TryRead(), it will enter its unique number.

RingBuffer will keep track of the last position each consumer read in std::vector<long> mReadIdxVec

If mReadIdxVec[consumer unique number] >= mMaxReadIdx, TryRead() will return a nullptr.

Question: Within Push() method, will it be a problem if multiple producers on different threads call this line mVector[index] = pMsg; at the same time?

Can this code be improved further? Any comments are welcomed!

---

RingBuffer.h

#ifndef __RING_BUFFER_H_
#define __RING_BUFFER_H_

#include <atomic>

class RingBuffer
{
    private:
        std::vector<Foo *> mVector;
        std::atomic<long> mWriteIdx{0}; //mWriteIdx will not exceed 2,000,000,000
        std::atomic<long> mMaxReadIdx{0}; //mMaxReadIdx will not exceed 2,000,000,000
        std::vector<long> mReadIdxVec;

    public:
        RingBuffer(int pNumOfConsumers);
        void Push(Foo *);
        Foo * TryRead(const int&);
};

#endif

---

RingBuffer.cc

#include "RingBuffer.h"
#define MAX_SIZE 1000

RingBuffer::RingBuffer(int pNumOfConsumers) 
{
    mVector.reserve(MAX_SIZE);
    mVector.assign(MAX_SIZE, nullptr);
    mReadIdxVec.assign(pNumOfConsumers, 0);
}

void RingBuffer::Push(Foo * pMsg)
{
    if(!pMsg) return;
    long writeIdx = mWriteIdx++;
    long index = writeIdx % MAX_SIZE;
    if(writeIdx >= MAX_SIZE) delete mVector[index];
    mVector[index] = pMsg;
    mMaxReadIdx++;
}

Foo * RingBuffer::TryRead(const int& pConsumer) 
{
    if(mReadIdxVec[pConsumer] >= mMaxReadIdx) return nullptr;
    long index =  mReadIdxVec[pConsumer] % MAX_SIZE;
    mReadIdxVec[pConsumer]++;
    return mVector[index];
}

Answer 1

You have comments in the definition of RingBuffer on the value limits for some members, but there is nothing in the code that enforces that or checks to see if those limits are exceeded.

Why does TryRead take its parameter by const int&? The reference is unnecessary. Just pass in the int.

Why is MAX_SIZE a macro? It should be a constexpr (if your compiler supports it) or static const int if not.

In the RingBuffer constructor, you can combine the reserve and assign calls into one, and include it in the member initializer list.

Push will happily and silently overwrite things that have been added to your ring buffer but not consumed.

The way your Push is set up, you can have multiple threads write to the same memory location (if, for instance, there are enough other calls to Push so that the value calculated for index wraps back around).

It is also possible for mMaxReadIdx to indicate a slot is available for reading when it isn't. For example, thread A starts a Push, and gets a write index. Thread B then starts another Push, gets a write index, stores its value and increments mMaxReaddx before thread A has stored its message. The slot used by thread A will be readable by a TryRead but won't have data in it yet.

TryRead can return the same message to multiple consumers, since each consumer has its own idea of where the bottom of the buffer is.

RingBuffer::RingBuffer(int pNumOfConsumers): mVector(MAX_SIZE, nullptr)

Answer 2

To resolve the problem of mMaxReadIdx being incremented by Thread B before Thread A has stored its message, could we consider the use of an atomic boolean as below?

---

Ringbuffer.h

#ifndef __RING_BUFFER_H_
#define __RING_BUFFER_H_

#include <atomic>

class RingBuffer
{
    private:
      std::vector<Foo> mVector;
      std::atomic<long> mWriteIdx{0}; //mWriteIdx will not exceed 2,000,000,000
      std::atomic<long> mMaxReadIdx{0}; //mMaxReadIdx will not exceed 2,000,000,000
      std::vector<long> mReadIdxVec;
      std::atomic<bool> mIsPushing{false};

    public:
      RingBuffer(int pNumOfConsumers);
      void Push(Foo);
      Foo * TryRead(int);
};

#endif

---

#include "RingBuffer.h"

static constexpr int const& MAX_SIZE = 5000;

RingBuffer::RingBuffer(int pNumOfConsumers) 
{
    for(int i=0; i < MAX_SIZE; i++)
    {
        Foo f;
        mVector.push_back(f);
    }
    mReadIdxVec.assign(pNumOfConsumers, 0);
}

void RingBuffer::Push(Foo pMsg)
{
    long writeIdx = mWriteIdx++;
    long index = writeIdx % MAX_SIZE;
    while(mIsPushing.load(std::memory_order_seq_cst)){};
    mIsPushing.store(true,std::memory_order_seq_cst);
    if(writeIdx >= MAX_SIZE) delete mVector[index];
    mVector[index] = pMsg;
    mMaxReadIdx++;
    mIsPushing.store(false,std::memory_order_seq_cst);
}

Foo * RingBuffer::TryRead(const int& pConsumer) 
{
    if(mReadIdxVec[pConsumer] >= mMaxReadIdx) return nullptr;
    long index =  mReadIdxVec[pConsumer] % MAX_SIZE;
    mReadIdxVec[pConsumer]++;
    return mVector[index];
}

---

Ringbuffer.h (Lock-free)

#ifndef __RING_BUFFER_H_
#define __RING_BUFFER_H_

#include <atomic>

class Foo
{
    public:
        long bar = 0;
        double bar2 = 0;
        std::atomic<bool> isSetting{false};

        void set(long pBar, double pBar2)
        {
            isSetting = true;
            bar = pBar;
            bar2 = pBar2;
            isSetting = false;
        }
        Foo(){};
        Foo(const Foo &m2) //copy constructor required as they are default-deleted for atomics
        {
            set(m2.bar, m2.bar2);
        }
};

class RingBuffer
{
    private:
      std::vector<Foo> mVector;
      std::atomic<long> mWriteIdx{0}; //mWriteIdx will not exceed 2,000,000,000
      std::atomic<long> mMaxReadIdx{0}; //mMaxReadIdx will not exceed 2,000,000,000
      std::vector<long> mReadIdxVec;

    public:
      RingBuffer(int pNumOfConsumers);
      void Push(Foo);
      Foo * TryRead(int);
};

#endif

---

#include "RingBuffer.h"

static constexpr int const& MAX_SIZE = 5000;

RingBuffer::RingBuffer(int pNumOfConsumers) 
{
    for(int i=0; i < MAX_SIZE; i++)
    {
        Foo f;
        mVector.push_back(f);
    }
    mReadIdxVec.assign(pNumOfConsumers, 0);
}

void RingBuffer::Push(Foo pMsg)
{
    long writeIdx = mWriteIdx++;
    long index = writeIdx % MAX_SIZE;
    mVector[index].set(pMsg.bar, pMsg.bar2);
}

Foo * RingBuffer::TryRead(const int& pConsumer) 
{
   if(mReadIdxVec[pConsumer] >= mWriteIdx) return nullptr;
   if(mVector[index].isSetting) return nullptr;
   long index = mReadIdxVec[pConsumer] % MAX_SIZE;
   mReadIdxVec[pConsumer]++;
   return &mVector[index];
}