multiprocessing en Python pour les tâches lourdes CPU

1. Aperçu

Quand le code Python doit faire beaucoup de calculs, par exemple traiter des images, calculer des checksums ou simuler des nombres, vous verrez que l’utilisation de threads n’accélère presque pas. La cause est le GIL (Global Interpreter Lock). Cet article explique brièvement ce problème. Ensuite l’article guide l’utilisation du module multiprocessing et de ProcessPoolExecutor pour diviser le travail entre plusieurs process. Après lecture, vous saurez comment paralléliser une tâche CPU-bound, écrire des tests pour cela et éviter les erreurs courantes.

Tous les exemples utilisent Python 3.12+ et ne nécessitent que la bibliothèque standard, pas besoin d’installation supplémentaire.

2. Pourquoi le threading n’aide pas

2.1. Une tâche lourde CPU

Nous utilisons une fonction de comptage des nombres premiers écrite en Python pur. C’est du travail CPU typique :

# cpu_task.py
import math

def count_primes(start: int, end: int) -> int:
    count = 0
    for n in range(max(start, 2), end):
        if all(n % d for d in range(2, math.isqrt(n) + 1)):
            count += 1
    return count

La fonction parcourt chaque nombre dans [start, end) et essaie de le diviser par les nombres de 2 à sa racine carrée. La fonction ne lit pas de fichiers, n’effectue pas d’appels réseau. Elle ne fait que calculer.

2.2. Mesure du temps : séquentiel et thread

# bench_threads.py
import time
from concurrent.futures import ThreadPoolExecutor
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

def run_sequential() -> int:
    return sum(count_primes(a, b) for a, b in RANGES)

def run_threads() -> int:
    starts, ends = zip(*RANGES)
    with ThreadPoolExecutor(max_workers=8) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    for fn in (run_sequential, run_threads):
        t0 = time.perf_counter()
        result = fn()
        print(f"{fn.__name__}: {result} primes, {time.perf_counter() - t0:.2f}s")

RANGES divise la plage [0, 1_000_000) en 8 parties. run_threads assigne chaque partie à un thread. Résultats sur une machine à 8 cœurs (les résultats sur votre machine seront différents) :

run_sequential: 78498 primes, 6.41s
run_threads: 78498 primes, 6.58s

La version avec threads est même un peu plus lente. Dans CPython standard, à tout moment, un seul thread exécute le bytecode Python, parce qu’il doit tenir le GIL. Les huit threads ne font que se relayer sur un seul cœur, et cela a également le coût supplémentaire du changement de contexte entre threads.

Remarque : À partir de Python 3.13, il y a une version de construction free-threaded (python3.13t) qui supprime le GIL. Cette version est toujours en phase expérimentale, et de nombreuses extensions C ne la supportent pas encore. Pour la plupart des projets actuels, utiliser des process reste le moyen le plus fiable.

3. Bases de multiprocessing

Chaque processus a son propre interpréteur et son propre GIL. Ainsi, N processus peuvent s’exécuter réellement en parallèle sur N cœurs.

3.1. Créer des Process manuellement

# basic_process.py
from multiprocessing import Process, Queue
from cpu_task import count_primes

def worker(start: int, end: int, results: Queue) -> None:
    results.put(count_primes(start, end))

if __name__ == "__main__":
    results = Queue()
    ranges = [(0, 500_000), (500_000, 1_000_000)]
    processes = [Process(target=worker, args=(a, b, results)) for a, b in ranges]

    for p in processes:
        p.start()
    total = sum(results.get() for _ in processes)
    for p in processes:
        p.join()

    print("Total primes:", total)

Explication étape par étape :

• Process(target=..., args=...) décrit un processus enfant. Il ne s’exécute pas jusqu’à ce que vous appeliez start().
• Le processus enfant ne partage pas la mémoire avec le processus parent. Pour renvoyer les résultats, vous devez utiliser un canal de communication. Ici c’est Queue.
• Le programme appelle results.get() avant join(). Si vous appelez join() en premier alors que la queue contient encore de grandes données non lues, le programme peut se bloquer.

Total primes: 78498

Cette approche est flexible mais verbeux. En pratique, vous gerez rarement manuellement Process.

3.2. Utiliser Pool

Pool crée un groupe de workers à l’avance et distribue le travail automatiquement :

# pool_example.py
from multiprocessing import Pool
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

if __name__ == "__main__":
    with Pool(processes=8) as pool:
        counts = pool.starmap(count_primes, RANGES)
    print(counts)
    print("Total:", sum(counts))

starmap déplie chaque tuple dans RANGES en paramètres start, end. Les résultats sont renvoyés dans le même ordre que l’entrée :

[10746, 9895, 9622, 9480, 9637, 9554, 9454, 9610]
Total: 78498

Si vous ne passez pas processes, Pool utilise os.cpu_count() workers. Le bloc with nettoiera les workers à la fin.

4. ProcessPoolExecutor : une API plus moderne

concurrent.futures.ProcessPoolExecutor a la même interface que ThreadPoolExecutor. Grâce à cela, vous pouvez passer des threads aux process en changeant simplement le nom de la classe :

# parallel.py
import os
from concurrent.futures import ProcessPoolExecutor
from cpu_task import count_primes

def split_range(start: int, end: int, parts: int) -> list[tuple[int, int]]:
    step = (end - start + parts - 1) // parts
    return [(s, min(s + step, end)) for s in range(start, end, step)]

def parallel_count_primes(start: int, end: int, workers: int | None = None) -> int:
    workers = workers or os.cpu_count() or 1
    chunks = split_range(start, end, workers)
    starts, ends = zip(*chunks)
    with ProcessPoolExecutor(max_workers=workers) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    import time
    t0 = time.perf_counter()
    print(parallel_count_primes(0, 1_000_000, workers=8))
    print(f"{time.perf_counter() - t0:.2f}s")

78498
1.27s

Le temps a diminué d’environ 6,4 secondes à 1,3 seconde. La vitesse n’augmente pas d’exactement 8 fois pour deux raisons. Premièrement, les chunks n’ont pas une charge de travail uniforme : les chunks avec de grands nombres prennent plus de temps à vérifier. Deuxièmement, le démarrage des processus et la transmission des données prennent aussi du temps.

Si vous voulez traiter chaque résultat dès qu’il est prêt, utilisez submit avec as_completed :

from concurrent.futures import ProcessPoolExecutor, as_completed
from cpu_task import count_primes

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(count_primes, a, a + 250_000): a
                   for a in range(0, 1_000_000, 250_000)}
        for future in as_completed(futures):
            print(f"Plage à partir de {futures[future]}: {future.result()} primes")

Si un worker lève une exception, future.result() la relève dans le processus parent. Vous pouvez donc attraper les erreurs avec try/except comme d’habitude.

5. Nombreuses petites tâches : chunksize et imap_unordered

Chaque fois que vous envoyez une tâche à un worker, Python doit pickle les paramètres, les envoyer via une pipe, puis les unpickle. Si vous avez 100 000 très petites tâches, ce coût peut être plus important que le calcul lui-même. Vous résolvez cela en regroupant les tâches en batches avec chunksize :

# chunks.py
from multiprocessing import Pool

def is_prime(n: int) -> bool:
    if n < 2:
        return False
    return all(n % d for d in range(2, int(n ** 0.5) + 1))

if __name__ == "__main__":
    numbers = range(1_000_000)
    with Pool() as pool:
        total = sum(pool.imap_unordered(is_prime, numbers, chunksize=10_000))
    print(total)

• chunksize=10_000 fait que chaque envoi contient 10 000 nombres au lieu de 1. Le nombre de communications entre les processus est considérablement réduit.
• imap_unordered retourne les résultats de manière lazy et n’en préserve pas l’ordre. Ce mode convient quand vous avez seulement besoin d’agréger, par exemple en additionnant, et que vous voulez économiser la mémoire.
• ProcessPoolExecutor.map accepte aussi le paramètre chunksize. Sa valeur par défaut est 1, alors souvenez-vous de l’ajuster quand vous avez de nombreuses petites tâches.

6. Initialiser les ressources une fois par worker

Supposons que chaque tâche nécessite une table de consultation coûteuse à créer. Si vous recréez la table à chaque appel de fonction, c’est très gaspilleur. Le paramètre initializer vous permet d’exécuter une fonction exactement une fois au démarrage du worker :

# initializer_example.py
from concurrent.futures import ProcessPoolExecutor

_price_table: dict[str, float] = {}

def load_prices(currency: str) -> None:
    global _price_table
    rate = 25_000 if currency == "VND" else 1
    _price_table = {f"SKU-{i}": i * 1.5 * rate for i in range(100_000)}

def order_total(skus: list[str]) -> float:
    return sum(_price_table[s] for s in skus)

if __name__ == "__main__":
    orders = [["SKU-1", "SKU-2"], ["SKU-10", "SKU-20", "SKU-30"]]
    with ProcessPoolExecutor(max_workers=2, initializer=load_prices,
                             initargs=("VND",)) as executor:
        print(list(executor.map(order_total, orders)))

[112500.0, 2250000.0]

La variable _price_table est une variable globale dans chaque worker. Le processus parent ne voit pas ces changements. C’est un moyen sûr pour chaque worker d’avoir sa propre ressource sans avoir à envoyer la table de consultation à chaque tâche.

7. Tester le code parallèle

Vous pouvez tester le code utilisant les process avec pytest comme le code habituel. La condition est que la fonction worker doit être définie au niveau du module :

# test_parallel.py
from cpu_task import count_primes
from parallel import parallel_count_primes, split_range

def test_count_primes_known_value():
    assert count_primes(0, 100) == 25

def test_split_range_covers_whole_interval():
    chunks = split_range(0, 10, 3)
    assert chunks == [(0, 4), (4, 8), (8, 10)]

def test_parallel_matches_sequential():
    expected = count_primes(0, 50_000)
    assert parallel_count_primes(0, 50_000, workers=4) == expected

pip install pytest
pytest -q test_parallel.py

...                                                    [100%]
3 passed in 0.62s

Le test le plus important est test_parallel_matches_sequential. Il s’assure que la division des chunks ne rate aucun élément ni n’en compte deux fois. C’est une erreur très courante lors de l’écriture manuelle de split_range.

8. Erreurs courantes et quand ne pas l’utiliser

Oublier if __name__ == "__main__":.

Avec la méthode de démarrage spawn (par défaut sur Windows et macOS) ou forkserver (par défaut sur Linux à partir de Python 3.14), le processus enfant réimporte le module principal. Si le code créant le pool est au niveau du module, chaque processus enfant crée un nouveau pool. Vous obtiendrez alors une RuntimeError ou une création infinie de process.

Passer des lambdas ou des fonctions imbriquées.

Tout ce qui est envoyé aux workers doit être picklable :

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(lambda x: x * x, range(5))))

_pickle.PicklingError: Can't pickle <function <lambda> at 0x...>: attribute lookup <lambda> on __main__ failed

Définissez la fonction avec def au niveau du module. Si vous devez fixer certains paramètres, utilisez functools.partial.

Passer des données trop volumineuses.

Une liste de millions d’éléments passée à chaque tâche sera pickled plusieurs fois. Le temps de copie peut dépasser le temps de calcul. Il y a plusieurs solutions : envoyer uniquement des indices ou des chemins de fichiers, utiliser initializer, ou utiliser multiprocessing.shared_memory avec des données de type array.

Dépendre du comportement de fork.

Le code qui fonctionne sur les anciennes versions de Linux grâce à fork (le processus enfant hérite des variables globales) peut échouer sur macOS, Windows ou Python 3.14. Pour un comportement cohérent, choisissez explicitement la méthode de démarrage :

import multiprocessing as mp

if __name__ == "__main__":
    mp.set_start_method("spawn")

Remarque : Ne pas utiliser multiprocessing pour les tâches I/O-bound comme les appels HTTP ou les requêtes de base de données. Lors de l’attente d’E/S, le thread libère toujours le GIL, donc ThreadPoolExecutor ou asyncio sont beaucoup plus légers. De même, ne pas utiliser multiprocessing quand chaque tâche ne prend que quelques microsecondes sans regroupement, ou quand le calcul est déjà dans une bibliothèque C qui libère le GIL (NumPy, fonctions de hachage sur grandes données, etc.). Dans ces cas, les threads ou la vectorisation suffisent généralement.

Créer trop de workers.

Augmenter le nombre de workers au-delà du nombre de cœurs CPU n’accélère pas le code CPU-bound. Cela gaspille seulement la RAM, car chaque processus est un interpréteur complet. Commencer avec os.cpu_count() est raisonnable.

9. Conclusion

Le GIL empêche les threads d’accélérer le code Python pur lourd CPU. Chaque processus a son propre GIL, donc diviser le travail entre plusieurs processus est le moyen simple d’utiliser tous les cœurs CPU. Dans la plupart des cas, ProcessPoolExecutor est le choix le plus concis. Pool est utile quand vous avez besoin de imap_unordered ou starmap. Process manuel n’est destiné qu’aux situations particulières.

Quand vous utilisez multiprocessing, vous devez vous souvenir de quelques choses :

• Toujours avoir if __name__ == "__main__":.
• Ne passez que des fonctions et des données picklables.
• Augmentez chunksize quand vous avez de nombreuses petites tâches.
• Utilisez initializer pour les ressources lourdes.
• Écrivez un test comparant les résultats parallèles avec les résultats séquentiels pour vous assurer que le chunking est correct.