Taking OCaml and Eio for a spin

My first blog post on this website was about Haskell. I've always been interested in functional programming but the complexity of Haskell never seemed worth.

I have been vaguely aware of OCaml for a long time but never dipped my toes in. It has been having a bit of a renaissance recently with multicore support, effects and a more systems focussed spin called OxCaml from Jane Street. Now seems a great time to learn it and I was curious about Eio, a concurrency framework. It uses the aforementioned effects but more interesting to me is it is meant to be deterministic, which could be helpful for testing.

Books & Tutorials

The first thing to do is pick a tutorial or a book. A quick google search later I had found Real World OCaml. Assuming that it would be similar to Real World Haskell (which I enjoyed and was very hands on) and assuming I already had sufficient FP chops, I dived in.

Impatient to look at syntax and semantics I skipped over the more prosey Prologue. If I had read it I would have been told that the book chooses to use Jane Street's Base standard library replacement. It seems that historically the real standard library was quite anemic, only intended to cover the needs of the compiler. This is no longer true and it has slowly been growing. Still, it doesn't seem unreasonable to me that a book called Real World OCaml use the libraries from the biggest real world user of the language.

Anyway, I surely learnt my lesson to read things more carefully and went back to the beginning to start again. The first half of the book is well written and I happily followed along. I started to get lost around GADTs and first class modules though. I think the book lacks for exercises, so once something unfamilar comes up you don't have the full grasp of the basics to bail you out.

I went searching for something a bit easier and found CS3110 which has both videos and text to suit all tastes. It also had exercises which helped a lot. I would suggest to people that they start with this course and then graduate onto RWO for the more advanced features. I like RWO although if you are looking for a project based book like RWH, you'll be disappointed.

How It Feels In The Hand

Overall OCaml is very enjoyable to write code in. It has a nice mix of functional and imperative programming. For example, to loop through a list you are either using a recursive function or List.map and friends:

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
*)

but you still have mutable variables and the like if you need them:

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
*)

I will say that the syntax feels quite wordy, e.g. the let .. in, match .. with constructs. I think syntax is the least interesting part of a programming language to me - I was happy even back in the days where Rust had sigils - so it doesn't bother me too much. Having named arguments (with the ~) is very nice.

The compiler bothers me more. When it encounters an error it seems to give up on the rest of the file. This makes the iteration loop quite slow - write code, get an error, fix the error, rebuild. There's very little chance to fix errors in bulk, especially if you're focussed on one module.

Eio

Eio is an effects-based IO concurrency library. It's relatively new but it has some interesting features such as io_uring support. Ultimately my goal is to write an implementation of the Raft consensus algorithm and try test it.

The Raft paper says to use remote procedure calls (RPCs) to communicate between the servers. A dumb RPC client/server seems like a great first project. Let's come up with a very simple protocol: a procedure call has a request id (int), a name (string) and a single argument (string). Integers are a single byte and a string is serialised as a byte for the length and then the data. As an example, to call the Echo procedure with Hello as request 10, you would send:

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

A response will just be the request id and the string return value.

Before we start, we need to familiarise ourselves with some jargon Eio uses for its building blocks:

• Fiber - a thread of execution. Not a system thread though!
• Switch - something that groups fibers together. A bit like a sync.WaitGroup in Go
• Flow - something you can read or write to
• Stream - a bounded, thread safe queue. Like a Go channel
• Promise - I'm sure you know this one.

Starting with the server we will need to take a Flow and loop, reading procedure calls off it, invoking them and sending a response:

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

We define a type to hold a connection, as a server can have many of them. Note that we aren't using the flow directly here, instead we are using Buf_read or Buf_write types. That's so we can use buffered reading/writing and be more efficient. Operations on a Flow are done directly on the underlying resource (socket, file etc). Happily these buffered reader/writers have useful helper functions for reading/writing types which we use here to send/receive the request id.

read_string/write_string come from 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

Next we just need a helper to get the buffered flows and run the loop:

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}

and a test executable:

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)

and we're ready to test it:

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

Nice!

Client

For the client let's get a bit more fancy and make it safe to use concurrently. To do that we need to make sure that we are only writing a single procedure call to the socket at any one time. This will be done using two fibers: one for reading and one for writing. The writing one will take its work off a stream that we add to when a method is invoked:

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;

But how do we get the response back to the caller? To do that we will send a promise on our stream as well and store it away in a hash table, with the key being the request id. This can be resolved later when the server has responded. Note that in Eio promises are actually split into two halves - the side that waits (Promise.t) and the side that resolves (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

The receive loop can then just read responses and resolve promises:

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

We can use it like so, from multiple different 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

and see that it works:

> ./_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

Concluion

Overall OCaml has been fun to learn and easy to use. It sits at a number of sweet spots - FP but you can always do imperative, compiled but fast compiler, a choice between exceptions and errors as velues etc. It has quite a small userbase and I think it shows. It's not clear what is the best way to get started and there's not that much code out there to study.

Eio will feel familar to anyone who has used Go. Whilst hacking around with it I did keep getting tripped up with flows being closed which turned out to be because I was spinning up a fiber when I should have blocked (like in the callback to run_server).

Hopefully I'll be back soon with a working Raft implementation.