Skip to main content
Rollback to Revision 1
Source Link
Jamal
  • 34.9k
  • 13
  • 133
  • 237

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

// 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;
};
Editing more
Source Link
sun
  • 33
  • 5
#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;
};
#include <atomic>
#include <iostream>
#include <thread>
#include <chrono>
#include <emmintrin.h>


#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;
};
#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;
};
Edit based on the answers below
Source Link
sun
  • 33
  • 5

// EDIT - Based on the answers below.

#include <atomic>
#include <iostream>
#include <thread>
#include <chrono>
#include <emmintrin.h>


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

// EDIT - Based on the answers below.

#include <atomic>
#include <iostream>
#include <thread>
#include <chrono>
#include <emmintrin.h>


#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;
};
Rollback to Revision 1
Source Link
Jamal
  • 34.9k
  • 13
  • 133
  • 237
Loading
Changes with the answers.
Source Link
sun
  • 33
  • 5
Loading
Tweeted twitter.com/StackCodeReview/status/654339541407563776
Source Link
sun
  • 33
  • 5
Loading