Concurrence et Synchronisation

Problème de la section critique§

Une section critique est une partie du code qui accède à des ressources partagées (mémoire, fichiers, périphériques) et qui ne doit être exécutée que par un seul thread ou processus à la fois.

Race condition§

Une race condition survient quand le résultat d’une opération dépend de l’ordre d’exécution de plusieurs threads, de façon non déterministe.

import threading

compteur = 0

def incrementer():
    global compteur
    for _ in range(100000):
        compteur += 1  # RACE CONDITION : pas atomique !
        # Equivalent à :
        # temp = compteur    # lecture
        # temp = temp + 1    # calcul
        # compteur = temp    # écriture
        # Un autre thread peut s'intercaler entre ces étapes

threads = [threading.Thread(target=incrementer) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(compteur)  # < 1 000 000 très probablement — race condition !

Conditions de validité d’une solution à la section critique§

1. Exclusion mutuelle : un seul processus à la fois dans la section critique
2. Progression : si aucun processus n’est en section critique, un processus qui veut y entrer doit pouvoir le faire dans un temps fini
3. Attente bornée : un processus en attente finit toujours par entrer (pas de famine)

Mutex (Mutual Exclusion)§

Verrou binaire : un seul thread peut le détenir à la fois.

import threading

compteur = 0
mutex = threading.Lock()

def incrementer():
    global compteur
    for _ in range(100000):
        with mutex:          # acquisition automatique + libération garantie
            compteur += 1    # section critique

threads = [threading.Thread(target=incrementer) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(compteur)  # 1 000 000 — garanti

Différence mutex vs spinlock :

• Mutex : le thread bloqué est mis en sommeil (le scheduler l’enlève du CPU). Efficace si l’attente est longue.
• Spinlock : le thread bloqué boucle activement (“spinning”) jusqu’à ce que le verrou soit libre. Consomme du CPU mais évite le coût du changement de contexte. Recommandé seulement pour des sections critiques très courtes sur des systèmes multi-cœurs.

Sémaphores§

Généralisent le mutex avec un compteur. Permettent de contrôler l’accès à un nombre limité de ressources identiques.

import threading
import time

# Sémaphore binaire (équivalent mutex)
semaphore_binaire = threading.Semaphore(1)

# Sémaphore compteur — limiter à 3 connexions simultanées
pool_connexions = threading.Semaphore(3)

def acceder_base():
    with pool_connexions:  # acquire()
        print(f"Thread {threading.current_thread().name} : connexion active")
        time.sleep(1)
    print(f"Thread {threading.current_thread().name} : connexion libérée")

threads = [threading.Thread(target=acceder_base, name=f"T{i}") for i in range(8)]
for t in threads: t.start()
for t in threads: t.join()
# Au plus 3 threads simultanément dans la section

Sémaphore vs Mutex : un mutex appartient au thread qui l’a acquis (seul ce thread peut le libérer). Un sémaphore peut être acquis par un thread et libéré par un autre — utile pour la synchronisation par signal (ex : producteur-consommateur).

Moniteurs§

Un moniteur est une abstraction de plus haut niveau qui combine des données partagées avec les verrous qui les protègent. En Python, threading.Condition en est une implémentation.

import threading

class Buffer:
    def __init__(self, taille):
        self.taille = taille
        self.elements = []
        self.condition = threading.Condition()

    def produire(self, item):
        with self.condition:
            while len(self.elements) >= self.taille:
                self.condition.wait()  # Libère le verrou et attend
            self.elements.append(item)
            print(f"Produit : {item}, buffer : {self.elements}")
            self.condition.notify_all()  # Réveille les consommateurs

    def consommer(self):
        with self.condition:
            while len(self.elements) == 0:
                self.condition.wait()  # Libère le verrou et attend
            item = self.elements.pop(0)
            print(f"Consommé : {item}, buffer : {self.elements}")
            self.condition.notify_all()  # Réveille les producteurs
            return item

Problèmes classiques§

Producteur-Consommateur§

Un ou plusieurs producteurs génèrent des données ; un ou plusieurs consommateurs les traitent. Problème : synchronisation pour éviter que le producteur écrive dans un buffer plein ou que le consommateur lise un buffer vide.

import threading
import queue
import time

file = queue.Queue(maxsize=5)

def producteur():
    for i in range(10):
        item = f"item-{i}"
        file.put(item)  # Bloque si la file est pleine
        print(f"Produit : {item}")
        time.sleep(0.1)

def consommateur():
    while True:
        item = file.get()  # Bloque si la file est vide
        print(f"Consommé : {item}")
        file.task_done()
        time.sleep(0.3)

p = threading.Thread(target=producteur)
c = threading.Thread(target=consommateur, daemon=True)
p.start()
c.start()
p.join()
file.join()

Lecteurs-Rédacteurs§

Plusieurs threads peuvent lire simultanément, mais un seul peut écrire (et aucun ne doit lire pendant l’écriture).

import threading

class VerrouLecteursRedacteurs:
    def __init__(self):
        self._lecteurs = 0
        self._mutex_lecteurs = threading.Lock()
        self._mutex_ecriture = threading.Lock()

    def acquerir_lecture(self):
        with self._mutex_lecteurs:
            self._lecteurs += 1
            if self._lecteurs == 1:
                self._mutex_ecriture.acquire()  # Premier lecteur bloque l'écriture

    def liberer_lecture(self):
        with self._mutex_lecteurs:
            self._lecteurs -= 1
            if self._lecteurs == 0:
                self._mutex_ecriture.release()  # Dernier lecteur libère l'écriture

    def acquerir_ecriture(self):
        self._mutex_ecriture.acquire()

    def liberer_ecriture(self):
        self._mutex_ecriture.release()

Note : cette implémentation favorise les lecteurs et peut affamer les rédacteurs si des lecteurs arrivent en continu. Des variantes équitables existent.

Dîner des philosophes§

5 philosophes autour d’une table circulaire. 5 fourchettes entre eux. Pour manger, chacun a besoin de 2 fourchettes. Problème : interblocage si tous prennent leur fourchette gauche simultanément.

import threading
import time
import random

NOMBRE_PHILOSOPHES = 5
fourchettes = [threading.Lock() for _ in range(NOMBRE_PHILOSOPHES)]

def philosophe(id):
    gauche = id
    droite = (id + 1) % NOMBRE_PHILOSOPHES

    for _ in range(3):
        # Solution : le philosophe 0 prend dans l'ordre inverse (casse la symétrie)
        premiere, seconde = (droite, gauche) if id == 0 else (gauche, droite)

        with fourchettes[premiere]:
            with fourchettes[seconde]:
                print(f"Philosophe {id} mange")
                time.sleep(random.uniform(0.1, 0.3))

        print(f"Philosophe {id} pense")
        time.sleep(random.uniform(0.1, 0.3))

threads = [threading.Thread(target=philosophe, args=(i,)) for i in range(NOMBRE_PHILOSOPHES)]
for t in threads: t.start()
for t in threads: t.join()

Interblocage (Deadlock)§

Conditions de Coffman§

Un interblocage survient si et seulement si les quatre conditions suivantes sont réunies simultanément :

1. Exclusion mutuelle : au moins une ressource est non partageable
2. Détention et attente : un processus détient une ressource et attend d’en acquérir d’autres
3. Pas de préemption : une ressource ne peut être libérée que volontairement
4. Attente circulaire : P1 attend une ressource de P2, P2 attend une ressource de P3, …, Pn attend une ressource de P1

Stratégies§

Prévention : éliminer une des conditions de Coffman. Ex : imposer un ordre d’acquisition des verrous (élimine l’attente circulaire — solution dans le dîner des philosophes).

Évitement — Algorithme du banquier (Dijkstra) : avant d’allouer une ressource, l’OS vérifie qu’il reste dans un “état sûr” (il existe un ordre d’exécution possible pour tous les processus). Nécessite de connaître les besoins maximaux à l’avance.

Détection et récupération : laisser les deadlocks se produire, les détecter (graphe d’attente avec cycle), puis les résoudre (tuer un processus, préempter une ressource).

Ignorer (autruche) : la plupart des OS (Linux, Windows) ignorent les deadlocks car leur prévention coûterait trop en performance. Si un deadlock se produit, l’utilisateur tue les processus.

# Détecter un deadlock potentiel — voir l'état des threads
kill -3 PID_JAVA   # Thread dump Java
# Python : faulthandler
import faulthandler
faulthandler.dump_traceback()

Famine (Starvation)§

Un thread attend indéfiniment car d’autres threads sont toujours prioritaires. Différent du deadlock : les autres threads progressent, mais celui en famine ne progresse jamais.

Solutions : vieillissement (augmenter progressivement la priorité d’un thread en attente), scheduling équitable (Round Robin).

Variables atomiques§

Pour des opérations simples, les variables atomiques évitent d’utiliser des mutex.

# Python n'a pas de type atomique natif comme Java (AtomicInteger)
# Mais le GIL rend certaines opérations atomiques en CPython

# Pour du vrai parallélisme (multiprocessing ou threads C) :
from multiprocessing import Value, Array

compteur = Value('i', 0)  # entier partagé entre processus

with compteur.get_lock():
    compteur.value += 1

Programmation asynchrone (asyncio)§

Alternative aux threads pour la concurrence I/O-bound. Un seul thread, cooperative multitasking via des coroutines.

import asyncio
import aiohttp  # client HTTP asynchrone

async def telecharger(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    # Téléchargements concurrents sans threads
    resultats = await asyncio.gather(*[telecharger(url) for url in urls])
    print(f"{len(resultats)} pages téléchargées")

asyncio.run(main())

Quand utiliser quoi :