Thread-Safe and Lock-Free - Queue Implementation

Question

I was trying to create a lock-free queue implementation in Java, mainly for personal learning. The queue should be a general one, allowing any number of readers and/or writers concurrently.

Would you please review it, and suggest any improvements/issues you find?

(Cross-post from StackOverflow)

import java.util.concurrent.atomic.AtomicReference;

public class LockFreeQueue<T> {
    private static class Node<E> {
        final E value;
        volatile Node<E> next;

        Node(E value) {
            this.value = value;
        }
    }

    private AtomicReference<Node<T>> head, tail;

    public LockFreeQueue() {
        // have both head and tail point to a dummy node
        Node<T> dummyNode = new Node<T>(null);
        head = new AtomicReference<Node<T>>(dummyNode);
        tail = new AtomicReference<Node<T>>(dummyNode);
    }

    /**
     * Puts an object at the end of the queue.
     */
    public void putObject(T value) {
        Node<T> newNode = new Node<T>(value);
        Node<T> prevTailNode = tail.getAndSet(newNode);
        prevTailNode.next = newNode;
    }

    /**
     * Gets an object from the beginning of the queue. The object is removed
     * from the queue. If there are no objects in the queue, returns null.
     */
    public T getObject() {
        Node<T> headNode, valueNode;

        // move head node to the next node using atomic semantics
        // as long as next node is not null
        do {
            headNode = head.get();
            valueNode = headNode.next;
            // try until the whole loop executes pseudo-atomically
            // (i.e. unaffected by modifications done by other threads)
        } while (valueNode != null && !head.compareAndSet(headNode, valueNode));

        T value = (valueNode != null ? valueNode.value : null);

        // release the value pointed to by head, keeping the head node dummy
        if (valueNode != null)
            valueNode.value = null;

        return value;
    }
}

Answer 1

Yes.

• The combination of volatile and compare-and-swap operations is enough to make sure that the Node objects are safely published.
• The compare-and-swap must be before the assignment to a volatile variable in both methods, so you're fine there. They do not need to happen atomically.

The queue exhibits odd behavior. Let's say thread 1 stops in putObject() after the CAS but before the last assignment. Next thread 2 executes putObject() in its entirety. Next thread three calls getObject(), and it cannot see either of the first two objects, even though thread 2 is completely done. There's a small chain being built up in-memory. Only after thread 1 completes putObject() are both objects visible on the queue, which is somewhat surprising, and has weaker semantics than most non-blocking data structures.

Another way of looking at the odd API is that it's nice to have a non-blocking put() but it's very strange to have a non-blocking get(). It means that the queue must be used with repeated polling and sleeping.

Answer 2

I don't like your 'put' routine. The 'odd behavior' others have noticed means the algorithm loses one of the main advantages of being 'lock-free': immunity from priority inversion or asynchronously-terminated threads. If a thread gets waylaid in the middle of a queue-write operation, the queue will be completely broken unless or until the queue finishes its work.

The solution is to CompareAndSet the last node's "next" pointer before updating the "tail" pointer; if the "last node"'s "next" pointer is non-null, that means it's not the last node anymore and one must traverse links to find the real last node, repeat the CompareAndSet on it, and hope it's still the last node.

Answer 3

While OP's solution looks right for me, Ron's improvement invents a race condition:

Thread 1 (in getObject()):

} while (!refHead.compareAndSet(head, next));

T value = next.value;

Thread 1 is suspended here, so it did not yet execute

next.value = null;

We know that value!=null and refHead now is next, so refHead.get().value!=null

Thread 2 (in getObject()):

    head = refHead.get();
    assert head.value == null;

Here the assert bites even that everything is will be ok again after Thread 1 continues.

Answer 4

I tweaked your code a little but I think your approach is good.

As I alluded to in the comment, there's no guarantee of fairness between threads compareAndSetting the head, so a really unlucky thread could be stuck for a while if there are a lot of consumers.

I don't think this data structure should hold nulls, since there's no way to distinguish between getting a null or getting from an empty queue, so I throw the NPE.

I refactored the logic a bit in getObject to remove redundant null checks.

And I renamed some vars to be hungarianlike.

import java.util.concurrent.atomic.AtomicReference;

public class LockFreeQueue<T> {
private static class Node<E> {
    E value;
    volatile Node<E> next;

    Node(E value) {
        this.value = value;
    }
}

private final AtomicReference<Node<T>> refHead, refTail;
public LockFreeQueue() {
    // have both head and tail point to a dummy node
    Node<T> dummy = new Node<T>(null);
    refHead = new AtomicReference<Node<T>>(dummy);
    refTail = new AtomicReference<Node<T>>(dummy);
}

/**
 * Puts an object at the end of the queue.
 */
public void putObject(T value) {
    if (value == null) throw new NullPointerException();

    Node<T> node = new Node<T>(value);
    Node<T> prevTail = refTail.getAndSet(node);
    prevTail.next = node;
}

/**
 * Gets an object from the beginning of the queue. The object is removed
 * from the queue. If there are no objects in the queue, returns null.
 */
public T getObject() {
    Node<T> head, next;

    // move head node to the next node using atomic semantics
    // as long as next node is not null
    do {
        head = refHead.get();
        next = head.next;
        if (next == null) {
            // empty list
            return null;
        }
        // try until the whole loop executes pseudo-atomically
        // (i.e. unaffected by modifications done by other threads)
    } while (!refHead.compareAndSet(head, next));

    T value = next.value;

    // release the value pointed to by head, keeping the head node dummy
    next.value = null;

    return value;
}
}

Answer 5

Similar structure and operations for it to get lock free multi-producer single-consumer (MPSC) queue described by Dmitry Vyukov.

I have used it as an internal queue for Scalaz actors to work with it in MPSC mode.