Dominar la comunicación entre subprocesos en Python:sincronización e intercambio de datos

La comunicación entre subprocesos se refiere al proceso de habilitar la comunicación y sincronización entre subprocesos dentro de un programa multiproceso de Python.

Generalmente, los subprocesos en Python comparten el mismo espacio de memoria dentro de un proceso, lo que les permite intercambiar datos y coordinar sus actividades a través de variables compartidas, objetos y mecanismos de sincronización especializados proporcionados por el módulo de subprocesos.

Para facilitar la comunicación entre subprocesos, el módulo de subprocesos proporciona varias primitivas de sincronización, como objetos de bloqueos, eventos, condiciones y semáforos. En este tutorial aprenderá cómo utilizar el objeto Evento y Condición para proporcionar comunicación entre subprocesos en un programa multiproceso.

El objeto del evento

Un objeto Event gestiona el estado de un indicador interno para que los subprocesos puedan esperar o establecerse. El objeto de evento proporciona métodos para controlar el estado de este indicador, lo que permite que los subprocesos sincronicen sus actividades en función de condiciones compartidas.

La bandera es inicialmente falsa y se vuelve verdadera con el método set() y se restablece a falsa con el método clear(). El método wait() se bloquea hasta que la bandera sea verdadera.

Los siguientes son los métodos clave del objeto Evento:

• is_set():Devuelve Verdadero si y sólo si el indicador interno es verdadero.
• set():establece el indicador interno en verdadero. Se despiertan todos los hilos que esperan que esto se haga realidad. Los hilos que llaman a wait() una vez que el indicador es verdadero no se bloquearán en absoluto.
• clear():restablece el indicador interno a falso. Posteriormente, los subprocesos que llamen a wait() se bloquearán hasta que se llame a set() para establecer el indicador interno en verdadero nuevamente.
• esperar (tiempo de espera =Ninguno):bloquear hasta que el indicador interno sea verdadero. Si la bandera interna es verdadera al ingresar, regrese inmediatamente. De lo contrario, bloquee hasta que otro hilo llame a set() para establecer el indicador en verdadero, o hasta que se agote el tiempo de espera opcional. Cuando el argumento de tiempo de espera está presente y no es Ninguno, debe ser un número de coma flotante que especifique un tiempo de espera para la operación en segundos.

Ejemplo

El siguiente código intenta simular el flujo de tráfico controlado por el estado del semáforo, ya sea VERDE o ROJO.

Hay dos subprocesos en el programa que apuntan a dos funciones diferentes. La función signal_state() establece y restablece periódicamente el evento que indica el cambio de señal de VERDE a ROJO.

La función Traffic_flow() espera a que se establezca el evento y ejecuta un bucle hasta que permanece configurado.

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")

Salida

Al ejecutar el código anterior obtendrá el siguiente resultado:

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

El objeto de condición

El objeto Condición en el módulo de subprocesos de Python proporciona un mecanismo de sincronización más avanzado. Permite que los hilos esperen una notificación de otro hilo antes de continuar. El objeto Condición siempre está asociado con un candado y proporciona mecanismos para la señalización entre subprocesos.

La siguiente es la sintaxis de la clase threading.Condition() −

threading.Condition(lock=None)

A continuación se muestran los métodos clave del objeto Condición:

• adquirir(*args):adquiere el bloqueo subyacente. Este método llama al método correspondiente en el bloqueo subyacente; el valor de retorno es lo que devuelve ese método.
• release():libera el bloqueo subyacente. Este método llama al método correspondiente en el bloqueo subyacente; no hay valor de retorno.
• wait(timeout=None):este método libera el bloqueo subyacente y luego lo bloquea hasta que lo activa una llamada notify() o notify_all() para la misma variable de condición en otro subproceso, o hasta que se agote el tiempo de espera opcional. Una vez que se despierta o se agota el tiempo, vuelve a adquirir el bloqueo y regresa.
• wait_for(predicate, timeout=None):este método de utilidad puede llamar a wait() repetidamente hasta que se cumpla el predicado o hasta que se agote el tiempo de espera. El valor de retorno es el último valor de retorno del predicado y se evaluará como Falso si se agotó el tiempo de espera del método.
• notify(n=1):este método activa como máximo n de los subprocesos que esperan la variable de condición; no es una operación si no hay hilos en espera.
• notify_all():activa todos los subprocesos que esperan en esta condición. Este método actúa como notify(), pero activa todos los hilos en espera en lugar de uno. Si el subproceso que realiza la llamada no ha adquirido el bloqueo cuando se llama a este método, se genera un RuntimeError.

Ejemplo

Este ejemplo demuestra una forma simple de comunicación entre subprocesos utilizando el objeto Condición del módulo de subprocesos de Python. Aquí thread_a y thread_b se comunican mediante un objeto Condición, thread_a espera hasta recibir una notificación de thread_b. thread_b duerme durante 2 segundos antes de notificar a thread_a y luego finaliza.

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()

Salida

Al ejecutar el código anterior obtendrá el siguiente resultado:

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

Ejemplo

Aquí hay otro código que demuestra cómo se utiliza el objeto Condición para proporcionar comunicación entre subprocesos. En esto, el hilo t2 ejecuta la función taskB() y el hilo t1 ejecuta la función taskA(). El hilo t1 adquiere la condición y la notifica.

En ese momento, el hilo t2 está en estado de espera. Después de que se libera la condición, el hilo en espera procede a consumir el número aleatorio generado por la función de notificación.

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")

Cuando ejecute este código, producirá el siguiente resultado:

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