// EDIT - Based on the answers below.
#include <atomic>
#include <iostream>
#include <thread>
#include <chrono>
#include <emmintrin.h>
namespace {
#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?
// This is not right but you get the idea of SNZI. Need to bound this with the size of the buffer.
//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;
__declspec(align(CACHE_LINE_SiZE)) std::atomic_long read_count;
};
typedef struct MessageBlock {
std::atomic_bool claimed;
bool fresh;
char* message;
uint32_t bytesToRead;
MessageBlock()
{
claimed.store(false);
fresh = false;
message = new char[ACQMSG_BUF_SIZE];
}
~MessageBlock()
{
delete[] message;
}
// Need padding for 64 byte size end to avoid false sharing. But watch out the size if you add more members of change members.
// The last is for alignment members
char PAD[CACHE_LINE_SiZE - sizeof(std::atomic_bool) * 2 - sizeof(uint32_t)-sizeof(char*) - 2];
} MB;
class WaitFreeBuffer
{
public:
WaitFreeBuffer(size_t sizeOfBuffer = ACQMSG_BUF_SIZE){
current_read_index = 0;
current_write_index = 0;
};
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++;
_mm_pause(); // Do a 1 - 2 cycle pause for avoiding pipeline clogging & flush problem. Ref - https://software.intel.com/sites/default/files/m/d/4/1/d/8/17689_w_spinlock.pdf
}
auto& current = message_buffer[current_write_index];
current.claimed.store(true, std::memory_order_release);
current.bytesToRead = sizeToWrite;
::memcpy(current.message, buffer, sizeToWrite);
current.fresh = true;
current.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;
auto& current = message_buffer[current_read_index];
if (!current.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.
bool current_claimed = false;
do {
isAlreadyClaimed = false;
while (current.claimed.load(std::memory_order_relaxed))
{
_mm_pause(); // Do a 1 - 2 cycle pause for avoiding pipeline clogging & flush problem. Ref - https://software.intel.com/sites/default/files/m/d/4/1/d/8/17689_w_spinlock.pdf
attempts++;
if (attempts >= MAX_CLAIM_ATTEMPTS)
{
return false; // Failed to claim a slot for read
}
}
bool success = current.claimed.compare_exchange_strong(isAlreadyClaimed, true);
if (success)
{
current_claimed = true;
}
} while (!current_claimed);
sizeRead = current.bytesToRead;
::memcpy(buffer, current.message, sizeRead);
current.fresh = false;
current.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;
};