feat: add nanoev.picos optional lib

dune-project

@@ -18,6 +18,11 @@
  (name nanoev)
  (synopsis "Tiny event loop around `select`")
  (depends ocaml dune base-unix)
- (tags (unix select async)))
+ (depopts
+  (picos
+   (and (>= 0.5) (< 0.7))))
+ (tags
+  (unix select async)))
+
 
 ; See the complete stanza docs at https://dune.readthedocs.io/en/stable/reference/dune-project/index.html

nanoev.opam

@@ -13,6 +13,9 @@ depends: [
   "base-unix"
   "odoc" {with-doc}
 ]
+depopts: [
+  "picos" {>= "0.5" & < "0.7"}
+]
 build: [
   ["dune" "subst"] {dev}
   [

src/picos/dune

@@ -0,0 +1,6 @@
+
+(library
+  (name nanoev_picos)
+  (public_name nanoev.picos)
+  (optional) ; picos
+  (libraries threads picos nanoev))

src/picos/nanoev_picos.ml

@@ -0,0 +1,107 @@
+let ( let@ ) = ( @@ )
+
+module Global_ = struct
+  type st =
+    | None
+    | Some of {
+        active: bool Atomic.t;
+        nanoev: Nanoev.t;
+        th: Thread.t;
+      }
+
+  let st : st Atomic.t = Atomic.make None
+  let lock = Mutex.create ()
+
+  let with_lock lock f =
+    Mutex.lock lock;
+    match f () with
+    | exception e ->
+      Mutex.unlock lock;
+      raise e
+    | x ->
+      Mutex.unlock lock;
+      x
+
+  let bg_thread_ ~active ~evloop () : unit =
+    while Atomic.get active do
+      Nanoev.step evloop
+    done
+
+  let[@inline] has_bg_thread () = Atomic.get st <> None
+
+  let setup_bg_thread (ev : Nanoev.t) : unit =
+    let@ () = with_lock lock in
+    (match Atomic.get st with
+    | Some st ->
+      Atomic.set st.active false;
+      Nanoev.wakeup_from_outside st.nanoev;
+      Thread.join st.th
+    | None -> ());
+    let active = Atomic.make true in
+    Atomic.set st
+    @@ Some
+         {
+           active;
+           nanoev = ev;
+           th = Thread.create (bg_thread_ ~active ~evloop:ev) ();
+         }
+end
+
+let has_bg_thread = Global_.has_bg_thread
+let setup_bg_thread = Global_.setup_bg_thread
+
+let[@inline] get_loop_exn_ () : Nanoev.t =
+  match Atomic.get Global_.st with
+  | None -> failwith "No nanoev loop installed."
+  | Some st -> st.nanoev
+
+let[@inline] unwrap_ = function
+  | None -> ()
+  | Some (exn, bt) -> Printexc.raise_with_backtrace exn bt
+
+let retry_read_ fd f =
+  let ev = get_loop_exn_ () in
+  let rec loop () =
+    match f () with
+    | res -> res
+    | exception
+        Unix.Unix_error ((Unix.EAGAIN | Unix.EWOULDBLOCK | Unix.EINTR), _, _) ->
+      let trigger = Picos.Trigger.create () in
+      Nanoev.on_readable ev fd trigger () (fun trigger () ->
+          Picos.Trigger.signal trigger);
+      Picos.Trigger.await trigger |> unwrap_;
+      loop ()
+  in
+  loop ()
+
+let retry_write_ fd f =
+  let ev = get_loop_exn_ () in
+  let rec loop () =
+    match f () with
+    | res -> res
+    | exception
+        Unix.Unix_error ((Unix.EAGAIN | Unix.EWOULDBLOCK | Unix.EINTR), _, _) ->
+      let trigger = Picos.Trigger.create () in
+      Nanoev.on_writable ev fd trigger () (fun trigger () ->
+          Picos.Trigger.signal trigger);
+      Picos.Trigger.await trigger |> unwrap_;
+      loop ()
+  in
+  loop ()
+
+let read fd buf i len : int = retry_read_ fd (fun () -> Unix.read fd buf i len)
+let accept fd = retry_read_ fd (fun () -> Unix.accept fd)
+
+let write fd buf i len : int =
+  retry_write_ fd (fun () -> Unix.write fd buf i len)
+
+let connect fd addr = retry_write_ fd (fun () -> Unix.connect fd addr)
+
+let sleep t =
+  if t > 0. then (
+    let ev = get_loop_exn_ () in
+    let trigger = Picos.Trigger.create () in
+    Nanoev.run_after_s ev t trigger () (fun trigger () ->
+        Picos.Trigger.signal trigger);
+    Picos.Trigger.await trigger |> unwrap_
+  )

src/picos/nanoev_picos.mli

@@ -0,0 +1,15 @@
+(** Basic interface with picos *)
+
+val setup_bg_thread : Nanoev.t -> unit
+(** Install this event loop in a background thread *)
+
+val has_bg_thread : unit -> bool
+(** [has_bg_thread ()] is [true] iff a background thread is running a nanoev loop *)
+
+(** {2 Non blocking IO primitives} *)
+
+val read : Unix.file_descr -> bytes -> int -> int -> int
+val connect : Unix.file_descr -> Unix.sockaddr -> unit
+val write : Unix.file_descr -> bytes -> int -> int -> int
+val accept : Unix.file_descr -> Unix.file_descr * Unix.sockaddr
+val sleep : float -> unit