Files sans verrous

Quelques raccourcis :

Ce qui suit détaille un exemple complet et opérationnel de programme implémentant un (en fait, plusieurs, mais de manière générique) duo concurrent comprenant un producteur et un consommateur, transigeant à travers une file d'attente sans verrous. Nous verrons trois implémentations :

Dans les trois cas, l'exposé d'origine est de Herb Sutter lui-même. Le code proposé ici n'est qu'une adaptation maison de son propre discours. Je vous invite à lire les articles originaux pour enrichir l'éclairage que vous pourrez en tirer.

Outils généraux

Quelques outils généraux apparaîtront ci-dessous. Les voici.

Fonction trim()

Pour des raisons algorithmiques, nous aurons recours à une fonction éliminant les blancs aux extrémités d'une chaîne de caractères. Plusieurs langages offrent une telle fonction de manière standard, mais pas C++ (à ma connaissance), donc en voici une « maison ».

Nous utiliserons pour ce faire la fonction visible à droite, mais vous pouvez en implémenter une par vous-mêmes si vous le souhaitez (c'est amusant à coder).

 
//
// ... inclure <string>, <algorithm> et <locale>...
//
template <class C>
   std::basic_string<C> trim(std::basic_string<C> s) {
      using namespace std;
      auto loc = locale{ "" };
      auto pred = [&](C c) { return !isspace(c, loc); };
      auto ndeb = distance(begin(s), find_if(begin(s), end(s), pred));
      auto nfin = distance(s.rbegin(), find_if(s.rbegin(), s.rend(), pred));
      return (ndeb || nfin)? s.substr(ndeb, s.size() - (ndeb + nfin)) : s;
   }

Fonction get_cache_line_size()

Sous Microsoft Windows, l'implémentation de get_cache_line_size() que j'ai utilisé est présentée à droite. Il est probable que cette approche soit perfectible (pour moi, c'était du code un peu jetable), mais si ça peut vous rendre service...

Dans la version C++ 11 du code proposé, cette fonction est la seule qui ne soit pas portable. Évidemment, dans la version pré-C++ 11, par la force des choses, il y a un peu plus de code propre à la plateforme.

 
class NoCacheLineDescriptionFound {};
struct PlusDeProcesseursQue {
   int nb_estime;
   PlusDeProcesseursQue(int nb_estime)
      : nb_estime{ nb_estime }
   {
   }
};
int get_cache_line_size();
//
// ...
//
#include <windows.h>
int get_cache_line_size() {
   enum { NPROCS = 16 }; // on y va « safe »
   SYSTEM_LOGICAL_PROCESSOR_INFORMATION buff[NPROCS];
   unsigned long return_length = sizeof(buff);
   if (!GetLogicalProcessorInformation(buff, &return_length))
      throw PlusDeProcesseursQue{ NPROCS };
   for (unsigned long i = 0; i < return_length; ++i)
      if (buff[i].Relationship == RelationCache) {
         CACHE_DESCRIPTOR &cache = buff[i].Cache;
         if (cache.Level == 1)
            return cache.LineSize;
      }
   throw NoCacheLineDescriptionFound{};
}

Programme de test

Nous monterons tout d'abord le programme utilisant les files sans verrous, dans le but de clarifier la démarche qui sera suivie par la suite et de faciliter la lecture du reste du code.

Tout d'abord, notons que pour les fins de l'exemple, nous implémenterons les trois cas annoncés en introduction dans un même fichier d'en-tête, lock_free_queue.h, dans des espaces nommés distincts (version_0, version_1 et version_2 respectivement).

Le programme passera d'une implémentation à l'autre par une simple directive using. Le nom utilisé pour isoler les diverses versions de file sans verrou sera file_test<T>::type dans l'un ou l'autre de ces espaces nommés (les noms des véritables classes varieront selon les implémentations).

Si votre compilateur n'est pas tout à fait à jour, vous pouvez remplacer dans le code chaque occurrence de

file_synchro<T>

...par...

typename file_test<T>::type;
 
#include "lock_free_queue.h"
#include <iostream>
#include <string>
#include <functional>
#include <cassert>
#include <locale>
#include <algorithm>
#include <atomic>
#include <thread>
using namespace std;
//using version_0::file_test;
//using version_1::file_test;
using version_2::file_test;
template <class T>
   using file_synchro = file_test<T>;

Les threads producteur et consommateur manipuleront chacun deux entités, soit une file synchronisée sans verrous et un booléen permettant de signaler la fin des opérations. Remarquez au passage :

Les threads producteur et consommateur seront génériques (oui, c'est légal), pour que le code client puisse les générer sur la base des types à transiger. Puisque nous avons accès aux threads standards de C++ 11, le code proposé ici sera relativement simple. En effet :

  • La fonction producteur<T,G>() est générique sur la base du type des données à manipuler et sur la base d'une source de données de ce type (l'objet gen de type G). Tant que fini ne sera pas devenu vrai, la file synchronisée sans verrous (l'objet file) sera alimenté par ce que produira gen
  • La fonction consomateur<T>() est aussi générique sur la base du type des données à manipuler. Tant que fini ne sera pas devenu vrai, elle consommera un T de la la file synchronisée sans verrous (l'objet file) et affichera ce T à la console – ceci est fait pour fins de débogage, car ce n'est pas du tout nécessaire à notre démonstration

Dans cet exemple, il sera nécessaire que le type T expose un constructeur par défaut et qu'il soit possible d'en sérialiser une instance sur un flux en sortie.

 
template <class T, class G>
   void producteur(file_synchro<T> &file, G gen, atomic<bool> &fini) {
      while (!fini)
         file.produce(gen());
   }
template <class T>
   void consommateur(file_synchro<T> &file, atomic<bool> &fini) {
      for (T val; !fini; )
         if (file.consume(val))
            cout << "Consomme \"" << val << '\"' << endl;
   }

L'un des générateurs de valeurs utilisés dans le thread producteur sera un foncteur, plus précisément une instance du type decoupeur proposé à droite, qui générera mot à mot et de manière cyclique le contenu d'un message significatif. Notez que son attribut texte ne terminera jamais par un blanc (il s'agit d'un invariant qu'assure le constructeur de cette classe), ce qui permet d'alléger le code de son opérateur (). Je n'ai pas utilisé une expression λ ici (ça aurait été lourd, il me semble).

Le lien entre cur et texte est une sorte d'invariant silencieux. C'est la raison pour laquelle j'ai représenté cur par un indice plutôt que par un itérateur : si cur était un itérateur, dans le cas où un decoupeur serait copié, sa copie serait un itérateur sur la string originale, pas sur la copie... Nous aurions alors un sérieux problème sur les bras, qui pourrait rester silencieux selon le contexte d'utilisation.

Utiliser un indice, puis recalculer les itérateurs correspondants au besoin, est ici un choix plus sain.

 
class decoupeur {
   string texte;
   string::size_type cur;
   locale loc;
public:
   decoupeur(const string &msg = "J'aime mon prof, bon, pis c'est d'meme!", const locale &loc = locale(""))
      : texte{trim(msg)}, loc{loc}
   {
      assert(!texte_.empty());
      cur = {};
   }
   string operator()() {
      auto it = find_if(begin(texte)+cur, end(texte), [&](char c) {
         return isspace(c, loc);
      });
      auto s = string(begin(texte)+cur, it);
      if (it == end(texte))
         cur = {};
      else
         cur = distance(
            begin(texte), find_if(it, end(texte), [&](char c) {
               return !isspace(c, loc);
            })
         );
      return s;
   }
};

Enfin, le code de test est proposé à droite. On y remarquera une fonction tester() générique sur la base du type à transiger et du type de générateur utilisé par le thread producteur, de même qu'un programme principal relativement simple.

La fonction tester() crée une file sans verrous correctement typée et un signal de fin atomique, lance deux threads qui opéreront sur ces objets (et, dans le cas du producteur, sur l'objet gen), puis lit une touche et assure la bonne terminaison des threads.

Remarquez la ligne en commentaires, qui permet d'afficher à la console la taille d'une Cache Line pour la cache L1 sur un ordinateur donné avec la fonction get_cache_line_size(). Cette information servira dans l'une des implémentations de file sans verrou (plus bas), mais ne peut pas (à ma connaissance) être obtenue de manière statique, donc à la compilation. Pour l'obtenir, j'ai exécuté mon programme une 1re fois, affiché cette valeur, puis utilisé cete information pour apposer une valeur à une constante dans l'implémentation pour laquelle elle s'avérait opportune.

Dans le programme principal, la génération d'entiers passe par une λ. Notez le mot-clé mutable appliqué à la λ, d'ailleurs :

  • La valeur initiale générée sera une copie de la variable locale cur
  • Par défaut, en C++, les valeurs capturées par copie dans une λ sont const
  • Ici, nous souhaitons que la λ retourne un entier différent à chaque appel, donc que la valeur capturée soit modifiable
  • C'est ce que permet le mot-clé mutable dans ce cas-ci
 
template <class T, class G>
   void tester(G gen) {
      file_synchro<T> file;
      atomic<bool> fini { false };
      thread th [] = {
         thread(producteur<T,G>, ref(file), gen, ref(fini)),
         thread(consommateur<T>, ref(file), ref(fini))
      };
      char c;
      cin >> c;
      fini = true;
      for (auto &thr : th)
         thr.join();
   }
int main() {
   //cout << get_cache_line_size() << endl;
   {
      int cur = 0;
      tester<int>([=]() mutable { return cur++; });
   }
   tester<std::string>(decoupeur{});
}

Les files sans verrous elles-mêmes

Abordons maintenant les ébauches de files sans verrous, en commençant par une implémentation qui est presque correcte, mais comprend une condition de course sournoise. Nous ferons ensuite une digression par une brève discussion de l'atomicité, et nous reviendrons vers deux implémentations correctes de l'idée de file sans verrous.

Implémentation 0

La première implémentation examinée ici est due à Petru Marginean (article original : http://www.drdobbs.com/high-performance-computing/208801974). Elle semble fonctionner mais en fait, elle comprend des conditions de course (pour une critique de Sutter en ce sens : http://www.drdobbs.com/cpp/210600279). Ce qui est sournois avec de telles conditions de course est qu'elles tendent à ne causer de problèmes que de manière occasionnelle, et sont donc très difficiles à dépister et à régler.

Cette première implémentation supporte un seul producteur et un seul consommateur. J'ai conservé le nom de classe utilisé dans l'article original mais je me suis permis quelques ajustements pour les fins de la présentation.

La proposition de Marginean repose sur une encapsulation d'une liste doublement chaînée standard, std::list, tirée directement de STL. Elle tient à jour des marqueurs sur la tête et sur la queue de cette liste; ces marqueurs sont importants du fait qu'ils isolent le substrat sous-jacent (std::list) de la logique de la liste sans verrous elle-même.

Modifier tete signifiera publier le retrait d'un élément de la liste (sa consommation), alors que modifier queue signifiera y publier l'insertion d'un élément. Puisque tete ne sera pas nécessairement égal à liste.begin(), les éléments dans l'intervalle (liste.begin()..tete( devront être nettoyés au moment opportun, dans l'optique d'éviter de polluer la mémoire toute entière.

Les méthodes clés de chacune des trois files sans verrous que nous explorerons seront :

  • Son constructeur par défaut, qui mettra en place les invariants essentiels de la classe
  • La méthode produce(), permettant d'y insérer un élément
  • La méthode consume(), permettamt d'essayer d'en retirer un élément (et retournant true seulement si cela a fonctionné, donc seulement si la file n'était pas vide), et
  • Un destructeur, du moins dans deux des trois implémentations. Dans le cas de l'implémentation de Marginean, que nous examinons ici, le destructeur serait superflu du fait que la gestion toute entière des données est encapsulée dans le substrat std::list
 
#ifndef LOCK_FREE_QUEUE_H
#define LOCK_FREE_QUEUE_H
#include <list>
#include <algorithm>
#include <atomic>
namespace version_0 {
   template <class T>
      class LockFreeQueue {
         using container_type = std::list<T>;
         container_type liste;
         using iterator = typename container_type::iterator;
         iterator tete, queue;

Ici, le constructeur par défaut insère initialement un élément bidon dans la file, pour marquer une distance entre le début (queue) et la fin (tete) de la file. Un T par défaut est utilisé à cette fin ici, ce qui impose une contrainte sur le type T utilisé dans cete implémentation. Cette démarcation ente le début et la fin est une clé de l'approche mise de l'avant ici : si cet invariant est maintenu, alors le consommateur (unique) et le producteur (aussi unique) ne toucheront jamais au même noeud.

L'insertion d'un élément se fait en fin de liste (liste.push_back()), mais n'est publié (modification à queue) qu'une fois l'insertion réussie. Au sens du consommateur, tant que la modification à queue n'a pas pris effet, l'élément n'est pas encore réellement inséré dans la file.

Marginean préconise de saisir cette opportunité pour faire du nettoyage des éléments inutilisés : invoquer erase() sur l'intervalle (begin(liste)..tete( en utilisant bien sûr une copie de ces deux itérateurs permet d'y arriver.

L'extraction d'un élément tient d'abord compte du fait qu'il y a une démarcation d'un noeud entre la tete et la queue : si queue est juste après tete, alors la file est considérée vide. Par la suite, tete_ est déplacé, l'élément est saisi (T::operator=(const T&) doit être défini) et un signal de réussite est retourné au code client.

 
      public:
         LockFreeQueue(const LockFreeQueue&) = delete;
         LockFreeQueue& operator=(const LockFreeQueue&) = delete;
         LockFreeQueue() {
            liste.push_back(T{}); // marqueur de démarcation
            tete = std::begin(liste);
            queue = std::end(liste);
         }
         void produce(const T &elem) {
            liste.push_back(elem); // ajout
            queue = std::end(liste);  // publication de l'ajout
            liste.erase(std::begin(liste), tete); // nettoyage
         }
         bool consume(T &elem) {
            iterator prochain = tete;
            ++prochain;
            if (prochain != queue) { // vide?
               tete = prochain; // publication de la saisie
               elem = *tete;     // remise de l'élément au client
               return true;      // tigidou
            }
            return false;        // elle était vide, finalement
         }
      };

Pour les fins des tests présentés ici, même si les diverses files sans verrous que nous utiliserons porteront des noms distincts, nous regrouperons ces noms sous une seule et même dénomination. Ainsi, dans chaque espace nommé, le type file_test<T> correspondra au type de file sans verrou du même espace.

 
   template <class T>
      using file_test = LockFreeQueue<T>;
}
// ... à suivre ...

Problémes de la version 0

Quels sont les problèmes de l'implémentation de Marginean? Ici encore, Sutter nous éclaire.

Le principal problème et que les attributs tete_ et queue_ sont sujets à des conditions de course. Il faudrait que les opérations sur ces attributs soient à la fois atomiques et ordonnancées, au sens du mot clé volatile dans son acception Java ou C#, mais ils ne sont ni l'un, ni l'autre. Une entité sur laquelle les opérations sont atomiques doit typiquement être au plus aussi grande qu'un mot mémoire. En effet, l'atomicité (surtout sur une plateforme munie de plusieurs processeurs ou de plusieurs coeurs) exige un support du substrat matériel (du code machine) pour permettre des lectures et des écritures atomiques par un processeur.

À propos de l'atomicité, un article du Code Project (oui, je sais, c'est pas toujours très rigoureux, mais certains sont pas si mal) sur les approches sans verrous, disponible à l'adresse http://www.codeproject.com/KB/threads/LockFree.aspx, décrit une opération atomique comme étant « an operation that is guaranteed to finish and not be interrupted half-way by another thread ». L'article poursuit :

« On almost all modern processors, reads and writes of naturally aligned native types are atomic. This means that as long as the memory bus is at least as wide as the type being read or written the CPU reads and writes will happen in a single bus transaction. This makes it impossible for other threads to see them half-completed. For x86 and x64 platforms, types larger than 8 bytes are not guaranteed to be atomic. »

Sur un entier, l'affectation d'un littéral (int i = 3;) sera typiquement atomique mais une opération d'autoincrémentation (i++) pourra prendre jusqu'à trois opérations machines, et ne le sera donc pas.

Pour ce qui est des risques de réordonnancement, examinons quelques lignes de la méthode produce() (à droite). Les instructions A et B peuvent sembler indépendantes l'une de l'autre (car liste.end() n'est pas le dernier noeud valide de liste mais bien ce qui suit ce noeud); elles risquent donc d'être réordonnancées par un processeur, ce qui peut poser un risque du fait que l'écriture sur queue est une publication d'état.

 
void produce(const T &elem) {
   liste.push_back(elem); // A
   queue = end(liste);  // B
   // ...

Dans le même ordre d'idées, examinons quelques lignes de la méthode consume() (à droite). Cet extrait pose deux risques :

  • Tout d'abord, la modification de tete est une publication d'état, qui a lieu avant que la consommation du contenu pointé ne se soit faite. Heureusement, cela peut être aisément corrigé
  • Ensuite, même sans cet irritant, la possible réorganisation du code pose encore ici problème : les instructions C et D semblent (à une réécriture près) indépendantes l'une de l'autre et peuvent être réordonnancées par un processeur, ce qui peut poser un risque du fait que l'écriture sur tete est une publication d'état
 
bool consume(T &elem) {
   // ...
   if (prochain != queue) {
      tete = prochain; // C
      elem = *tete;    // D
      // ...

À propos du réordonnancement, nous devons être prudents face aux réorganisations du code susceptibles d'être réalisées soit par le compilateur, soit par le processeur.

Les réorganisations réalisées par un compilateur peuvent être en partie prévenues par le recours à la qualification volatile. Le standard ISO de 1998 (qui n'est pas le plus détaillé en ce qui a trait à la multiprogrammation, c'est le moins qu'on puisse dire) définit le comportement d'un programme C++ en termes du comportement attendu d'une machine abstraite, indiquant entre autres :

 « The observable behavior of the abstract machine is its sequence of reads and writes to volatile data and calls to library functions » (§1.9, 6) et « Accessing an object designated by a volatile lvalue, modifying an object, calling a library function, or calling a function that does any of these operations are all side effects, which are changes in the state of the execution environment. Evaluation of an expression might produce side effects. At certain specified points in the execution sequence called sequence points, all side effects of previous evaluations shall be complete and no side effects of subsequent evaluations shall have taken place  » (§1.9, 7).

Autre problème : le compilateur peut appliquer des transformations brisant notre logique. Par exemple, les deux cas proposés à droite sont possibles :

  • Le premier cas (en haut) introduit une valeur bidon pour faciliter le débogage, et s'avère une technique répandue avec plusieurs compilateurs. Dans notre logique, toutefois, la non-écriture dans tete avant que nous ne l'ayons jugé opportun est crucial, l'écriture dans cet attribut constituant l'une des clés de l'algorithme
  • Le second cas montre une optimisation spéculative où l'attribut tete est modifié d'abord et où la modification est annulée ultérieurement si le constat est fait que cette écriture était mal venue. Dans le cas de l'algorithme proposé ici, cette optimisation brise la sémantique du code

Clairement, bien que cette première implémentation fonctionne dans bien des cas, elle est à risque, et nous ne pouvons pas nous arrêter ici. 

if (prochain != queue) {
   tete = 0xDEADBEEF; // oups!
   tete = prochain;
   elem = *tete;
 
__temp = tete;
tete = prochain;       // spéculatif
if (prochain == queue) // note: test inversé
   tete = __temp;      // annuler si la spéculation fait patate
else
   elem = *tete;

Les réordonnancements réalisés par les processeurs varient selon les technologies. Un article de 2007, somme toute détaillé sur le sujet et expliquant comment Linux s'en sort, peut être consulté sur http://www.rdrop.com/users/paulmck/scalability/paper/ordering.2007.09.19a.pdf. Heureusement, C++ 11 encadre les optimisations possibles sur les variables atomiques, dans l'optique de garder le code généré séquentiellement cohérent; les variables atomiques nous permettent donc de garder une certaine capacité de raisonner à partir du code source, malgré tout.

Implémentation 1

Cette deuxième implémentation est due à Herb Sutter (article original : http://www.drdobbs.com/cpp/211601363). Elle repose sur des variables atomiques servant de verrous booléens pour les extrémités de la file. Cette version supporte un plusieurs consommateurs et plusieurs producteurs.

Le recours à diverses variables atomiques dans la structure même de la file sans verrous nous demande de nous assurer (sur Microsoft Windows, du moins) que chacune d'elles se trouve alignée sur la frontière du début d'une Cache Line. Avec un compilateur C++ 11 le supportant, nous aurions aussi pu utiliser alignas() pour en arriver au même résultat mais de manière plus élégante.

La taille de la Cache Line sur l'ordinateur sur lequel ce code a été écrit est de 64 bytes, valeur obtenue par un appel à la fonction get_cache_line_size(). Je ne connais pas d'approche de programmation statique pour obtenir ces informations au moment d'écrire ceci.

Cette approche construit manuellement une file, pour éviter le problème de la non-atomicité des itérateurs de std::list. Remarquez que tous les noeuds sont de la taille de la Cache Line (dû au padding qui se trouve inséré dans chacun d'eux), et remarquez que chaque attribut de la file est aussi aligné sur cette frontière.

Ici, deux simili verrous contrôlent les accès aux extrémités (consumerLock, un booléen atomique partagé entre les consommateurs, et producerLock, un booléen atomique partagé entre les consommateurs).

Les noeuds ne sont pas atomiques, mais les booléens qui contrôlent l'accès aux extrémités de la file, eux, le sont. 

// ... suite ...
namespace version_1 {
   static const int CACHE_LINE_SIZE = 64; // obtenu par un appel offline a get_cache_line_size()
   template <class T>
      class LowLockQueue {
         struct Node {
            Node(T* val)
               : val{val}, next{}
            {
            }
            T* val;
            std::atomic<Node*> next;
            char pad[CACHE_LINE_SIZE - sizeof(T*)- sizeof(std::atomic<Node*>)];
         };
         char pad0[CACHE_LINE_SIZE];
         // un consommateur à la fois
         Node* first;
         char pad1[CACHE_LINE_SIZE - sizeof(Node*)];
         // partagé entre les consommateurs
         std::atomic<bool> consumerLock;
         char pad2[CACHE_LINE_SIZE - sizeof(std::atomic<bool>)];
         // un producteur à la fois
         Node* last; 
         char pad3[CACHE_LINE_SIZE - sizeof(Node*)];
         // partagé entre les producteurs
         std::atomic<bool> producerLock;
         char pad4[CACHE_LINE_SIZE - sizeof(std::atomic<bool>)];

Le constructeur par défaut, à l'image de celui dans l'approche 0 plus haut, insère un noeud séparateur qulque peu bidon (ici, un noeud contenant un pointeur nul), et initialise les deux « verrous » à false.

Le destructeur, quant à lui, détruit à la fois les noeuds et ce vers quoi ils pointent. À titre de rappel, appliquer l'opérateur delete sur un pointeur nul est sans effet, donc même le noeud séparateur (qui contient un pointeur nul) peut être détruit sans problème de cette manière.

 
      public:
         LowLockQueue(const LowLockQueue&) = delete;
         LowLockQueue& operator=(const LowLockQueue&) = delete;
         LowLockQueue()
            : producerLock{}, consumerLock{}
         {
            first = last = new Node{nullptr};
         }
         ~LowLockQueue() {
            while(first) {
               Node* p = first;
               first = p->next;
               delete p->val; // sans danger si !(p->val_)
               delete p;
            }
         }
// ...

L'insertion d'un nouvel élément dans la file attend d'abord l'obtention de l'exclusivité d'accès sur producerLock (un Spin Lock, semblable à ce qu'on trouve dans une section critique de Windows) puis insère le nouveau noeud et publie l'insertion une fois celle-ci complétée.

Notez qu'en cas de levée d'exceptions sur new Node, ce code n'est pas sécuritaire (le new T{val} qui y serait exécuté en premier lieu mènerait à une fuite de ressources).

L'extraction d'un élément de la file attend d'abord l'obtention de l'exclusivité d'accès sur consumerLock (un autre Spin Lock), puis fait l'essentiel de ses opérations sur des variable temporaires (zeFirst et zeNext). La publication de nouveaux états (first = zeNext) n'est faite qu'une fois les opérations sur la file complétées.

La méthode exchange() d'une variable atomique décrit une opération « lire-modifier-écrire » qui retourne l'ancienne valeur de la variable ainsi modifiée. Ainsi, si une variable lock est de type atomic<bool>, alors l'expression suivante :

while(lock.exchange(true))
   ;

signifie « tant que j'essaie d'écrire true dans lock et qu'il s'avère que lock était déjà true auparavant » (car si lock était true avant la modification atomique, alors le verrou que cette variable représente n'était pas disponible à ce moment).

 
// ...
         void produce(const T &val) {
            Node* p = new Node{new T{val}};
            while (producerLock.exchange(true))
               ;   // obtenir l'exclusivite
            last->next = p;      // publier aux consommateurs
            last = p;            // avancer last
            producerLock = {};   // terminer l'exclusivite
         }
         bool consume(T &result) {
            while(consumerLock.exchange(true))
               ; // obtenir l'exclusivite
            Node* zeFirst = first;
            Node* zeNext = first->next;
            if(zeNext) {// si non-vide
               T* val = zeNext->val; // extraire la valeur du noeud
               zeNext->val = {};
               first = zeNext;       // avancer first_
               consumerLock = {};    // terminer l'exclusivite
               result = *val;        // recuperer la valeur
               delete val;             // nettoyage
               delete zeFirst;
               return true;	 		   // tigidou
            }
            consumerLock_ = {};  // terminer l'exclusivite
            return false;        // il n'y avait rien dans la file
         }
      };
   template <class T>
      using file_test = LowLockQueue<T>;
}
// ...

Implémentation 2

Cette troisième implémentation est aussi due à Herb Sutter (article original : http://www.drdobbs.com/high-performance-computing/210604448). Tout comme la première implémentation, elle évite les verrous booléens sur les extrémités, mais les remplace ici par des pointeurs atomiques.

Cette version est beaucoup plus simple et beaucoup plus courte que la précédente. En fait, elle s'apparente surtout à la version 0, à ceci près qu'elle n'a pas recours à std::list comme substrat mais bien à une implémentation maison, à base de pointeurs (à plus forte partie, de pointeurs atomiques).

 
// ...
namespace version_2 {
   template <class T>
      class LockFreeQueue {
         struct Node {
            Node(T val)
               : val{val}, next{}
            {
            }
            T val;
            Node* next;
         };
         Node* first;                       // seulement pour le producteur
         std::atomic<Node*> divider, last;  // partages

Le constructeur par défaut insère un noeud bidon dès le début, comme dans les deux autres approches. Ici, puisque les noeuds contiennent des valeurs (des T) plutôt que des indirections (des T*), le noeud séparateur contiendra un T par défaut (un T{}), ce qui impose une contrainte sur le type T.

 Le destructeur détruit tous les noeuds, incluant bien sûr le noeud bidon.

 
      public:
         LockFreeQueue(const LockFreeQueue&) = delete;
         LockFreeQueue& operator=(const LockFreeQueue&) = delete;
         LockFreeQueue() {
            // ajout d'un separateur bidon
            first = divider = last = new Node{T{}};
         }
         ~LockFreeQueue() {
            while(first) {
               Node* p = first;
               first = p->next;
               delete p;
            }
         }

L'approche globale respecte celle mise de l'avant pour l'approche 0. Une phase de nettoyage des éléments inutilisés est réalisée suite à la production d'une nouvelle valeur (ce qui ressemble à l'approche suivie par la plupart des moteurs de collecte d'ordures, qui réclament la mémoire inutilisée seulement lors d'une nouvelle allocation dynamique de mémoire).

La clé est le recours à des noeuds atomiques (en fait, à des pointeurs atomiques sur des noeuds) pour le séparateur et pour le point de consommation, adjointe à un moment choisi avec soin pour la publication des changements apportés à la structure (ici, au moment de modification de divider, qui suit la consommation d'un élément).

Notez qu'un pointeur atomique n'offre pas l'opérateur ->, ce qui explique par exemple le recours à l'écriture (*last).next plutôt qu'à last->next.

Les appels à load() sur les atomiques lors de la comparaison sont nécessaires. Un load() sur une atomique est une opération en lecture retournant une copie de la valeur de l'atomique au moment de l'appel.

En effet, on ne peut réaliser une comparaison avec == ou != qui soit atomique puisque cette opération est binaire (elle implique deux opérandes) alors que l'atomicité ne s'opère que sur une donnée à la fois.

Évidemment, il est possible que les valeurs de l'une ou de l'autre des atomiques ait changé entre le moment du plus récent load() et celui du test avec != alors il importe que l'algorithme n'en souffre pas.

 
         void produce(const T &elem) {
            (*last).next = new Node{elem};
            last = (*last).next;     // publication
            while(first != divider) { // collecte des ordures :)
               Node* p = first;
               first = first->next;
               delete p;
            }
         }
         bool consume(T &elem) {
            if(divider.load() != last.load()) { // si non-vide
               elem = (*divider).next->val;
               divider = (*divider).next; // publication
               return true;               // tigidou
            }
            return false;                  // elle était vide
         }
      };
   template <class T>
      using file_test = LockFreeQueue<T>;
}
#endif

Voilà.

Lectures complémentaires

Quelques liens pour enrichir le propos :

Valid XHTML 1.0 Transitional

CSS Valide !