Atomiques et lecture-modification-écriture

Une mise à jour perdue arrive parce qu'une lecture et une écriture sont deux tours séparés par un intervalle. Une opération atomique ferme cet intervalle : la lecture, la modification et l'écriture se font en une seule étape indivisible. Avec elle, le compteur est juste sous tous les ordonnancements : 4 ordonnancements au lieu de 28, et aucun n'est faux.

Dans notre machine, un tour va d'un yield au suivant, donc rien de ce qui est écrit entre deux yield ne peut être entrelacé. Une fonction qui ne contient aucun yield se comporte donc comme une opération atomique. La plus courante est fetch-and-add : elle lit l'ancienne valeur, écrit old + n et renvoie old, le tout dans un seul tour. Un thread compteur qui l'appelle n'a qu'un yield pour tout l'incrément au lieu de deux, si bien que les deux threads ne peuvent plus lire la même ancienne valeur.

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}, [])

Les deux ordres finissent à {'x': 2} et aucun thread n'est resté vivant. Avec un tour par incrément, le programme n'a que les deux ordonnancements [0, 1] et [1, 0] (4 dans l'étude, pour 1 et 3 incréments), et aucun ne peut perdre une mise à jour.

Fetch-and-add ne sait qu'ajouter. Pour tout le reste il faut l'outil général, comparer-et-échanger (CAS). Le motif est une boucle : lire l'ancienne valeur, en calculer la nouvelle, et demander au CAS de l'installer seulement si personne n'a changé la valeur entre-temps. Si le CAS échoue, quelqu'un est passé avant, donc on relit et on réessaie. Ci-dessous, deux compteurs CAS : quand le thread 1 lit avant que le thread 0 ait écrit, son CAS échoue une fois, il compte un réessai, et la seconde tentative réussit. La mise à jour n'est jamais perdue, car une écriture périmée est refusée.

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}, [])

Après [0, 1, 0, 1], le dictionnaire partagé est {'x': 1, 'retries': 1} et le thread 1 est encore vivant : il avait lu 0, mais le thread 0 avait déjà rangé 1, donc son comparer-et-échanger a été refusé. En donnant au thread 1 ses deux tours suivants (relire, réessayer), on finit à x = 2. Notez le prix : le perdant a fait le travail deux fois. Fetch-and-add ne réessaie jamais ; le CAS réessaie autant de fois que les threads se heurtent.

L'explorateur du sujet précédent supposait que chaque thread prend un nombre fixe de tours, lengths. Une boucle de réessai casse cela : un thread qui perd une course prend des tours supplémentaires, et leur nombre dépend de l'ordonnancement. La solution est de parcourir les ordonnancements comme un arbre. À partir d'un préfixe de tours, run_prefix rejoue le programme et dit quels threads sont encore vivants ; s'il n'y en a aucun, l'ordonnancement est complet et son état final est compté, sinon on se ramifie sur chacun d'eux. Cette recherche en profondeur visite chaque ordonnancement complet exactement une fois. values_of lit ensuite un nom partagé dans tous les résultats.

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]

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]

Les deux réparations sont justes sous tous les ordonnancements, et elles diffèrent par le coût. Fetch-and-add fait le travail en un tour par incrément : 4 ordonnancements. Le compteur CAS a 144 ordonnancements, car une collision ajoute des tours, et dans le pire un thread réessaie 3 fois. C'est le compromis du code sans verrou : personne n'attend un verrou, mais le gagnant de chaque course oblige le perdant à refaire son travail.

Le labo suit le cours : fetch-and-add, comparer-et-échanger, un thread compteur bâti sur chacun, l'explorateur qui gère des threads à nombre de tours variable, un utilitaire qui lit les valeurs d'un nom sur tous les résultats et l'étude de trois compteurs sur les jours de hausse de SPY.

• Fonction · questionS'appuie sur
• fetch_add(shared, name, n)Ajouter n, et qu'y avait-il avant ?De l'arithmétique simple
• compare_and_swap(shared, name, expected, new)La valeur valait-elle encore ce que j'ai lu ?De l'arithmétique simple
• atomic_increments(k)Un compteur sans intervalle ?fetch_add
• cas_increments(k)Un compteur qui réessaie au lieu de perdre ?compare_and_swapfetch_add
• explore_all(make_threads, shared)Tous les résultats, quel que soit le nombre de tours ?run_prefix
• values_of(outcomes, name)Quelles valeurs ce nom prend-il ?De l'arithmétique simple
• study(bars)Comment les trois compteurs se comparent-ils sur SPY ?explore_allvalues_of

Sujet suivant : nous avons supposé que chaque thread voit chaque écriture immédiatement et dans l'ordre. Un vrai processeur ne le promet pas. Le modèle mémoire : acquire et release ajoute un tampon d'écriture à la machine, montre un lecteur qui voit un drapeau avant les données qu'il protège, et le répare avec l'ordre le moins coûteux.