wip: unix event loop

dune-project

@@ -33,6 +33,20 @@
  (tags
   (thread pool domain futures fork-join)))
 
+(package
+ (name moonpool-io)
+ (synopsis "Non blocking IO for Moonpool based on Unix")
+ (depends
+  (ocaml (>= 5.0))
+  dune
+  (poll (and (>= 3.0) (< 4.0)))
+  (iostream (>= 0.2.2))
+  (trace :with-test))
+ (depopts
+   mtime)
+ (tags
+  (moonpool io unix async)))
+
 (package
   (name moonpool-lwt)
   (synopsis "Event loop for moonpool based on Lwt-engine (experimental)")

moonpool-io.opam

@@ -0,0 +1,34 @@
+# This file is generated by dune, edit dune-project instead
+opam-version: "2.0"
+version: "0.6"
+synopsis: "Non blocking IO for Moonpool based on Unix"
+maintainer: ["Simon Cruanes"]
+authors: ["Simon Cruanes"]
+license: "MIT"
+tags: ["moonpool" "io" "unix" "async"]
+homepage: "https://github.com/c-cube/moonpool"
+bug-reports: "https://github.com/c-cube/moonpool/issues"
+depends: [
+  "ocaml" {>= "5.0"}
+  "dune" {>= "3.0"}
+  "poll" {>= "3.0" & < "4.0"}
+  "iostream" {>= "0.2.2"}
+  "trace" {with-test}
+  "odoc" {with-doc}
+]
+depopts: ["mtime"]
+build: [
+  ["dune" "subst"] {dev}
+  [
+    "dune"
+    "build"
+    "-p"
+    name
+    "-j"
+    jobs
+    "@install"
+    "@runtest" {with-test}
+    "@doc" {with-doc}
+  ]
+]
+dev-repo: "git+https://github.com/c-cube/moonpool.git"

src/fib/fiber.ml

@@ -10,7 +10,7 @@ type cancel_callback = Exn_bt.t -> unit
 let prom_of_fut : 'a Fut.t -> 'a Fut.promise =
   Fut.Private_.unsafe_promise_of_fut
 
-module Private_ = struct
+module Private0 = struct
   type 'a t = {
     id: Handle.t;  (** unique identifier for this fiber *)
     state: 'a state A.t;  (** Current state in the lifetime of the fiber *)
@@ -42,7 +42,7 @@ module Private_ = struct
     | Terminating_or_done _ -> true
 end
 
-include Private_
+include Private0
 
 let create_ ~ls ~runner () : 'a t =
   let id = Handle.generate_fresh () in
@@ -317,3 +317,9 @@ let yield () : unit =
     check_if_cancelled_ self;
     Suspend_.yield ();
     check_if_cancelled_ self
+
+module Private_ = struct
+  include Private0
+
+  let cancel_from_outside = resolve_as_failed_
+end

src/fib/fiber.mli

@@ -31,6 +31,7 @@ module Private_ : sig
   type any = Any : _ t -> any [@@unboxed]
 
   val get_cur : unit -> any option
+  val cancel_from_outside : _ t -> Exn_bt.t -> unit
 end
 
 (**/**)

src/fib/main.ml

@@ -8,7 +8,9 @@ let main (f : Runner.t -> 'a) : 'a =
     Fiber.on_result fiber (fun _ -> Runner.shutdown_without_waiting runner);
     (* run the main thread *)
     Fifo_pool.Private_.run_thread st runner ~on_exn:(fun e bt ->
-        raise (Oh_no (Exn_bt.make e bt)));
+        let ebt = Exn_bt.make e bt in
+        Fiber.Private_.cancel_from_outside fiber ebt;
+        raise (Oh_no ebt));
     match Fiber.peek fiber with
     | Some (Ok x) -> x
     | Some (Error ebt) -> Exn_bt.raise ebt