アクター | Lambda Prelude

アクターは Lambda Prelude における常駐サービスの単位。OCaml 5 のエフェクトハンドラ上に直接実装されている — Eio・Miou・POSIX 専用 API には依存しない。同じアクターコードが Windows と Linux、インタプリタと AOT で動く。

3 つの通信プリミティブ

アクター表層は spawn / send / receive の 3 つの通信プリミティブと、自分自身への参照を返す Actor self から始まる。

Actor self                  "現在の fiber のアクター参照"
Actor spawn: [:me | body]   "新規 fiber を生成、body は自身の参照を受ける"
actor ! message             "fire-and-forget 送信"
actor receive               "メッセージが来るまでブロック"

最初の ping-pong は次のとおり。

parent := Actor self.

child := Actor spawn: [:me |
  msg := me receive.
  msg printNl.
  parent ! #pong
].

child ! #ping.
parent receive printNl.

状態はループ引数として生きる

共有可変状態はない。アクターの状態は再帰ループの引数として表現する。「状態を更新する」とは、新しい値で自分自身を呼び直すこと。

Object subclass: #Counter.

Counter class >> run: me count: n =
  me receive: {
    Inc      -> [:msg | Counter run: me count: n + 1].
    GetCount -> [:msg | msg reply ! n. Counter run: me count: n]
  }.

counter := Actor spawn: [:me | Counter run: me count: 0].

n は呼び出しフレームの中に閉じている — 他の fiber から読み書きされない。

型付き選択受信

me receive: { Class -> [:msg | ...] } はクラス・タグディスパッチ表。スケジューラがメールボックスを先頭から末尾へ走査し、最初に一致したハンドラを実行する。一致しないメッセージは元の位置に残る — これが選択受信。

Object subclass: #Inc.
Object subclass: #Get fields: (reply).

Counter class >> run: me state: n =
  me receive: {
    Inc -> [:msg | Counter run: me state: n + 1].
    Get -> [:msg | msg reply ! n. Counter run: me state: n]
  }.

ディスパッチ表は各メッセージクラスを Hindley-Milner 推論に通すので、ハンドラ本体は msg を具体型として扱える — Obj.magic も型を回避する抜け穴もない。

受信タイムアウト

receive:after:do: は一致したメッセージかタイムアウト分岐のどちらかを返す。

me receive: { Tick -> [:t | handle: t] }
   after: 1000
   do:    [Log warn: 'no tick for 1 s'].

Future と `ask:`

ask: は send + receive の上に乗った request / response の糖衣。応答が到着したときに解決する Future を返す。

fut := worker ask: (Compute fields: { x: 7 }).
fut await printNl.

Future の手動生成も用意してあるが、通常経路は ask:。

リンク / モニタ / 終了シグナル捕捉

失敗の意味論は Erlang の語彙をローカルモデルに当てはめたもの。

リンク — 対称。どちらか一方が異常終了すると他方も終了する。2 つのアクターのライフタイムを縛る。
モニタ — 非対称。観察対象が死ぬと観察側のメールボックスに Down メッセージが届くが、観察側は走り続ける。
終了シグナル捕捉 — リンク経由の死の連鎖を Exit メッセージに変換し、アクター本体がハンドラで受け取れるようにする。

Object subclass: #Boom.
Object subclass: #Worker.

Worker class >> loop: me =
  me receive: {
    Boom -> [:msg | Error raise: 'kaboom!']
  }.

main   := Actor self.
worker := Actor spawn: [:me | Worker loop: me].

main monitor: worker.
worker ! (Boom fields: {}).

main receive: {
  Down -> [:d |
    'Down received' printNl.
    d reason printNl
  ]
}.

'main still alive' printNl.

Down は死んだアクターの参照と失敗理由を運ぶ。観察側は走り続ける — リンクを貼っていない限り連鎖死は起きない。

Actor spawn: と、その後に続く monitor: は 2 つの別々のステップで、ワーカードメイン上では子がその間に走って死にうる。そのため既に死んだアクターへの登録は、登録を捨てるのではなくその死を報告する: monitor: は死亡時に記録された理由を載せた Down を返し、linkTo: は raise する (呼び出し側が終了シグナルを捕捉していれば Exit を届ける)。遅れて観察を始めた側も、早くから観察していた側と同じものを見る。

モニタもリンクも付いていないアクターの異常死は stderr に報告される。以前はこの死が痕跡を残さなかった — 理由は観察者のために記録されるが、観察者がいなければ捨てられていたので、エラーを踏んだワーカーはただ存在しなくなり、証拠は「行われなかった仕事」だけだった。--quiet は未マッチメッセージ警告と一緒にこの報告も止める。

スーパーバイザ

スーパーバイザツリーは専用構文ではなく、ふつうのクラスとして書く。スーパーバイザは子を監視し、Down メッセージを契機に再起動する。再起動予算を使い切った場合は停止し、上位スーパーバイザに監視されていればそこで扱われる。

stdlib の Supervision モジュールは再利用可能な戦略 (one-for-one / one-for-all / rest-for-one) と再起動ポリシーを提供するが、内部機構は monitor: + spawn: の組合せに過ぎない。

マルチドメインスケジューリング

ランタイムはマルチドメイン構成 — アクターは OS スレッドをまたいで spawn でき、クロスドメインの inbox 受け渡しとワークスティーリングが効く。AOT バイナリには同じランタイムが一緒にコンパイルされているので、まったく同一のマルチドメインスケジューラを備える。CPU バウンドな仕事はどちらのモードでもコア間に拡がり、別個の単一ドメインランタイムは存在しない。

境界つきソケット I/O

TCP connect:port:timeout:、TCPSocket >> readLine:timeout: / readBytes:timeout:、TLS wrapClient:host:timeout:、TLS の read / write はすべてアクタースケジューラと統合されている。遅い上流は呼び出し側 fiber を wait_readable / wait_writable で待機させるだけで、OS スレッドはブロックしない — 他のアクターは走り続ける。ホスト名解決もバックグラウンドの解決スレッドで走り、connect 自体も非ブロッキングなので、遅い DNS サーバも遅いハンドシェイクもドメインを止めない。

境界つき read には、UDPSocket >> receiveFrom: を含めた全ソケット共通の 1 つの規則がある: 正常な EOF では Maybe none を返し、呼び出し側が設定した期限を過ぎたら raise する。タイムアウトは Error try:onError: で捕まえるものであって、ifPresent:ifAbsent: を「不在値」としてすり抜けることはない。

TCP listen: は全インタフェースを、TCP listenHost: addr port: port は 1 つのローカルアドレスを束縛する。1 段上の HttpServer startHost:port:handler: / startGracefulHost:port:handler: も同じ。したがって 2 つのサーバが別アドレスで同じポートを持てるので、複数ノードをローカルで動かす時も本番と同じポート番号のまま、アドレスでノードを分けられる。

接続したソケットには全て TCP_NODELAY を設定する — accept、connect、Remote クライアント、HTTP クライアント、Redis プラグイン。レスポンスもステータス行 / 各ヘッダ / 空行 / ボディを別々に書くのではなく、バッファに組み立てて 1 回で書く。Nagle アルゴリズムは小さな書き込みを、直前の書き込みが相手に ack されるまで保留する。その ack を相手のスタックが 40 ms ほど遅延させるので、2 回に分けて出ていく往復は軒並み止まっていた — ここで話す request/response プロトコル (HTTP、JSON-RPC、TERIOS、grain) はどれもその形をしている。Linux での実測は、生の TCP 往復が 88 ms → 18 µs、Remote の RPC 呼び出しが 44 ms → 70 µs。Windows ではこの停止は起きないので、デプロイ先のプラットフォームを計測するまで残り続けていた。

TLS ブリッジは Tls.Engine を直接駆動する — ハンドシェイク / アプリケーション read / write がそれぞれランタイムの I/O wait プリミティブの上で待機する。

ロードマップから外しているもの

Erlang / Akka Cluster 形式の分散アクター — ノード間 spawn / link / monitor / スーパーバイザのワイヤ越し版 — はロードマップ外。ローカルアクターモデルはそのまま維持する。

複数ノード対応 — 仮想アクター (grain)

複数ノード方式は仮想アクター (grain) として実装済み。grain は fiber 参照ではなく安定した (クラス, id) の組で宛先指定されるアクターで、ライブ状態は固定したプロセスメモリではなく外部ストアに置く。ネットワーク上のオブジェクト参照ではなく (クラス, id) で到達するので、アイドル時は活性化を落とし（メモリを解放し）、次の呼び出しで外部ストアから状態を読み直して再活性化でき、GC すべきノード間参照グラフが存在しない。アプリケーションは grain クラスに通常のメソッドを定義し、その (クラス, id) 宛てにメッセージを送る — 分散ライフサイクルは stdlib/grain.lp の Grain モジュールが担う。

ランタイムは「(クラス, id) ごとに生きたオブジェクトは 1 つ」を 3 つの役割で保つ:

ディレクトリ (サーバごとに 1 つ) — id -> owner マップを持つ単一スレッドアクター。プロセス内で get-or-create が競合しない。
オーナ (アクティブな id ごとに 1 つ) — 状態をメモリに保持し、その id への全呼び出しをメールボックス経由で順序実行 (id 単位の直列化)。アイドルタイムアウトで非活性化し、メモリはアクティブな作業集合だけを追う。
リース (クロスノード) — id の活性化時に Redis リースを取り、TTL の間そのノードに固定する。状態書き込みは単調増加のフェンストークンを伴い、失効した旧オーナは新しいオーナが既に書いた状態を上書きできない。ノードが正常に落ちればリースが失効し、別ノードが保存状態から活性化する。

単一活性化は、プロセス内では無条件に成立する — ディレクトリの get-or-create は競合しない。プロセスを跨ぐ場合は 2 つの前提の下で成立する: 単一かつ線形化可能な Redis であること、オーナがリース TTL を越えて停止しないこと。フェンスは書き込みを順序付けるが、2 オーナの read-modify-write を直列化はしない。前提が崩れる状況 — 最新トークンを失う Redis のフェイルオーバー、旧オーナが生きたままのパーティション、リースより長い GC 停止 — では 2 オーナが一時的に併存し、更新が失われうる。そこからの復旧はアプリケーション側の仕事で、system-of-record への冪等な書き戻しがその手段になる。

単一ノードモード (for:initial:redis:) はディレクトリとオーナのみ、クロスノードモード (for:initial:redis:url:) がリースを足す。ルーティングは redirect ベースで、クライアント側に隠れている: id を所有しないノードはオーナの URL を返し、Remote プロキシがその redirect を透過的に追い、id ごとにオーナをキャッシュする — アプリケーションコードは入口 URL を 1 つ持ち続けるだけでよい。静的なメンバリスト (for:initial:redis:url:members:) を渡すと各 id は優先ノードへハッシュされるので、最初の接触が常に 1 つの入口ノードに集中しても活性化はクラスタ全体に散る。優先ノードに到達できない場合、プロキシは takeover フラグを立てて入口ノードへ再試行し、死んだノードのリースが失効した時点で入口ノードがその id を引き取る。Redis は揮発的な活性化ストアであって永続的な記録ではない — system-of-record DB への永続化はアプリケーションの責務。

ストアは差し替え可能。GrainServer は GrainStore プロトコル (stateFor: / persist:json:token: / acquireFor: / renew:token: / release:token:) 越しに話す。RedisGrainStore が組込バックエンドで、アプリケーションは独自ストア (例: アクター保持のインメモリマップ) を for:initial:store: で注入できる。

選択受信 / リンク / モニタはあくまでローカル限定 — ノードをまたいだリンク連鎖は導入しない。遠隔参照 Remote at: url for: #Class id: はリテラルのクラスを渡すと型に乗り、未知セレクタはコンパイル時エラーになる (非リテラルの for: は従来通り動的型付け)。

Remote の呼び出しは接続を再利用する。ソケットは呼び出しごとに張り直さず host:port ごとに保持し (1 エンドポイントあたり最大 8 本)、取り出した接続を相手が既に閉じていた場合は新しい接続で 1 度だけ再試行する。Nagle の停止に隠されていなかった Windows での実測では、同一エンドポイントへの逐次 JSON-RPC 呼び出しが 4543 µs (220 calls/sec) から 1411 ± 256 µs (709 calls/sec) になった。保持した接続に待機時間による追い出しは無いので、多数の宛先へ散発的に呼ぶ側はファイルディスクリプタを掴んだままになる。