Multithreading en Python

Concurrence Python : Le `Threading`

La concurrence permet à plusieurs tâches de s’exécuter ou de gérer leur exécution sur des périodes de temps qui se chevauchent. Python offre plusieurs façons d’atteindre la concurrence, le threading étant une approche courante.

Avantages de l’utilisation des threads :

Mémoire partagée : Les threads au sein du même processus partagent le même espace mémoire, ce qui rend la communication et le partage de données entre eux relativement simples par rapport aux processus séparés (multiprocessing).
Efficacité des ressources : Les threads (souvent appelés processus légers) consomment généralement moins de ressources système (mémoire, temps d’initialisation) que des processus complets.
Réactivité : Utile pour maintenir les applications réactives (par exemple, gérer les événements d’interface graphique tout en effectuant des tâches en arrière-plan) ou gérer plusieurs opérations liées aux E/S (I/O-bound) simultanément (comme les requêtes réseau ou les opérations de fichiers).

Qu’est-ce qu’un thread ?

Un thread est la plus petite séquence d’instructions programmées qui peut être gérée indépendamment par un ordonnanceur. Un processus peut contenir plusieurs threads, tous s’exécutant dans le même contexte de processus.

Chaque thread possède son propre contexte d’exécution (incluant un pointeur d’instruction et des registres).
Les threads peuvent être mis en pause (mis en veille) ou céder le contrôle, permettant à d’autres threads de s’exécuter.
L’ordonnanceur du système d’exploitation gère le basculement entre les threads (multitasking préemptif).

Le Verrou Global de l’Interpréteur (`GIL`)

De manière cruciale, dans l’interpréteur CPython standard, le Verrou Global de l’Interpréteur (GIL) garantit qu’un seul thread exécute du bytecode Python à la fois au sein d’un processus unique, même sur des processeurs multi-cœurs. Cela simplifie la gestion de la mémoire mais limite le véritable parallélisme pour les tâches gourmandes en CPU.

Implications du GIL :

Le threading est efficace pour les tâches liées aux E/S (I/O-bound) (où les threads passent du temps à attendre des opérations externes comme les E/S réseau ou disque). Pendant qu’un thread attend, le GIL peut être libéré, permettant à un autre thread de s’exécuter.
Le threading ne fournit pas d’accélération pour les tâches liées au CPU (CPU-bound) (tâches effectuant des calculs intensifs) dans CPython, car un seul thread exécute du code Python à la fois.
Pour le parallélisme lié au CPU, utilisez le module multiprocessing, qui crée des processus séparés, chacun avec son propre interpréteur et espace mémoire, contournant le GIL.
La bibliothèque asyncio fournit un modèle de concurrence alternatif utilisant une boucle d’événements, adapté au code réseau structuré de haut niveau et à d’autres tâches liées aux E/S, souvent plus efficace que le threading pour gérer un très grand nombre de connexions concurrentes.

Le module `threading` (Recommandé)

Le module threading est l’interface moderne de haut niveau pour travailler avec les threads en Python. Il s’appuie sur le module _thread de plus bas niveau (et maintenant obsolète) (anciennement thread en Python 2). Préférez toujours threading à _thread pour le nouveau code.

Composants clés du `threading` :

threading.Thread : La classe représentant un thread d’exécution.
threading.Lock, threading.RLock : Primitives de synchronisation pour prévenir les race conditions (conditions de concurrence) lorsque plusieurs threads accèdent à des ressources partagées.
threading.Event : Un mécanisme simple de communication entre threads (un thread signale un événement, d’autres l’attendent).
threading.Condition, threading.Semaphore, threading.Barrier : Outils de synchronisation plus avancés.
threading.current_thread() : Retourne l’objet Thread actuel.
threading.active_count() : Retourne le nombre d’objets Thread actifs.

Création et démarrage de threads

Il existe deux principales façons de créer des threads en utilisant threading.Thread :

1. Passage d’un objet appelable (fonction) au constructeur :

C’est souvent l’approche la plus simple pour les tâches directes.

#!/usr/bin/env python3

import threading
import time

def print_time(thread_name, delay):
    """Affiche l'heure actuelle toutes les 'delay' secondes, 5 fois."""
    count = 0
    while count < 5:
        time.sleep(delay)
        count += 1
        current_time = time.strftime("%H:%M:%S", time.localtime())
        print(f"{thread_name}: {current_time} (Count: {count})")
    print(f"{thread_name} se termine.")

# Crée des threads en passant la fonction cible et les arguments
try:
    thread1 = threading.Thread(target=print_time, args=("Thread-1", 1))
    thread2 = threading.Thread(target=print_time, args=("Thread-2", 2))

    # Démarre les nouveaux threads
    thread1.start()
    thread2.start()

    # Attend que les deux threads se terminent avant que le thread principal ne se termine
    thread1.join()
    thread2.join()

except Exception as e:
    print(f"Erreur : impossible de démarrer le thread - {e}")

print("Thread principal terminé.")

2. Héritage de `threading.Thread` :

Redéfinissez la méthode __init__ (optionnel, appelle le __init__ parent) et la méthode run() (obligatoire, contient la logique d’exécution du thread).

#!/usr/bin/env python3

import threading
import time

class MyThread(threading.Thread):
    def __init__(self, thread_id, name, delay):
        # Appelle l'initialiseur de la classe Thread
        super().__init__()
        # Ou : threading.Thread.__init__(self)
        self.thread_id = thread_id
        self.name = name
        self.delay = delay

    def run(self):
        """Cette méthode est exécutée lorsque thread.start() est appelé."""
        print(f"Démarrage de {self.name}")
        # Appelle la fonction contenant la logique principale du thread
        print_task(self.name, self.delay)
        print(f"Sortie de {self.name}")

def print_task(thread_name, delay):
    """Une tâche simple à effectuer par le thread."""
    count = 0
    while count < 3:
        time.sleep(delay)
        count += 1
        current_time = time.strftime("%H:%M:%S", time.localtime())
        print(f"{thread_name}: {current_time} (Count: {count})")

# Crée de nouveaux threads en instanciant la sous-classe
thread1 = MyThread(1, "Thread-A", 1)
thread2 = MyThread(2, "Thread-B", 1.5)

# Démarre de nouveaux Threads (cela appelle la méthode run())
thread1.start()
thread2.start()

# Attend la fin des threads
thread1.join()
thread2.join()

print("Thread principal terminé.")

Méthodes clés des threads :

start() : Démarre l’activité du thread. Il appelle la méthode run() (ou la fonction cible) dans un thread de contrôle séparé.
run() : La méthode représentant l’activité du thread. Redéfinissez-la dans les sous-classes.
join([timeout]) : Attend que le thread se termine. Bloque le thread appelant. Si timeout est donné, il attend au maximum ce nombre de secondes.
is_alive() : Retourne True si le thread est toujours en cours d’exécution, False sinon.
name / getName() / setName() : Obtient ou définit le nom du thread.
ident : L’identifiant unique du thread (ou None s’il n’est pas démarré).

Synchronisation des threads (`Lock`)

Lorsque plusieurs threads accèdent à des ressources partagées (comme des variables ou des structures de données), des race conditions peuvent se produire, entraînant des résultats imprévisibles. Des primitives de synchronisation comme les verrous (locks) sont utilisées pour contrôler l’accès.

Un threading.Lock fournit une exclusion mutuelle. Un seul thread peut ‘acquérir’ le verrou à la fois. Les autres threads tentant de l’acquérir seront bloqués jusqu’à ce que le verrou soit ‘libéré’.

Utilisation des verrous avec l’instruction `with` (recommandé) :

L’instruction with simplifie l’utilisation des verrous en acquérant automatiquement le verrou à l’entrée du bloc et en le libérant à la sortie, même si des erreurs se produisent.

#!/usr/bin/env python3

import threading
import time

shared_counter = 0
counter_lock = threading.Lock() # Crée un objet verrou

class WorkerThread(threading.Thread):
    def __init__(self, name):
        super().__init__(name=name)

    def run(self):
        global shared_counter
        print(f"{self.name} démarre.")
        for _ in range(100000):
            # Acquérir le verrou avant d'accéder à la ressource partagée
            with counter_lock:
                # --- Début de la section critique ---
                current_value = shared_counter
                # Simule un temps de traitement
                time.sleep(0.000001)
                shared_counter = current_value + 1
                # --- Fin de la section critique ---
            # Le verrou est automatiquement libéré ici
        print(f"{self.name} a terminé.")

# Crée des threads
thread1 = WorkerThread("Worker-1")
thread2 = WorkerThread("Worker-2")

# Démarre les threads
thread1.start()
thread2.start()

# Attend la fin des threads
thread1.join()
thread2.join()

print(f"Thread principal terminé. Valeur finale du compteur : {shared_counter}")
# Sans le verrou, la valeur finale serait probablement inférieure à 200000 en raison de conditions de concurrence.

Utilisation manuelle des verrous (moins recommandé) :

Vous pouvez appeler manuellement lock.acquire() et lock.release(), mais vous devez vous assurer que release() est appelé, généralement en utilisant un bloc try...finally.

# À l'intérieur de la méthode run (alternative à 'with')
counter_lock.acquire()
try:
    # Accède à la ressource partagée
    current_value = shared_counter
    shared_counter = current_value + 1
finally:
    counter_lock.release() # S'assure que le verrou est toujours libéré

Utilisation de files d’attente pour la communication entre threads (module `queue`)

Le module queue (notez le ‘q’ minuscule en Python 3) fournit des classes de files d’attente thread-safe, idéales pour passer des données entre les threads producteurs et consommateurs.

Méthodes courantes de file d’attente :

queue.Queue(maxsize=0) : Crée une file d’attente FIFO (Premier Entré, Premier Sorti). maxsize=0 signifie une taille infinie.
put(item, block=True, timeout=None) : Ajoute un item à la file d’attente. Si block est True et que la file d’attente est pleine (si maxsize > 0), attend qu’un espace soit disponible (ou que le timeout expire).
get(block=True, timeout=None) : Supprime et retourne un item de la file d’attente. Si block est True et que la file d’attente est vide, attend qu’un item soit disponible (ou que le timeout expire).
qsize() : Retourne le nombre approximatif d’éléments dans la file d’attente.
empty() : Retourne True si la file d’attente est vide, False sinon.
full() : Retourne True si la file d’attente est pleine, False sinon.
task_done() : Appelez ceci après avoir traité un élément récupéré avec get(). Utilisé avec join().
join() : Bloque jusqu’à ce que tous les éléments de la file d’attente aient été récupérés et traités (c’est-à-dire, task_done() appelé pour chaque élément).

Autres types de files d’attente : LifoQueue (Dernier Entré, Premier Sorti), PriorityQueue.

Exemple : Modèle Producteur-Consommateur

#!/usr/bin/env python3

import queue
import threading
import time

# Une file d'attente thread-safe
work_queue = queue.Queue(10)
exit_flag = False

class ProducerThread(threading.Thread):
    def __init__(self, name, q):
        super().__init__(name=name)
        self.q = q

    def run(self):
        print(f"{self.name} démarre.")
        for i in range(1, 6):
            if exit_flag:
                break
            item = f"Data_{i}"
            self.q.put(item) # Ajoute l'élément à la file d'attente
            print(f"{self.name} a produit {item}")
            time.sleep(0.5)
        print(f"{self.name} a terminé la production.")

class ConsumerThread(threading.Thread):
    def __init__(self, name, q):
        super().__init__(name=name)
        self.q = q

    def run(self):
        print(f"{self.name} démarre.")
        while not exit_flag or not self.q.empty():
            try:
                # Obtient l'élément de la file d'attente, attend jusqu'à 1 seconde si vide
                item = self.q.get(block=True, timeout=1)
                print(f"{self.name} consomme {item}")
                # Simule un temps de traitement
                time.sleep(1)
                self.q.task_done() # Signale que le traitement est terminé
            except queue.Empty:
                if exit_flag:
                    break # Quitte si le drapeau est défini et la file d'attente est vide
                else:
                    continue # La file d'attente est temporairement vide, réessayez
            except Exception as e:
                 print(f"{self.name} a rencontré une erreur : {e}")
                 self.q.task_done() # Marque toujours la tâche comme terminée si une erreur se produit
        print(f"{self.name} a terminé la consommation.")


# Crée des threads
producer = ProducerThread("Producer", work_queue)
consumer1 = ConsumerThread("Consumer-1", work_queue)
consumer2 = ConsumerThread("Consumer-2", work_queue)

# Démarre les threads
producer.start()
consumer1.start()
consumer2.start()

# Attend que le producteur termine
producer.join()

# Attend que la file d'attente soit vide (tous les éléments traités)
print("Attente de la fin des consommateurs...")
work_queue.join() # Bloque jusqu'à ce que task_done() soit appelé pour tous les éléments

# Signale aux consommateurs de se terminer
print("Définition du drapeau de sortie.")
exit_flag = True

# Attend que les consommateurs se terminent proprement
consumer1.join()
consumer2.join()

print("Thread principal terminé.")

Multithreading en Python

Concurrence Python : Le Threading

Qu’est-ce qu’un thread ?

Le Verrou Global de l’Interpréteur (GIL)

Le module threading (Recommandé)

Composants clés du threading :