WIP: Add blocking

Author: polytypic

README.md

@@ -59,6 +59,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)
 
@@ -523,25 +524,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/Tx/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/Tx/index.html#val-commit)
-will essentially perform a
-[busy-wait](https://en.wikipedia.org/wiki/Busy_waiting), which should usually be
-avoided.
-
 ### Programming with explicit transaction log passing
 
 The [`Xt`](https://ocaml-multicore.github.io/kcas/doc/kcas/Kcas/Xt/index.html)
@@ -1494,6 +1476,89 @@ 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 scheduler : Xt.scheduler =
+    let open struct
+      type t = {
+        mutex : Mutex.t;
+        condition : Condition.t;
+        mutable signaled : bool;
+      }
+    end in
+    let key = Domain.DLS.new_key @@ fun () -> {
+      mutex = Mutex.create ();
+      condition = Condition.create ();
+      signaled = false;
+    } in
+    fun suspend ->
+      let t = Domain.DLS.get key in
+      suspend (fun () ->
+        Mutex.lock t.mutex;
+        t.signaled <- true;
+        Mutex.unlock t.mutex;
+        Condition.broadcast t.condition);
+      Mutex.lock t.mutex;
+      while not t.signaled do
+        Condition.wait t.condition t.mutex
+      done;
+      t.signaled <- false;
+      Mutex.unlock t.mutex
+val scheduler : Xt.scheduler = <fun>
+```
+
+```ocaml
+# let push ~xt stack elem = Xt.of_tx ~xt (push stack elem)
+val push : xt:'a Xt.t -> 'b list Loc.t -> 'b -> unit = <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 ~scheduler { 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 ~scheduler { 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/kcas.ml

@@ -5,6 +5,13 @@
 
 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 +37,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 +223,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 =
@@ -539,6 +559,37 @@ module Xt = struct
 
   let call { tx } = tx [@@inline]
 
+  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 attempt (mode : Mode.t) tx =
     let xt =
       let casn = Atomic.make (mode :> status)
@@ -547,6 +598,7 @@ module Xt = struct
       { casn; cass; post_commit }
     in
     let result = tx ~xt in
+    let awaiters = take_awaiters ~xt [] xt.cass in
     match xt.cass with
     | NIL -> Action.run xt.post_commit result
     | CASN (loc, state, NIL, NIL) ->
@@ -557,22 +609,97 @@ module Xt = struct
           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
+        if determine_for_owner xt.casn cass then (
+          resume_awaiters awaiters;
+          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 attempt ?(mode = Mode.lock_free) tx = attempt mode tx.tx [@@inline]
 
-  let commit ?(backoff = Backoff.default) ?(mode = Mode.obstruction_free) tx =
-    commit backoff mode tx.tx
-    [@@inline]
+  type scheduler = ((unit -> unit) -> unit) -> unit
+
+  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 attempt ?(mode = Mode.lock_free) tx = attempt mode tx.tx [@@inline]
+  let rec commit backoff (mode : Mode.t) scheduler_opt tx =
+    let xt =
+      let casn = Atomic.make (mode :> status)
+      and cass = NIL
+      and post_commit = Action.noop in
+      { casn; cass; post_commit }
+    in
+    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 scheduler_opt 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 scheduler_opt tx
+            | exception Mode.Interference ->
+                commit (Backoff.once backoff) Mode.lock_free scheduler_opt tx))
+    | exception Exit -> (
+        match scheduler_opt with
+        | None -> commit (Backoff.once backoff) mode scheduler_opt tx
+        | Some scheduler -> (
+            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 scheduler_opt tx
+                else
+                  match determine_for_owner xt.casn xt.cass with
+                  | true ->
+                      if Atomic.get self == `Init then
+                        scheduler (fun resume ->
+                            if
+                              not
+                                (Atomic.compare_and_set self `Init
+                                   (`Waiting resume))
+                            then resume ());
+                      (* TODO: remove awaiters *)
+                      commit
+                        (Backoff.once (Backoff.reset backoff))
+                        mode scheduler_opt tx
+                  | false -> commit (Backoff.once backoff) mode scheduler_opt tx
+                  | exception Mode.Interference ->
+                      commit (Backoff.once backoff) Mode.lock_free scheduler_opt
+                        tx)
+            | exception Exit ->
+                commit (Backoff.once backoff) mode scheduler_opt tx))
+
+  let commit ?(backoff = Backoff.default) ?(mode = Mode.obstruction_free)
+      ?scheduler { tx } =
+    commit backoff mode scheduler tx
+    [@@inline]
 
   let of_tx tx ~xt =
     let (_, cass, post_commit), x = tx (xt.casn, xt.cass, xt.post_commit) in

src/kcas.mli

@@ -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]. *)
@@ -494,7 +502,13 @@ module Xt : sig
       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
+  type scheduler = ((unit -> unit) -> unit) -> unit
+  (** Type of a scheduler that can be called to suspend execution of the current
+      domain or fiber until the resume function given by the scheduler is
+      called. *)
+
+  val commit :
+    ?backoff:Backoff.t -> ?mode:Mode.t -> ?scheduler:scheduler -> '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.
@@ -505,7 +519,8 @@ module Xt : sig
 
       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. *)
+      locations are modified by other domains unless a suitable {!scheduler} is
+      given. *)
 
   (** {1 Conversions} *)