Atômicas e leitura-modificação-escrita

Uma atualização perdida acontece porque uma leitura e uma escrita são dois turnos com um intervalo entre eles. Uma operação atômica fecha o intervalo: a leitura, a mudança e a escrita acontecem num passo indivisível. Com ela o contador está certo sob todos os escalonamentos: 4 escalonamentos em vez de 28, e nenhum errado.

Na nossa máquina um turno vai de um yield ao seguinte, então nada escrito entre dois yield pode ser intercalado. Uma função que não contém nenhum yield se comporta, portanto, como uma operação atômica. A mais comum é fetch-and-add: lê o valor antigo, escreve old + n e devolve old, tudo dentro de um turno. Uma thread contadora que a chama tem um só yield para todo o incremento em vez de dois, então as duas threads não podem mais ler o mesmo valor antigo.

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

As duas ordens terminam em {'x': 2} e nenhuma thread fica viva. Com um turno por incremento o programa só tem os dois escalonamentos [0, 1] e [1, 0] (4 no estudo, para 1 e 3 incrementos), e nenhum pode perder uma atualização.

Fetch-and-add só soma. Para todo o resto é preciso a ferramenta geral, a compara-e-troca (CAS). O padrão é um laço: ler o valor antigo, calcular o novo a partir dele e pedir ao CAS que o instale só se ninguém mudou o valor nesse meio-tempo. Se o CAS falha, alguém chegou antes, então lê-se de novo e tenta-se outra vez. A seguir, dois contadores CAS: quando a thread 1 lê antes de a thread 0 escrever, seu CAS falha uma vez, conta uma nova tentativa e a segunda tentativa tem sucesso. A atualização nunca se perde, porque uma escrita obsoleta é recusada.

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

Depois de [0, 1, 0, 1] o dicionário compartilhado é {'x': 1, 'retries': 1} e a thread 1 ainda está viva: leu 0, mas a thread 0 já tinha guardado 1, então sua compara-e-troca foi recusada. Dando à thread 1 seus dois turnos seguintes (ler de novo, trocar de novo) termina em x = 2. Note o preço: o perdedor fez o trabalho duas vezes. Fetch-and-add nunca tenta de novo; o CAS tenta tantas vezes quanto as threads colidem.

O explorador do tópico anterior supunha que cada thread toma um número fixo de turnos, lengths. Um laço de nova tentativa quebra isso: uma thread que perde uma corrida toma turnos extras, e quantos depende do escalonamento. A solução é percorrer os escalonamentos como uma árvore. A partir de um prefixo de turnos, run_prefix refaz o programa e diz quais threads ainda estão vivas; se nenhuma está, o escalonamento está completo e seu estado final é contado, e se há, ramifica-se em cada uma. Esta busca em profundidade visita cada escalonamento completo exatamente uma vez. values_of lê depois um nome compartilhado de todos os resultados.

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]

As duas correções estão certas sob todos os escalonamentos e diferem no custo. Fetch-and-add faz o trabalho em um turno por incremento: 4 escalonamentos. O contador CAS tem 144 escalonamentos, porque uma colisão acrescenta turnos, e no pior deles uma thread tenta de novo 3 vezes. É o compromisso do código sem travas: ninguém espera uma trava, mas o vencedor de cada corrida obriga o perdedor a repetir seu trabalho.

O laboratório segue a aula: fetch-and-add, compara-e-troca, uma thread contadora construída sobre cada um, o explorador que lida com threads de número de turnos variável, um auxiliar que lê os valores de um nome em todos os resultados e o estudo de três contadores nos dias de alta do SPY.

• Função · perguntaBaseia-se em
• fetch_add(shared, name, n)Somar n, e o que havia antes?Aritmética simples
• compare_and_swap(shared, name, expected, new)O valor ainda era o que li?Aritmética simples
• atomic_increments(k)Um contador sem intervalo?fetch_add
• cas_increments(k)Um contador que tenta de novo em vez de perder?compare_and_swapfetch_add
• explore_all(make_threads, shared)Todos os resultados, seja qual for o número de turnos?run_prefix
• values_of(outcomes, name)Que valores esse nome assume?Aritmética simples
• study(bars)Como os três contadores se comparam no SPY?explore_allvalues_of

Próximo tópico: supusemos que cada thread vê cada escrita imediatamente e em ordem. Um processador real não promete isso. O modelo de memória: acquire e release acrescenta um buffer de escrita à máquina, mostra um leitor vendo uma flag antes dos dados que ela protege e o corrige com a ordem que menos custa.