Fix handling of very long IO vectors

`Flow.write` doesn't place any limit on how long the list of vectors can
be, but real operating systems do have limits and will fail if given too
many.

Also, Eio_posix was allocating the array on the stack, which could fail
if it was very large.

Author: talex5

lib_eio_linux/eio_linux.ml

@@ -73,14 +73,23 @@ let _fast_copy_try_splice src dst =
 
 (* XXX workaround for issue #319, PR #327 *)
 let fast_copy_try_splice src dst = fast_copy src dst
+    
+let rec list_take n = function
+  | [] -> []
+  | x :: xs ->
+    if n = 0 then []
+    else x :: list_take (n - 1) xs
+
+let truncate_to_iomax xs =
+  if List.compare_length_with xs Uring.iov_max <= 0 then xs
+  else list_take Uring.iov_max xs
 
 (* Copy using the [Read_source_buffer] optimisation.
    Avoids a copy if the source already has the data. *)
 let copy_with_rsb rsb dst =
+  let write xs = Low_level.writev_single dst (truncate_to_iomax xs) in
   try
-    while true do
-      rsb (Low_level.writev_single dst)
-    done
+    while true do rsb write done
   with End_of_file -> ()
 
 (* Copy by allocating a chunk from the pre-shared buffer and asking
@@ -161,11 +170,11 @@ module Flow = struct
     Low_level.readv ~file_offset t bufs
 
   let pwrite t ~file_offset bufs =
-    Low_level.writev_single ~file_offset t bufs
+    Low_level.writev_single ~file_offset t (truncate_to_iomax bufs)
 
   let read_methods = []
 
-  let single_write t bufs = Low_level.writev_single t bufs
+  let single_write t bufs = Low_level.writev_single t (truncate_to_iomax bufs)
 
   let copy t ~src =
     match Eio_unix.Resource.fd_opt src with

lib_eio_posix/eio_posix_stubs.c

@@ -25,6 +25,7 @@
 #include <caml/bigarray.h>
 #include <caml/socketaddr.h>
 #include <caml/custom.h>
+#include <caml/fail.h>
 
 #include "fork_action.h"
 
@@ -66,6 +67,27 @@ CAMLprim value caml_eio_posix_getrandom(value v_ba, value v_off, value v_len) {
   CAMLreturn(Val_long(ret));
 }
 
+/* Allocates an array of C iovecs using the cstructs in the array [v_bufs]. */
+static struct iovec *alloc_iov(value v_bufs) {
+  struct iovec *iov;
+  int n_bufs = Wosize_val(v_bufs);
+
+  if (n_bufs == 0) return NULL;
+  iov = calloc(n_bufs, sizeof(struct iovec));
+  if (iov == NULL)
+    caml_raise_out_of_memory();
+
+  for (int i = 0; i < n_bufs; i++) {
+    value v_cs = Field(v_bufs, i);
+    value v_ba = Field(v_cs, 0);
+    value v_off = Field(v_cs, 1);
+    value v_len = Field(v_cs, 2);
+    iov[i].iov_base = (uint8_t *)Caml_ba_data_val(v_ba) + Long_val(v_off);
+    iov[i].iov_len = Long_val(v_len);
+  }
+  return iov;
+}
+
 /* Fill [iov] with pointers to the cstructs in the array [v_bufs]. */
 static void fill_iov(struct iovec *iov, value v_bufs) {
   int n_bufs = Wosize_val(v_bufs);
@@ -97,11 +119,11 @@ CAMLprim value caml_eio_posix_writev(value v_fd, value v_bufs) {
   CAMLparam1(v_bufs);
   ssize_t r;
   int n_bufs = Wosize_val(v_bufs);
-  struct iovec iov[n_bufs];
-
-  fill_iov(iov, v_bufs);
+  struct iovec *iov;
 
+  iov = alloc_iov(v_bufs);
   r = writev(Int_val(v_fd), iov, n_bufs);
+  free(iov);
   if (r < 0) uerror("writev", Nothing);
 
   CAMLreturn(Val_long(r));
@@ -125,11 +147,11 @@ CAMLprim value caml_eio_posix_pwritev(value v_fd, value v_bufs, value v_offset)
   CAMLparam2(v_bufs, v_offset);
   ssize_t r;
   int n_bufs = Wosize_val(v_bufs);
-  struct iovec iov[n_bufs];
-
-  fill_iov(iov, v_bufs);
+  struct iovec *iov;
 
+  iov = alloc_iov(v_bufs);
   r = pwritev(Int_val(v_fd), iov, n_bufs, Int63_val(v_offset));
+  free(iov);
   if (r < 0) uerror("pwritev", Nothing);
 
   CAMLreturn(Val_long(r));

lib_eio_posix/flow.ml

@@ -27,6 +27,25 @@ let eio_of_stat x =
     ctime  = float_of_time (Low_level.ctime_sec x) (Low_level.ctime_nsec x);
   }
 
+let truncate_to_iomax xs =
+  let rec count i = function
+    | [] -> i
+    | _ when i = Config.iov_max -> Config.iov_max
+    | _ :: xs -> count (i + 1) xs
+  in
+  let len = count 0 xs in
+  let arr = Array.make len Cstruct.empty in
+  let rec fill i xs =
+    if i = len then arr
+    else (
+      match xs with
+      | x :: xs ->
+        Array.set arr i x;
+        fill (i + 1) xs
+      | [] -> assert false
+    ) in
+  fill 0 xs
+
 module Impl = struct
   type tag = [`Generic | `Unix]
 
@@ -41,7 +60,7 @@ module Impl = struct
 
   let single_write t bufs =
     try
-      Low_level.writev t (Array.of_list bufs)
+      Low_level.writev t (truncate_to_iomax bufs)
     with Unix.Unix_error (code, name, arg) ->
       raise (Err.wrap code name arg)
 
@@ -66,17 +85,17 @@ module Impl = struct
   let read_methods = []
 
   let pread t ~file_offset bufs =
-    let got = Low_level.preadv ~file_offset t (Array.of_list bufs) in
+    let got = Low_level.preadv ~file_offset t (truncate_to_iomax bufs) in
     if got = 0 then raise End_of_file
     else got
 
-  let pwrite t ~file_offset bufs = Low_level.pwritev ~file_offset t (Array.of_list bufs)
+  let pwrite t ~file_offset bufs = Low_level.pwritev ~file_offset t (truncate_to_iomax bufs)
 
   let send_msg t ~fds data =
-    Low_level.send_msg ~fds t (Array.of_list data)
+    Low_level.send_msg ~fds t (truncate_to_iomax data)
 
   let recv_msg_with_fds t ~sw ~max_fds data =
-    let _addr, n, fds = Low_level.recv_msg_with_fds t ~sw ~max_fds (Array.of_list data) in
+    let _addr, n, fds = Low_level.recv_msg_with_fds t ~sw ~max_fds (truncate_to_iomax data) in
     n, fds
 
   let seek = Low_level.lseek

lib_eio_posix/include/discover.ml

@@ -8,7 +8,7 @@ let optional_flags = [
 
 let () =
   C.main ~name:"discover" (fun c ->
-      let c_flags = ["-D_LARGEFILE64_SOURCE"] in
+      let c_flags = ["-D_LARGEFILE64_SOURCE"; "-D_XOPEN_SOURCE=700"; "-D_DARWIN_C_SOURCE"] in
       let includes = ["sys/types.h"; "sys/stat.h"; "fcntl.h"] in
       let extra_flags, missing_defs =
         C.C_define.import c ~c_flags ~includes
@@ -22,7 +22,7 @@ let () =
       in
       let present_defs =
         C.C_define.import c ~c_flags
-          ~includes:["sys/types.h"; "sys/stat.h"; "fcntl.h"]
+          ~includes:["sys/types.h"; "sys/stat.h"; "fcntl.h"; "limits.h"]
           C.C_define.Type.(extra_flags @ [
             "O_RDONLY", Int;
             "O_RDWR", Int;
@@ -41,6 +41,8 @@ let () =
 
             "AT_FDCWD", Int;
             "AT_SYMLINK_NOFOLLOW", Int;
+
+            "IOV_MAX", Int;
           ])
         |> List.map (function
             | name, C.C_define.Value.Int v when List.mem name optional_flags ->

tests/flow.md

@@ -173,3 +173,27 @@ Make sure we don't crash on SIGPIPE:
 +Connection_reset (good)
 - : unit = ()
 ```
+
+## IO_MAX
+
+Sending a very long vector over a flow should just send it in chunks, not fail:
+
+```ocaml
+# Eio_main.run @@ fun env ->
+  Switch.run @@ fun sw ->
+  let r, w = Eio_unix.pipe sw in
+  let a = Cstruct.of_string "abc" in
+  let vecs = List.init 10_000 (Fun.const a) in
+  Fiber.both
+    (fun () ->
+       Eio.Flow.write w vecs;
+       Eio.Flow.close w
+    )
+    (fun () ->
+       let got = Eio.Flow.read_all r in
+       traceln "Read %d bytes" (String.length got);
+       assert (got = Cstruct.to_string (Cstruct.concat vecs))
+    )
++Read 30000 bytes
+- : unit = ()
+```