Lock-free multi-producer / multi-consumer queue in C++

Question

I've been working on a lockless multi-producer, multi-consumer queue in an effort to learn as much as I can about concurrency, without the use of mutual exclusion. The queue uses a bounded ring buffer to store/load the data from. Two cursors are used two track the next available index on the buffer, one for producers and one for consumers. The cursors are a custom sequential container type, which uses memory barriers to order access to some data. Inspiration is taken from the LMAX Disruptor here. Once a producer or consumer has successfully incremented the cursor, it can then safely do its work on the "claimed" index, knowing any following producer or consumer will be accessing a higher index. The queue size is specified by the user, but ceiled up to the nearest power of two, so a bitwise-and can be used in place of modulo to automatically wrap the index back to zero.. The queue works as it is right now, I've got two different version of the enqueu/dequeue functions, with one using a CAS loop to increment the cursor, and the other using a fetch_add.

The main issue I'm having is that the fetch_add version of enqueue is about 10x faster than the CAS version, which I don't really understand. Then second, I'm getting quite a bit of false sharing reported in Intel VTune Profiler, even though I've padded everything.

Anyway, here's the code:

The Enqueue/Dequeue & EnqueueCAS/DequeueCAS functions are inside the TMPMCQueue class at the bottom.

// SPDX-License-Identifier: GPL-2.0-or-later
/** Lockless Multi-Producer Multi-Consumer Queue Type.
 * Author: Primrose Taylor
 */

#ifndef MPMCQUEUE_H
#define MPMCQUEUE_H

#include "stdio.h"
#include "stdlib.h"

#include <atomic>

#define PLATFORM_CACHE_LINE_SIZE 64

/**
 * A container which can ensure that access to it's data will be sequentially consistent across all accessing threads,
 * but allows for getting the data via a custom memory order.
 */
template <typename T>
class TSequentialContainer
{
public:
    TSequentialContainer()
    {
        static_assert(
            std::is_copy_constructible_v<T>     ||
            std::is_copy_assignable_v<T>        ||
            std::is_move_assignable_v<T>        ||
            std::is_move_constructible_v<T>,
            "Can't use non-copyable, non-assignable, non-movable, or non-constructible type!"
        );
    }

    explicit TSequentialContainer(const T& InitialValue)
    {
        TSequentialContainer();
        Data.store(InitialValue, std::memory_order_seq_cst);
    }
    
    /**
     * Get the data, using an acquire fence to ensure that any prior write is visible to this load.
     */
    T Get() const
    {
        const T OutCopy = Data.load(std::memory_order_relaxed);
        std::atomic_thread_fence(std::memory_order_acquire);
        return OutCopy;
    }

    /**
     * Load the data with relaxed semantics. NOTE: NOT THREAD SAFE!
     */
    T GetRelaxed() const
    {
        return Data.load(std::memory_order_relaxed);
    }
    
    T GetCustom(const std::memory_order MemoryOrder) const
    {
        return Data.load(MemoryOrder);
    }
    
    /**
     * Set the data, first performing a release fence.
     * The release fence will ensure that any subsequent read will see this write.
     */
    void Set(const T& NewData) 
    {
        std::atomic_thread_fence(std::memory_order_release);
        Data.store(NewData, std::memory_order_relaxed);
    }
    
    /**
     * Set the data by first performing a release fence, then storing the data,
     * then performing a full fence.
     */
    void SetFullFence(const T& NewData)
    {
        std::atomic_thread_fence(std::memory_order_release);
        Data.store(NewData, std::memory_order_relaxed);
        std::atomic_thread_fence(std::memory_order_seq_cst);
    }
    
    void SetCustom(const T& NewData, const std::memory_order MemoryOrder)
    {
        Data.store(NewData, MemoryOrder);
    }
    
    /**
     * Perform a CAS operation on the stored data.
     * Uses release semantics if works.
     * Uses relaxed semantics if failed.
     */
    bool CompareAndSet(T& Expected, const T& NewValue)
    {
        return Data.compare_exchange_weak(Expected, NewValue,
            std::memory_order_release, std::memory_order_relaxed);
    }
    
protected:
    uint_fast8_t PadToAvoidContention0[PLATFORM_CACHE_LINE_SIZE] = { };
    /**
     * An atomic variable which holds the data.
     */
    std::atomic<T> Data;
    uint_fast8_t PadToAvoidContention1[PLATFORM_CACHE_LINE_SIZE] = { };

private:
    TSequentialContainer(const TSequentialContainer&) = delete;
    TSequentialContainer& operator=(const TSequentialContainer&) = delete;
};

/**
 * A simple child class of the @link TSequentialContainer which uses an int64 instead of a template.
 * Providing some extra functions specific to modifying an integer.
 */
class FSequentialInteger : public TSequentialContainer<int_fast64_t>
{
public:
    FSequentialInteger(const int_fast64_t InitialValue = 0)
        : TSequentialContainer()
    {
        SetFullFence(InitialValue);
    }
    
    /**
     * Uses a fetch_add with Acquire/Release semantics to increment the integer.
     *
     * @return Returns the original value of the integer.
     */
    int_fast64_t AddAndGetOldValue(const int_fast64_t Value)
    {
        return Data.fetch_add(Value, std::memory_order_acq_rel);
    }
    
    /**
     * @link AddAndGetOldValue()
     */
    int_fast64_t AddAndGetNewValue(const int_fast64_t Value)
    {
        return AddAndGetOldValue(Value) + Value;
    }

    /**
     * @link AddAndGetNewValue()
     * @link AddAndGetOldValue()
     */
    int_fast64_t IncrementAndGetOldValue()
    {
        return AddAndGetOldValue(1);
    }

    /**
     * @link IncrementAndGetOldValue()
     */
    void Increment()
    {
        IncrementAndGetOldValue();
    }

    void IncrementRelaxed()
    {
        Data.fetch_add(1, std::memory_order_relaxed);
    }

    void operator=(const int_fast64_t NewValue)
    {
        SetFullFence(NewValue);
    }
};

/**
 * Enum used to represent each status output from the Enqueue/Dequeue functions inside @link TMPMCQueue
 */
enum class EMPMCQueueErrorStatus : uint_fast8_t
{
    TRANSACTION_SUCCESS,
    BUFFER_FULL,
    BUFFER_EMPTY,
    BUFFER_NOT_INITIALIZED,
    COPY_FAILED,
    COPY_SUCCESS,
    BUFFER_COPY_FAILED,
    BUFFER_COPY_SUCCESS
};

/**
 * A Lockless Multi-Producer, Multi-Consumer Queue that uses
 * a bounded ring buffer to store the data.
 * @link TSequentialContainer A sequential container of type T, which uses memory barriers to sync access to it's data.
 * @link FSequentialInteger A sequential integer container, which uses memory barriers to sync access to it's data.
 * @link EMPMCQueueErrorStatus Enum used to represent each status output from the Enqueue/Dequeue functions.
 *
 * @template T The type to use for the queue.
 * @template TQueueSize The size you want the queue to be. This will be rounded UP to the nearest power of two.
 *
 * @biref A Lockless Multi-Producer, Multi-Consumer Queue.
 */
template <typename T, uint_fast64_t TQueueSize>
class TMPMCQueue final
{
private:
    using FElementType = T;
    using FCursor = FSequentialInteger;

public:
    TMPMCQueue()
    {
        if(TQueueSize == 0 || TQueueSize == UINT64_MAX)
        {
            return;
        }

        /**
         * Ceil the queue size to the nearest power of 2
         * @cite https://graphics.stanford.edu/~seander/bithacks.html#RoundUpPowerOf2
         */
        uint_fast64_t NearestPower = TQueueSize;
        {
            NearestPower--;
            NearestPower |= NearestPower >> 1; // 2 bit
            NearestPower |= NearestPower >> 2; // 4 bit
            NearestPower |= NearestPower >> 4; // 8 bit
            NearestPower |= NearestPower >> 8; // 16 bit
            NearestPower |= NearestPower >> 16; // 32 bit
            NearestPower |= NearestPower >> 32; // 64 bit
            NearestPower++;
        }

        IndexMask.store(NearestPower - 1); // Set the IndexMask to be one less than the NearestPower

        /** Allocate the ring buffer. */
        RingBuffer = (FBufferNode*)calloc(NearestPower, sizeof(FBufferNode));
        for(uint_fast64_t i = 0; i < NearestPower; ++i)
        {
            RingBuffer[i].Data = (FElementType*)malloc(sizeof(FElementType));
        }
        
        ConsumerCursor.SetFullFence(0);
        ProducerCursor.SetFullFence(0);
    }

    ~TMPMCQueue()
    {
        if(RingBuffer == nullptr)
            return;
        
        free(RingBuffer);
        RingBuffer = nullptr;
    }

    /**
     * Add a new element to the queue.
     *
     * @link Dequeue()
     * @link FSequentialInteger::Get()
     * @link TSequentialContainer::IncrementAndGetOldValue()
     * @param NewElement The new element to add to the queue.
     *
     * @return An error status, used to check if the add worked.
     */
    EMPMCQueueErrorStatus Enqueue(const FElementType& NewElement)
    {
        /** Get the Consumer & Producer cursor values, using an acquire fence */
        const int_fast64_t CurrentConsumerCursor = ConsumerCursor.Get();
        const int_fast64_t CurrentProducerCursor = ProducerCursor.Get();
        
        /** Return false if the buffer is full */
        if((CurrentProducerCursor + 1) == CurrentConsumerCursor)
        {
            return EMPMCQueueErrorStatus::BUFFER_FULL;
        }
        
        /** Perform a fetch_add with acquire_release semantics */
        const int_fast64_t ClaimedIndex = ProducerCursor.IncrementAndGetOldValue(); // fetch_add
        /** Calculate the index, avoiding the use of modulo */
        const int_fast64_t ClaimedIndexMask = ClaimedIndex & IndexMask.load(std::memory_order_relaxed);
        
        /** Update the index on the ring buffer with the new element */
        *RingBuffer[ClaimedIndexMask].Data = NewElement;
        
        return EMPMCQueueErrorStatus::TRANSACTION_SUCCESS;
    }

    EMPMCQueueErrorStatus EnqueueCAS(const FElementType& NewElement)
    {
        /** Get the Consumer & Producer cursor values, using an acquire fence */
        const int_fast64_t CurrentConsumerCursor = ConsumerCursor.Get();
        const int_fast64_t CurrentProducerCursor = ProducerCursor.Get();
        
        /** Return false if the buffer is full */
        if((CurrentProducerCursor + 1) == CurrentConsumerCursor)
        {
            return EMPMCQueueErrorStatus::BUFFER_FULL;
        }
        
        int_fast64_t ClaimedIndex = CurrentProducerCursor;
        
        while(!ProducerCursor.CompareAndSet(ClaimedIndex, ClaimedIndex + 1)) // CAS loop
        {
            ClaimedIndex = ProducerCursor.Get();
            _mm_pause();
        }
        
        /** Calculate the index, avoiding the use of modulo */
        const int_fast64_t ThisIndexMask = ClaimedIndex & IndexMask.load(std::memory_order_relaxed);
        
        /** Update the index on the ring buffer with the new element */
        *RingBuffer[ThisIndexMask].Data = NewElement;
        
        return EMPMCQueueErrorStatus::TRANSACTION_SUCCESS;
    }
    
    /**
     * Claim an element from the queue.
     *
     * @link Enqueue()
     * @link FSequentialInteger::Get()
     * @link TSequentialContainer::IncrementAndGetOldValue()
     * @param Output A reference to the variable to store the output in.
     *
     * @link EMPMCQueueErrorStatus
     * @return An error status, used to check if the add worked.
     */
    EMPMCQueueErrorStatus Dequeue(FElementType& Output)
    {
        /** Get the Consumer & Producer cursor values, using an acquire fence */
        const int_fast64_t CurrentConsumerCursor = ConsumerCursor.Get();
        const int_fast64_t CurrentProducerCursor = ProducerCursor.Get();

        // Check if the buffer is empty
        if(CurrentConsumerCursor == CurrentProducerCursor)
        {
            return EMPMCQueueErrorStatus::BUFFER_EMPTY;
        }

        /** Perform a fetch_add with acquire_release semantics */
        const int_fast64_t ClaimedIndex = ConsumerCursor.IncrementAndGetOldValue(); // fetch_add
        /** Calculate the index, avoiding the use of modulo */
        const int_fast64_t ClaimedIndexMask = ClaimedIndex & IndexMask.load(std::memory_order_relaxed);
        
        /** Store the claimed element from the ring buffer in the Output var */
        Output = *RingBuffer[ClaimedIndexMask].Data;
        
        return EMPMCQueueErrorStatus::TRANSACTION_SUCCESS;
    }

    EMPMCQueueErrorStatus DequeueCAS(FElementType& Output)
    {
        const int_fast64_t CurrentConsumerCursor = ConsumerCursor.Get();
        const int_fast64_t CurrentProducerCursor = ProducerCursor.Get();
        
        if(CurrentConsumerCursor == CurrentProducerCursor)
        {
            return EMPMCQueueErrorStatus::BUFFER_EMPTY;
        }
        
        int_fast64_t ClaimedIndex = CurrentConsumerCursor;

        while(!ConsumerCursor.CompareAndSet(ClaimedIndex, ClaimedIndex + 1)) // CAS loop
        {
            ClaimedIndex = ConsumerCursor.Get();
            _mm_pause();
        }
        
        /** Calculate the index, avoiding the use of modulo */
        const int_fast64_t ThisIndexMask = ClaimedIndex & IndexMask.load(std::memory_order_relaxed);
        
        /** Update the index on the ring buffer with the new element */
        Output = *RingBuffer[ThisIndexMask].Data;
        
        return EMPMCQueueErrorStatus::TRANSACTION_SUCCESS;
    }

private:
    struct FBufferNode
    {
        FBufferNode() noexcept
            : Data(nullptr)
        {
        }
        
        uint_fast8_t PadToAvoidContention0[PLATFORM_CACHE_LINE_SIZE] = { };
        FElementType* Data;
        uint_fast8_t PadToAvoidContention1[PLATFORM_CACHE_LINE_SIZE] = { };
    };
    
private:
    uint_fast8_t PadToAvoidContention0[PLATFORM_CACHE_LINE_SIZE] = { };
    /** Stores a value that MUST be one less than a power of two e.g 1023.
    * Used to calculate an index for access to the @link RingBuffer.
    */
    std::atomic<uint_fast64_t>                  IndexMask; 
    uint_fast8_t PadToAvoidContention1[PLATFORM_CACHE_LINE_SIZE] = { };
    /**
     * This is the pointer to the ring buffer which holds the queue's data.
     * This is allocated in the default constructor using calloc.
     */
    FBufferNode*                                RingBuffer;
    uint_fast8_t PadToAvoidContention2[PLATFORM_CACHE_LINE_SIZE] = { };
    /**
     * The cursor that holds the next available index on the ring buffer for Consumers.
     */
    FCursor                                     ConsumerCursor;
    uint_fast8_t PadToAvoidContention3[PLATFORM_CACHE_LINE_SIZE] = { };
    /**
     * The cursor that holds the next available index on the ring buffer for Producers.
     */
    FCursor                                     ProducerCursor;
    uint_fast8_t PadToAvoidContention4[PLATFORM_CACHE_LINE_SIZE] = { };

private:
    TMPMCQueue(const TMPMCQueue&) = delete;
    TMPMCQueue& operator=(const TMPMCQueue&) = delete;
};

#endif // MPMCQUEUE_H

A quick example usage, use TIMES_TO_CYCLE and THREAD_COUNT macros to try different workloads:

#include "MPMCQueue.h"

/** Define how many times to Produce/Consume an element. */
#define TIMES_TO_CYCLE  10000000
/** Total number of threads. Must be a multiple of two for this example. */
#define THREAD_COUNT    16

/** Declare a queue which stores data of type int, and has a size of 500,000.
  * The size will be ceiled up to the nearest power of two. */
static TMPMCQueue<int, 500000> MyQueue;

/** Atomic counter used to track how many threads have completed. */
static FSequentialInteger ThreadsCompleteCount;

int main()
{
    /** Create a producer & consumer each iteration. */
    for(int_fast64_t i = 0; i < (THREAD_COUNT / 2); ++i)
    {
        // Create a Producer
        std::thread([]()
        {
            const int NumberToAddLoads = 100;
            /** Execute TIMES_TO_CYCLE Enqueue operations. */
            for(int_fast64_t j = 0; j < TIMES_TO_CYCLE; ++j)
            {
                MyQueue.Enqueue(NumberToAddLoads); // fetch_add version
                // MyQueue.EnqueueCAS(NumberToAddLoads); // CAS version
            }

            /** Increment the counter to indicate that this thread has finished. */
            ThreadsCompleteCount.Increment();
        }).detach(); // Detach the thread from the main thread.
        
        // Create a Consumer
        std::thread([]()
        {
            int NumberToHoldLoads = 0;
            /** Execute TIMES_TO_CYCLE Dequeue operations. */
            for(int_fast64_t j = 0; j < TIMES_TO_CYCLE; ++j)
            {
                MyQueue.Dequeue(NumberToHoldLoads); // fetch_add version
                // MyQueue.DequeueCAS(NumberToHoldLoads); // CAS version
            }
            /** Increment the counter to indicate that this thread has finished. */ 
            ThreadsCompleteCount.Increment();
        }).detach(); // Detach the thread from the main thread.
    }

    /** Spin pause until threads are complete. */
    while(ThreadsCompleteCount.GetRelaxed() < THREAD_COUNT)
    {
        _mm_pause();
    }

    return 0;
}

fetch_add enqueue assembly:

mov rdx, qword ptr [rip+0x3fb9]
nop
mov rcx, qword ptr [rip+0x4079]
nop
inc rcx
cmp rcx, rdx
jz 0x140001317 <Block 4>
Block 3:
mov rdx, r8
lock xadd qword ptr [rip+0x4064], rdx
mov rcx, qword ptr [rip+0x3ec5]
and rcx, rdx
imul rdx, rcx, 0x88
mov rcx, qword ptr [rip+0x3efc]
mov rdx, qword ptr [rdx+rcx*1+0x40]
mov dword ptr [rdx], 0x64

And here's the much larger enqueue using a CAS loop:

mov rdx, qword ptr [rip+0x3f89]
nop
mov rcx, qword ptr [rip+0x4049]
nop
lea rax, ptr [rcx+0x1]
cmp rax, rdx
jz 0x140001362 <Block 7>                            
Block 3:                            
nop dword ptr [rax], eax
Block 4:                            
lea rdx, ptr [rcx+0x1]                          
mov rax, rcx                            
lock cmpxchg qword ptr [rip+0x4028], rdx                            
jz 0x14000133e <Block 6>                            
Block 5:                            
mov rcx, qword ptr [rip+0x401f]                         
nop                         
pause                           
jmp 0x140001320 <Block 4>                           
Block 6:                            
mov rax, qword ptr [rip+0x3e7b]
nop
and rax, rcx
imul rcx, rax, 0x88
mov rax, qword ptr [rip+0x3eb1]
mov rcx, qword ptr [rcx+rax*1+0x40]
mov dword ptr [rcx], 0x64

I'm on Windows 10 with an i9 9900k. I hope the above information is useful. I'm not used to profiling something like this, so please forgive any naiveity.

Any help would be great, cheers.

Answer 1

Your Template Constraints Could be Requirements

You could turn your static_assert check into either a requires clause on the template, or a concept. The compiler will still reject any type that does not meet the requirements, but you would be able to prove and overload, and also, you will get a better error message about which constraint a class that doesn’t compile fails to meet.

In particular, you could write a version that updates the data in the array directly if that is a wait-free operation, and a version that updates a pointer to data if not.

Your Safety Check is not Correct

You use two separate operations to retrieve the two cursors, compare their values (which might not both have been updated) to see whether it is safe to proceed, then increment the producer cursor (which might have changed in the interim).

Without having done a rigorous analysis, it appears at a glance that one thing that might go wrong with this enqueue or dequeue is that you might not have seen an update to the producer cursor, leading you to spuriously conclude that the buffer is not full when it is, and then increment the producer cursor again after it updates, causing the cursor to wrap around and the container to believe the buffer is empty.

What you want is to store the counters in a struct. (See update below.)

Even then, I don’t see what prevents an element from being dequeued before it is fully enqueued. TMPMCQueue::dequeue only checks that its retrieved values of the two cursors (at separate times) are not equal. They are not necessarily equal after the producer has incremented the producer cursor. So what prevents a consumer from seeing that the producer has moved the producer cursor past the element it is writing, then moving the consumer cursor to the element and reading it before it is fully written?

The Element Type

I’m not sure why you store pointers, as you are not doing receive-copy-update by atomically updating the pointer. You instead malloc dynamic memory for each, then dereference and copy the source data to this buffer. You also add roughly twice as much padding as you need, but not an exact multiple of the padding size.

Your elements should be aligned with alignas. If you need to pad, which you probably don’t, you could add an array of char[cache_size - sizeof(T)%cache_size].

One approach you could take would be to store the elements in a properly-aligned struct containing the data, and make the array of those. If the algorithm is correct, overwriting the data in the array should be more efficient than retrieving and writing through a pointer. If you do it this way, you must ensure the read cursor does not reach your slot until you have updated.

Another approach you could take is to move a std::atomic<std::unique_ptr> that you receive from the producers into the buffer, thus updating a large element in a singe wait-free atomic operation without copying the data. These smart pointers would need to allocate from some thread-local pool of memory if the threads are not to contend for the heap. Dequeueing an element then transfers ownership of the pointer to the consumer thread, which can recycle the memory, or free it by letting the smart pointer expire.

Alternatively, you could receive a weak pointer to the producer data, which must live long enough to be read.

Minor Stylistic Nits

Your spin locks could use the STL to call std::this_thread::yield in the idle-lock, rather than use a non-standard function.

You don’t need to restrict yourself to powers of two when you do compare-and-swap. If you do, though, there are bit-twiddling tricks involving -x^~x that can get you the bitmask you want.

Update on Atomically Updating the Counters

I’d previously suggested a data structure for the counters that did not work properly on some compilers. If you want to load or update two 64-bit counters with a single atomic operation, you need to declare a struct containing two counters as members, then wrap that in std::atomic.

ICC 2021 is the best at optimizing atomic updates to this structure, compiling compare_exchange_weak to a lock cmpxchg16b instruction on x86_64-v4. Clang++ 13.0.0 can also make the same optimization, but only if you align your struct to 16-byte boundaries. G++ 11.2 cannot.

The following code:

#include <atomic>
#include <cassert>
#include <stddef.h>
#include <thread>

/* Clang++ 13.0.0 needs alignas(16) to inline atomic updates to cmpxchg16b, on x86_64-v4.
 * ICC 2021 does not.  g++ 11.2 fails to, either way.
 */
#if __clang__ && __x86_64__
#  define COUNTER_ALIGNMENT alignas(16)
#else
#  define COUNTER_ALIGNMENT /**/
#endif

struct COUNTER_ALIGNMENT counter_pair {
    size_t producer_counter;
    size_t consumer_counter;
};

static_assert( std::atomic<counter_pair>::is_always_lock_free, "" );

std::atomic<counter_pair> counters(counter_pair{});

size_t inc_producer_counter()
{
  counter_pair current_counters = counters.load(std::memory_order_acquire);
  const size_t new_producer_counter = current_counters.producer_counter + 1;

  if ( counters.compare_exchange_weak(current_counters,
                                      { new_producer_counter,
                                        current_counters.consumer_counter },
                                      std::memory_order_release )
     )
  {
    return new_producer_counter;
  } else {
    std::this_thread::yield();
    return inc_producer_counter();
  }
}

compiles to the following assembly on clang++ -std=c++20 -O3 -march=x86_64-v4 with version 13.0.0:

inc_producer_counter():              # @inc_producer_counter()
        push    rbx
.LBB0_2:                                # =>This Inner Loop Header: Depth=1
        xor     eax, eax
        xor     edx, edx
        xor     ecx, ecx
        xor     ebx, ebx
        lock            cmpxchg16b      xmmword ptr [rip + counters]
        lea     rbx, [rax + 1]
        mov     rcx, rdx
        lock            cmpxchg16b      xmmword ptr [rip + counters]
        je      .LBB0_3
        call    sched_yield
        jmp     .LBB0_2
.LBB0_3:
        mov     rax, rbx
        pop     rbx
        ret
counters:
        .zero   16

Or, if 32-bit counters are enough, you can change size_t to uint_least32_t.

Answer 2

Disclaimer

I'm not that competent in lock free datastructures, so I just pulled out a book called "The art of multiprocessor programming" and went through the code line by line looking for errors mentioned in the book. If you see serious error please downvote and write your own answer explaining the problem.

---

Memory leak and undefined behavior

Destructor of the queue does not free the individual elements nor does it invoke destructor for the elements. I did not notice static_asserts for value type being trivially destructible.

Cache line alignment

I would just write this

template <typename T>
struct aligned_atomic {
    alignas(std::hardware_destructive_interference_size) std::atomic<T> data;
};

Which is actually shown in the examples. Unfortunately gcc and clang doesn't have the constant yet, but MSVC does.

This makes it portable and avoids all of the unnecessary infrastructure around it. Also, the idea is that the size of the whole struct is divisible by 64, so the equation should be padding_size = 64 - sizeof(T) % 64 then

alignas(64) struct aligned_buffer_t {
    T data;
    std::byte _padding_bytes[padding_size];
};

So that now sizeof(aligned_buffer_t) % 64 == 0. Notice that this time alignas is on the struct, because otherwise the padding bytes would stick out from the cache line.

Too much cache line alignment

Inside the queue, there is too much padding. If the variables are used together, they shouldn't be aligned separately, but together, so that sizeof of all of them is less than std::hardware_constructive_interference_size. At least the producer/consumer cursors belong together, perhaps with the data pointer.

Atomic is not needed on const variables initialized in constructor

Index mask is only initialized once and never changed. There is no need to put atomic on it, as accessing the variables during object construction (from outside the constructor) is undefined behavior.

Broken synchronization ?

        const int_fast64_t ClaimedIndex = ProducerCursor.IncrementAndGetOldValue(); // fetch_add

This line in Enqueue seems like it needs CAS. Let me give you an example where the current version breaks. Imagine threads A and B that both did the check while the queue had only one empty space left. A gets unscheduled so only B gets to try and fills the last empty spot. After A gets scheduled again, it thinks that there is still spot left so it will just increment and overflow.

There is similar overflow problem in EnqueueCAS. If CAS failed it still has to check for fullness. The line shown below also seems redundant, as it should already be updated:

            ClaimedIndex = ProducerCursor.Get();

Use consume if you can

It seems like acquire operations are not needed and simple consume will suffice. From what I understood, the latter is faster on ARM and weaker memory order architectures.

Answer 3

It does not work properly

Davislor already mentioned that in his post, but let me make it abundantly clear by showing you a simplified version of your Enqueue() and Dequeue() functions:

Enqueue(const FElementType& NewElement)
{
    auto ClaimedIndex = ClaimIndexForEnqueue();
    *RingBuffer[ClaimedIndex].Data = NewElement;
}

Dequeue(FElementType& Output)
{
    auto ClaimedIndex = ClaimIndexForDequeue();
    Output = *RingBuffer[ClaimedIndex].Data;
}

Even if you update the indices correctly atomically in the first line of each function, there is nothing that guarantees atomic access to *RingBuffer[ClaimedIndex].Data in the second line of each function.

Avoid reinventing the wheel

When I see the comments in this piece of code:

/** Perform a fetch_add with acquire_release semantics */
const int_fast64_t ClaimedIndex = ConsumerCursor.IncrementAndGetOldValue(); // fetch_add

I cannot help but think: wow, that person really wants to do a fetch_add() with acquire-release semantics, why didn't they just use a std::atomic<int_fast64_t> and write:

const int_fast64_t ClaimedIndex = ConsumerCursor.fetch_add(1, std::memory_order_acq_rel);

That way, no comments are necessary. I suggest you use a properly aligned std::atomic directly, and perhaps provide helper functions for those things not provided by std::atomic directly. For example:

template<typename T>
void SetFullFence(std::atomic<T> &Data, const T& NewData)
{
    std::atomic_thread_fence(std::memory_order_release);
    Data.store(NewData, std::memory_order_relaxed);
    std::atomic_thread_fence(std::memory_order_seq_cst);
}

Alternatively, if you really want to wrap std::atomic in your own class, at least use the same naming conventions as std::atomic.