Como testar corretamente estruturas de dados concorrentes

 (matklad.github.io)
2 pontos por GN⁺ 2024-07-07 | 1 comentários | Compartilhar no WhatsApp
  • Usando como exemplo um contador concorrente quebrado em Rust, o texto revela problemas que testes comuns de carga com threads deixam passar por meio de controle de ordem de execução reproduzível e minimizável
  • Um wrapper de AtomicU32 para testes insere pause(), e uma managed thread para antes e depois das operações atômicas, sendo retomada na ordem escolhida pelo teste
  • Um teste simples pode fazer 100 threads incrementarem 100 vezes cada uma e produzir falhas como 9598 em vez do valor esperado 10000, mas isso depende de timing, o que dificulta reproduzir, depurar e reduzir o caso
  • Um teste baseado em propriedades com arbtest reproduz o mesmo interleaving com a mesma seed e minimiza o caso de falha até 0: increment, 1: increment, 0: unpause, 1: unpause
  • Expandindo a mesma estrutura com exhaustigen, é possível enumerar todos os interleavings até 5 incrementos, e após corrigir com fetch_add, 81133 interleavings passam no teste

Contador concorrente não atômico

  • O exemplo usa AtomicU32 do Rust, mas increment() faz load seguido de store(value + 1), então a operação de incremento em si não é atômica
  • A estrutura Counter é simples
    • value: AtomicU32
    • increment() lê o valor com SeqCst, soma 1 ao valor lido e grava de volta
    • get() lê o valor atual com SeqCst
  • Duas threads podem ler o mesmo valor e depois gravar o mesmo resultado incrementado, fazendo uma atualização se perder

Por que testes comuns com threads não bastam

  • A verificação mais simples é fazer várias threads incrementarem repetidamente o mesmo contador e checar o valor final
    • thread_count = 100
    • increment_count = 100
    • o valor esperado é 10000
  • Um exemplo de execução falha com left: 9598, right: 10000
  • Esse método depende fortemente do timing de escalonamento
    • é difícil reproduzir de forma determinística a mesma falha
    • é difícil depurar
    • se o número de threads ou de incrementos for reduzido, o teste pode passar por sorte, dificultando minimizar o caso de falha

Lidando com interleavings com testes baseados em propriedades

  • Testes baseados em propriedades (PBT) combinam bem com testes de máquina de estados
    • é fácil gerar entradas arbitrárias
    • pode-se definir a propriedade de que o resultado da execução concorrente deve ser igual ao de um modelo sequencial
    • isso também combina com a necessidade de minimizar entradas que falham
  • A dificuldade é que não é fácil fazer threads reais do sistema operacional avançarem um passo por vez no momento desejado
  • A solução é estruturar o teste para escolher uma thread arbitrária a cada iteração e avançá-la um passo
    • é preciso conseguir inserir outra thread entre o load e o store de uma mesma thread
    • para isso, é criada uma API de managed thread que permite controlar diretamente as threads

AtomicU32 de teste e inserção de pause

  • Em builds de teste, usa-se managed_thread::AtomicU32 em vez de std::sync::atomic::AtomicU32
    • #[cfg(test)] use managed_thread::AtomicU32
    • #[cfg(not(test))] use std::sync::atomic::AtomicU32
  • O wrapper AtomicU32 chama pause() antes e depois de load() e store()
    • load: pause() → load real → pause()
    • store: pause() → store real → pause()
  • Graças a esses pontos de inserção, o teste pode parar e retomar threads ao redor das operações atômicas, controlando a ordem de execução

Formato da API de managed thread

  • O teste cria duas managed threads dentro de std::thread::scope
    • como são scoped threads, elas podem pegar emprestados dados locais da stack
    • o estado é passado como referência ao contador, como em spawn(scope, &counter)
  • A managed thread não executa uma função main específica desde o início; ela executa closures enviadas pela thread de controle via submit()
    • t.submit(|c| c.increment())
    • a thread executa a closure sobre seu estado T
  • O loop de teste realiza ações aleatórias em cada thread enquanto ainda houver entropia
    • se a thread estiver parada, chama unpause()
    • se não estiver parada, executa increment() via submit()
    • o modelo sequencial counter_model também é incrementado o mesmo número de vezes
  • No final, todas as threads são finalizadas com join() e compara-se counter_model com o valor real de counter.get()

Implementação de pause e unpause

  • pause() usa thread_local! para encontrar o contexto da managed thread atual sem alterar a API de Counter em teste
    • o contexto é compartilhado como Arc<SharedContext>
    • SharedContext contém Mutex<State> e Condvar
  • Os estados são Ready, Running e Paused
    • Ready: aguardando a próxima closure
    • Running: a managed thread está em execução
    • Paused: a execução está parada em um ponto de pause()
  • Quando a managed thread chega a pause(), ela muda o estado de Running para Paused e avisa a thread de controle pela condition variable
  • unpause() muda o estado de Paused para Running, acorda a managed thread e depois espera até que o estado deixe de ser Running
    • isso impede que a thread de controle e a managed thread continuem executando ao mesmo tempo
    • em qualquer momento, apenas uma das duas executa, reduzindo a não determinismo

Reprodução e minimização da falha

  • A execução com arbtest encontra uma falha no contador quebrado
    • no exemplo, o valor do modelo é 4 e o valor real é 3
    • a seed da falha é 0x4fd7ddff00000020
  • Ao informar a mesma seed, é possível obter novamente o mesmo interleaving, o que facilita reproduzir a falha
  • Com .minimize(), o caso de falha é reduzido a uma execução mais curta
    • a seed mínima final é 0x9c2a13a600000001
    • o trace mínimo tem quatro passos
      • 0: increment
      • 1: increment
      • 0: unpause
      • 1: unpause
  • Nesse caso mínimo, o valor esperado é 2, mas o valor real é 1, expondo o defeito do incremento baseado em load/store

Expansão para enumerar todos os interleavings

  • A mesma estrutura pode ser adaptada para usar enumeração em vez de interleavings aleatórios
  • Com exhaustigen, é escrito um teste que explora todos os interleavings até 5 incrementos
    • o teste evita iterações inúteis e é estruturado para sempre fazer unpause ou submeter um increment
  • A implementação quebrada encontra o mesmo bug
    • um exemplo de falha é left: 2, right: 1
  • Ao corrigir Counter::increment() para fetch_add(1, SeqCst), o teste passa
    • o wrapper AtomicU32 também recebe pause() antes e depois de fetch_add()
    • o resultado da execução é all 81133 interleavings are fine!
    • o tempo de execução é real 8.65s, CPU 8.16s e RSS 63.91mb

Expansão para modelo de memória fraca e model checking

  • Na implementação de brinquedo atual, AtomicU32 delega para um atomic real
  • Uma ideia de expansão é fazer cada atomic manter o conjunto de valores já escritos e, ao ler, retornar um valor arbitrário consistente com um modelo de memória fraca
  • A exploração de interleavings também pode ficar mais inteligente do que uma busca aleatória
    • com uma abordagem de model checking, é possível verificar se todos os interleavings significativamente diferentes foram considerados
    • como na abordagem de Generate All The Things, dá para enumerar todos os interleavings dentro de um pequeno limite

Por que é possível minimizar sem shrinking

  • O arbtest usado parece uma interface de PRNG familiar, mas usa um PRNG finito
    • se você continuar pedindo valores aleatórios, em algum momento ele retorna Err(OutOfEntropy)
    • por isso o código de teste usa ? e while !rng.is_empty()
  • Quando o teste consome toda a entropia, ele termina mais cedo; assim, reduzir a entropia disponível também encurta a execução do teste
  • A implementação interna é conceitualmente parecida com &mut &[u8]
    • cada solicitação de número aleatório encurta o slice de bytes
    • quanto menor o slice inicial, mais simples tende a ser o teste
  • Graças a isso, os casos de falha podem ficar mais curtos sem precisar implementar manualmente uma lógica separada de shrinking
  • O código-fonte do exemplo está em properly-concurrent

Leituras relacionadas

1 comentários

 
GN⁺ 2024-07-07
Opiniões no Hacker News

  • Estou criando uma biblioteca chamada Temper em Rust com uma abordagem parecida: https://github.com/reitzensteinm/temper/tree/main
    Mas, para modelar as implicações estranhas criadas pelo modelo de memória completo do Rust, é preciso ir muito além: é necessário um livro-razão que rastreie quais escritas cada thread percebeu. Dependendo da ordem de memória atômica, fences de leitura/escrita etc., podem surgir garantias do tipo: se você percebeu a escrita X, então também precisa necessariamente perceber a escrita Y
    Acredito que seja uma das coleções mais completas de casos de teste para os modelos de memória de C++/Rust, reunindo quase tudo o que pude encontrar em livros, no padrão C++, no Stack Overflow, em blogs etc. Por exemplo, o arquivo para Rust Atomics and Locks, da Mara Bos, está aqui: https://github.com/reitzensteinm/temper/blob/main/memlog/tes...
    Loom, mencionado no texto, é uma biblioteca parecida, mas muito mais madura, que permite testar exaustivamente componentes de nível mais alto, como mutexes ou filas: https://github.com/tokio-rs/loom No entanto, ela não modela o próprio modelo de memória de forma tão detalhada quanto o Temper, e eu estava pensando em portar os casos de teste para o Loom
    Fui inspirado por uma apresentação de Will Wilson sobre testes no FoundationDB; hoje ele está na Antithesis criando uma solução baseada em hipervisor que realiza esse tipo de teste em contêineres Docker arbitrários: https://www.youtube.com/watch?v=4fFDFbi3toc, https://antithesis.com/
    Acredito firmemente que essa área vai crescer muito nos próximos 10 anos. WebAssembly está em um ponto ideal: completo o bastante para compilar software arbitrário, mas simples o suficiente para que criar algo como a Antithesis não precise ser um projeto de 5 anos para uma equipe de elite que já lançou um banco de dados

  • Implementei um snapshot atômico de memória compartilhada em Rust e levei os testes automatizados o mais a sério que consegui: https://github.com/kaymanb/todc/tree/main/todc-mem
    No começo usei o Loom citado no texto, mas depois troquei para o shuttle: https://github.com/tokio-rs/loom, https://github.com/awslabs/shuttle
    O shuttle usa uma abordagem randomizada em vez de exploração exaustiva como o Loom, mas o escalonador ainda oferece garantias probabilísticas para encontrar bugs. Na prática, achei o shuttle mais rápido e capaz de escalar para cenários de teste mais complexos
    De forma parecida com a abordagem do texto, se uma determinada agenda de execução causar falha no teste, é possível salvar a semente aleatória. A capacidade de reproduzir rapidamente um teste com falha é muito importante e permite escrever casos de teste explícitos para bugs que já foram encontrados e corrigidos: https://github.com/kaymanb/todc/blob/0e2874a70ec8beed8fae773...

  • No lado Kotlin/Java, o Lincheck da JetBrains é uma boa biblioteca para esse tipo de trabalho: https://github.com/JetBrains/lincheck
    Gosto especialmente do fato de ser declarativa e da forma como exibe os resultados de linearizabilidade

  • Fico curioso se existe alguma biblioteca tipo Loom para C++. Tenho algumas estruturas de dados lock-free que gostaria de testar

  • Se entendi corretamente, esta abordagem tem limitações em relação a garantias fracas de progresso
    Embora o cálculo no texto não seja totalmente trivial, podemos pensar em um loop de cmpxchg que, em hardware real e com um escalonador real, tem uma probabilidade extremamente baixa de ser interrompido em uma CPU específica. Se o número de CPUs for n, no pior caso a probabilidade de progredir é 1/n, mas neste método de teste ela se torna 1/t^p. Aqui, t é o número de tarefas, que pode ser muito maior que o número de CPUs, e p é o número de pausas dentro do corpo desse loop, que facilmente chega a 3 ou mais. Isso já é suficiente para fazer um algoritmo que funciona na prática parecer quebrado
    Por outro lado, mesmo quando se quer tratar progresso fraco como bug e exigir progresso forte, este método não parece oferecer uma ferramenta útil
    Ainda assim, ele claramente é útil para muitos problemas de concorrência

    • Acho que 1/t^p não está certo; vejo isso simplesmente como 1/t. Afinal, depois de um tempo t, alguma tarefa certamente terá progredido e, se houver t tarefas, a probabilidade de que tenha sido a minha é 1/t
      A confusão principal parece estar no fato de que ser interrompido não significa necessariamente perder no CAS

  • Sobre o trecho “Para ser sincero, há um pouco de conhecimento prévio aqui. A menos que você faça algo muito amaldiçoado com assembly inline, não acho que seja possível evitar a criação de threads reais. Se algo chama a função pause() e queremos fazê-lo parar até receber instruções futuras, isso precisa acontecer dentro de uma thread que mantenha uma pilha separada da pilha do teste”, fico curioso se não daria para usar algum tipo de runtime assíncrono
    Parece que se está instrumentando operações atômicas para obter multitarefa cooperativa. Talvez eu precise tomar mais café, mas parece mais simples fazer isso sem threads

    • Usar async seria conveniente, mas outro requisito é que não queremos alterar a API observada externamente do software em teste. Como async é “contagioso”, uma API síncrona precisa usar uma implementação síncrona

  • Uma desvantagem desta abordagem é que o próprio código em teste precisa ser modificado para se adequar ao código de teste
    Parece que daria para fazer a mesma coisa iniciando duas threads e usando ptrace para executá-las passo a passo, inserindo “aleatoriamente” a execução de instruções. Algo como o modo chaos do rr
    No entanto, algumas instruções podem não ser atômicas; então, se é que isso é possível sem emulação, talvez fosse necessário executar passo a passo na unidade de “microcódigo atômico”

    • Parece o hipervisor da Antithesis

  • Para usar o Loom, parece que é necessária compilação condicional e, embora isso seja aceitável ao testar uma biblioteca, é bastante intrusivo
    #[cfg(loom)]
    pub(crate) use loom::sync::atomic::AtomicUsize;
    #[cfg(not(loom))]
    pub(crate) use std::sync::atomic::AtomicUsize;
    Fico curioso se existe alguma linguagem que permita usar melhor um escalonador próprio

  • Se quisermos ser realmente minuciosos, parece que poderíamos executar o teste com ptrace e avançar as threads passo a passo para criar diferentes interleavings no nível de instruções. Fico curioso se alguém já viu isso na prática
    Em casos como este, em que não é possível instrumentar o código, haveria uma alternativa para testes de caixa-preta?

    • Já usei esse tipo de abordagem para testar manipuladores de sinal assíncronos, mas nesse caso o número de combinações é muito mais favorável. Se a thread principal executa n instruções, são necessárias apenas n execuções, rodando de 0 até n instruções antes de inserir o sinal; depois disso, o manipulador de sinal roda até o fim e a thread principal também roda até o fim. O tempo total é O(n^2)
      Mas, se houver t threads, cada uma executando n instruções, e todas puderem interromper umas às outras em cada fronteira, isso fica impraticável para valores realistas de n. Acho que seria preciso reduzir o problema selecionando e simulando apenas as operações com comportamento interessante

  • Parece bem interessante, então vou experimentar. Ainda assim, não deve capturar todos os tipos de erro. Cada chamada a pause() cria sincronização entre as threads, então será que alguns problemas de corrida de dados não ficariam ocultos? Em Rust, talvez isso não seja um problema