lib.bend source

lib.bend on the hub · documented module

# RingBuf — fixed-capacity FIFO ring buffer for U32 values.# Depends only on Base. Capacity is a Nat so any non-negative capacity is valid.## The backing storage is a fixed List of slots. `start` points at the logical# front and `len` is the number of live values; physical indices wrap modulo# `capacity`. Empty slots contain 0 and are never observable through the API.## `push` is fail-if-full: it returns None when the buffer is full and Some of# the updated buffer otherwise. It never overwrites an existing value.import Basetype RingBuf is Data:  RB{capacity: Nat, start: Nat, len: Nat, slots: List<&2, U32>}type RingBuf.Push is Data:  PushResult{buffer: RingBuf}type RingBuf.Pop is Data:  PopResult{value: U32, rest: RingBuf}# Allocate exactly capacity slots. A zero-capacity buffer is valid but cannot# accept pushes.def RingBuf.new(+capacity: Nat) -> RingBuf:  RB{capacity, 0n, 0n, List.replicate(U32, capacity, 0)}def RingBuf.empty(capacity: Nat) -> RingBuf:  RingBuf.new(capacity)def RingBuf.capacity(r: RingBuf) -> Nat:  match r:    case RB{capacity, start, len, slots}:      capacitydef RingBuf.length(r: RingBuf) -> Nat:  match r:    case RB{capacity, start, len, slots}:      lendef RingBuf.is_empty(r: RingBuf) -> Bool:  Nat.is_eq(RingBuf.length(r), 0n)def RingBuf.is_full(+r: RingBuf) -> Bool:  Nat.is_eq(RingBuf.length(r), RingBuf.capacity(r))# Read a slot. Internal buffers always have a slot at every valid index.def RingBuf.read(slots: List<&2, U32>, index: Nat) -> U32:  Maybe.default(&2, U32, List.get(&2, U32, slots, index), 0)def RingBuf.push.go(  +capacity: Nat, +start: Nat, +len: Nat, slots: List<&2, U32>, x: U32,  available: Bool) -> Maybe<&2, RingBuf.Push>:  match available:    case False{}:      None{}    case True{}:      index = Nat.mod(Nat.add(start, len), capacity)      next_slots = List.set(&2, U32, slots, index, x)      Some{PushResult{RB{capacity, start, 1n+len, next_slots}}}def RingBuf.push(r: RingBuf, x: U32) -> Maybe<&2, RingBuf.Push>:  match r:    case RB{+capacity, +start, +len, slots}:      RingBuf.push.go(capacity, start, len, slots, x, Nat.is_lt(len, capacity))def RingBuf.pop.go(  +capacity: Nat, +start: Nat, len: Nat, +slots: List<&2, U32>) -> Maybe<&2, RingBuf.Pop>:  match len:    case 0n:      None{}    case 1n+p:      x = RingBuf.read(slots, start)      next_start = Nat.mod(1n+start, capacity)      Some{PopResult{x, RB{capacity, next_start, p, slots}}}def RingBuf.pop(r: RingBuf) -> Maybe<&2, RingBuf.Pop>:  match r:    case RB{capacity, start, len, slots}:      RingBuf.pop.go(capacity, start, len, slots)def RingBuf.peek(r: RingBuf) -> Maybe<&2, U32>:  match r:    case RB{capacity, start, len, slots}:      match len:        case 0n:          None{}        case 1n+p:          Some{RingBuf.read(slots, start)}def RingBuf.to_list.go(  +work: List<&2, Unit>, +capacity: Nat, +slots: List<&2, U32>, +index: Nat) -> List<&2, U32>:  match work:    case Nil{}:      Nil{}    case +u <> rest:      next_index = Nat.mod(1n+index, capacity)      tail = RingBuf.to_list.go(rest, capacity, slots, next_index)      RingBuf.read(slots, index) <> taildef RingBuf.to_list(r: RingBuf) -> List<&2, U32>:  match r:    case RB{capacity, start, len, slots}:      work = List.replicate(Unit, len, Unit{})      RingBuf.to_list.go(work, capacity, slots, start)