Single consumer and single producer lock-free circular buffer

Question

I'm pretty new to multithreading. The code below is a simple implementation of a single consumer/single producer circular buffer that does not use any locking, that I wrote for fun.

The question is, will this kind of container work in actual real world situation or would it crash miserably because of some data race that I have not noticed?

#include <array>
#include <memory>
#include <atomic>
#include <stdexcept>
#include <iostream>
#include <thread>
#include <random>


template<typename T, int S=5>
class CircularBuffer{
public:
  CircularBuffer():start(0),end(0){
  }
  bool AddElement(std::unique_ptr<T> && obj,bool block=true){

    int end_idx=end.load();
    if(block)
      while(end_idx+1==start.load());//Busy wait until there is place
    else if(end_idx+1==start.load())
      return false;

    buffer[end_idx++]=std::move(obj);
    if(end_idx==buffer.size())
      end.store(0);
    else
      end.store(end_idx); 
    return true;
  }
  std::unique_ptr<T> GetElement(bool block=true){
      int start_idx=start.load();
      if(block)
        while(end.load()==start_idx);
      else if(end.load()==start_idx)
        return std::unique_ptr<T>();

      auto ret_obj=std::move(buffer[start_idx]);
      start_idx++;
      if(start_idx==buffer.size())
    start.store(0);
      else
    start.store(start_idx);
      return ret_obj;      
  }

private:  
  std::array<std::unique_ptr<T>,S+1> buffer;
  std::atomic<int> start;
  std::atomic<int> end;

};

std::random_device rd;
std::mt19937 gen(rd());
std::uniform_int_distribution<> dis(0,500);

void Producer(CircularBuffer<int> * buffer){

  int i=0;
  while(i<20){
    std::this_thread::sleep_for(std::chrono::milliseconds(dis(gen)));
    auto val=dis(gen);
    std::cout << "Adding " << val <<std::endl;
    buffer->AddElement(std::unique_ptr<int>(new int(val)));
    i++;
  }
}
void Consumer(CircularBuffer<int> * buffer){

  int i=0;
  while(i<20){
    std::this_thread::sleep_for(std::chrono::milliseconds(dis(gen)));
    std::cout << "Got " << *buffer->GetElement() << std::endl;
    i++;
  }

}

int main(){

  CircularBuffer<int> buffer;

  std::thread t[2];
  t[0]=std::thread(Producer,&buffer);
  t[1]=std::thread(Consumer,&buffer);

  t[0].join();
  t[1].join();
}

Answer 1

A few notes, mostly unrelated to the actual lock-free nature of the code:

1. Some of your names are a lot less meaningful than I'd like. A prime example is your template parameter S in: template<typename T, int S=5>. This should probably be renamed to something like Size instead. Although not quite as problematic, I'd say your dis and gen have pretty much the same problem as well. (Note that although T as the type parameter is just as short, it's so common I'd accept it, much like I'd accept i as the index for a loop).

2. Although some people avoid them, I'd at least consider using something like size_t for the types of CircularBuffer::begin and CircularBuffer::end. It appears that neither should ever be negative.

3. You might want to consider using std::atomic_int (or std::atomic_uint) instead of std::atomic<int>. It's unlikely to make much difference, but does give the library design a little extra chance for optimization (i.e., atomic_int might just be a typedef for atomic<int>, but could be a separate type instead, if the library designer sees an advantage in doing so).

4. It may just be a side-effect of some strange tab expansion (or something similar) but at least as posted, your indentation is pretty strange in a few places. For one obvious example:

     if(start_idx==buffer.size())
   start.store(0);
     else
   start.store(start_idx);
   

5. I'd have to do some looking to be absolutely certain, but I believe your use of a global for the random number generator causes a data race that leads to undefined behavior. Specifically, a pseudo-random number generator has an internal state that's modified when you obtain a random number from it. Doing that modification simultaneously from two or more threads gives undefined behavior.

6. This sequence:

     start_idx++;
     if(start_idx==buffer.size())
   start.store(0);
     else
   start.store(start_idx);
   

Looks to me like it's subject to race conditions. For example, let's assume start_idx starts out equal to buffer_size-1, then we execute the code above from two threads:

thread A: start_idx++;  
thread B: start_idx++;

thread A: if (start_idx == buffer.size())
thread B: if (start_idx == buffer.size())

Both if statements return false--start_idx is now one greater than buffer.size(). We then store start_idx into start.

The next time the function is called, it attempts to use:

auto ret_obj=std::move(buffer[start_idx]);

...and start_idx indexes past the end of buffer, giving undefined behavior.

Limiting to single-producer, single-consumer may be enough to prevent this problem from manifesting, but 1) I find that an essentially useless model, and 2) I'm not at all sure it's sufficient even then.

Although it may sound defeatist, let me give one little bit of advice: when you first do multithreaded programming, you're a lot better off sticking to using locks, mutexes, and so on. Getting just that much right will keep you busy for quite a while.

(Only) once you've gotten to the point that it's easy to handle that, then start to move on to working at lock-free programming (but be aware that it's an area where even the very best, brightest and most experienced still get things wrong quite frequently.