OCaml と Eio で始める並行 RPC サーバー構築

はじめに：なぜ OCaml なのか

私は以前から関数型プログラミングに興味を持っていましたが、Haskell の複雑さに挫折した経験がありました。一方、OCaml は最近再び注目を集めています。

• 再興の要因: マルチコアサポートやエフェクト（Effect）の導入により進化。Jane Street が開発する「OxCaml」のようなシステム開発特化アプローチも登場。
• 並行処理フレームワーク Eio: 決定論的（deterministic）な設計を採用しており、テストが非常に容易な点に魅力を感じました。

学習リソースの選び方

まずは適切な教材を選ぶことが重要です。

『Real World OCaml』

• Jane Street が開発・維持する標準ライブラリ『Base』を使用している点が特徴です。
• 注意: 以前は標準ライブラリが貧弱でしたが、現在は徐々に成長しており、実社会での利用も正当化されています。
• 課題: 後半の GADT や一次階級モジュールといった高度な概念では、演習問題が少なく理解に苦しみました。

『CS3110』コース（推奨）

基礎を固めるにはこちらの教材が最適です。

• 形式: 動画とテキストの視聴覚教材を提供。
• 利点: 演習問題が充実しており、基礎概念の定着に役立ちます。
• 学習順序: まずは『CS3110』で基礎を学び、その後『Real World OCaml』へ移行するのがおすすめです。

『Real World Haskell』との比較

• プロジェクト指向や特定のライブラリ構成を求める人にとっては、『Real World OCaml』は失望する可能性があります。

コードを書く感覚：How It Feels In The Hand

OCaml でコードを書くのは、関数型と命令型プログラミングの良いバランスが取れているため非常に楽しいです。

再帰と高階関数の選択

リストを反復処理する際、再帰関数と

List.map

（および同様の高階関数）のどちらを使うかを選択できます。

let print_recursive list =
  let rec loop i list =
    match list with
    | [] -> ()
    | x :: xs ->
      Printf.printf "[%d] => %i";
      loop (i+1) xs
  in
  loop 0 list

let print_fn = List.iteri (fun i x -> Printf.printf "[%d] => %i\n" i x)

let () =
  print_recursive ["foo"; "bar"; "baz"];
  print_fn ["foo"; "bar"; "baz"]
(* 
[0] => foo
[1] => bar
[2] => baz
[0] => foo
[1] => bar
[2] => baz
*)

可変変数の活用

必要であれば、

mutable

キーワードで可変変数を利用することも可能です。

type client = { mutable request_id : int }

let dispatch cli ~msg =
  let request_id = cli.request_id in
  cli.request_id <- request_id + 1; (* リクエスト ID を更新 *)
  
  Printf.printf "Dispatching %d/%s to server\n" request_id msg

コンパイラーのエラー処理

OCaml のコンパイラーはエラー発生時にファイル全体を再構築する傾向があり、開発サイクル（記述→エラー修正→再構築）が少し重い場合があります。特に一つのモジュールに集中している場合、効率的な修正が困難になりがちです。

Eio で RPC サーバーを実装

基礎知識

Eio はエフェクトに基づく IO 並行処理ライブラリです。以下の用語を理解しておきましょう。

• Fiber: OS スレッドではない実行スレッド。
• Switch: ファイバーをグループ化したもの（Go の

  sync.WaitGroup

  に似る）。
• Flow: 読み書き可能なデータの流れ。
• Stream: 有界でスレッドセーフなキュー（Go のチャンネルに似る）。
• Promise: 非同期計算の結果を表す型。

サーバーの実装

サーバー側では、Flow を直接使うのではなく、

Buf_read

や

Buf_write

を使用してバッファ付きの読み書きを行います。これにより効率を向上させることができます。

パケット構造

単純なプロトコルとして、リクエスト ID（1 バイト）、名前（文字列）、引数（文字列）をシリアル化します。 例：リクエスト ID 10 で「Hello」を送る場合。

00000000: 0a04 4563 686f 0548 656c 6c6f            ..Echo.Hello

コード例

サーバーは接続を受け付け、フローを読み取ります。

open Utils

type conn = {
  src : Eio.Buf_read.t;
  sink : Eio.Buf_write.t;
}

let rec conn_loop conn =
  let request_id = Eio.Buf_read.uint8 conn.src in
  let procedure_name = read_string conn.src in
  let body = read_string conn.src in
  
  traceln "rx %d/%s(%s)" request_id procedure_name body;

  (* RPC ロジック: Echo や Capitalise など *)
  let res = match procedure_name with
  | "Echo" -> body
  | "Capitalise" -> String.capitalize_ascii body
  | _ -> failwith "unknown procedure"
  in

  traceln "tx %d/%s(%s) => %s" request_id procedure_name body res;
  
  Eio.Buf_write.uint8 conn.sink request_id;
  write_string conn.sink res;

  conn_loop conn

テスト実行

サーバーを起動し、nc（netcat）でテストできます。

Echo プロシージャ

> printf '\x00\x04Echo\x05Hello' | nc localhost 1470 -q0
Hello⏎

Capitalise プロシージャ

> printf '\x01\x0aCapitalise\x07matthew' | nc localhost 1470 -q0
Matthew⏎

クライアント実装：並行処理と安全な呼び出し

単一のプロシージャ呼出しのみを送信するように制御し、並行して動作させるクライアントを作成します。

リクエスト管理

書き込み用ファイバーでは、ストリームからタスクを読み取ります。

type request = { procedure : string; arg : string; }

type t = {
  mutable request_id : int;
  writeq : request Eio.Stream.t; (* リクエストのストリーム *)
}

let rec send_loop cli w =
  let req = Eio.Stream.take cli.writeq in
  let request_id = take_and_inc_request_id cli in
  write_request request;
  send_loop cli w

プロミスとレスポンス処理

サーバーからの応答を返すためには、Promise を使用します。Eio では Promise は

Promise.t

（待機側）と

Promise.u

（解決側）に分かれています。

• 送信時: リクエストに解決用の Resolver を設定し、ハッシュテーブルに格納。
• 受信時: 応答を読み取り、対応する Promise を解決。

type waiter = { resolver : string Eio.Promise.u; }

type request = {
  procedure : string;
  arg : string;
  resolver : string Eio.Promise.u;
}

type t = {
  mutable request_id : int;
  writeq : request Eio.Stream.t;
  waiters : (int, waiter) Hashtbl.t;
}

動作確認

複数のファイバーから並行して呼び出しを実行できます。

open Eio.Std

let port = 1470

let () =
  Eio_main.run @@ fun env ->
  Eio.Switch.run @@ fun sw ->
  let net = Eio.Stdenv.net env in

  let addr = `Tcp (Eio.Net.Ipaddr.V4.loopback, port) in
  let flow = Eio.Net.connect ~sw net addr in

  let cli = Dumb_rpc.Client.create () in
  
  (* サーバー側ファイバーをバックグラウンドで開始 *)
  Eio.Fiber.fork_daemon ~sw (fun () -> Dumb_rpc.Client.run cli ~flow; `Stop_daemon);
  
  (* クライアント側から並行して呼び出し *)
  for i = 0 to 5 do
    Eio.Fiber.fork ~sw (fun () ->
      let arg = Printf.sprintf "arg-%d" i in
      traceln "Invoking Echo(%s)" arg;
      let res = Dumb_rpc.Client.invoke cli ~procedure:"Echo" ~arg in
      traceln "Got response Echo(%s) => %s" arg res
    )
  done

出力結果の一部:

+Invoking Echo(arg-0)
+Invoking Echo(arg-1)
...
+tx 0/Echo(arg-0)
+rx 0 => arg-0
+Got response Echo(arg-0) => arg-0
...

結論

OCaml は、関数型でありながら命令型も可能という「美味しいスポット」を兼ね備え、高速なコンパイラーと強力なエラーハンドリングを提供しています。ユーザー数はまだ少ないですが、入門への道筋は明確になりつつあります。

Eio は Go を使った経験があれば馴染みやすいですが、非同期処理の概念に慣れない場合、Flow の閉じ方やファイバーの起動タイミングなどで躓くことがあります。近い将来、動作する Raft 合意形成アルゴリズムの実装についての記事に戻る予定です。