package kcas_data

  1. Overview
  2. Docs

Source file queue.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
open Kcas

type 'a t = {
  front : 'a Elems.t Loc.t;
  middle : 'a Elems.t Loc.t;
  back : 'a Elems.t Loc.t;
}

let alloc ~front ~middle ~back =
  (* 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
  { back; middle; front }

let create () = alloc ~front:Elems.empty ~middle:Elems.empty ~back:Elems.empty

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

module Xt = struct
  let is_empty ~xt { back; middle; front } =
    (* 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
  let push = add

  (** Cooperative helper to move elems from back to middle. *)
  let back_to_middle ~middle ~back =
    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
    in
    try Xt.commit { tx } with Exit -> ()

  let take_opt_finish ~xt front elems =
    let elems = Elems.rev elems in
    Xt.set ~xt front (Elems.tl_safe elems);
    Elems.hd_opt elems

  let take_opt ~xt { back; middle; front } =
    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
      else
        let elems = Xt.exchange ~xt back Elems.empty in
        if elems != Elems.empty then take_opt_finish ~xt front elems 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
    Xt.set ~xt front elems;
    Elems.hd_opt elems

  let peek_opt ~xt { back; middle; front } =
    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
      else
        let elems = Xt.exchange ~xt back Elems.empty in
        if elems != Elems.empty then peek_opt_finish ~xt front elems 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 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
    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
    Xt.set ~xt q1.front front;
    Xt.set ~xt q1.middle middle;
    Xt.set ~xt q1.back back

  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
    |> 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
    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
    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 push = add

let take_opt q =
  match Loc.update q.front Elems.tl_safe |> Elems.hd_opt with
  | None -> Kcas.Xt.commit { tx = Xt.take_opt q }
  | some -> some

let take_blocking q = Kcas.Xt.commit { tx = Xt.take_blocking q }
let take_all q = Kcas.Xt.commit { tx = Xt.take_all q }

let peek_opt q =
  match Loc.get q.front |> Elems.hd_opt with
  | None -> Kcas.Xt.commit { tx = Xt.peek_opt q }
  | some -> some

let peek_blocking q = Kcas.Xt.commit { tx = Xt.peek_blocking q }
let clear q = Kcas.Xt.commit { tx = Xt.clear q }
let swap q1 q2 = Kcas.Xt.commit { tx = Xt.swap q1 q2 }
let to_seq q = Kcas.Xt.commit { tx = Xt.to_seq q }
let iter f q = Seq.iter f @@ to_seq q
let fold f a q = Seq.fold_left f a @@ to_seq q

exception Empty

let of_option = function None -> raise Empty | Some value -> value [@@inline]
let peek s = peek_opt s |> of_option
let top = peek
let take s = take_opt s |> of_option