Make eio server/client wait for an event

Learning
1

Hi,

My attempt to create a reactive, event-driven server and client isn’t correct. I just try to loop recursively. My intention is to make the server and client react to a message and send an acknowledgement.
I plan to use timers in the future to send messages.

My simple attempt works once and ends with a broken socket.

open Eio.Std

(* Prefix all trace output with "client: " *)
let traceln fmt = traceln ("client: " ^^ fmt)

module Read = Eio.Buf_read
module Write = Eio.Buf_write

let rec loop flow clock  =
  Eio.Fiber.yield ();
  Write.with_flow flow @@ fun to_server ->
  Write.string to_server "Hello";
  Write.char to_server ' ';
  Write.string to_server "from client\n";
  let reply = Read.(parse_exn take_all) flow ~max_size:100 in
  traceln "Got reply %S" reply;
  Eio.Time.sleep clock 0.01;
  loop flow clock

let run ~net ~clock ~addr  =
Fiber.yield ();
  Eio.Time.sleep clock 0.01;
  traceln "Connecting to server at %a..." Eio.Net.Sockaddr.pp addr;
  Switch.run  @@ fun sw ->
  let flow = Eio.Net.connect ~sw net addr in
  loop flow clock

I am reusing the example from the eio documentations. The server also loops in the same way.
This doesn’t work as expected.

Thanks.

2

This reads until the other side closes the flow, after which you can’t read any more from it.

You can either:

  • Create a reader from the flow once, and then use it to parse multiple commands, or
  • Use a parser that executes commands as it reads them.

For an example of the first option, see Buffered Reading and Parsing.

For the second, you’d want something like:

let run ~net ~clock ~addr  =
  Eio.Time.sleep clock 0.01;
  traceln "Connecting to server at %a..." Eio.Net.Sockaddr.pp addr;
  Switch.run  @@ fun sw ->
  let flow = Eio.Net.connect ~sw net addr in
  Write.with_flow flow @@ fun to_server ->
  Read.parse_exn (loop clock to_server) flow ~max_size:100
3

Guess failed to mention that I was trying to keep both the server and client up without shutting down resources. I use Switch.run in the main.ml but if the server and clients loop indefinitely then it is not going to clean and shutdown. That is what I assumed.

But here the loop does not keep it up.

open Eio.Std

(* Prefix all trace output with "client: " *)
let traceln fmt = traceln ("client: " ^^ fmt)

module Read = Eio.Buf_read
module Write = Eio.Buf_write

let rec loop flow clock  =
  Eio.Fiber.yield ();
  Write.with_flow flow @@ fun to_server ->
   Write.string to_server "Hello";
   Write.char to_server ' ';
   Write.string to_server "from client\n";
    let reply = Read.parse_exn Read.line flow ~max_size:100 in
    traceln "Got reply %S" reply;
    Eio.Time.sleep clock 0.01;
  loop flow clock

let run ~net ~clock ~addr  =
  Fiber.yield ();
  Eio.Time.sleep clock 0.01;
  traceln "Connecting to server at %a..." Eio.Net.Sockaddr.pp addr;
  Switch.run  @@ fun sw ->
    let flow = Eio.Net.connect ~sw net addr in
    loop flow clock

Have I understood this wrongly ? This is the code that executes without errors. But the socket is closed somewhere.

Moreover I assumed I could parse the messages indefinitely without looping like the reactor pattern. Now I believe that I may have to code the abstraction on top of eio. Is that correct ?

I am trying not to recursively loop but react to message event by executing a handler. Something like Netty.(netty.io)

4

I’m not quite sure what you’re trying to do, but your code is creating one connection and then creating a new buffered reader on each iteration of the loop. That isn’t going to work. You need to create the reader (and, ideally, the writer) once per connection.

This should work better:

let rec loop clock to_server from_server  =
  Write.string to_server "Hello";
  Write.char to_server ' ';
  Write.string to_server "from client\n";
  let reply = Read.line from_server in
  traceln "Got reply %S" reply;
  Eio.Time.sleep clock 0.01;
  loop clock to_server from_server

let run ~net ~clock ~addr  =
  Eio.Time.sleep clock 0.01;
  traceln "Connecting to server at %a..." Eio.Net.Sockaddr.pp addr;
  Switch.run  @@ fun sw ->
  let flow = Eio.Net.connect ~sw net addr in
  Write.with_flow flow @@ fun to_server ->
  Read.parse_exn (loop clock to_server) flow ~max_size:100
5

Thanks. I fixed that and also my server code had to be fixed like this. A newline was required and I have forked a Fiber. But it worked.

open Eio.Std

(* Prefix all trace output with "server: " *)
let traceln fmt = traceln ("server: " ^^ fmt)

module Read = Eio.Buf_read
module Write = Eio.Buf_write

(* Read one line from [client] and respond with "OK". *)
let rec handle_client to_client from_client =
  traceln "Received: %S" (Read.line from_client);
  Write.string to_client "OK\n";
  Write.flush to_client;
  traceln "Written to client ";
  handle_client to_client from_client


let rec server_loop socket =

  Switch.run @@ fun sw ->
  let flow,addr = Eio.Net.accept ~sw socket in
  traceln "Accepted connection from %a" Eio.Net.Sockaddr.pp addr;
  Fiber.fork ~sw (fun () ->
    Write.with_flow flow @@ fun to_client ->
    let from_client = Read.of_flow flow ~max_size:100 in
    handle_client to_client from_client
  );
  server_loop socket 

(* Accept incoming client connections on [socket].
   We can handle multiple clients at the same time.
   Never returns (but can be cancelled). *)
let run socket =
 server_loop  socket

I was hoping to avoid the recursive calls because I thought eio implements the reactor pattern. and I just need to setup the socket but don’t need to look for messages or accept connections by looping and checking.

6

Eio.Net.run_server will run an accept loop for you.

If you don’t want handle_client to be recursive, you could just call it in a loop:

while true do handle_client to_client from_client done

Though it should probably be called handle_message in that case.