Aller au contenu

Ce TP accompagne le chapitre Concurrence et parallélisme.

Remarque

On compile les programmes de ce TP avec l'option -pthread :

gcc -Wall -Wextra -g -pthread -o prog prog.c

Si vous avez un problème à l'édition de liens, il faut rajouter -lpthread. L'option -fsanitize=thread (à la place de -fsanitize=address) détecte à l'exécution les accès concurrents non protégés à une même variable (data races).

Attention, sous macOS, les sémaphores anonymes de semaphore.h ne sont pas implémentés : sem_init renvoie -1 et les programmes de la seconde partie ne fonctionnent pas. Il faut alors travailler sous Linux (machine virtuelle, serveur de l'établissement...).

Type universel de pointeurs

En C, le type void * joue un rôle particulier :

  • toute valeur de type T *a peut être convertie en un void * ;
  • toute valeur de type void *v peut être convertie en un T *.

Ces conversions sont implicites en C, il n'y a rien à écrire.

Ainsi, le code suivant est correct :

#include <stdio.h>

void incremente(void *a)
{
    int *t = a;
    *t = *t + 1;
}

int main(void)
{
    int n = 3;
    incremente(&n);
    printf("n = %d\n", n);
}

Ainsi, il est possible de faire des fonctions génériques en utilisant le type void * comme une sorte de 'a en OCaml. Mais contrairement à OCaml, aucune vérification n'est faite : c'est au programmeur de savoir vers quel type pointe réellement un void *.

Remarque

Comme le type int est, sur les architectures usuelles, plus petit que le type void * en termes de taille, il est possible de faire des conversions explicites pour ne pas avoir à définir une variable comme n ici.

Création de threads

Pour créer des threads, on utilise la bibliothèque pthread dont l'en-tête est pthread.h.

Pour créer un thread :

  • on définit une valeur pthread_t t qui va contenir les données du thread ;
  • on appelle pthread_create(&t, NULL, f, arg); où f est une fonction de prototype void *f(void *arg); et arg est une valeur de type void *.

Par exemple, on définit la fonction incremente qui va prendre un entier en argument par pointeur et l'incrémenter.

void *incremente(void *arg)
{
    int *n = arg;
    *n = *n + 1;

    return NULL; // pas de retour
}

On peut alors lancer cette fonction dans un thread :

pthread_t t;
int v = 3;
pthread_create(&t, NULL, incremente, &v);

Pour attendre qu'un thread t termine, on appelle pthread_join(t, NULL) :

pthread_join(t, NULL);
printf("%d\n", v); // affiche 4

Remarque

Il est également possible d'avoir une valeur de retour mais c'est hors programme.

Algorithme de Peterson

On considère le programme à deux threads suivant :

#include <pthread.h>
#include <stdio.h>

int compteur = 0;
int N = 100000;

void *incremente(void *arg)
{
    int *id = arg; // le numéro du thread, inutilisé pour l'instant
    (void)id;
    for(int i = 0; i < N; i++) {
        compteur = compteur + 1;
    }
    return NULL;
}

int main(void)
{
    pthread_t t0, t1;
    int i0 = 0;
    int i1 = 1;
    pthread_create(&t0, NULL, incremente, &i0);
    pthread_create(&t1, NULL, incremente, &i1);
    pthread_join(t0, NULL);
    pthread_join(t1, NULL);
    printf("compteur = %d (attendu %d)\n",
        compteur, 2 * N);
}

Question 1

Observez le comportement en changeant la valeur de N.

Corrigé

Pour de petites valeurs de N, on obtient souvent 2 * N : le premier thread a fini avant que le second ne démarre. Pour de grandes valeurs, le résultat est inférieur à 2 * N et varie d'une exécution à l'autre : des incrémentations sont perdues quand les deux threads lisent la même valeur de compteur avant d'écrire (voir Exemple de section critique).

Avec l'option -O2, on peut même obtenir exactement N : le compilateur remplace la boucle par une unique instruction compteur = compteur + N, et les deux threads lisent 0 en même temps. En C, ces accès concurrents non protégés sont des data races, dont le comportement n'est pas défini, et -fsanitize=thread les signale.

On va essayer de réaliser des fonctions de prototypes :

void verrouille(int id);
void deverrouille(int id);

où id est le numéro du thread, 0 ou 1 ici, et telles qu'entre les appels à verrouille(id); et deverrouille(id); on puisse placer une section critique et garantir :

  • l'exclusion mutuelle : un seul thread peut exécuter la section critique à un moment donné ;
  • l'absence d'interblocage : si un thread effectue un appel à verrouille, alors au moins un thread, peut-être un autre, va rentrer en section critique ensuite ;
  • l'absence de famine : si un thread effectue un appel à verrouille, il ne va pas attendre indéfiniment.

Remarque

L'absence de famine implique l'absence d'interblocage (voir Propriétés d'un verrou).

On va alors, dans l'exemple précédent, remplacer incremente par :

void *incremente(void *arg)
{
    int *id = arg; // le numéro du thread
    for(int i = 0; i < N; i++) {
        verrouille(*id);
        compteur = compteur + 1;
        deverrouille(*id);
    }
    return NULL;
}

Premier essai

On propose le pseudo-code suivant :

flag[0] = false
flag[1] = false

verrouille(i):
    flag[i] = true
    tant que flag[j] == true où j est le numéro de l'autre thread
        attendre

deverrouille(i):
    flag[i] = false

Question 2

Implémentez et testez cet algorithme.

Que se passe-t-il ? Est-ce prévisible ?

Corrigé 
bool flags[2] = { false, false };

void verrouille(int id)
{
    flags[id] = true;
    while(flags[1-id]) { // attente active
    }
}

void deverrouille(int id)
{
    flags[id] = false;
}

Le programme se bloque : les deux threads finissent par lever leur drapeau en même temps, puis chacun attend indéfiniment que l'autre baisse le sien. C'est un interblocage, prévisible : voir Première idée.

Question 3

Montrez que ce verrou satisfait l'exclusion mutuelle.

Corrigé

Voir la preuve du cours dans Première idée : si les deux threads étaient en section critique, chacun aurait écrit son drapeau avant que l'autre n'écrive le sien, ce qui est impossible.

Second essai

On propose maintenant le pseudo-code suivant :

victime = 0

verrouille(i):
    victime = i
    tant que victime == i
        attendre

deverrouille(i):
    ne fait rien

Question 4

Implémentez et testez cet algorithme.

Que se passe-t-il ? Est-ce prévisible ?

Corrigé 
int victime = 0;

void verrouille(int id)
{
    victime = id;
    while(victime == id) { // attente active
    }
}

void deverrouille(int id)
{
}

Un thread n'est débloqué que lorsque l'autre verrouille à son tour. Dès qu'un thread a fini ses N incrémentations, l'autre reste bloqué indéfiniment : le programme ne termine pas. C'est prévisible, voir Deuxième idée.

Question 5

Montrez que ce verrou satisfait l'exclusion mutuelle.

Corrigé

Voir la preuve dans Deuxième idée : le dernier thread à avoir écrit dans victime ne peut pas en être sorti.

Algorithme de Peterson

On va combiner les deux mauvaises idées précédentes pour en déduire une bonne idée.

On va utiliser flags et victime, mais cette fois, on attend tant que flags[j] == true et victime == i.

Question 6

Implémentez et testez cet algorithme.

Que se passe-t-il ? Est-ce prévisible ?

Corrigé 
bool flags[2] = { false, false };
int victime = 0;

void verrouille(int id)
{
    int autre = 1 - id;
    flags[id] = true;
    victime = id;
    while(flags[autre] && victime == id) { // attente active
    }
}

void deverrouille(int id)
{
    flags[id] = false;
}

Le programme termine, mais le compteur peut être légèrement inférieur à 2 * N (par exemple 199999), alors que l'algorithme est prouvé correct : les hypothèses du modèle mémoire ne sont pas satisfaites par les processeurs modernes, qui peuvent réordonner l'écriture de flags[id] et la lecture de flags[autre]. Avec -O2, le programme peut même se bloquer : le compilateur, qui ne sait pas que d'autres threads modifient ces variables, lit flags[autre] et victime une seule fois avant la boucle.

Remarque

Il est possible que ce verrou marche presque en pratique ; la raison est liée aux optimisations de la gestion de la mémoire, où les écritures ne sont pas réalisées au bon moment.

Sur un processeur x86, on peut rajouter l'instruction :

asm volatile ("mfence" ::: "memory");

avant le while pour forcer l'écriture avant la lecture. Cette instruction n'existe pas sur d'autres processeurs (ARM, comme les Mac récents) ; on peut utiliser à la place la fonction __sync_synchronize(); de gcc et clang, qui place la barrière mémoire adaptée au processeur.

Normalement, ça suffit ici à réparer le verrou sans optimisations (-O0). Pour un programme correct au sens de la norme, il faut déclarer les variables partagées comme atomiques avec stdatomic.h (atomic_bool flags[2]; et atomic_int victime;) : leurs accès sont alors réalisés dans l'ordre du programme et ne sont plus optimisés. C'est hors programme.

Question 7

Prouvez que ce verrou satisfait toutes les propriétés voulues.

Corrigé

Voir les preuves de l'exclusion mutuelle et de l'absence de famine dans Algorithme de Peterson ; l'absence d'interblocage découle de l'absence de famine.

Mutex

Il y a dans pthread un mécanisme de verrou fiable : les mutex.

On déclare un mutex m avec une variable globale :

pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;

On peut également l'initialiser avec

pthread_mutex_init(&m, NULL);

On peut alors verrouiller et déverrouiller m avec :

pthread_mutex_lock(&m); // verrouille
pthread_mutex_unlock(&m); // déverrouille

Même si ici cela ne va rien faire, c'est une bonne pratique d'appeler

pthread_mutex_destroy(&m);

quand on a fini d'utiliser le mutex.

Question 8

Reprenez le code d'incrémentation avec un mutex.

Corrigé 
pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;

void *incremente(void *arg)
{
    (void)arg;
    for(int i = 0; i < N; i++) {
        pthread_mutex_lock(&m);
        compteur = compteur + 1;
        pthread_mutex_unlock(&m);
    }
    return NULL;
}

On obtient toujours 2 * N, y compris avec -O2, et -fsanitize=thread ne signale plus rien. On appelle pthread_mutex_destroy(&m); à la fin de main, après les join.

Incrémentation atomique

Une autre manière de réparer ce programme est d'utiliser l'instruction atomique non portable suivante :

__sync_add_and_fetch(&compteur, 1);

Ici, on va réaliser l'incrémentation d'un seul coup sans avoir de possibilité d'accès concurrent (voir Instructions atomiques).

Question 9

Vérifiez que cela fonctionne.

Corrigé

Il suffit de remplacer, dans la boucle de incremente, l'instruction compteur = compteur + 1; (et les appels de verrouillage) par __sync_add_and_fetch(&compteur, 1);. On obtient bien 2 * N. C'est l'instruction fetch and add du processeur ; elle est plus rapide qu'un mutex mais ne protège qu'une seule opération sur une seule variable.

Sémaphores

pthread définit également des sémaphores, dans l'en-tête semaphore.h. Ce sont des compteurs de ressources qui permettent :

  • d'être initialisés avec un nombre de ressources ;
  • d'être acquis en attendant que le compteur soit \(> 0\) et en le décrémentant ;
  • d'être rendus en incrémentant le compteur.

Ce mécanisme permet de transmettre des signaux, puisque un thread peut attendre un sémaphore qui est nul et qui va passer à 1 quand un autre thread veut lui envoyer un signal.

Avec pthread, on utilise :

sem_t s; // définit un sémaphore

sem_init(&s, 0, v); // l'initialise avec la valeur
                    // initiale v

sem_wait(&s); // attend pour décrémenter (acquisition)

sem_post(&s); // incrémente (rendu)

sem_destroy(&s); // quand on a fini de l'utiliser

Rendez-vous

Question 10

Réalisez deux threads effectuant chacun l'affichage AB de deux caractères, mais en faisant en sorte que la trace soit toujours AABB.

Indication

On pourra utiliser deux sémaphores initialement à 0, intitulés, c'est un indice, t_est_la[0] et t_est_la[1].

Corrigé

Chaque thread signale qu'il est arrivé, puis attend l'autre (voir aussi Rendez-vous à deux threads).

#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>

sem_t t_est_la[2];

void *f(void *arg)
{
    int *id = arg;
    putchar('A');

    sem_post(&t_est_la[*id]);     // je signale que je suis là
    sem_wait(&t_est_la[1 - *id]); // j'attends l'autre

    putchar('B');
    return NULL;
}

int main(void)
{
    pthread_t t0, t1;
    int i0 = 0;
    int i1 = 1;
    sem_init(&t_est_la[0], 0, 0);
    sem_init(&t_est_la[1], 0, 0);
    pthread_create(&t0, NULL, f, &i0);
    pthread_create(&t1, NULL, f, &i1);
    pthread_join(t0, NULL);
    pthread_join(t1, NULL);
    putchar('\n');
    sem_destroy(&t_est_la[0]);
    sem_destroy(&t_est_la[1]);
    return 0;
}

Il faut faire le sem_post avant le sem_wait : dans l'autre ordre, les deux threads s'attendraient mutuellement (interblocage).

Rendez-vous généralisé : la barrière

Question 11

Même question que la précédente mais pour \(10\) threads.

Il ne faut pas que le nombre de primitives de synchronisation dépende du nombre de threads.

Indication

Une variable compte, un mutex pour la gérer et un sémaphore nul pour signaler qu'on peut passer à la suite.

Corrigé

Le dernier thread arrivé ouvre la barrière, puis chaque thread qui passe laisse passer le suivant (voir Rendez-vous à n threads : barrière).

#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>
#include <stdbool.h>

#define NB_THREADS 10

int compte = 0;
pthread_mutex_t verrou_compte = PTHREAD_MUTEX_INITIALIZER;
sem_t barriere; // initialement à 0

void *f(void *arg)
{
    (void)arg;
    putchar('A');

    pthread_mutex_lock(&verrou_compte);
    compte = compte + 1;
    bool dernier = (compte == NB_THREADS); // lu dans la section critique
    pthread_mutex_unlock(&verrou_compte);

    if (dernier)
        sem_post(&barriere);

    // tourniquet : chaque thread qui passe laisse passer le suivant
    sem_wait(&barriere);
    sem_post(&barriere);

    putchar('B');
    return NULL;
}

int main(void)
{
    pthread_t t[NB_THREADS];
    sem_init(&barriere, 0, 0);
    for(int i = 0; i < NB_THREADS; i++)
        pthread_create(&t[i], NULL, f, NULL);
    for(int i = 0; i < NB_THREADS; i++)
        pthread_join(t[i], NULL);
    putchar('\n');
    sem_destroy(&barriere);
    return 0;
}

On teste compte == NB_THREADS dans la section critique : lire compte après l'avoir déverrouillé serait une data race.

Barrière réutilisable

Question 12

On souhaite créer un type barriere et des fonctions :

// initialise la barrière pour synchroniser n threads
void init_barriere(barriere *b, int n);

// attend que les n threads aient atteint la barrière
void attends_barriere(barriere *b);

Contrairement au cas précédent, on souhaite que la barrière soit réutilisable.

Il ne faut pas que le nombre de primitives de synchronisation dépende du nombre de threads.

Corrigé

Avec un seul tourniquet, un thread rapide pourrait repasser la barrière avant que les autres ne l'aient franchie. On utilise donc deux tourniquets, l'un fermé quand l'autre est ouvert, comme dans Barrière réutilisable.

struct barriere {
    int n;                 // nombre de threads à synchroniser
    int compte;            // nombre de threads arrivés
    pthread_mutex_t verrou;
    sem_t entree;          // tourniquet d'entrée, initialement fermé (0)
    sem_t sortie;          // tourniquet de sortie, initialement ouvert (1)
};
typedef struct barriere barriere;

void init_barriere(barriere *b, int n)
{
    b->n = n;
    b->compte = 0;
    pthread_mutex_init(&b->verrou, NULL);
    sem_init(&b->entree, 0, 0);
    sem_init(&b->sortie, 0, 1);
}

void detruit_barriere(barriere *b)
{
    pthread_mutex_destroy(&b->verrou);
    sem_destroy(&b->entree);
    sem_destroy(&b->sortie);
}

void attends_barriere(barriere *b)
{
    // phase 1 : on attend que les n threads soient arrivés
    pthread_mutex_lock(&b->verrou);
    b->compte = b->compte + 1;
    if (b->compte == b->n) {
        sem_wait(&b->sortie);  // on ferme la sortie
        sem_post(&b->entree);  // on ouvre l'entrée
    }
    pthread_mutex_unlock(&b->verrou);
    sem_wait(&b->entree);
    sem_post(&b->entree);

    // phase 2 : on attend que les n threads soient passés
    pthread_mutex_lock(&b->verrou);
    b->compte = b->compte - 1;
    if (b->compte == 0) {
        sem_wait(&b->entree);  // on referme l'entrée
        sem_post(&b->sortie);  // on ouvre la sortie
    }
    pthread_mutex_unlock(&b->verrou);
    sem_wait(&b->sortie);
    sem_post(&b->sortie);
}

Avec 10 threads exécutant 5 fois putchar('A'); attends_barriere(&b); putchar('B'); attends_barriere(&b);, la trace est toujours formée de 10 A, 10 B, 10 A, etc.

Problèmes

Dans tous les problèmes suivants, on demande d'implémenter une solution qui se comporte bien.

Producteurs-consommateurs

On considère des tâches qui sont gérées dans un tampon de \(n\) cases.

On a :

  • \(p\) threads producteurs qui veulent écrire une nouvelle tâche dans le tampon, quitte à attendre s'il est plein ;
  • \(c\) threads consommateurs qui veulent récupérer une tâche et libérer la place dans le tampon, quitte à attendre s'il est vide.
Corrigé

On gère le tampon comme une file circulaire, avec un sémaphore comptant les places libres, un autre comptant les tâches présentes et un mutex protégeant les indices (voir Producteurs-consommateurs).

#define N 4            // taille du tampon

int tampon[N];
int debut = 0;         // indice de la prochaine tâche à retirer
int fin = 0;           // indice de la prochaine case libre

sem_t places;          // places libres, initialisé à N
sem_t taches;          // tâches présentes, initialisé à 0
pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;

void depose(int tache) // appelé par les producteurs
{
    sem_wait(&places);
    pthread_mutex_lock(&m);
    tampon[fin] = tache;
    fin = (fin + 1) % N;
    pthread_mutex_unlock(&m);
    sem_post(&taches);
}

int retire(void)       // appelé par les consommateurs
{
    sem_wait(&taches);
    pthread_mutex_lock(&m);
    int tache = tampon[debut];
    debut = (debut + 1) % N;
    pthread_mutex_unlock(&m);
    sem_post(&places);
    return tache;
}

L'ordre des opérations compte : si un producteur prenait le mutex avant d'attendre une place libre, il bloquerait les consommateurs qui pourraient la libérer.

Lecteurs-écrivains

On considère maintenant une ressource comme une base de données dont l'accès est une section critique, et deux types de threads :

  • autant de lecteurs qu'on le souhaite peuvent être dans la section critique ;
  • les écrivains ont un accès exclusif à la section critique (ce qui bloque également les lecteurs).
Corrigé

Un sémaphore salle_vide, initialisé à 1, est pris par chaque écrivain et par le premier lecteur qui entre ; il est rendu par chaque écrivain et par le dernier lecteur qui sort. Le nombre de lecteurs présents est protégé par un mutex.

int nb_lecteurs = 0;   // lecteurs en section critique
pthread_mutex_t verrou_lecteurs = PTHREAD_MUTEX_INITIALIZER;
sem_t salle_vide;      // initialisé à 1

void debut_lecture(void)
{
    pthread_mutex_lock(&verrou_lecteurs);
    nb_lecteurs = nb_lecteurs + 1;
    if (nb_lecteurs == 1)  // le premier lecteur ferme la salle aux écrivains
        sem_wait(&salle_vide);
    pthread_mutex_unlock(&verrou_lecteurs);
}

void fin_lecture(void)
{
    pthread_mutex_lock(&verrou_lecteurs);
    nb_lecteurs = nb_lecteurs - 1;
    if (nb_lecteurs == 0)  // le dernier lecteur la rouvre
        sem_post(&salle_vide);
    pthread_mutex_unlock(&verrou_lecteurs);
}

void debut_ecriture(void)
{
    sem_wait(&salle_vide);
}

void fin_ecriture(void)
{
    sem_post(&salle_vide);
}

Cette solution présente un risque de famine pour les écrivains : si des lecteurs se relaient sans cesse, nb_lecteurs ne revient jamais à 0. On peut y remédier en ajoutant un sémaphore « tourniquet » que les écrivains ferment en arrivant et que chaque lecteur doit franchir avant debut_lecture.

Dîner des philosophes

\(n\) philosophes sont autour d'une table avec une baguette entre deux d'entre eux. Ils ont besoin de deux baguettes pour manger.

On cherche une solution garantissant les bonnes propriétés et maximisant le parallélisme, c'est-à-dire le nombre de philosophes mangeant en même temps.

Corrigé

Voir l'étude détaillée dans Le dîner des philosophes : la solution naïve (un sémaphore par baguette) peut provoquer un interblocage, et la solution par états évite l'interblocage tout en permettant à des philosophes non voisins de manger simultanément, mais présente un risque de famine.

Dîner convivial

\(n\) personnes sont autour d'une marmite pouvant servir \(m\) rations avant d'être vide. Quand elle est vide, elles vont réveiller le cuisinier pour qu'il remplisse la marmite et elles attendent que la marmite soit remplie pour se resservir.

Le cuisinier ne doit remplir la marmite que lorsqu'elle est vide.

Corrigé

Le nombre de rations est protégé par un mutex. Le convive qui trouve la marmite vide réveille le cuisinier avec un premier sémaphore, puis attend, en gardant le verrou, le signal du second sémaphore : les autres convives attendent ainsi sur le verrou et le cuisinier n'est réveillé qu'une fois par marmite vide.

#define M 5                 // rations par marmite

int rations = 0;            // la marmite est vide au départ
pthread_mutex_t verrou_marmite = PTHREAD_MUTEX_INITIALIZER;
sem_t marmite_vide;         // signal vers le cuisinier, initialisé à 0
sem_t marmite_pleine;       // signal vers le convive, initialisé à 0

void se_servir(void)        // appelé par les convives avant de manger
{
    pthread_mutex_lock(&verrou_marmite);
    if (rations == 0) {
        sem_post(&marmite_vide);   // on réveille le cuisinier
        sem_wait(&marmite_pleine); // et on attend qu'il ait rempli
    }
    rations = rations - 1;
    pthread_mutex_unlock(&verrou_marmite);
}

void *cuisinier(void *arg)
{
    (void)arg;
    while (true) {
        sem_wait(&marmite_vide);
        rations = M;               // la marmite est vide ici
        sem_post(&marmite_pleine);
    }
    return NULL;
}

Le cuisinier modifie rations sans prendre le verrou, mais sans concurrence : le seul convive qui pourrait y accéder attend marmite_pleine et les autres attendent le verrou qu'il détient. Les sémaphores assurent que l'écriture du cuisinier est visible du convive.