eio_windows: fix anonymous pipes on Windows

lib_eio_windows/low_level.ml

@@ -38,25 +38,27 @@ let rec do_nonblocking ty fn fd =
       );
     do_nonblocking ty fn fd
 
+let transfer label ty op fd =
+  match Fd.is_blocking fd with
+  | true ->
+    Fd.use_exn label fd @@ fun fd ->
+    in_worker_thread ~label (fun () -> op fd)
+  | false ->
+    (match ty with Read -> await_readable fd | Write -> await_writable fd);
+    Fd.use_exn label fd @@ fun fd ->
+    do_nonblocking ty op fd
+
 let read fd buf start len =
-  await_readable fd;
-  Fd.use_exn "read" fd @@ fun fd ->
-  do_nonblocking Read (fun fd -> Unix.read fd buf start len) fd
+  transfer "read" Read (fun fd -> Unix.read fd buf start len) fd
 
 let read_cstruct fd (buf:Cstruct.t) =
-  await_readable fd;
-  Fd.use_exn "read_cstruct" fd @@ fun fd ->
-  do_nonblocking Read (fun fd -> Unix.read_bigarray fd buf.buffer buf.off buf.len) fd
+  transfer "read_cstruct" Read (fun fd -> Unix.read_bigarray fd buf.buffer buf.off buf.len) fd
 
 let write fd buf start len =
-  await_writable fd;
-  Fd.use_exn "write" fd @@ fun fd ->
-  do_nonblocking Write (fun fd -> Unix.write fd buf start len) fd
+  transfer "write" Write (fun fd -> Unix.write fd buf start len) fd
 
 let write_cstruct fd (buf:Cstruct.t) =
-  await_writable fd;
-  Fd.use_exn "write_cstruct" fd @@ fun fd ->
-  do_nonblocking Write (fun fd -> Unix.write_bigarray fd buf.buffer buf.off buf.len) fd
+  transfer "write_cstruct" Write (fun fd -> Unix.write_bigarray fd buf.buffer buf.off buf.len) fd
 
 let sleep_until time =
   Sched.enter @@ fun t k ->
@@ -305,8 +307,6 @@ let ftruncate fd len =
 
 let pipe ~sw =
   let unix_r, unix_w = Unix.pipe ~cloexec:true () in
-  let r = Fd.of_unix ~sw ~blocking:false ~close_unix:true unix_r in
-  let w = Fd.of_unix ~sw ~blocking:false ~close_unix:true unix_w in
-  Unix.set_nonblock unix_r;
-  Unix.set_nonblock unix_w;
+  let r = Fd.of_unix ~sw ~blocking:true ~close_unix:true unix_r in
+  let w = Fd.of_unix ~sw ~blocking:true ~close_unix:true unix_w in
   r, w

lib_eio_windows/test/test.ml

@@ -85,6 +85,7 @@ let () =
   Alcotest.run ~bail:true "eio_windows" [
     "net", Test_net.tests env;
     "fs", Test_fs.tests env;
+    "pipe", Test_pipe.tests;
     "timeout", Timeout.tests env;
     "random", Random.tests env;
     "dla", Dla.tests;

lib_eio_windows/test/test_pipe.ml

@@ -0,0 +1,40 @@
+(* Tests for anonymous pipes *)
+
+open Eio.Std
+
+let read_all flow =
+  let b = Buffer.create 16 in
+  Eio.Flow.copy flow (Eio.Flow.buffer_sink b);
+  Buffer.contents b
+
+let test_transfer () =
+  Switch.run @@ fun sw ->
+  let r, w = Eio_unix.pipe sw in
+  Eio.Flow.copy_string "hello" w;
+  Eio.Flow.close w;
+  Alcotest.(check string) "transfer" "hello" (read_all r)
+
+let test_read_before_write () =
+  Switch.run @@ fun sw ->
+  let r, w = Eio_unix.pipe sw in
+  Fiber.both
+    (fun () ->
+       let buf = Cstruct.create 8 in
+       let n = Eio.Flow.single_read r buf in
+       Alcotest.(check string) "data" "ping" (Cstruct.to_string ~len:n buf))
+    (fun () -> Eio.Flow.copy_string "ping" w)
+
+let test_eof () =
+  Switch.run @@ fun sw ->
+  let r, w = Eio_unix.pipe sw in
+  Eio.Flow.close w;
+  let buf = Cstruct.create 1 in
+  match Eio.Flow.single_read r buf with
+  | _ -> Alcotest.fail "read should have signaled eof"
+  | exception End_of_file -> ()
+
+let tests = [
+  "transfer", `Quick, test_transfer;
+  "read-before-write", `Quick, test_read_before_write;
+  "eof", `Quick, test_eof;
+]