Algorithme du boulanger

L'article suppose que vous avez lu la présentation générale des algorithmes de synchronisation classiques, de même que Algos-Section-Critique-General.html qui explique la démarche d'ensemble, présente les risques et décrit les classes Producteur et Consommateur qui sont réinvesties ici.

On le nomme parfois aussi algorithme de la boulangerie, ou encore simplement algorithme de Lamport, mais l'ami Lamport a plusieurs algorithmes connus à son actif alors...

Le code qui suit est une implémentation de l'algorithme du boulanger, de Leslie Lamport. Cet algorithme permet la synchronisation de $n$ unités d'exécution (threads, processus) partageant une zone de mémoire, sans avoir recours à des outils de synchronisation comme ceux que l'on trouve dans les systèmes d'exploitation contemporains. Il repose sur un ingénieux système de votation.

Intuition générale : on cherche à atteindre un consensus en procédant à un vote, tenant compte d'un identifiant unique à chaque unité d'exécution. Ceci permet de créer un ordonnancement pour départager des situations qui seraient, autrement, ambiguës.

Le principe

Le principe de cet algorithme va comme suit :

Diverses unités d'exécution, munies d'identifiants entiers distincts les uns des autres

Lorsqu'une unité souhaite accéder à la section critique, elle démarre un vote de manière visible (un booléen dans un tableau à l'indice correspondant à son identifiant)

Tout demandeur d'accès à la section critique vote en insérant une valeur entière strictement positive dans un tableau de votes, à l'indice correspondant à son identifiant (ceci correspond à un signal de priorité d'accès, défini au préalable)

Une unité d'exécution quittant la zone critique remet son vote à zéro

L'unité $i$ a accès à la section critique quand son vote est l'entrée correspondant à la valeur la plus élevée

Détails spécifiques à l'algorithme du boulanger

L'algorithme du boulanger utilise deux états partagés, placés dans etats_synchro à droite :

Un tableau de booléens, un par unité de traitement, où l'entrée à un indice donné est vraie seulement pendant que l'unité de traitement portant ce numéro est en train de piger un numéro (en quelque sorte, quand elle est en train de voter), et
Un tableau d'entiers, un par unité de traitement, où l'entrée à un indice donné est le numéro pigé par cette unité de traitement (son vote), ou zéro si elle ne participe pas au vote en cours

Le service lock() permet d'accéder à la section critique. Sa mécanique va comme suit :

La période de pige d'un numéro se situe entre le moment où entering devient true et celui où il redevient false pour une unité d'exécution donnée. Plusieurs unités d'exécution peuvent piger concurremment
La pige est simple : identifier la plus haute valeur déjà pigée, et y ajouter un. Ceci ne garantit par l'unicité de la valeur pigée, dû au caractère concurrent de ce segment
Ensuite, pour chaque unité d'exécution, attendre la fin de sa pige (s'il y a lieu) puis attendre que l'unité de traitement ait eu l'opportunité de s'exécuter. L'ordre d'exécution est simple : le plus petit entier pigé passe d'abord, et dans le cas où les circonstances font en sorte que deux unités d'exécution ont pigé la même valeur, la priorité est donnée à l'unité de traitement ayant le plus petit identifiant
Quand l'unité d'exécution active a examiné toutes les autres unités d'exécution, alors c'est nécessairement à son tour

Quitter la zone critique implique remettre le vote à zéro, tout simplement, ce qui assurera un bon redémarrage lors de la prochaine votation.

#include "Producteur.h"
#include "Consommateur.h"
#include <cassert>
#include <algorithm>
#include <iostream>
#include <string>
#include <fstream>
#include <thread>
#include <random>
using namespace std;

class etats_synchro
{
   enum { NTHREADS = 2 };
   bool entering[NTHREADS];
   int numero[NTHREADS];
public:
   etats_synchro()
   {
      fill(begin(entering), end(entering), false);
      fill(begin(numero), end(numero), 0);
   }
   void lock(int n)
   {
      entering[n] = true;
      numero[n] = 1 + *(max_element(begin(numero), end(numero)));
      entering[n] = false;
      for (int i = 0; i < NTHREADS; i++)
      {
         // attendre que le thread i ait pigé un numéro
          while (entering[i])
            ;
         // Attendre que tous les threads ayant un plus petit numéro que nous
         // aient fini leur boulot. Si deux threads ont pigé le même numéro (une
         // possibilité ici), alors on celui qui a le plus petit indice a priorité
         while (numero[i] &&
                ((numero[i] < numero[n]) || ((numero[i] == numero[n]) && (i < n))))
            ;
      }
   }
   void unlock(int n)
      { numero[n] = 0; }
};

L'algorithme du boulanger est implémenté par la fonction lamport(), à droite, exécutée dans un thread distinct. Le code clé de cet algorithme est celui qui gère l'entrée dans la section jugée critique, de même que celui qui en gère la sortie.

L'une des grandes vertus de cet algorithme est qu'en surface, il a le comportement attendu des verrous avec lesquels nous sommes plus familiers aujourd'hui. En effet, même dans la terminologie, nous pouvons voir poindre ce qui deviendra éventuellement le mutex que nous connaissons bien.

template <class T>
   void lamport(T && oper, etats_synchro &etats, bool &fin)
   {
      auto id = oper.id;
      while (!fin)
      {
         oper.avant_partie_critique();
         etats.lock(id);
         oper.partie_critique();
         etats.unlock(id);
         oper.apres_partie_critique();
      }
   }

Sans grande surprise, le programme principal lance les diverses unités de traitement souhaitées, assure leur identification correcte et leur supplée le booléen qu'elles partageront et qui jouera pour elles le rôle d'un signal de fin, puis attend la fin des threads et nettoie les ressources attribuées au préalable.

int main()
{
   etats_synchro etats;
   random_device rd;
   mt19937 prng{ rd() };
   auto gen_work = [&]() -> int {
      uniform_int_distribution<int> de{1,50};
      return de(prng);
   };
   ofstream sortie{"out.txt"};
   clog.rdbuf(sortie.rdbuf());
   bool fin = {};
   int id = {};
   string transit;
   Producteur prod{ id++, transit, gen_work };
   Consommateur cons{ id++, transit, sortie };
   thread th[] = {
      thread{[&] {
         lamport(prod, etats, fin);
      }},
      thread{[&] {
         lamport(cons, etats, fin);
      }}
   };
   char c; cin >> c; fin = true;
   for(auto &thr : th)
      thr.join();
   clog.rdbuf(nullptr);
}