Ein verlorenes Update entsteht, weil Lesen und Schreiben zwei Züge mit einer Lücke dazwischen sind. Eine atomare Operation schließt die Lücke: Lesen, Ändern und Schreiben geschehen in einem unteilbaren Schritt. Damit stimmt der Zähler unter jedem Schedule: 4 Schedules statt 28, und keiner ist falsch.
Ein Zug, keine Lücke
In unserer Maschine reicht ein Zug von einem yield zum nächsten, also kann nichts, was zwischen zwei yield steht, verzahnt werden. Eine Funktion ohne yield verhält sich daher wie eine atomare Operation. Die häufigste ist Fetch-and-Add: Sie liest den alten Wert, schreibt old + n und gibt old zurück, alles in einem Zug. Ein Zähl-Thread, der sie aufruft, hat nur ein yield für das ganze Inkrement statt zwei, sodass beide Threads nicht mehr denselben alten Wert lesen können.
def fetch_add(shared, name, n): old = shared.get(name, 0) shared[name] = old + n return olddef atomic_increments(k): def thread(shared): for _ in range(k): yield fetch_add(shared, "x", 1) return threadtwo = lambda shared: [atomic_increments(1)(shared), atomic_increments(1)(shared)]for prefix in ([0, 1], [1, 0]): print(prefix, run_prefix(two, prefix))
[0, 1] ({'x': 2}, [])
[1, 0] ({'x': 2}, [])Beide Reihenfolgen enden bei {'x': 2}, und kein Thread bleibt am Leben. Mit einem Zug pro Inkrement hat das Programm nur die beiden Schedules [0, 1] und [1, 0] (4 in der Studie, bei 1 und 3 Inkrementen), und keiner kann ein Update verlieren.
Compare-and-Swap und die Wiederholungsschleife
Fetch-and-Add kann nur addieren. Für alles andere braucht man das allgemeine Werkzeug, Compare-and-Swap (CAS). Das Muster ist eine Schleife: den alten Wert lesen, daraus den neuen berechnen und CAS bitten, ihn nur zu setzen, wenn inzwischen niemand den Wert geändert hat. Schlägt CAS fehl, war jemand schneller, also liest man neu und versucht es wieder. Unten zwei CAS-Zähler: Liest Thread 1, bevor Thread 0 geschrieben hat, schlägt sein CAS einmal fehl, er zählt einen Wiederholversuch, und der zweite Versuch gelingt. Das Update geht nie verloren, weil ein veraltetes Schreiben abgelehnt wird.
def compare_and_swap(shared, name, expected, new): if shared.get(name, 0) != expected: return False shared[name] = new return Truedef cas_increments(k): def thread(shared): for _ in range(k): while True: yield old = shared.get("x", 0) yield if compare_and_swap(shared, "x", old, old + 1): break fetch_add(shared, "retries", 1) return threadcas = lambda shared: [cas_increments(1)(shared), cas_increments(1)(shared)]print(run_prefix(cas, [0, 1, 0, 1]))print(run_prefix(cas, [0, 1, 0, 1, 1, 1]))
({'x': 1, 'retries': 1}, [1])
({'x': 2, 'retries': 1}, [])Nach [0, 1, 0, 1] ist das gemeinsame Dictionary {'x': 1, 'retries': 1}, und Thread 1 lebt noch: Er las 0, aber Thread 0 hatte schon 1 gespeichert, sein Compare-and-Swap wurde also abgelehnt. Gibt man Thread 1 seine zwei nächsten Züge (neu lesen, neu tauschen), endet es bei x = 2. Beachten Sie den Preis: Der Verlierer hat die Arbeit doppelt gemacht. Fetch-and-Add wiederholt nie; CAS wiederholt so oft, wie Threads kollidieren.
Erkunden, wenn die Zahl der Züge variiert
Der Explorer des letzten Themas nahm an, dass jeder Thread eine feste Anzahl von Zügen nimmt, lengths. Eine Wiederholungsschleife bricht das: Ein Thread, der ein Rennen verliert, nimmt zusätzliche Züge, und wie viele, hängt vom Schedule ab. Die Lösung ist, die Schedules als Baum zu durchsuchen. Von einem Präfix von Zügen aus spielt run_prefix das Programm erneut ab und sagt, welche Threads noch leben; lebt keiner mehr, ist der Schedule vollständig und sein Endzustand wird gezählt, sonst verzweigt man auf jeden. Diese Tiefensuche besucht jeden vollständigen Schedule genau einmal. values_of liest danach einen gemeinsamen Namen aus allen Ergebnissen.
def explore_all(make_threads, shared=None): outcomes = {} def visit(prefix): final, alive = run_prefix(make_threads, prefix, shared) if not alive: key = tuple(sorted(final.items())) outcomes[key] = outcomes.get(key, 0) + 1 return for tid in alive: visit(prefix + [tid]) visit([]) return outcomesdef values_of(outcomes, name): return sorted({dict(key).get(name, 0) for key in outcomes})print(explore_all(cas))racy = lambda shared: [increments(1)(shared), increments(1)(shared)]print(values_of(explore_all(racy), "x"), values_of(explore_all(cas), "x"))
{(('x', 2),): 2, (('retries', 1), ('x', 2)): 4}
[1, 2] [2]In dieser Maschine sieht jeder Thread jede Operation gleichzeitig und in Reihenfolge. Ein echter Prozessor verspricht das für gewöhnliche Lese- und Schreibzugriffe nicht, und selbst eine atomare Operation lässt die Kerne um die Cache-Zeile streiten. Das nächste Thema entfernt die erste Annahme.
returns = [float(r) for r in load_bars()["close"].pct_change().dropna()][:6]ups = [sum(1 for r in returns[start::2] if r > 0) for start in (0, 1)]for name, factory in (("racy", increments), ("atomic", atomic_increments), ("cas", cas_increments)): outcomes = explore_all(lambda shared, f=factory: [f(k)(shared) for k in ups]) print(name, sum(outcomes.values()), "schedules, x in", values_of(outcomes, "x"))
racy 28 schedules, x in [1, 2, 3, 4] atomic 4 schedules, x in [4] cas 144 schedules, x in [4]
Beide Reparaturen sind unter jedem Schedule korrekt und unterscheiden sich in den Kosten. Fetch-and-Add erledigt die Arbeit in einem Zug pro Inkrement: 4 Schedules. Der CAS-Zähler hat 144 Schedules, weil eine Kollision Züge hinzufügt, und im schlimmsten wiederholt ein Thread 3-mal. Das ist der Kompromiss von sperrfreiem Code: Niemand wartet auf eine Sperre, aber der Gewinner jedes Rennens zwingt den Verlierer, seine Arbeit zu wiederholen.
Eine Operation, die einen gemeinsamen Wert liest, daraus einen neuen berechnet und ihn in einem unteilbaren Schritt zurückschreibt. Fetch-and-Add und Compare-and-Swap sind die beiden, die Sie am häufigsten nutzen; jede sperrfreie Struktur wird aus ihnen gebaut.
Was Sie bauen werden
Das Labor folgt der Lektion: Fetch-and-Add, Compare-and-Swap, ein Zähl-Thread auf Basis von jedem, der Explorer für Threads mit variabler Zugzahl, ein Helfer, der die Werte eines Namens über alle Ergebnisse liest, und die Studie dreier Zähler an den Aufwärtstagen von SPY.
- Funktion · FrageBaut auf
- fetch_add(shared, name, n)n addieren, und was war vorher da?Einfache Arithmetik
- compare_and_swap(shared, name, expected, new)War der Wert noch, was ich las?Einfache Arithmetik
- atomic_increments(k)Ein Zähler ohne Lücke?fetch_add
- cas_increments(k)Ein Zähler, der wiederholt statt zu verlieren?compare_and_swapfetch_add
- explore_all(make_threads, shared)Alle Ergebnisse, egal wie viele Züge?run_prefix
- values_of(outcomes, name)Welche Werte nimmt dieser Name an?Einfache Arithmetik
- study(bars)Wie schneiden die drei Zähler bei SPY ab?explore_allvalues_of
Nächstes Thema: Wir haben angenommen, dass jeder Thread jedes Schreiben sofort und in Reihenfolge sieht. Ein echter Prozessor verspricht das nicht. Das Speichermodell: Acquire und Release fügt der Maschine einen Store-Puffer hinzu, zeigt einen Leser, der ein Flag vor den Daten sieht, die es schützt, und behebt das mit der Ordnung, die am wenigsten kostet.
Lab: Was bewertet wird
7 Aufgaben · 100 Pkt.- Fetch-and-Add5 Pkt.
- Compare-and-Swap10 Pkt.
- Ein Zähler mit Fetch-and-Add10 Pkt.
- Ein Zähler mit Wiederholungsschleife20 Pkt.
- Alle Schedules erkunden20 Pkt.
- Die Werte eines Namens5 Pkt.
- Die Studie mit der SPY-Tabelle30 Pkt.