尝试 OCaml 与 Eio

我在此网站的第一篇博客是关于 Haskell 的。我一直对函数式编程感兴趣，但 Haskell 的复杂性似乎并不值得。

Lobsters · Jul 22, 2026

我在此网站的第一篇博客是关于 Haskell 的。我一直对函数式编程感兴趣，但 Haskell 的复杂性似乎并不值得。

我很早就听说过 OCaml，但从未涉足。最近，由于多核支持、 effects 以及 Jane Street 推出的更偏向系统的 OxCaml， OCaml 迎来了一波复兴。现在似乎是学习它的好时机，我对 Eio 也很感兴趣——它是一个并发框架。它使用了上述 effects，但更吸引我的是它旨在实现确定性，这可能有助于测试。

书籍与教程

首先要做的是选择一本教程或书籍。快速搜索一番后，我找到了 Real World OCaml。我假设它会和 Real World Haskell（我很喜欢，而且非常实用）类似，并且我已经具备足够的 FP 基础，于是便直接开始阅读。

我急于了解语法和语义，跳过了更具叙述性的序言。如果我读了序言，就会知道这本书选择使用 Jane Street 的 Base 标准库替代品。历史上，真正的标准库一直比较贫瘠，只覆盖编译器的需求。现在已不再如此，它在慢慢壮大。不过，对于一本名为 Real World OCaml 的书使用这门语言最大的实际用户提供的库，我并不觉得不合理。

无论如何，我肯定吸取了教训，要更仔细地阅读内容，于是回到开头重新开始。前半部分写得很好，我愉快地跟随着学习。到了 GADTs 和一等模块时，我开始感到困惑。我认为这本书缺少练习题，因此当遇到不熟悉的内容时，你无法充分掌握基础知识来帮助自己。

我去寻找更简单的内容，找到了 CS3110，它同时提供视频和文本以适应不同口味。它还提供练习题，这帮助很大。我建议人们先从这门课程开始，然后再转到 RWO 学习更高级的功能。我喜欢 RWO，不过如果你想要一本像 RWH 那样基于项目的书籍，你会感到失望。

上手感受

总体而言，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
*)

但如果你需要，你仍然可以使用可变变量等：

type client = { mutable request_id : int }

let dispatch cli ~msg =
  let request_id = cli.request_id in
  cli.request_id <- request_id + 1; (* update request_id *)
  
  Printf.printf "Dispatching %d/%s to server\n" request_id msg

let () =
  dispatch_cli ~msg:"Hello";
  dispatch_cli ~msg:"World"

(*
  => Dispatching 0/Hello to server
  => Dispatching 1/World to server
*)

我要说的是，语法感觉相当冗长，例如 let .. in、match .. with 等结构。我认为语法是编程语言中最不有趣的部分——即使在 Rust 还使用符号的年代，我也很喜欢——所以这并不太困扰我。拥有命名参数（使用 ~）非常不错。

编译器更让我烦恼。当它遇到错误时，似乎会放弃文件的其余部分。这使得迭代循环相当缓慢—— 编写代码，遇到错误，修复错误，重新构建。几乎没有批量修复错误的机会，尤其是当你专注于一个模块时。

Eio

Eio 是一个基于 effects 的 IO 并发库。它相对较新，但有一些有趣的功能，例如 io_uring 支持。我的最终目标是实现 Raft 共识算法并尝试测试它。

Raft 论文建议使用远程过程调用（RPC）在服务器之间进行通信。一个简单的 RPC 客户端/服务器似乎是一个很好的第一个项目。让我们设计一个非常简单的协议：一个过程调用有一个请求 ID（int）、一个名称（string）和一个参数（string）。整数用一个字节表示，字符串序列化为一个表示长度的字节，然后是数据。例如，要调用 Echo 过程，请求 ID 为 10，参数为 Hello，你将发送：

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

响应将只是请求 ID 和字符串返回值。

在开始之前，我们需要熟悉 Eio 用于其构建模块的一些术语：

Fiber - 一个执行线程。不是系统线程！
Switch - 将 fibers 分组的东西。有点像 Go 中的 sync.WaitGroup
Flow - 可以读取或写入的东西
Stream - 一个有界的、线程安全的队列。类似于 Go 的 channel
Promise - 我相信你知道这个。

从服务器开始，我们需要接受一个 Flow 并循环，从中读取过程调用，执行它们并发送响应：

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;

  (* In a real RPC system we might dispatch the invocation to a worker pool *)
  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

我们定义一个类型来保存连接，因为服务器可以有多个连接。请注意，这里我们没有直接使用 flow，而是使用 Buf_read 或 Buf_write 类型。这样我们就可以使用缓冲读/写，更加高效。Flow 上的操作是直接在底层资源（socket、文件等）上完成的。值得高兴的是，这些缓冲读写器有用于读/写类型的有用辅助函数，我们在这里使用它们来发送/接收请求 ID。

read_string/write_string 来自 Utils：

let read_string src =
  let len = Eio.Buf_read.uint8 src in
  Eio.Buf_read.take len src
  
let write_string sink s =
  Eio.Buf_write.uint8 sink (String.length s);
  Eio.Buf_write.string sink s

接下来，我们只需要一个辅助函数来获取缓冲流并运行循环：

let handle_conn flow =
  let src = Eio.Buf_read.of_flow ~max_size:1024 flow in
  Eio.Buf_write.with_flow flow @@ fun sink ->
  conn_loop {src; sink}

以及一个测试可执行文件：

open Dumb_rpc
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 handle_client flow addr =
    traceln "Accepted connection from %a" Eio.Net.Sockaddr.pp addr;
    Dumb_rpc.Server.handle_conn ~sw flow
  in

  let addr = `Tcp (Eio.Net.Ipaddr.V4.loopback, port) in
  let socket = Eio.Net.listen net ~sw ~reuse_addr:true ~backlog:5 addr in
  Eio.Net.run_server socket handle_client
    ~on_error:(traceln "Error handling connection: %a" Fmt.exn)

然后我们可以测试它：

> printf '\x00\x04Echo\x05Hello' | nc localhost 1470 -q0
Hello⏎
> printf '\x01\x0aCapitalise\x07matthew' | nc localhost 1470 -q0
Matthew⏎

不错！

客户端

对于客户端，让我们更花哨一点，使其可以安全地并发使用。为此，我们需要确保在任何时候都只向 socket 写入一个过程调用。这将通过两个 fibers 来完成：一个用于读取，一个用于写入。写入的 fiber 将从一个 stream 中获取其工作，当方法被调用时，我们将工作添加到该 stream 中：

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

type t = {
  mutable request_id : int;
  writeq : request Eio.Stream.t; (* Read right to left: stream of request *)
}

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

let invoke cli ~procedure ~arg =
  let req = {procedure; arg;} in
  Eio.Stream.add cli.writeq req;

但是我们如何将响应返回给调用者呢？为此，我们将在 stream 上发送一个 promise，并将其存储在哈希表中，键是请求 ID。当服务器响应时，可以稍后解析它。请注意，在 Eio 中，promise 实际上分为两半——等待的一方（Promise.t）和解析的一方（Promise.u）。

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;
}

let rec send_loop cli w =
  let req = Eio.Stream.take cli.writeq in
  let request_id = take_and_inc_request_id cli in
  Hashtbl.add cli.waiters request_id { resolver = req.resolver };
  
  traceln "tx %d/%s(%s)" request_id req.procedure req.arg;
  write_request request;

  send_loop cli w

let invoke cli ~procedure ~arg =
  let (promise, resolver) = Eio.Promise.create () in
  let req = {procedure; arg; resolver; } in
  Eio.Stream.add cli.writeq req;

  Eio.Promise.await promise

然后，接收循环可以只读取响应并解析 promise：

let rec recv_loop cli r =
  let (request_id, body) = read_response
  
  traceln "rx %d => %s" request_id body;
  let waiter = Hashtbl.find cli.waiters request_id in
  Promise.resolve waiter.resolver body;
  
  recv_loop cli r

我们可以从多个不同的 fibers 中使用它，如下所示：

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

可以看到它可以工作：

> ./_build/default/bin/client.exe
+Invoking Echo(arg-0)
+Invoking Echo(arg-1)
+Invoking Echo(arg-2)
+Invoking Echo(arg-3)
+Invoking Echo(arg-4)
+Invoking Echo(arg-5)
+tx 0/Echo(arg-0)
+tx 1/Echo(arg-1)
+tx 2/Echo(arg-2)
+tx 3/Echo(arg-3)
+rx 0 => arg-0
+Got response Echo(arg-0) => arg-0
+tx 4/Echo(arg-4)
+rx 1 => arg-1
+Got response Echo(arg-1) => arg-1
+rx 2 => arg-2
+tx 5/Echo(arg-5)
+Got response Echo(arg-2) => arg-2
+rx 3 => arg-3
+rx 4 => arg-4
+Got response Echo(arg-3) => arg-3
+Got response Echo(arg-4) => arg-4
+rx 5 => arg-5
+Got response Echo(arg-5) => arg-5

结论

总体而言，学习 OCaml 很有趣，使用起来也很容易。它处于多个甜蜜点——函数式编程，但你总能使用命令式；编译型，但编译器很快；可以在异常和值错误之间选择等等。它的用户群相当小，这一点显而易见。如何开始学习并不清楚，而且没有太多代码可以研究。

Eio 对任何使用过 Go 的人来说都会感到熟悉。在使用它时，我确实一直被 flow 被关闭的问题困扰，事实证明这是因为我应该阻塞时却启动了一个 fiber（就像在 run_server 的回调中）。

希望我很快就能带着一个可工作的 Raft 实现回来。

Read the original source

Lobsters

Publisher

0 Comments

No comments yet. Be the first to share your thoughts.