.github/workflows/gh-pages.yml

@@ -22,7 +22,7 @@ jobs:
       # temporary until it's in a release
       - run: opam pin picos 0.6.0 -y -n
 
-      - run: opam install odig moonpool moonpool-lwt -t
+      - run: opam install odig moonpool moonpool-lwt moonpool-io -t
 
       - run: opam exec -- odig odoc --cache-dir=_doc/ moonpool moonpool-lwt
 

.github/workflows/main.yml

@@ -31,7 +31,7 @@ jobs:
 
       - run: opam pin picos 0.6.0 -y -n
 
-      - run: opam install -t moonpool moonpool-lwt --deps-only
+      - run: opam install -t moonpool moonpool-lwt moonpool-io --deps-only
       - run: opam exec -- dune build @install
 
       # install some depopts
@@ -62,7 +62,7 @@ jobs:
       # temporary until it's in a release
       - run: opam pin https://github.com/ocaml-multicore/picos.git -y -n
 
-      - run: opam install -t moonpool moonpool-lwt --deps-only
+      - run: opam install -t moonpool moonpool-lwt moonpool-io --deps-only
       - run: opam exec -- dune build @install
       # install some depopts
       - run: opam install thread-local-storage trace domain-local-await

dune-project

@@ -52,5 +52,17 @@
     (trace-tef :with-test)
     (odoc :with-doc)))
 
+(package
+  (name moonpool-io)
+  (synopsis "Async IO for moonpool, relying on picos (experimental)")
+  (allow_empty)  ; on < 5.0
+  (depends
+    (moonpool (= :version))
+    (picos_io (and (>= 0.5) (< 0.7)))
+    (ocaml (>= 5.0))
+    (trace :with-test)
+    (trace-tef :with-test)
+    (odoc :with-doc)))
+
 
 ; See the complete stanza docs at https://dune.readthedocs.io/en/stable/dune-files.html#dune-project

moonpool-io.opam

@@ -0,0 +1,33 @@
+# This file is generated by dune, edit dune-project instead
+opam-version: "2.0"
+version: "0.9"
+synopsis: "Async IO for moonpool, relying on picos (experimental)"
+maintainer: ["Simon Cruanes"]
+authors: ["Simon Cruanes"]
+license: "MIT"
+homepage: "https://github.com/c-cube/moonpool"
+bug-reports: "https://github.com/c-cube/moonpool/issues"
+depends: [
+  "dune" {>= "3.0"}
+  "moonpool" {= version}
+  "picos_io" {>= "0.5" & < "0.7"}
+  "ocaml" {>= "5.0"}
+  "trace" {with-test}
+  "trace-tef" {with-test}
+  "odoc" {with-doc}
+]
+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"

moonpool-lwt.opam

@@ -11,8 +11,6 @@ depends: [
   "dune" {>= "3.0"}
   "moonpool" {= version}
   "ocaml" {>= "5.0"}
-  "qcheck-core" {with-test & >= "0.19"}
-  "hmap" {with-test}
   "lwt"
   "base-unix"
   "trace" {with-test}

src/io/dune

@@ -0,0 +1,7 @@
+(library
+ (name moonpool_io)
+ (public_name moonpool-io)
+ (synopsis "Async IO for moonpool, using Picos")
+ (enabled_if
+  (>= %{ocaml_version} 5.0))
+ (libraries moonpool moonpool.fib picos_io picos_io.select picos_io.fd))

src/io/moonpool_io.ml

@@ -0,0 +1,15 @@
+[@@@deprecated "just use lwt or eio or something else"]
+
+module Fd = Picos_io_fd
+module Unix = Picos_io.Unix
+module Select = Picos_io_select
+
+let fd_of_unix_fd : Unix.file_descr -> Fd.t = Fun.id
+
+(** [main f] runs [f()] inside a scheduler. *)
+let main (f : Moonpool.Runner.t -> 'a) : 'a = Moonpool_fib.main f
+
+(** {2 Async read/write} *)
+
+let read = Unix.read
+let write = Unix.write

src/sync/dune

@@ -0,0 +1,5 @@
+(library
+ (name moonpool_sync)
+ (public_name moonpool.sync)
+ (synopsis "Cooperative synchronization primitives for Moonpool")
+ (libraries moonpool picos picos_std.sync picos_std.event))

src/sync/event.ml

@@ -0,0 +1,11 @@
+include Picos_std_event.Event
+
+let[@inline] of_fut (fut : _ Moonpool.Fut.t) : _ t =
+  from_computation (Moonpool.Fut.Private_.as_computation fut)
+
+module Infix = struct
+  let[@inline] ( let+ ) x f = map f x
+  let ( >|= ) = ( let+ )
+end
+
+include Infix

src/sync/event.mli

@@ -0,0 +1,12 @@
+include module type of struct
+  include Picos_std_event.Event
+end
+
+val of_fut : 'a Moonpool.Fut.t -> 'a t
+
+module Infix : sig
+  val ( >|= ) : 'a t -> ('a -> 'b) -> 'b t
+  val ( let+ ) : 'a t -> ('a -> 'b) -> 'b t
+end
+
+include module type of Infix

src/sync/lock.ml

@@ -0,0 +1,38 @@
+module Mutex = Picos_std_sync.Mutex
+
+type 'a t = {
+  mutex: Mutex.t;
+  mutable content: 'a;
+}
+
+let create content : _ t = { mutex = Mutex.create (); content }
+
+let with_ (self : _ t) f =
+  Mutex.lock self.mutex;
+  try
+    let x = f self.content in
+    Mutex.unlock self.mutex;
+    x
+  with e ->
+    Mutex.unlock self.mutex;
+    raise e
+
+let[@inline] mutex self = self.mutex
+let[@inline] update self f = with_ self (fun x -> self.content <- f x)
+
+let[@inline] update_map l f =
+  with_ l (fun x ->
+      let x', y = f x in
+      l.content <- x';
+      y)
+
+let get l =
+  Mutex.lock l.mutex;
+  let x = l.content in
+  Mutex.unlock l.mutex;
+  x
+
+let set l x =
+  Mutex.lock l.mutex;
+  l.content <- x;
+  Mutex.unlock l.mutex

src/sync/lock.mli

@@ -0,0 +1,56 @@
+(** Mutex-protected resource.
+
+    This lock is a synchronous concurrency primitive, as a thin wrapper around
+    {!Mutex} that encourages proper management of the critical section in RAII
+    style:
+
+    {[
+      let (let@) = (@@)
+
+
+      …
+      let compute_foo =
+        (* enter critical section *)
+        let@ x = Lock.with_ protected_resource in
+        use_x;
+        return_foo ()
+        (* exit critical section *)
+      in
+      …
+    ]}
+
+    This lock is based on {!Picos_sync.Mutex} so it is [await]-safe.
+
+    @since 0.7 *)
+
+type 'a t
+(** A value protected by a cooperative mutex *)
+
+val create : 'a -> 'a t
+(** Create a new protected value. *)
+
+val with_ : 'a t -> ('a -> 'b) -> 'b
+(** [with_ l f] runs [f x] where [x] is the value protected with the lock [l],
+    in a critical section. If [f x] fails, [with_lock l f] fails too but the
+    lock is released. *)
+
+val update : 'a t -> ('a -> 'a) -> unit
+(** [update l f] replaces the content [x] of [l] with [f x], while protected by
+    the mutex. *)
+
+val update_map : 'a t -> ('a -> 'a * 'b) -> 'b
+(** [update_map l f] computes [x', y = f (get l)], then puts [x'] in [l] and
+    returns [y], while protected by the mutex. *)
+
+val mutex : _ t -> Picos_std_sync.Mutex.t
+(** Underlying mutex. *)
+
+val get : 'a t -> 'a
+(** Atomically get the value in the lock. The value that is returned isn't
+    protected! *)
+
+val set : 'a t -> 'a -> unit
+(** Atomically set the value.
+
+    {b NOTE} caution: using {!get} and {!set} as if this were a {!ref} is an
+    anti pattern and will not protect data against some race conditions. *)

src/sync/moonpool_sync.ml

@@ -0,0 +1,11 @@
+[@@@ocaml.deprecated "use Picos_std_sync directly or single threaded solutions"]
+
+module Mutex = Picos_std_sync.Mutex
+module Condition = Picos_std_sync.Condition
+module Lock = Lock
+module Event = Event
+module Semaphore = Picos_std_sync.Semaphore
+module Lazy = Picos_std_sync.Lazy
+module Latch = Picos_std_sync.Latch
+module Ivar = Picos_std_sync.Ivar
+module Stream = Picos_std_sync.Stream