WIP: Add blocking

Author: polytypic

README.md

@@ -46,7 +46,6 @@ is distributed under the [ISC license](LICENSE.md).
     - [A transactional lock-free stack](#a-transactional-lock-free-stack)
     - [A transactional lock-free queue](#a-transactional-lock-free-queue)
     - [Composing transactions](#composing-transactions)
-    - [About transactions](#about-transactions)
     - [A transactional lock-free leftist heap](#a-transactional-lock-free-leftist-heap)
     - [A composable Michael-Scott style queue](#a-composable-michael-scott-style-queue)
 - [Designing lock-free algorithms with k-CAS](#designing-lock-free-algorithms-with-k-cas)
@@ -58,6 +57,7 @@ is distributed under the [ISC license](LICENSE.md).
     - [Understanding transactions](#understanding-transactions)
     - [A three-stack lock-free queue](#a-three-stack-lock-free-queue)
     - [A rehashable lock-free hash table](#a-rehashable-lock-free-hash-table)
+  - [Blocking](#blocking)
   - [Beware of torn reads](#beware-of-torn-reads)
 - [Development](#development)
 
@@ -504,25 +504,6 @@ Or transfer elements between different transactional data structures:
 The ability to compose transactions allows algorithms and data-structures to be
 used for a wider variety of purposes.
 
-#### About transactions
-
-The transaction mechanism provided by **kcas** is quite intentionally designed
-to be very simple and efficient. This also means that it cannot provide certain
-features, because adding such features would either add significant dependencies
-or overheads to the otherwise simple and efficient implementation. In
-particular, the transactions provided by **kcas** do not directly provide
-blocking or the ability to wait for changes to shared memory locations before
-retrying a transaction. The way
-[`commit`](https://ocaml-multicore.github.io/kcas/doc/kcas/Kcas/Xt/index.html#val-commit)
-works is that it simply retries the transaction in case it failed. To avoid
-contention, a
-[`backoff`](https://ocaml-multicore.github.io/kcas/doc/kcas/Kcas/Backoff/index.html)
-mechanism is used, but otherwise
-[`commit`](https://ocaml-multicore.github.io/kcas/doc/kcas/Kcas/Xt/index.html#val-commit)
-will essentially perform a
-[busy-wait](https://en.wikipedia.org/wiki/Busy_waiting), which should usually be
-avoided.
-
 #### A transactional lock-free leftist heap
 
 Let's implement something a bit more complicated,
@@ -1434,6 +1415,54 @@ What we have here is a lock-free hash table with rehashing that should not be
 highly prone to starvation. In other respects this is a fairly naive hash table
 implementation. You might want to think about various ways to improve upon it.
 
+### Blocking
+
+Consider the following approach to unconditionally pop an element from a stack:
+
+```ocaml
+# let pop ~xt stack =
+    match Xt.update ~xt stack tl_safe with
+    | [] -> raise Exit
+    | elem :: _ -> elem
+val pop : xt:'a Xt.t -> 'b list Loc.t -> 'b = <fun>
+```
+
+A transaction that attempts to use the above on an empty stack will keep on
+retrying until some other domain pushes an element to the stack. This sort of
+[busy-wait](https://en.wikipedia.org/wiki/Busy_waiting) should generally be
+avoided.
+
+```ocaml
+# let stack () : _ stack = Loc.make ~awaitable:true []
+val stack : unit -> 'a stack = <fun>
+```
+
+```ocaml
+# let input_stack : int stack = stack ()
+val input_stack : int stack = <abstr>
+# let output_stack : int stack = stack ()
+val output_stack : int stack = <abstr>
+```
+
+```ocaml
+# let other = Domain.spawn @@ fun () ->
+    let x = Xt.commit { tx = pop input_stack } in
+    Xt.commit { tx = push output_stack x }
+val other : unit Domain.t = <abstr>
+```
+
+```ocaml
+# Xt.commit { tx = push input_stack 42 }
+- : unit = ()
+# Xt.commit { tx = pop output_stack }
+- : int = 42
+```
+
+```ocaml
+# Domain.join other
+- : unit = ()
+```
+
 ### Beware of torn reads
 
 The algorithm underlying **kcas** ensures that it is not possible to read

src/backoff.mli

@@ -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/kcas.ml

@@ -3,8 +3,16 @@
  * Copyright (c) 2023, Vesa Karvonen <vesa.a.j.k@gmail.com>
  *)
 
+module Trigger = Trigger
 module Backoff = Backoff
 
+module AwaitableId = struct
+  let id = Atomic.make (Int.min_int asr 1)
+  let get_unique () = Atomic.fetch_and_add id 1
+  let is id = id < 0
+  let to_awaiters id = Int.min_int - 1 - id
+end
+
 module Id = struct
   let id = Atomic.make 1
   let get_unique () = Atomic.fetch_and_add id 1
@@ -30,9 +38,11 @@ end = struct
     [@@inline]
 end
 
+type awaiter = [ `Init | `Resumed | `Waiting of unit -> unit ] Atomic.t
 type determined = [ `After | `Before ]
 
-type 'a loc = { state : 'a state Atomic.t; id : int }
+type 'a loc = { state : 'a state Atomic.t; id : int; awaiters : awaiters }
+and awaiters = awaiter list loc
 and 'a state = { mutable before : 'a; mutable after : 'a; mutable casn : casn }
 and cass = CASN : 'a loc * 'a state * cass * cass -> cass | NIL : cass
 and casn = status Atomic.t
@@ -214,12 +224,23 @@ let cas loc before state =
 let inc x = x + 1
 let dec x = x - 1
 
+let make_awaiters id =
+  let state = Atomic.make @@ new_state [] and id = AwaitableId.to_awaiters id in
+  let rec awaiters = { state; id; awaiters } in
+  awaiters
+
 module Loc = struct
   type 'a t = 'a loc
 
-  let make after =
-    { state = Atomic.make @@ new_state after; id = Id.get_unique () }
+  let make ?(awaitable = false) after =
+    let state = Atomic.make @@ new_state after in
+    let id =
+      if awaitable then AwaitableId.get_unique () else Id.get_unique ()
+    in
+    let awaiters = if awaitable then make_awaiters id else Obj.magic () in
+    { state; id; awaiters }
 
+  let is_awaitable loc = AwaitableId.is loc.id
   let get_id loc = loc.id [@@inline]
 
   let get loc =
@@ -391,38 +412,106 @@ module Xt = struct
 
   let call { tx } = tx [@@inline]
 
-  let attempt (mode : Mode.t) tx =
+  let rec take_awaiters ~xt awaiters = function
+    | NIL -> awaiters
+    | CASN (loc, state, l, r) ->
+        let awaiters =
+          if l != NIL then take_awaiters ~xt awaiters l else awaiters
+        in
+        if AwaitableId.is loc.id then
+          let awaiters =
+            if state.before != state.after then
+              match exchange ~xt loc.awaiters [] with
+              | [] -> awaiters
+              | aws -> aws :: awaiters
+            else awaiters
+          in
+          take_awaiters ~xt awaiters r
+        else awaiters
+
+  let resume_awaiters awaiters =
+    awaiters
+    |> List.iter @@ List.iter
+       @@ fun awaiter ->
+       match Atomic.exchange awaiter `Resumed with
+       | `Waiting resume -> resume ()
+       | _ -> ()
+    [@@inline never]
+
+  let resume_awaiters = function
+    | [] -> ()
+    | awaiters -> resume_awaiters awaiters
+    [@@inline]
+
+  let rec add_awaiter ~xt awaiter = function
+    | NIL -> ()
+    | CASN (loc, _, l, r) ->
+        if l != NIL then add_awaiter ~xt awaiter l;
+        if AwaitableId.is loc.id then (
+          modify ~xt loc.awaiters (List.cons awaiter);
+          add_awaiter ~xt awaiter r)
+
+  let rec reset casn = function
+    | NIL -> NIL
+    | CASN (loc, state, l, r) as old ->
+        let l' = reset casn l and r' = reset casn r in
+        if is_cmp casn state then
+          if l == l' && r == r' then old else CASN (loc, state, l', r')
+        else
+          let state' = Atomic.get loc.state in
+          let current = eval state' in
+          if current != state.before then exit () else CASN (loc, state', l, r)
+
+  let rec commit backoff (mode : Mode.t) tx =
     let xt =
       let casn = Atomic.make (mode :> status)
       and cass = NIL
       and post_commit = Action.noop in
       { casn; cass; post_commit }
     in
-    let result = tx ~xt in
-    match xt.cass with
-    | NIL -> Action.run xt.post_commit result
-    | CASN (loc, state, NIL, NIL) ->
-        if is_cmp xt.casn state then Action.run xt.post_commit result
-        else
-          let before = state.before in
-          state.before <- state.after;
-          if cas loc before state then Action.run xt.post_commit result
-          else exit ()
-    | cass ->
-        if determine_for_owner xt.casn cass then
-          Action.run xt.post_commit result
-        else exit ()
-
-  let rec commit backoff mode tx =
-    match attempt mode tx with
-    | result -> result
-    | exception Mode.Interference ->
-        commit (Backoff.once backoff) Mode.lock_free tx
-    | exception Exit -> commit (Backoff.once backoff) mode tx
-
-  let commit ?(backoff = Backoff.default) ?(mode = Mode.obstruction_free) tx =
-    commit backoff mode tx.tx
+    match tx ~xt with
+    | result -> (
+        let awaiters = take_awaiters ~xt [] xt.cass in
+        match xt.cass with
+        | NIL -> Action.run xt.post_commit result
+        | CASN (loc, state, NIL, NIL) ->
+            if is_cmp xt.casn state then Action.run xt.post_commit result
+            else
+              let before = state.before in
+              state.before <- state.after;
+              if cas loc before state then Action.run xt.post_commit result
+              else commit (Backoff.once backoff) mode tx
+        | cass -> (
+            match determine_for_owner xt.casn cass with
+            | true ->
+                resume_awaiters awaiters;
+                Action.run xt.post_commit result
+            | false -> commit (Backoff.once backoff) mode tx
+            | exception Mode.Interference ->
+                commit (Backoff.once backoff) Mode.lock_free tx))
+    | exception Exit -> (
+        match reset xt.casn xt.cass with
+        | cass -> (
+            xt.cass <- cass;
+            let self = Atomic.make `Init in
+            add_awaiter ~xt self cass;
+            if xt.cass == cass then commit (Backoff.once backoff) mode tx
+            else
+              match determine_for_owner xt.casn xt.cass with
+              | true ->
+                  (if Atomic.get self == `Init then
+                     let t = Trigger.prepare_for_await () in
+                     if Atomic.compare_and_set self `Init (`Waiting t.release)
+                     then t.await ());
+                  (* TODO: remove awaiters *)
+                  commit (Backoff.once (Backoff.reset backoff)) mode tx
+              | false -> commit (Backoff.once backoff) mode tx
+              | exception Mode.Interference ->
+                  commit (Backoff.once backoff) Mode.lock_free tx)
+        | exception Exit -> commit (Backoff.once backoff) mode tx)
+
+  let commit ?(backoff = Backoff.default) ?(mode = Mode.obstruction_free) { tx }
+      =
+    commit backoff mode tx
     [@@inline]
-
-  let attempt ?(mode = Mode.lock_free) tx = attempt mode tx.tx [@@inline]
 end

src/kcas.mli

@@ -1,7 +1,7 @@
 (** {1 Auxiliary modules} *)
 
+module Trigger : module type of Trigger
 module Backoff : module type of Backoff
-(** Randomized exponential backoff mechanism. *)
 
 (** {1 Individual locations}
 
@@ -13,9 +13,17 @@ module Loc : sig
   type !'a t
   (** Type of shared memory locations. *)
 
-  val make : 'a -> 'a t
+  val make : ?awaitable:bool -> 'a -> 'a t
   (** [make initial] creates a new shared memory location with the [initial]
-      value. *)
+      value.
+
+      By default a location is considered to be unawaitable to avoid overheads.
+      {!Xt.commit} cannot block to await for changes to unawaitable locations.
+      To make a location awaitable, simply pass [~awaitable:true]. *)
+
+  val is_awaitable : 'a t -> bool
+  (** [is_awaitable r] determines whether the shared memory location [r] is
+      awaitable. *)
 
   val get_id : 'a t -> int
   (** [get_id r] returns the unique id of the shared memory location [r]. *)
@@ -239,24 +247,14 @@ module Xt : sig
   val call : 'a tx -> xt:'x t -> 'a
   (** [call ~xt tx] is equivalent to [tx.Xt.tx ~xt]. *)
 
-  val attempt : ?mode:Mode.t -> 'a tx -> 'a
-  (** [attempt tx] attempts to atomically perform the transaction over shared
-      memory locations recorded by calling [tx] with a fresh explicit
-      transaction log.  If used in {!Mode.obstruction_free} may raise
-      {!Mode.Interference}.  Otherwise either raises [Exit] on failure to commit
-      the transaction or returns the result of the transaction.  The default for
-      [attempt] is {!Mode.lock_free}. *)
-
   val commit : ?backoff:Backoff.t -> ?mode:Mode.t -> 'a tx -> 'a
-  (** [commit tx] repeats [attempt tx] until it does not raise [Exit] or
-      {!Mode.Interference} and then either returns or raises whatever attempt
-      returned or raised.
+  (** [commit tx] repeatedly calls [tx] to record a log of shared memory
+      accesses and attempts to perform them atomically until it succeeds and
+      then returns whatever [tx] returned.  [tx] may raise [Exit] to explicitly
+      request a retry or any other exception than {!Mode.Interference} to abort
+      the transaction.
 
       The default for [commit] is {!Mode.obstruction_free}.  However, after
       enough attempts have failed during the verification step, [commit]
-      switches to {!Mode.lock_free}.
-
-      Note that, aside from using exponential backoff to reduce contention, the
-      transaction mechanism has no way to intelligently wait until shared memory
-      locations are modified by other domains. *)
+      switches to {!Mode.lock_free}. *)
 end

src/kcas_data/kcas_data.ml

@@ -54,6 +54,11 @@ module Hashtbl = Hashtbl
 module Queue = Queue
 module Stack = Stack
 
+(** {1 [Eio] style synchronization primitives} *)
+
+module Promise = Promise
+module Semaphore = Semaphore
+
 (** {1 Utilities} *)
 
 module Accumulator = Accumulator