Maîtriser la communication inter-threads en Python : Synchronisation et partage de données

La communication inter-threads désigne le processus permettant la communication et la synchronisation entre threads au sein d'un programme Python multi-thread.

Généralement, les threads en Python partagent le même espace mémoire au sein d'un processus, ce qui leur permet d'échanger des données et de coordonner leurs activités via des variables partagées, des objets et des mécanismes de synchronisation spécialisés fournis par le module threading.

Pour faciliter cette communication, le module threading fournit divers primitives de synchronisation : verrous (Locks), événements (Events), conditions (Conditions) et sémaphores (Semaphores). Ce tutoriel explique comment utiliser les objets Event et Condition pour la communication entre threads.

L'objet Event

Un objet Event gère l'état d'un drapeau interne que les threads peuvent attendre ou définir. Il fournit des méthodes pour contrôler cet état, permettant aux threads de synchroniser leurs activités basées sur des conditions partagées.

Le drapeau est initialement faux, devient vrai avec set() et revient à faux avec clear(). La méthode wait() bloque jusqu'à ce que le drapeau soit vrai.

Méthodes clés

• is_set() : Retourne True si le drapeau interne est vrai.
• set() : Définit le drapeau à vrai. Tous les threads en attente sont réveillés.
• clear() : Réinitialise le drapeau à faux. Les appels ultérieurs à wait() bloqueront jusqu'à un nouveau set().
• wait(timeout=None) : Bloque jusqu'à ce que le drapeau soit vrai. Si déjà vrai, retourne immédiatement. Sinon, attend qu'un autre thread appelle set() ou que le délai timeout (en secondes) expire.

Exemple : Simulation de feu de circulation

Deux threads : signal_state() alterne le feu (VERT/ROUGE), traffic_flow() attend le vert pour faire passer les véhicules.

from threading import Event, Thread
import time
terminate = False
def signal_state():
    global terminate
    while not terminate:
        time.sleep(0.5)
        print("Agent : FEU VERT")
        event.set()
        time.sleep(1)
        print("Agent : FEU ROUGE")
        event.clear()
def traffic_flow():
    global terminate
    num = 0
    while num < 10 and not terminate:
        print("En attente du feu VERT")
        event.wait()
        print("Feu VERT ... Circulation autorisée")
        while event.is_set() and not terminate:
            num += 1
            print("Véhicule n°", num, "passe")
            time.sleep(1)
        print("Feu ROUGE ... Attente")
event = Event()
t1 = Thread(target=signal_state)
t2 = Thread(target=traffic_flow)
t1.start()
t2.start()
time.sleep(5)
terminate = True
t1.join()
t2.join()
print("Fin du thread principal")

Sortie typique

En attente du feu VERT
Agent : FEU VERT
Feu VERT ... Circulation autorisée
Véhicule n° 1 passe
Agent : FEU ROUGE
Feu ROUGE ... Attente
En attente du feu VERT
Agent : FEU VERT
Feu VERT ... Circulation autorisée
Véhicule n° 2 passe
Véhicule n° 3 passe
Agent : FEU ROUGE
Feu ROUGE ... Attente
...

L'objet Condition

L'objet Condition offre un mécanisme de synchronisation plus avancé. Il permet aux threads d'attendre une notification d'un autre thread avant de continuer. Il est toujours associé à un verrou.

Syntaxe

threading.Condition(lock=None)

Méthodes clés

• acquire(*args) : Acquiert le verrou sous-jacent.
• release() : Libère le verrou.
• wait(timeout=None) : Libère le verrou et bloque jusqu'à notification (notify() ou notify_all()) ou timeout. Réacquiert le verrou au réveil.
• wait_for(predicate, timeout=None) : Appelle wait() jusqu'à ce que predicate soit vrai ou timeout.
• notify(n=1) : Réveille au plus n threads en attente.
• notify_all() : Réveille tous les threads en attente. Lève RuntimeError si le verrou n'est pas acquis.

Exemple 1 : Notification simple

from threading import Condition, Thread
import time
c = Condition()
def thread_a():
    print("Thread A démarré")
    with c:
        print("Thread A en attente...")
        c.wait()
        print("Thread A notifié !")
    print("Thread A terminé")
def thread_b():
    print("Thread B démarré")
    with c:
        time.sleep(2)
        print("Notification du Thread A...")
        c.notify()
    print("Thread B terminé")
Thread(target=thread_a).start()
Thread(target=thread_b).start()

Sortie :

Thread A démarré
Thread A en attente...
Thread B démarré
Notification du Thread A...
Thread B terminé
Thread A notifié !
Thread A terminé

Exemple 2 : Producteur-Consommateur

from threading import Condition, Thread
import time, random
numbers = []
def producteur(c):
    for _ in range(5):
        with c:
            num = random.randint(1, 10)
            print("Nombre généré :", num)
            numbers.append(num)
            print("Notification envoyée")
            c.notify()
        time.sleep(0.3)
def consommateur(c):
    for _ in range(5):
        with c:
            print("En attente de mise à jour")
            while not numbers:
                c.wait()
            print("Nombre récupéré :", numbers.pop())
        time.sleep(0.3)
c = Condition()
t1 = Thread(target=consommateur, args=(c,))
t2 = Thread(target=producteur, args=(c,))
t1.start()
t2.start()
t1.join()
t2.join()
print("Terminé")

Sortie typique :

En attente de mise à jour
Nombre généré : 2
Notification envoyée
Nombre récupéré : 2
Nombre généré : 5
Notification envoyée
En attente de mise à jour
Nombre récupéré : 5
...