6
\$\begingroup\$

I'm implementing a lock-free, multiple consumer, multiple producer FIFO queue/pipe as an exercise in thinking about atomicity in operations.

My main concern is correctness of operation, my second concern is good practices around atomics and general C++11. Performance is interesting but not important for this exercise.

Without futher ado, here's the code:

#include <atomic>
#include <exception>

// For dump
#include <iostream>
#include <string>

/// <summary> A lock free queue implementation.
/// 
/// Design notes: Here be dragons. The queue is implemented as a single linked 
/// list with a head, divider and tail pointer. These are always ordered such 
/// that "head -> divider -> tail" and are always non-null. The divider's next
/// pointer points to the first node with data or is null. In other words, this
/// means that "divider == tail -> empty container". Nodes between head and 
/// divider are empty and will be freed lazily. </summary>
///
/// <remarks> * Thread Safety   : Full.
/// * Exception Safety: Basic. </remarks>
///
/// <tparam name="T"> Generic type parameter. </tparam>
template<typename T>
class lockfree_queue{
    struct link;
    using link_ptr = std::atomic < link* > ;

    struct link{
        link() noexcept = default;
        link(const link&) = delete;
        link& operator = (const link&) = delete;

        link_ptr m_next{ nullptr };
    };

    struct node : link{
        template<typename... Args>
        node(Args&&... args)
            : m_data(std::forward<Args>(args)...)
        {}
        T m_data;
    };

public:
    using size_type = std::size_t;
    using value_type = T;

    /// <summary> Destructor, it's the users responsibility to make sure that 
    ///           no one uses the class after it's destruction and that no 
    ///           thread is in any of the function bodies. </summary>
    ~lockfree_queue(){
        free_nodes(m_head.m_next.load());
    }

    /// <summary> Tests if this container is empty. This operation only makes
    ///           sense if there is only one thread reading/consuming the queue.
    ///           </summary>
    /// <returns> True if the queue is empty, false otherwise. </returns>
    bool empty() const noexcept{
        return m_divider.load() == m_tail.load();
    }

    /// <summary> Gets the instantaneous number of elements in the queue. 
    ///           Mostly useful as a debug probe to monitor the queue size. 
    ///           </summary>
    /// <returns> The number of elements in the queue. </returns>
    size_type size() const noexcept{
        return m_size;
    }

    /// <summary> Emplaces a new node on the queue. If the construction of the 
    ///           data throws, the queue is unmodified. </summary>
    /// <tparam name="Args"> Type of the arguments. </tparam>
    /// <param name="args"> Variable arguments providing the arguments to 
    ///                     construct the data with.</param>
    template<typename... Args>
    void emplace(Args&&... args){
        auto l_new_node = new node(std::forward<Args>(args)...);

        // m_tail->m_next can have two states:
        // 1) It's non-null, means an insertion is in progress but has not bee completed.
        // 2) It's null, means no insertion is in progress.
        // m_tail->m_next will only be written from this function.

        // This loop does a CAS with m_tail->m_next to see if it is null and if it is it
        // inserts the new node. At which point any concurrent push will retry until (3)
        // below completes.
        link* l_null = nullptr;
        while (!m_tail.load()->m_next.compare_exchange_weak(l_null, l_new_node));
        m_tail = l_new_node; // 3) Commit/publish the new tail.

        m_size++;
    }

    void dump(){
        auto n = &m_head;

        while (n != nullptr) {
            std::string special = "";
            if (n == m_divider.load())
                special += "D";
            if (n == m_tail.load())
                special += "T";

            std::cout << "[(" << special << ")";
            if (n != &m_head)
                std::cout << "\"" << static_cast<node*>(n)->m_data << "\"";
            else
                std::cout << "sentinel";
            std::cout << "(" << n << ")] -> ";

            n = n->m_next.load();
        }
        std::cout << "[null]" << std::endl;
    }

    /// <summary> Consumes one item from the queue. The item is move assigned 
    ///           to result. If the assignment throws, the queue is will have
    ///           dropped consumed item but is otherwise unmodified. </summary>
    /// <param name="result"> [in,out] The result. </param>
    /// <returns> An auto. </returns>
    bool consume(T& result){
        link* l_divider = nullptr;
        link* l_snack = nullptr;

        // Try to temporarily unlink the head if it is not already unlinked and
        // it's not the divider
        auto l_head = m_head.m_next.load();
        if (l_head == nullptr || &m_head == m_divider.load() || 
            !m_head.m_next.compare_exchange_strong(l_head, nullptr)){

            l_head = nullptr; // We didn't get to unlink the head this time.
        }

        do{
            // The divider's next pointer points to the next node with data.
            l_divider = m_divider.load();
            l_snack = l_divider->m_next.load(); // divider is never null.

            if (nullptr == l_snack)
                return false; // empty

            // If the CAS below succeeds, then no one has moved the divider since
            // we loaded the new divider position (which is non-null) and we have
            // moved the divider to the next node without interruption.
        } while (!m_divider.compare_exchange_weak(l_divider, l_snack));
        m_size--;


        try{
            result = std::move(static_cast<node*>(l_snack)->m_data);
            cleanup_pop(l_head, l_divider);
        }
        catch (...){
            cleanup_pop(l_head, l_divider);
            std::rethrow_exception(std::current_exception());
        }
        return true;
    }

private:
    void free_nodes(link* from, link* up_until = nullptr) noexcept {
        assert(from != &m_head);

        while (from != up_until){
            auto next = from->m_next.load();
            // All links but the head are nodes, necessary to destroy data.
            delete static_cast<node*>(from);
            from = next;
        }
    }

    void cleanup_pop(link* l_head, link* l_divider) noexcept {
        if (l_head){
            // The head has been unlinked by us and we are the only ones 
            // with a handle to the detached head. We can now safely free
            // all nodes from the detached head up until the divider.
            auto new_divider = l_divider->m_next.load();
            free_nodes(l_head, new_divider);

            // Oh, and re-link the head
            m_head.m_next = new_divider;
        }
    }

    link m_head;
    link_ptr m_divider{ &m_head };
    link_ptr m_tail{ &m_head };
    std::atomic<size_type> m_size{ 0 };
};

int main(){

    lockfree_queue<double> q;

    assert(true == q.empty());
    assert(0 == q.size());
    q.dump();
    q.emplace(0);
    q.dump();

    q.emplace(1);
    q.dump();
    q.emplace(2);
    q.dump();
    q.emplace(3);
    q.dump();

    double ans;
    q.consume(ans);
    q.dump();
    assert(ans == 0);

    q.consume(ans);
    q.dump();
    assert(ans == 1);

    q.consume(ans);
    q.dump();
    assert(ans == 2);

    q.consume(ans);
    q.dump();
    assert(ans == 3);

    q.emplace(3.14);
    q.dump();
    q.consume(ans);
    q.dump();
    assert(ans == 3.14);
}

I'm also interested in if anyone has some ideas on good test cases for the correctness under concurrency.

\$\endgroup\$
5
\$\begingroup\$

ABA problem

I was able to break your queue (but it wasn't easy). I inserted some code to freeze one thread here in consume():

    do{
        // The divider's next pointer points to the next node with data.
        l_divider = m_divider.load();
        l_snack = l_divider->m_next.load(); // divider is never null.

        if (nullptr == l_snack)
            return false; // empty

        // Special hack to freeze one thread at a dangerous spot.
        if (freeze) {
            freeze = 0;
            frozen = 1;
            while (frozen);
        }

        // If the CAS below succeeds, then no one has moved the divider since
        // we loaded the new divider position (which is non-null) and we have
        // moved the divider to the next node without interruption.
    } while (!m_divider.compare_exchange_weak(l_divider, l_snack));

At this point, one thread was trying to move the divider from A to B like this:

divider(A) -> B -> C -> D
    trying to swap A with B to end up like this:
divider(B) -> C -> D

So the thread was frozen with l_divider being A and l_snack being B.

Then I ran another thread and caused it to consume the whole queue (ABCD all freed). In that other thread, I used emplace() to put new nodes on the stack, and I carefully manipulated the allocator to force this situation:

divider(A) -> C -> D

When I say I manipulated the allocator, I mean I did an extra allocation to make sure that B was skipped. At that point, I set frozen = 0 to unfreeze the first thread. What happened was that it swapped A with B like this:

divider(B) -> ?

But of course B was no longer part of the queue. So after that, any future consumes were broken. I actually made B point at itself, so the consumes kept consuming B forever.

This problem is known as the ABA problem in case you have not already learned about it.

\$\endgroup\$
  • \$\begingroup\$ I've heard ABA mentioned but I didn't know what it was until now. Thanks, do you have an suggestions for how to avoid the problem in this case? \$\endgroup\$ – Emily L. May 14 '15 at 10:55
  • \$\begingroup\$ @EmilyL. This other code review question did it with versioning. \$\endgroup\$ – JS1 May 14 '15 at 11:05
  • \$\begingroup\$ Hmm am I correct in that emplace also suffers from the ABA problem? If the producer (A) is pre-empted after the load but before the CAS. Another thread produces new items and consumes them to the point of the loaded pointer by (A) being freed before the CAS resulting in access to released memory? \$\endgroup\$ – Emily L. May 14 '15 at 11:16
  • \$\begingroup\$ @EmilyL. You may be right about emplace having a problem. I thought that for some reason that you had a sentinel at the tail but now I see that what tail was pointing to could be freed by some other thread. \$\endgroup\$ – JS1 May 14 '15 at 12:08

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.