3
\$\begingroup\$

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;
}
\$\endgroup\$
1
\$\begingroup\$

After a first reading through I found a few things that might be problems or bad style, even a memory leak.

define bad, constexpr good!

#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(0), read_count(0) { // as much as possible should go in the init list.
    }
    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;

why not use __declspec(align(CACHE_LINE_SiZE)) here too?

    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;

Mega leak no destructor, why not use std::array message;

        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;

if you really want to do this check with

static_assert(sizeof(MessageBlock) == CACHE_LINE_SiZE, "not aligned to cache");

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)
        {

You are deleting other peoples message, they should do it them self.

            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)
            {

At some point the 2 threads will come so much out of synch that one will fail.

                return false;
            }
            attempts++;
        }

These lines are far too long in my opinion, you can reduce all their length if you say

auto& mbuf = message_buffer[current_write_index]; // a reference
mbuf.claimed.store(true, std::memory_order_release);
mbuf.bytesToRead = sizeToWrite;
::memcpy(mbuf.message, buffer, sizeToWrite);
mbuf.fresh = true;
mbuf.claimed.store(false, std::memory_order_release);



        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;

far too few iterations to test it.

        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;

Why malloc this???

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

EDIT: To make a CAS with less cacheline trafic do something like this

do {
  while (var.load(relax) != whatIWant)
    // intel inline pause for 12 cycles
    // linux nanosleep(100) supposed to sleep for 100ns 
    pause() or nanosleep(100) 
  CAS
} while (!done)
\$\endgroup\$
  • \$\begingroup\$ Thanks for your comments and for catching the leak I missed. I thought a lot about the wait-freeness of the buffer but kind of forgot about allocations and readable code :). I don't understand your comment about "At some point the 2 threads will come so much out of synch that one will fail.". Can you elaborate? \$\endgroup\$ – sun Oct 15 '15 at 14:14
  • \$\begingroup\$ I meant one will return without it having been able to read respectively write a buffer. \$\endgroup\$ – Surt Oct 15 '15 at 14:25
  • \$\begingroup\$ True. It is a lossy buffer and it can fail reads and writes. Does it make sense? \$\endgroup\$ – sun Oct 15 '15 at 14:29
  • \$\begingroup\$ Depending on your usage you could add code like I added at the end to both your read and write. Using mutex and condition vars is sure to bring about a task switch at some point costing from 1200ns to 50K ns + any duration before it gets switch in again. \$\endgroup\$ – Surt Oct 15 '15 at 14:37
  • \$\begingroup\$ Thanks. I suppose you mean the __mm_pause() intrinsics. A requirement I forgot the mention is writes need to happen as fast as they can but it can overwrite the last written slot if no slot is available. And the sequence of writes need to be mainaintained in increasing order. So it can go from 1-2-3 to 1-2-4 but not 1-4-3. \$\endgroup\$ – sun Oct 19 '15 at 4:13
0
\$\begingroup\$

Struct size calculation

In this struct definition you are trying to make the struct equal to a cache line size but your padding calculation looks a bit fragile:

typedef struct MessageBlock{
    std::atomic_bool claimed;
    bool fresh;
    int bytesToRead;
    char* 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.
    char PAD[CACHE_LINE_SiZE - sizeof(std::atomic_bool) * 2 - sizeof(int)-sizeof(char*) - 4];
} MB;

Here your padding assumes the rest of the struct is the size of two bools, one int, one pointer, and 4 extra bytes. I'm not sure where those 4 extra bytes come from but perhaps it is the alignment gap between bytesToRead and message on a 64-bit system. However, on a 32-bit system, or on a system where an int is 64 bits, those 4 extra bytes may not exist. You may want to reorganize your struct to make the padding more portable. For example, use uint32_t instead of int and move the pointer to the beginning of the struct.

\$\endgroup\$
  • \$\begingroup\$ It is alignment gap. I should have reorganized to remove the misalignment. Thank you for your comment. \$\endgroup\$ – sun Oct 14 '15 at 6:14
0
\$\begingroup\$

EDIT - Based on the comments above.

#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;
};
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.