Make pipes nonblocking in Eio_unix.pipe, zap Eio_linux.pipe

Make sure we use pipes as nonblocking see #319 in Eio_unix.pipe.
Zap Eio_linux.pipe since it's only used in tests.

Author: haesbaert

lib_eio_linux/eio_linux.ml

@@ -1292,15 +1292,6 @@ let stdenv ~run_event_loop =
     method debug = Eio.Private.Debug.v
   end
 
-let pipe sw =
-  let r, w = Unix.pipe () in
-  (* XXX workaround for issue #319, PR #327 *)
-  Unix.set_nonblock r;
-  Unix.set_nonblock w;
-  let r = source (FD.of_unix ~sw ~seekable:false ~close_unix:true r) in
-  let w = sink (FD.of_unix ~sw ~seekable:false ~close_unix:true w) in
-  r, w
-
 let monitor_event_fd t =
   let buf = Cstruct.create 8 in
   while true do
@@ -1455,6 +1446,9 @@ let rec run : type a.
             )
           | Eio_unix.Private.Pipe sw -> Some (fun k ->
               let r, w = Unix.pipe ~cloexec:true () in
+              (* See issue #319, PR #327 *)
+              Unix.set_nonblock r;
+              Unix.set_nonblock w;
               let r = (flow (FD.of_unix ~sw ~seekable:false ~close_unix:true r) :> <Eio.Flow.source; Eio.Flow.close; Eio_unix.unix_fd>) in
               let w = (flow (FD.of_unix ~sw ~seekable:false ~close_unix:true w) :> <Eio.Flow.sink; Eio.Flow.close; Eio_unix.unix_fd>) in
               continue k (r, w)

lib_eio_linux/eio_linux.mli

@@ -74,10 +74,6 @@ type stdenv = <
 val get_fd : <has_fd; ..> -> FD.t
 val get_fd_opt : #Eio.Generic.t -> FD.t option
 
-val pipe : Switch.t -> source * sink
-(** [pipe sw] is a source-sink pair [(r, w)], where data written to [w] can be read from [r].
-    It is implemented as a Unix pipe. *)
-
 (** {1 Main Loop} *)
 
 val run :

lib_eio_linux/tests/test.ml

@@ -20,7 +20,7 @@ let read_one_byte ~sw r =
 let test_poll_add () =
   Eio_linux.run @@ fun _stdenv ->
   Switch.run @@ fun sw ->
-  let r, w = Eio_linux.pipe sw in
+  let r, w = Eio_unix.pipe sw in
   let thread = read_one_byte ~sw r in
   Fiber.yield ();
   let w = Option.get (Eio_linux.get_fd_opt w) in
@@ -33,7 +33,7 @@ let test_poll_add () =
 let test_poll_add_busy () =
   Eio_linux.run ~queue_depth:2 @@ fun _stdenv ->
   Switch.run @@ fun sw ->
-  let r, w = Eio_linux.pipe sw in
+  let r, w = Eio_unix.pipe sw in
   let a = read_one_byte ~sw r in
   let b = read_one_byte ~sw r in
   Fiber.yield ();
@@ -50,7 +50,7 @@ let test_copy () =
   Eio_linux.run ~queue_depth:3 @@ fun _stdenv ->
   Switch.run @@ fun sw ->
   let msg = "Hello!" in
-  let from_pipe, to_pipe = Eio_linux.pipe sw in
+  let from_pipe, to_pipe = Eio_unix.pipe sw in
   let buffer = Buffer.create 20 in
   Fiber.both
     (fun () -> Eio.Flow.copy from_pipe (Eio.Flow.buffer_sink buffer))
@@ -67,8 +67,8 @@ let test_direct_copy () =
   Eio_linux.run ~queue_depth:4 @@ fun _stdenv ->
   Switch.run @@ fun sw ->
   let msg = "Hello!" in
-  let from_pipe1, to_pipe1 = Eio_linux.pipe sw in
-  let from_pipe2, to_pipe2 = Eio_linux.pipe sw in
+  let from_pipe1, to_pipe1 = Eio_unix.pipe sw in
+  let from_pipe2, to_pipe2 = Eio_unix.pipe sw in
   let buffer = Buffer.create 20 in
   let to_output = Eio.Flow.buffer_sink buffer in
   Switch.run (fun sw ->
@@ -85,9 +85,15 @@ let test_direct_copy () =
 let test_iovec () =
   Eio_linux.run ~queue_depth:4 @@ fun _stdenv ->
   Switch.run @@ fun sw ->
-  let from_pipe, to_pipe = Eio_linux.pipe sw in
-  let from_pipe = Eio_linux.get_fd from_pipe in
-  let to_pipe = Eio_linux.get_fd to_pipe in
+  let from_pipe, to_pipe = Eio_unix.pipe sw in
+  let from_pipe =
+    Eio_unix.FD.take from_pipe |>
+    Eio_linux.FD.of_unix ~sw ~seekable:false ~close_unix:true
+  in
+  let to_pipe =
+    Eio_unix.FD.take to_pipe |>
+    Eio_linux.FD.of_unix ~sw ~seekable:false ~close_unix:true
+  in
   let message = Cstruct.of_string "Got [   ] and [   ]" in
   let rec recv = function
     | [] -> ()