Repository: ocaml-multicore/picos
Pull request #75 (2024-04-03 23:47:17)

Add interrupt support and implement async IO for blocking FDs
This PR
- enhances `Picos_stdio` to handle `EINTR` errors by retrying the computation,
- adds support for "interrupts" into `Picos_select` that generate signals, and
- uses the interrupt support for transparently asynchronous IO with blocking FDs.



Patch:
diff --git a/lib/picos_select/dune b/lib/picos_select/dune
index 8fde796..639a8ad 100644
--- a/lib/picos_select/dune
+++ b/lib/picos_select/dune
@@ -8,6 +8,7 @@
   (re_export picos.fd)
   (re_export unix)
   picos.domain
+  picos.tls
   picos_thread_atomic
   threads.posix
   psq
diff --git a/lib/picos_select/picos_select.ml b/lib/picos_select/picos_select.ml
index c9810b7..ae92c8f 100644
--- a/lib/picos_select/picos_select.ml
+++ b/lib/picos_select/picos_select.ml
@@ -1,5 +1,10 @@
 open Picos
 
+let intr_sig = Sys.sigusr2
+let is_intr_sig s = s == intr_sig
+let intr_sigs = [ intr_sig ]
+let intr_pending = Atomic.make 0
+
 type cancel_at =
   | Cancel_at : {
       time : Mtime.span;
@@ -174,9 +179,19 @@ and select_thread_continue s rd wr ex (rd_fds, wr_fds, ex_fds) =
   let wr = process_fds wr_fds wr (Picos_thread_atomic.exchange s.new_wr []) in
   let ex = process_fds ex_fds ex (Picos_thread_atomic.exchange s.new_ex []) in
   let tos = process_timeouts s in
+  let tos =
+    let n = Atomic.get intr_pending in
+    if n = 0 then tos
+    else begin
+      Unix.kill (Unix.getpid ()) intr_sig;
+      let idle = 0.000_001 (* 1μs *) in
+      if tos < 0.0 || idle <= tos then idle else tos
+    end
+  in
   select_thread s tos rd wr ex
 
 let select_thread s =
+  if not Sys.win32 then Thread.sigmask SIG_BLOCK intr_sigs |> ignore;
   begin
     try
       let pipe_inn, pipe_out = Unix.pipe ~cloexec:true () in
@@ -283,3 +298,92 @@ let await_on file_descr op =
   with exn ->
     Computation.cancel computation exit_exn_bt;
     raise exn
+
+(* *)
+
+module Intr = struct
+  type intr_status = Cleared | Signaled
+
+  type _ tdt =
+    | Nothing : [> `Nothing ] tdt
+    | Req : {
+        state : state;
+        mutable computation : intr_status Computation.t;
+      }
+        -> [> `Req ] tdt
+
+  type t = T : [< `Nothing | `Req ] tdt -> t [@@unboxed]
+
+  let cleared =
+    let computation = Computation.create () in
+    Computation.return computation Cleared;
+    computation
+
+  let intr_key =
+    Picos_tls.new_key @@ fun () : [ `Req ] tdt ->
+    Req { state = get (); computation = cleared }
+
+  let handle _ =
+    let (Req r) = Picos_tls.get intr_key in
+    Computation.return r.computation Signaled
+
+  let intr_action trigger (Req r : [ `Req ] tdt) id =
+    match Computation.await r.computation with
+    | Cleared ->
+        (* No signal needs to be delivered. *)
+        remove_action trigger r.state id
+    | Signaled ->
+        (* Signal was delivered before timeout. *)
+        remove_action trigger r.state id;
+        if Atomic.fetch_and_add intr_pending 1 = 0 then begin
+          (* We need to make sure at least one select thread will keep on
+             triggering interrupts. *)
+          wakeup r.state `Alive
+        end
+    | exception Exit ->
+        (* The timeout was triggered.  This must have been called from the
+           select thread, which will soon trigger an interrupt. *)
+        Atomic.incr intr_pending
+
+  let () =
+    if not Sys.win32 then begin
+      let previously_blocked = Thread.sigmask SIG_BLOCK intr_sigs in
+      assert (not (List.exists is_intr_sig previously_blocked));
+      let old_behavior = Sys.signal intr_sig (Sys.Signal_handle handle) in
+      assert (old_behavior == Signal_default)
+    end
+
+  let nothing = T Nothing
+
+  let[@alert "-handler"] req ~seconds =
+    if Sys.win32 then
+      invalid_arg "Picos_select.Intr is not supported on Windows"
+    else begin
+      let time = to_deadline ~seconds in
+      let (Req r as req) = Picos_tls.get intr_key in
+      assert (not (Computation.is_running r.computation));
+      let id = next_id r.state in
+      let computation = Computation.with_action req id intr_action in
+      r.computation <- computation;
+      let entry = Cancel_at { time; exn_bt = exit_exn_bt; computation } in
+      add_timeout r.state id entry;
+      let was_blocked : int list = Thread.sigmask SIG_UNBLOCK intr_sigs in
+      assert (List.exists is_intr_sig was_blocked);
+      T req
+    end
+
+  let clr = function
+    | T Nothing -> ()
+    | T (Req r) ->
+        let was_blocked : int list = Thread.sigmask SIG_BLOCK intr_sigs in
+        assert (not (List.exists is_intr_sig was_blocked));
+        if not (Computation.try_return r.computation Cleared) then begin
+          while
+            let count = Atomic.get intr_pending in
+            count <= 0
+            || not (Atomic.compare_and_set intr_pending count (count - 1))
+          do
+            Thread.yield ()
+          done
+        end
+end
diff --git a/lib/picos_select/picos_select.mli b/lib/picos_select/picos_select.mli
index 8e4a5cb..9cb1998 100644
--- a/lib/picos_select/picos_select.mli
+++ b/lib/picos_select/picos_select.mli
@@ -31,3 +31,32 @@ val return_on : 'a Computation.t -> Picos_fd.t -> [ `R | `W | `E ] -> 'a -> unit
 
 val await_on : Picos_fd.t -> [ `R | `W | `E ] -> Picos_fd.t
 (** [await_on fd op] awaits until [fd] becomes available for [op]. *)
+
+module Intr : sig
+  (** A mechanism to interrupt blocking {!Unix} IO operations.
+
+      ⚠️ The mechanism uses the {!Sys.sigusr2} signal which should not be used
+      for other purposes at the same time. *)
+
+  type t
+  (** Represents an optional interrupt request. *)
+
+  val nothing : t
+  (** A constant for a no request.  {{!clr} [clr nothing]} does nothing. *)
+
+  val req : seconds:float -> t
+  (** [req ~seconds] requests an interrupt in the form of a signal delivered to
+      the thread that made the request within the specified number of [seconds].
+      Blocking {!Unix} IO calls typically raise an error with the {{!Unix.EINTR}
+      [Unix.EINTR]} error code when they are interrupted by a signal.
+
+      Regardless of whether the signal gets triggered or a system call gets
+      interrupted, the request must be {{!clr} cleared}.
+
+      ⚠️ Due to limitations of the OCaml system modules and unlike with typical
+      timeout mechanisms, the interrupt may also be triggered sooner. *)
+
+  val clr : t -> unit
+  (** [clr req] either cancels or acknowledges the interrupt request.  Every
+      {!req} must be cleared! *)
+end
diff --git a/lib/picos_stdio/dune b/lib/picos_stdio/dune
index 333c8ba..2cb2fc7 100644
--- a/lib/picos_stdio/dune
+++ b/lib/picos_stdio/dune
@@ -5,4 +5,5 @@
  (libraries
   (re_export unix)
   (re_export picos.fd)
+  picos.htbl
   picos.select))
diff --git a/lib/picos_stdio/picos_stdio.ml b/lib/picos_stdio/picos_stdio.ml
index ca26392..a46226c 100644
--- a/lib/picos_stdio/picos_stdio.ml
+++ b/lib/picos_stdio/picos_stdio.ml
@@ -1,67 +1,141 @@
+let nonblock_fds = Picos_htbl.create ~hashed_type:(module Picos_fd.Resource) ()
+
 module Unix = struct
   include Unix
 
   type file_descr = Picos_fd.t
 
   let is_oob flag = MSG_OOB == flag
+  let is_nonblock flag = O_NONBLOCK == flag
 
   (* The retry wrappers below are written to avoid closure allocations. *)
 
-  let progress_1 fd x1 fn op name =
-    try fn (Picos_fd.unsafe_get fd) x1
-    with
-    (* The documentation of [bind] and [connect] does not mention [EAGAIN] or
-       [EWOULDBLOCK], but on Windows we get those errors. *)
-    | Unix.Unix_error ((EAGAIN | EINPROGRESS | EWOULDBLOCK), _, _) ->
-      begin
-        let fd = Picos_select.await_on fd op in
-        match Unix.getsockopt_error (Picos_fd.unsafe_get fd) with
-        | None -> ()
-        | Some error -> raise (Unix.Unix_error (error, name, ""))
-      end
+  (* [EAGAIN] (and [EWOULDBLOCK]) indicates that the operation would have
+     blocked and so we await using the [select] thread for the file descriptor.
+
+     [EINTR] indicates that the operation was interrupted by a signal, which
+     usually shouldn't happen with non-blocking operations.  We don't want to
+     block, so we do the same thing as with [EAGAIN]. *)
+
+  let intr_req fd =
+    if Sys.win32 || Picos_htbl.mem nonblock_fds (Picos_fd.unsafe_get fd) then
+      Picos_select.Intr.nothing
+    else Picos_select.Intr.req ~seconds:0.000_001 (* 1μs *)
+
+  let intr_clr = Picos_select.Intr.clr
 
   let rec again_0 fd fn op =
-    try fn (Picos_fd.unsafe_get fd)
-    with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
-      again_0 (Picos_select.await_on fd op) fn op
+    let intr = intr_req fd in
+    match fn (Picos_fd.unsafe_get fd) with
+    | result ->
+        intr_clr intr;
+        result
+    | exception Unix.Unix_error ((EAGAIN | EINTR | EWOULDBLOCK), _, _) ->
+        intr_clr intr;
+        again_0 (Picos_select.await_on fd op) fn op
+    | exception exn ->
+        intr_clr intr;
+        raise exn
 
   let rec again_cloexec_0 ?cloexec fd fn op =
-    try fn ?cloexec (Picos_fd.unsafe_get fd)
-    with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
-      again_cloexec_0 ?cloexec (Picos_select.await_on fd op) fn op
+    let intr = intr_req fd in
+    match fn ?cloexec (Picos_fd.unsafe_get fd) with
+    | result ->
+        intr_clr intr;
+        result
+    | exception Unix.Unix_error ((EAGAIN | EINTR | EWOULDBLOCK), _, _) ->
+        intr_clr intr;
+        again_cloexec_0 ?cloexec (Picos_select.await_on fd op) fn op
+    | exception exn ->
+        intr_clr intr;
+        raise exn
 
   let rec again_3 fd x1 x2 x3 fn op =
-    try fn (Picos_fd.unsafe_get fd) x1 x2 x3
-    with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
-      again_3 (Picos_select.await_on fd op) x1 x2 x3 fn op
+    let intr = intr_req fd in
+    match fn (Picos_fd.unsafe_get fd) x1 x2 x3 with
+    | result ->
+        intr_clr intr;
+        result
+    | exception Unix.Unix_error ((EAGAIN | EINTR | EWOULDBLOCK), _, _) ->
+        intr_clr intr;
+        again_3 (Picos_select.await_on fd op) x1 x2 x3 fn op
+    | exception exn ->
+        intr_clr intr;
+        raise exn
 
   let rec again_4 fd x1 x2 x3 x4 fn op =
-    try fn (Picos_fd.unsafe_get fd) x1 x2 x3 x4
-    with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
-      again_4 (Picos_select.await_on fd op) x1 x2 x3 x4 fn op
+    let intr = intr_req fd in
+    match fn (Picos_fd.unsafe_get fd) x1 x2 x3 x4 with
+    | result ->
+        intr_clr intr;
+        result
+    | exception Unix.Unix_error ((EAGAIN | EINTR | EWOULDBLOCK), _, _) ->
+        intr_clr intr;
+        again_4 (Picos_select.await_on fd op) x1 x2 x3 x4 fn op
+    | exception exn ->
+        intr_clr intr;
+        raise exn
 
   let rec again_5 fd x1 x2 x3 x4 x5 fn op =
-    try fn (Picos_fd.unsafe_get fd) x1 x2 x3 x4 x5
-    with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) ->
-      again_5 (Picos_select.await_on fd op) x1 x2 x3 x4 x5 fn op
-
-  let rcn ?dispose fd =
-    let fd = Picos_fd.create ?dispose fd in
-    match Unix.set_nonblock (Picos_fd.unsafe_get fd) with
-    | () -> fd
-    | exception Unix.Unix_error (Unix.ENOTSOCK, _, _) when Sys.win32 ->
-        (* TODO: Non-blocking is supported only on sockets on Windows. *)
-        fd
-
-  let stdin = rcn ~dispose:false Unix.stdin
-  and stdout = rcn ~dispose:false Unix.stdout
-  and stderr = rcn ~dispose:false Unix.stderr
+    let intr = intr_req fd in
+    match fn (Picos_fd.unsafe_get fd) x1 x2 x3 x4 x5 with
+    | result ->
+        intr_clr intr;
+        result
+    | exception Unix.Unix_error ((EAGAIN | EINTR | EWOULDBLOCK), _, _) ->
+        intr_clr intr;
+        again_5 (Picos_select.await_on fd op) x1 x2 x3 x4 x5 fn op
+    | exception exn ->
+        intr_clr intr;
+        raise exn
+
+  (* [EINPROGRESS] indicates that a socket operation is being performed
+     asynchronously.  We await using the [select] thread for the operation to
+     complete and then get the error from the socket. *)
+
+  let progress_1 fd x1 fn op name =
+    let intr = intr_req fd in
+    match fn (Picos_fd.unsafe_get fd) x1 with
+    | () -> intr_clr intr
+    | exception
+        Unix.Unix_error ((EAGAIN | EINPROGRESS | EINTR | EWOULDBLOCK), _, _) ->
+      begin
+        (* The documentation of [bind] and [connect] does not mention [EAGAIN]
+           (or [EWOULDBLOCK]), but on Windows we do seem to get those errors
+           from [connect].
+
+           The documentation of [bind] does not mention [EINTR].  Matching on
+           that shouldn't cause issues with [bind].
+
+           For [connect] both [EINPROGRESS] and [EINTR] mean that connection
+           will be established asynchronously and we use [select] to wait. *)
+        intr_clr intr;
+        let fd = Picos_select.await_on fd op in
+        match Unix.getsockopt_error (Picos_fd.unsafe_get fd) with
+        | None -> ()
+        | Some error -> raise (Unix.Unix_error (error, name, ""))
+      end
+    | exception exn ->
+        intr_clr intr;
+        raise exn
+
+  let stdin = Picos_fd.create ~dispose:false Unix.stdin
+  and stdout = Picos_fd.create ~dispose:false Unix.stdout
+  and stderr = Picos_fd.create ~dispose:false Unix.stderr
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/open.html *)
-  let openfile path flags file_perm = rcn (Unix.openfile path flags file_perm)
+  let openfile path flags file_perm =
+    let fd = Unix.openfile path flags file_perm in
+    if List.exists is_nonblock flags then
+      Picos_htbl.try_add nonblock_fds fd () |> ignore;
+    Picos_fd.create fd
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/close.html *)
-  let close fd = Picos_fd.decr ~close:true fd
+  let close fd =
+    let _ : bool =
+      Picos_htbl.try_remove nonblock_fds (Picos_fd.unsafe_get fd)
+    in
+    Picos_fd.decr ~close:true fd
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/fsync.html *)
   let fsync fd = again_0 fd Unix.fsync `W
@@ -141,14 +215,20 @@ module Unix = struct
   (* *)
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/dup.html *)
-  let dup ?cloexec fd = rcn (Unix.dup ?cloexec (Picos_fd.unsafe_get fd))
+  let dup ?cloexec fd =
+    Picos_fd.create (Unix.dup ?cloexec (Picos_fd.unsafe_get fd))
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/dup.html *)
   let dup2 ?cloexec src dst =
     Unix.dup2 ?cloexec (Picos_fd.unsafe_get src) (Picos_fd.unsafe_get dst)
 
-  (*let set_nonblock _ = failwith "TODO"*)
-  (*let clear_nonblock _ = failwith "TODO"*)
+  let set_nonblock fd =
+    Unix.set_nonblock (Picos_fd.unsafe_get fd);
+    Picos_htbl.try_add nonblock_fds (Picos_fd.unsafe_get fd) () |> ignore
+
+  let clear_nonblock fd =
+    Unix.clear_nonblock (Picos_fd.unsafe_get fd);
+    Picos_htbl.try_remove nonblock_fds (Picos_fd.unsafe_get fd) |> ignore
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/fcntl.html *)
   let set_close_on_exec fd = Unix.set_close_on_exec (Picos_fd.unsafe_get fd)
@@ -161,7 +241,7 @@ module Unix = struct
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/pipe.html *)
   let pipe ?cloexec () =
     let inn, out = Unix.pipe ?cloexec () in
-    (rcn inn, rcn out)
+    (Picos_fd.create inn, Picos_fd.create out)
 
   (* *)
 
@@ -208,17 +288,17 @@ module Unix = struct
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/socket.html *)
   let socket ?cloexec socket_domain socket_type protocol =
-    rcn (Unix.socket ?cloexec socket_domain socket_type protocol)
+    Picos_fd.create (Unix.socket ?cloexec socket_domain socket_type protocol)
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/socketpair.html *)
   let socketpair ?cloexec socket_domain socket_type mystery =
     let fst, snd = Unix.socketpair ?cloexec socket_domain socket_type mystery in
-    (rcn fst, rcn snd)
+    (Picos_fd.create fst, Picos_fd.create snd)
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/accept.html *)
   let accept ?cloexec fd =
     let fd, sockaddr = again_cloexec_0 ?cloexec fd Unix.accept `R in
-    (rcn fd, sockaddr)
+    (Picos_fd.create fd, sockaddr)
 
   (* https://pubs.opengroup.org/onlinepubs/9699919799/functions/bind.html *)
   let bind fd sockaddr = progress_1 fd sockaddr Unix.bind `W "bind"
diff --git a/lib/picos_stdio/picos_stdio.mli b/lib/picos_stdio/picos_stdio.mli
index ed74f08..cdafc62 100644
--- a/lib/picos_stdio/picos_stdio.mli
+++ b/lib/picos_stdio/picos_stdio.mli
@@ -1,27 +1,22 @@
-(** Basic IO facilities based on OCaml standard libraries for {!Picos}.
-
-    🚧 This library is currently considered experimental and unfinished.  Please
-    report any problems and issues you might encounter. *)
+(** Basic IO facilities based on OCaml standard libraries for {!Picos}. *)
 
 module Unix : sig
   (** A transparently asynchronous replacement for a subset of the {{!Deps.Unix}
       [Unix]} module that comes with OCaml.
 
-      All the operations in this module that return file descriptors implicitly
-      set the {{!file_descr} file descriptors} to {{!Deps.Unix.set_nonblock}
-      non-blocking mode} and operations on file descriptors, such as {!read} and
+      In this module operations on file descriptors, such as {!read} and
       {!write} and others, implicitly block, in a scheduler friendly manner, to
-      await for the file descriptor to become available for the operation.
+      await for the file descriptor to become available for the operation.  This
+      works best with file descriptors {{!set_nonblock} set to non-blocking mode}.
 
       ⚠️ Beware that this does not currently try to work around any limitations
-      of the {{!Deps.Unix} [Unix]} module that comes with OCaml.  For example,
-      on Windows, only sockets can be put into non-blocking mode.
+      of the {{!Deps.Unix} [Unix]} module that comes with OCaml.  In particular,
+      on Windows, only sockets can be put into non-blocking mode.  Also, on
+      Windows, scheduler friendly blocking only works properly with non-blocking
+      file descriptors, i.e. sockets.
 
       Please consult the documentation of the {{!Deps.Unix} [Unix]} module that
-      comes with OCaml.
-
-      🚧 This module is currently considered experimental and unfinished.
-      Please report any problems and issues you might encounter. *)
+      comes with OCaml. *)
 
   type file_descr = Picos_fd.t
   (** Opaque type alias for {{!Deps.Unix.file_descr} [Unix.file_descr]}.
@@ -259,10 +254,8 @@ module Unix : sig
   val access : string -> access_permission list -> unit
   val dup : ?cloexec:bool -> file_descr -> file_descr
   val dup2 : ?cloexec:bool -> file_descr -> file_descr -> unit
-
-  (*val set_nonblock : file_descr -> unit*)
-  (*val clear_nonblock : file_descr -> unit*)
-
+  val set_nonblock : file_descr -> unit
+  val clear_nonblock : file_descr -> unit
   val set_close_on_exec : file_descr -> unit
   val clear_close_on_exec : file_descr -> unit
   val mkdir : string -> file_perm -> unit


Test patch:
diff --git a/test/dune b/test/dune
index 65198fe..8470909 100644
--- a/test/dune
+++ b/test/dune
@@ -55,6 +55,13 @@
 
 ;;
 
+(test
+ (name test_select)
+ (modules test_select)
+ (libraries picos.threaded picos.select alcotest domain_shims))
+
+;;
+
 (test
  (name test_mpsc_queue)
  (package picos)
diff --git a/test/test_select.ml b/test/test_select.ml
new file mode 100644
index 0000000..bc01d91
--- /dev/null
+++ b/test/test_select.ml
@@ -0,0 +1,37 @@
+open Picos
+
+let test_intr () =
+  let inn, out = Unix.pipe ~cloexec:true () in
+  let main () =
+    Picos_threaded.run ~forbid:false @@ fun () ->
+    let computation = Computation.create () in
+    let n_threads = 10 in
+    let n = Atomic.make n_threads in
+    for _ = 1 to n_threads do
+      let main () =
+        for _ = 1 to 1_000 do
+          let req = Picos_select.Intr.req ~seconds:0.000_001 in
+          match Unix.read inn (Bytes.create 1) 0 1 with
+          | _ -> assert false
+          | exception Unix.Unix_error (EINTR, _, _) -> Picos_select.Intr.clr req
+        done;
+        if 1 = Atomic.fetch_and_add n (-1) then Computation.finish computation
+      in
+      Fiber.spawn ~forbid:false computation [ main ]
+    done;
+    Computation.wait computation
+  in
+  let domains = Array.init 3 @@ fun _ -> Domain.spawn main in
+  let finally () =
+    Array.iter Domain.join domains;
+    Unix.close inn;
+    Unix.close out
+  in
+  Fun.protect ~finally main
+
+let () =
+  [
+    ( "Intr",
+      if Sys.win32 then [] else [ Alcotest.test_case "" `Quick test_intr ] );
+  ]
+  |> Alcotest.run "Picos_select"
diff --git a/test/test_server_and_client.ml b/test/test_server_and_client.ml
index 1ce9494..f3e5467 100644
--- a/test/test_server_and_client.ml
+++ b/test/test_server_and_client.ml
@@ -2,6 +2,20 @@ open Foundation.Finally
 open Elements
 open Picos_stdio
 
+let is_ocaml4 = String.starts_with ~prefix:"4." Sys.ocaml_version
+
+let use_nonblock =
+  Sys.win32
+  || begin
+       Random.self_init ();
+       Random.bool ()
+     end
+
+let set_nonblock fd =
+  if use_nonblock then
+    try Unix.set_nonblock fd
+    with _exn -> Printf.printf "  (Failed to set_nonblock)\n%!"
+
 let main () =
   Bundle.run @@ fun bundle ->
   let n = 100 in
@@ -16,12 +30,14 @@ let main () =
         finally Unix.close @@ fun () ->
         Unix.socket ~cloexec:true PF_INET SOCK_STREAM 0
       in
+      set_nonblock socket;
       Unix.bind socket Unix.(ADDR_INET (inet_addr_loopback, 0));
       server_addr := Some (Unix.getsockname socket);
       Unix.listen socket 1;
       Printf.printf "  Server listening\n%!";
       Unix.accept ~cloexec:true socket |> fst
     in
+    set_nonblock client;
     let bytes = Bytes.create n in
     let n = Unix.read client bytes 0 (Bytes.length bytes) in
     Printf.printf "  Server read %d\n%!" n;
@@ -36,6 +52,7 @@ let main () =
       finally Unix.close @@ fun () ->
       Unix.socket ~cloexec:true PF_INET SOCK_STREAM 0
     in
+    set_nonblock socket;
     let server_addr =
       let rec loop retries =
         match !server_addr with
@@ -61,5 +78,7 @@ let main () =
   Printf.printf "Server and Client test: OK\n%!"
 
 let () =
-  Printf.printf "Using fibers on OCaml 5 and threads on OCaml 4:\n%!";
+  Printf.printf "Using %sblocking sockets and %s:\n%!"
+    (if use_nonblock then "non-" else "")
+    (if is_ocaml4 then "threads on OCaml 4" else "fibers on OCaml 5");
   Test_scheduler.run main