Séance 3 — Programmation parallèle avancée (synchronisation)
bleutitreSéance 3 : Programmation parallèle avancée (synchronisation)
Les défis de la programmation parallèle
Dans un programme parallèle, plusieurs fils s’exécutent en même temps et se partagent des ressources : variables, structures de données, fichiers, connexions. Que se passe-t-il si deux fils veulent utiliser la même ressource en même temps ? Ou si l’un a besoin d’une donnée que l’autre est en train de modifier ? Comment s’assurer qu’un résultat ne sort pas avant que toutes ses parties soient prêtes ?
C’est la question centrale de la programmation parallèle : comment gérer l’accès concurrent aux ressources partagées et synchroniser les fils pour qu’ils collaborent efficacement ?
Les problèmes fondamentaux
Lorsque plusieurs fils s’exécutent en parallèle et partagent des ressources, trois problèmes majeurs émergent :
- Courses critiques (race conditions) : Quand deux fils accèdent à la même donnée simultanément, l’un au moins en écriture
- Interblocage : Quand des fils s’attendent mutuellement indéfiniment
- Livelock : Quand des fils restent actifs mais ne progressent pas
Courses critiques et sections critiques
Le problème des courses critiques
Une course critique (race condition) se produit quand le résultat d’un programme dépend de l’ordre d’exécution non déterministe de plusieurs fils accédant à des ressources partagées.
from threading import Thread
# Variable globale partagée
compteur = 0
def incrementer():
global compteur
for i in range(1000000):
# Cette ligne n'est PAS atomique !
compteur = compteur + 1
# Elle se décompose en 3 opérations :
# 1. Lire compteur
# 2. Ajouter 1
# 3. Écrire le résultat
# Lancement de 2 threads
thread1 = Thread(target=incrementer)
thread2 = Thread(target=incrementer)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print(f"Valeur finale : {compteur}")
# Attendu : 2000000
# Possible : toute valeur de 2 à 2000000, selon l'ordre des filsExécuté sous CPython 3.13, ce programme affiche Valeur finale : 2000000, et le même résultat revient à chaque essai. Ce résultat juste ne prouve rien : depuis la version 3.10, l’interpréteur ne passe d’un fil à l’autre qu’à certains points de la boucle (les sauts en arrière, en fin de tour de boucle, et les appels de fonction), jamais entre la lecture et l’écriture de compteur = compteur + 1. Rien ne garantit ce comportement, qui change avec la version, avec un interpréteur sans verrou global, ou dès que la boucle fait autre chose : trois exécutions sous CPython 3.9 donnent 1 415 771, 1 481 718, 1 549 458, et trois sous CPython 3.13t, la version sans verrou global, 1 358 692, 1 027 324, 1 231 345. Une attente entre la lecture et l’écriture fait apparaître la course :
def incrementer(n):
global compteur
for i in range(n):
v = compteur # 1. lire
time.sleep(0) # le fil cède la main
compteur = v + 1 # 2. ajouter 1, 3. écrireAvec deux fils de \(100\,000\) incréments chacun, ce programme affiche attendu : 200000 obtenu : 100000 : la moitié des incréments est perdue. La Figure 1 montre comment un incrément se perd : le second fil lit compteur avant que le premier ait écrit, puis écrase son écriture.
Après l’étape 2 : le fil 2 lit avant que le fil 1 ait écrit
Après l’étape 6 : le fil 2 écrase l’écriture du fil 1
Combien de résultats différents ce programme peut-il donner ? Chaque fil exécute ses trois opérations dans l’ordre, mais les opérations des deux fils peuvent s’entrelacer de toutes les façons. Pour un incrément par fil, il y a \(\binom{6}{3}=20\) entrelacements des six opérations, et seuls 2 d’entre eux donnent la bonne valeur 2 : ceux où un fil a fini d’écrire avant que l’autre lise. La Figure 2 donne, pour 1, 2 et 3 incréments par fil, le nombre d’entrelacements qui mènent à chaque valeur finale. Dès deux incréments par fil, la plus petite valeur possible est 2, et non le nombre d’incréments d’un fil : F1 lit 0 ; F2 fait \(n-1\) incréments ; F1 écrit 1, ce qui efface le travail de F2 ; F2 lit 1 ; F1 fait ses \(n-1\) incréments restants ; F2 ajoute 1 à la valeur 1 qu’il a lue et écrit 2 en dernier, ce qui efface le travail de F1. Le rejeu de cet ordre donne bien 2 pour \(n = 2, 3, 4\). Avec un million d’incréments par fil, le résultat peut donc aller de 2 à \(2\,000\,000\).
Sections critiques
Une section critique est une portion de code qui accède à une ressource partagée et qui ne doit pas être exécutée par plusieurs fils simultanément.
Pour résoudre les courses critiques, il faut garantir l’exclusion mutuelle : un seul fil à la fois peut exécuter la section critique.
Identifier les sections critiques dans le code suivant :
solde_bancaire = 1000
historique = []
def retrait(montant):
global solde_bancaire
# Ligne A : Vérification
if solde_bancaire >= montant:
# Ligne B : Calcul local
nouveau_solde = solde_bancaire - montant
time.sleep(0.001) # Simule un délai
# Ligne C : Mise à jour
solde_bancaire = nouveau_solde
# Ligne D : Logging
historique.append(f"Retrait de {montant}")
return True
return FalseQuelles lignes forment une section critique ? Que se passe-t-il si deux fils appellent retrait(800) simultanément alors que le solde est de 1000 F CFA ?
Afficher la correction
Section critique : Lignes A, B et C forment une section critique car elles lisent, modifient et écrivent solde_bancaire.
Problème potentiel :
- Fil 1 : Vérifie 1000 >= 800 ✓
- Fil 2 : Vérifie 1000 >= 800 ✓
- Fil 1 : Calcule nouveau_solde = 200
- Fil 2 : Calcule nouveau_solde = 200
- Fil 1 : Écrit solde = 200
- Fil 2 : Écrit solde = 200
Résultat : 1600 F versés pour un compte de 1000 F, et le solde affiche 200 F, une valeur qui paraît normale : c’est le solde après un seul retrait. L’écriture du fil 2 a écrasé celle du fil 1, et un débit de 800 F n’est enregistré nulle part (le vrai solde serait de \(-600\) F). Une exécution avec deux fils le confirme : les deux appels rendent True, le solde final vaut 200 et l’historique contient deux lignes.
Note : La ligne D (logging) fait aussi partie de la section critique car elle modifie historique, une ressource partagée.
Correction : un verrou autour des lignes A à D (section suivante) fait de la vérification et de la mise à jour un seul bloc. Le second fil trouve alors un solde de 200 F, et son retrait est refusé : les appels rendent True puis False.
Erreur fréquente : ne protéger que la ligne C. Les deux fils passent quand même la vérification avec le même solde de 1000 F : la vérification et l’action doivent être dans la même section critique. Autre piège, sans la déclaration global solde_bancaire, l’affectation de la ligne C rend solde_bancaire locale à la fonction, et la ligne A lève UnboundLocalError dès le premier appel.
Mécanismes de synchronisation
Verrou (mutex)
Un verrou (mutex, verrou d’exclusion mutuelle) est un mécanisme de synchronisation qui permet de protéger une section critique en garantissant qu’un seul fil peut l’exécuter à la fois.
Opérations fondamentales :
lock(): Acquérir le verrou (bloquer si déjà verrouillé)unlock(): Libérer le verrou
from threading import Thread, Lock
compteur = 0
mutex = Lock() # Création du mutex
def incrementer_securise():
global compteur
for i in range(1000000):
mutex.acquire() # Acquérir le verrou
try:
compteur = compteur + 1
finally:
mutex.release() # TOUJOURS libérer !
# Ou mieux, avec context manager :
def incrementer_propre():
global compteur
for i in range(1000000):
with mutex: # Acquisition et libération automatiques
compteur = compteur + 1Avec deux fils et un million d’incréments chacun, les deux versions donnent exactement 2000000. Un fil qui trouve le verrou pris est bloqué : il attend dans la file du verrou sans consommer de calcul, et un autre fil en attente reçoit le verrou à chaque release(). La Figure 3 suit trois fils qui incrémentent compteur sous la protection d’un même verrou. Python ne garantit pas l’ordre dans lequel les fils en attente reçoivent le verrou ; la figure suit leur ordre d’arrivée.
Après l’étape 3 : F1 dans la section critique, F2 et F3 bloqués
Après l’étape 6 : chacun est passé à son tour, compteur vaut 3
Erreurs fréquentes avec les verrous :
Oublier de libérer :
mutex.acquire() compteur += 1 # Si une exception se produit ici, le mutex n'est jamais libéré ! # Tous les autres threads sont bloqués indéfiniment mutex.release()Solution : Toujours utiliser un bloc try-finally ou un context manager (with).
Verrouiller trop longtemps :
with mutex: compteur += 1 calcul_long() # ✗ Mauvais ! Autres threads bloqués ecrire_fichier() # ✗ Opération I/O lenteSolution : Ne protéger que l’accès à la ressource partagée.
Interblocage par verrouillage multiple :
# Thread 1 with mutex_A: with mutex_B: # Attend mutex_B ... # Thread 2 with mutex_B: with mutex_A: # Attend mutex_A ... # DEADLOCK ! Chacun attend l'autre
Sémaphores
Un sémaphore est un compteur qui limite le nombre de fils pouvant accéder simultanément à une ressource.
- Sémaphore binaire (compteur = 0 ou 1) : Équivalent à un verrou pour l’exclusion mutuelle ; mais un sémaphore n’a pas de propriétaire : n’importe quel fil peut faire
release() - Sémaphore compteur (compteur > 1) : Limite le nombre d’accès concurrents
Opérations :
acquire(): Décrémenter (bloquer si compteur = 0)release(): Incrémenter (et réveiller un fil en attente)
from threading import Semaphore, Thread
import time
# Maximum 3 connexions simultanées à la base de données
semaphore_bd = Semaphore(3)
def requete_base_donnees(id_thread):
print(f"Thread {id_thread} : En attente de connexion...")
with semaphore_bd: # Acquire automatique
print(f"Thread {id_thread} : Connecté à la BD")
time.sleep(2) # Simulation requête
print(f"Thread {id_thread} : Requête terminée")
# Release automatique en sortant du with
print(f"Thread {id_thread} : Connexion fermée")
# Lancer 10 threads
threads = []
for i in range(10):
t = Thread(target=requete_base_donnees, args=(i,))
t.start()
threads.append(t)
for t in threads:
t.join()Les trois premiers fils prennent les trois places ; les sept autres attendent dans acquire(). Chaque fois qu’un fil sort du bloc with, son release() réveille un fil en attente, qui prend la place : jamais plus de 3 connexions à la fois. Les fils entrent donc par vagues de 3, et la durée totale est de \(\lceil 10/3\rceil = 4\) vagues de 2 s, soit 8 s. Une exécution mesure au plus 3 connexions simultanées et 8,00 s au total. Comme pour le verrou, l’ordre de réveil n’est pas garanti ; ici, les fils sont entrés dans l’ordre d’arrivée. La Figure 4 montre le compteur du sémaphore et la frise des connexions.
Dans la sortie, deux messages se retrouvent parfois sur la même ligne (Thread 1 : En attente de connexion...Thread 0 : Connecté à la BD) : print écrit le texte, puis le saut de ligne, et un autre fil peut écrire entre les deux. L’affichage est lui aussi une ressource partagée.
\(t=0\) : trois fils connectés, compteur à 0, sept en attente
\(t=6\) s : le dernier fil entre, le compteur remonte à 2
Semaphore(3), 2 s par requête. Frise : en bleu, connecté ; hachuré, en attente ; trait rouge : instant de l’étape ; cadre rouge : modifié à l’étape.
Barrières de synchronisation
Une barrière est un point de synchronisation où tous les fils doivent attendre que tous les autres fils l’aient atteint avant de continuer.
from threading import Thread, Barrier
import time, random
# 4 threads doivent se synchroniser
barriere = Barrier(4)
def travailleur(id_thread):
print(f"Thread {id_thread} : Phase 1 - Chargement des données")
time.sleep(random.uniform(0.5, 2))
print(f"Thread {id_thread} : Données chargées")
# Point de synchronisation
barriere.wait() # Attendre que les 4 threads arrivent ici
print(f"Thread {id_thread} : Phase 2 - Traitement")
time.sleep(random.uniform(0.5, 2))
print(f"Thread {id_thread} : Traitement terminé")
# Deuxième barrière
barriere.wait()
print(f"Thread {id_thread} : Phase 3 - Finalisation")
# Résultat : Tous les threads terminent phase 1 avant de commencer phase 2Le même objet Barrier sert aux deux points de synchronisation : il se réarme dès que les quatre fils l’ont franchi. Chaque fil attend donc le plus lent, et la durée totale est la somme des phases les plus lentes. Dans une exécution, les durées tirées pour la phase 1 sont 0,86 ; 1,05 ; 1,44 ; 0,52 s (fils 0 à 3) : le fil 3 arrive le premier, le fil 2 le dernier, à 1,44 s ; avec les phases 2, la durée totale est \(1,44 + 1,76 = 3,20\) s, et la mesure donne 3,20 s (Figure 5).
Un programme de simulation parallèle a 3 fils qui doivent :
- Initialiser leurs données (temps variable)
- Se synchroniser
- Calculer en parallèle (100 itérations)
- Se synchroniser après chaque itération
- Afficher les résultats finaux
Implémenter ce programme avec des barrières. Combien de barrières sont nécessaires ?
Afficher la correction
from threading import Thread, Barrier
import time, random
N_THREADS = 3
N_ITERATIONS = 100
# Deux barrières : une pour l'init, une pour les itérations
barriere_init = Barrier(N_THREADS)
barriere_iteration = Barrier(N_THREADS)
resultats = [0] * N_THREADS
def simulation(id_thread):
# Phase 1 : Initialisation
print(f"Thread {id_thread} : Initialisation...")
time.sleep(random.uniform(0.1, 0.5))
resultats[id_thread] = id_thread * 10
# Attendre que tous soient initialisés
barriere_init.wait()
print(f"Thread {id_thread} : Début simulation")
# Phase 2 : Calcul itératif
for iteration in range(N_ITERATIONS):
# Calcul local
resultats[id_thread] += random.random()
# Synchronisation : attendre tous les threads
barriere_iteration.wait()
# Tous les threads sont au même point ici
if id_thread == 0:
print(f"Itération {iteration+1}/{N_ITERATIONS} complète")
# Phase 3 : Finalisation
print(f"Thread {id_thread} : Résultat final = {resultats[id_thread]:.2f}")
threads = [Thread(target=simulation, args=(i,)) for i in range(N_THREADS)]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"Simulation terminée. Résultats : {[f'{r:.2f}' for r in resultats]}")Nombre de barrières :
- 1 barrière pour l’initialisation : garantit que tous les fils sont prêts
- 1 barrière dans la boucle (appelée 100 fois) : synchronise chaque itération
- Total : 2 objets Barrier, mais la barrière d’itération est appelée 100 fois
Chaque fil appelle donc wait() \(1 + 100 = 101\) fois (compté à l’exécution : 101 pour chacun des trois fils). Comme un objet Barrier se réarme, un seul objet pourrait servir aux deux usages ; deux objets rendent le programme plus lisible.
Erreur fréquente : créer Barrier(4) pour 3 fils. Le quatrième participant n’arrive jamais : les trois fils restent bloqués dans wait().
Interblocage et livelock
Interblocage (deadlock)
Un interblocage se produit quand deux ou plusieurs fils sont bloqués indéfiniment, chacun attendant qu’un autre libère une ressource.
Les quatre conditions nécessaires pour un interblocage (conditions de Coffman) :
- Exclusion mutuelle : Les ressources ne peuvent pas être partagées
- Hold and wait : Un fil garde une ressource et en attend une autre
- Pas de préemption : On ne peut pas forcer un fil à libérer une ressource
- Attente circulaire : T1 attend T2, T2 attend T3, …, Tn attend T1
Les quatre conditions sont nécessaires : empêcher une seule d’entre elles rend l’interblocage impossible.
from threading import Lock, Thread
import time
mutex_A = Lock()
mutex_B = Lock()
def thread_1():
print("Thread 1 : Demande mutex_A...")
with mutex_A:
print("Thread 1 : mutex_A acquis")
time.sleep(0.1) # Permet à Thread 2 d'acquérir mutex_B
print("Thread 1 : Demande mutex_B...")
with mutex_B: # ✗ BLOQUÉ ! Thread 2 a mutex_B
print("Thread 1 : mutex_B acquis")
print("Thread 1 : Travail terminé")
def thread_2():
print("Thread 2 : Demande mutex_B...")
with mutex_B:
print("Thread 2 : mutex_B acquis")
time.sleep(0.1) # Permet à Thread 1 d'acquérir mutex_A
print("Thread 2 : Demande mutex_A...")
with mutex_A: # ✗ BLOQUÉ ! Thread 1 a mutex_A
print("Thread 2 : mutex_A acquis")
print("Thread 2 : Travail terminé")
# Lancer les threads
t1 = Thread(target=thread_1)
t2 = Thread(target=thread_2)
t1.start()
t2.start()
t1.join() # Attendra indéfiniment
t2.join()À l’exécution, les six premiers messages s’affichent (trois par fil : il demande son premier verrou, l’obtient, puis demande le second) et plus rien ensuite : trois secondes plus tard, les deux fils sont toujours vivants et les deux verrous toujours pris.
Graphe d’attente. Pour voir un interblocage, on dessine les fils (cercles) et les ressources (carrés) : une flèche d’une ressource vers un fil indique que le fil la détient, une flèche d’un fil vers une ressource qu’il la demande. En remplaçant chaque demande par une flèche vers le détenteur de la ressource, on obtient le graphe d’attente entre fils. Il y a interblocage dès que ce graphe contient un cycle : c’est la condition d’attente circulaire. La Figure 6 montre l’exemple au moment où le cycle se ferme, puis la correction par un ordre global (solution 1 ci-dessous). Un programme qui parcourt tous les ordres d’exécution possibles des deux fils trouve exactement un état bloqué, celui-ci ; avec l’ordre global, aucun.
T2 demande A : cycle T1 \(\rightarrow\) T2 \(\rightarrow\) T1, interblocage
Avec l’ordre global (A puis B) : T2 attend A sans rien tenir, pas de cycle
L’interblocage classique : le dîner des philosophes
5 philosophes sont assis autour d’une table. Entre chaque philosophe, il y a une fourchette. Pour manger, un philosophe doit prendre la fourchette à sa gauche ET celle à sa droite.
Scénario d’interblocage :
- Tous les philosophes prennent simultanément la fourchette à leur gauche
- Chacun attend maintenant la fourchette à sa droite
- Mais toutes les fourchettes de droite sont déjà prises (chacune par le voisin de droite, qui l’a prise comme fourchette gauche)
- Interblocage ! Personne ne peut manger, tous attendent indéfiniment
Dans le graphe d’attente, chaque philosophe attend son voisin de droite : P0 \(\rightarrow\) P1 \(\rightarrow\) P2 \(\rightarrow\) P3 \(\rightarrow\) P4 \(\rightarrow\) P0, un cycle (Figure 7). Avec un ordre global sur les fourchettes (chacun prend d’abord celle de plus petit numéro), le philosophe P4 commence par la fourchette 0 : s’il ne l’obtient pas, il attend sans rien tenir, et son voisin P3 peut prendre les deux siennes. Programmés avec 5 fils et 5 verrous, les deux cas se comportent ainsi : dans le premier, chaque fil tient une fourchette et attend l’autre ; dans le second, les cinq philosophes mangent leurs trois repas.
Prévention de l’interblocage
Solution 1 : Ordre global d’acquisition
# Toujours acquérir les mutex dans le MÊME ordre
def thread_1_securise():
with mutex_A: # Toujours A en premier
with mutex_B: # Puis B
print("Thread 1 : Travail")
def thread_2_securise():
with mutex_A: # MÊME ordre : A en premier
with mutex_B: # Puis B
print("Thread 2 : Travail")
# Plus de deadlock possible !
# Si T1 a mutex_A, T2 attend mutex_A (pas de cycle)L’ordre global supprime l’attente circulaire. Avec plus de deux verrous, il faut un ordre total (par numéro ou par nom), le même pour tous les fils.
Solution 2 : Timeout
def thread_avec_timeout():
if mutex_A.acquire(timeout=1.0): # Attendre max 1 seconde
try:
if mutex_B.acquire(timeout=1.0):
try:
print("Travail effectué")
finally:
mutex_B.release()
else:
print("Timeout mutex_B, abandon")
finally:
mutex_A.release()
else:
print("Timeout mutex_A, abandon")Après une seconde, le fil renonce et rend ce qu’il tient : le délai attaque la condition de non-préemption (ou de hold and wait, selon les auteurs). Il ne prouve pas qu’il y avait un interblocage ; il borne l’attente. Avec deux fils qui prennent les verrous en ordres opposés et le même délai : au bout de 1 s, le délai du premier fil expire, il abandonne et rend son verrou ; l’autre l’obtient juste avant son propre délai et travaille (80 exécutions sur 80, machine au repos). Si les deux délais expirent ensemble, les deux abandonnent : il faut réessayer, et pas au même rythme.
Solution 3 : Try-lock et backoff
def thread_avec_backoff():
while True:
if mutex_A.acquire(blocking=False): # Non-bloquant
try:
if mutex_B.acquire(blocking=False):
try:
print("Travail effectué")
return # Succès !
finally:
mutex_B.release()
else:
# Échec mutex_B, libérer mutex_A et réessayer
pass
finally:
mutex_A.release()
# Attendre un peu avant de réessayer
time.sleep(random.uniform(0.001, 0.01))Avec blocking=False, acquire rend False sans attendre si le verrou est pris : le fil n’attend jamais en tenant un verrou. L’attente aléatoire évite que les deux fils recommencent au même instant ; dans une exécution avec deux fils en ordres opposés, les deux réussissent, en 1 et 2 essais.
Livelock
Un livelock se produit quand des fils restent actifs mais ne progressent pas, car ils réagissent continuellement aux actions des autres sans avancer.
Deux personnes se croisent dans un couloir étroit :
- Personne A se déplace à gauche
- Personne B se déplace aussi à gauche (pour éviter A)
- Personne A voit B et se déplace à droite
- Personne B voit A et se déplace aussi à droite
- Personne A retourne à gauche…
- Les deux bougent constamment mais aucune n’avance !
Contrairement à l’interblocage où tout est figé, dans un livelock les fils sont actifs mais improductifs.
def thread_livelock_A():
while True:
if not mutex_A.locked():
mutex_A.acquire()
if mutex_B.locked(): # B occupé
mutex_A.release() # Libérer A pour être "poli"
time.sleep(0.001) # Attendre un peu
continue
mutex_B.acquire()
# Travail...
mutex_B.release()
mutex_A.release()
break
def thread_livelock_B():
while True:
if not mutex_B.locked():
mutex_B.acquire()
if mutex_A.locked(): # A occupé
mutex_B.release() # Libérer B pour être "poli"
time.sleep(0.001) # Attendre un peu
continue
mutex_A.acquire()
# Travail...
mutex_A.release()
mutex_B.release()
break
# Problème : tant qu'ils avancent au même rythme, les deux
# threads se libèrent mutuellement en boucle sans progresserLa Figure 8 déroule ce code en supposant que les deux fils exécutent leurs lignes au même rythme : au deuxième tour, l’état est exactement celui du premier, et rien ne permet d’en sortir. En pratique, les écarts de temps entre fils brisent souvent la symétrie. Exécuté tel quel 20 fois, ce programme ne fait aucune reprise : le fil A termine avant que B ait démarré. Avec un départ simultané (une barrière) et 10 ms de travail après la prise du premier verrou, un des deux fils recommence une fois, puis passe, et cela 20 fois sur 20. Rien ne garantit que l’écart apparaisse : le livelock est un risque, pas une fatalité, et un délai aléatoire le rend improbable.
Tour 2 : même état qu’au tour 1, aucun travail fait
Backoff aléatoire : A attend moins longtemps et fait son travail
Solution au livelock : Ajouter un délai aléatoire (backoff exponentiel) pour briser la symétrie.
Patrons (patterns) de parallélisation
Patron fork-join
Le patron Fork-Join consiste à :
- Fork : Diviser une tâche en sous-tâches parallèles
- Exécution : Traiter les sous-tâches en parallèle
- Join : Attendre que toutes les sous-tâches se terminent
- Fusion : Combiner les résultats
from concurrent.futures import ThreadPoolExecutor
import time
def sous_tache(n):
"""Simule une tâche qui prend du temps"""
print(f"Début sous-tâche {n}")
time.sleep(1)
resultat = n * n
print(f"Fin sous-tâche {n}, résultat = {resultat}")
return resultat
def fork_join_exemple():
donnees = [1, 2, 3, 4, 5, 6, 7, 8]
# Fork : Créer un pool de workers
with ThreadPoolExecutor(max_workers=4) as executor:
# Lancer toutes les tâches en parallèle
futures = [executor.submit(sous_tache, n) for n in donnees]
# Join : Attendre et collecter les résultats
resultats = [future.result() for future in futures]
# Fusion : Combiner les résultats
total = sum(resultats)
print(f"Somme totale : {total}")
return total
fork_join_exemple()Les 8 sous-tâches d’une seconde passent sur 4 workers en 2 vagues : le programme affiche Somme totale : 204 après 2 s environ (2,00 s mesurées), au lieu de 8 s à la suite. Les appels submit font créer au pool jusqu’à 4 workers, qu’il garde d’une tâche à l’autre : il évite de créer un fil par sous-tâche (Figure 9).
Patron pipeline
Le patron Pipeline consiste à diviser le traitement en étapes successives, chacune dans son fil : les étapes s’exécutent en parallèle, chacune sur un élément différent, et des files relient une étape à la suivante.
from queue import Queue
from threading import Thread
# Files pour passer les données entre étapes
queue_chargement = Queue()
queue_traitement = Queue()
def etape_chargement(fichiers):
"""Étape 1 : Charger les images"""
for fichier in fichiers:
print(f"Chargement : {fichier}")
image = charger_image(fichier) # Fonction hypothétique
queue_chargement.put(image)
queue_chargement.put(None) # Signal de fin
def etape_traitement():
"""Étape 2 : Traiter les images"""
while True:
image = queue_chargement.get()
if image is None:
queue_traitement.put(None)
break
print(f"Traitement : {image}")
image_traitee = appliquer_filtre(image)
queue_traitement.put(image_traitee)
def etape_sauvegarde():
"""Étape 3 : Sauvegarder les résultats"""
while True:
image = queue_traitement.get()
if image is None:
break
print(f"Sauvegarde : {image}")
sauvegarder_image(image)
# Lancer le pipeline
fichiers = ["img1.jpg", "img2.jpg", "img3.jpg"]
t1 = Thread(target=etape_chargement, args=(fichiers,))
t2 = Thread(target=etape_traitement)
t3 = Thread(target=etape_sauvegarde)
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()Les fonctions charger_image, appliquer_filtre et sauvegarder_image sont hypothétiques. Pendant que l’étape 2 traite une image, l’étape 1 charge la suivante et l’étape 3 sauvegarde la précédente. Pour \(n\) éléments et des étapes de durées \(d_1, \dots, d_k\), la durée est \[T = (d_1 + \dots + d_k) + (n-1) \times \max_i d_i :\] la première image traverse toutes les étapes, puis une image sort à chaque passage de l’étape la plus lente. Avec des étapes de 0,1 s, 0,3 s et 0,2 s pour les trois images, \(T = 0{,}6 + 2 \times 0{,}3 = 1{,}2\) s, contre \(3 \times 0{,}6 = 1{,}8\) s à la suite ; une exécution mesure 1,20 s (Figure 10).
Patron producteur-consommateur
Le patron Producteur-Consommateur utilise une file partagée où :
- Les producteurs ajoutent des éléments
- Les consommateurs retirent et traitent des éléments
- La file gère automatiquement la synchronisation
from queue import Queue, Empty
from threading import Thread
import time
import random
# File thread-safe (synchronisée automatiquement)
file_taches = Queue(maxsize=5) # Taille limitée
def producteur(id_prod, n_taches):
"""Génère des tâches"""
for i in range(n_taches):
tache = f"Tâche-{id_prod}-{i}"
print(f"Producteur {id_prod} : Création {tache}")
file_taches.put(tache) # Bloque si file pleine
time.sleep(random.uniform(0.1, 0.5))
print(f"Producteur {id_prod} : Terminé")
def consommateur(id_cons):
"""Traite les tâches"""
while True:
try:
# Timeout pour éviter l'attente infinie
tache = file_taches.get(timeout=2)
print(f"Consommateur {id_cons} : Traitement {tache}")
time.sleep(random.uniform(0.2, 0.8)) # Simule travail
file_taches.task_done() # Signaler que la tâche est finie
except Empty:
print(f"Consommateur {id_cons} : Plus de tâches, arrêt")
break
# Lancer 2 producteurs et 3 consommateurs
producteurs = [Thread(target=producteur, args=(i, 5))
for i in range(2)]
consommateurs = [Thread(target=consommateur, args=(i,))
for i in range(3)]
for t in producteurs + consommateurs:
t.start()
for t in producteurs:
t.join()
# Attendre que toutes les tâches soient traitées
file_taches.join()
print("Toutes les tâches sont terminées !")put attend une place libre quand la file contient déjà 5 éléments, get attend un élément quand elle est vide. Chaque put ajoute une tâche non terminée, chaque task_done en retire une, et file_taches.join() attend qu’il n’en reste aucune. La Figure 11 montre la file qui se remplit quand les consommateurs sont occupés, puis le producteur bloqué sur une file pleine. Dans une exécution, les 10 tâches sont traitées (3, 4 et 3 par consommateur) et la file n’a jamais contenu plus de 2 tâches : les consommateurs suivent le rythme des producteurs.
L’arrêt par délai (get(timeout=2)) a un défaut : un producteur qui reste plus de 2 s sans déposer arrête les consommateurs trop tôt. Un signal de fin ne dépend d’aucune durée : après les producteurs, le programme dépose un None par consommateur, et chaque consommateur s’arrête quand il en reçoit un (Lab 4).
File pleine : P0 bloqué dans put
C1 retire la plus ancienne tâche, P0 dépose la sienne
Queue(maxsize=5) ; tâche \(i\)-\(k\) : \(k\)-ième tâche du producteur \(i\). Ligne jaune : opération de l’étape ; cadre rouge : modifié à l’étape.
Ordonnancement et équilibrage de charge
Types d’équilibrage de charge
L’équilibrage de charge consiste à répartir le travail entre les fils/processus pour optimiser l’utilisation des ressources.
Deux approches principales :
- Statique : Répartition fixée à l’avance
- Dynamique : Répartition adaptative pendant l’exécution
Équilibrage statique :
- Avantages : peu de surcoût, prédictible
- Inconvénients : Inefficace si les tâches ont des durées variables
- Usage : Tâches homogènes, charge prévisible
Équilibrage dynamique :
- Avantages : S’adapte aux tâches hétérogènes, meilleure utilisation
- Inconvénients : surcoût de coordination (une file partagée)
- Usage : Tâches hétérogènes, charge imprévisible
La Figure 12 compare les deux approches sur 12 tâches de durées inégales (5, 4, 3, puis neuf tâches de 1 ou 2 s, soit 22 s de travail) et 4 fils. En statique, chaque fil reçoit 3 tâches consécutives : le fil 0 hérite des trois plus longues et finit à 12 s, pendant que les autres s’arrêtent après 3 ou 4 s. En dynamique, chaque fil libre prend la tâche suivante dans une file : les petites tâches remplissent les trous et tout est fini à 6 s, près de la borne \(22/4 = 5{,}5\) s.
100 fichiers, dont la taille varie de 1 Ko à 1 Go, sont à traiter sur 4 processeurs.
a) Avec équilibrage statique (25 fichiers par processeur), quel est le problème probable ?
b) Proposer une stratégie d’équilibrage dynamique.
c) Implémenter cette stratégie en Python avec ThreadPoolExecutor.
Afficher la correction
a) Problème statique :
Si on divise en 4 groupes de 25 fichiers, un processeur pourrait recevoir tous les gros fichiers (1 Go chacun) et finir très tard, pendant que les autres processeurs restent inactifs après avoir traité leurs petits fichiers (1 Ko).
Résultat : 3 processeurs inactifs, 1 processeur surchargé, mauvaise efficacité.
b) Stratégie dynamique :
- Option 1 - File de tâches : Tous les fichiers dans une file, chaque processeur prend le prochain fichier disponible
- Option 2 - Tri par taille : Trier les fichiers par taille décroissante, traiter les gros d’abord (avec la file de l’option 1, les petits fichiers comblent la fin)
- Option 3 - Work stealing : Les processeurs inactifs “volent” des tâches aux processeurs occupés
c) Implémentation (Option 1 - File de tâches) :
from concurrent.futures import ThreadPoolExecutor, as_completed
import os
import time
def traiter_fichier(chemin_fichier):
"""Traite un fichier (durée proportionnelle à la taille)"""
taille = os.path.getsize(chemin_fichier)
print(f"Début traitement : {chemin_fichier} ({taille} octets)")
# Simuler traitement proportionnel à la taille
time.sleep(taille / 1_000_000) # 1 sec par Mo
print(f"Fin traitement : {chemin_fichier}")
return chemin_fichier, taille
def equilibrage_dynamique(liste_fichiers, n_workers=4):
"""Équilibrage dynamique avec ThreadPoolExecutor"""
with ThreadPoolExecutor(max_workers=n_workers) as executor:
# Soumettre toutes les tâches (non-bloquant)
futures = {executor.submit(traiter_fichier, f): f
for f in liste_fichiers}
# Traiter les résultats au fur et à mesure
resultats = []
for future in as_completed(futures):
fichier, taille = future.result()
resultats.append((fichier, taille))
print(f"✓ Complété : {fichier}")
return resultats
# Utilisation
fichiers = [f"fichier_{i}.dat" for i in range(100)]
resultats = equilibrage_dynamique(fichiers, n_workers=4)
print(f"\nTotal : {len(resultats)} fichiers traités")Explication :
ThreadPoolExecutor implémente automatiquement une file de tâches avec équilibrage dynamique :
- Les 4 workers prennent des tâches dans la file
- Dès qu’un worker finit, il prend la prochaine tâche
- Pas de worker inactif tant qu’il reste des tâches
- Bonne utilisation des ressources : seul le dernier fichier commencé peut finir pendant que les autres workers ont terminé
Mesure. Avec 100 fichiers fictifs dont les tailles sont tirées de 1 Ko à 1 Go (loi log-uniforme ; 1 Go simulé par 0,5 s d’attente), l’équilibrage statique en blocs de 25 dure 1,03 s et l’équilibrage dynamique 0,93 s (trois essais chacun) ; la borne, total divisé par 4, est de 0,85 s.
Erreurs fréquentes : le code ci-dessus lit la taille de fichiers fichier_i.dat qui doivent exister : exécuté sans ces fichiers, il lève FileNotFoundError. Et ThreadPoolExecutor convient ici parce que le traitement simulé est une attente ; pour un vrai calcul, les fils de CPython ne s’exécutent pas en parallèle (verrou global de l’interpréteur, dans la version standard ; la version 3.13t sans GIL les exécute en parallèle), et il faut des processus, avec ProcessPoolExecutor et la même interface.
Synthèse et bonnes pratiques
Checklist des problèmes de concurrence
Identifier les ressources partagées
- Variables globales ?
- Structures de données partagées ?
- Fichiers, bases de données ?
Protéger les sections critiques
- Utiliser des verrous (locks) pour l’exclusion mutuelle
- Garder les sections critiques courtes
- Toujours libérer les verrous (with statement)
Éviter les interblocages
- Ordre global d’acquisition des verrous
- Utiliser des timeouts
- Éviter d’acquérir plusieurs verrous simultanément
Choisir le bon patron
- Fork-Join pour tâches indépendantes
- Pipeline pour traitement en étapes
- Producteur-Consommateur pour flux de données
Équilibrage de charge
- Dynamique pour tâches hétérogènes
- Statique pour tâches homogènes
- Utiliser des pools de fils
Comparaison des primitives de synchronisation
| Primitive | Usage | Surcoût | Quand utiliser ? |
|---|---|---|---|
| Verrou (Lock) | Exclusion mutuelle | Faible | Protéger ressources partagées |
| Sémaphore | Limiter accès | Faible | Pool de ressources limité |
| Barrière | Synchronisation | Moyen | Phases parallèles |
| Queue | Communication | Moyen | Producteur-consommateur |
| Event | Signal binaire | Faible | Notification |
Règles d’or de la programmation parallèle
- Minimiser le partage : Privilégier les données locales
- Granularité appropriée : Ni trop fine (surcoût), ni trop grosse (déséquilibre)
- Éviter l’attente active : Utiliser des primitives de synchronisation
- Tester exhaustivement : Les bugs de concurrence sont non-déterministes
- Documenter les invariants : Quelle ressource est protégée par quel verrou ?
- Mesurer les performances : Le parallélisme peut parfois ralentir !
Exercice de synthèse
Implémenter un serveur web qui :
- Accepte des requêtes HTTP concurrentes
- Utilise un pool de fils workers (max 10)
- Limite les connexions actives avec un sémaphore
- Maintient des statistiques partagées (compteur de requêtes)
- Évite les courses critiques
- Gère proprement l’arrêt avec barrières
Contraintes :
- Utiliser threading.Thread, Lock, Semaphore, Barrier
- Le compteur de statistiques doit être exact (pas de course critique)
- Tous les fils doivent se terminer proprement
Structure suggérée :
class ServeurWeb:
def __init__(self, max_workers=10, max_connexions=5):
self.semaphore = Semaphore(max_connexions)
self.mutex_stats = Lock()
self.compteur_requetes = 0
# ...
def traiter_requete(self, requete):
# Implémenter ici
pass
def statistiques(self):
# Retourner stats de manière thread-safe
passAfficher la correction
Les workers sont des fils qui prennent les requêtes dans une file ; le sémaphore borne les connexions actives, un verrou protège les compteurs, et une barrière à max_workers + 1 participants (les workers et le fil principal) assure l’arrêt : le fil principal dépose un None par worker, puis attend à la barrière que tous aient fini.
def worker(self):
while (requete := self.file.get()) is not None:
self.traiter_requete(requete)
self.fin.wait() # Barrier(max_workers + 1)
def traiter_requete(self, requete):
with self.semaphore: # au plus max_connexions
time.sleep(0.05) # réponse HTTP simulée
with self.mutex_stats:
self.compteur_requetes += 1
def statistiques(self):
with self.mutex_stats:
return self.compteur_requetes
def arreter(self):
for _ in self.workers:
self.file.put(None) # un signal par worker
self.fin.wait() # attendre tous les workersAvec 10 workers, 5 connexions au plus et 40 requêtes de 0,05 s, une exécution compte exactement 40 requêtes, au plus 5 actives à la fois, en 0,40 s, soit \(\lceil 40/5\rceil \times 0{,}05\) ; aucun worker n’est vivant après l’arrêt.
Points à retenir
- Courses critiques : Accès concurrent non synchronisé à des ressources partagées
- Section critique : Code qui accède à une ressource partagée
- Verrou : Garantit l’exclusion mutuelle (un seul fil à la fois)
- Sémaphore : Limite le nombre de fils concurrents (compteur)
- Barrière : Point de synchronisation où tous les fils attendent
- Interblocage : Attente circulaire (fils bloqués indéfiniment)
- Livelock : Fils actifs mais ne progressent pas
- Fork-Join : Diviser, traiter en parallèle, fusionner
- Pipeline : Traitement en étapes parallèles ; \(T = \sum d_i + (n-1)\max d_i\)
- Producteur-Consommateur : File partagée entre producteurs et consommateurs
- Équilibrage statique : Répartition fixe (tâches homogènes)
- Équilibrage dynamique : Répartition adaptative (tâches hétérogènes)
Prochaine séance et références
- Prochaine séance : Séance 4, architectures distribuées et NUMA (mémoire locale et distante, placement des données, cohérence par répertoire)
- TD associé : TD de la Séance 3 (entrelacements, sections critiques, sémaphores et barrières, graphes d’attente, files bornées, pipeline et équilibrage)
- Lab Python : Lab 4, synchroniser : course et verrou, vérifier puis agir, sémaphore, producteurs et consommateurs, pipeline, interblocage
- Références : A. Silberschatz, P. B. Galvin, G. Gagne, Operating System Concepts, 10e éd., Wiley, 2018, chapitres 6 à 8 ; E. W. Dijkstra, Cooperating Sequential Processes, EWD 123, 1965 (sémaphores, philosophes) ; C. A. R. Hoare, Communicating Sequential Processes, Prentice Hall, 1985 (mise en scène des philosophes) ; E. G. Coffman, M. J. Elphick, A. Shoshani, System Deadlocks, ACM Computing Surveys 3(2), 1971, p. 67-78 ; M. McCool, A. D. Robison, J. Reinders, Structured Parallel Programming, Morgan Kaufmann, 2012 (fork-join, pipeline) ; documentation Python, modules
threading,queueetconcurrent.futures
Pour cette séance
Les slides
À lire en ligne
- TD — Séance 3 — Programmation parallèle avancée (synchronisation)
- Lab 4 — Synchroniser : verrous, sémaphores, files et interblocage
À télécharger
- Notebooks des labs de la séance 3 (ZIP, 7 Ko)