internal/relation_failures/queue.bend
bendvy-ecs-core@0.1.0.1 · 74 lines · 0 definitions
1import Base
2
3# Owned descriptor-local FIFO metadata. Back is newest-first, front oldest-first.
4# Cached size makes publication and whole-batch capacity trim amortized linear.
5type Batch<E: Data> is Data:
6 Batch{tick: Nat,count: Nat,values: List<&2,E>}
7type Queue<E: Data> is Data:
8 Queue{front: List<&2,Batch<E>>,back: List<&2,Batch<E>>,size: Nat,dropped: Nat}
9
10def empty(~E: Data) -> Queue<E>: Queue{[],[],0n,0n}
11
12def count_loop(~E: Data,items: List<&2,E>,acc: Nat) -> Nat:
13 match items:
14 case []: acc
15 case _ <> rest: count_loop(~E,rest,Nat.add(1n,acc))
16
17def publish_nonempty(~E: Data,queue: Queue<E>,+tick: Nat,+items: List<&2,E>,+count: Nat) -> Queue<E>:
18 match queue:
19 case Queue{front,back,size,dropped}: Queue{front,Batch{tick,count,items} <> back,Nat.add(size,count),dropped}
20
21def publish_checked(~E: Data,nonempty: Bool,queue: Queue<E>,tick: Nat,items: List<&2,E>,count: Nat) -> Queue<E>:
22 match nonempty:
23 case False{}: queue
24 case True{}: publish_nonempty(~E,queue,tick,items,count)
25
26def publish(~E: Data,queue: Queue<E>,tick: Nat,+items: List<&2,E>) -> Queue<E>:
27 publish_checked(~E,Nat.is_lt(0n,count_loop(~E,items,0n)),queue,tick,items,count_loop(~E,items,0n))
28
29def normalize(~E: Data,queue: Queue<E>) -> Queue<E>:
30 match queue:
31 case Queue{Nil{},back,size,dropped}: Queue{List.reverse(&2,Batch<E>,back),[],size,dropped}
32 case Queue{front,back,size,dropped}: Queue{front,back,size,dropped}
33
34def trim_step(~E: Data,remove: Bool,+tick: Nat,+count: Nat,+items: List<&2,E>,+rest: List<&2,Batch<E>>,+back: List<&2,Batch<E>>,+size: Nat,+dropped: Nat,ih: Queue<E> -> Queue<E>) -> Queue<E>:
35 match remove:
36 case False{}: Queue{Batch{tick,count,items} <> rest,back,size,dropped}
37 case True{}: ih(Queue{rest,back,Nat.sub(size,count),Nat.max(dropped,tick)})
38
39def trim_normalized(~E: Data,queue: Queue<E>,+boundary: Nat,+capacity: Nat,ih: Queue<E> -> Queue<E>) -> Queue<E>:
40 match queue:
41 case Queue{Nil{},back,size,dropped}: Queue{[],back,size,dropped}
42 case Queue{Batch{+tick,+count,+items} <> +rest,+back,+size,+dropped}: trim_step(~E,Bool.or(Nat.is_le(tick,boundary),Nat.is_lt(capacity,size)),tick,count,items,rest,back,size,dropped,ih)
43
44def trim_loop(~E: Data,fuel: Nat,queue: Queue<E>,+boundary: Nat,+capacity: Nat) -> Queue<E>:
45 match fuel:
46 case Zero{}: queue
47 case Succ{rest}: trim_normalized(~E,normalize(~E,queue),boundary,capacity,next => trim_loop(~E,rest,next,boundary,capacity))
48
49def trim(~E: Data,queue: Queue<E>,boundary: Nat,capacity: Nat) -> Queue<E>:
50 match queue:
51 case Queue{+front,+back,+size,+dropped}: trim_loop(~E,Nat.add(size,1n),Queue{front,back,size,dropped},boundary,capacity)
52
53# Materialization is tail-safe; no recursive append of a growing retained prefix.
54def prepend_items(~E: Data,items: List<&2,E>,acc: List<&2,E>) -> List<&2,E>:
55 match items:
56 case []: acc
57 case item <> rest: prepend_items(~E,rest,item <> acc)
58
59def visible_items(~E: Data,visible: Bool,items: List<&2,E>,acc: List<&2,E>) -> List<&2,E>:
60 match visible:
61 case False{}: acc
62 case True{}: prepend_items(~E,items,acc)
63
64def read_loop(~E: Data,batches: List<&2,Batch<E>>,+cursor: Nat,acc: List<&2,E>) -> List<&2,E>:
65 match batches:
66 case []: acc
67 case Batch{tick,_,items} <> rest: read_loop(~E,rest,cursor,visible_items(~E,Nat.is_lt(cursor,tick),items,acc))
68
69def read_back(~E: Data,+front: List<&2,Batch<E>>,+back: List<&2,Batch<E>>,+cursor: Nat) -> List<&2,E>:
70 List.reverse(&2,E,read_loop(~E,List.reverse(&2,Batch<E>,back),cursor,read_loop(~E,front,cursor,[])))
71
72def read(~E: Data,queue: Queue<E>,cursor: Nat) -> List<&2,E>:
73 match queue:
74 case Queue{front,back,_,_}: read_back(~E,front,back,cursor)