A Second Look at Concurrency (Actor)
Last time, we discussed using message queues to pass messages between threads (or processes) to achieve inter-thread communication.
Shared Message Queue
This approach is based on the assumption that concurrency is solely for improving system throughput. Under this assumption, every worker performs the same task, so they can share a single message queue. It doesn’t matter which worker picks up the next instruction.
What if we are not siblings?
What if we need to implement a system where each worker has a distinct role and works asynchronously (e.g., our IP-phone)? Suppose we still use the method above, with all workers sharing a single message queue. A problem arises: messages cannot be sent to specific workers.
How can we solve this? A simple and straightforward solution is to add an identifier to each message, indicating its intended recipient:
queue = Queue([('Send_to_worker_A', 'do_something1'),
('Send_to_worker_B', 'do_something2'),
('Send_to_worker_B', 'do_something3'),
('Send_to_worker_C', 'do_something4'),
('Send_to_worker_A', 'do_something5')])
It seems the problem is solved, right? But we’ve introduced a new issue.
What if I accidentally open someone else’s letter?
If a message intended for worker A is received by worker B, it’s essentially useless to worker B. Meanwhile, worker A, who should have received it, does not, causing the message to be lost. This wastes worker B’s time, consumes worker A’s resources, and could even lead the entire system into some bizarre bug due to a misstep.
How do we solve this problem? Intuitively, adding the following strategy to each worker seems to fix it: if a worker receives a message meant for someone else, put it back into the message queue.
class Worker_B:
def run(self):
while True:
if not self.queue.empty():
data = self.queue.get()
if data['to_whom'] == 'worker_B':
do_something_with_data
else:
self.queue.put(data)
It appears we’ve patched the problem… but in reality, we’ve introduced yet another issue.
This seems like a dead end
First, a message queue is a queue, meaning FIFO (First-In, First-Out). Even if we introduce a priority queue, this fundamental characteristic doesn’t change. Therefore, a message that should have been the first delivered to worker A, after being handled by worker B, becomes the last in the queue for worker A. This could lead the system into a strange bug.
Second, even if we manage to reinsert the misdirected message back to the front of the queue, without introducing a locking mechanism to make worker B’s receive-check-return operation atomic, other workers might continue receiving messages while worker B is performing the return operation. This could still cause significant timing issues. Introducing a lock mechanism can solve logical errors, but it would mean only one worker can truly work during each polling cycle. Other workers that receive wrong messages and execute the receive-check-return operation would waste their time slices, leading to inefficiency.
Therefore, for a system that introduces concurrency for asynchronous work, having all workers share a single message queue doesn’t seem like a good approach.
The fundamental solution to this problem is to create a dedicated message queue (mailbox) for each group of workers performing the same task, or even assign a personal message queue to each worker, completely abandoning the ability to share memory between threads. The latter method is the protagonist of today’s discussion: the Actor Model (Participant Pattern).
Actor Model
First, let’s look at the definition and concepts of the Actor Model from Wikipedia:
Concepts
In computer science, the actor model is a mathematical model of concurrent computation. An “actor” is a fundamental unit of concurrent computation: upon receiving a message, an actor can make local decisions, create more actors, send more messages, and determine how to respond to the next message received. The actor model was proposed in a 1973 paper by Carl Hewitt, Peter Bishop, and Richard Steiger.
The philosophy promoted by the actor model is “everything is an actor,” similar to object-oriented programming’s “everything is an object.” However, object-oriented programming is typically sequential, while the actor model is inherently concurrent. An actor is a computational entity that responds to received messages and can concurrently:
- Send a finite number of messages to other actors;
- Create a finite number of new actors;
- Specify the behavior to be used for the next message it receives.
The above operations do not assume sequential execution and can therefore be performed in parallel. The decoupling of the sender from the sent message is a fundamental advantage of the actor model. This allows for asynchronous communication while satisfying the control structures of message passing. Message recipients are distinguished by addresses, sometimes called “mailing addresses.” Thus, an actor can only communicate with actors whose addresses it possesses. It can obtain addresses from received messages or from the actors it creates. Characteristics of the actor model include: parallel computation within or between actors, dynamic creation of actors, addresses included in messages, interaction only through direct asynchronous message passing, and no constraints on the order of message arrival.
Implementation
Actors are simple enough, born for concurrency, and have sufficient encapsulation to isolate changes (e.g., not caring whether it’s multi-threaded or multi-process).
Let’s look at a typical Python thread-based implementation of an Actor:
from queue import Queue
from threading import Thread, Event
class WorkerExit(Exception):
# Exception used to terminate the task
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
To explain simply, this Actor (which I named BaseWorker and use as a parent class in real projects) maintains two data structures: a Queue as the mailbox and an Event as a mechanism to block the main thread.
The core operation exposed as an external interface here is only the send() method. Note that we do not restrict the type of messages that can be passed, which offers immense flexibility.
Inside the Actor, we use a thread to run the run() method combined with the recv() method to perform the specified work. It’s worth mentioning that we set a sentinel signal, WorkerExit, to stop the task. Please note that WorkerExit works by being recognized and then raised as an exception. In the exception handling here, we could even do more things, but for now, we simply stop the thread when this exception is caught. The implementation of this exception handling relies on _bootstrap(), which wraps the run() method.
Under the philosophy of the Actor Model, we can extend this simple example to go much further.
Summary
Concurrent programming requires a more abstract way of thinking than usual and can serve as good practice for the principle of “high cohesion, low coupling.”
In my exploration of this entirely new field for me, the Actor Model has indeed been a guiding light, leading me to truly get started. Hence, I share it here. :)