47
\$\begingroup\$

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;
    }
}
\$\endgroup\$
  • \$\begingroup\$ There's a race in getObject()'s head.get().next versus putObject()'s prevTailNode.next = newNode. I wish the codereview site had line numbers. After the tail.getAndSet(newNode) has occurred but before prevTailNode.next assignment has gone trhough, getObject may call head.get() and then read headNode.next which will be null at that moment, resulting in getObject() returning null incorrectly. \$\endgroup\$ – Ron Jan 27 '11 at 14:18
  • \$\begingroup\$ @Ron I talk about how that behavior is strange in my answer, but it is at least thread-safe. \$\endgroup\$ – Craig Jan 27 '11 at 18:10
  • \$\begingroup\$ @Craig Ok I see. getObject returning null is ok. I guess a performance drawback is whether the call to head.compareAndSet in getObject has any guarantees about making forward progress. \$\endgroup\$ – Ron Jan 27 '11 at 19:52
  • 1
    \$\begingroup\$ @Ron That's true. Usually non-blocking structures are written to give better performance than blocking structures, but there's no guarantee that's the case here. Especially since the thread that's doing the CAS operations cannot help other threads make forward progress. \$\endgroup\$ – Craig Jan 27 '11 at 20:45
21
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ Thanks for your comments. I didn't make the value field final to be able to set it to null to avoid a potential memory leak. (Check the last if statement in getObject()). \$\endgroup\$ – Hosam Aly Jan 27 '11 at 13:45
  • \$\begingroup\$ As for the odd behavior, I agree with you, but that was a design decision to simplify the code and improve performance. I don't think it would matter if it's used for simple producer-consumer scenarios. What do you think? \$\endgroup\$ – Hosam Aly Jan 27 '11 at 13:47
  • \$\begingroup\$ @Hosam I do think it's problematic on the consumer side. \$\endgroup\$ – Craig Jan 27 '11 at 18:30
  • \$\begingroup\$ @Hosam: My rule of thumb is this: either final or volatile. If you won't make it final, at least make it volatile. (This is exactly what Craig said, and I want to reinforce it.) \$\endgroup\$ – Chris Jester-Young Jan 27 '11 at 20:26
  • \$\begingroup\$ @Craig: Thanks. Sorry for my late comment, but you probably know how hot it has been in Egypt during the past period. It's true that this queue would require repeated polling and sleeping. This is (IMHO) appropriate in scenarios where a simple message queue-like behavior is needed. Nevertheless, I like your idea about a non-blocking get(). I will think about how to implement it. Thanks again. \$\endgroup\$ – Hosam Aly Feb 13 '11 at 19:06
4
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ Thanks @supercat. Your point about asynchronously-terminated threads is certainly valuable, but I am not sure how your suggestion fixes it. Could you please explain it further with some code? \$\endgroup\$ – Hosam Aly Feb 13 '11 at 19:13
  • \$\begingroup\$ @Hosam: The real last item of the queue is the one whose "next" pointer is null. To add an item to the queue, make the real last item, i.e. the one whose "next" pointer is null, point to the new one. Although the queue keeps a "last-item" pointer, it won't always point to the very last item; it may point to an item which used to be the last item, but isn't any more. As long as code which attempts to add an item is prepared to search for the real last item, all that is required for correctness is that the "last item" pointer point somewhere within the queue. \$\endgroup\$ – supercat Feb 14 '11 at 0:48
  • \$\begingroup\$ @Hosam: Adding an item to the queue does two things: -1- It makes the real last item's "next" pointer point to the new item; -2- It tries to update the queue's "last item" pointer. If there are two simultaneous attempts to do step #1, the CompareExchange of one will succeed and the other will fail. The one whose CompareExchange fails will search for the new end of the queue and add itself after that. Note that the attempt to update the end-of-queue pointer is in a sense optional; if it never got updated, the queue would get slower and slower, but occasional failures are no problem. \$\endgroup\$ – supercat Feb 14 '11 at 0:55
4
\$\begingroup\$

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.

\$\endgroup\$
3
\$\begingroup\$

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;
}
}
\$\endgroup\$
  • \$\begingroup\$ Same problem as the original code... value needs to be volatile. \$\endgroup\$ – Craig Jan 27 '11 at 20:13
  • \$\begingroup\$ I claim it doesn't need to be volatile. Threads writing to and reading from the value field will have encountered memory fences via the AtomicReferences. \$\endgroup\$ – Ron Jan 27 '11 at 21:07
  • \$\begingroup\$ Oh yeah! I need to read JCIP again! \$\endgroup\$ – Craig Jan 27 '11 at 21:42
  • \$\begingroup\$ Thanks. Throwing the NPE is a good idea, and removing the redundant null checks is a welcome addition. But as @Tino pointed out, the assert could fail in a race condition (codereview.stackexchange.com/questions/224/…). \$\endgroup\$ – Hosam Aly Feb 13 '11 at 19:35
2
\$\begingroup\$

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.

\$\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.