Ein Ringpuffer für einen Producer und einen Consumer

Die einfachste sperrfreie Warteschlange braucht gar kein Compare-and-Swap. Geben Sie jeder Position genau einen Schreiber: Nur der Producer schreibt den Tail, nur der Consumer den Head. Ein veralteter Lesezugriff auf die Position der Gegenseite ist immer sicher, sodass ein Ringpuffer Ticks ohne Sperre zwischen zwei Threads weitergeben kann. Der Explorer beweist das über 291 Schedules und zeigt, dass es in 458 von 1.072 bricht, sobald ein zweiter Producer dazukommt.

Ein Ringpuffer hat zwei Zähler, die nur wachsen. Der Producer schreibt Slot tail % capacity und veröffentlicht dann durch Erhöhen von tail; der Consumer liest Slot head % capacity und gibt ihn dann durch Erhöhen von head frei. Der Puffer ist voll, wenn tail - head == capacity, und leer, wenn head == tail. Warum ist das ohne Sperre sicher? Der Producer ist der einzige Schreiber von tail, kann also eine lokale Kopie führen. Er liest head, um Platz zu suchen, und ist dieser Wert veraltet, ist der veraltete Head kleiner als der echte, der Producer hält den Puffer also für voller, als er ist, und verwirft schlimmstenfalls ein Element, überschreibt aber nie einen ungelesenen Slot. Auch der veraltete tail des Consumers ist kleiner, er sieht den Puffer schlimmstenfalls leer. Das ist das Single-Writer-Prinzip.

def initial_state():    return {"head": 0, "tail": 0, "pushed": (), "popped": ()}def ring_producer(items, capacity):    def thread(shared):        tail = 0        for item in items:            yield            head = shared.get("head", 0)            if tail - head < capacity:                yield                shared["slot%d" % (tail % capacity)] = item                yield                tail += 1                shared["tail"] = tail                shared["pushed"] = shared.get("pushed", ()) + (item,)    return threaddef ring_consumer(attempts, capacity):    def thread(shared):        head = 0        for _ in range(attempts):            yield            tail = shared.get("tail", 0)            if head != tail:                yield                item = shared.get("slot%d" % (head % capacity))                yield                head += 1                shared["head"] = head                shared["popped"] = shared.get("popped", ()) + (item,)    return threadtwo = lambda shared: [ring_producer((7, 8), 2)(shared), ring_consumer(2, 2)(shared)]for prefix in ([0, 0, 0], [0, 0, 0, 1, 1, 1], [0, 0, 0, 0, 0, 0, 1, 1, 1, 1]):    print(prefix, run_prefix(two, prefix, initial_state())[0])

[0, 0, 0] {'head': 0, 'tail': 1, 'pushed': (7,), 'popped': (), 'slot0': 7}
[0, 0, 0, 1, 1, 1] {'head': 1, 'tail': 1, 'pushed': (7,), 'popped': (7,), 'slot0': 7}
[0, 0, 0, 0, 0, 0, 1, 1, 1, 1] {'head': 1, 'tail': 2, 'pushed': (7, 8), 'popped': (7,), 'slot0': 7, 'slot1': 8}

Drei Züge des Producers schieben den ersten Tick hinein: Er liest den Head, legt das Element in slot0 und veröffentlicht tail: 1. Nach drei Zügen des Consumers ist head: 1 und popped: (7,). In der letzten Zeile hat der Producer beide Elemente geschoben (slot0: 7, slot1: 8), und der Consumer hat das erste genommen; Thread 1 lebt noch, weil er einen Versuch übrig hat.

Was muss eine Warteschlange versprechen? Zwei Dinge. Die entnommenen Elemente sind die eingefügten, in derselben Reihenfolge und ohne Lücken: popped ist ein Präfix von pushed (die FIFO-Eigenschaft). Und es wird nichts entnommen, was nie eingefügt wurde: len(popped) <= len(pushed). Die Threads protokollieren, was sie einfügen und entnehmen, genau in dem Zug, der die Operation beendet, sodass violations beides an jedem Endzustand prüfen kann. verify führt dann das ganze Programm unter jedem Schedule mit explore_all aus und zählt die Schedules, die jede Invariante brechen. Ein leeres broken-Dictionary über alle Schedules ist ein Beweis für diese Programmgröße.

def violations(final):    pushed, popped = tuple(final.get("pushed", ())), tuple(final.get("popped", ()))    broken = []    if popped != pushed[: len(popped)]:        broken.append("order")    if len(popped) > len(pushed):        broken.append("count")    return sorted(broken)def verify(producers, attempts, capacity):    def make_threads(shared):        return [ring_producer(items, capacity)(shared) for items in producers] + [ring_consumer(attempts, capacity)(shared)]    outcomes = explore_all(make_threads, initial_state())    broken = {}    pushed = popped = 0    for key, count in outcomes.items():        final = dict(key)        pushed = max(pushed, len(final.get("pushed", ())))        popped = max(popped, len(final.get("popped", ())))        for name in violations(final):            broken[name] = broken.get(name, 0) + count    return {"schedules": sum(outcomes.values()), "broken": dict(sorted(broken.items())), "pushed": pushed, "popped": popped}print(violations({"pushed": (7, 8), "popped": (8,)}))print(verify([(7, 8)], 2, 2))print(verify([(7, 8)], 2, 1))

['order']
{'schedules': 101, 'broken': {}, 'pushed': 2, 'popped': 2}
{'schedules': 25, 'broken': {}, 'pushed': 2, 'popped': 2}

Ein Producer mit zwei Elementen in einen Puffer mit zwei Slots, gelesen von einem Consumer mit zwei Versuchen, hat 101 Schedules, und keiner bricht eine Invariante. Bei einem Puffer mit nur einem Slot muss der Producer ein Element verwerfen, wenn der Consumer langsam ist, bleibt aber korrekt: 25 Schedules, keine Verletzung. Die entnommene Liste (8,), nachdem (7, 8) eingefügt wurde, ist erwartungsgemäß eine Verletzung der Reihenfolge.

Das Sicherheitsargument stützt sich auf eine Tatsache: tail hat einen Schreiber. Fügen Sie einen zweiten Producer hinzu, und zwei Threads führen je ihre eigene Vorstellung vom Tail. Beide lesen denselben head, berechnen denselben Slot, und das zweite Schreiben überschreibt das erste. Der Explorer findet es: 458 der 1.072 Schedules zweier Producer (je ein Tick) enden mit einem popped, das kein Präfix von pushed ist. Die Abhilfe ist kein Flicken an diesem Puffer, sondern eine andere Struktur, eine Multi-Producer-Warteschlange, die spätere Themen mit Compare-and-Swap bauen.

print(verify([(7,), (8,)], 2, 2))

{'schedules': 1072, 'broken': {'order': 458}, 'pushed': 2, 'popped': 1}

ticks = [int(round(float(c) * 100)) for c in load_bars()["close"].iloc[:4]]print(ticks)print("one producer: ", verify([ticks[:2]], 3, 2))print("two producers:", verify([ticks[:1], ticks[1:2]], 2, 2))

[9369, 9305, 9354, 9349]
one producer:  {'schedules': 291, 'broken': {}, 'pushed': 2, 'popped': 2}
two producers: {'schedules': 1072, 'broken': {'order': 458}, 'pushed': 2, 'popped': 1}

Die ganze Lektion steckt in zwei Zahlen: Ein Schreiber pro Position genügt für die Korrektheit auf 291 Schedules, und zwei Schreiber auf derselben Position liegen bei 43 % von 1.072 falsch. Der Explorer hat nicht geraten; er hat jede Verzahnung ausgeführt.

Das Labor folgt der Lektion: der leere Ring, die Producer- und Consumer-Threads, die Prüfung der zwei Invarianten, der Verifizierer, der den Explorer startet, und die Studie, die SPY-Schlusskurse mit einem und mit zwei Producern durch den Ring schickt.

• Funktion · FrageBaut auf
• initial_state()Wie sieht ein leerer Ring aus?Einfache Arithmetik
• ring_producer(items, capacity)Wie schiebt der Producer?Einfache Arithmetik
• ring_consumer(attempts, capacity)Wie entnimmt der Consumer?Einfache Arithmetik
• violations(final)Welche Invarianten bricht dieser Zustand?Einfache Arithmetik
• verify(producers, attempts, capacity)Stimmt der Ring unter jedem Schedule?explore_allviolations
• study(bars)Was bricht mit einem zweiten Producer?verify

Nächstes Thema: Der Single-Writer-Trick hat eine Grenze, und Compare-and-Swap auch. Das ABA-Problem und getaggte Zeiger baut einen sperrfreien Stack auf Compare-and-Swap, zeigt einen Schedule, in dem der Tausch gelingt, obwohl sich der Stack darunter verändert hat, und behebt es mit einem Versions-Tag.