A fila sem travas mais simples não precisa de compara-e-troca nenhuma. Dê a cada posição um único escritor: só o produtor escreve a cauda, só o consumidor escreve a cabeça. Uma leitura obsoleta da posição do outro lado é sempre segura, então um buffer circular pode passar ticks entre duas threads sem trava. O explorador prova isso em 291 escalonamentos e mostra a quebra em 458 de 1072 quando entra um segundo produtor.
Um círculo com um único escritor por posição
Um buffer circular mantém dois contadores que só crescem. O produtor escreve a casa tail % capacity e depois publica subindo tail; o consumidor lê a casa head % capacity e depois a libera subindo head. O buffer está cheio quando tail - head == capacity e vazio quando head == tail. Por que é seguro sem trava? O produtor é o único escritor de tail, então pode guardar uma cópia local. Ele lê head para procurar espaço e, se o valor está obsoleto, a cabeça obsoleta é menor que a real, então o produtor acha o buffer mais cheio do que está e, no máximo, descarta um item, sem nunca sobrescrever uma casa não lida. O tail obsoleto do consumidor também é menor, então no máximo ele vê o buffer vazio. Este é o princípio do escritor único.
def initial_state(): return {"head": 0, "tail": 0, "pushed": (), "popped": ()}def ring_producer(items, capacity): def thread(shared): tail = 0 for item in items: yield head = shared.get("head", 0) if tail - head < capacity: yield shared["slot%d" % (tail % capacity)] = item yield tail += 1 shared["tail"] = tail shared["pushed"] = shared.get("pushed", ()) + (item,) return threaddef ring_consumer(attempts, capacity): def thread(shared): head = 0 for _ in range(attempts): yield tail = shared.get("tail", 0) if head != tail: yield item = shared.get("slot%d" % (head % capacity)) yield head += 1 shared["head"] = head shared["popped"] = shared.get("popped", ()) + (item,) return threadtwo = lambda shared: [ring_producer((7, 8), 2)(shared), ring_consumer(2, 2)(shared)]for prefix in ([0, 0, 0], [0, 0, 0, 1, 1, 1], [0, 0, 0, 0, 0, 0, 1, 1, 1, 1]): print(prefix, run_prefix(two, prefix, initial_state())[0])
[0, 0, 0] {'head': 0, 'tail': 1, 'pushed': (7,), 'popped': (), 'slot0': 7}
[0, 0, 0, 1, 1, 1] {'head': 1, 'tail': 1, 'pushed': (7,), 'popped': (7,), 'slot0': 7}
[0, 0, 0, 0, 0, 0, 1, 1, 1, 1] {'head': 1, 'tail': 2, 'pushed': (7, 8), 'popped': (7,), 'slot0': 7, 'slot1': 8}Três turnos do produtor empurram o primeiro tick: ele lê a cabeça, guarda o item em slot0 e publica tail: 1. Depois de três turnos do consumidor, head: 1 e popped: (7,). Na última linha o produtor empurrou os dois itens (slot0: 7, slot1: 8) e o consumidor pegou o primeiro; a thread 1 ainda está viva porque lhe resta uma tentativa.
Verificar os invariantes sobre todos os escalonamentos
O que uma fila deve prometer? Duas coisas. Os itens retirados são os itens inseridos, na mesma ordem e sem lacunas: popped é um prefixo de pushed (a propriedade FIFO). E nada é retirado que nunca foi inserido: len(popped) <= len(pushed). As threads registram o que inserem e retiram no mesmo turno que termina a operação, então violations pode verificar ambas em qualquer estado final. verify executa então o programa todo sob cada escalonamento com explore_all e conta os escalonamentos que quebram cada invariante. Um dicionário broken vazio sobre todos os escalonamentos é uma prova, para aquele tamanho de programa.
def violations(final): pushed, popped = tuple(final.get("pushed", ())), tuple(final.get("popped", ())) broken = [] if popped != pushed[: len(popped)]: broken.append("order") if len(popped) > len(pushed): broken.append("count") return sorted(broken)def verify(producers, attempts, capacity): def make_threads(shared): return [ring_producer(items, capacity)(shared) for items in producers] + [ring_consumer(attempts, capacity)(shared)] outcomes = explore_all(make_threads, initial_state()) broken = {} pushed = popped = 0 for key, count in outcomes.items(): final = dict(key) pushed = max(pushed, len(final.get("pushed", ()))) popped = max(popped, len(final.get("popped", ()))) 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(violations({"pushed": (7, 8), "popped": (8,)}))print(verify([(7, 8)], 2, 2))print(verify([(7, 8)], 2, 1))
['order']
{'schedules': 101, 'broken': {}, 'pushed': 2, 'popped': 2}
{'schedules': 25, 'broken': {}, 'pushed': 2, 'popped': 2}Um produtor com dois itens num buffer de duas casas, lido por um consumidor com duas tentativas, tem 101 escalonamentos e nenhum quebra um invariante. Com um buffer de uma casa só o produtor precisa descartar um item quando o consumidor é lento, mas continua correto: 25 escalonamentos, nenhuma violação. A lista retirada (8,) depois de (7, 8) inserido é, como esperado, uma violação da ordem.
Cada variável compartilhada tem exatamente uma thread que a escreve, e qualquer número que só a lê. Com um único escritor não há corrida nessa variável, e um leitor que vê um valor antigo continua seguro se o valor antigo só erra para o lado seguro.
O que o quebra: um segundo produtor
O argumento de segurança se apoia num fato: tail tem um escritor. Acrescente um segundo produtor e duas threads mantêm cada uma sua ideia da cauda. Ambas leem o mesmo head, calculam a mesma casa e a segunda escrita sobrescreve a primeira. O explorador acha: 458 dos 1072 escalonamentos de dois produtores (um tick cada) terminam com um popped que não é prefixo de pushed. A solução não é um remendo neste buffer, e sim outra estrutura, uma fila de vários produtores, que os próximos tópicos constroem com compara-e-troca.
print(verify([(7,), (8,)], 2, 2)){'schedules': 1072, 'broken': {'order': 458}, 'pushed': 2, 'popped': 1}Esta máquina executa cada operação em ordem. Num processador real o produtor deve publicar tail com uma escrita release e o consumidor lê-lo com um carregamento acquire, senão o consumidor pode ver a cauda nova antes do item (os dados obsoletos do tópico anterior). Ponha também head e tail em linhas de cache diferentes, senão as duas threads disputam uma linha (falso compartilhamento).
ticks = [int(round(float(c) * 100)) for c in load_bars()["close"].iloc[:4]]print(ticks)print("one producer: ", verify([ticks[:2]], 3, 2))print("two producers:", verify([ticks[:1], ticks[1:2]], 2, 2))
[9369, 9305, 9354, 9349]
one producer: {'schedules': 291, 'broken': {}, 'pushed': 2, 'popped': 2}
two producers: {'schedules': 1072, 'broken': {'order': 458}, 'pushed': 2, 'popped': 1}A lição inteira cabe em dois números: um escritor por posição basta para a correção em 291 escalonamentos, e dois escritores na mesma posição erram em 43% de 1072. O explorador não adivinhou; executou todas as intercalações.
O que você vai construir
O laboratório segue a aula: o buffer vazio, as threads produtora e consumidora, a verificação dos dois invariantes, o verificador que roda o explorador e o estudo que envia fechamentos do SPY pelo buffer com um e com dois produtores.
- Função · perguntaBaseia-se em
- initial_state()Como é um buffer vazio?Aritmética simples
- ring_producer(items, capacity)Como o produtor empurra?Aritmética simples
- ring_consumer(attempts, capacity)Como o consumidor retira?Aritmética simples
- violations(final)Quais invariantes este estado quebra?Aritmética simples
- verify(producers, attempts, capacity)O buffer está certo sob todos os escalonamentos?explore_allviolations
- study(bars)O que quebra com um segundo produtor?verify
Próximo tópico: o truque do escritor único tem um limite, e o compara-e-troca também. O problema ABA e os ponteiros etiquetados constrói uma pilha sem travas sobre compara-e-troca, mostra um escalonamento em que a troca tem sucesso embora a pilha tenha mudado por baixo, e a corrige com uma etiqueta de versão.
Lab: o que é avaliado
6 tarefas · 100 pts- O buffer vazio5 pts
- O produtor20 pts
- O consumidor20 pts
- Os invariantes10 pts
- Verificar sobre todos os escalonamentos15 pts
- O estudo com a tabela do SPY30 pts