machine_stream_queue.bend
bendvy-ecs-core@0.1.0.1 · 86 lines · 0 definitions
1import Base
2
3# Owned descriptor-local FIFO metadata. Back is newest-first, front oldest-first.
4# Cached event size and batch count retain empty-batch compatibility without scans.
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,batchCount: Nat}
9
10def empty(~E: Data) -> Queue<E>: Queue{[],[],0n,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
17# The compatibility batch-list lowering also uses count zero for empty batches.
18def publish_nonempty(~E: Data,queue: Queue<E>,+tick: Nat,+items: List<&2,E>,+count: Nat) -> Queue<E>:
19 match queue:
20 case Queue{front,back,size,dropped,batches}: Queue{front,Batch{tick,count,items} <> back,Nat.add(size,count),dropped,1n+batches}
21
22def publish_checked(~E: Data,nonempty: Bool,queue: Queue<E>,tick: Nat,items: List<&2,E>,count: Nat) -> Queue<E>:
23 match nonempty:
24 case False{}: queue
25 case True{}: publish_nonempty(~E,queue,tick,items,count)
26
27def publish(~E: Data,queue: Queue<E>,tick: Nat,+items: List<&2,E>) -> Queue<E>:
28 publish_checked(~E,Nat.is_lt(0n,count_loop(~E,items,0n)),queue,tick,items,count_loop(~E,items,0n))
29
30def normalize(~E: Data,queue: Queue<E>) -> Queue<E>:
31 match queue:
32 case Queue{Nil{},back,size,dropped,batches}: Queue{List.reverse(&2,Batch<E>,back),[],size,dropped,batches}
33 case Queue{front,back,size,dropped,batches}: Queue{front,back,size,dropped,batches}
34
35def 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,+batches:Nat,ih: Queue<E> -> Queue<E>) -> Queue<E>:
36 match remove:
37 case False{}: Queue{Batch{tick,count,items} <> rest,back,size,dropped,batches}
38 case True{}: ih(Queue{rest,back,Nat.sub(size,count),Nat.max(dropped,tick),Nat.sub(batches,1n)})
39
40# Match the canonical two-pass policy even for non-monotonic supplied ticks:
41# first remove the expired prefix, then evict whole batches for capacity only.
42def past_normalized(~E:Data,queue:Queue<E>,+boundary:Nat,ih:Queue<E> -> Queue<E>) -> Queue<E>:
43 match queue:
44 case Queue{Nil{},back,size,dropped,batches}:Queue{[],back,size,dropped,batches}
45 case Queue{Batch{+tick,+count,+items} <> +rest,+back,+size,+dropped,+batches}:trim_step(~E,Nat.is_le(tick,boundary),tick,count,items,rest,back,size,dropped,batches,ih)
46def past_loop(~E:Data,fuel:Nat,queue:Queue<E>,+boundary:Nat) -> Queue<E>:
47 match fuel:
48 case Zero{}:queue
49 case Succ{rest}:past_normalized(~E,normalize(~E,queue),boundary,next => past_loop(~E,rest,next,boundary))
50def capacity_normalized(~E:Data,queue:Queue<E>,+capacity:Nat,ih:Queue<E> -> Queue<E>) -> Queue<E>:
51 match queue:
52 case Queue{Nil{},back,size,dropped,batches}:Queue{[],back,size,dropped,batches}
53 case Queue{Batch{+tick,+count,+items} <> +rest,+back,+size,+dropped,+batches}:trim_step(~E,Nat.is_lt(capacity,size),tick,count,items,rest,back,size,dropped,batches,ih)
54def capacity_loop(~E:Data,fuel:Nat,queue:Queue<E>,+capacity:Nat) -> Queue<E>:
55 match fuel:
56 case Zero{}:queue
57 case Succ{rest}:capacity_normalized(~E,normalize(~E,queue),capacity,next => capacity_loop(~E,rest,next,capacity))
58def capacity_pass(~E:Data,queue:Queue<E>,capacity:Nat) -> Queue<E>:
59 match queue:
60 case Queue{front,back,size,dropped,+batches}:capacity_loop(~E,Nat.add(batches,1n),Queue{front,back,size,dropped,batches},capacity)
61def trim(~E:Data,queue:Queue<E>,boundary:Nat,capacity:Nat) -> Queue<E>:
62 match queue:
63 case Queue{front,back,size,dropped,+batches}:capacity_pass(~E,past_loop(~E,Nat.add(batches,1n),Queue{front,back,size,dropped,batches},boundary),capacity)
64
65# Materialization is tail-safe; no recursive append of a growing retained prefix.
66def prepend_items(~E: Data,items: List<&2,E>,acc: List<&2,E>) -> List<&2,E>:
67 match items:
68 case []: acc
69 case item <> rest: prepend_items(~E,rest,item <> acc)
70
71def visible_items(~E: Data,visible: Bool,items: List<&2,E>,acc: List<&2,E>) -> List<&2,E>:
72 match visible:
73 case False{}: acc
74 case True{}: prepend_items(~E,items,acc)
75
76def read_loop(~E: Data,batches: List<&2,Batch<E>>,+cursor: Nat,acc: List<&2,E>) -> List<&2,E>:
77 match batches:
78 case []: acc
79 case Batch{tick,_,items} <> rest: read_loop(~E,rest,cursor,visible_items(~E,Nat.is_lt(cursor,tick),items,acc))
80
81def read_back(~E: Data,+front: List<&2,Batch<E>>,+back: List<&2,Batch<E>>,+cursor: Nat) -> List<&2,E>:
82 List.reverse(&2,E,read_loop(~E,List.reverse(&2,Batch<E>,back),cursor,read_loop(~E,front,cursor,[])))
83
84def read(~E: Data,queue: Queue<E>,cursor: Nat) -> List<&2,E>:
85 match queue:
86 case Queue{front,back,_,_,_}: read_back(~E,front,back,cursor)