Concurrence et parallélisme

Concurrence et parallélisme#

Hide code cell source

 
import matplotlib.pyplot as plt
import matplotlib.patches as patches
import numpy as np
import seaborn as sns

sns.set_theme(style="whitegrid", palette="muted", font_scale=1.1)

Concurrence vs parallélisme#

Ces deux notions sont souvent confondues, mais elles décrivent des réalités bien distinctes.

La concurrence désigne la capacité d’un programme à gérer plusieurs tâches en même temps du point de vue logique, sans que ces tâches soient nécessairement exécutées physiquement en parallèle. Un seul processeur peut exécuter des tâches concurrentes en les entrecoupant rapidement — c’est ce que fait asyncio avec sa boucle d’événements, et ce que fait le système d’exploitation avec les threads.

Le parallélisme désigne l’exécution physiquement simultanée de plusieurs tâches sur plusieurs cœurs de processeur ou plusieurs machines. Pour bénéficier d’un vrai parallélisme en Python, il faut contourner le GIL, ce qui implique généralement d’utiliser multiprocessing ou des extensions C.

Définition 38 (Concurrence et parallélisme)

  • Concurrence (concurrency) : plusieurs tâches progressent en se partageant des ressources (temps CPU, mémoire). Elles peuvent s’exécuter sur un seul cœur par entrelacement.

  • Parallélisme (parallelism) : plusieurs tâches s’exécutent simultanément sur plusieurs cœurs. Le parallélisme implique la concurrence, mais l’inverse n’est pas vrai.

Python propose trois modèles de concurrence, chacun adapté à un type de problème :

  1. asyncio : concurrence coopérative dans un seul thread, idéale pour les I/O.

  2. threading : concurrence préemptive avec plusieurs threads dans un seul processus, limitée par le GIL pour le CPU.

  3. multiprocessing : parallélisme réel avec plusieurs processus indépendants, chacun ayant son propre GIL — idéal pour le calcul intensif.

Le GIL (Global Interpreter Lock)#

Le GIL (Global Interpreter Lock) est un verrou global qui garantit qu’un seul thread Python peut exécuter du bytecode Python à la fois au sein d’un même interpréteur CPython. Il existe pour des raisons historiques : la gestion du comptage de références (mécanisme de ramasse-miettes de CPython) n’est pas thread-safe sans ce verrou.

Définition 39 (GIL)

Le GIL (Global Interpreter Lock) est un mutex qui protège l’état interne de l’interpréteur CPython. Il est acquis avant chaque exécution de bytecode Python et libéré périodiquement (ou lors d’opérations I/O). Sa conséquence principale : plusieurs threads Python ne peuvent pas exécuter du code Python en parallèle — même sur une machine multicoeur. En revanche, le GIL est libéré lors des opérations I/O et des appels à des extensions C qui le relâchent explicitement (NumPy, OpenSSL, etc.).

Conséquences pratiques du GIL#

  • Tâches I/O-bound : le GIL est libéré pendant les attentes d’I/O. Les threads permettent donc une vraie concurrence pour les programmes qui attendent beaucoup.

  • Tâches CPU-bound : le GIL empêche toute exécution parallèle. Utiliser plusieurs threads pour accélérer un calcul Python est contre-productif — la contention sur le GIL peut même ralentir le programme.

  • Extensions C : NumPy, pandas, OpenSSL, etc. peuvent relâcher le GIL pendant leurs opérations internes, permettant un vrai parallélisme pour ces parties-là.

Le no-GIL Python 3.13#

Depuis Python 3.13 (PEP 703), une version expérimentale sans GIL est disponible. Elle permet d’activer un mode free-threaded (python3.13t) où plusieurs threads peuvent exécuter du bytecode Python simultanément. Cette évolution majeure nécessite que les bibliothèques tierces soient adaptées pour être thread-safe sans compter sur le GIL. Elle est encore expérimentale en 2024-2025, mais représente l’avenir du parallélisme en Python.

threading#

Le module threading fournit des threads POSIX/Windows au niveau du système d’exploitation. Malgré le GIL, ils sont utiles pour les tâches I/O-bound : pendant qu’un thread attend une réponse réseau, un autre peut effectuer son traitement.

import threading
import time

resultats = {}

def recuperer_donnees(cle: str, duree: float):
    """Simule une requête réseau bloquante."""
    time.sleep(duree)
    resultats[cle] = f"Données de {cle} ({duree}s)"

# Exécution avec threads : les trois attentes se déroulent en parallèle
threads = [
    threading.Thread(target=recuperer_donnees, args=("API_A", 0.15)),
    threading.Thread(target=recuperer_donnees, args=("API_B", 0.10)),
    threading.Thread(target=recuperer_donnees, args=("API_C", 0.12)),
]

debut = time.perf_counter()
for t in threads:
    t.start()
for t in threads:
    t.join()   # attendre la fin de chaque thread
duree = time.perf_counter() - debut

print(f"Durée totale : {duree:.3f}s (séquentiel aurait pris ~0.37s)")
for cle, val in resultats.items():
    print(f"  {cle}: {val}")

Durée totale : 0.151s (séquentiel aurait pris ~0.37s)
  API_B: Données de API_B (0.1s)
  API_C: Données de API_C (0.12s)
  API_A: Données de API_A (0.15s)

Synchronisation entre threads#

Les threads partagent la mémoire du processus. Toute modification d’un objet partagé doit être protégée.

import threading

class CompteurThreadSafe:
    def __init__(self):
        self._valeur = 0
        self._verrou = threading.Lock()

    def incrementer(self):
        with self._verrou:   # acquiert et libère automatiquement
            self._valeur += 1

    @property
    def valeur(self):
        return self._valeur

compteur = CompteurThreadSafe()
threads = [
    threading.Thread(target=lambda: [compteur.incrementer() for _ in range(1000)])
    for _ in range(10)
]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Valeur finale : {compteur.valeur}")   # Toujours 10000

Valeur finale : 10000

Définition 40 (Primitives de threading)

  • threading.Lock : verrou simple. Un seul thread peut l’acquérir à la fois.

  • threading.RLock : verrou réentrant — le même thread peut l’acquérir plusieurs fois sans blocage (utile en cas de récursion).

  • threading.Event : drapeau binaire. Les threads peuvent wait() jusqu’à ce qu’un autre appelle set().

  • threading.Condition : combinaison d’un verrou et d’une notification. Permet à des threads d’attendre qu’une condition soit vraie (wait()/notify()/notify_all()).

  • threading.Semaphore(n) : permet à au plus n threads d’entrer dans une section simultanément.

multiprocessing#

Le module multiprocessing crée des processus séparés, chacun avec son propre interpréteur Python et donc son propre GIL. C’est la seule façon d’obtenir un vrai parallélisme CPU en Python standard.

import multiprocessing
import time

def calculer_primes(borne: int) -> int:
    """Compte les nombres premiers jusqu'à borne (CPU-intensif)."""
    def est_premier(n):
        if n < 2:
            return False
        for i in range(2, int(n**0.5) + 1):
            if n % i == 0:
                return False
        return True
    return sum(1 for n in range(2, borne) if est_premier(n))

if __name__ == "__main__":
    bornes = [50_000, 60_000, 55_000, 65_000]

    # Séquentiel
    debut = time.perf_counter()
    resultats_seq = [calculer_primes(b) for b in bornes]
    duree_seq = time.perf_counter() - debut
    print(f"Séquentiel : {duree_seq:.3f}s → {resultats_seq}")

    # Multiprocessing avec Pool
    debut = time.perf_counter()
    with multiprocessing.Pool() as pool:
        resultats_par = pool.map(calculer_primes, bornes)
    duree_par = time.perf_counter() - debut
    print(f"Parallèle  : {duree_par:.3f}s → {resultats_par}")
    print(f"Accélération : ×{duree_seq/duree_par:.1f}")

Séquentiel : 0.255s → [5133, 6057, 5590, 6493]

Parallèle  : 0.224s → [5133, 6057, 5590, 6493]
Accélération : ×1.1

Communication entre processus#

Les processus ne partagent pas la mémoire. La communication se fait par sérialisation (pickle) via des Queue ou des Pipe.

import multiprocessing

def producteur(queue: multiprocessing.Queue, items: list):
    for item in items:
        queue.put(item)
    queue.put(None)   # signal de fin

def consommateur(queue: multiprocessing.Queue, resultats: list):
    while True:
        item = queue.get()
        if item is None:
            break
        resultats.append(item * 2)

if __name__ == "__main__":
    q = multiprocessing.Queue()
    # Les listes ne se partagent pas entre processus : on utilise Manager
    with multiprocessing.Manager() as manager:
        resultats = manager.list()
        p1 = multiprocessing.Process(target=producteur, args=(q, [1, 2, 3, 4, 5]))
        p2 = multiprocessing.Process(target=consommateur, args=(q, resultats))
        p1.start(); p2.start()
        p1.join();  p2.join()
        print(list(resultats))   # [2, 4, 6, 8, 10] (ordre peut varier)

[2, 4, 6, 8, 10]

Remarque 36

shared_memory (Python 3.8+) : le sous-module multiprocessing.shared_memory permet de partager des blocs de mémoire bruts entre processus sans sérialisation, ce qui est crucial pour les grands tableaux NumPy. On crée un SharedMemory, on le mappe dans un ndarray dans chaque processus, et les lectures/écritures sont visibles partout — au prix d’une synchronisation manuelle.

concurrent.futures#

Le module concurrent.futures fournit une interface unifiée et de haut niveau pour les deux modèles (threading et multiprocessing), via les classes ThreadPoolExecutor et ProcessPoolExecutor.

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
import time

def tache_io(n: int) -> str:
    time.sleep(0.05)
    return f"IO tâche {n} terminée"

# ThreadPoolExecutor : idéal pour I/O-bound
debut = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as executor:
    # submit : soumettre une tâche, retourne un Future
    futurs = {executor.submit(tache_io, i): i for i in range(10)}

    # as_completed : itérer sur les futurs dans l'ordre d'achèvement
    for futur in as_completed(futurs):
        n = futurs[futur]
        try:
            resultat = futur.result()
        except Exception as e:
            print(f"  Tâche {n} a échoué : {e}")
        else:
            pass   # print(resultat)

duree = time.perf_counter() - debut
print(f"10 tâches I/O en {duree:.3f}s (séquentiel : ~0.50s)")

10 tâches I/O en 0.102s (séquentiel : ~0.50s)

from concurrent.futures import ProcessPoolExecutor

def carre(n: int) -> int:
    return n * n

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        # map : interface simple, même signature que map() natif
        resultats = list(executor.map(carre, range(10)))
    print(resultats)   # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

```{prf:example} Comparaison submit vs map :label: example-20-01

  • executor.submit(fn, *args) : soumet une tâche unique, retourne un Future. Utilisez as_completed() pour traiter les résultats au fur et à mesure de leur achèvement.

  • executor.map(fn, iterable) : soumet toutes les tâches, retourne les résultats dans l’ordre d’entrée (bloque jusqu’à ce que chaque résultat soit disponible). Plus simple mais moins flexible.


## Quand utiliser quoi

```{code-cell} python
:tags: [hide-input]

fig, ax = plt.subplots(figsize=(13, 6))
ax.set_xlim(0, 13)
ax.set_ylim(0, 7)
ax.axis('off')
ax.set_title(
    "Choisir le bon modèle de concurrence en Python",
    fontsize=14, fontweight='bold', pad=15
)

couleurs = {
    "asyncio": "#27ae60",
    "threading": "#3498db",
    "multiprocessing": "#e74c3c",
    "entete": "#2c3e50",
}

# En-têtes des colonnes
colonnes = [("asyncio", 1.0), ("threading", 5.0), ("multiprocessing", 9.0)]
for label, x in colonnes:
    c = couleurs[label]
    rect = patches.FancyBboxPatch((x, 5.5), 3.5, 1.0,
        boxstyle="round,pad=0.1", facecolor=c, alpha=0.85, edgecolor='white', lw=2)
    ax.add_patch(rect)
    ax.text(x + 1.75, 6.05, label, ha='center', va='center',
            fontsize=12, fontweight='bold', color='white')

# Lignes du tableau
lignes = [
    ("Type de tâche",   "I/O-bound",          "I/O-bound",          "CPU-bound"),
    ("GIL",             "Un seul thread",      "Limité par le GIL",  "Contourné"),
    ("Mémoire partagée","Oui (mono-thread)",   "Oui (avec verrous)", "Non (IPC)"),
    ("Overhead",        "Très faible",         "Faible",             "Élevé"),
    ("Complexité",      "Moyenne",             "Élevée (verrous)",   "Élevée (IPC)"),
    ("Cas d'usage",     "Serveur, webscraping","Téléchargements",    "Calcul, ML"),
]
couleurs_lignes = ["#f8f9fa", "#ffffff"] * 10

for i, (critere, val_async, val_thread, val_mp) in enumerate(lignes):
    y = 4.6 - i * 0.75
    fond = "#f0f0f0" if i % 2 == 0 else "#ffffff"

    # Cellule critère
    rect = patches.FancyBboxPatch((0.1, y - 0.3), 0.8, 0.6,
        boxstyle="round,pad=0.05", facecolor=couleurs["entete"], alpha=0.1,
        edgecolor='none')
    ax.add_patch(rect)
    ax.text(0.5, y, critere, ha='center', va='center',
            fontsize=8.5, fontweight='bold', color=couleurs["entete"])

    # Valeurs pour chaque modèle
    for j, (val, (_, x)) in enumerate(zip([val_async, val_thread, val_mp], colonnes)):
        c = list(couleurs.values())[j]
        rect = patches.FancyBboxPatch((x, y - 0.3), 3.5, 0.6,
            boxstyle="round,pad=0.05", facecolor=fond, edgecolor='#dddddd', lw=0.8)
        ax.add_patch(rect)
        ax.text(x + 1.75, y, val, ha='center', va='center', fontsize=8.5,
                color='#2c3e50')

plt.tight_layout()
plt.show()

En résumé, la règle de décision est simple :

  • Votre programme attend des I/O (réseau, disque, base de données) et vous voulez gérer des milliers de connexions simultanées → asyncio.

  • Votre programme attend des I/O mais doit utiliser des bibliothèques synchrones → threading via ThreadPoolExecutor.

  • Votre programme calcule intensivement (traitement d’images, machine learning, simulations) → multiprocessing via ProcessPoolExecutor.

  • Vous avez besoin des deux (I/O + CPU) → combinez asyncio avec ProcessPoolExecutor via loop.run_in_executor().

Résumé#

Dans ce chapitre, nous avons exploré les trois modèles de concurrence de Python :

  • La concurrence (gérer plusieurs tâches en se partageant le CPU) diffère du parallélisme (exécuter plusieurs tâches sur plusieurs cœurs simultanément). Python offre les deux, selon le modèle choisi.

  • Le GIL est un verrou global qui limite CPython à un seul thread de bytecode à la fois. Il est libéré lors des I/O et des extensions C. Python 3.13 introduit un mode expérimental sans GIL.

  • threading convient aux tâches I/O-bound : les attentes se chevauchent malgré le GIL. Les primitives (Lock, RLock, Event, Condition, Semaphore) protègent les ressources partagées.

  • multiprocessing contourne le GIL en lançant plusieurs processus indépendants — chacun avec son propre interpréteur. Idéal pour le calcul CPU-intensif. La communication passe par Queue, Pipe ou shared_memory.

  • concurrent.futures unifie les deux approches avec ThreadPoolExecutor et ProcessPoolExecutor, une interface de haut niveau basée sur submit, map et as_completed.

Dans le dernier chapitre, nous clôturons le livre avec les bonnes pratiques, les idiomes pythoniques, le style PEP 8 et les anti-patterns à éviter.

Contenu