Pile synchronisée sans verrous

Écrire une pile synchronisée n'est pas en soi une tâche très ardue. Écrire une pile synchronisée sans avoir recours à des verrous est une chose plus délicate, mais qui peut en valoir la peine.

Cet article présente deux implémentations de piles synchronisées : une implémentation reposant sur un verrou (un mutex) et une autre basée sur des opérations atomiques. Les deux implémentations offriront une interface identique, ceci pour simplifier les tests et les comparaisons. Une courte analyse des résultats obtenus à l'aide d'une application jouet pour ces deux implémentations suit.

Implémentation avec verrou

L'implémentation avec verrou sera la plus simple des deux implémentations examinées ici, et de loin d'ailleurs.

Nous utiliserons deux classes exposant la même interface publique, au nom d'un espace nommé près. J'ai nommé l'implémentation avec verrou lock_full::stack, faute d'une meilleure idée, et pour la distinguer de lock_free::stack (plus bas). Les deux implémentations seront génériques.	`#ifndef LOCK_FULL_STACK_H #define LOCK_FULL_STACK_H #include <mutex> #include <stack> namespace lock_full { template <class T> class stack {`
L'implémentation avec verrou reposera, sans surprises, sur un std::stack<T> pour la fonctionnalité de pile en tant que telle, et sur un std::mutex pour ce qui est de la synchronisation. L'idée ici est de ne pas réinventer la roue, et de tirer le maximum des outils existants.	`std::stack<T> impl; std::mutex m;`
Sans que ce ne soit nécessaire dans ce cas, j'ai fait de lock_full::stack une classe incopiable et ne supportant pas le mouvement. En pratique, le côté incopiable était déjà assuré par la présence d'un attribut incopiable (le mutex), alors que le mouvement aurait été banal à implémenter. J'ai fait ce choix par souci de conformité entre les implémentations, n'ayant pas fait le choix d'implémenter ces services dans le cas de la version sans verrou. Le constructeur par défaut explicitement marqué comme tel est nécessaire du fait que l'exposition d'autres constructeurs, même supprimés, entraînait la non-génération du constructeur par défaut, alors que nous en avons besoin ici.	`public: stack() = default; ~stack() = default; stack(const stack&) = delete; stack(stack&&) = delete; stack& operator=(const stack&) = delete; stack& operator=(stack&&) = delete;`
L'ajout d'un élément sur le dessus de la pile (méthode push()) est trivial, se composant d'une paire synchronisation / délégation.	`void push(const T &val) { std::lock_guard<std::mutex> _{ m }; impl.push(val); }`
L'extraction d'un élément de la pile (méthode pop()) est une opération presque triviale, mais cette évidence est trompeuse. Notez que dans ce cas, la signature de lock_full::stack<T>::pop() est différente de celle de std::stack<T>::pop(), du fait que la version présentée ici doit se conformer aux contraintes de Thread-Safety et doit permettre à la fois d'essayer de retirer un élément et de constater le succès ou l'échec (p. ex. : lorsque la pile est vide) de cette tentative. Exposer une méthode empty() ici serait contre-productif, menant tout droit vers des cas de TOCTTOU.	`bool pop(T &val) { std::lock_guard<std::mutex> _{ m }; if (impl.empty()) return false; val = impl.top(); impl.pop(); return true; } }; } #endif`

Ce code est très simple, avouons-le. Il est aussi fonctionnel et plutôt efficace, mais tout ajout et toute extraction s'y verra sérialisée dû au passage obligé par l'obtention d'un mutex.

Implémentation sans verrou

Tel qu'annoncé, programmer une structure de données fonctionnellement équivalente à la précédente mais sans avoir recours à un verrou est une aventure d'un tout autre ordre.

Tel qu'annoncé, nos classes lock_full::stack et lock_free::stack sont toutes deux génériques et exposent des interfaces publiques essentiellement identiques, de manière à rendre leurs codes clients respectifs interchangeables.	`#ifndef LOCK_FREE_STACK_H #define LOCK_FREE_STACK_H #include <atomic> namespace lock_free { template <class T> class stack {`
Sur le plan de l'implémentation, notre version sans verrou reposera sur des noeuds chaînés les uns les autres et accessibles à partir d'une noeud servant de tête de liste. C'est ce noeud qui sera le point de contention en situation de concurrence, celui qui aura besoin de synchronisation, ce qui explique que ce soit celui que nous entreposerons dans un pointeurs atomique.	`struct noeud { T val; noeud succ; noeud(const T &val) : val{ val }, succ{} { } noeud(T &&val) : val{ std::move(val) }, succ{} { } }; std::atomic<noeud> tete;`
Par souci de sécurité et de cohésion, j'ai fait de lock_free::stack une classe incopiable et ne supportant pas le mouvement.. Implémenter la copie aurait été un petit cauchemar ici, et bien que le mouvement aurait pu être implémenté, je ne souhaitais pas distraire du propos de cet article. Le constructeur par défaut créera explicitement une pile vide.	`public: stack(const stack&) = delete; stack(stack&&) = delete; stack& operator=(const stack&) = delete; stack& operator=(stack&&) = delete; stack() : tete{ nullptr } { }`
Pour réaliser l'ajout d'une valeur sur la pile (un push()), notre stratégie va comme suit : Avant toute synchronisation, par souci d'efficacité, nous instancions le noeud p que nous souhaitons ajouter sur le dessus de la pile Initialement, le successeur de p est le noeud sur le dessus de la pile au début de l'exécution. C'est une démarche a priori optimiste, car s'il n'y a pas de contention sur la tête, alors il ne restera plus qu'à faire de p la tête de la pile et le tour sera joué La mécanique de permutation atomique qui suit, ou compare_exchange_weak(), exprime l'idée suivante : « Tant que je ne serai pas parvenu à placer p à la place de tete, mets la valeur de tete telle que lue ici dans p->succ » Cette répétitive se poursuivra tant que notre méthode ne sera pas celle qui aura réussi à remplacer la valeur de tete par celle de p. À chaque échec (chaque cas où un tiers aura modifié tete et interféré avec notre propre code), la valeur qui sera constatée dans tete sera placée dans p->succ, ce qui explique que le corps de la répétitive soit vide (sinon, après tout, ce que nous y aurions fait aurait été de relire le pointeur tete et de le placer dans p->succ) Une fois la répétitive terminée, p est sur le dessus de la pile, au mieux de notre connaissance; évidemment, entre la fin de la répétitive et la fin de la méthode, il se peut qu'un autre push() ou qu'un autre pop() aient été faits, mais notre propre travail aura été accompli Notez que je me suis permis la fantaisie d'utiliser un modèle mémoire relaxé pour cette opération. Comprenez-vous pour quelle raison ce relâchement est raisonnable dans ce cas-ci?	`void push(const T &val) { auto p = new noeud{ val }; p->succ = tete.load(); while (!tete.compare_exchange_weak(p->succ, p, std::memory_order_relaxed)) // brrrr... ; }`
Le pop() est lui aussi plus complexe que sa contrepartie avec verrous. Voici la stratégie appliquée : Un pointeur p est initialisé au pointeur tete, le load() donnant accès à la valeur de cette variable atomique Si p est nul initialement, alors la pile est vide, et nous quittons tout simplement en signalant ce fait au code client La mécanique de permutation atomique qui suit, ou compare_exchange_weak(), exprime l'idée suivante : « Tant que je ne serai pas parvenu à placer p->succ à la place de tete, mets la valeur de tete telle que lue ici dans p » Cette répétitive se poursuivra tant que notre méthode ne sera pas celle qui aura réussi à remplacer la valeur de tete par celle de p->succ. À chaque échec (chaque cas où un tiers aura modifié tete et interféré avec notre propre code), la valeur qui sera constatée dans tete sera placée dans p. Ici, s'il s'avère que le p obtenu ce faisant soit nul, cela signifie qu'un tiers a extrait le dernier noeud de la pile et que celle-ci est désormais vide; dans ce cas, nous quittons en signalant ce fait au code client Une fois l'exécution de la répétitive terminée, nous avons un accès exclusif à p, qui n'est techniquement plus sur la pile. Nous pouvons donc en extraire la valeur en toute quiétude, le supprimer et signaler au code client le succès de l'extraction.	`bool pop(T &val) { noeud *p = tete.load(); if (!p) return false; while (!tete.compare_exchange_weak(p, p->succ)) if (!p) return false; // Ici, on a acces exclusif à p, qui était la tete auparavant val = p->val; delete p; return true; }`
Puisque nous gérons manuellement nos noeuds, le destructeur de lock_free::stack doit explicitement se débarrasser de ceux encore sous sa gouverne. Notre implémentation détruira simplement les noeuds un à la fois, tant qu'il en restera. Pour alléger le code, nous passerons ici par raw_pop(), une version privée et simplifiée de pop() qui ne se préoccupe ni de synchronisation, ni de la valeur des noeuds.	`~stack() { while (raw_pop()) ; }`
L'implémentation de raw_pop() est décrite à droite. Simplement, cette méthode lit la valeur de la tête (un pointeur atomique, ce qui explique le load()), détruit ce noeud et passe au suivant, retournant true seulement si un noeud a été détruit. Dans le cas où la tête est nulle, donc où la pile est vide, la méthode retourne false ce qui permet au code client de constater cet état de fait.	`private: bool raw_pop() { auto tete_cur = tete.load(); if (!tete_cur) return false; auto p = tete_cur->succ; delete tete_cur; tete.store(p); return true; } }; } #endif`

Ouf. C'est plein de subtilités, n'est-ce pas?

Programme (artificiel) de test

Nous utiliserons ces deux implémentations à l'aide d'un programme de test un peu artificiel. J'insiste sur le côté artificiel du test, car sans avoir un reflet fidèle d'un cas d'utilisation réel de nos structures de données, le raisonnement sur nos résultats ne sera pas immédiat et notre analyse devra être des plus prudentes.

Outre nos deux implémentations de piles synchronisées, le code de test présenté ici sera strictement basé sur des outils standards. Notez que j'ai évité un using namespace std pour éviter toute confusion entre nos classes stack et std::stack.	`#include "lock_free_stack.h" #include "lock_full_stack.h" #include <thread> #include <iostream> #include <utility> #include <chrono> using namespace std::chrono;`
Pour faire semblant que le code de tests utilisera les valeurs extraites des diverses piles, j'ai écrit la fonction unused() qui fait semblant d'utiliser le paramètre qui lui est passé. C'est un pur mensonge, mais c'est un truc bien pratique que j'ai pris à Herb Sutter.	`template <class T> void unused(T &&) { }`
La fonction tester(f,args...) appellera f(args...) et retournera une paire contenant le résultat de cet appel et le temps requis pour son exécution, sans plus. Elle ne serait pas à propos si f(args...) était void, mais ce ne sera pas un irritant ici.	`template <class F, class ... Args> auto tester(F f, Args && ... args) { auto avant = system_clock::now(); auto result = f(std::forward<Args>(args)...); return std::make_pair(result, system_clock::now() - avant); }`
Le programme de test comparera les deux implémentations de piles synchronisées sous deux angles : Un où les accès sont concurrents, et avec une très forte contention (un thread fera des push() sans arrêt et l'autre n'aura de cesse de faire des pop()), et Un autre qui fera tous les push(), puis tous les pop(), le tout en séquence, donc où il n'y aura aucune contention L'idée est de mesurer l'impact de la synchronisation dans deux perspectives un peu extrêmes (et artificielles). Nous comptabiliserons chaque fois le nombre de push() et le nombre de pop(), mais notez que ces mesures sont d'une utilité limitée : dans le cas séquentiel, il y aura autant de pop() que de push(), alors que dans le cas concurrent, les pop() seront typiquement moins nombreux que les push() car un signal d'arrêt sera donné par le dernier push() et interrompra le travail du thread réalisant les extractions. Nous n'aurons pas de fuites de ressources dû au destructeur de notre pile. Nos deux fonctions de tests retourneront le nombre de push() et le nombre de pop(), pour fins d'affichage et pour enrichir l'analyse.	`template <template <class> class S> std::pair<int, int> test_concurrent_stack(int n) { S<int> ze_stack; using namespace std; atomic<bool> fini{}; int npushes = 0, npops = 0; thread popper{ [&ze_stack, &fini, &npops]() { while (!fini) { int val; if (ze_stack.pop(val)) { unused(val); ++npops; } } } }; thread pusher{ [&ze_stack, &fini, &npushes, n]() { for (int i = 1; i <= n; ++i) { ze_stack.push(i); ++npushes; } fini = true; } }; popper.join(); pusher.join(); return make_pair(npushes, npops); }`
	`template <template <class> class S> std::pair<int, int> test_stack(int n) { S<int> ze_stack; using namespace std; int npushes = 0, npops = 0; for (int i = 1; i <= n; ++i) { ze_stack.push(i); ++npushes; } int val; while (ze_stack.pop(val)) { unused(val); ++npops; } return make_pair(npushes, npops); }`
Le programme de test appelle simplement les fonctions ci-dessus et affiche les résultats obtenus.	template <class S> void afficher_stats(const char *nom_test, S stats) { using namespace std; auto npushes = stats.first.first; auto npops = stats.first.second; auto dt = stats.second; cout << non_test << "\n\tNb push: " << npushes << "\n\tNb pop: " << npops << "\n\tTemps total: " << duration_cast<milliseconds>(dt).count() << " ms." << "\n\tNb operations / milliseconde: " << static_cast<double>(npushes + npops) / duration_cast<milliseconds>(dt).count() << endl; } int main() { using namespace std; enum { N = 5'000'000 }; cout << "Sans concurrence, aucune contention\n"; auto free_stats_seq = tester(test_stack<lock_free::stack>, N); auto full_stats_seq = tester(test_stack<lock_full::stack>, N); afficher_stats("Lock free", free_stats_seq); afficher_stats("Lock full", full_stats_seq); cout << "-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-\n"; cout << "Avec concurrence, forte contention\n"; auto free_stats = tester(test_concurrent_stack<lock_free::stack>, N); auto full_stats = tester(test_concurrent_stack<lock_full::stack>, N); afficher_stats("Lock free", free_stats); afficher_stats("Lock full", full_stats); }

Résultats et analyse

Un résultat d'exécution possible pour ce programme jouet serait :

Sans concurrence, aucune contention
Lock free
        Nb push: 5000000
        Nb pop: 5000000
        Temps total: 383 ms.
        Nb operations / milliseconde: 26109.7
Lock full
        Nb push: 5000000
        Nb pop: 5000000
        Temps total: 467 ms.
        Nb operations / milliseconde: 21413.3
-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
Avec concurrence, forte contention
Lock free
        Nb push: 5000000
        Nb pop: 4999999
        Temps total: 899 ms.
        Nb operations / milliseconde: 11123.5
Lock full
        Nb push: 5000000
        Nb pop: 4964462
        Temps total: 1544 ms.
        Nb operations / milliseconde: 6453.67

En pratique, pour la version séquentielle, le code sans verrous sera toujours plus rapide que le code avec verrou, et ce par une bonne marge. La différence entre ces deux implémentations sera exacerbée en mode Debug, où le code généré sera plus naïf, mais même en mode Release (ci-dessus) on peut voir une bonne différence entre les deux.

Pour le mode à forte contention, dans l'exécution montrée ci-dessus, il se trouve que la version sans verrous a été significativement plus rapide que la version avec verrous, mais ce constat cache un piège, et vous constaterez que sur plusieurs exécutions, il arrive fréquemment que la version avec verrous soit plus rapide. Il ne faut pas en tirer de constats trop fermes cependant, une version avec autant de contention n'étant pas nécessairement représentative de cas d'exécution réalistes.