From 005564929035e8ccbd5dde1eeeff0f5ea2761387 Mon Sep 17 00:00:00 2001
From: Vesa Karvonen <vesa.a.j.k@gmail.com>
Date: Sat, 20 May 2023 20:18:34 +0300
Subject: [PATCH] Add optional `capacity` argument to `Queue.create` and
 `Stack.create`

---
 src/kcas_data/elems.ml               |  11 ++
 src/kcas_data/elems.mli              |   1 +
 src/kcas_data/list_with_capacity.ml  | 121 +++++++++++++++++++
 src/kcas_data/list_with_capacity.mli |  24 ++++
 src/kcas_data/queue.ml               | 167 +++++++++++++++++----------
 src/kcas_data/queue.mli              |  11 +-
 src/kcas_data/queue_intf.ml          |   7 +-
 src/kcas_data/stack.ml               |  77 +++++++-----
 src/kcas_data/stack.mli              |  13 ++-
 src/kcas_data/stack_intf.ml          |   7 +-
 test/kcas_data/queue_test.ml         |  12 ++
 test/kcas_data/stack_test.ml         |  13 +++
 12 files changed, 370 insertions(+), 94 deletions(-)
 create mode 100644 src/kcas_data/list_with_capacity.ml
 create mode 100644 src/kcas_data/list_with_capacity.mli

diff --git a/src/kcas_data/elems.ml b/src/kcas_data/elems.ml
index 0481548b..b3e74fd2 100644
--- a/src/kcas_data/elems.ml
+++ b/src/kcas_data/elems.ml
@@ -41,3 +41,14 @@ let rev_prepend_to_seq t tl =
         | `Reversed t' -> t'
       in
       prepend_to_seq t tl ()
+
+let rec of_list_rev tl length = function
+  | [] -> tl
+  | x :: xs ->
+      let length = length + 1 in
+      of_list_rev { value = x; tl; length } length xs
+
+let of_list_rev = function
+  | [] -> empty
+  | x :: xs -> of_list_rev (singleton x) 1 xs
+  [@@inline]
diff --git a/src/kcas_data/elems.mli b/src/kcas_data/elems.mli
index 2bf19d3a..084d61e2 100644
--- a/src/kcas_data/elems.mli
+++ b/src/kcas_data/elems.mli
@@ -18,3 +18,4 @@ val prepend_to_seq : 'a t -> 'a Seq.t -> 'a Seq.t
 val to_seq : 'a t -> 'a Seq.t
 val of_seq_rev : 'a Seq.t -> 'a t
 val rev_prepend_to_seq : 'a t -> 'a Seq.t -> 'a Seq.t
+val of_list_rev : 'a list -> 'a t
diff --git a/src/kcas_data/list_with_capacity.ml b/src/kcas_data/list_with_capacity.ml
new file mode 100644
index 00000000..421c5823
--- /dev/null
+++ b/src/kcas_data/list_with_capacity.ml
@@ -0,0 +1,121 @@
+open Kcas
+
+type 'a t = { capacity : int; length : int; list : 'a list; limit : int }
+
+let empty_unlimited =
+  { capacity = Int.max_int; length = 0; list = []; limit = Int.max_int }
+
+let make_empty ~capacity =
+  if capacity = Int.max_int then empty_unlimited
+  else { capacity; length = 0; list = []; limit = capacity }
+  [@@inline]
+
+let make ~capacity ~length ~list ~limit = { capacity; length; list; limit }
+  [@@inline]
+
+let to_rev_elems t = Elems.of_list_rev t.list [@@inline]
+let is_empty t = t.length = 0 [@@inline]
+let length t = t.length [@@inline]
+let capacity t = t.capacity [@@inline]
+let limit t = t.limit [@@inline]
+let list t = t.list [@@inline]
+
+let tl_safe = function
+  | { list = []; _ } as t -> t
+  | { capacity; length; list = _ :: list; _ } as t ->
+      let limit = if capacity = Int.max_int then capacity else t.limit in
+      { capacity; length = length - 1; list; limit }
+  [@@inline]
+
+let tl_or_retry = function
+  | { list = []; _ } -> Retry.later ()
+  | { capacity; length; list = _ :: list; _ } as t ->
+      let limit = if capacity = Int.max_int then capacity else t.limit in
+      { capacity; length = length - 1; list; limit }
+  [@@inline]
+
+let hd_opt t = match t.list with [] -> None | x :: _ -> Some x [@@inline]
+
+let hd_or_retry t = match t.list with [] -> Retry.later () | x :: _ -> x
+  [@@inline]
+
+let hd_unsafe t = List.hd t.list [@@inline]
+
+let cons_safe x ({ capacity; _ } as t) =
+  if capacity = Int.max_int then
+    let { length; list; _ } = t in
+    { capacity; length = length + 1; list = x :: list; limit = capacity }
+  else
+    let { length; limit; _ } = t in
+    if length < limit then
+      let { list; _ } = t in
+      { capacity; length = length + 1; list = x :: list; limit }
+    else t
+  [@@inline]
+
+let cons_or_retry x ({ capacity; _ } as t) =
+  if capacity = Int.max_int then
+    let { length; list; _ } = t in
+    { capacity; length = length + 1; list = x :: list; limit = capacity }
+  else
+    let { length; limit; _ } = t in
+    if length < limit then
+      let { list; _ } = t in
+      { capacity; length = length + 1; list = x :: list; limit }
+    else Retry.later ()
+  [@@inline]
+
+let move ({ capacity; _ } as t) =
+  if capacity = Int.max_int then empty_unlimited
+  else
+    let { length; _ } = t in
+    if length = 0 then t
+    else
+      let { limit; _ } = t in
+      { capacity; length = 0; list = []; limit = limit - length }
+  [@@inline]
+
+let move_last ({ capacity; _ } as t) =
+  if capacity = Int.max_int then empty_unlimited
+  else
+    let { length; _ } = t in
+    let limit = capacity - length in
+    if length = 0 && t.limit = limit then t
+    else { capacity; length = 0; list = []; limit }
+
+let clear ({ capacity; _ } as t) =
+  if capacity = Int.max_int then empty_unlimited
+  else if t.length = 0 && t.limit = capacity then t
+  else make_empty ~capacity
+  [@@inline]
+
+let rec prepend_to_seq xs tl =
+  match xs with
+  | [] -> tl
+  | x :: xs -> fun () -> Seq.Cons (x, prepend_to_seq xs tl)
+
+let to_seq { list; _ } = prepend_to_seq list Seq.empty
+
+let rev_prepend_to_seq { length; list; _ } tl =
+  if length <= 1 then prepend_to_seq list tl
+  else
+    let t = ref (`Original list) in
+    fun () ->
+      let t =
+        match !t with
+        | `Original t' ->
+            (* This is domain safe as the result is always equivalent. *)
+            let t' = List.rev t' in
+            t := `Reversed t';
+            t'
+        | `Reversed t' -> t'
+      in
+      prepend_to_seq t tl ()
+
+let of_list ?(capacity = Int.max_int) list =
+  let length = List.length list in
+  let limit = Int.min 0 (capacity - length) in
+  { capacity; length; list; limit }
+
+let of_seq_rev ?capacity xs =
+  of_list ?capacity (Seq.fold_left (fun xs x -> x :: xs) [] xs)
diff --git a/src/kcas_data/list_with_capacity.mli b/src/kcas_data/list_with_capacity.mli
new file mode 100644
index 00000000..ad427009
--- /dev/null
+++ b/src/kcas_data/list_with_capacity.mli
@@ -0,0 +1,24 @@
+type !'a t
+
+val empty_unlimited : 'a t
+val make_empty : capacity:int -> 'a t
+val make : capacity:int -> length:int -> list:'a list -> limit:int -> 'a t
+val is_empty : 'a t -> bool
+val length : 'a t -> int
+val capacity : 'a t -> int
+val limit : 'a t -> int
+val list : 'a t -> 'a list
+val cons_safe : 'a -> 'a t -> 'a t
+val cons_or_retry : 'a -> 'a t -> 'a t
+val move : 'a t -> 'a t
+val move_last : 'a t -> 'a t
+val clear : 'a t -> 'a t
+val to_rev_elems : 'a t -> 'a Elems.t
+val to_seq : 'a t -> 'a Seq.t
+val rev_prepend_to_seq : 'a t -> 'a Seq.t -> 'a Seq.t
+val of_seq_rev : ?capacity:int -> 'a Seq.t -> 'a t
+val tl_safe : 'a t -> 'a t
+val tl_or_retry : 'a t -> 'a t
+val hd_opt : 'a t -> 'a option
+val hd_or_retry : 'a t -> 'a
+val hd_unsafe : 'a t -> 'a
diff --git a/src/kcas_data/queue.ml b/src/kcas_data/queue.ml
index 911ab85e..e7102e24 100644
--- a/src/kcas_data/queue.ml
+++ b/src/kcas_data/queue.ml
@@ -2,127 +2,172 @@ open Kcas
 
 type 'a t = {
   front : 'a Elems.t Loc.t;
-  middle : 'a Elems.t Loc.t;
-  back : 'a Elems.t Loc.t;
+  back : 'a List_with_capacity.t Loc.t;
+  middle : 'a List_with_capacity.t Loc.t;
 }
 
-let alloc ~front ~middle ~back =
+let alloc ~front ~back ~middle =
   (* We allocate locations in specific order to make most efficient use of the
      splay-tree based transaction log. *)
   let front = Loc.make front
-  and middle = Loc.make middle
-  and back = Loc.make back in
+  and back = Loc.make back
+  and middle = Loc.make middle in
   { back; middle; front }
 
-let create () = alloc ~front:Elems.empty ~middle:Elems.empty ~back:Elems.empty
+let create ?(capacity = Int.max_int) () =
+  if capacity < 0 then invalid_arg "Queue.create: capacity must be non-negative";
+  let back = List_with_capacity.make_empty ~capacity in
+  alloc ~front:Elems.empty ~back ~middle:List_with_capacity.empty_unlimited
 
 let copy q =
-  let tx ~xt = (Xt.get ~xt q.front, Xt.get ~xt q.middle, Xt.get ~xt q.back) in
-  let front, middle, back = Xt.commit { tx } in
-  alloc ~front ~middle ~back
+  let tx ~xt = (Xt.get ~xt q.front, Xt.get ~xt q.back, Xt.get ~xt q.middle) in
+  let front, back, middle = Xt.commit { tx } in
+  alloc ~front ~back ~middle
 
 module Xt = struct
-  let is_empty ~xt { back; middle; front } =
+  let is_empty ~xt q =
     (* We access locations in order of allocation to make most efficient use of
        the splay-tree based transaction log. *)
-    Xt.get ~xt front == Elems.empty
-    && Xt.get ~xt middle == Elems.empty
-    && Xt.get ~xt back == Elems.empty
-
-  let length ~xt { back; middle; front } =
-    Elems.length (Xt.get ~xt front)
-    + Elems.length (Xt.get ~xt middle)
-    + Elems.length (Xt.get ~xt back)
-
-  let add ~xt x q = Xt.unsafe_modify ~xt q.back @@ Elems.cons x
+    Xt.get ~xt q.front == Elems.empty
+    && List_with_capacity.is_empty (Xt.get ~xt q.back)
+    && Xt.get ~xt q.middle == List_with_capacity.empty_unlimited
+
+  let length ~xt q =
+    Elems.length (Xt.get ~xt q.front)
+    + List_with_capacity.length (Xt.get ~xt q.back)
+    + List_with_capacity.length (Xt.get ~xt q.middle)
+
+  let try_add ~xt x q =
+    let lwc = Xt.unsafe_update ~xt q.back (List_with_capacity.cons_safe x) in
+    let capacity = List_with_capacity.capacity lwc in
+    capacity = Int.max_int
+    ||
+    let back_length = List_with_capacity.length lwc in
+    back_length < List_with_capacity.limit lwc
+    ||
+    let other_length =
+      List_with_capacity.length (Xt.get ~xt q.middle)
+      + Elems.length (Xt.get ~xt q.front)
+    in
+    let limit = capacity - other_length in
+    back_length < limit
+    &&
+    (Xt.set ~xt q.back
+       (List_with_capacity.make ~capacity ~length:(back_length + 1)
+          ~list:(x :: List_with_capacity.list lwc)
+          ~limit);
+     true)
+
+  let add ~xt x q = Retry.unless (try_add ~xt x q)
   let push = add
 
   (** Cooperative helper to move elems from back to middle. *)
-  let back_to_middle ~middle ~back =
+  let back_to_middle ~back ~middle =
     let tx ~xt =
-      let xs = Xt.exchange ~xt back Elems.empty in
-      if xs == Elems.empty || Xt.exchange ~xt middle xs != Elems.empty then
-        raise Exit
+      let xs = Xt.unsafe_update ~xt back List_with_capacity.move in
+      if
+        List_with_capacity.length xs = 0
+        || Xt.exchange ~xt middle xs != List_with_capacity.empty_unlimited
+      then raise Exit
     in
     try Xt.commit { tx } with Exit -> ()
 
-  let take_opt_finish ~xt front elems =
-    let elems = Elems.rev elems in
+  let take_opt_finish ~xt front lwc =
+    let elems = List_with_capacity.to_rev_elems lwc in
     Xt.set ~xt front (Elems.tl_safe elems);
     Elems.hd_opt elems
 
-  let take_opt ~xt { back; middle; front } =
+  let take_opt ~xt { front; back; middle } =
     let elems = Xt.unsafe_update ~xt front Elems.tl_safe in
     if elems != Elems.empty then Elems.hd_opt elems
     else (
       if not (Xt.is_in_log ~xt middle || Xt.is_in_log ~xt back) then
-        back_to_middle ~middle ~back;
-      let elems = Xt.exchange ~xt middle Elems.empty in
-      if elems != Elems.empty then take_opt_finish ~xt front elems
+        back_to_middle ~back ~middle;
+      let lwc = Xt.exchange ~xt middle List_with_capacity.empty_unlimited in
+      if lwc != List_with_capacity.empty_unlimited then
+        take_opt_finish ~xt front lwc
       else
-        let elems = Xt.exchange ~xt back Elems.empty in
-        if elems != Elems.empty then take_opt_finish ~xt front elems else None)
+        let lwc = Xt.unsafe_update ~xt back List_with_capacity.move_last in
+        if List_with_capacity.length lwc <> 0 then take_opt_finish ~xt front lwc
+        else None)
 
   let take_blocking ~xt q = Xt.to_blocking ~xt (take_opt q)
 
-  let peek_opt_finish ~xt front elems =
-    let elems = Elems.rev elems in
+  let peek_opt_finish ~xt front lwc =
+    let elems = List_with_capacity.to_rev_elems lwc in
     Xt.set ~xt front elems;
     Elems.hd_opt elems
 
-  let peek_opt ~xt { back; middle; front } =
+  let peek_opt ~xt { front; back; middle } =
     let elems = Xt.get ~xt front in
     if elems != Elems.empty then Elems.hd_opt elems
     else (
       if not (Xt.is_in_log ~xt middle || Xt.is_in_log ~xt back) then
-        back_to_middle ~middle ~back;
-      let elems = Xt.exchange ~xt middle Elems.empty in
-      if elems != Elems.empty then peek_opt_finish ~xt front elems
+        back_to_middle ~back ~middle;
+      let lwc = Xt.exchange ~xt middle List_with_capacity.empty_unlimited in
+      if lwc != List_with_capacity.empty_unlimited then
+        peek_opt_finish ~xt front lwc
       else
-        let elems = Xt.exchange ~xt back Elems.empty in
-        if elems != Elems.empty then peek_opt_finish ~xt front elems else None)
+        let lwc = Xt.unsafe_update ~xt back List_with_capacity.move_last in
+        if List_with_capacity.length lwc <> 0 then peek_opt_finish ~xt front lwc
+        else None)
 
   let peek_blocking ~xt q = Xt.to_blocking ~xt (peek_opt q)
 
-  let clear ~xt { back; middle; front } =
-    Xt.set ~xt front Elems.empty;
-    Xt.set ~xt middle Elems.empty;
-    Xt.set ~xt back Elems.empty
+  let clear ~xt q =
+    Xt.set ~xt q.front Elems.empty;
+    Xt.unsafe_modify ~xt q.back List_with_capacity.clear;
+    Xt.set ~xt q.middle List_with_capacity.empty_unlimited
 
   let swap ~xt q1 q2 =
     let front = Xt.get ~xt q1.front
-    and middle = Xt.get ~xt q1.middle
-    and back = Xt.get ~xt q1.back in
+    and back = Xt.get ~xt q1.back
+    and middle = Xt.get ~xt q1.middle in
     let front = Xt.exchange ~xt q2.front front
-    and middle = Xt.exchange ~xt q2.middle middle
-    and back = Xt.exchange ~xt q2.back back in
+    and back = Xt.exchange ~xt q2.back back
+    and middle = Xt.exchange ~xt q2.middle middle in
     Xt.set ~xt q1.front front;
-    Xt.set ~xt q1.middle middle;
-    Xt.set ~xt q1.back back
+    Xt.set ~xt q1.back back;
+    Xt.set ~xt q1.middle middle
 
   let seq_of ~front ~middle ~back =
     (* Sequence construction is lazy, so this function is O(1). *)
     Seq.empty
-    |> Elems.rev_prepend_to_seq back
-    |> Elems.rev_prepend_to_seq middle
+    |> List_with_capacity.rev_prepend_to_seq back
+    |> List_with_capacity.rev_prepend_to_seq middle
     |> Elems.prepend_to_seq front
 
-  let to_seq ~xt { front; middle; back } =
-    let front = Xt.get ~xt front
-    and middle = Xt.get ~xt middle
-    and back = Xt.get ~xt back in
+  let to_seq ~xt q =
+    let front = Xt.get ~xt q.front
+    and back = Xt.get ~xt q.back
+    and middle = Xt.get ~xt q.middle in
     seq_of ~front ~middle ~back
 
-  let take_all ~xt { front; middle; back } =
-    let front = Xt.exchange ~xt front Elems.empty
-    and middle = Xt.exchange ~xt middle Elems.empty
-    and back = Xt.exchange ~xt back Elems.empty in
+  let take_all ~xt q =
+    let front = Xt.exchange ~xt q.front Elems.empty
+    and back = Xt.unsafe_update ~xt q.back List_with_capacity.clear
+    and middle = Xt.exchange ~xt q.middle List_with_capacity.empty_unlimited in
     seq_of ~front ~middle ~back
 end
 
 let is_empty q = Kcas.Xt.commit { tx = Xt.is_empty q }
 let length q = Kcas.Xt.commit { tx = Xt.length q }
-let add x q = Loc.modify q.back @@ Elems.cons x
+
+let try_add x q =
+  let lwc = Loc.update q.back (List_with_capacity.cons_safe x) in
+  let capacity = List_with_capacity.capacity lwc in
+  capacity = Int.max_int
+  ||
+  let back_length = List_with_capacity.length lwc in
+  back_length < List_with_capacity.limit lwc
+  || Kcas.Xt.commit { tx = Xt.try_add x q }
+
+let add x q =
+  let lwc = Loc.update q.back (List_with_capacity.cons_safe x) in
+  if List_with_capacity.capacity lwc <> Int.max_int then
+    if List_with_capacity.length lwc = List_with_capacity.limit lwc then
+      Kcas.Xt.commit { tx = Xt.add x q }
+
 let push = add
 
 let take_opt q =
diff --git a/src/kcas_data/queue.mli b/src/kcas_data/queue.mli
index f4dd49a9..c4740d5c 100644
--- a/src/kcas_data/queue.mli
+++ b/src/kcas_data/queue.mli
@@ -19,8 +19,15 @@ type !'a t
 exception Empty
 (** Raised when {!take} or {!peek} is applied to an empty queue. *)
 
-val create : unit -> 'a t
-(** [create ()] returns a new empty queue. *)
+val create : ?capacity:int -> unit -> 'a t
+(** [create ()] returns a new empty queue.
+
+    The optional [capacity] can be used to specify the maximum number of
+    elements that may be stored in the queue at any point.
+
+    {b WARNING}: A [capacity] of [0] is allowed, but it means that no elements
+    can be passed through the queue.  In other words, a zero capacity queue is
+    effectively closed. *)
 
 val copy : 'a t -> 'a t
 (** [copy q] returns a copy of the queue [q]. *)
diff --git a/src/kcas_data/queue_intf.ml b/src/kcas_data/queue_intf.ml
index dfebb3dd..5fdb3a99 100644
--- a/src/kcas_data/queue_intf.ml
+++ b/src/kcas_data/queue_intf.ml
@@ -22,11 +22,16 @@ module type Ops = sig
       and modifications of the queue have no effect on the sequence. *)
 
   val add : ('x, 'a -> 'a t -> unit) fn
-  (** [add x q] adds the element [x] at the end of the queue [q]. *)
+  (** [add x q] adds the element [x] at the end of the queue [q], or blocks
+      waiting until the queue is no longer full. *)
 
   val push : ('x, 'a -> 'a t -> unit) fn
   (** [push] is a synonym for {!add}. *)
 
+  val try_add : ('x, 'a -> 'a t -> bool) fn
+  (** [try_add x q] tries to add the element [x] to the end of the queue [q] and
+      returns [true] on success or [false] when the queue is full. *)
+
   val peek_opt : ('x, 'a t -> 'a option) fn
   (** [peek_opt q] returns the first element in queue [q], without removing it
       from the queue, or returns [None] if the queue is empty. *)
diff --git a/src/kcas_data/stack.ml b/src/kcas_data/stack.ml
index 2335732e..d6242ef2 100644
--- a/src/kcas_data/stack.ml
+++ b/src/kcas_data/stack.ml
@@ -1,41 +1,66 @@
 open Kcas
 
-type 'a t = 'a Elems.t Loc.t
+type 'a t = 'a List_with_capacity.t Loc.t
 
-let create () = Loc.make Elems.empty
-let copy s = Loc.make @@ Loc.get s
-let of_seq xs = Loc.make (Elems.of_seq_rev xs)
+let create ?(capacity = Int.max_int) () =
+  Loc.make (List_with_capacity.make_empty ~capacity)
+
+let copy s = Loc.make (Loc.get s)
+
+let of_seq ?(capacity = Int.max_int) xs =
+  Loc.make (List_with_capacity.of_seq_rev ~capacity xs)
 
 module Xt = struct
-  let length ~xt s = Xt.get ~xt s |> Elems.length
-  let is_empty ~xt s = Xt.get ~xt s == Elems.empty
-  let push ~xt x s = Xt.unsafe_modify ~xt s @@ Elems.cons x
-  let pop_opt ~xt s = Xt.unsafe_update ~xt s Elems.tl_safe |> Elems.hd_opt
-  let pop_all ~xt s = Elems.to_seq @@ Xt.exchange ~xt s Elems.empty
+  let length ~xt s = List_with_capacity.length (Xt.get ~xt s)
+  let is_empty ~xt s = List_with_capacity.is_empty (Xt.get ~xt s)
+  let push ~xt x s = Xt.unsafe_modify ~xt s (List_with_capacity.cons_or_retry x)
+
+  let try_push ~xt x s =
+    let lwc = Xt.unsafe_update ~xt s (List_with_capacity.cons_safe x) in
+    List_with_capacity.length lwc < List_with_capacity.capacity lwc
+
+  let pop_opt ~xt s =
+    Xt.unsafe_update ~xt s List_with_capacity.tl_safe
+    |> List_with_capacity.hd_opt
+
+  let pop_all ~xt s =
+    List_with_capacity.to_seq (Xt.unsafe_update ~xt s List_with_capacity.clear)
 
   let pop_blocking ~xt s =
-    Xt.unsafe_update ~xt s Elems.tl_safe |> Elems.hd_or_retry
+    Xt.unsafe_update ~xt s List_with_capacity.tl_or_retry
+    |> List_with_capacity.hd_unsafe
 
-  let top_opt ~xt s = Xt.get ~xt s |> Elems.hd_opt
-  let top_blocking ~xt s = Xt.get ~xt s |> Elems.hd_or_retry
-  let clear ~xt s = Xt.set ~xt s Elems.empty
+  let top_opt ~xt s = List_with_capacity.hd_opt (Xt.get ~xt s)
+  let top_blocking ~xt s = List_with_capacity.hd_or_retry (Xt.get ~xt s)
+  let clear ~xt s = Xt.unsafe_modify ~xt s List_with_capacity.clear
   let swap ~xt s1 s2 = Xt.swap ~xt s1 s2
-  let to_seq ~xt s = Elems.to_seq @@ Xt.get ~xt s
+  let to_seq ~xt s = List_with_capacity.to_seq (Xt.get ~xt s)
 end
 
-let length s = Loc.get s |> Elems.length
-let is_empty s = Loc.get s == Elems.empty
-let push x s = Loc.modify s @@ Elems.cons x
-let pop_opt s = Loc.update s Elems.tl_safe |> Elems.hd_opt
-let pop_all s = Loc.exchange s Elems.empty |> Elems.to_seq
-let pop_blocking s = Loc.update s Elems.tl_or_retry |> Elems.hd_unsafe
-let top_opt s = Loc.get s |> Elems.hd_opt
-let top_blocking s = Loc.get_as Elems.hd_or_retry s
-let clear s = Loc.set s Elems.empty
+let length s = List_with_capacity.length (Loc.get s)
+let is_empty s = List_with_capacity.is_empty (Loc.get s)
+let push x s = Loc.modify s (List_with_capacity.cons_or_retry x)
+
+let try_push x s =
+  let lwc = Loc.update s (List_with_capacity.cons_safe x) in
+  List_with_capacity.length lwc < List_with_capacity.capacity lwc
+
+let pop_opt s =
+  List_with_capacity.hd_opt (Loc.update s List_with_capacity.tl_safe)
+
+let pop_all s =
+  List_with_capacity.to_seq (Loc.update s List_with_capacity.clear)
+
+let pop_blocking s =
+  List_with_capacity.hd_unsafe (Loc.update s List_with_capacity.tl_or_retry)
+
+let top_opt s = List_with_capacity.hd_opt (Loc.get s)
+let top_blocking s = Loc.get_as List_with_capacity.hd_or_retry s
+let clear s = Loc.modify s List_with_capacity.clear
 let swap s1 s2 = Kcas.Xt.commit { tx = Kcas.Xt.swap s1 s2 }
-let to_seq s = Elems.to_seq @@ Loc.get s
-let iter f s = Elems.iter f @@ Loc.get s
-let fold f a s = Elems.fold f a @@ Loc.get s
+let to_seq s = List_with_capacity.to_seq (Loc.get s)
+let iter f s = List.iter f (List_with_capacity.list (Loc.get s))
+let fold f a s = List.fold_left f a (List_with_capacity.list (Loc.get s))
 
 exception Empty
 
diff --git a/src/kcas_data/stack.mli b/src/kcas_data/stack.mli
index d84fe9a6..82af844c 100644
--- a/src/kcas_data/stack.mli
+++ b/src/kcas_data/stack.mli
@@ -18,13 +18,20 @@ type !'a t
 exception Empty
 (** Raised when {!pop} or {!top} is applied to an empty stack. *)
 
-val create : unit -> 'a t
-(** [create ()] returns a new empty stack. *)
+val create : ?capacity:int -> unit -> 'a t
+(** [create ()] returns a new empty stack.
+
+    The optional [capacity] can be used to specify the maximum number of
+    elements that may be stored in the stack at any point.
+
+    {b WARNING}: A [capacity] of [0] is allowed, but it means that no elements
+    can be passed through the stack.  In other words, a zero capacity stack is
+    effectively closed. *)
 
 val copy : 'a t -> 'a t
 (** [copy s] returns a copy of the stack [s]. *)
 
-val of_seq : 'a Seq.t -> 'a t
+val of_seq : ?capacity:int -> 'a Seq.t -> 'a t
 (** [of_seq xs] creates a stack from the sequence [xs]. *)
 
 (** {1 Compositional interface} *)
diff --git a/src/kcas_data/stack_intf.ml b/src/kcas_data/stack_intf.ml
index 0a54c365..8f237ff4 100644
--- a/src/kcas_data/stack_intf.ml
+++ b/src/kcas_data/stack_intf.ml
@@ -22,7 +22,12 @@ module type Ops = sig
       and modifications of the stack have no effect on the sequence. *)
 
   val push : ('x, 'a -> 'a t -> unit) fn
-  (** [push x s] adds the element [x] to the top of the stack [s]. *)
+  (** [push x s] adds the element [x] to the top of the stack [s], or blocks
+      waiting until the stack is no longer full. *)
+
+  val try_push : ('x, 'a -> 'a t -> bool) fn
+  (** [try_push x s] tries to add the element [x] to the top of the stack [s]
+      and returns [true] on success or [false] when the stack is full. *)
 
   val pop_opt : ('x, 'a t -> 'a option) fn
   (** [pop_opt s] removes and returns the topmost element of the stack [s], or
diff --git a/test/kcas_data/queue_test.ml b/test/kcas_data/queue_test.ml
index a6a59b09..d1b834cd 100644
--- a/test/kcas_data/queue_test.ml
+++ b/test/kcas_data/queue_test.ml
@@ -1,6 +1,18 @@
 open Kcas
 open Kcas_data
 
+let () =
+  let s = Queue.create ~capacity:2 () in
+  assert (Queue.try_add 101 s);
+  assert (Queue.try_add 42 s);
+  assert (not (Queue.try_add 15 s));
+  let d = Domain.spawn @@ fun () -> Queue.add 5 s in
+  assert (Queue.take s = 101);
+  let tx ~xt = Retry.unless (Queue.Xt.take_blocking ~xt s = 42) in
+  Xt.commit { tx };
+  assert (Queue.take_blocking s = 5);
+  Domain.join d
+
 let () =
   let q = Queue.create () in
   Queue.add 101 q;
diff --git a/test/kcas_data/stack_test.ml b/test/kcas_data/stack_test.ml
index fd9b2ff2..b36842de 100644
--- a/test/kcas_data/stack_test.ml
+++ b/test/kcas_data/stack_test.ml
@@ -1,5 +1,18 @@
+open Kcas
 open Kcas_data
 
+let () =
+  let s = Stack.create ~capacity:2 () in
+  assert (Stack.try_push 101 s);
+  assert (Stack.try_push 42 s);
+  assert (not (Stack.try_push 15 s));
+  let d = Domain.spawn @@ fun () -> Stack.push 5 s in
+  assert (Stack.pop s = 42);
+  let tx ~xt = Retry.unless (Stack.Xt.pop_blocking ~xt s = 5) in
+  Xt.commit { tx };
+  assert (Stack.pop s = 101);
+  Domain.join d
+
 let () =
   let s = Stack.create () in
   assert (Stack.length s = 0);