Segunda Inmersión en Concurrencia (Actor)
En la entrega anterior hablamos sobre el uso de colas de mensajes para pasar mensajes entre hilos (o procesos) y así lograr comunicación entre ellos.
La Cola de Mensajes Compartida
Esto se basa en la premisa de que la concurrencia solo sirve para mejorar el rendimiento del sistema. Bajo esta suposición, cada worker realiza el mismo trabajo, por lo que pueden compartir una única cola de mensajes; no importa quién tome la siguiente instrucción.
¿Y si no somos iguales?
¿Qué pasa si necesitamos implementar un sistema donde cada worker tenga una función específica y trabaje de forma asíncrona (por ejemplo, nuestro teléfono IP)? Si seguimos usando el método anterior, donde todos los workers comparten una cola de mensajes, surge un problema: los mensajes no pueden enviarse a un worker específico.
¿Cómo solucionamos esto? Una solución simple y directa es añadir una etiqueta a cada mensaje, indicando a quién va dirigido:
queue = Queue([('Enviar_a_worker_A', 'hacer_algo1'),
('Enviar_a_worker_B', 'hacer_algo2'),
('Enviar_a_worker_B', 'hacer_algo3'),
('Enviar_a_worker_C', 'hacer_algo4'),
('Enviar_a_worker_A', 'hacer_algo5')])
Parece que el problema está resuelto, ¿verdad? Pero hemos introducido uno nuevo.
Si accidentalmente abro una carta para otro
Si un mensaje destinado al worker A es recibido por el worker B, este mensaje es inútil para B. Mientras tanto, el worker A, que debería haberlo recibido, no lo hace, por lo que el mensaje se pierde. Se desperdicia el tiempo del worker B, se malgasta el del worker A, e incluso podría introducir un error extraño en todo el sistema debido a este salto.
¿Cómo solucionamos este problema? Intuitivamente, añadir la siguiente estrategia a cada worker parece resolverlo: si recibe un mensaje para otro, lo vuelve a poner en la cola.
class Worker_B:
def run(self):
while True:
if not self.queue.empty():
data = self.queue.get()
if data['para_quien'] == 'worker_B':
hacer_algo_con_los_datos
else:
self.queue.put(data)
Parece que con un parche hemos solucionado el problema… pero en realidad hemos introducido otro nuevo.
Esto parece un callejón sin salida
Primero, una cola de mensajes es una cola, lo que implica FIFO (Primero en Entrar, Primero en Salir). Incluso si introducimos colas de mensajes con prioridad, esto no cambia en esencia. Así, un mensaje que debería ser el primero para el worker A, después de ser manipulado por el worker B, se convierte en el último en la cola para el worker A, lo que podría llevar al sistema a un error extraño.
Segundo, incluso si logramos que el mensaje mal entregado se vuelva a insertar al principio de la cola, si no introducimos un mecanismo de bloqueo (lock) para que la operación recibir-comprobar-devolver del worker B sea atómica, entonces mientras el worker B ejecuta la operación de devolución, otros workers podrían estar ya realizando operaciones de recepción de mensajes. Esto aún podría causar grandes problemas de sincronización. Introducir bloqueos resolvería los errores lógicos, pero haría que en cada ciclo de sondeo solo un worker pudiera trabajar realmente, mientras que los otros workers que reciben mensajes erróneos y ejecutan `recibir-comprobar-devolver* perderían su cuota de tiempo inútilmente, reduciendo la eficiencia.
Por lo tanto, para sistemas que introducen concurrencia para lograr trabajo asíncrono, que cada worker comparta la misma cola de mensajes no parece una buena solución.
La solución radical a este problema es crear una cola de mensajes (buzón) específica para cada grupo de workers que realizan la misma tarea, o incluso asignar a cada worker su propia cola de mensajes, renunciando por completo a la capacidad de memoria compartida entre hilos. Este último método es el protagonista de hoy: el Modelo de Actores (Patrón Actor).
El Modelo de Actores
Primero, veamos la definición y conceptos del Modelo de Actores según Wikipedia:
Conceptos
En ciencias de la computación, el modelo de actores (inglés: Actor model) es un modelo de computación concurrente. Un “actor” es una abstracción conceptual en programación, considerada la unidad básica de computación concurrente: cuando un actor recibe un mensaje, puede tomar decisiones, crear más actores, enviar más mensajes y determinar cómo responder a los siguientes mensajes. El modelo de actores fue propuesto en 1973 en un artículo de Carl Hewitt, Peter Bishop y Richard Steiger.
La filosofía que promueve el modelo de actores es “todo es un actor”, similar al “todo es un objeto” de la programación orientada a objetos, pero mientras la POO suele ser secuencial, el modelo de actores es paralelo. Un actor es una entidad computacional que responde a los mensajes recibidos y, de forma paralela, puede:
- Enviar un número finito de mensajes a otros actores.
- Crear un número finito de nuevos actores.
- Especificar el comportamiento al recibir el siguiente mensaje.
Las operaciones anteriores no asumen ejecución secuencial, por lo que pueden realizarse en paralelo. La desvinculación entre el emisor y el mensaje ya enviado es la ventaja fundamental del modelo de actores. Esto permite comunicación asíncrona y satisface las estructuras de control de paso de mensajes. Los receptores de mensajes se distinguen por direcciones, a veces llamadas “direcciones de correo”. Por lo tanto, un actor solo puede comunicarse con actores cuyas direcciones posee. Puede obtener direcciones a través de la información recibida o de las direcciones de los actores que crea. Las características del modelo de actores son: computación paralela dentro o entre actores, los actores pueden crearse dinámicamente, las direcciones de los actores se incluyen en los mensajes, la interacción solo se realiza a través de comunicación asíncrona directa de mensajes, y no se restringe el orden de llegada de los mensajes.
Implementación
Los Actores son lo suficientemente simples, nacidos para la concurrencia, y tienen una encapsulación lo suficientemente buena para aislar cambios (por ejemplo, sin importar si es multihilo o multiproceso).
Veamos una implementación típica de un Actor en Python usando hilos:
from queue import Queue
from threading import Thread, Event
class WorkerExit(Exception):
# Excepción para finalizar la tarea
pass
class BaseWorker(object):
# Actor
def __init__(self):
self._mailbox = Queue()
super(BaseWorker, self).__init__()
def send(self, msg):
self._mailbox.put(msg)
def recv(self):
msg = self._mailbox.get()
if msg is WorkerExit:
raise WorkerExit()
return msg
def close(self):
self.send(WorkerExit)
def start(self):
self._terminated = Event()
t = Thread(target=self._bootstrap)
t.daemon = True
t.start()
def _bootstrap(self):
try:
self.run()
except WorkerExit:
pass
finally:
self._terminated.set()
def join(self):
self._terminated.wait()
def run(self):
raise NotImplementedError
Explicándolo brevemente, este Actor (al que llamé BaseWorker, usado como clase base en proyectos reales) mantiene dos estructuras de datos: un Queue como buzón y un Event como mecanismo para bloquear el hilo principal.
Aquí, la operación central que sirve como interfaz externa es solo el método send(). Nótese que no restringimos el tipo de mensaje que se puede pasar, lo que otorga una flexibilidad enorme.
Dentro del Actor, usamos un hilo para ejecutar el método run() combinado con recv() para realizar el trabajo específico asignado. Vale la pena mencionar que configuramos una señal centinela, WorkerExit, para detener la tarea. Nótese que WorkerExit funciona lanzando una excepción cuando es identificada, y en el manejo de excepciones podríamos hacer incluso más cosas, aunque aquí simplemente detenemos la ejecución del hilo al capturar esta excepción. Esta implementación del manejo de excepciones depende del método _bootstrap() que envuelve a run().
Bajo la filosofía del Modelo de Actores, podemos extender este ejemplo simple para ir mucho más allá.
Conclusión
La programación concurrente requiere una forma de pensar más abstracta de lo habitual, y también puede ser una buena práctica del principio “alta cohesión, bajo acoplamiento”.
En mi exploración de este campo completamente nuevo para mí, el modelo de Actores ha sido ciertamente una luz guía, guiándome para entrar de verdad en este mundo, por eso lo comparto aquí. :)