Thread-Safe und Lock-Free - Queue-Implementierung

Ja.

• Die Kombination aus flüchtigen Operationen und Compare-and-Swap-Operationen reicht aus, um sicherzustellen, dass die Node-Objekte sicher veröffentlicht werden.
• Der Compare-and-Swap muss volatilein beiden Methoden vor der Zuweisung zu einer Variablen stehen, damit Sie dort gut zurechtkommen. Sie müssen nicht atomar geschehen.

Die Warteschlange zeigt ein merkwürdiges Verhalten. Nehmen wir an, Thread 1 stoppt putObject()nach dem CAS, aber vor der letzten Zuweisung. Der nächste Thread 2 wird putObject()vollständig ausgeführt. Beim nächsten Thread drei Aufrufe getObject(), und es kann keines der ersten beiden Objekte angezeigt werden, obwohl Thread 2 vollständig ausgeführt wurde. Im Speicher wird eine kleine Kette aufgebaut. Erst nach Abschluss von Thread 1 putObject()sind beide Objekte in der Warteschlange sichtbar, was etwas überraschend ist und eine schwächere Semantik aufweist als die meisten nicht blockierenden Datenstrukturen.

Eine andere Sichtweise auf die ungerade API ist, dass es nett ist, eine Nichtblockierung zu haben, put()aber es ist sehr merkwürdig, eine Nichtblockierung zu haben get(). Dies bedeutet, dass die Warteschlange bei wiederholten Abfragen und beim Schlafen verwendet werden muss.

Ich mag deine 'Put'-Routine nicht. Das "ungerade Verhalten", das andere bemerkt haben, bedeutet, dass der Algorithmus einen der Hauptvorteile des "Lock-Free" -Verfahrens verliert: Immunität gegen Prioritätsinversion oder asynchron terminierte Threads. Wenn ein Thread mitten in eine Warteschlangenschreiboperation eingelegt wird, wird die Warteschlange vollständig unterbrochen, es sei denn, bis die Warteschlange ihre Arbeit beendet hat.

Die Lösung besteht darin, CompareAndSet den Zeiger "next" des letzten Knotens zu setzen, bevor der Zeiger "tail" aktualisiert wird. Wenn der "nächste" Zeiger des letzten Knotens nicht null ist, bedeutet dies, dass er nicht mehr der letzte Knoten ist, und man muss Verknüpfungen durchlaufen, um den echten letzten Knoten zu finden. Wiederholen Sie das CompareAndSet darauf und hoffen Sie, dass es immer noch der letzte Knoten ist .

Während die Lösung von OP für mich richtig aussieht, erfindet Rons Verbesserung eine Rennbedingung:

Thread 1 (in getObject ()):

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

T value = next.value;

Thread 1 ist hier ausgesetzt, wurde also noch nicht ausgeführt

next.value = null;

Wir wissen, dass value! = Null und refHead jetzt als nächstes kommt, also refHead.get (). Value! = Null

Thread 2 (in getObject ()):

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

Hier beißt die Behauptung sogar, dass alles wieder in Ordnung ist, nachdem Thread 1 fortgefahren ist.

Ich habe Ihren Code ein wenig angepasst, aber ich denke, Ihr Ansatz ist gut.

Wie ich in dem Kommentar anspielte, gibt es keine Garantie für die Fairness zwischen den Threads compareAndSetting des Kopfes, so dass ein wirklich unglücklicher Thread für eine Weile hängen bleiben könnte, wenn es viele Verbraucher gibt.

Ich denke nicht, dass diese Datenstruktur Nullen enthalten sollte, da es keine Möglichkeit gibt, zwischen dem Erhalten einer Null und dem Abrufen aus einer leeren Warteschlange zu unterscheiden, also werfe ich die NPE.

Ich habe die Logik in getObject ein wenig überarbeitet, um redundante Nullprüfungen zu entfernen.

Und ich habe einige Vars umbenannt, um ungarisch zu sein.

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;
}
}

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.