Files à plusieurs producteurs et plusieurs consommateurs

Avec un producteur et un consommateur, une position a un seul écrivain. Avec plusieurs, les positions doivent être distribuées atomiquement, et un ticket ne suffit pas : le producteur qui détient la position 5 n'a pas encore écrit son élément. Un numéro de séquence sur chaque case dit à qui est le tour de la case. Sur 2 756 ordonnancements de deux producteurs et 1 600 de deux consommateurs, cette file ne se casse jamais.

Une file bornée de capacity cases utilise deux compteurs qui ne font que croître. Un producteur prend la position suivante par un comparer-et-échanger sur enq ; un consommateur prend la suivante par un comparer-et-échanger sur deq. La position p vit dans la case p % capacity. Le comparer-et-échanger rend les réservations sûres : deux producteurs n'obtiennent jamais la même position. Mais réserver une position n'est pas écrire l'élément. Entre l'échange et l'écriture, la case est réservée et vide, et un consommateur qui la lirait maintenant lirait n'importe quoi. Chaque case porte donc un numéro de séquence qui n'avance qu'au bon moment.

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}

Lisez les trois exécutions. Après [0, 0, 0], le producteur a réservé la position 0 (enq: 1), rangé data0 et mis seq0 à 1 : l'élément est prêt. Après les trois tours du consommateur, got0: 7 et seq0: 2 : la case est libre pour le tour suivant. Dans la dernière exécution, le consommateur regarde entre la réservation du producteur et sa publication, voit seq0 = 0 au lieu de 1, et repart les mains vides. Il ne lit pas n'importe quoi. Il dit simplement « rien de prêt pour l'instant », et l'élément sera trouvé par une tentative ultérieure.

Que doit promettre la file maintenant ? Les producteurs consignent chaque élément sous sa position (put3), les consommateurs consignent ce qu'ils ont pris sous la position réservée (got3). Si la file fonctionne, chaque position sortie contient l'élément qui y a été mis (pas de mismatch), et les positions sorties sont 0, 1, 2, ... sans trou ni répétition (pas de gap). Cela donne à la file son ordre : l'ordre des positions. violations lit ces deux faits dans un état final et verify lance l'explorateur pour un nombre quelconque de producteurs et de consommateurs. Ici, un seul producteur de deux éléments contre deux consommateurs, et un seul élément contre un consommateur.

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}

Pour mesurer sa valeur, retirons-le. Le consommateur négligent ci-dessous réserve une position avec le comparer-et-échanger puis lit aussitôt la case, sans regarder seq. Contre un producteur d'un seul élément, il a 20 ordonnancements, et dans 10 d'entre eux il réserve une position avant que le producteur ait rien publié et renvoie None comme élément, ou le mauvais élément. La vraie file a 4 ordonnancements pour le même programme et aucun ne casse. Le numéro de séquence est exactement le contrôle qui transforme un ticket réservé en élément livré.

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}

Les deux formes de la file sont justes sur tous les ordonnancements, l'explorateur en témoin, tandis qu'une version sans le contrôle de séquence échoue dans la moitié de son espace (bien plus petit). Producteurs et consommateurs ne s'attendent jamais : un consommateur qui ne trouve rien de prêt revient simplement, et un producteur qui trouve la file pleine renvoie False. C'est le prix et le cadeau d'une file sans verrou : les opérations peuvent échouer, mais ne bloquent jamais.

Le labo suit le cours : la file vide avec ses numéros de séquence, enqueue et dequeue avec réservation par comparer-et-échanger et publication, la vérification des deux invariants, le vérificateur qui lance l'explorateur pour un nombre quelconque de producteurs et de consommateurs, et l'étude sur deux cours SPY.

• Fonction · questionS'appuie sur
• init_queue(capacity)À quoi ressemble une file vide ?De l'arithmétique simple
• enqueue(shared, item, capacity)Comment un producteur réserve-t-il et publie-t-il ?De l'arithmétique simple
• dequeue(shared, capacity)Comment un consommateur réserve-t-il et libère-t-il ?De l'arithmétique simple
• violations(final)Quels invariants un état final viole-t-il ?De l'arithmétique simple
• verify(producers, consumers, capacity)La file est-elle juste sous tous les ordonnancements ?explore_allviolations
• study(bars)Les deux formes tiennent-elles sur les cours SPY ?verify

Sujet suivant : vous avez maintenant prouvé la justesse de petits programmes en exécutant tous les ordonnancements. Tester le code concurrent compare cela aux outils qu'utilisent les vraies équipes : un détecteur de courses qui observe une exécution et signale les accès qui pourraient se heurter, et un vérificateur de modèles qui fait ce que fait votre explorateur, mais avec les astuces qui lui permettent de passer à l'échelle.