This is for a single producer and single consumer wait free ring buffer. The writes need to be wait free for sure. It pre-allocates messages slots and uses a claim strategy to capture a buffer for write or read. For writes, it uses a ping pong strategy if the buffer is full which means if it find a read happening in the slot it was about the claim, it will go back to last slot to overwrite. If the read claims that by that time, then write will move forward again and eventually succeed. Read uses a SNZI to check if there are messages in the buffer to read.
Do let me know if there are any issues with the buffer? (Correctness, concurrency hazards, performance or anything I can do to make it better). This is my firtst attempt to write a wait free data structure so any help will be appreciated.
#include <atomic>
#include <iostream>
#include <thread>
#include <chrono>
#define CIRCULAR_BUFFER_SIZE 0xA00000 // 10 MB
#define ACQMSG_BUF_SIZE 5120 // Largest Message is 4688 bytes.
#define CACHE_LINE_SiZE 64 // 64 bytes cache line size for x86-64 processors
#define NUMBER_OF_SLOTS 2190 //2190 = 10 MB (Buffer Size) / 4.688 KB (Size of each buffer)
#define MAX_CLAIM_ATTEMPTS 500000
// Not really a SNZI but still... Can this be a racey SNZI and not use atomics?
//Then there will be conservation of reads and writes.
// This will be used to check if there are items to read in the buffer. Don't need exact count, just an estimate.
class BinarySNZI
{
public:
BinarySNZI()
{
write_count.store(0);
read_count.store(0);
}
void inc()
{
write_count++;
}
void dec()
{
read_count++;
}
long count()
{
return write_count.load(std::memory_order_relaxed) - read_count.load(std::memory_order_relaxed);
}
private:
// Separate into individual cache line.
__declspec(align(CACHE_LINE_SiZE)) std::atomic_long write_count;
char PAD1[CACHE_LINE_SiZE - sizeof(std::atomic_long)];
std::atomic_long read_count;
};
typedef struct MessageBlock{
std::atomic_bool claimed;
bool fresh;
int bytesToRead;
char* message;
MessageBlock()
{
claimed.store(false);
fresh = false;
message = new char[ACQMSG_BUF_SIZE];
}
// Need padding for 64 byte size end to avoid false sharing. But watch out the size if you add more members of change members.
char PAD[CACHE_LINE_SiZE - sizeof(std::atomic_bool) * 2 - sizeof(int)-sizeof(char*) - 4];
} MB;
class WaitFreeBuffer
{
public:
WaitFreeBuffer(size_t sizeOfBuffer = ACQMSG_BUF_SIZE){
current_read_index = 0;
current_write_index = 0;
};
~WaitFreeBuffer(){
for (int i = 0; i < NUMBER_OF_SLOTS; ++i)
{
delete[] message_buffer[i].message;
}
};
bool Write(int sizeToWrite, char* buffer){
int attempts = 0;
bool isAlreadyClaimed = false;
bool hasSteppedBack = false;
bool hasPingPonged = false;
// You don't want to keep CAS looping as this creates a lot of cache line traffic. A better way is, I don't need a
// CAS here. I can only use a load. And do a store when the loop breaks
while (message_buffer[current_write_index].claimed.load())
{
isAlreadyClaimed = false;
if (attempts >= MAX_CLAIM_ATTEMPTS/2 && ! hasSteppedBack)
{
hasSteppedBack = true;
attempts = 0;
current_write_index = current_write_index - 1;
if (current_write_index == -1)
current_write_index = NUMBER_OF_SLOTS - 1;
}
if (attempts >= MAX_CLAIM_ATTEMPTS/2 && hasSteppedBack)
{
attempts = 0;
hasPingPonged = true;
current_write_index++ ;
current_write_index = current_write_index % NUMBER_OF_SLOTS;
}
if (attempts >= MAX_CLAIM_ATTEMPTS / 2 && hasPingPonged)
{
return false;
}
attempts++;
}
message_buffer[current_write_index].claimed.store(true, std::memory_order_release);
message_buffer[current_write_index].bytesToRead = sizeToWrite;
::memcpy(message_buffer[current_write_index].message, buffer, sizeToWrite);
message_buffer[current_write_index].fresh = true;
message_buffer[current_write_index].claimed.store(false, std::memory_order_release);
current_write_index++;
current_write_index = current_write_index % NUMBER_OF_SLOTS;
if (! hasPingPonged)
counter.inc();
return true;
};
// Update current read index
bool Read(int& sizeRead, char* buffer){
int attempts = 0;
if (!message_buffer[current_read_index].fresh)
{
// If there is nothing fresh you may want to yield the thread in the client side
return false;
}
bool isAlreadyClaimed = false;
// You don't want to keep looping as this creates a lot of cache line traffic. Need a better way
while (!message_buffer[current_read_index].claimed.compare_exchange_strong(isAlreadyClaimed, true))
{
isAlreadyClaimed = false;
if (attempts >= MAX_CLAIM_ATTEMPTS)
{
return false;
}
attempts++;
}
sizeRead = message_buffer[current_read_index].bytesToRead;
::memcpy(buffer, message_buffer[current_read_index].message, sizeRead);
message_buffer[current_read_index].fresh = false;
message_buffer[current_read_index].claimed.store(false, std::memory_order_release);
current_read_index++;
current_read_index = current_read_index % NUMBER_OF_SLOTS;
counter.dec();
return true;
};
bool HasNewItems()
{
return counter.count() > 0;
}
private:
// Do I need number of unread count?
__declspec(align(CACHE_LINE_SiZE)) MB message_buffer[NUMBER_OF_SLOTS];
__declspec(align(CACHE_LINE_SiZE)) int current_read_index;
__declspec(align(CACHE_LINE_SiZE)) int current_write_index;
__declspec(align(CACHE_LINE_SiZE)) BinarySNZI counter;
};
int _tmain(int argc, _TCHAR* argv[])
{
WaitFreeBuffer buffer;
std::thread writer([&buffer](){
bool r;
for (int i = 0; i < 2000; i++){
r = buffer.Write(sizeof(int), (char*)(&i));
if (!r)
std::cout << "Write failed for " << i << std::endl;
}
});
std::thread reader([&buffer](){
int size;
void* b = malloc(sizeof(int));
bool r;
while(buffer.HasNewItems()){
r = buffer.Read(size, (char*)b);
if (r) {
std::cout << *((int*)b) << std::endl;
}
else {
std::cout << "Read failed " << std::endl;
}
}
});
writer.join();
reader.join();
return 0;
}
