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];
}

Welcome to Code Review! I see a Foo in your code. Who's that and where did he come from? For future reference, please take a look at the How to get the best value out of Code Review - Asking Questions. — Mast
– Mast ♦, Commented Aug 16, 2018 at 5:57
Is it intended that each consumer individually has to visit every item pushed into the ringbuffer? Elements already processed by other consumers don't seem to get skipped. — hoffmale
– hoffmale, Commented Aug 16, 2018 at 12:47

1201ProgramAlarm · Accepted Answer · 2018-08-16 03:47:18Z

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)

"mMaxReadIdx to indicate a slot is available for reading when it isn't". Thanks for pointing this out! :) I will have to think of a way to prevent this.. If anyone have some ideas, please share! — interceptwind
– interceptwind, Commented Aug 16, 2018 at 4:15
Also thanks for pointing out the unnecessary reference in TryRead and MAX_SIZE as macro. — interceptwind
– interceptwind, Commented Aug 16, 2018 at 4:21
I am not too worried about the limits not being enforced cause I am certain they won't be exceeded for my use-case. — interceptwind
– interceptwind, Commented Aug 16, 2018 at 4:26

L.Koh · Accepted Answer · 2018-08-30 02:32:35Z

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];
}

That's basically a spin lock, so not exactly lock-free. I don't think that's intended by the OP. A truly lock-free solution could add a std::atomic<bool> per element in the ringbuffer which indicates whether said element is ready to be read. // Also, the acquisition of the spin lock isn't correct. Multiple threads might read mIsPushing as false before some thread stores true in it again. A better approach would use compare_exchange_strong. — hoffmale
– hoffmale, Commented Aug 24, 2018 at 5:02
@hoffmale do you mean using compare_exchange_strong instead of using store? — L.Koh
– L.Koh, Commented Aug 24, 2018 at 7:11
I was originally thinking of replacing while(mIsPushing.load(std::memory_order_seq_cst)){}; mIsPushing.store(true,std::memory_order_seq_cst); with auto expected = false; while(!mIsPushing.compare_exchange_strong(expected, true)) expected = false;. Thinking about it some more, I realized that this could be simplified to while(mIsPushing.exchange(true));. // The problem in the original snippet is the gap between checking mIsPushing and setting it. A thread might read false, but before it is able to set mIsPushing to true, some other thread might also have read false. — hoffmale
– hoffmale, Commented Aug 24, 2018 at 8:32
Ah yes what you said is similar to the example here as well: en.cppreference.com/w/cpp/atomic/atomic_exchange. Thanks for the explanation! — L.Koh
– L.Koh, Commented Aug 24, 2018 at 9:04

Stack Exchange Network

Customized lock-free RingBuffer

2 Answers 2

You must log in to answer this question.

Hot Network Questions

Customized lock-free RingBuffer

2 Answers 2

You must log in to answer this question.

Related

Hot Network Questions