atomic_queue - thread-safe and lock-free implementation of the FIFO data structure pattern

Question

please review my implementation of a FIFO data structure that guarantees thread-safe access without locking.

I removed the license header block and all Doxygen comments. The original version (along with some tests) is available from here: https://github.com/fat-lobyte/atomic_queue

It should compile with

• Clang >= 3.1, with -stdlib=libc++
• GCC => 4.7, with -D_GLIBCXX_USE_SCHED_YIELD
• Visual Studio >= 2012 (_MSC_VER >= 1700), without emplace_back() member function

I would ask you to check if

• It actually is thread safe (this is the main concern)
• It is compliant with the C++11 standard
• It performs reasonably and no resources are wasted
• If its possible to implement it more simple without losing performance or flexibility

Additionally, if someone is a real pro in lock-free programming, they can help me answer the following question: Can I relax some of the load/store/exchange operations memory ordering constraints (pass an std::memory_order constant)? I do not understand the memory model properly, and quite frankly I find it is a really difficult to grasp.

Help would be very appreciated. Thanks in advance!

#ifndef ATOMIC_QUEUE_HPP_INCLUDED
#define ATOMIC_QUEUE_HPP_INCLUDED

#include <cstddef>
#include <memory>
#include <atomic>
#include <thread>

// At the time of writing, MSVC didn't know noexcept
#if defined(_MSC_VER) && _MSC_VER <= 1700
#   define noexcept throw()
#endif

namespace aq {

namespace detail {

/** A node in the linked list. */
template <typename T>
struct node
{
    T t /**< Value */;
    std::atomic<node<T>*> next /**< Next node in list */;
};


template <typename T, typename Allocator = std::allocator<T> >
class atomic_queue_base
{
public:

    atomic_queue_base(const Allocator& alc = Allocator()) noexcept
        : size_(0u), front_(nullptr), back_(nullptr),
        alc_(alc)
    { }


    ~atomic_queue_base() noexcept
    {
        node<T>* fr = front_;

        while(fr)
        {
            node<T>* next = fr->next;
            NodeAllocatorTraits::destroy(alc_, fr);
            NodeAllocatorTraits::deallocate(alc_, fr, 1);
            fr = next;
        }

    }


    void push_back(const T& t)
    {
        auto new_node = NodeAllocatorTraits::allocate(alc_, 1);

        try {
            ValueAllocator alc(alc_);
            ValueAllocatorTraits::construct(
               alc,
               &new_node->t, t
            );
        } catch(...)
        {
            NodeAllocatorTraits::deallocate(alc_, new_node, 1);
            throw;
        }
        new_node->next = nullptr;

        push_node(new_node);
    }


    void push_back(T&& t)
    {
        auto new_node = NodeAllocatorTraits::allocate(alc_, 1);

        try {
            ValueAllocator alc(alc_);
            ValueAllocatorTraits::construct(
               alc, &new_node->t, std::move(t)
            );
        } catch(...)
        {
            NodeAllocatorTraits::deallocate(alc_, new_node, 1);
            throw;
        }
        new_node->next = nullptr;

        push_node(new_node);
    }


#if !(defined(_MSC_VER) && _MSC_VER <= 1700)
    template<typename... Args>
    void emplace_back(Args&&... args)
    {
        auto new_node = NodeAllocatorTraits::allocate(alc_, 1);

        try {
            ValueAllocator alc(alc_);
            ValueAllocatorTraits::construct(
               alc, &new_node->t, std::forward<Args>(args)...
            );
        } catch(...)
        {
            NodeAllocatorTraits::deallocate(alc_, new_node, 1);
            throw;
        }
        new_node->next = nullptr;

        push_node(new_node);
    }
#endif

    T* pop_front() noexcept
    {
        node<T>* old_front = front_;
        node<T>* new_front;

        do {
            if (!old_front) return nullptr; // nothing to pop
            new_front = old_front->next;
        } while(!front_.compare_exchange_weak(old_front, new_front));

        --size_;

        // if the old front was also the back, the queue is now empty.
        new_front = old_front;
        if(back_.compare_exchange_strong(new_front, nullptr))
            old_front->next = old_front;

        return reinterpret_cast<T*>(old_front);
    }


    void deallocate(T* obj) noexcept
    {
        if (!obj) return;

        // call destructor
        NodeAllocatorTraits::destroy(alc_, reinterpret_cast<node<T>*>(obj));

        // nodes with next == 0 are still referenced by an executing
        // push_back() function and the next ptr will be modified.
        // Since we don't want push_back() to write to deallocated memory, we hang
        // in a loop until the node has a non-zero next ptr.
        while(!reinterpret_cast<node<T>*>(obj)->next.load())
            std::this_thread::yield();

        NodeAllocatorTraits::deallocate(
            alc_, reinterpret_cast<node<T>*>(obj), 1
        );
    }


    std::size_t size() const noexcept
    { return size_; }

protected:

    void push_node(node<T>* new_node) noexcept
    {
        node<T>* old_back = back_.exchange(new_node);
        node<T>* null_node = nullptr;

        // if front_ was set to null (no node yet), we have to update the front_
        // pointer.
        if(!front_.compare_exchange_strong(null_node, new_node))
            // if front_ is not null, then there was a previous node.
            // We have to update this nodes next pointer.
            old_back->next  = new_node;

        ++size_;
    }

    typedef Allocator ValueAllocator;
    typedef std::allocator_traits<ValueAllocator> ValueAllocatorTraits;

    // Rebind allocator traits for ValueAllocator to our own NodeAllocator
    typedef
        typename ValueAllocatorTraits::template rebind_traits<node<T> >
#if defined(_MSC_VER) && _MSC_VER <= 1700
        ::other
#endif
        NodeAllocatorTraits;

    // Get actual allocator type from traits
    typedef typename NodeAllocatorTraits::allocator_type NodeAllocator;

    std::atomic_size_t size_ /**< Current size of queue. Not reliable. */;
    std::atomic<node<T>*> front_ /**< Front of the queue. */;
    std::atomic<node<T>*> back_ /**< Back of the queue. */;
    NodeAllocator alc_ /**< Allocator for node<T> objects. */;
};

} // namespace detail

using detail::atomic_queue_base;

} // namespace aq



#endif // ifndef ATOMIC_QUEUE_HPP_INCLUDED

Answer 1

Pretty sure this is broken in multi-threaded code:

    do
    {
        // Point A.
        if (!old_front) return nullptr;
        new_front = old_front->next;  
    }
    while(!front_.compare_exchange_weak(old_front, new_front));

What happens if:

1. Thread A. Reaches point A
2. Thread A is de-scheduled and is not currently running.
3. Thread B. runs through the above code and works
4. Thread B. destroyes the node it poped off the front.
5. Thread A. Is re-scheduled to run.
6. Thread A. Any use of old_front is invalid as it points at a node that was deleted by Thread B
7. Thread A. Even if the node was not destroyed it is no longer the head of the list.
   Thus any usage of next is suspect.

Answer 2

I think the design is not very useful (apart from any possible thread safety issues). pop() just returns a pointer and the user is trusted with eventually de-allocating it. This cannot be exception safe (leaking memory when an exception is thrown and the pointer returned by pop() lost). This makes the code useless in many situations where you want to use a threadsafe stack (and is exactly what you don't want to do in C++, it smells like C).

Anthony Williams (in "C++ concurrency in action") implements a lock-free stack that returns a std::shared_ptr<T> in pop(). I have no experience with that, but implementing std::shared_ptr to be threadsafe and lock-free cannot be trivial (I don't know how good the implementations provided in the various libstdc++ are).