feat(picos): check for cancellation; expose more in Base.Raw

.github/workflows/gh-pages.yml

@@ -25,6 +25,7 @@ jobs:
       - run: opam install odig moonpool trace
       - run: opam pin . -y -n
       - run: opam install . --deps-only
+      - run: opam install .
 
       - run: opam exec -- odig odoc --cache-dir=_doc/ nanoev nanoev-picos nanoev-posix nanoev_tiny_httpd
 

dune-project

@@ -76,7 +76,7 @@
   picos
   picos_std
   (tiny_httpd
-   (>= 0.17)))
+   (>= 0.19)))
  (tags
   (nanoev http)))
 

nanoev_tiny_httpd.opam

@@ -14,7 +14,7 @@ depends: [
   "nanoev-picos" {= version}
   "picos"
   "picos_std"
-  "tiny_httpd" {>= "0.17"}
+  "tiny_httpd" {>= "0.19"}
   "odoc" {with-doc}
 ]
 build: [

src/picos/base.ml

@@ -1,12 +1,18 @@
 open Common_
 
 let get_loop_exn_ : unit -> Nanoev.t = Global_ev.get_nanoev_exn
+let[@inline] check_fiber_ () : unit = Picos.Fiber.(check @@ current ())
 
 let[@inline] unwrap_ = function
   | None -> ()
   | Some (exn, bt) -> Printexc.raise_with_backtrace exn bt
 
+let[@inline] on_readable_ fd x y f : unit =
+  let ev = get_loop_exn_ () in
+  Nanoev.on_readable ev fd x y f
+
 let[@unroll 1] rec retry_read_ fd f =
+  check_fiber_ ();
   match f () with
   | res -> res
   | exception
@@ -18,15 +24,19 @@ let[@unroll 1] rec retry_read_ fd f =
     (* Trace_.message "read must wait"; *)
     let trigger = Picos.Trigger.create () in
     let closed_r = ref false in
-    let ev = get_loop_exn_ () in
+    on_readable_ fd trigger closed_r (fun ~closed trigger closed_r ->
-    Nanoev.on_readable ev fd trigger closed_r (fun ~closed trigger closed_r ->
         closed_r := closed;
         Picos.Trigger.signal trigger);
     Picos.Trigger.await trigger |> unwrap_;
     if !closed_r then raise Closed;
     retry_read_ fd f
 
+let[@inline] on_writable_ fd x y f : unit =
+  let ev = get_loop_exn_ () in
+  Nanoev.on_writable ev fd x y f
+
 let[@unroll 1] rec retry_write_ fd f =
+  check_fiber_ ();
   match f () with
   | res -> res
   | exception
@@ -36,10 +46,9 @@ let[@unroll 1] rec retry_write_ fd f =
           _,
           _ ) ->
     (* Trace_.message "write must wait"; *)
-    let ev = get_loop_exn_ () in
     let trigger = Picos.Trigger.create () in
     let closed_r = ref false in
-    Nanoev.on_writable ev fd trigger closed_r (fun ~closed trigger closed_r ->
+    on_writable_ fd trigger closed_r (fun ~closed trigger closed_r ->
         closed_r := closed;
         Picos.Trigger.signal trigger);
     Picos.Trigger.await trigger |> unwrap_;
@@ -84,16 +93,22 @@ let connect fd addr = retry_write_ fd (fun () -> Unix.connect fd addr)
 let[@inline] max_fds () =
   Option.fold ~none:1024 ~some:Nanoev.max_fds @@ Global_ev.get_nanoev ()
 
+let run_after_s_ t x y f : unit =
+  let ev = get_loop_exn_ () in
+  Nanoev.run_after_s ev t x y f
+
 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 () ->
+    run_after_s_ t trigger () (fun trigger () -> Picos.Trigger.signal trigger);
-        Picos.Trigger.signal trigger);
+    Picos.Trigger.await trigger |> unwrap_;
-    Picos.Trigger.await trigger |> unwrap_
+    check_fiber_ ()
   )
 
 module Raw = struct
+  let run_after_s = run_after_s_
+  let on_readable = on_readable_
+  let on_writable = on_writable_
   let retry_read = retry_read_
   let retry_write = retry_write_
 end

src/picos/base.mli

@@ -32,6 +32,14 @@ val sleep : float -> unit
 (** Suspend current fiber for [n] seconds *)
 
 module Raw : sig
+  val run_after_s : float -> 'a -> 'b -> ('a -> 'b -> unit) -> unit
+
+  val on_writable :
+    Unix.file_descr -> 'a -> 'b -> (closed:bool -> 'a -> 'b -> unit) -> unit
+
+  val on_readable :
+    Unix.file_descr -> 'a -> 'b -> (closed:bool -> 'a -> 'b -> unit) -> unit
+
   val retry_read : Unix.file_descr -> (unit -> 'a) -> 'a
   val retry_write : Unix.file_descr -> (unit -> 'a) -> 'a
 end

src/tiny_httpd/nanoev_tiny_httpd.ml

@@ -259,10 +259,10 @@ open struct
     slice.len <- 0
 end
 
-let create ?(masksigpipe = not Sys.win32) ?max_connections ?max_buf_pool_size
+let create ?enable_logging ?(masksigpipe = not Sys.win32) ?max_connections
-    ?(timeout = 0.0) ?buf_size ?(get_time_s = Unix.gettimeofday)
+    ?max_buf_pool_size ?(timeout = 0.0) ?buf_size
-    ?(addr = "127.0.0.1") ?(port = 8080) ?sock ?middlewares ~new_thread () :
+    ?(get_time_s = Unix.gettimeofday) ?(addr = "127.0.0.1") ?(port = 8080) ?sock
-    TH.Server.t =
+    ?head_middlewares ?middlewares ~new_thread () : TH.Server.t =
   let max_connections = get_max_connection_ ?max_connections () in
   let max_pool_size =
     match max_buf_pool_size with
@@ -300,4 +300,5 @@ let create ?(masksigpipe = not Sys.win32) ?max_connections ?max_buf_pool_size
     let tcp_server () = tcp_server_builder
   end in
   let backend = (module B : TH.Server.IO_BACKEND) in
-  TH.Server.create_from ?buf_size ?middlewares ~backend ()
+  TH.Server.create_from ?enable_logging ?buf_size ?head_middlewares ?middlewares
+    ~backend ()

src/tiny_httpd/nanoev_tiny_httpd.mli

@@ -1,6 +1,7 @@
 module TH = Tiny_httpd_core
 
 val create :
+  ?enable_logging:bool ->
   ?masksigpipe:bool ->
   ?max_connections:int ->
   ?max_buf_pool_size:int ->
@@ -10,6 +11,7 @@ val create :
   ?addr:string ->
   ?port:int ->
   ?sock:Unix.file_descr ->
+  ?head_middlewares:TH.Server.Head_middleware.t list ->
   ?middlewares:([ `Encoding | `Stage of int ] * TH.Server.Middleware.t) list ->
   new_thread:((unit -> unit) -> unit) ->
   unit ->