8
\$\begingroup\$

I'm looking for some feedback on my lock-free queue, based on Disruptor, mainly for any potential concurrency issues, such as where I need additional fences. It looks correct to me, and I can't seem to able to break it, but I'm only testing it on my x86 machine.

#pragma once

#include <atomic>
#include <cstdint>

template <typename T> class LockFreeMPMCQueue
{
    public:
    explicit LockFreeMPMCQueue( size_t size )
        : m_data( new T[size] ), m_size( size ), m_head_1( 0 ), m_head_2( 0 ), m_tail_1( 0 ), m_tail_2( 0 )
    {
    }

    virtual ~LockFreeMPMCQueue() { delete[] m_data; }

    // non-copyable
    LockFreeMPMCQueue( const LockFreeMPMCQueue<T>& ) = delete;
    LockFreeMPMCQueue( const LockFreeMPMCQueue<T>&& ) = delete;
    LockFreeMPMCQueue<T>& operator=( const LockFreeMPMCQueue<T>& ) = delete;
    LockFreeMPMCQueue<T>& operator=( const LockFreeMPMCQueue<T>&& ) = delete;

    bool try_enqueue( const T& value )
    {
        const std::uint64_t head = m_head_2.load( std::memory_order_relaxed );
        std::uint64_t tail = m_tail_1.load( std::memory_order_relaxed );

        const std::uint64_t count = tail - head;

        // count could be greater than size if between the reading of head, and the reading of tail, both head
        // and tail have been advanced
        if( count >= m_size )
        {
            return false;
        }

        if( !std::atomic_compare_exchange_strong_explicit(
            &m_tail_1, &tail, tail + 1, std::memory_order_relaxed, std::memory_order_relaxed ) )
        {
            return false;
        }

        m_data[tail % m_size] = value;

        while( m_tail_2.load( std::memory_order_relaxed ) != tail )
        {
            std::this_thread::yield();
        }

        // Release - read/write before can't be reordered with writes after
        // Make sure the write of the value to m_data is
        // not reordered past the write to m_tail_2
        std::atomic_thread_fence( std::memory_order_release );
        m_tail_2.store( tail + 1, std::memory_order_relaxed );

        return true;
    }

    bool try_dequeue( T& out )
    {
        const std::uint64_t tail = m_tail_2.load( std::memory_order_relaxed );
        std::uint64_t head = m_head_1.load( std::memory_order_relaxed );

        if( head >= tail )
        {
            return false;
        }

        if( !std::atomic_compare_exchange_strong_explicit(
            &m_head_1, &head, head + 1, std::memory_order_relaxed, std::memory_order_relaxed ) )
        {
            return false;
        }

        out = m_data[head % m_size];

        while( m_head_2.load( std::memory_order_relaxed ) != head )
        {
            std::this_thread::yield();
        }

        // Release - read/write before can't be reordered with writes after
        // Make sure the read of value from m_data is
        // not reordered past the write to m_head_2
        std::atomic_thread_fence( std::memory_order_release );
        m_head_2.store( head + 1, std::memory_order_relaxed );

        return true;
    }

    size_t capacity() const { return m_size; }

    private:
    T* m_data;
    size_t m_size;

    // Make sure each index is on its own cache line
    char pad1[56];
    std::atomic<std::uint64_t> m_head_1;
    char pad2[56];
    std::atomic<std::uint64_t> m_head_2;
    char pad3[56];
    std::atomic<std::uint64_t> m_tail_1;
    char pad4[56];
    std::atomic<std::uint64_t> m_tail_2;
};
\$\endgroup\$
  • \$\begingroup\$ For those less familiar with lock-free synchronization, what are the reasons behind the padding for each of the atomic counters? \$\endgroup\$ – glampert Mar 20 '16 at 18:33
  • \$\begingroup\$ @glampert It's to make sure the indexes are on separate cache lines, I'm making the assumption that a cache line is 64 bytes, which it is for most current x86 processors (I actually got the padding size wrong originally, I've amended that). If an index is written to on thread A, if the index is cached in thread B then it will be updated, this comes at the cost of time. Without the padding, all the indexes are likely on the same cache line, so a write to any of them results in the updating of the other threads caches. More info here youtube.com/watch?v=WDIkqP4JbkE \$\endgroup\$ – Joe Mar 20 '16 at 23:11
3
\$\begingroup\$

Missing barrier

When I looked at your code, I could immediately tell there was something wrong because you use a release barrier without an acquire barrier that pairs with it.

You have this barrier in try_enqueue() between writing to m_data and writing to m_tail_2:

std::atomic_thread_fence( std::memory_order_release );

But you don't have any corresponding acquire barrier in try_dequeue() between reading from m_tail_2 and reading from m_data. You should put in a barrier like this:

bool try_dequeue( T& out )
{
    // ...

    // Add this barrier which pairs with the other release barrier
    std::atomic_thread_fence( std::memory_order_acquire );

    out = m_data[head % m_size];

    // ...
}

Without this barrier, the cpu could read m_data[head] speculatively before the read of m_tail_2, and you could end up getting an older value instead of the value just written.

Sequence of events that lead to error

  1. Initially, m_data[0] = 0
  2. Cpu 1: try_enqueue(): m_data[0] = 5
  3. Cpu 1: try_enqueue(): fence(release)
  4. Cpu 1: try_enqueue(): m_tail_2 = 1
  5. Cpu 2: try_dequeue(): reads m_tail_2 as 1
  6. Cpu 2: try_dequeue(): exchange(head, head+1) (no effect)
  7. Cpu 2: try_dequeue(): reads m_data[0] as 0

Due to weak memory ordering, on step 7 above, m_data[0] can be read as 0 instead of 5 because there was no acquire barrier between steps 5 and 7. The strong exchange on step 6 did not perform a barrier because it was called with std::memory_order_relaxed, std::memory_order_relaxed as the memory ordering.

Note that there are 2 possible ways that memory accesses can be "reordered". The first is that the compiler might shift lines of code around, and therefore, your accesses are actually performed out of order with respect to a program execution timeline. This is not what is happening in my example above. The second way is that the memory system may reorder accesses. In other words, even if two reads occur one after the other in the program execution timeline, the second read may read an earlier value than the first read. This second case is what causes step 7 above to fail.

Possible missing barrier

Similarly, you might need an acquire barrier in try_enqueue() before your write to m_data, which pairs with the release barrier in try_dequeue(). I say "might" because this one is harder to explain. The barrier is only needed if the cpu would do a speculative write to m_data before reading m_head_2. But because m_head_2 is used to determine whether or not to actually write to m_data, there is a data dependency there which would seem to guarantee that m_head_2 is read before writing to m_data. However, there are processors such as the DEC Alpha which have strange behavior with respect to data dependencies. So I'm not 100% sure about whether this barrier is needed, and if it is needed, it may only be needed for DEC Alpha and similar processors.

\$\endgroup\$
  • \$\begingroup\$ atomic_compare_exchange_strong_explicit might change the value of head, so I was under the impression that the read of m_data[head] couldn't be reordered before the atomic_compare_exchange_strong_explicit. Is that wrong? \$\endgroup\$ – Joe Mar 21 '16 at 9:18
  • \$\begingroup\$ @Joe - you are right. None of these fences are required on x86. cmpxchg acts as a fence. \$\endgroup\$ – a25bedc5-3d09-41b8-82fb-ea6c353d75ae Mar 21 '16 at 9:27
  • \$\begingroup\$ @a25bedc5-3d09-41b8-82fb-ea6c353d75ae but would they be required on PPC/Arm/Dec Alpha? \$\endgroup\$ – Joe Mar 21 '16 at 10:15
  • \$\begingroup\$ @Joe I added a section to explain the sequence of events that would lead to an error. \$\endgroup\$ – JS1 Mar 21 '16 at 18:41
  • 1
    \$\begingroup\$ @Joe, I don't really know about weak-ordered platforms. I don't even know a single person who worked low level on them. I'm more interested to know how did my username become a guid! \$\endgroup\$ – a25bedc5-3d09-41b8-82fb-ea6c353d75ae Mar 22 '16 at 18:29

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.