eio_windows: unregister FDs on cancel

lib_eio_posix/test/dune

@@ -3,8 +3,8 @@
   (enabled_if (= %{os_type} "Unix"))
   (deps (package eio_posix)))
 
-(test
-  (name open_beneath)
+(tests
+  (names open_beneath test_await)
   (package eio_posix)
   (build_if (= %{os_type} "Unix"))
   (libraries eio_posix))

lib_eio_posix/test/test_await.ml

@@ -0,0 +1,25 @@
+open Eio.Std
+
+let () =
+  Eio_posix.run @@ fun _ ->
+  let a, b = Unix.(socketpair PF_UNIX SOCK_STREAM 0) in
+  (* Start awaiting readable/writable state, but cancel immediately. *)
+  try
+    Eio.Cancel.sub (fun cc ->
+        Fiber.all [
+          (fun () -> Eio_unix.await_readable a);
+          (fun () -> Eio_unix.await_writable b);
+          (fun () -> Eio.Cancel.cancel cc Exit);
+        ];
+        assert false
+      )
+  with Eio.Cancel.Cancelled _ ->
+    (* Now wait for something else. Will fail if the old FDs are still being waited on. *)
+    let c, d = Unix.(socketpair PF_UNIX SOCK_STREAM 0) in
+    Unix.close a;
+    Unix.close b;
+    Fiber.first
+      (fun () -> Eio_unix.await_readable c)
+      (fun () -> Eio_unix.await_writable d);
+    Unix.close c;
+    Unix.close d

lib_eio_windows/sched.ml

@@ -271,6 +271,8 @@ let await_readable t (k : unit Suspended.t) fd =
     if was_empty then update t waiters fd;
     Fiber_context.set_cancel_fn k.fiber (fun ex ->
         Lwt_dllist.remove node;
+        if Lwt_dllist.is_empty waiters.read then
+          update t waiters fd;
         t.active_ops <- t.active_ops - 1;
         enqueue_failed_thread t k ex
       );
@@ -287,6 +289,8 @@ let await_writable t (k : unit Suspended.t) fd =
     if was_empty then update t waiters fd;
     Fiber_context.set_cancel_fn k.fiber (fun ex ->
         Lwt_dllist.remove node;
+        if Lwt_dllist.is_empty waiters.write then
+          update t waiters fd;
         t.active_ops <- t.active_ops - 1;
         enqueue_failed_thread t k ex
       );

lib_eio_windows/test/test.ml

@@ -1,3 +1,5 @@
+open Eio.Std
+
 module Timeout = struct
   let test clock () =
     let t0 = Unix.gettimeofday () in
@@ -48,6 +50,35 @@ module Dla = struct
   ]
 end
 
+module Await_fd = struct
+  let test_cancel () =
+    let a, b = Unix.(socketpair PF_UNIX SOCK_STREAM 0) in
+    (* Start awaiting readable/writable state, but cancel immediately. *)
+    try
+      Eio.Cancel.sub (fun cc ->
+          Fiber.all [
+            (fun () -> Eio_unix.await_readable a);
+            (fun () -> Eio_unix.await_writable b);
+            (fun () -> Eio.Cancel.cancel cc Exit);
+          ];
+          assert false
+        )
+    with Eio.Cancel.Cancelled _ ->
+      (* Now wait for something else. Will fail if the old FDs are still being waited on. *)
+      let c, d = Unix.(socketpair PF_UNIX SOCK_STREAM 0) in
+      Unix.close a;
+      Unix.close b;
+      Fiber.first
+        (fun () -> Eio_unix.await_readable c)
+        (fun () -> Eio_unix.await_writable d);
+      Unix.close c;
+      Unix.close d
+
+  let tests = [
+    "cancel", `Quick, test_cancel;
+  ]
+end
+
 
 let () =
   Eio_windows.run @@ fun env ->
@@ -56,5 +87,6 @@ let () =
     "fs", Test_fs.tests env;
     "timeout", Timeout.tests env;
     "random", Random.tests env;
-    "dla", Dla.tests
-  ]
+    "dla", Dla.tests;
+    "await", Await_fd.tests;
+  ]