Add blocking

Author: polytypic

README.md

@@ -7,12 +7,12 @@
 (homepage "https://github.com/ocaml-multicore/kcas")
 (license "ISC")
 (package (name kcas)
- (synopsis "Multi-word compare-and-set library")
+ (synopsis "Software transactional memory based on lock-free multi-word compare-and-set")
  (depends 
   (ocaml (>= 5.0))
   (mdx (and (>= 1.10.0) :with-test))))
 (package (name kcas_data)
- (synopsis "Compositional lock-free data structures")
+ (synopsis "Compositional lock-free data structures and primitives for communication and synchronization")
  (depends
   (kcas (= :version))
   (mdx (and (>= 1.10.0) :with-test))))

dune-project

@@ -1,6 +1,7 @@
 # This file is generated by dune, edit dune-project instead
 opam-version: "2.0"
-synopsis: "Multi-word compare-and-set library"
+synopsis:
+  "Software transactional memory based on lock-free multi-word compare-and-set"
 maintainer: ["KC Sivaramakrishnan <sk826@cl.cam.ac.uk>"]
 authors: ["KC Sivaramakrishnan <sk826@cl.cam.ac.uk>"]
 license: "ISC"

kcas.opam

@@ -1,6 +1,7 @@
 # This file is generated by dune, edit dune-project instead
 opam-version: "2.0"
-synopsis: "Compositional lock-free data structures"
+synopsis:
+  "Compositional lock-free data structures and primitives for communication and synchronization"
 maintainer: ["KC Sivaramakrishnan <sk826@cl.cam.ac.uk>"]
 authors: ["KC Sivaramakrishnan <sk826@cl.cam.ac.uk>"]
 license: "ISC"

kcas_data.opam

@@ -16,6 +16,8 @@
  * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  *)
 
+(** Randomized exponential backoff mechanism. *)
+
 type t
 (** Type of backoff values. *)
 

src/backoff.mli

@@ -0,0 +1,43 @@
+type mode = [ `Cancelable | `Protected ]
+type t = { release : unit -> unit; await : unit -> unit }
+
+module Default = struct
+  type t = { mutex : Mutex.t; condition : Condition.t; mutable alarm : bool }
+
+  let init () =
+    let t =
+      let mutex = Mutex.create ()
+      and condition = Condition.create ()
+      and alarm = false in
+      { mutex; condition; alarm }
+    in
+    let methods =
+      let release () =
+        if not t.alarm then (
+          Mutex.lock t.mutex;
+          t.alarm <- true;
+          Mutex.unlock t.mutex;
+          Condition.signal t.condition)
+      and await () =
+        if not t.alarm then (
+          Mutex.lock t.mutex;
+          while not t.alarm do
+            Condition.wait t.condition t.mutex
+          done;
+          Mutex.unlock t.mutex)
+      in
+      { release; await }
+    in
+    fun (_ : mode) ->
+      t.alarm <- false;
+      methods
+end
+
+let key = Domain.DLS.new_key Default.init
+
+let using ~prepare_for_await ~while_running =
+  let old = Domain.DLS.get key in
+  Domain.DLS.set key prepare_for_await;
+  Fun.protect ~finally:(fun () -> Domain.DLS.set key old) while_running
+
+let prepare_for_await mode = Domain.DLS.get key mode

src/domain_local_await.ml

@@ -0,0 +1,57 @@
+(** A scheduler independent blocking mechanism. *)
+
+type mode =
+  [ `Cancelable
+    (** In the [`Cancelable] mode {!prepare_for_await} and {!t.await} are
+        allowed to raise an (unspecified) exception that indicates that the
+        caller's fiber has been canceled (and should terminate).  If an
+        exception is raised, then the caller should perform whatever cleanup is
+        necessary to e.g. avoid space leaks. *)
+  | `Protected
+    (** In the [`Protected] mode {!prepare_for_await} and {!t.await} should act
+        as if the caller's fiber has not been canceled.  The [`Protected] mode
+        should only be used in special cases such as when await might be needed
+        during cleanup actions during stack unwinding. *)
+  ]
+(** Specifies a mode of operation regarding cancellation. *)
+
+type t = {
+  release : unit -> unit;
+      (** [t.release ()] resumes the corresponding caller of [t.await ()] or
+          does nothing in case the corresponding [t.await ()] has already
+          resumed or the target fiber has been canceled.
+
+          {b NOTE}: An implementation of [t.release ()] should never fail. *)
+  await : unit -> unit;
+      (** [t.await ()] suspends the caller at most until [t.release ()] is
+          called. *)
+}
+(** Represents an asynchronous trigger.
+
+    {b NOTE}: {!release} and {!await} should be domain safe and ideally
+    optimized with the assumption that {!release} may be called multiple times
+    and even before {!await} is called.  Furthermore, {!await} may be called at
+    most once. *)
+
+val prepare_for_await : mode -> t
+(** [prepare_for_await mode] prepares and returns a trigger [t] for (at most)
+    one use of [t.await ()] by calling the [prepare] function registered for the
+    current domain.
+
+    {b NOTE}: It is allowed for two different calls of [prepare_for_await] to
+    return the same trigger and e.g. share a single trigger per domain or per
+    fiber or even just have one single trigger. *)
+
+val using : prepare_for_await:(mode -> t) -> while_running:(unit -> 'a) -> 'a
+(** [using ~prepare_for_await ~while_running] registers the given asynchronous
+    trigger mechanism for the current domain for the duration of running the
+    given scheduler.  In other words, this sets the implementation of
+    {!prepare_for_await} for the current domain.
+
+    {b NOTE}: The given [prepare_for_await] function is called every time
+    {!prepare_for_await} is called while the scheduler is running.
+
+    {b NOTE}: This is normally only called by libraries that implement
+    schedulers and the specified [prepare_for_await] typically returns a trigger
+    mechanism {!t} that tightly integrates with the scheduler by e.g. performing
+    an effect to suspend the current fiber when {!t.await} is called. *)