Colas de varios productores y varios consumidores

Con un productor y un consumidor una posición tiene un escritor. Con muchos, las posiciones deben repartirse atómicamente, y un ticket no basta: el productor que tiene la posición 5 aún no ha escrito su elemento. Un número de secuencia en cada casilla dice a quién le toca la casilla. Sobre 2756 planificaciones de dos productores y 1600 de dos consumidores, esta cola nunca se rompe.

Una cola acotada de capacity casillas usa dos contadores que solo crecen. Un productor toma la siguiente posición con un comparar e intercambiar sobre enq; un consumidor toma la siguiente con uno sobre deq. La posición p vive en la casilla p % capacity. El comparar e intercambiar hace seguras las reservas: dos productores nunca obtienen la misma posición. Pero reservar una posición no es escribir el elemento. Entre el intercambio y la escritura la casilla está reservada y vacía, y un consumidor que la leyera ahora leería basura. Cada casilla lleva por eso un número de secuencia que solo avanza en el momento correcto.

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}

Lea las tres ejecuciones. Tras [0, 0, 0] el productor ha reservado la posición 0 (enq: 1), guardado data0 y subido seq0 a 1: el elemento está listo. Tras los tres turnos del consumidor, got0: 7 y seq0: 2: la casilla queda libre para la siguiente vuelta. En la última ejecución el consumidor mira entre la reserva del productor y su publicación, ve seq0 = 0 en vez de 1 y se va con las manos vacías. No lee basura. Solo dice «nada listo todavía», y el elemento lo encontrará un intento posterior.

¿Qué debe prometer ahora la cola? Los productores registran cada elemento bajo su posición (put3), los consumidores registran lo que tomaron bajo la posición que reservaron (got3). Si la cola funciona, cada posición sacada contiene el elemento que se puso allí (sin mismatch), y las posiciones sacadas son 0, 1, 2, ... sin huecos ni repeticiones (sin gap). Eso da a la cola su orden: el orden de las posiciones. violations lee esos dos hechos de un estado final y verify lanza el explorador para cualquier número de productores y consumidores. Aquí, un solo productor con dos elementos contra dos consumidores, y un solo elemento contra un consumidor.

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}

Para ver su valor, quitémoslo. El consumidor descuidado siguiente reserva una posición con el comparar e intercambiar y luego lee la casilla enseguida, sin mirar seq. Contra un productor con un solo elemento tiene 20 planificaciones, y en 10 de ellas reserva una posición antes de que el productor haya publicado nada y devuelve None como elemento, o el elemento equivocado. La cola real tiene 4 planificaciones para el mismo programa y ninguna se rompe. El número de secuencia es exactamente la comprobación que convierte un ticket reservado en un elemento entregado.

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}

Ambas formas de la cola son correctas sobre todas las planificaciones, con el explorador como testigo, mientras que una versión sin la comprobación de secuencia falla en la mitad de su espacio (mucho menor). Productores y consumidores nunca se esperan: un consumidor que no encuentra nada listo simplemente vuelve, y un productor que encuentra la cola llena devuelve False. Ese es el precio y el regalo de una cola sin cerrojos: las operaciones pueden fallar, pero nunca bloquean.

El laboratorio sigue la lección: la cola vacía con sus números de secuencia, el enqueue y el dequeue con reserva por comparar e intercambiar y publicación, la comprobación de los dos invariantes, el verificador que lanza el explorador para cualquier número de productores y consumidores, y el estudio sobre dos cierres de SPY.

• Función · preguntaSe apoya en
• init_queue(capacity)¿Cómo es una cola vacía?Aritmética simple
• enqueue(shared, item, capacity)¿Cómo reserva y publica un productor?Aritmética simple
• dequeue(shared, capacity)¿Cómo reserva y libera un consumidor?Aritmética simple
• violations(final)¿Qué invariantes rompe un estado final?Aritmética simple
• verify(producers, consumers, capacity)¿Es correcta la cola bajo todas las planificaciones?explore_allviolations
• study(bars)¿Resisten ambas formas con cierres de SPY?verify

Siguiente tema: ya ha demostrado la corrección de programas pequeños ejecutando todas las planificaciones. Probar código concurrente lo compara con las herramientas que usan los equipos reales: un detector de carreras que observa una ejecución y señala los accesos que podrían competir, y un verificador de modelos que hace lo que hace su explorador, pero con los trucos que le permiten escalar.