queue.bend

bend-kit-scheduler@0.4.1.0 · 175 lines · 29 definitions

1# Pure generic scheduler queue; Nat deadlines and never-reused FIFO IDs.
2import Base
3import bend-kit-collections@0.1.4.1/heap.bend as H
4
5type Ticket<-A: Data> is Data:
6  Ticket{id: Nat, deadline: Nat, payload: A}
7
8type State<-A: Data> is Data:
9  State{next_id: Nat, pending: H.Heap<Ticket<A>>, running: List<&2, Ticket<A>>}
10
11def id(-A: Data, t: Ticket<A>) -> Nat:
12  match t:
13    case Ticket{i, d, x}:
14      i
15
16def deadline(-A: Data, t: Ticket<A>) -> Nat:
17  match t:
18    case Ticket{i, d, x}:
19      d
20
21def payload(-A: Data, t: Ticket<A>) -> A:
22  match t:
23    case Ticket{i, d, x}:
24      x
25
26def le.same(b: Bool, i: Nat, j: Nat, d: Nat, e: Nat) -> Bool:
27  match b:
28    case True{}:
29      Nat.is_le(i, j)
30    case False{}:
31      Nat.is_le(d, e)
32
33def le(-A: Data, a: Ticket<A>, b: Ticket<A>) -> Bool:
34  match a b:
35    case Ticket{i, +d, x} Ticket{j, +e, y}:
36      le.same(Nat.is_eq(d, e), i, j, d, e)
37
38def new(-A: Data) -> State<A>:
39  State{0n, H.Heap.new(Ticket<A>), Nil{}}
40
41def enqueue(~A: Data, d: Nat, x: A, s: State<A>) -> State<A> & Nat:
42  match s:
43    case State{+i, h, r}:
44      (State{1n+i, H.Heap.push(~Ticket<A>, ~(a => b => le(A, a, b)), Ticket{i, d, x}, h), r}, i)
45
46def select.due(~A: Data, b: Bool, i: Nat, h: H.Heap<Ticket<A>>, r: List<&2, Ticket<A>>, t: Ticket<A>, l: H.Heap<Ticket<A>>, q: H.Heap<Ticket<A>>) -> State<A> & Maybe<&2, Ticket<A>>:
47  match b:
48    case False{}:
49      (State{i, h, r}, None{})
50    case True{}:
51      +t = t
52      (State{i, H.Heap.merge(~Ticket<A>, ~(a => b => le(A, a, b)), l, q), t <> r}, Some{t})
53
54def select(~A: Data, now: Nat, s: State<A>) -> State<A> & Maybe<&2, Ticket<A>>:
55  match s:
56    case State{i, H.HLeaf{}, r}:
57      (State{i, H.HLeaf{}, r}, None{})
58    case State{i, H.HNode{+n, +t, +l, +q}, r}:
59      select.due(~A, Nat.is_le(deadline(A, t), now), i, H.HNode{n, t, l, q}, r, t, l, q)
60
61# ponytail: eager O(n) cancellation; indexed deletion if cancellation dominates.
62def remove.list.keep(-A: Data, b: Bool, t: Ticket<A>, rest: List<&2, Ticket<A>>) -> List<&2, Ticket<A>>:
63  match b:
64    case True{}:
65      rest
66    case False{}:
67      t <> rest
68
69def remove.list(-A: Data, xs: List<&2, Ticket<A>>, +i: Nat) -> List<&2, Ticket<A>>:
70  match xs:
71    case Nil{}:
72      Nil{}
73    case Con{+t, rest}:
74      remove.list.keep(A, Nat.is_eq(id(A, t), i), t, remove.list(A, rest, i))
75
76def remove.heap.keep(~A: Data, b: Bool, t: Ticket<A>, l: H.Heap<Ticket<A>>, r: H.Heap<Ticket<A>>) -> H.Heap<Ticket<A>>:
77  match b:
78    case True{}:
79      H.Heap.merge(~Ticket<A>, ~(a => b => le(A, a, b)), l, r)
80    case False{}:
81      +l = l
82      +r = r
83      H.HNode{1n+Nat.add(H.Heap.size(Ticket<A>, l), H.Heap.size(Ticket<A>, r)), t, l, r}
84
85def remove.heap(~A: Data, h: H.Heap<Ticket<A>>, +i: Nat) -> H.Heap<Ticket<A>>:
86  match h:
87    case H.HLeaf{}:
88      H.HLeaf{}
89    case H.HNode{n, +t, l, r}:
90      remove.heap.keep(~A, Nat.is_eq(id(A, t), i), t, remove.heap(~A, l, i), remove.heap(~A, r, i))
91
92def cancel(~A: Data, i: Nat, s: State<A>) -> State<A>:
93  match s:
94    case State{n, h, r}:
95      +i = i
96      State{n, remove.heap(~A, h, i), remove.list(A, r, i)}
97
98# Reference observation for laws; completion uses the single-pass take below.
99def find.list.choose(-A: Data, b: Bool, t: Ticket<A>, rest: Maybe<&2, Ticket<A>>) -> Maybe<&2, Ticket<A>>:
100  match b:
101    case True{}:
102      Some{t}
103    case False{}:
104      rest
105
106def find.list(-A: Data, xs: List<&2, Ticket<A>>, +i: Nat) -> Maybe<&2, Ticket<A>>:
107  match xs:
108    case Nil{}:
109      None{}
110    case Con{+t, rest}:
111      find.list.choose(A, Nat.is_eq(id(A, t), i), t, find.list(A, rest, i))
112
113type Taken<-A: Data> is Data:
114  Taken{remaining: List<&2, Ticket<A>>, found: Maybe<&2, Ticket<A>>}
115
116# Remove all matching entries and return the first token in one traversal.
117def take.keep(-A: Data, b: Bool, t: Ticket<A>, result: Taken<A>) -> Taken<A>:
118  match b:
119    case True{}:
120      match result:
121        case Taken{rest, found}:
122          Taken{rest, Some{t}}
123    case False{}:
124      match result:
125        case Taken{rest, found}:
126          Taken{t <> rest, found}
127
128def take.list(-A: Data, xs: List<&2, Ticket<A>>, +i: Nat) -> Taken<A>:
129  match xs:
130    case Nil{}:
131      Taken{Nil{}, None{}}
132    case Con{+t, rest}:
133      take.keep(A, Nat.is_eq(id(A, t), i), t, take.list(A, rest, i))
134
135def complete.taken(-A: Data, result: Taken<A>, n: Nat, h: H.Heap<Ticket<A>>) -> State<A> & Maybe<&2, Ticket<A>>:
136  match result:
137    case Taken{r, found}:
138      (State{n, h, r}, found)
139
140def complete(-A: Data, i: Nat, s: State<A>) -> State<A> & Maybe<&2, Ticket<A>>:
141  match s:
142    case State{n, h, r}:
143      complete.taken(A, take.list(A, r, i), n, h)
144
145def requeue.found(~A: Data, found: Maybe<&2, Ticket<A>>, d: Nat, s: State<A>) -> State<A> & Bool:
146  match found:
147    case None{}:
148      (s, False{})
149    case Some{Ticket{i, old, x}}:
150      match s:
151        case State{n, h, r}:
152          (State{n, H.Heap.push(~Ticket<A>, ~(a => b => le(A, a, b)), Ticket{i, d, x}, h), r}, True{})
153
154def requeue.completed(~A: Data, r: State<A> & Maybe<&2, Ticket<A>>, d: Nat) -> State<A> & Bool:
155  (s, found) = r
156  requeue.found(~A, found, d, s)
157
158def requeue(~A: Data, i: Nat, d: Nat, s: State<A>) -> State<A> & Bool:
159  requeue.completed(~A, complete(A, i, s), d)
160
161# Non-consuming observations used by adapters and laws.
162def next_id(-A: Data, s: State<A>) -> Nat:
163  match s:
164    case State{n, h, r}:
165      n
166
167def pending(-A: Data, s: State<A>) -> H.Heap<Ticket<A>>:
168  match s:
169    case State{n, h, r}:
170      h
171
172def running(-A: Data, s: State<A>) -> List<&2, Ticket<A>>:
173  match s:
174    case State{n, h, r}:
175      r