Python multiprocessing : traiter des lots sur CPU

Le module multiprocessing de Python exécute le travail dans des processus distincts. Un traitement Python qui sollicite le CPU peut ainsi utiliser plusieurs cœurs. Cette approche convient aux lots divisibles en tâches indépendantes, dont le calcul justifie le démarrage des processus et les échanges de données.

Commencez par une version séquentielle correcte. Comparez ensuite ses résultats avec ceux d'un petit pool avant de modifier le nombre de processus ou de choisir une machine plus puissante. Ce guide suit cette démarche avec un calcul reproductible, puis montre comment signaler les erreurs de chaque tâche avec ProcessPoolExecutor.

Quand utiliser un pool de processus

Cherchez un travail qui passe du temps à calculer plutôt qu'à attendre : calculs numériques indépendants, transformations exécutées principalement en Python ou validation de lots séparés. Si chaque tâche dépend de l'état changeant d'une autre, revoyez d'abord le découpage.

Dans les versions classiques de CPython, le verrou global de l'interpréteur limite l'exécution simultanée du bytecode Python par les threads. Les processus constituent une option pour les calculs intensifs en Python. Les threads peuvent toutefois convenir aux attentes réseau ou disque, et certaines bibliothèques natives libèrent ce verrou. Python propose aussi des versions facultatives sans ce verrou global. Vérifiez donc votre interpréteur et vos bibliothèques. La documentation Python sur les threads, en anglais, explique cette distinction.

Un pool sur une seule machine ne fournit pas de file de tâches persistante entre serveurs. Si le travail doit survivre aux redémarrages de l'application, permettre des reprises ou s'exécuter sur plusieurs machines, consultez notre guide sur la concurrence des workers Celery, en anglais.

Un exemple complet avec multiprocessing

Enregistrez le code suivant dans batch.py. Il compte les nombres premiers dans quatre intervalles : la borne inférieure est incluse et la borne supérieure exclue. L'algorithme volontairement simple par divisions successives produit un travail CPU indépendant et des résultats faciles à vérifier. Il sert d'exemple pédagogique, pas de méthode recommandée pour générer efficacement des nombres premiers.

Ces exemples ont été exécutés sous forme de fichiers sur Windows avec CPython 3.12.14. Ils utilisent uniquement la bibliothèque standard et choisissent explicitement spawn. La validation ne comprend pas d'exécution sur Linux ni sur une VM Hivenet.

"""Small CPU batch example. Run as a file, not in a notebook cell."""
import multiprocessing as mp
from math import isqrt


def count_primes(bounds):
    start, stop = bounds
    if start > stop:
        raise ValueError("start must not exceed stop")
    total = 0
    for n in range(max(2, start), stop):
        if all(n % d for d in range(2, isqrt(n) + 1)):
            total += 1
    return total


def parallel_counts(jobs, workers=2, chunksize=1):
    if workers < 1 or chunksize < 1:
        raise ValueError("workers and chunksize must be positive")
    ctx = mp.get_context("spawn")
    with ctx.Pool(processes=workers) as pool:
        return pool.map(count_primes, jobs, chunksize=chunksize)


def main():
    jobs = [(0, 100), (100, 200), (200, 300), (300, 400)]
    serial = [count_primes(job) for job in jobs]
    parallel = parallel_counts(jobs, workers=2)
    assert parallel == serial
    for job, count in zip(jobs, parallel):
        print(f"{job}: {count}")
    print(f"total: {sum(parallel)}")
    print(f"matches serial: {parallel == serial}")


if __name__ == "__main__":
    main()

Depuis le dossier contenant le fichier, lancez :

python batch.py

Voici la sortie enregistrée :

(0, 100): 25
(100, 200): 21
(200, 300): 16
(300, 400): 16
total: 78
matches serial: True

Les versions séquentielle et parallèle trouvent toutes deux 78 nombres premiers inférieurs à 400. La validation a aussi comparé la fonction de calcul à un crible développé séparément, au lieu de se limiter à deux appels de la même fonction. Les lots vides, tâches répétées, entrées inversées, différentes tailles de blocs et erreurs des processus ont été vérifiés.

Le parent construit la liste des tâches et rassemble les résultats. Chaque processus reçoit deux entiers et renvoie un entier ; aucune tâche ne modifie les données d'une autre. Pour traiter des fichiers, cette organisation peut transmettre un chemin et renvoyer un petit résumé. Ouvrez le fichier dans le processus de travail au lieu de transmettre un fichier déjà ouvert. Pour les étapes d'extraction et de chargement, consultez notre guide de pipeline ETL Python, en anglais.

Le rôle du bloc main et de spawn

Avec spawn, le processus enfant démarre un nouvel interpréteur et importe le module principal. Le bloc if __name__ == "__main__": empêche cet import de lancer un autre pool. Définissez les fonctions de travail au niveau du module et exécutez ces exemples comme des scripts, plutôt que de les coller dans une session interactive.

La méthode de démarrage par défaut dépend du système et de la version de Python. Python 3.14 a remplacé fork par forkserver comme choix par défaut sur les plateformes POSIX compatibles. L'appel explicite get_context("spawn") clarifie le choix pour cet exemple sans modifier le choix global de l'application. Consultez la documentation Python 3.14 sur le démarrage des processus, en anglais, avant d'adapter un programme qui suppose que Linux utilise toujours fork.

Les arguments et résultats traversent les frontières des processus. La documentation de pickle, en anglais, précise les objets sérialisables et explique pourquoi les fonctions doivent être importables par leur nom. Utilisez ici des données ordinaires et des fonctions définies au niveau du module. Évitez de soumettre des lambdas, fonctions imbriquées, fichiers ouverts ou connexions actives à une base de données. Ne chargez pas de données sérialisées provenant d'une source non fiable.

Map, chunksize et ordre des résultats

Dans cet exemple, Pool.map attend les résultats et les renvoie dans l'ordre des entrées. Le zip final associe donc chaque intervalle au bon compte, même si les processus terminent dans un ordre différent. La référence de multiprocessing, en anglais, décrit les API de pool et les solutions pour récupérer les résultats progressivement.

Avec chunksize=1, chaque bloc planifié contient un intervalle. Une valeur supérieure regroupe plusieurs tâches. De plus grands blocs peuvent réduire le coût de distribution, mais un processus qui reçoit un bloc coûteux peut terminer longtemps après les autres. Choisissez cette valeur à partir de tâches représentatives ; quatre petits intervalles ne suffisent pas à déterminer un réglage de production.

Le bloc with libère le pool. Ici, l'appel bloquant à map est terminé avant la sortie du bloc. Un contexte Pool se termine en arrêtant ses processus : ne soumettez pas de travail asynchrone avant de quitter le bloc en pensant qu'il se poursuivra. Pour une fermeture explicite qui laisse les tâches finir, appelez close() pour refuser les nouvelles tâches, puis join() pour attendre.

Inspecter chaque tâche avec ProcessPoolExecutor

Un objet future permet de récupérer le résultat ou l'erreur d'une tâche soumise. Enregistrez ce second fichier sous le nom executor_demo.py, à côté de batch.py. Un intervalle est volontairement invalide.

"""Report each task result, including one deliberate invalid range."""
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor
from batch import count_primes


def main():
    jobs = [(0, 100), (10, 1), (100, 200)]
    ctx = mp.get_context("spawn")
    with ProcessPoolExecutor(max_workers=2, mp_context=ctx) as executor:
        futures = [executor.submit(count_primes, job) for job in jobs]
        for job, future in zip(jobs, futures):
            try:
                count = future.result()
            except ValueError as exc:
                print(f"{job}: ERROR {exc}")
            else:
                print(f"{job}: {count}")


if __name__ == "__main__":
    main()

Lancez le second script :

python executor_demo.py

La sortie enregistrée conserve l'erreur tout en affichant les résultats valides :

(0, 100): 25
(10, 1): ERROR start must not exceed stop
(100, 200): 21

future.result() lève l'exception de la tâche dans le processus parent. Cet exemple intercepte seulement la ValueError attendue ; il ne transforme pas silencieusement toute erreur d'infrastructure en exécution réussie. L'affichage suit l'ordre de soumission et peut donc attendre une première tâche lente. Utilisez as_completed avec une association explicite entre chaque future et sa tâche si vous préférez l'ordre d'achèvement.

Le contexte de l'exécuteur attend les tâches soumises lors de sa fermeture. Un délai dépassé pendant l'attente d'une future n'arrête pas le calcul en cours, et l'annulation d'une future ne peut pas annuler une tâche déjà en cours. Si les délais comptent, prévoyez des tâches de durée bornée et une politique explicite de reprise ou d'arrêt. La référence de concurrent.futures, en anglais, décrit ces comportements. N'appelez pas les méthodes d'un exécuteur ou d'une future depuis une fonction soumise au même pool de processus.

Choisir le nombre de processus à partir de mesures

Deux processus suffisent à montrer le fonctionnement. Ce n'est pas une recommandation de dimensionnement. Pour votre charge, gardez les mêmes données et comparez une exécution séquentielle à quelques nombres explicites de processus. Relevez la durée totale, démarrage compris, le pic de mémoire, l'exactitude des résultats et les erreurs. Répétez les essais ; ne transformez pas le meilleur résultat isolé en estimation de capacité.

Surveillez trois limites fréquentes. Une tâche très courte peut coûter plus cher à distribuer qu'à calculer. De gros arguments ou résultats augmentent les coûts de sérialisation et de mémoire. Les bibliothèques numériques qui créent leurs propres threads peuvent multiplier le travail concurrent lorsqu'elles sont lancées dans plusieurs processus. Vérifiez leurs réglages avant d'agrandir encore le pool.

Pour une grande source de tâches, ne recopiez pas sans adaptation la liste de toutes les futures de cet exemple. Limitez le travail en attente et consommez les résultats progressivement. Python 3.14 a ajouté buffersize à map de l'exécuteur pour limiter le travail soumis dont les résultats n'ont pas encore été fournis. Ce paramètre n'existe pas dans Python 3.12 utilisé pour les tests. Consultez la documentation des exécuteurs de Python 3.14, en anglais.

Exécuter le traitement sur une VM CPU

Les instances CPU de Compute with Hivenet, présentées en anglais, fournissent une VM isolée avec accès SSH pour le traitement de données et les tâches en arrière-plan. Un pool de processus s'exécute dans cette VM ; il ne se répartit pas automatiquement sur plusieurs instances.

Chez Hivenet, chaque vCPU correspond à un thread matériel. Le nombre de vCPU n'est donc pas le nombre de cœurs physiques indépendants. Choisissez une petite instance qui satisfait le besoin en mémoire, mesurez la version séquentielle sur cette machine, puis testez un pool de taille limitée. Consignez les versions de Python et des dépendances. Les exemples locaux de ce guide vérifient le fonctionnement, sans démontrer un gain de vitesse ni mesurer la capacité d'une instance Hivenet.