Comment bien tester correctement les structures de données concurrentes

 (matklad.github.io)
2 points par GN⁺ 2024-07-07 | 1 commentaires | Partager sur WhatsApp
  • À partir d’un compteur concurrent Rust défectueux, l’article montre comment mettre en évidence, via un contrôle de l’ordonnancement d’exécution reproductible et minimisable, des problèmes que les tests de charge classiques sur threads ne détectent pas
  • Un wrapper AtomicU32 dédié aux tests insère pause(), et des managed threads s’arrêtent avant et après les opérations atomiques puis reprennent selon l’ordre choisi par le test
  • Un test simple peut faire exécuter 100 incréments 100 fois chacun par 100 threads et produire un échec comme 9598 au lieu de la valeur attendue 10000, mais cela dépend du timing et reste difficile à reproduire, déboguer et réduire
  • Un test de propriété basé sur arbtest permet de rejouer le même interleaving avec la même seed et de minimiser le cas d’échec jusqu’à 0: increment, 1: increment, 0: unpause, 1: unpause
  • En étendant la même approche avec exhaustigen, on peut énumérer tous les interleavings jusqu’à 5 incréments maximum, et après correction avec fetch_add, les 81133 interleavings passent

Compteur concurrent non atomique

  • L’exemple utilise AtomicU32 de Rust, mais increment() effectue load puis store(value + 1), donc l’opération d’incrément elle-même n’est pas atomique
  • La structure Counter est simple
    • value: AtomicU32
    • increment() lit la valeur avec SeqCst, ajoute 1 à la valeur lue, puis la réécrit
    • get() lit la valeur courante avec SeqCst
  • Deux threads peuvent lire la même valeur puis écrire le même résultat incrémenté, ce qui fait disparaître une mise à jour

Pourquoi les tests classiques sur threads sont insuffisants

  • La vérification la plus simple consiste à faire incrémenter en boucle le même compteur par plusieurs threads, puis à vérifier la valeur finale
    • thread_count = 100
    • increment_count = 100
    • la valeur attendue est 10000
  • Un exemple d’exécution échoue avec left: 9598, right: 10000
  • Cette méthode dépend fortement du timing d’ordonnancement
    • difficile de reproduire exactement le même échec de manière déterministe
    • difficile à déboguer
    • si l’on réduit le nombre de threads ou d’incréments, le test peut réussir par chance, ce qui rend la minimisation du cas d’échec difficile

Gérer les interleavings avec les tests basés sur les propriétés

  • Les tests basés sur les propriétés (PBT) se prêtent bien aux tests de machines à états
    • il est facile de générer des entrées aléatoires
    • on peut poser comme propriété que le résultat concurrent doit être identique à celui d’un modèle d’exécution séquentielle
    • cela correspond aussi au besoin de minimiser les entrées en échec
  • La difficulté vient du fait qu’il est compliqué de faire progresser de vrais threads OS pas à pas exactement au moment voulu
  • La solution consiste à choisir à chaque itération un thread aléatoire et à le faire avancer d’une étape
    • il faut pouvoir insérer un autre thread entre le load et le store d’un thread donné
    • pour cela, l’auteur construit une API de managed thread pilotée directement

AtomicU32 de test et insertion de pause

  • Dans les builds de test, std::sync::atomic::AtomicU32 est remplacé par un managed_thread::AtomicU32 maison
    • #[cfg(test)] use managed_thread::AtomicU32
    • #[cfg(not(test))] use std::sync::atomic::AtomicU32
  • Le wrapper AtomicU32 appelle pause() avant et après load() et store()
    • load: pause() → vrai load → pause()
    • store: pause() → vrai store → pause()
  • Grâce à ces points d’insertion, le test peut arrêter et relancer les threads autour des opérations atomiques pour contrôler l’ordre d’exécution

Forme de l’API de managed thread

  • Le test crée deux managed threads dans un std::thread::scope
    • comme il s’agit de scoped threads, ils peuvent emprunter des données locales de pile
    • on passe par exemple une référence au compteur comme état avec spawn(scope, &counter)
  • Un managed thread ne lance pas immédiatement une fonction main spécifique ; il exécute à la place une closure envoyée par le thread de contrôle via submit()
    • t.submit(|c| c.increment())
    • le thread exécute la closure sur son propre état T
  • Tant qu’il reste de l’entropie, la boucle de test choisit aléatoirement une action pour chaque thread
    • si le thread est arrêté, on appelle unpause()
    • sinon, on lance increment() via submit()
    • le modèle séquentiel counter_model est lui aussi incrémenté du même nombre de fois
  • À la fin, tous les threads sont join() puis on compare counter_model avec counter.get()

Implémentation de pause et unpause

  • pause() utilise thread_local! pour retrouver le contexte du managed thread courant sans modifier l’API Counter testée
    • le contexte est partagé via Arc<SharedContext>
    • SharedContext contient Mutex<State> et Condvar
  • Les états sont Ready, Running et Paused
    • Ready : en attente de la prochaine closure
    • Running : le managed thread est en cours d’exécution
    • Paused : le thread est arrêté sur un point pause()
  • Quand un managed thread atteint pause(), il fait passer son état de Running à Paused et notifie le thread de contrôle via la condition variable
  • unpause() fait passer l’état de Paused à Running, réveille le managed thread, puis attend à nouveau tant que l’état reste Running
    • cela évite que le thread de contrôle et le managed thread continuent à s’exécuter en même temps
    • à tout moment, un seul des deux avance, ce qui réduit la non-détermination

Reproduire et minimiser les échecs

  • L’exécution avec arbtest trouve un échec sur le compteur défectueux
    • dans l’exemple, la valeur du modèle est 4 et la valeur réelle 3
    • la seed d’échec est 0x4fd7ddff00000020
  • En réutilisant la même seed, on obtient exactement le même interleaving, ce qui facilite la reproduction de l’échec
  • Avec .minimize(), le cas d’échec se réduit à une exécution plus courte
    • la seed du cas minimal final est 0x9c2a13a600000001
    • la trace minimale tient en quatre étapes
      • 0: increment
      • 1: increment
      • 0: unpause
      • 1: unpause
  • Dans ce cas minimal, la valeur attendue est 2 mais la valeur réelle est 1, ce qui met clairement en évidence le défaut de l’incrément basé sur load/store

Extension à l’énumération de tous les interleavings

  • La même structure peut être adaptée d’une stratégie aléatoire vers une stratégie d’énumération
  • En utilisant exhaustigen, l’auteur écrit un test qui explore tous les interleavings jusqu’à 5 incréments maximum
    • le test évite les itérations factices et veille à toujours soit unpause un thread, soit lui soumettre un increment
  • L’implémentation défectueuse révèle le même bug
    • exemple d’échec : left: 2, right: 1
  • En corrigeant Counter::increment() avec fetch_add(1, SeqCst), le test passe
    • le wrapper AtomicU32 ajoute aussi pause() avant et après fetch_add()
    • le résultat d’exécution devient all 81133 interleavings are fine!
    • temps d’exécution : real 8.65s, CPU : 8.16s, RSS : 63.91mb

Extension vers les modèles mémoire faibles et le model checking

  • Dans cette implémentation jouet, AtomicU32 délègue actuellement à de vraies opérations atomiques
  • Une idée d’extension consiste à conserver, pour chaque atomic, l’ensemble des valeurs écrites et à renvoyer à la lecture une valeur aléatoire cohérente avec un modèle mémoire faible
  • L’exploration des interleavings peut elle aussi devenir plus intelligente que le simple aléatoire
    • une approche de type model checking permettrait de vérifier que tous les interleavings réellement différents ont été pris en compte
    • on pourrait aussi énumérer tous les interleavings sur de petits domaines, comme dans l’approche Generate All The Things

Pourquoi la minimisation fonctionne sans shrinking

  • Le arbtest utilisé ressemble à une interface classique de PRNG, mais repose sur un PRNG fini
    • si l’on continue à demander des valeurs aléatoires, il finit par renvoyer Err(OutOfEntropy)
    • c’est pour cela que le code de test utilise ? et while !rng.is_empty()
  • Quand le test épuise son entropie, il s’arrête plus tôt ; réduire l’entropie disponible raccourcit donc aussi l’exécution du test
  • Son implémentation interne se rapproche conceptuellement de &mut &[u8]
    • à chaque demande de nombre aléatoire, la tranche d’octets rétrécit
    • plus la tranche initiale est courte, plus le test devient simple
  • Grâce à cette approche, les cas d’échec peuvent se raccourcir sans qu’il soit nécessaire d’implémenter soi-même une logique de shrinking séparée
  • Le code source de l’exemple est disponible dans properly-concurrent

À lire aussi

1 commentaires

 
GN⁺ 2024-07-07
Commentaires sur Hacker News

  • En Rust, une bibliothèque appelée Temper est en cours de développement avec une approche similaire : https://github.com/reitzensteinm/temper/tree/main
    Cela dit, pour modéliser les implications étranges produites par l’ensemble du modèle mémoire de Rust, il faut aller beaucoup plus loin et disposer d’un registre qui suit quelles écritures chaque thread a observées. Selon l’ordonnancement mémoire atomique, les barrières de lecture/écriture, etc., observer l’écriture X peut impliquer qu’il faut aussi nécessairement avoir observé l’écriture Y
    Je pense que c’est probablement l’une des plus grandes collections de cas de test pour le modèle mémoire C++/Rust, avec presque tout ce qu’on peut trouver dans des livres, la norme C++, Stack Overflow, des blogs, etc. Par exemple, le fichier pour Rust Atomics and Locks de Mara Bos est ici : https://github.com/reitzensteinm/temper/blob/main/memlog/tes...
    Loom, mentionné dans l’article, est une bibliothèque similaire mais bien plus aboutie, qui permet de tester de manière exhaustive des composants de plus haut niveau comme les mutex ou les files : https://github.com/tokio-rs/loom mais elle ne modélise pas le modèle mémoire lui-même de façon aussi fine que Temper, et j’envisageais de porter les cas de test vers Loom
    J’ai été inspiré par la présentation de Will Wilson sur les tests de FoundationDB, et il construit actuellement chez Antithesis une solution basée sur un hyperviseur pour effectuer ce type de tests sur des conteneurs Docker arbitraires : https://www.youtube.com/watch?v=4fFDFbi3toc, https://antithesis.com/
    Je suis convaincu que ce domaine va énormément grandir au cours des 10 prochaines années. WebAssembly se situe à un point d’équilibre remarquable : suffisamment complet pour compiler des logiciels arbitraires, tout en restant assez simple pour que construire quelque chose comme Antithesis ne devienne pas un projet de cinq ans réservé à une équipe d’élite ayant déjà lancé une base de données

  • J’ai implémenté en Rust des instantanés atomiques en mémoire partagée, et j’ai pris les tests automatisés aussi sérieusement que possible : https://github.com/kaymanb/todc/tree/main/todc-mem
    Au début, j’ai utilisé Loom comme dans l’article, puis je suis ensuite passé à shuttle : https://github.com/tokio-rs/loom, https://github.com/awslabs/shuttle
    Au lieu d’une exploration exhaustive comme Loom, shuttle utilise une approche aléatoire, mais son ordonnanceur fournit toujours des garanties probabilistes pour la découverte de bugs. À l’usage, shuttle s’est révélé plus rapide et a mieux évolué vers des scénarios de test plus complexes
    Comme dans l’approche de l’article, si un ordonnanceancement précis fait échouer un test, on peut enregistrer la graine aléatoire. La capacité à reproduire rapidement un test en échec est très importante, et permet d’écrire des cas de test explicites pour des bugs déjà trouvés et corrigés : https://github.com/kaymanb/todc/blob/0e2874a70ec8beed8fae773...

  • Côté Kotlin/Java, Lincheck de JetBrains est une bonne bibliothèque pour ce genre de travail : https://github.com/JetBrains/lincheck
    J’aime particulièrement son aspect déclaratif et la façon dont il affiche les résultats de linéarisabilité

  • Je me demande s’il existe une bibliothèque de type Loom aussi en C++. J’ai des structures de données lock-free que j’aimerais tester

  • Si j’ai bien compris, cette approche a des limites en ce qui concerne la faible garantie de progression
    Le calcul principal n’est pas totalement trivial, mais sur du vrai matériel avec un vrai ordonnanceur, on peut imaginer une boucle cmpxchg dont la probabilité d’être interrompue sur un CPU donné est extrêmement faible. S’il y a n CPU, dans le pire des cas la probabilité de progression est de 1/n, alors qu’avec cette méthode de test elle devient 1/t^p. Ici, t est le nombre de tâches, qui peut être bien supérieur au nombre de CPU, et p est le nombre de pauses dans le corps de la boucle, qui peut facilement être de 3 ou plus. C’est largement suffisant pour faire paraître cassé un algorithme qui fonctionne en pratique
    À l’inverse, même si l’on veut considérer la faible progression comme un bug et exiger une forte progression, cette méthode ne semble pas vraiment fournir d’outil utile
    Cela dit, elle est clairement utile pour beaucoup de problèmes de concurrence

    • 1/t^p ne me semble pas juste, j’aurais tendance à dire simplement 1/t. Au bout du compte, après t unités de temps, au moins une tâche aura forcément progressé, et s’il y a t tâches, la probabilité que ce soit la mienne est de 1/t
      La confusion centrale semble être qu’être interrompu ne signifie pas forcément perdre au CAS

  • À propos du passage « Pour être honnête, il y a ici un peu de connaissance préalable. À moins de faire quelque chose de profondément maudit avec de l’assembleur inline, je ne vois pas comment éviter de créer de vrais threads. Si quelque chose doit appeler une fonction pause() et qu’on veut le laisser bloqué jusqu’à nouvel ordre, cela doit se produire dans un thread qui garde une pile distincte de celle du test », je me demande s’il ne serait pas possible d’utiliser un type de runtime asynchrone
    On dirait qu’on instrumente les opérations atomiques pour obtenir du multitâche coopératif. Il me faut peut-être plus de café, mais faire ça sans threads semble plus simple

    • L’asynchrone serait pratique, mais une autre exigence est de ne pas vouloir modifier l’API observable de l’extérieur du logiciel testé. L’asynchrone étant « contagieux », une API synchrone doit avoir une implémentation synchrone

  • Un inconvénient de cette approche est qu’il faut modifier le code testé lui-même pour l’adapter au code de test
    On pourrait sans doute obtenir le même effet en lançant deux threads et en faisant du pas à pas avec ptrace, en intercalant « aléatoirement » l’exécution des instructions. Un peu comme le mode chaos de rr
    Cela dit, certaines instructions peuvent ne pas être atomiques, donc à moins de faire de l’émulation, il faudrait probablement un moyen d’exécuter en pas à pas à l’échelle du « microcode atomique », si c’est seulement possible

    • Ça ressemble à l’hyperviseur d’Antithesis

  • Utiliser Loom semble nécessiter de la compilation conditionnelle ; pour tester une bibliothèque, ça va sans doute, mais c’est assez intrusif
    #[cfg(loom)]
    pub(crate) use loom::sync::atomic::AtomicUsize;
    #[cfg(not(loom))]
    pub(crate) use std::sync::atomic::AtomicUsize;
    Je me demande s’il existe des langages qui facilitent davantage l’usage de son propre ordonnanceur

  • Pour être vraiment exhaustif, on pourrait sans doute exécuter le test avec ptrace et faire avancer les threads en pas à pas afin de produire différents entrelacements au niveau des instructions. Je me demande si quelqu’un a déjà réellement vu cette approche en pratique
    Quand on ne peut pas instrumenter le code comme ici, existe-t-il une alternative pour faire des tests en boîte noire ?

    • J’ai déjà utilisé ce genre d’approche pour tester des gestionnaires de signaux asynchrones, mais dans ce cas le nombre de combinaisons est bien plus favorable. Si le thread principal exécute n instructions, il suffit de n exécutions où l’on injecte le signal après 0 à n instructions, puis le gestionnaire de signal s’exécute jusqu’au bout avant que le thread principal ne termine aussi. Le temps total est de O(n^2)
      Mais s’il y a t threads exécutant chacun n instructions et qu’ils peuvent tous s’interrompre mutuellement à chaque frontière, cela devient difficilement praticable pour des valeurs réalistes de n. Il faudrait probablement réduire le problème en ne retenant et en simulant que les opérations au comportement intéressant

  • Ça a l’air assez cool, il faudra que j’essaie. Cela dit, ça ne détectera pas tous les types d’erreurs. Chaque appel à pause() introduit une synchronisation entre les threads ; est-ce que cela ne risque pas de masquer certains problèmes de course aux données ? En Rust, ce n’est peut-être pas un problème