Mutex portables

Ce qui suit montre brièvement comment il est possible de passer d'une implémentation non-portable d'un outil système, par exemple un mutex de Microsoft Windows, à une implémentation portable au sens où rien dans le code client ne laissera transparaître un couplage avec une plateforme spécifique. On y verra aussi comment automatiser la libération de ressources telles qu'un mutex ainsi construit, dans une optique de résilience et de stabilité.

Évidemment, si votre compilateur offre une implémentation standard des mutex et des fils d'exécution, utilisez-les.

Point de départ – version non-portable

Pour illustrer l'intérêt de la démarche, voici un petit programme parsemé d'artéfacts d'une plateforme spécifique (Microsoft Windows, mais on aurait pu faire de même avec une autre plateforme) :

Le programme principal conclura son exécution lorsque tous les fils d'exécution qu'il aura lancé se seront arrêtés. Ceci nous donne :

Fichier principal.cpp
#include <random>
#include <iostream>
#include <algorithm>
#include <vector>
#include <windows.h>
using namespace std;
unsigned long __stdcall chic_thread(void *);
HANDLE chic_mutex;
struct Etat {
   int &donnee;
   int indice;
   Etat(int &donnee, int indice) noexcept : donnee{ donnee }, indice{ indice } {
   }
};
int main() {
   enum { NTHREADS = 10 };
   HANDLE h[NTHREADS];
   vector<Etat> v;
   v.reserve(NTHREADS);
   int donnee = 10;
   chic_mutex = CreateMutex(0, FALSE, 0);
   for (int i = 0; i < NTHREADS; ++i) {
      v.emplace_back(donnee, i);
      h[i] = CreateThread(0, 0, chic_thread, &v[i], 0, 0);
   }
   WaitForMultipleObjects(NTHREADS, h, TRUE, INFINITE);
   for_each(begin(h), end(h), CloseHandle);
   CloseHandle(chic_mutex);
}
unsigned long __stdcall chic_thread(void *p) {
   auto &etat = *static_cast<Etat *>(p);
   random_device rd;
   mt19937 prng { rd() };
   uniform_int_distribution<> de{ 0, etat.indice };
   bool poursuivre = true;
   do {
      if (WaitForSingleObject(chic_mutex, 0)==WAIT_OBJECT_0) {
         etat.donnee = de(prng); // code risqué
         if (etat.donnee == 0) {
            cout << "Thread #" << etat.indice << "; arret" << endl;
            poursuivre = false;
         }
         ReleaseMutex(chic_mutex);
      }
      Sleep(1);
   } while(poursuivre);
   return {};
}

Cette utilisation de fils d'exécution synchronisés dépend, pour le moment, de l'emploi d'une variable globale (chic_mutex). Si nous devions passer à une autre plateforme (par exemple un système POSIX tel que Linux), nous devrions ajuster ou remplacer :

On peut comprendre qu'à moins de se destiner à une seule et unique plateforme, travailler à l'aide d'outils non-portables alourdit de manière significative la mise à jour et l'entretien d'un programme.

Pour en savoir plus sur l'idiome de classe Incopiable, voir ../Developpement/Schemas-conception.html#incopiable

La dualité construction et destruction du Mutex applique l'idiome RAII : ../Developpement/Schemas-conception.html#raii

Réduire le couplage superficiel – classe Mutex

Découpons maintenant le tout un peu plus pour en arriver à localiser le HANDLE représentant le mutex pour le système d'exploitation dans une instance d'une classe Mutex de notre cru. Parmi les avantages d'une telle approche, on trouve :

On obtiendra ce qui suit.

Mutex.h
#ifndef MUTEX_H
#define MUTEX_H
#include "Incopiable.h"
#include <windows.h>
class Mutex : Incopiable {
   HANDLE m;
public:
   Mutex() : m{ CreateMutex(0, FALSE, 0) } {
   }
   bool obtenir(int delai) const noexcept {
      return WaitForSingleObject(m, delai)) == WAIT_OBJECT_0;
   }
   bool obtenir() const noexcept {
      return obtenir(INFINITE);
   }
   void relacher() const noexcept {
      ReleaseMutex(m);
   }
   ~Mutex() {
      CloseHandle(m);
   }
};
#endif
Mutex.cpp
#include "Mutex.h" // et c'est tout!

Le fichier Mutex.cpp pourrait être omis.

Personnellement, j'aime bien avoir un fichier source même quand le code d'une classe tient au complet dans son fichier d'en-tête, pour être capable de ne compiler, au besoin, que cette classe. Cela sert en période de développement, quand on veut mettre au point une chose à la fois.

Suite à ce changement, le programme devient comme suit.

Fichier principal.cpp
#include "Mutex.h"
#include <random>
#include <iostream>
#include <algorithm>
#include <vector>
#include <windows.h>
using namespace std;
unsigned long __stdcall chic_thread(void *);
Mutex chic_mutex;
struct Etat {
   int &donnee;
   int indice;
   Etat(int &donnee, int indice) noexcept : donnee{ donnee }, indice{ indice } {
   }
};
int main() {
   enum { NTHREADS = 10 };
   HANDLE h[NTHREADS];
   vector<Etat> v;
   v.reserve(NTHREADS);
   int donnee = 10;
   for (int i = 0; i < NTHREADS; ++i) {
      v.emplace_back(donnee, i);
      h[i] = CreateThread(0, 0, chic_thread, &v[i], 0, 0);
   }
   WaitForMultipleObjects(NTHREADS, h, TRUE, INFINITE);
   for_each(begin(h), end(h), CloseHandle);
}
unsigned long __stdcall chic_thread(void *p) {
   auto &etat = *static_cast<Etat *>(p);
   random_revice rd;
   mt19937 prng { rd() };
   uniform_int_distribution<> de{ 0, etat.indice };
   bool poursuivre = true;
   do {
      if (chic_mutex.obtenir(0)) {
         etat.donnee = de(prng); // code risqué
         if (etat.donnee == 0) {
            cout << "Thread #" << etat.indice << "; arret" << endl;
            poursuivre = false;
         }
         chic_mutex.relacher();
      }
      Sleep(1);
   } while(poursuivre);
   return {};
}

On remarquera un allègement léger du programme principal. Nous avons fait du progrès, mais il reste encore beaucoup à faire.

Parenthèse pour C++ 11 : incopiable mais déplaçable

Il est sage de faire en sorte que Mutex ne soit pas copiable du fait que, si un Mutex était copiable, alors des programmes comme celui proposé ci-dessous planteraient dramatiquement mais seraient légaux aux yeux du compilateur.

#include "Mutex.h"
void f(Mutex) {
   Mutex m;
   f(m);
} // boum!

En effet, le Mutex local à main() est une variable dont la portée s'étend de sa définition à la fin du bloc dans lequel elle est définie (donc jusqu'à l'accolade fermante de main()). En permettant de la passer par valeur à f(), cela provoquerait une copie de m, ce qui copierait implicitement le HANDLE qui s'y trouve. La copie locale à f() y serait utilisée localement, puis détruite. La destruction de la copie de m solliciterait son destructeur, ce qui fermerait le HANDLE contenu dans m (une copie d'un HANDLE représente la même ressource que le HANDLE original, après tout). Ce faisant, à la fin de main(), l'original chercherait, de par le destructeur de m, à fermer un HANDLE déjà fermé, soit m.m... Boum!

Bloquer la copie, dans les circonstances, est clairement la chose à faire... mais depuis C++ 11, il est possible de définir une sémantique de mouvement pour des objets qui ne sont pas intrinsèquement copiables (entre autres – le mouvement peut être utile pour beaucoup d'autres types d'objets aussi, en particulier pour des objets qui sont copiables mais longs à copier).

Dans le cas d'un Mutex, l'ajout de la sémantique de mouvement donnerait la classe suivante.

Mutex.h
#ifndef MUTEX_H
#define MUTEX_H
#include "Incopiable.h"
#include <cassert>
#include <algorithm>
#include <windows.h>
class Mutex : Incopiable {
   HANDLE m;
public:
   Mutex() : m{ CreateMutex(0, FALSE, 0) } {
   }
   bool obtenir(int delai) const noexcept {
      assert(m);
      return WaitForSingleObject(m, delai)) == WAIT_OBJECT_0;
   }
   bool obtenir() const noexcept {
      return obtenir(INFINITE);
   }
   void relacher() const noexcept {
      assert(m);
      ReleaseMutex(m);
   }
   ~Mutex()  {
      if (m) CloseHandle(m);
   }
   Mutex(Mutex && autre) noexcept : m{ std::move(autre.m ) } {
      autre.m = INVALID_HANDLE_VALUE;
   }
   Mutex& operator=(Mutex && autre) noexcept {
      m = std::move(autre.m);
      autre.m = INVALID_HANDLE_VALUE;
      return *this;
   }
};
#endif
Mutex.cpp
#include "Mutex.h" // et c'est tout!

Les changements clés vont comme suit :

Meilleure synchronisation – verrou RAII

Les mutex sont de bons outils dans la mesure où ils sont bien utilisés. Cela signifie que les mutex doivent être obtenus au moment opportun, et relâchés précisément au bon moment. L'ennui principal d'une obtention et d'une libération manuelles d'un mutex est que, si une exception est levée pendant que le mutex est saisi par une méthode, alors ce mutex ne sera jamais relâché.

Par exemple, le code ci-dessous pourrait ne jamais relâcher le mutex m :

Mutex m; // globale... Beurk
void f();
void g() {
   m.obtenir();
   f(); // ceci lèvera-t-il une exception? On ne le sait pas ici
   m.relacher();
}

Dans de vrais systèmes, ce risque est bien réel, et est porteur de conséquences.

On peut résoudre ce problème avec ce que j'appellerai un Autoverrou, cas particulier d'objet RAII :

Une implémentation possible suit.

#ifndef MUTEX_H
#define MUTEX_H
// ... classe Mutex, omise pour fins d'économie ...
class Autoverrou {
   const Mutex &m;
public:
   Autoverrou(const Mutex &m) noexcept : m{ m } {
       m.obtenir();
   }
   ~Autoverrou() {
      m.relacher();
   }
};
#endif

Le secret ici est que dans un sous-programme donné, les destructeurs de toutes les variables locales seront appelés lors de la complétion du sous-programme, que cette complétion se produise suite à une levée d'exception ou qu'elle résulte de la fin normale du sous-programme (passage par return, ou rencontre de l'accolade fermante du sous-programme).

Du moment où une variable locale de type Autoverrou sera instanciée, le Mutex sera obtenu. À la fin de la vie d'un Autoverrou, son Mutex sera libéré. On utilisera donc des Autoverrou comme variables locales à des méthodes lorsqu'on voudra délimiter de manière sécurisée la durée d'obtention d'un Mutex.

Fichier principal.cpp
#include "Mutex.h"
#include <random>
#include <iostream>
#include <algorithm>
#include <vector>
#include <windows.h>
using namespace std;
unsigned long __stdcall chic_thread(void *);
Mutex chic_mutex;
struct Etat {
   int &donnee;
   int indice;
   Etat(int &donnee, int indice) noexcept : donnee{ donnee }, indice{ indice } {
   }
};
int main() {
   enum { NTHREADS = 10 };
   HANDLE h[NTHREADS];
   vector<Etat> v;
   v.reserve(NTHREADS);
   int donnee = 10;
   for (int i = 0; i < NTHREADS; ++i) {
      v.emplace_back(donnee, i);
      h[i] = CreateThread(0, 0, chic_thread, &v[i], 0, 0);
   }
   WaitForMultipleObjects(NTHREADS, h, TRUE, INFINITE);
   for_each(begin(h), end(h), CloseHandle);
}
unsigned long __stdcall chic_thread(void *p) {
   auto &etat = *static_cast<Etat *>(p);
   random_revice rd;
   mt19937 prng { rd() };
   uniform_int_distribution<> de{ 0, etat.indice };
   bool poursuivre = true;
   do {
      // notez le recours localisé à l'Autoverrou : nous ne voulons pas
      // tenir le mutex plus longtemps que nécessaire pour éviter de
      // bloquer inutilement les autres fils d'exécution
      int lancer = de(prng);
      {
         Autoverrou av {chic_mutex};
         etat.donnee = lancer; // code risqué
      }
      if (lancer == 0) {
         cout << "Thread #" << etat.indice << "; arret" << endl;
         poursuivre = false;
      }
      Sleep(1);
   } while(poursuivre);
   return {};
}

On remarquera un allègement léger du code de chic_thread(), mais surtout un accroissement de la résilience du programme. Cette version perd toutefois un peu de souplesse en attendant systématiquement l'obtention du mutex (la précédente se suspendait si l'obtention était un échec), mais il existe une solution simple si vous en avez besoin.

Enfin : implémentation conforme aux usages de C++ 11

Allons-y maintenant avec une version plus proche des usages de C++ 11. Notez que malgré l'illustration qui suit, puisque ce standard offre un support plein et entier des mutex portables standards, vous devriez utiliser ces derniers si votre compilateur est à jour (plutôt qu'une version maison comme celle proposée ici).

Mutex.h
#ifndef IMUTEX_H
#define IMUTEX_H
#include <cassert>
#include <memory>
class Mutex { // implicitement incopiable, car l'attribut p est incopiable
   class Impl;
   std::unique_ptr<Impl> p;
public:
   Mutex();
   Mutex(Mutex &&) noexcept;
   Mutex& operator=(Mutex &&) noexcept;
   ~Mutex();
   bool obtenir() const noexcept;
   bool obtenir(int ms) const noexcept;
   void relacher() const noexcept;
};
// ... code d'Autoverrou, omis par souci d'économie...
#endif
Mutex.cpp
#include "Mutex.h"
#include "Incopiable.h"
#include <windows.h> // par exemple
class Mutex::Impl : Incopiable {
   HANDLE m;
public:
   Impl() : m{ CreateMutex(0, FALSE, 0) } {
   }
   ~Impl() {
      CloseHandle(m);
   }
   bool obtenir() const noexcept {
      return obtenir(INFINITE);
   }
   bool obtenir(int ms) const noexcept {
      return WaitForSingleObject(m, ms) == WAIT_OBJECT_0;
   }
   void relacher() const noexcept {
      ReleaseMutex(m);
   }
};
Mutex::Mutex() : p{ make_unique<Impl>() } {
}
Mutex::Mutex(Mutex &&autre) = default;
Mutex& Mutex::operator=(Mutex &&autre) = default;
Mutex::~Mutex() = default;

Remarquez la simplicité de cette implémentation. En effet :

Valid XHTML 1.0 Transitional

CSS Valide !