gazagnaire.org
Eio network backends and test helpers
eio-net: add the systhread vectored-write offload
Stream.offload_single_write performs one writev on a systhread through a
stub that releases the runtime lock, so the calling domain keeps running
fibers while the kernel copies the buffers into the socket. Semantics
match Eio.Flow.single_write, including short writes and waiting for
writability, so callers loop exactly as they do today.
author
Thomas Gazagnaire
date
(Jul 22, 2026, 10:47 PM UTC)
commit
15f05e98
15f05e9838ebfbb23508c60ded245a3195d1ad5f
parent
991cafd7
991cafd7f7f35887b77aa638098bdcef5849283e
+94
-1
+1
-1
lib/unix/dune
+1
-1
lib/unix/dune
+36
lib/unix/eio_net_unix.ml
+36
lib/unix/eio_net_unix.ml
···
71
71
in
72
72
{ base with Eio_net.Frame.recv_batch; send_batch }
73
73
end
74
+
75
+
module Stream = struct
76
+
external writev : Unix.file_descr -> (Cstruct.buffer * int * int) array -> int
77
+
= "caml_eio_net_writev"
78
+
79
+
let iov_max = 256
80
+
81
+
let offload_single_write fd bufs =
82
+
let bufs = List.filteri (fun i _ -> i < iov_max) bufs in
83
+
let iov =
84
+
Array.of_list
85
+
(List.map
86
+
(fun c -> (c.Cstruct.buffer, c.Cstruct.off, c.Cstruct.len))
87
+
bufs)
88
+
in
89
+
let rec attempt () =
90
+
let wrote =
91
+
Eio_unix.Fd.use_exn "Eio_net_unix.Stream.offload_single_write" fd
92
+
(fun unix_fd ->
93
+
Eio_unix.run_in_systhread ~label:"eio-net-writev" (fun () ->
94
+
writev unix_fd iov))
95
+
in
96
+
if wrote >= 0 then wrote
97
+
else begin
98
+
Eio_unix.Fd.use_exn "Eio_net_unix.Stream.offload_single_write" fd
99
+
Eio_unix.await_writable;
100
+
attempt ()
101
+
end
102
+
in
103
+
attempt ()
104
+
105
+
let offload_writer flow =
106
+
match Eio_unix.Resource.fd_opt flow with
107
+
| None -> None
108
+
| Some fd -> Some (offload_single_write fd)
109
+
end
+16
lib/unix/eio_net_unix.mli
+16
lib/unix/eio_net_unix.mli
···
20
20
4 MiB on a datagram descriptor. This is best-effort: a kernel may clamp
21
21
the request to its configured maximum. *)
22
22
end
23
+
24
+
module Stream : sig
25
+
val offload_single_write : Eio_unix.Fd.t -> Cstruct.t list -> int
26
+
(** [offload_single_write fd bufs] performs one vectored write on a systhread
27
+
through a writev stub that releases the runtime lock, so the calling
28
+
domain keeps running fibers while the kernel copies [bufs] into the socket
29
+
buffer. Semantics match {!Eio.Flow.single_write}: the call returns after
30
+
writing at least one byte, waiting for writability when the socket is
31
+
full, and may consume fewer bytes than offered -- including fewer buffers,
32
+
since one call passes at most 256 iovecs. *)
33
+
34
+
val offload_writer : _ Eio.Resource.t -> (Cstruct.t list -> int) option
35
+
(** [offload_writer flow] is [Some write] when [flow] exposes an OS file
36
+
descriptor, where [write] is {!offload_single_write} on it, and [None]
37
+
otherwise. *)
38
+
end
+41
lib/unix/writev_offload.c
+41
lib/unix/writev_offload.c
···
1
+
#include <caml/alloc.h>
2
+
#include <caml/bigarray.h>
3
+
#include <caml/fail.h>
4
+
#include <caml/memory.h>
5
+
#include <caml/mlvalues.h>
6
+
#include <caml/threads.h>
7
+
#include <caml/unixsupport.h>
8
+
#include <errno.h>
9
+
#include <sys/uio.h>
10
+
11
+
#define EIO_NET_IOV_MAX 256
12
+
13
+
/* Vectored write of cstruct (buffer, offset, length) triples with the
14
+
* runtime lock released, so a systhread caller leaves its domain free to
15
+
* run fibers while the kernel copies the batch into the socket buffer.
16
+
* Returns the byte count, or -1 when the socket would block. Bigarray data
17
+
* never moves, so the pointers stay valid across the blocking section. */
18
+
CAMLprim value caml_eio_net_writev(value v_fd, value v_bufs)
19
+
{
20
+
CAMLparam2(v_fd, v_bufs);
21
+
struct iovec iov[EIO_NET_IOV_MAX];
22
+
int count = (int)Wosize_val(v_bufs);
23
+
if (count <= 0 || count > EIO_NET_IOV_MAX)
24
+
caml_invalid_argument("caml_eio_net_writev: iovec count");
25
+
for (int i = 0; i < count; i++) {
26
+
value triple = Field(v_bufs, i);
27
+
iov[i].iov_base = (char *)Caml_ba_data_val(Field(triple, 0))
28
+
+ Long_val(Field(triple, 1));
29
+
iov[i].iov_len = (size_t)Long_val(Field(triple, 2));
30
+
}
31
+
int fd = Int_val(v_fd);
32
+
caml_release_runtime_system();
33
+
ssize_t wrote = writev(fd, iov, count);
34
+
int saved_errno = errno;
35
+
caml_acquire_runtime_system();
36
+
if (wrote >= 0)
37
+
CAMLreturn(Val_long(wrote));
38
+
if (saved_errno == EAGAIN || saved_errno == EWOULDBLOCK)
39
+
CAMLreturn(Val_long(-1));
40
+
caml_unix_error(saved_errno, "writev", Nothing);
41
+
}