並行処理(Actor)の再考
前回では、スレッド（プロセス）間でメッセージをやり取りするためにメッセージキューを利用し、それによってスレッド間通信を実現する方法について述べました。
共有メッセージキュー
これは、並行処理がシステムのスループット向上だけを目的としているという前提に基づいています。この仮定では、各ワーカーが行う作業は同じであるため、一つのメッセージキューを共有し、誰が次の命令を取得しても問題ありません。
もし私たちが同じ役割を持たないなら
例えば、各ワーカーがそれぞれの役割を果たし、非同期に動作するシステム（例えばIP電話）を実装する必要がある場合はどうすればよいでしょうか？上記の方法を引き続き使用し、すべてのワーカーが一つのメッセージキューを共有すると、問題が発生します：特定のワーカーにメッセージを送信することができません。
この問題をどう解決するでしょうか？単純で直接的な解決策は、各メッセージに識別子を付けて、そのメッセージの送信先を指定することです：
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')])
問題はこれで解決したように見えますね？しかし、新たな問題を導入してしまいました。
もし誤って他人宛の手紙を開封してしまったら
もしworker A宛のメッセージがworker Bに届いてしまった場合、彼にとってそのメッセージは実際には何の役にも立ちません。一方、本来そのメッセージを受け取るべきworker Aはそれを受け取れず、その結果メッセージは失われてしまいます。worker Bの時間を無駄にし、worker Aの時間を浪費するだけでなく、順序の飛び越しによってシステム全体が何らかの奇妙なバグに陥る可能性さえあります。
この問題はどう解決すればよいでしょうか？直感的には、各ワーカーに以下のような戦略を追加すれば問題は解決するように思えます：他人宛のメッセージを受け取ったら、そのメッセージをメッセージキューに戻す。
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)
一見、問題を修正したように見えます…しかし実際には、また新たな問題を導入してしまいました。
これは行き止まりのように見える
第一に、メッセージキューはキューであり、これはFIFO（先入れ先出し）を意味します。優先順位付きメッセージキューを導入したとしても、本質的にはこの点は変わりません。このため、本来最初にworker Aに渡されるべきメッセージが、worker Bによってかき回された結果、現在のメッセージキューの中でworker Aに渡されるのが最も遅いメッセージになってしまい、システムが奇妙なバグに陥る可能性があります。
第二に、たとえ誤って配信されたメッセージをメッセージキューの先頭に戻す方法を考えたとしても、誤ったメッセージを受け取ったworker Bの受信-確認-戻す操作をアトミック操作にするロック機構を導入しなければ、worker Bが戻す操作を実行している間に、他のワーカーがすでにメッセージ受信操作を続行している可能性があり、依然としてタイミングに関する大きな問題を引き起こす可能性があります。ロック機構を導入すれば論理的な誤りは解決できますが、これにより各ポーリングサイクルで実際に作業できるのは一つのワーカーだけになり、誤ったメッセージを受け取って受信-確認-戻す操作を実行する他のワーカーは、自分のタイムスライスを無駄にすることになり、効率が低下します。
したがって、非同期作業のために並行処理を導入するシステムにとっては、各ワーカーが同じメッセージキューを共有することは良い方法ではないようです。
この問題を根本的に解決する方法は、同じタスクを実行するワーカーのグループごとに専用のメッセージキュー（メールボックス）を設けること、あるいは各ワーカーに独自のメッセージキューを設け、スレッド間でのメモリ共有能力を完全に放棄することです。後者の方法が、今日お話しする主役である**Actorモデル（アクターモデル）**です。
Actorモデル
まず、WikipediaによるActorモデルの定義と概念を見てみましょう：
概念
コンピュータ科学において、アクターモデル（英語: Actor model）は並行計算におけるモデルの一つです。「アクター」はプログラム上の抽象概念であり、並行計算の基本単位と見なされます：アクターがメッセージを受信すると、いくつかの決定を行い、さらに多くのアクターを作成し、より多くのメッセージを送信し、次のメッセージにどのように応答するかを決定することができます。アクターモデルは、1973年にCarl Hewitt、Peter Bishop、Richard Steigerの論文で提唱されました。
アクターモデルが提唱する哲学は「すべてがアクターである」というもので、これはオブジェクト指向プログラミングの「すべてがオブジェクトである」という考えに似ていますが、オブジェクト指向プログラミングは通常逐次実行されるのに対し、アクターモデルは並列実行されます。アクターは計算実体であり、受信したメッセージに応答すると同時に、並列に以下のことができます：
- 他のアクターに有限数のメッセージを送信する；
- 有限数の新しいアクターを作成する；
- 次のメッセージを受信したときの動作を指定する。
上記の操作には逐次実行の前提が含まれていないため、並列に行うことができます。送信者と送信済みメッセージの分離は、アクターモデルの根本的な利点です。これにより、非同期通信が可能になり、メッセージパッシングの制御構造が満たされます。メッセージの受信者はアドレスによって識別され、これは「メールアドレス」とも呼ばれることがあります。したがって、アクターは自分がアドレスを持っているアクターとのみ通信できます。アクターは受信した情報からアドレスを取得したり、自分が作成したアクターのアドレスを取得したりすることができます。アクターモデルの特徴は、アクター内部またはアクター間で並列計算が行われ、アクターは動的に作成でき、アクターのアドレスはメッセージに含まれ、相互作用は直接的な非同期メッセージ通信のみを通じて行われ、メッセージの到着順序に制限がないことです。
実装
Actorは十分にシンプルで、並行処理のために生まれ、変化を隔離するのに十分なカプセル化を備えています（例：マルチスレッドかマルチプロセスかを気にしない）。
典型的なActorのPythonスレッド実装を見てみましょう：
from queue import Queue
from threading import Thread, Event
class WorkerExit(Exception):
# タスク終了用の例外
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
簡単に説明すると、このActor（私はBaseWorkerと名付け、実際のプロジェクトでは親クラスとして使用しています）は、2つのデータ構造を維持します：Queueをメールボックスとして、Eventをメインスレッドをブロックする終了手段として使用します。
ここでの外部インターフェースとなる核心的な操作はsend()メソッドだけです。注意してください、私たちは渡すことができるメッセージの型を制限しません。これは非常に高い柔軟性を意味します。
Actor内部では、一つのスレッドを使用してrun()メソッドとrecv()メソッドを組み合わせ、指定された具体的な作業を実行します。特筆すべきは、タスクを停止するための番兵信号WorkerExitを設定している点です。WorkerExitの動作原理は、それが識別された後に例外として送出されることです。この例外処理の部分では、さらに多くのことを行うこともできますが、ここではこの例外をキャッチしたときにスレッドの実行を停止するだけです。この例外処理の実装は、run()メソッドをラップする_bootstrap()メソッドに依存しています。
Actorモデルの哲学の下で、このシンプルな例を拡張し、さらに先へ進むことができます。
まとめ
並行プログラミングには、通常よりもさらに抽象的な思考が必要であり、「高凝集、低結合」という思想の良い実践例ともなり得ます。
私にとってこの全く新しい分野の探求において、Actorは確かに道を照らす灯台であり、私を本当の意味で入門させてくれました。そのため、ここで共有させていただきます。 :)