Combinaison de future et as_completed pour la gestion de tâches concurrrentes

Cet exemple illustre l'utilisation conjointe de future et as_completed pour orchestrer des opérations parallèles en Python.

from concurrent.futures import ThreadPoolExecutor, as_completed

def traiter_element(element):
    # Simulation d'un traitement
    import time
    time.sleep(1)
    return f"Élément {element} traité"

elements = ["alpha", "beta", "gamma"]

with ThreadPoolExecutor(max_workers=2) as pool:
    correspondances = {pool.submit(traiter_element, el): el for el in elements}
    
    for tache in as_completed(correspondances):
        identifiant = correspondances[tache]
        try:
            resultat = tache.result()
            print(f"Traitement réussi de {identifiant} : {resultat}")
        except Exception as exc:
            print(f"Échec du traitement de {identifiant} : {exc}")

Exemple de sortie :

Traitement réussi de alpha : Élément alpha traité
Traitement réussi de beta : Élément beta traité
Traitement réussi de gamma : Élément gamma traité

Fonctionnalités distinctes de future et as_completed

1. L'objet future

• Un future encapsule une opération asynchrone, enregistre son statut d'exécution et conserve son résultat éventuel.
• Généré par la méthode submit des exécuteurs de concurrent.futures, il permet de superviser une tâche individuelle.

Attributs et méthodes essentiels

Gestion de l'état

• future.done() : renvoie un booléen indiquant si la tâche est achevée.
• future.running() : renvoie un booléen indiquant si la tâche est active.
• future.cancelled() : renvoie un booléen indiquant si la tâche a été annulée.

Accès aux résultats

• future.result(timeout=None) : récupère le résultat de la tâche. Bloque si la tâche est en cours jusqu'à sa fin ou expiration du délai. Propage les exceptions levées par la tâche.
• future.exception(timeout=None) : renvoie l'exception levée par la tâche ou None.
• future.add_done_callback(fn) : exécute la fonction fn lorsque la tâche se termine.

Annulation

• future.cancel() : tente d'annuler la tâche (efficace seulement si elle n'a pas démarré).

2. La fonction as_completed

• as_completed est un utilitaire du module concurrent.futures qui agrège plusieurs future pour une surveillance groupée.
• Elle prend un itérable de futures et produit un générateur qui émet les tâches terminées dans leur ordre d'achèvement.

Propriétés clés

Traitement dans l'ordre de complétion

• Les résultats sont disponibles dès que chaque tâche finit, indépendamment de leur ordre de soumission.
• Évite l'attente séquentielle de toutes les tâches, contrairement à certaines alternatives.

Exécution non bloquante

• L'appel initial ne bloque pas, mais l'itération sur le générateur permet un traitement au rythme de l'achèvement des tâches.

Interactions entre future et as_completed

• future : abstraction pour une unité de travail unique, fournissant statut et résultat.
• as_completed : mécanisme pour paralléliser et traiter un ensemble de futures de manière réactive.

Deuxième exemple d'implémentation :

from concurrent.futures import ThreadPoolExecutor, as_completed

def calculer_puissance(base, exposant):
    return base ** exposant

operations = [(2, 4), (3, 3), (5, 2), (7, 1)]

with ThreadPoolExecutor(max_workers=3) as executeur:
    taches = [executeur.submit(calculer_puissance, b, e) for b, e in operations]
    
    for tache in as_completed(taches):
        print(f"Résultat obtenu : {tache.result()}")

Sortie possible : (l'ordre varie en fonction des temps d'exécution)

Résultat obtenu : 16
Résultat obtenu : 27
Résultat obtenu : 25
Résultat obtenu : 7

Contextes d'utilisation et différences opérationnelles