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
open! Base
open! Import
(** See [Adaptive_backoff.once] *)
let log_scale = 11
type 'a t = 'a Modes.Portended.t list Awaitable.t
let create ?padded () = Awaitable.make ?padded []
let sexp_of_t (type a : value mod contended) sexp_of_a (t : a t) =
Awaitable.get t |> List.map ~f:(fun { portended } -> portended) |> [%sexp_of: a list]
;;
let push t a =
let[@inline] rec loop () =
let before = Awaitable.get t in
match
Awaitable.compare_and_set
t
~if_phys_equal_to:before
~replace_with:({ portended = a } :: before)
with
| Set_here ->
(match before with
| [] -> Awaitable.signal t
| _ :: _ -> ())
| Compare_failed ->
Adaptive_backoff.once ~random_key:(Awaitable.random_key t) ~log_scale;
loop ()
in
loop () [@nontail]
;;
type state =
| Never_awaited
| Signaled
let pop await t =
let[@inline] rec loop state =
match Awaitable.get t with
| [] ->
(match Awaitable.await await t ~until_phys_unequal_to:[] with
| Terminated -> raise Await.Terminated
| Signaled -> loop Signaled)
| x :: xs as cur ->
(match Awaitable.compare_and_set t ~if_phys_equal_to:cur ~replace_with:xs with
| Set_here ->
(match state, xs with
| Signaled, _ :: _ -> Awaitable.signal t
| Never_awaited, _ | _, [] -> ());
x.portended
| Compare_failed ->
Adaptive_backoff.once ~random_key:(Awaitable.random_key t) ~log_scale;
loop state)
in
loop Never_awaited [@nontail]
;;
let pop_or_cancel await c t =
let[@inline] rec loop state : _ Or_canceled.t =
match Awaitable.get t with
| [] ->
(match Awaitable.await_or_cancel await c t ~until_phys_unequal_to:[] with
| Terminated -> raise Await.Terminated
| Canceled -> Canceled
| Signaled -> loop Signaled)
| x :: xs as cur ->
(match Awaitable.compare_and_set t ~if_phys_equal_to:cur ~replace_with:xs with
| Set_here ->
(match state, xs with
| Signaled, _ :: _ -> Awaitable.signal t
| Never_awaited, _ | _, [] -> ());
Completed x.portended
| Compare_failed ->
Adaptive_backoff.once ~random_key:(Awaitable.random_key t) ~log_scale;
loop state)
in
loop Never_awaited [@nontail]
;;
let pop_nonblocking t =
let[@inline] rec loop () =
match Awaitable.get t with
| [] ->
Adaptive_backoff.once_unless_alone ~random_key:(Awaitable.random_key t) ~log_scale;
Null
| x :: xs as cur ->
(match Awaitable.compare_and_set t ~if_phys_equal_to:cur ~replace_with:xs with
| Set_here -> This x.portended
| Compare_failed ->
Adaptive_backoff.once ~random_key:(Awaitable.random_key t) ~log_scale;
loop ())
in
loop () [@nontail]
;;
external portended_list
: 'a Modes.Portended.t list
-> 'a list @ contended portable
@@ portable
= "%identity"
let drain t = Awaitable.exchange t [] |> portended_list
let drain_blocking await t =
let[@inline] rec loop () =
match Awaitable.get t with
| [] ->
(match Awaitable.await await t ~until_phys_unequal_to:[] with
| Terminated -> raise Await.Terminated
| Signaled -> loop ())
| _ :: _ ->
(match Awaitable.exchange t [] with
| [] ->
Adaptive_backoff.once ~random_key:(Awaitable.random_key t) ~log_scale;
loop ()
| _ :: _ as list -> portended_list list)
in
loop () [@nontail]
;;
let drain_blocking_or_cancel await c t =
let[@inline] rec loop () =
match Awaitable.get t with
| [] ->
(match Awaitable.await_or_cancel await c t ~until_phys_unequal_to:[] with
| Terminated -> raise Await.Terminated
| Canceled -> Or_canceled.Canceled
| Signaled -> loop ())
| _ :: _ ->
(match Awaitable.exchange t [] with
| [] ->
Adaptive_backoff.once ~random_key:(Awaitable.random_key t) ~log_scale;
loop ()
| _ :: _ as list -> Completed (portended_list list))
in
loop () [@nontail]
;;
module For_testing = Awaitable.For_testing