Warteschlangen mit mehreren Producern und Consumern

Mit einem Producer und einem Consumer hat eine Position einen Schreiber. Mit vielen müssen die Positionen atomar vergeben werden, und ein Ticket genügt nicht: Der Producer, der Position 5 hält, hat sein Element noch nicht geschrieben. Eine Sequenznummer an jeder Zelle sagt, wer an der Reihe ist. Über 2.756 Schedules zweier Producer und 1.600 zweier Consumer bricht diese Warteschlange nie.

Eine beschränkte Warteschlange mit capacity Zellen nutzt zwei Zähler, die nur wachsen. Ein Producer nimmt die nächste Position mit einem Compare-and-Swap auf enq; ein Consumer die nächste mit einem auf deq. Position p liegt in Zelle p % capacity. Das Compare-and-Swap macht die Vergabe sicher: Zwei Producer bekommen nie dieselbe Position. Aber eine Position zu belegen heißt nicht, das Element zu schreiben. Zwischen Tausch und Speichern ist die Zelle belegt und leer, und ein Consumer, der sie jetzt läse, läse Müll. Jede Zelle trägt deshalb eine Sequenznummer, die nur im richtigen Moment weitergeschaltet wird.

def init_queue(capacity):    shared = {"enq": 0, "deq": 0}    for cell in range(capacity):        shared["seq%d" % cell] = cell    return shareddef enqueue(shared, item, capacity):    while True:        yield        pos = shared["enq"]        cell = pos % capacity        diff = shared["seq%d" % cell] - pos        if diff < 0:            return False        if diff > 0:            continue        yield        if compare_and_swap(shared, "enq", pos, pos + 1):            break    yield    shared["data%d" % cell] = item    shared["seq%d" % cell] = pos + 1    shared["put%d" % pos] = item    return Truedef dequeue(shared, capacity):    while True:        yield        pos = shared["deq"]        cell = pos % capacity        diff = shared["seq%d" % cell] - (pos + 1)        if diff < 0:            return None        if diff > 0:            continue        yield        if compare_and_swap(shared, "deq", pos, pos + 1):            break    yield    item = shared.get("data%d" % cell)    shared["seq%d" % cell] = pos + capacity    shared["got%d" % pos] = item    return itemsolo = lambda shared: [producer_thread((7,), 2)(shared), consumer_thread(1, 2)(shared)]for prefix in ([0, 0, 0], [0, 0, 0, 1, 1, 1], [0, 0, 1, 1, 0, 1, 1]):    print(prefix, run_prefix(solo, prefix, init_queue(2))[0])

[0, 0, 0] {'enq': 1, 'deq': 0, 'seq0': 1, 'seq1': 1, 'data0': 7, 'put0': 7}
[0, 0, 0, 1, 1, 1] {'enq': 1, 'deq': 1, 'seq0': 2, 'seq1': 1, 'data0': 7, 'put0': 7, 'got0': 7}
[0, 0, 1, 1, 0, 1, 1] {'enq': 1, 'deq': 0, 'seq0': 1, 'seq1': 1, 'data0': 7, 'put0': 7}

Lesen Sie die drei Läufe. Nach [0, 0, 0] hat der Producer Position 0 belegt (enq: 1), data0 gespeichert und seq0 auf 1 erhöht: Das Element ist bereit. Nach den drei Zügen des Consumers ist got0: 7 und seq0: 2: Die Zelle ist für die nächste Runde frei. Im letzten Lauf schaut der Consumer zwischen Belegung und Veröffentlichung des Producers nach, sieht seq0 = 0 statt 1 und geht mit leeren Händen. Er liest keinen Müll. Er sagt nur „noch nichts bereit“, und ein späterer Versuch findet das Element.

Was muss die Warteschlange jetzt versprechen? Die Producer protokollieren jedes Element unter seiner Position (put3), die Consumer protokollieren, was sie nahmen, unter der Position, die sie belegten (got3). Funktioniert die Warteschlange, enthält jede entnommene Position das dort eingefügte Element (kein mismatch), und die entnommenen Positionen sind 0, 1, 2, ... ohne Lücken oder Wiederholungen (kein gap). Das gibt der Warteschlange ihre Ordnung: die Ordnung der Positionen. violations liest diese beiden Tatsachen aus einem Endzustand, und verify startet den Explorer für beliebig viele Producer und Consumer. Hier ein einzelner Producer mit zwei Elementen gegen zwei Consumer und ein einzelnes Element gegen einen Consumer.

def violations(final):    puts = {int(key[3:]): value for key, value in final.items() if key.startswith("put")}    gots = {int(key[3:]): value for key, value in final.items() if key.startswith("got")}    broken = []    if any(gots[p] != puts.get(p) for p in gots):        broken.append("mismatch")    if sorted(gots) != list(range(len(gots))):        broken.append("gap")    return sorted(broken)def verify(producers, consumers, capacity):    def make_threads(shared):        return [producer_thread(items, capacity)(shared) for items in producers] + [consumer_thread(n, capacity)(shared) for n in consumers]    outcomes = explore_all(make_threads, init_queue(capacity))    broken = {}    pushed = popped = 0    for key, count in outcomes.items():        final = dict(key)        pushed = max(pushed, sum(1 for k in final if k.startswith("put")))        popped = max(popped, sum(1 for k in final if k.startswith("got")))        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(verify([(7,)], [1], 2))print(verify([(7, 8)], [1, 1], 2))

{'schedules': 4, 'broken': {}, 'pushed': 1, 'popped': 1}
{'schedules': 1600, 'broken': {}, 'pushed': 2, 'popped': 2}

Um den Wert zu sehen, nehmen wir sie weg. Der nachlässige Consumer unten belegt eine Position mit dem Compare-and-Swap und liest die Zelle dann sofort, ohne auf seq zu schauen. Gegen einen Producer mit einem einzigen Element hat er 20 Schedules, und in 10 davon belegt er eine Position, bevor der Producer etwas veröffentlicht hat, und liefert None als Element oder das falsche. Die echte Warteschlange hat für dasselbe Programm 4 Schedules, und keiner bricht. Die Sequenznummer ist genau die Prüfung, die ein belegtes Ticket in ein geliefertes Element verwandelt.

def careless_dequeue(shared, capacity):    yield    pos = shared["deq"]    cell = pos % capacity    yield    if not compare_and_swap(shared, "deq", pos, pos + 1):        return None    yield    item = shared.get("data%d" % cell)    shared["got%d" % pos] = item    return itemdef careless_consumer(attempts, capacity):    def thread(shared):        for _ in range(attempts):            yield from careless_dequeue(shared, capacity)    return threadmake = lambda shared: [producer_thread((7,), 2)(shared), careless_consumer(1, 2)(shared)]outcomes = explore_all(make, init_queue(2))broken = sum(n for key, n in outcomes.items() if violations(dict(key)))print(sum(outcomes.values()), "schedules,", broken, "break the invariants")

20 schedules, 10 break the invariants

ticks = [int(round(float(c) * 100)) for c in load_bars()["close"].iloc[:2]]print(ticks)print("two producers:", verify([(ticks[0],), (ticks[1],)], [2], 2))print("two consumers:", verify([tuple(ticks)], [1, 1], 2))

[9369, 9305]
two producers: {'schedules': 2756, 'broken': {}, 'pushed': 2, 'popped': 2}
two consumers: {'schedules': 1600, 'broken': {}, 'pushed': 2, 'popped': 2}

Beide Formen der Warteschlange sind über alle Schedules korrekt, der Explorer ist der Zeuge, während eine Version ohne Sequenzprüfung in der Hälfte ihres (viel kleineren) Raums scheitert. Producer und Consumer warten nie aufeinander: Ein Consumer, der nichts Bereites findet, kehrt einfach zurück, und ein Producer, der die Warteschlange voll findet, gibt False zurück. Das ist Preis und Geschenk einer sperrfreien Warteschlange: Operationen können scheitern, blockieren aber nie.

Das Labor folgt der Lektion: die leere Warteschlange mit ihren Sequenznummern, Enqueue und Dequeue mit Compare-and-Swap-Belegung und Veröffentlichung, die Prüfung der zwei Invarianten, der Verifizierer, der den Explorer für beliebig viele Producer und Consumer startet, und die Studie an zwei SPY-Schlusskursen.

• Funktion · FrageBaut auf
• init_queue(capacity)Wie sieht eine leere Warteschlange aus?Einfache Arithmetik
• enqueue(shared, item, capacity)Wie belegt und veröffentlicht ein Producer?Einfache Arithmetik
• dequeue(shared, capacity)Wie belegt und gibt ein Consumer frei?Einfache Arithmetik
• violations(final)Welche Invarianten bricht ein Endzustand?Einfache Arithmetik
• verify(producers, consumers, capacity)Stimmt die Warteschlange unter jedem Schedule?explore_allviolations
• study(bars)Halten beide Formen bei SPY-Schlusskursen?verify

Nächstes Thema: Sie haben nun die Korrektheit kleiner Programme bewiesen, indem Sie alle Schedules ausgeführt haben. Nebenläufigen Code testen vergleicht das mit den Werkzeugen echter Teams: einem Race-Detektor, der einen Lauf beobachtet und Zugriffe markiert, die wettlaufen könnten, und einem Model Checker, der tut, was Ihr Explorer tut, aber mit den Tricks, die ihn skalieren lassen.