Filas de vários produtores e vários consumidores

Com um produtor e um consumidor uma posição tem um escritor. Com muitos, as posições precisam ser distribuídas atomicamente, e um tíquete não basta: o produtor que tem a posição 5 ainda não escreveu seu item. Um número de sequência em cada casa diz de quem é a vez da casa. Em 2756 escalonamentos de dois produtores e 1600 de dois consumidores, essa fila nunca quebra.

Uma fila limitada de capacity casas usa dois contadores que só crescem. Um produtor toma a próxima posição com um compara-e-troca em enq; um consumidor toma a próxima com um em deq. A posição p vive na casa p % capacity. O compara-e-troca torna as reservas seguras: dois produtores nunca recebem a mesma posição. Mas reservar uma posição não é escrever o item. Entre a troca e a escrita a casa está reservada e vazia, e um consumidor que a lesse agora leria lixo. Cada casa carrega por isso um número de sequência que só avança no momento certo.

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}

Leia as três execuções. Depois de [0, 0, 0] o produtor reservou a posição 0 (enq: 1), guardou data0 e subiu seq0 para 1: o item está pronto. Depois dos três turnos do consumidor, got0: 7 e seq0: 2: a casa fica livre para a próxima volta. Na última execução o consumidor olha entre a reserva do produtor e sua publicação, vê seq0 = 0 em vez de 1 e vai embora de mãos vazias. Ele não lê lixo. Só diz «nada pronto ainda», e o item será achado por uma tentativa posterior.

O que a fila deve prometer agora? Os produtores registram cada item sob sua posição (put3), os consumidores registram o que pegaram sob a posição que reservaram (got3). Se a fila funciona, cada posição retirada contém o item que foi posto ali (sem mismatch), e as posições retiradas são 0, 1, 2, ... sem buracos nem repetições (sem gap). Isso dá à fila sua ordem: a ordem das posições. violations lê esses dois fatos de um estado final e verify roda o explorador para qualquer número de produtores e consumidores. Aqui, um único produtor com dois itens contra dois consumidores, e um único item contra um 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 o valor, tire-o. O consumidor descuidado abaixo reserva uma posição com o compara-e-troca e depois lê a casa na hora, sem olhar seq. Contra um produtor de um só item ele tem 20 escalonamentos, e em 10 deles reserva uma posição antes de o produtor publicar qualquer coisa e devolve None como item, ou o item errado. A fila real tem 4 escalonamentos para o mesmo programa e nenhum quebra. O número de sequência é exatamente a verificação que transforma um tíquete reservado num item entregue.

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}

As duas formas da fila estão corretas sobre todos os escalonamentos, com o explorador como testemunha, enquanto uma versão sem a verificação de sequência falha na metade do seu espaço (bem menor). Produtores e consumidores nunca esperam uns pelos outros: um consumidor que não encontra nada pronto simplesmente volta, e um produtor que encontra a fila cheia devolve False. Esse é o preço e o presente de uma fila sem travas: as operações podem falhar, mas nunca bloqueiam.

O laboratório segue a aula: a fila vazia com seus números de sequência, o enqueue e o dequeue com reserva por compara-e-troca e publicação, a verificação dos dois invariantes, o verificador que roda o explorador para qualquer número de produtores e consumidores, e o estudo sobre dois fechamentos do SPY.

• Função · perguntaBaseia-se em
• init_queue(capacity)Como é uma fila vazia?Aritmética simples
• enqueue(shared, item, capacity)Como um produtor reserva e publica?Aritmética simples
• dequeue(shared, capacity)Como um consumidor reserva e libera?Aritmética simples
• violations(final)Quais invariantes um estado final quebra?Aritmética simples
• verify(producers, consumers, capacity)A fila está certa sob todos os escalonamentos?explore_allviolations
• study(bars)As duas formas se sustentam com fechamentos do SPY?verify

Próximo tópico: você agora provou a correção de programas pequenos executando todos os escalonamentos. Testar código concorrente compara isso com as ferramentas que as equipes reais usam: um detector de corridas que observa uma execução e sinaliza os acessos que poderiam competir, e um verificador de modelos que faz o que o seu explorador faz, mas com os truques que o deixam escalar.