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

La communication inter-thread fait référence au processus permettant la communication et la synchronisation entre les threads au sein d'un programme multithread Python.

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 de thread.

Pour faciliter la communication entre les threads, le module de thread fournit diverses primitives de synchronisation telles que des objets Verrous, Événements, Conditions et Sémaphores. Dans ce didacticiel, vous apprendrez à utiliser l'objet Événement et Condition pour assurer la communication entre les threads dans un programme multithread.

L'objet événement

Un objet Event gère l'état d'un indicateur interne afin que les threads puissent attendre ou se définir. L'objet événement fournit des méthodes pour contrôler l'état de cet indicateur, permettant aux threads de synchroniser leurs activités en fonction de conditions partagées.

L'indicateur est initialement faux et devient vrai avec la méthode set() et réinitialisé à faux avec la méthode clear(). La méthode wait() se bloque jusqu'à ce que l'indicateur soit vrai.

Voici les méthodes clés de l'objet Event -

• is_set() :renvoie True si et seulement si l'indicateur interne est vrai.
• set() :définit l'indicateur interne sur true. Tous les fils qui attendent que cela se réalise sont réveillés. Les threads qui appellent wait() une fois que l'indicateur est vrai ne bloqueront pas du tout.
• clear() :Réinitialise l'indicateur interne à false. Par la suite, les threads appelant wait() se bloqueront jusqu'à ce que set() soit appelé pour définir à nouveau l'indicateur interne sur true.
• wait(timeout=None) :Bloquer jusqu'à ce que l'indicateur interne soit vrai. Si l'indicateur interne est vrai à l'entrée, revenez immédiatement. Sinon, bloquez jusqu'à ce qu'un autre thread appelle set() pour définir l'indicateur sur true, ou jusqu'à ce que le délai d'attente facultatif se produise. Lorsque l'argument timeout est présent et non None, il doit s'agir d'un nombre à virgule flottante spécifiant un délai d'attente pour l'opération en secondes.

Exemple

Le code suivant tente de simuler le flux de circulation contrôlé par l'état du feu de circulation VERT ou ROUGE.

Il y a deux threads dans le programme, ciblant deux fonctions différentes. La fonction signal_state() définit et réinitialise périodiquement l'événement indiquant le changement du signal du VERT au ROUGE.

La fonction traffic_flow() attend que l'événement soit défini et exécute une boucle jusqu'à ce qu'il reste défini.

from threading import Event, Thread
import time
terminate = False
def signal_state():
 global terminate
 while not terminate:
 time.sleep(0.5)
 print("Traffic Police Giving GREEN Signal")
 event.set()
 time.sleep(1)
 print("Traffic Police Giving RED Signal")
 event.clear()
def traffic_flow():
 global terminate
 num = 0
 while num < 10 and not terminate:
 print("Waiting for GREEN Signal")
 event.wait()
 print("GREEN Signal ... Traffic can move")
 while event.is_set() and not terminate:
 num += 1
 print("Vehicle No:", num," Crossing the Signal")
 time.sleep(1)
 print("RED Signal ... Traffic has to wait")
event = Event()
t1 = Thread(target=signal_state)
t2 = Thread(target=traffic_flow)
t1.start()
t2.start()
# Terminate the threads after some time
time.sleep(5)
terminate = True
# join all threads to complete
t1.join()
t2.join()
print("Exiting Main Thread")

Sortie

En exécutant le code ci-dessus, vous obtiendrez le résultat suivant -

Waiting for GREEN Signal
Traffic Police Giving GREEN Signal
GREEN Signal ... Traffic can move
Vehicle No: 1 Crossing the Signal
Traffic Police Giving RED Signal
RED Signal ... Traffic has to wait
Waiting for GREEN Signal
Traffic Police Giving GREEN Signal
GREEN Signal ... Traffic can move
Vehicle No: 2 Crossing the Signal
Vehicle No: 3 Crossing the Signal
Traffic Police Giving RED Signal
Traffic Police Giving GREEN Signal
Vehicle No: 4 Crossing the Signal
Traffic Police Giving RED Signal
RED Signal ... Traffic has to wait
Traffic Police Giving GREEN Signal
Traffic Police Giving RED Signal
Exiting Main Thread

L'objet Condition

L'objet Condition du module de threading de Python fournit un mécanisme de synchronisation plus avancé. Il permet aux threads d'attendre une notification d'un autre thread avant de continuer. L'objet Condition est toujours associé à un verrou et fournit des mécanismes de signalisation entre les threads.

Voici la syntaxe de la classe threading.Condition() -

threading.Condition(lock=None)

Vous trouverez ci-dessous les méthodes clés de l'objet Condition −

• acquire(*args) :Acquérir le verrou sous-jacent. Cette méthode appelle la méthode correspondante sur le verrou sous-jacent; la valeur de retour est celle renvoyée par cette méthode.
• release() :libère le verrou sous-jacent. Cette méthode appelle la méthode correspondante sur le verrou sous-jacent; il n'y a pas de valeur de retour.
• wait(timeout=None) :Cette méthode libère le verrou sous-jacent, puis se bloque jusqu'à ce qu'il soit réveillé par un appel notify() ou notify_all() pour la même variable de condition dans un autre thread, ou jusqu'à ce que le délai d'attente facultatif se produise. Une fois réveillé ou expiré, il réacquiert le verrou et revient.
• wait_for(predicate, timeout=None) :Cette méthode utilitaire peut appeler wait() à plusieurs reprises jusqu'à ce que le prédicat soit satisfait ou jusqu'à ce qu'un délai d'attente se produise. La valeur de retour est la dernière valeur de retour du prédicat et sera évaluée à False si la méthode a expiré.
• notify(n=1) :cette méthode réveille au plus n des threads en attente de la variable de condition ; ce n'est pas une opération si aucun thread n'attend.
• notify_all() :Réveille tous les threads en attente dans cette condition. Cette méthode agit comme notify(), mais réveille tous les threads en attente au lieu d'un. Si le thread appelant n'a pas acquis le verrou lorsque cette méthode est appelée, une RuntimeError est levée.

Exemple

Cet exemple montre une forme simple de communication inter-thread utilisant l'objet Condition du module de thread de Python. Ici, thread_a et thread_b sont communiqués à l'aide d'un objet Condition, le thread_a attend de recevoir une notification de thread_b. le thread_b dort pendant 2 secondes avant de notifier thread_a puis se termine.

from threading import Condition, Thread
import time
c = Condition()
def thread_a():
 print("Thread A started")
 with c:
 print("Thread A waiting for permission...")
 c.wait()
 print("Thread A got permission!")
 print("Thread A finished")
def thread_b():
 print("Thread B started")
 with c:
 time.sleep(2)
 print("Notifying Thread A...")
 c.notify()
 print("Thread B finished")
Thread(target=thread_a).start()
Thread(target=thread_b).start()

Sortie

En exécutant le code ci-dessus, vous obtiendrez le résultat suivant -

Thread A started
Thread A waiting for permission...
Thread B started
Notifying Thread A...
Thread B finished
Thread A got permission!
Thread A finished

Exemple

Voici un autre code démontrant comment l'objet Condition est utilisé pour assurer la communication entre les threads. En cela, le thread t2 exécute la fonction taskB() et le thread t1 exécute la fonction taskA(). Le thread t1 acquiert la condition et la notifie.

À ce moment-là, le thread t2 est dans un état d’attente. Une fois la condition libérée, le thread en attente consomme le nombre aléatoire généré par la fonction de notification.

from threading import Condition, Thread
import time
import random
numbers = []
def taskA(c):
 for _ in range(5):
 with c:
 num = random.randint(1, 10)
 print("Generated random number:", num)
 numbers.append(num)
 print("Notification issued")
 c.notify()
 time.sleep(0.3)
def taskB(c):
 for i in range(5):
 with c:
 print("waiting for update")
 while not numbers: 
 c.wait()
 print("Obtained random number", numbers.pop())
 time.sleep(0.3)
c = Condition()
t1 = Thread(target=taskB, args=(c,))
t2 = Thread(target=taskA, args=(c,))
t1.start()
t2.start()
t1.join()
t2.join()
print("Done")

Lorsque vous exécutez ce code, il produira le résultat suivant -

waiting for update
Generated random number: 2
Notification issued
Obtained random number 2
Generated random number: 5
Notification issued
waiting for update
Obtained random number 5
Generated random number: 1
Notification issued
waiting for update
Obtained random number 1
Generated random number: 9
Notification issued
waiting for update
Obtained random number 9
Generated random number: 2
Notification issued
waiting for update
Obtained random number 2
Done