15
\$\begingroup\$

The code below implements an intrusive lock-free queue that supports multiple concurrent producers and consumers.

Some features:

  • Producers and consumers work on separate ends of the queue most of the time.
  • The fast path for producers and consumers has 4 atomic ops.
  • Version numbers are used to prevent the ABA problem.
  • Threads assist other threads in completing their operations, they're never blocked waiting for another thread to do something.

I'm particularly interested in comments regarding the correctness of this design. Unless proven otherwise I'm assuming that this code is broken, although I've tried to the best of my ability to have a correct lock-free implementation.

#include <cstdint>
#include <atomic>

struct LockfreeNode
{
  std::atomic<LockfreeNode>* next;
  uintptr_t version;

  LockfreeNode()
    : next(nullptr)
    , version(0)
  {
  }

  LockfreeNode(std::atomic<LockfreeNode>* next, uintptr_t version)
    : next(next)
    , version(version)
  {
  }
};

template <class T>
class LockfreeQueue
{
public:
  typedef LockfreeNode Node;
  typedef std::atomic<Node> AtomicNode;
  typedef typename AtomicNode T::* NodeMemberPointer;

  LockfreeQueue(NodeMemberPointer node)
    : mNodeMember(node)
  {
    mSentinel = Node(&mSentinel, PreviousVersion(1));
    mHead = Node(&mSentinel, 1);
    mTail = Node(&mSentinel, 1);
  }

  void Queue(const T& element)
  {
    AtomicNode* node = ToNode(element);

    QueueNode(node, 0);
  }

  T* Dequeue()
  {
    AtomicNode* result = DequeueNode();

    return result ? ToElement(result) : nullptr;
  }

private:
  void QueueNode(AtomicNode* node, uintptr_t generation)
  {
    // Capture the tail to figure out what the last node is. The last node can be a 
    // regular node or it can be the sentinel.
    //
    // - Regular node:
    //
    // A->s->B->C  N                    ; C is the last node, N is the new node
    // 1  2  2  2
    // ^        ^
    // h        t
    // 2        2
    //
    // - Sentinel:
    //
    // B->C->D->s  N
    // 2  2  2  2
    // ^        ^
    // h        t
    // 3        3

    Node tail = mTail.load(std::memory_order_acquire); // (Q.1)

    // Note that other threads could update the tail in (Q.5) in the meantime, 
    // invalidating our local copy above. In that case none of the operations below 
    // will have any effect and we'll end up looping again, retrying the queueing 
    // operation with a fresh copy of the tail captured in (Q.5).

    Node last;
    bool queued = false;

    // The loop below will continue until we've successfully queued the new node.
    // While looping, we'll potentially be helping complete other in-progress 
    // operations initiated by other threads.

    for (;;)
    {
      // Check whether the last node is the sentinel or not so that we can build 
      // the right comparand for the CAS operation in (Q.4).

      if (tail.next == &mSentinel)
      {
        // The last node is the sentinel, which means one of two things:
        //
        // - A consumer thread is potentially still in the process of bumping 
        // the sentinel, having executed (Q.4) and (Q.5) already but not 
        // necessarily the final head swing in (Q.3):
        //
        // s->B->C->D->s            ; Both 's' entries refer to the same sentinel
        // 2  2  2  2  2
        // ^           ^
        // h           t            ; The head might not have been swung to B yet
        // 2           3
        //
        // If the operation is still ongoing, all that remains to do is to swing
        // the head forward in (Q.3), allowing consumers to start dequeueing the 
        // next generation of elements that would now be to the left of the 
        // sentinel. If the operation has been completed already by another 
        // thread then our CAS in (Q.3) will do nothing because the head version 
        // number won't match the comparand there. After that, if we were 
        // originally trying to queue a regular node then we'll go ahead with the
        // rest of the loop iteration and try to do that (Q.4) using the 
        // comparand we'll build in 'last' below, otherwise we'll just return.
        //
        // - The list was empty and a producer thread is in the process of 
        // queueing the first node of a new generation, having linked it to the 
        // sentinel in (Q.4) but not having swung the tail forward to it in (Q.5)
        // yet:
        //
        //  s->B                    ; B is being queued by a producer
        //  2  2
        // ^ ^
        // h t                      ; The tail hasn't been updated yet
        // 2 2
        //
        // In this case we need to help that thread complete the queueing 
        // operation by swinging the tail in (Q.5) before doing anything else. 
        // Our CAS in (Q.3) will fail because the head version number won't match
        // the comparand, and the next CAS in (Q.4) will fail as well because the
        // sentinel version number won't match the comparand we'll build in
        // 'last' either. After that we'll try swinging the tail in (Q.5) and 
        // proceed to loop again to perform our original operation.

        // The first thing to do is capture the contents of the sentinel to 
        // determine what node follows it, so we can swing the head to that node 
        // in (Q.3) and so we can link the new node to it later in (Q.4). Note 
        // that we ignore the actual sentinel version here when building the 
        // 'last' comparand and always set it to the version prior to the tail's.
        // This is to make sure that the CAS in (Q.4) only succeeds in the case 
        // where the sentinel has nothing linked to the right:
        //
        // B->C->D->s               ; The sentinel isn't linked to anything
        // 2  2  2  2               ; The sentinel version is lower than the tail
        // ^        ^
        // h        t
        // 2        3
        //
        // And so that it fails in the case where another node was linked to it 
        // without the tail having been swung forward yet:
        //
        //  s->B                    ; The sentinel is already linked to something
        //  2  2                    ; The sentinel version is same as the tail
        // ^ ^
        // h t                      ; The tail needs to be swung forward first
        // 2 2

        last = mSentinel.load(std::memory_order_acquire); // (Q.2)
        last.version = PreviousVersion(tail.version);

        // We now have copy of the sentinel with a pointer to the node we need to
        // swing the head to.
        //
        // Other threads could invalidate this copy of the sentinel by linking a 
        // new node to it in (Q.4). In that case, the head update in (Q.3) would 
        // have necessarily been performed already by another thread, so our CAS 
        // operations in (Q.3) and (Q.4) would have no effect. We may still end 
        // up helping the other threads by swinging the tail forward for them in 
        // (Q.5) if they haven't done so already:
        //
        // B->C->D->s->E            ; E was queued by another thread
        // 2  2  2  3  3
        // ^        ^
        // h        t               ; We may be able to swing the tail later
        // 3        3
        //
        // Note that we don't need to read the current head here, since we know 
        // what its contents must be if it's still pointing to the sentinel:
        //
        // s->B->C->D->s   
        // 2  2  2  2  2
        // ^           ^
        // h           t            ; Head still pointing to the sentinel
        // 2           3            ; Head version number is same as sentinel's

        Node head(&mSentinel, last.version);
        Node newHead(last.next, tail.version);

        // Swing the head forward to complete the sentinel bump operation. If the
        // head had already been updated in (Q.3) by another thread then the CAS 
        // below will fail:
        //
        // B->C->D->s               ; B, C and D can now be dequeued
        // 2  2  2  2
        // ^        ^
        // h        t               ; The head was updated
        // 3        3
        //
        // This operation will also update the head version to match the tail.

        mHead.compare_exchange_strong(head, 
                                      newHead, 
                                      std::memory_order_acq_rel); // (Q.3)
      }
      else
      {
        // The last node is a regular node. Unless another producer has just 
        // linked a new node to it we know that the contents of that node will 
        // be a pointer  to the tail and the tail's version:
        //
        // A->s->B->C  N            ; C has known contents
        // 1  2  2  2
        // ^        ^
        // h        t
        // 2        2
        //
        // If on the other hand there is a node linked to it then the comparand 
        // below won't match and the CAS in (Q.4) will fail, after which we'll 
        // swing the tail forward to help the other thread complete its queueing
        // operation and then we'll retry our own operation:
        //
        // A->s->B->C->D  N         ; D was queued by another producer
        // 1  2  2  2  2
        // ^        ^
        // h        t               ; The tail may not have been updated yet
        // 2        2

        last = Node(&mTail, tail.version);
      }

      // If we're trying to bump the sentinel we'll know that it has now been 
      // bumped when the tail version is different than the one passed to this 
      // function. This is the only test we need to detect this condition, 
      // regardless of which part this thread played in the bumping. We can just 
      // check the tail version because it must have been updated in (Q.5) during 
      // the sentinel bump and, if so, the head must have been swung in (Q.3) 
      // already if we're at this point.

      if (node == &mSentinel && tail.version != generation)
      {
        queued = true;
      }

      // At this point we can return if the node has been successfully queued.

      if (queued)
      {
        break;
      }

      // If we're queueing a regular node (as opposed to the sentinel) then we need
      // to set the node's version number to that of the tail before linking it.
      // Sentinels instead have their version number updated by someone else in 
      // (Q.4) when a new node is queued after them.
      //
      // Also, we set the new node's next pointer to this queue's tail, which allows
      // us to make sure that in the event that this node is removed from this queue
      // and inserted in another queue that is using the same version number while a
      // producer thread here was about to link something to that node in (Q.4) that
      // CAS operation fails since the node's next pointer will necessarily be 
      // different. By making newly-queued nodes point to something unique to each 
      // queue (such as the head or tail) we ensure that the CAS will properly fail 
      // in this case.

      if (node != &mSentinel)
      {
        // This operation could be non-atomic.
        node->store(Node(&mTail, tail.version), std::memory_order_relaxed);
      }

      // Link the new node to the last node.
      //
      // - If the last node was the sentinel, then this operation will have the side
      // effect of incrementing the sentinel's version number to match the head and 
      // tail, signaling consumers that there is now a new generation of elements to
      // the right of the sentinel ready to be dequeued:
      //
      // B->C->D->s->N                ; N has been queued after the sentinel
      // 2  2  2  3  3                ; The sentinel's version has been updated
      // ^        ^
      // h        t
      // 3        3
      //
      // - If someone else has linked a node to the last node after we read the tail
      // but before we got to execute (Q.4) then the CAS below will fail, we'll get
      // back a copy of the pointer to that newly-queued node, and we'll go ahead 
      // and help that other thread complete its queueing operation by potentially 
      // swinging the tail forward in (Q.5) for them. We will then proceed to loop 
      // again and retry our operation with a fresh copy of the tail:
      //
      // A->s->B->C->D  N             ; D was queued by another thread before us
      // 1  2  2  2  2
      // ^        ^
      // h        t                   ; The tail might not have been updated yet
      // 2        2
      //
      // - If the last node was dequeued by a consumer in the meantime then we would
      // be accessing it after it's been removed from the data structure, which is 
      // safe under the assumption that storage for removed nodes is not released 
      // until this any other threads have finished accessing it.
      //
      // In this case the CAS below is guaranteed to fail due to one of the 
      // following three scenarios. First, if the consumer thread has executed 
      // (D.3) to acquire ownership of the node for removal then the node won't 
      // be pointing to the tail anymore (since it must have had something linked 
      // to it prior to the removal, at the very least the sentinel) so it won't 
      // match our CAS comparand that expects a pointer to the tail:
      //
      // C->D->E->s  N                ; C's being dequeued, no longer points to tail
      // 2  2  2  2
      //    ^     ^
      //    h     t
      //    3     3
      //
      // If the node has been readded to this queue after being dequeued, and it 
      // happens to be the last node again when we get to (Q.4) below, that CAS is
      // also going to fail because the node's version number must have increased 
      // when requeueing it:
      //
      // D->E->s->C  N                ; C has been requeued
      // 2  2  3  3                   ; C's version number has increased
      // ^        ^
      // h        t
      // 3        3
      //
      // Finally, if the node was removed from this queue and added to another 
      // queue, and the node was assigned the same version number on that queue 
      // as when it was on this one, then the CAS below will still fail because 
      // the node will no longer be pointing to this queue's tail; it will either 
      // be pointing to the other queue's tail if it's currently the last node 
      // there, or to some other node if it's not the last one. In either case it 
      // won't match our comparand that expects a pointer to our tail.

      Node newLast(node, tail.version);

      if (tail.next->compare_exchange_strong(last, 
                                             newLast, 
                                             std::memory_order_acq_rel)) // (Q.4)
      {
        // The new node has been successfully queued. We can exit the loop on the
        // next iteration.
        //
        // - If the new node was a regular node, it could have been linked to 
        // either another regular node or the sentinel. If it was queued to the 
        // sentinel then this operation had the side effect of incrementing the 
        // sentinel's version:
        //
        // B->C->D->s->N            ; N was linked to the sentinel
        // 2  2  2  3  3            ; The sentinel's version number was updated
        // ^        ^
        // h        t               ; The tail needs to be swung forward
        // 3        3
        //
        // or 
        //
        // A->s->B->C->N            ; N was linked to a regular node
        // 1  2  2  2  2
        // ^        ^
        // h        t               ; The tail needs to be swung forward
        // 2        2
        //
        // The only operation left to do in this case is to swing the tail forward
        // in (Q.5).
        //
        // - If the new node was the sentinel then it must have been linked to a 
        // regular node, and its version number won't be updated:
        //
        // s->B->C->s               ; The sentinel was queued
        // 2  2  2  2               ; The sentinel version hasn't been updated yet
        // ^     ^
        // h     t                  ; The head and tail need to be swung forward
        // 2     2
        //
        // In this case there are two operations left to do: swinging the tail in 
        // (Q.5) and swinging the head in (Q.3).

        last = newLast;
        queued = true;
      }

      // Swing the tail forward, using the contents of the last node that we 
      // captured above in (Q.4) to figure out what the next node to swing it to is.
      // It could be the node that we queued ourselves or it could be someone else's
      // node if they beat us to (Q.4).
      //
      // - If the new node is the sentinel then we need to increase the tail version
      // in this operation. New nodes linked after this point will use this new 
      // version number. This will effectively signal that a new generation of 
      // elements will start after the sentinel:
      //
      // s->B->C->s
      // 2  2  2  2
      // ^        ^
      // h        t
      // 2        3                   ; The tail version was updated
      //
      // - If someone else has updated the tail for us in (Q.5), then the CAS below
      // will fail due to a comparand mismatch, which can manifest in two different 
      // ways. First, if another thread has swung the tail and it's now pointing to 
      // a node different than the previous last node then the comparand pointer 
      // will differ:
      //
      // A->s->B->C->N                ; C was the last node
      // 1  2  2  2  2
      // ^           ^
      // h           t                ; The tail has been updated by someone else
      // 2           2                ; It's now pointing to something other than C
      //
      // On the other hand, if the previous last node was dequeued and then later 
      // requeued, and it happens to be the last node again when we execute (Q.5), 
      // then the comparand pointer will match but the version number will 
      // necessarily be different, since the node now belongs to a future 
      // generation:
      //
      // N->s->C                      ; C was dequeued and requeued
      // 2  3  3
      // ^     ^
      // h     t                      ; Tail version number is different than before
      // 2     3

      Node newTail(last.next, (last.next == &mSentinel) ? 
                              NextVersion(tail.version) : 
                              tail.version);

      if (mTail.compare_exchange_strong(tail, 
                                        newTail, 
                                        std::memory_order_acq_rel)) // (Q.5)
      {
        // We have succeeded in swinging the tail.
        //
        // - If the node pointed to by the tail is a regular node then we've now 
        // finished queueing it:
        //
        // A->s->B->C->N            ; N is now fully queued
        // 1  2  2  2  2
        // ^           ^
        // h           t
        // 2           2
        //
        // If this was our node then we'll exit the loop on the next iteration. 
        // Otherwise we've just helped someone else complete their operation and 
        // will loop again to retry ours.
        //
        // - If the node is the sentinel then it's not considered fully bumped 
        // until we swing the head in (Q.3) during the next iteration.
        //
        // s->B->C->s
        // 2  2  2  2
        // ^        ^
        // h        t               ; The head hasn't been updated yet
        // 2        2

        tail = newTail;
      }
    }
  }

  AtomicNode* DequeueNode()
  {
    AtomicNode* result = 0;

    // Capture the head to figure out what the first node is. The first node can be a
    // regular node or it can be the sentinel:
    //
    // - Regular node:
    //
    // A->s->B->C
    // 1  2  2  2
    // ^        ^
    // h        t
    // 2        2
    //
    // - Sentinel:
    //
    // s->B->C->D
    // 2  2  2  2
    // ^        ^
    // h        t
    // 2        2

    Node head = mHead.load(std::memory_order_acquire); // (D.1)

    // Loop until we've either successfully dequeued an element or the queue is empty.

    while (!result)
    {
      // We have captured the current head and know what the first node is.
      // 
      // - If the first node is a regular node, another consumer thread could 
      // dequeue it before us, updating the head in (D.3) in the process. We'll 
      // detect that in (D.3) too because our CAS there will fail with a comparand 
      // mismatch, which could be due to two reasons. First, if another consumer 
      // has dequeued that node then it must have updated the head to point to some
      // other node, so the node pointer in the comparand will differ:
      //
      // A  B->s->C                   ; A was dequeued by someone else
      //    1  2  2
      //    ^     ^
      //    h     t                   ; The head no longer points to A
      //    2     2
      //
      // On the other hand, if the node was dequeued and then later requeued, and it
      // happens to be the first node on the list again when we execute (D.3), then 
      // the comparand pointer will match but the head version number will 
      // necessarily be different, since the head must have crossed the sentinel 
      // at least once to reach the requeued node:
      //
      // A->s->D->E                   ; A was dequeued in gen 1 and readded in gen 2
      // 2  3  3  3
      // ^        ^
      // h        t
      // 3        3                   ; The head has a different version than before
      //
      // In both these cases we'll simply loop again to retry the operation.
      //
      // - If the first node is the sentinel and it's currently in the process of 
      // being bumped to the end of the list, other producer or consumer threads 
      // could update the head in (Q.3) during the final step of the sentinel bump 
      // operation before we get a chance to do it ourselves. In that case, we will 
      // enter QueueNode to bump the sentinel but will end up doing nothing because 
      // the work will have already been done:
      // 
      // s  B->C->D->s                ; The sentinel has already been bumped
      //    2  2  2  2
      //    ^        ^
      //    h        t
      //    3        3

      // Now capture the contents of the first node to figure out what the second 
      // node is:
      //
      // A->B->s->C                   ; B is the second node
      // 1  1  2  2
      // ^        ^
      // h        t
      // 2        2
      //
      // Note that the first node could have been dequeued by another thread as 
      // we're about to access its storage. We assume that storage for dequeued 
      // nodes is not freed until this thread and any other thread holding a pointer
      // to them is done accessing it:
      //
      // A  B->s->C                   ; A was dequeued by someone else
      //    1  2  2
      //    ^     ^
      //    h     t
      //    2     2

      Node first = head.next->load(std::memory_order_acquire); // (D.2)

      // Check whether the first node is a regular node or whether it's the 
      // sentinel. If it's a regular node we can go ahead and attempt the 
      // dequeueing.

      if (head.next != &mSentinel)
      {
        // The first node is a regular node.
        // We now attempt to swing the head to acquire ownership of the first 
        // node, which if successful means that we're allowed to dequeue it. 
        // If the CAS fails then another consumer has dequeued the node before 
        // us and we'll loop again to retry with a fresh copy of the head.

        Node newHead(first.next, head.version);

        if (mHead.compare_exchange_strong(head, 
                                          newHead, 
                                          std::memory_order_acq_rel)) // (D.3)
        {
          // We have acquired ownership of the first node, so we can return it.

          Node emptyNode(nullptr, 0);
          head.next->store(emptyNode, std::memory_order_release); // (D.4)

          result = head.next;
        }
      }
      else
      {
        // The first node is the sentinel.
        // Check whether the queue is empty or whether there are elements after
        // the sentinel. We can perform this check by just comparing the 
        // sentinel's version number with the head's. An empty queue has a 
        // sentinel with a lower version number than the head:
        //
        //  s
        //  0
        // ^ ^
        // h t
        // 1 1
        //
        // On the other hand, a non-empty queue has sentinel and head with 
        // identical version numbers. This is because during the linking of a 
        // new node to the right of the sentinel in (Q.4) the sentinel's version 
        // number must have been updated to match:
        //
        // s->A
        // 1  1
        // ^  ^
        // h  t
        // 1  1

        if (first.version != head.version)
        {
          // The queue is empty, so we return null.
          break;
        }

        // The queue is non-empty but the sentinel is at the front of the list.
        // Before we're allowed to dequeue any elements we need to bump the 
        // sentinel to the end of the queue. This operation is almost identical 
        // to queueing a regular node, so we let QueueNode handle all the details.
        // We must pass the current version number so that QueueNode can detect 
        // when the sentinel bump is complete.

        QueueNode(&mSentinel, head.version);

        head = mHead.load(std::memory_order_acquire); // (D.5)
      }
    }

    return result;
  }

  static uintptr_t NextVersion(uintptr_t version)
  {
    return version != UINTPTR_MAX ? version + 1 : uintptr_t(1);
  }

  static uintptr_t PreviousVersion(uintptr_t version)
  {
    return version != 1 ? version - 1 : UINTPTR_MAX;
  }

  T* ToElement(AtomicNode* node) const
  {
    uintptr_t offset = uintptr_t(&(static_cast<T*>(0)->*mNodeMember));

    return reinterpret_cast<T*>(uintptr_t(node) - offset);
  }

  AtomicNode* ToNode(const T& element) const
  {
    return const_cast<AtomicNode*>(&(element.*mNodeMember));
  }

private:
  __declspec(align(64)) NodeMemberPointer mNodeMember; // Shared, read-only
  __declspec(align(64)) AtomicNode mSentinel;          // Shared, read-write
  __declspec(align(64)) AtomicNode mHead;              // Mostly consumer-only
  __declspec(align(64)) AtomicNode mTail;              // Mostly producer-only
};
\$\endgroup\$
  • \$\begingroup\$ Any Benchmarks on this? \$\endgroup\$ – Filipe Santos Jun 27 '13 at 15:06
  • 1
    \$\begingroup\$ Filipe, now that I'm a bit more confident about the correctness of this implementation I ran some micro-benchmarks which you can see here. In short, the queue as shown above is decently fast, with some added spinwaits on retries it becomes quite fast, and with a few further tweaks to remove seemingly unnecessary atomic ops (Q.2 and the relaxed store in QueueNode, and D.3 in DequeueNode when dequeuing a regular node) it is the second-fastest queue I've found. Here's hoping it's correct. \$\endgroup\$ – Javier Blazquez Jun 30 '13 at 21:58
  • \$\begingroup\$ Thanks for the results. In comparison to dimitris queue your queue looks really promissing. But why do you get less throughput while increasing the Thread count? Am i missing something? \$\endgroup\$ – Filipe Santos Jul 2 '13 at 13:45
  • 1
    \$\begingroup\$ That's because the benchmark measures throughput under maximum contention, where threads do nothing except queueing and dequeueing elements (no work done per element). It's not a real world scenario, but I think it's very useful to compare how lock free implementations handle contention since that's what they're really good at. If work were done per element you would see throughput increase as threads increase. Dmitry's queue is pretty amazing, so I'd suggest checking that one out if you want something to use in production. It has slightly different properties than mine though. \$\endgroup\$ – Javier Blazquez Jul 3 '13 at 16:34
  • \$\begingroup\$ Do you have a simple usage example? I've tried to put together a quick test but am afraid I am missing something obvious. \$\endgroup\$ – user34333 Jan 3 '14 at 0:21
3
+50
\$\begingroup\$

I just have some stylistic things for a start:

  • Since your custom types are in PascalCase, your function names should instead be in camelCase or in snake_case. This isn't a requirement, but it makes it easier to distinguish between them.

  • The naming convention used for the private members looks like a form of Hungarian Notation, which is quite disliked in strongly-typed languages such as C++. It is more common to prefix members with m_ instead, though you may not need to do that here.

  • Some functions such as NextVersion(), PreviousVersion(), and ToElement() don't modify the argument, so those arguments should be const& in order to avoid a copy.

\$\endgroup\$

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.