12 mar. 2013

Programmation concurrente : threads et variables atomiques

thread.jpg

Les deux articles précédents sur ce sujet n'ont fait qu'aborder des concepts qui, certes utiles pour la compréhension des articles à venir, n'ont pas encore permis d'approcher les nouveautés de C++11 dans le domaine de la programmation concurrente. Le but de cet article est donc de rentrer enfin dans le vif du sujet, en décrivant les interfaces proposés par la librairie et les mécanismes implémentés dans le langage.

Vu l'énormité des apports à ce niveau, ce billet va se contenter d'une première approche en décrivant principalement deux points importants : les threads et les mécanismes de synchronisation.

C++11 et les threads

La classe std::thread, introduite dans le standard C++11, permet de créer un processus léger - un fil d'exécution ayant la même configuration mémoire que le processus parent. L'interface de la classe std::thread est relativement simple.

namespace std {
	class thread
	{
	public:
		// types définis
		typedef implementation-defined native_handle_type;
		class id;
thread(thread&) = delete; thread(const thread&) = delete; thread& operator=(const thread&) = delete;
thread() noexcept = default;
thread(thread&& __t) noexcept;
// le constructeur important de la classe template<typename _Callable, typename... _Args> explicit thread(_Callable&& f, _Args&&... args);
~thread();
thread& operator=(thread&& __t) noexcept;
void swap(thread& __t) noexcept;
// gestion de l'état de l'objet bool joinable() const noexcept; void join(); void detach();
// accesseurs thread::id get_id() const noexcept; native_handle_type native_handle(); static unsigned int hardware_concurrency() noexcept; }; }

Le premier point intéressant est que la classe std::thread n'est pas, contrairement à beaucoup d'autres classes de la librairie C++ standard, une classe template. Par contre, elle possède un constructeur template dont nous allons parler plus loin.

Le second point intéressant est qu'un thread n'est pas copiable mais implémente la sémantique de mouvement (via son operator=() et son move constructor).

Enfin, un troisième point à noter est que cette interface est fortement liée à celle des thread POSIX[1]. A voir les noms utilisés, cette observation est évidente. Les utilisateurs de Windows, habitués aux CreateThread() et autres WaitForSingleObject() peuvent se sentir un peu surpris par ce choix et il n'est pas impossible qu'ils soient un peu perdus. Voici un petit résumé sur la correspondance entre std::thread, Windows et POSIX. Ce résumé n'a pas pour but de vous livrer les détails d'implémentation interne de la classe std::thread mais de vous fournir les bases pour la compréhension de l'interface et une direction pour son implémentation.

  • thread::native_handle_type peut être implémenté avec un HANDLE sous Windows, et un pthread_t dans l'univers POSIX.
  • thread::join() attend la fin d'un thread. C'est l'équivalent d'un WaitForSingleObject() sur le HANDLE du thread sous Windows, ou d'un appel à pthread_join() en POSIX.
  • thread::detach() détache le fil d'exécution de l'instance de std::thread. Une fois détaché, il n'est plus possible d'appeler thread::join() sur ce thread. Le détachement du thread est intéressant dans le cadre de traitements courts qu'on est sûr de terminer (par exemple, un traitement dans lequel il n'y a pas de boucle, ou dans lequel la boucle est finie). Avant d'être détachée, le thread doit être valide (c'est à dire que thread::joinable() doit retourner true). Dans le cas contraire, une exception std::system_error est levée.
  • thread::hardware_concurrency() donne des indices sur l'architecture matérielle - est-ce une architecture mono ou multi-coeur/CPU ? Le standard ne dit pas que la réponse renvoyée par cette fonction est obligatoirement valide - si le système ne peut pas déterminer cette réponse alors la librairie standard peut répondre 0 (section 30.3.1.6 de la norme C++11). Dans un environnement POSIX, on utilisera pthread_getconcurrency(). Sous Windows, il faudra récupérer les informations sur le processeur avec GetSystemInfo().
  • la classe thread::id représente un identifiant pour le thread. Cet identifiant peut être comparé, affiché... Un algorithme de hashage spécialisé lui est dédié. De fait, une instance de thread::id peut être utilisé dans un conteneur associatif tel que std::set<>, std::map<> ou std::unordered_map<>. Lorsqu'une instance the_thread de std::thread est attachée à un thread d'exécution, alors on a la relation the_thread.get_id() != thread::id(). Si l'instance n'est pas attachée à un thread d'exécution ou après un appel à thread::detach(), c'est la relation the_thread.get_id() == thread::id() qui est vraie.

Un thread peut être créé dans deux états. L'utilisation du constructeur par défaut crée une instance qui est invalide (détachée d'un thread d'exécution), tandis que l'utilisation du constructeur template crée une instance valide (attaché à un thread d'exécution) pendant le temps d'exécution du thread lui-même. Pour garantir la validité de l'instance, le standard prévoie que la complétion de l'invocation de ce constructeur soit synchronisé avec le début de l'exécution du thread. On assure ainsi que l'expression the_thread.get_id() != thread::id() à la sortie du constructeur, et tant que thread::join() ou thread::detach() n'ont pas été appelés et ce, de manière indépendante à ce qui se passe au niveau du thread d'exécution lui-même. Ainsi, le thread d'exécution peut être terminé, la relation the_thread.get_id() != thread::id() reste vraie tant que le programme appelant ne détache pas ou ne rejoint pas le thread d'exécution.

A coté de la classe std::thread a été ajouté un nom d'espace nommé std::this_thread, dont la définition est :

namespace std {
	namespace this_thread {
		thread::id get_id() noexcept;
void yield() noexcept;
template <class Clock, class Duration> void sleep_until(const chrono::time_point<Clock, Duration>& abs_time); template <class Rep, class Period> void sleep_for(const chrono::duration<Rep, Period>& rel_time);; } }

Les fonctions this_thread::sleep_until() et this_thread::sleep_for() se passent de commentaires : elle permettent d'endormir le thread en cours d'exécution jusqu'à une date donnée ou pour un temps donné. Plus intéressant, la fonction this_thread::get_id() renvoie l'identifiant du thread en cours d'exécution. Cet identifiant n'est pas nécessairement lié à celui d'une instance de std::thread - la seule condition est qu'il ne soit pas égal à thread::id().

Mais la fonction la plus intéressante est sans conteste la fonction this_thread::yield() : cette fonction permet d'avertir le système d'exploitation que le thread n'a plus de travail à faire dans l'immédiat. Le système d'exploitation est donc libre de passer le contrôle à un autre processus (léger ou non). Cette fonction provoque donc un changement dans l'état du scheduler de l'OS. Sous Windows, un appel à this_thread::yield() est équivalent à un appel à Sleep(0). Dans un environnement POSIX, on utilisera shed_yield() (dans <sched.h>) à cette fin.

Nous avons a peu près tout abordé, mais sans vraiment discuter de la création d'un thread. Comme vous le pressentez, on utilise le constructeur template :

namespace std {
	class thread
	{
	public:
		// ...
		template<typename _Callable, typename... _Args>
			explicit thread(_Callable&& f, _Args&&... args);
// ... }; }

Il y a deux choses à voir dans ce constructeurs. La première chose est que tous les paramètres sont passés par rrvalue-reference[2]. Je vous fait grâce des détails d'implémentation, mais sachez tout de même que les arguments sont transférés au thread en cours de création - ils ne sont pas simplement copiés. Cette notion a son importance, parce que le thread devient le propriétaire des données qu'on lui passe en paramètre.

Le second point important est que le paramètre principal - le seul qui doit être présent - est un appelable[3] (callable dans le texte de la norme C++11). Un appelable est une entité du langage qui supporte d'être utilisé en conjonction avec operator(). Voici quelques exemple d'appelables.

  • une fonction libre ou une méthode statique dans une classe ;
  • un foncteur ou un autre type d'objet fonction
  • une expression lambda

C'est là un des points forts de la norme C++11 : là ou la plupart des API de gestion de thread se limitent à ne pouvoir utiliser que des fonctions libres ne prenant qu'un seul paramètre, la librairie standard C++11 réussit le tour de force de proposer une interface (et des implémentations dédiées) permettant l'utilisation d'un appelable quelconque avec autant de paramètres que souhaités.

Concrètement, voici quelques cas d'utilisation de la classe std::thread[4].

	#include <iostream>
	#include <thread>
	#include <chrono>
	#include <string>
void free_function() { auto start = std::chrono::high_resolution_clock::now(); for (unsigned int i=0; i<10000; ++i) { // schedule 10.000 times std::this_thread::yield(); } auto stop = std::chrono::high_resolution_clock::now(); auto duration = stop - start; std::cout << "free_function: " << duration.count() << " µs" << std::endl; }
struct functor { void operator()(unsigned int n, const std::string& s) { auto start = std::chrono::high_resolution_clock::now(); for (unsigned int i=0; i<n; ++i) { // schedule n times std::this_thread::yield(); } auto stop = std::chrono::high_resolution_clock::now(); auto duration = stop - start; std::cout << s << duration.count() << " µs" << std::endl; } };
int main() { std::thread t1(free_function); t1.join();
std::thread t2([](unsigned int n) { auto start = std::chrono::high_resolution_clock::now(); for (unsigned int i=0; i<n; ++i) { // schedule n times std::this_thread::yield(); } auto stop = std::chrono::high_resolution_clock::now(); auto duration = stop - start; std::cout << " lambda: " << duration.count() << " µs" << std::endl; }, 10000); t2.join();
std::string sparam(" callable: "); std::thread t3(functor(), 10000, std::cref(sparam)); t3.join(); }

Le premier thread est instancié avec une fonction libre. Le second est instancié avec une expression lambda prenant un argument (passé par valeur). Le troisième et dernier thread est instancie avec un foncteur prenant deux paramètres - le premier est passé par valeur, le second est passé par référence - la fonction std::cref() permet de créer une instance de std::reference_wrapper<>[5]. L'utilisation de std::cref() ou std::ref() pour passer un paramètre par référence (const ou non, selon la fonction utilisée) est nécessaire

Dans ce code d'exemple, chacun des threads exécute à peu de chose près le même code (en se basant ou non sur les paramètres reçus) : le thread schedule un certain nombre de fois et affiche ensuite son temps d'exécution avant de se terminer.

Points de synchronisation : accès atomiques

Bien évidemment, l'exemple présenté ci-dessus n'offre que peu d'intérêts. Le programme principal crée les thread et attend leurs terminaisons de manière séquentielle - ce qui n'aurait guère de sens dans un programme réel. Dans la grande majorité des cas, le programme principal crée un ou plusieurs threads et effectue d'autres traitements pendant que le ou les threads s'exécutent. Dans certains cas, les threads sont complètement indépendants les uns des autres - on parle de parallélisme embarrassant (car trivial). Dans ces cas, on peut se permettre de synchroniser la fin du programme sur la fin de tous les threads - ce n'est pas nécessairement un problème trivial avec l'interface proposée par la librairie standard, ainsi qu'on le verra par la suite.

Cependant, les problèmes trivialement parallèles ne sont hélas pas si courant. Dans une grande majorité des cas, les différents threads doivent communiquer ou utiliser une ressource partagée. On a vu dans les articles précédents que pour ce faire, une seule solution était envisageable : il faut utiliser des primitives de synchronisation.

Le standard C++11 offre plusieurs de ces primitives. La première (et la plus simple) est le type scalaire atomique. Les opérations effectués sur une donnée de ce type sont obligatoirement atomiques. Il existe deux types de données pour stocker des scalaires atomiques :

  • std::atomic<T> : T est un type intégral (bool,  char, int, long, long long - en version signée ou non) ou un type trivialement copiable (et, de préférence, assignable de manière statique) ou un type pointeur. Lorsque T est un type intégral, le type std::atomic<T> se résout à un type spécialisé offrant les meilleurs performances possibles. Dans le cas contraire, on utilise une implémentation spécialisée correspondant au type sous-jacent (par exemple, la spécialisation partielle std::atomic<T*> propose des méthodes qui n'ont de sens que pour un pointeur).
  • std::atomic_flag<> permet d'utiliser des primitives de type test and set (TAS, une forme améliorée de CAS).

std::atomic<> offre au moins les méthodes suivantes :

	namespace std {
		template <class T> struct atomic {
			atomic() noexcept = default;
atomic(const atomic&) = delete; atomic& operator=(const atomic&) = delete;
constexpr atomic(T); T operator=(T);
bool is_lock_free(); void store(T, memory_order); T load(memory_order); operator T() const; T exchange(T, memory_order);
bool compare_exchange_weak(T&, T, memory_order); bool compare_exchange_weak(T&, T, memory_order, memory_order); bool compare_exchange_strong(T&, T, memory_order); bool compare_exchange_strong(T&, T, memory_order, memory_order); }; }

Je me suis permis, dans ce résumé, de supprimer les qualifiant volatile et noexcept, ainsi que les valeurs par défaut de certains paramètres - si vous voulez les retrouver, je vous engage à consulter la norme C++11, section 29.5. Rapidement : on ne peut pas copier une valeur atomique, ni affecter une valeur atomique à une autre valeur atomique (pour la simple et bonne raison que deux opérations atomiques distinctes ne peuvent être réunies en une seule). Les opérations d'initialisation (utilisation du constructeur par valeur et de l'opérateur d'assignation) ne sont pas des opérations atomiques - il convient donc de les éviter si possible. De la même manière, l'accès à la valeur stockée via l'opérateur de transtypage n'est pas atomique.

Restent les opérations store(), load(), exchange() et compare_exchange_XXX(). Bien évidemment (il en faut un peu), ces opérations sont atomiques. Elles ont ceci de particulier qu'on peut avec des fonctions influer sur l'ordre des opérations mémoire effectuées. L'énumération memory_order (section 29.3 de la norme C++11) défini un certain nombre de constantes :

	namespace std {
		typedef enum memory_order {
			memory_order_relaxed, 
			memory_order_consume, 
			memory_order_acquire,
			memory_order_release, 
			memory_order_acq_rel, 
			memory_order_seq_cst
		} memory_order;
	}
  • memory_order_relaxed signifie que les opérations mémoire ne sont pas ordonnées - si vous vous rappelez l'article précédent de la série, cela veut simplement dire qu'aucune barrière mémoire n'est utilisée pour implémenter l'accès à la variable atomique. L'accès à la variable est atomique, le cache et la mémoire peuvent ne pas être mis à jour. A moins de savoir très exactement ce que vous faîtes, je vous conseille de ne jamais utiliser cette constante dans votre code. J'espère avoir été clair.
  • memory_order_seq_cst est la valeur par défaut utilisée pour les opérations store(), load(), exchange() est les versions à 3 paramètres de compare_exchange_XXX(). La constante est utilisée pour implémenter des barrière mémoires complètes au moment de l'acquisition et au moment de la libération de la donnée.
  • memory_order_acq_rel ajoute une barrière mémoire en aquisition avant l'accès à la donnée, et une barrière mémoire en libération après l'accès à la donnée.
  • memory_order_acquire et memory_order_release sont utilisées respectivement pour implémenter des barrières mémoire en aquisition et en libération.
  • memory_order_consume est utilisée pour implémenter un type plus rare de barrière mémoire en acquisition - la donnée est consommée par le code situé entre les deux barrières mémoires. Cette valeur n'est valable que pour l'opération de chargement de la donnée.

Les fonctions compare_exchange_XXX() sont un peu différentes. Notons d'abord que la version à 3 paramètres est équivalent à la version à 4 paramètres - en doublant le paramètre d'ordre mémoire[6] - et arrêtons nous sur la version à 4 paramètres.

	namespace std {
		template <class T> struct atomic {
			// ...
			bool compare_exchange_weak(
				T& expected, 
				T desired, 
				memory_order success, 
				memory_order failure);
			bool compare_exchange_strong(
				T& expected, 
				T desired, 
				memory_order success, 
				memory_order failure);
			// ...
		};
	}

L'algorithme utilisé par ces fonctions est le suivant :

	// utilisation de l'ordre mémoire success
	if (this->value == expected) {   // Compare-and-...
		this->value = desired;      // ..-Set
		return true;
	} else {
		// utilisation de l'ordre mémoire failure
		expected = this->value;
		return false;
	}

Bien évidemment, la comparaison suivie d'une mise à jour est implémentée sous la forme d'un CAS - c'est une opération atomique

Le fait que l'ordre mémoire failure soit utilisé en cas d'échec pour lire la valeur uniquement impose des limites sur les valeurs de failure. En effet, il ne sert à rien que cet ordre mémoire soit un ordre de libération. Du coup, les constantes memory_order_release et memory_order_acq_rel sont interdites par le standard C++11.

Le suffixe _weak ou _strong est utilisé pour marquer une différence de fonctionnement : en gros, la version _weak peut échouer même si les conditions de réussite sont respectées (c'est à dire que la valeur considérée est égale à expected). Une implémentation doit faire en sorte que ces erreurs soient le moins courantes possibles. La conséquence est que les fonctions _weak sont normallement utilisées dans des boucles, afin de compenser cette limitation.

Le code suivant présente une implémentation possible d'un spin lock minimal utilisant std::atomic<>.

	#ifndef spin_lock_h_included
	#define spin_lock_h_included
#include <atomic> #include <thread>
namespace example {
class spin_lock { private: std::atomic<unsigned int> m_spin;
bool try_acquire() { unsigned int expected = 0; return m_spin.compare_exchange_weak( expected, 1, std::memory_order_acq_rel, std::memory_order_acquire); }
public: spin_lock() : m_spin(0) // not atomic, but do we care at this point? { } ~spin_lock() = default; spin_lock(const spin_lock&) = delete; spin_lock& operator=(const spin_lock&) = delete;
void acquire() { while (!try_acquire()) std::this_thread::yield(); }
// to release the lock, we need to update its internal // atomic value and to put a release barrier (so that // any subsequent acquire will get the real value from // memory). void release() { m_spin.store(0, std::memory_order_release); } };
}
#endif // spin_lock_h_included

Conclusion

Cet article complète les deux articles précédents de la série en privilégiant cette fois une approche pragmatique. On voit que les concepts abordés alors (processus légers, CAS, barrières mémoire...) trouvent leur écho dans les interfaces proposées par la librairie standard du C++11 - plus exactement via les classes std::thread et std::atomic<T> qui ont été abordées ici. J'espère que cette introduction vous permettra de commencer à utiliser ces outils puissants dans votre code.

Il est entendu que les variables atomiques sont des outils puissants. Ils sont toutefois complexes à mettre en oeuvre, et ne sont pas nécessairement adaptés aux scénarios classiques d'utilisation. Il existe d'autres primitives de synchronisations plus rependus et plus puissants. Ces autres mécanismes - mutex et variables de condition - seront abordés dans le prochain article de cette série.

Notes

[1] pour information, le standard POSIX est visible gratuitement en ligne sur le site de l'Open Group. Il faut cependant s'enregistrer.

[2] un article futur s'attardera sur ce sujet relativement complexe, et sur les possibilités offertes par cette nouveauté du standard C++11.

[3] faute de meilleur terme ; j'aurais bien utilisé invocable, mais le mot n'existe pas en français...

[4] le code est compilé avec g++ 4.7.2. Une version antérieure (mais pas trop) de ce compilateur peut bien sûr être utilisée. Attention toutefois : il est nécessaire d'utiliser le flag de compilation -pthread pour compiler cet exemple (sans cela, vous aller recevoir des exceptions std::system_error dès que vous allez vouloir créer un thread).

[5] pour plus d'information, vous pouvez vous référer à la norme C++11 section 20.8.3.5, ou à la page correspondante du site cppreference.com.

[6] c'est un peu plus compliqué que ça : voir la section 29.6.5, §20 et §21 de la norme C++11 pour plus d'informations sur le sujet.

Commentaires

1. Le jeudi, mars 14 2013, 17:56 par Emmanuel Deloget

gdbivers m'a remonté une erreur : la capture dans l'expression lambda a disparu du billet. Cette erreur est maintenant corrigée. Merci encore à lui !

Ajouter un commentaire

Les commentaires peuvent être formatés en utilisant une syntaxe wiki simplifiée.

Fil des commentaires de ce billet