structure Lapis.Concurrent.AtomicCounter : Type

An atomic counter for generating unique IDs

• value : IO.Ref Nat

def Lapis.Concurrent.AtomicCounter.new : IO AtomicCounter

Create a new counter starting at 0

Equations

• Lapis.Concurrent.AtomicCounter.new = do let ref ← liftM (IO.mkRef 0) pure { value := ref }

def Lapis.Concurrent.AtomicCounter.newFrom (start : Nat) : IO AtomicCounter

Create a new counter starting at a given value

Equations

• Lapis.Concurrent.AtomicCounter.newFrom start = do let ref ← liftM (IO.mkRef start) pure { value := ref }

def Lapis.Concurrent.AtomicCounter.incrementAndGet (c : AtomicCounter) : IO Nat

Atomically increment and return the previous value

Equations

• c.incrementAndGet = ST.Ref.modifyGet c.value fun (n : Nat) => (n, n + 1)

def Lapis.Concurrent.AtomicCounter.get (c : AtomicCounter) : IO Nat

Get the current value without incrementing

Equations

• c.get = ST.Ref.get c.value

Unbounded Channel

structure Lapis.Concurrent.Channel.Unbounded (α : Type) : Type

An unbounded MPSC (multi-producer, single-consumer) channel. Multiple actors can send messages, one actor receives them in order. Uses atomic IO.Ref operations - no mutex needed.

• queue : IO.Ref (Array α)

  Queue of pending messages

• signal : IO.Ref (IO.Promise Unit)

  Promise for signaling new messages (recreated after each wait)

def Lapis.Concurrent.Channel.Unbounded.new {α : Type} : IO (Unbounded α)

Create a new unbounded channel

Equations

• One or more equations did not get rendered due to their size.

def Lapis.Concurrent.Channel.Unbounded.send {α : Type} (ch : Unbounded α) (msg : α) : IO Unit

Send a message to the channel (non-blocking). Uses atomic modify - safe for concurrent sends.

Equations

• ch.send msg = do ST.Ref.modify ch.queue fun (x : Array α) => x.push msg let sig ← ST.Ref.get ch.signal liftM (IO.Promise.resolve () sig)

partial def Lapis.Concurrent.Channel.Unbounded.recv {α : Type} (ch : Unbounded α) : IO α

Receive a message from the channel (blocking)

def Lapis.Concurrent.Channel.Unbounded.tryRecv {α : Type} (ch : Unbounded α) : IO (Option α)

Try to receive a message without blocking

Equations

• ch.tryRecv = ST.Ref.modifyGet ch.queue fun (q : Array α) => if h : 0 < q.size then (some q[0], q.extract 1) else (none, q)

def Lapis.Concurrent.Channel.Unbounded.isEmpty {α : Type} (ch : Unbounded α) : IO Bool

Check if the channel has pending messages

Equations

• ch.isEmpty = do let q ← ST.Ref.get ch.queue pure q.isEmpty

def Lapis.Concurrent.Channel.Unbounded.size {α : Type} (ch : Unbounded α) : IO Nat

Get the number of pending messages

Equations

• ch.size = do let q ← ST.Ref.get ch.queue pure q.size

Bounded Channel

structure Lapis.Concurrent.Channel.Bounded (α : Type) : Type

A bounded MPSC channel with backpressure. Senders block when the channel is full. Uses atomic IO.Ref operations - no mutex needed.

• queue : IO.Ref (Array α)

  Queue of pending messages

• capacity : Nat

  Maximum capacity

• notEmptySignal : IO.Ref (IO.Promise Unit)

  Signal for receivers (queue not empty)

• notFullSignal : IO.Ref (IO.Promise Unit)

  Signal for senders (queue not full)

def Lapis.Concurrent.Channel.Bounded.new {α : Type} (capacity : Nat) : IO (Bounded α)

Create a new bounded channel with given capacity

Equations

• One or more equations did not get rendered due to their size.

partial def Lapis.Concurrent.Channel.Bounded.send {α : Type} (ch : Bounded α) (msg : α) : IO Unit

Send a message to the channel (blocks if full)

def Lapis.Concurrent.Channel.Bounded.trySend {α : Type} (ch : Bounded α) (msg : α) : IO Bool

Try to send a message without blocking

Equations

• One or more equations did not get rendered due to their size.

partial def Lapis.Concurrent.Channel.Bounded.recv {α : Type} (ch : Bounded α) : IO α

Receive a message from the channel (blocking)

def Lapis.Concurrent.Channel.Bounded.tryRecv {α : Type} (ch : Bounded α) : IO (Option α)

Try to receive a message without blocking

Equations

• One or more equations did not get rendered due to their size.

def Lapis.Concurrent.Channel.Bounded.isFull {α : Type} (ch : Bounded α) : IO Bool

Check if the channel is full

Equations

• ch.isFull = do let q ← ST.Ref.get ch.queue pure (decide (q.size ≥ ch.capacity))

def Lapis.Concurrent.Channel.Bounded.isEmpty {α : Type} (ch : Bounded α) : IO Bool

Check if the channel is empty

Equations

• ch.isEmpty = do let q ← ST.Ref.get ch.queue pure q.isEmpty

Oneshot Channel

structure Lapis.Concurrent.Channel.Oneshot (α : Type) : Type

A oneshot channel for single request-response patterns. Can only be used once: one send, one receive.

• promise : IO.Promise α

def Lapis.Concurrent.Channel.Oneshot.new {α : Type} [Nonempty α] : IO (Oneshot α)

Create a new oneshot channel

Equations

• Lapis.Concurrent.Channel.Oneshot.new = do let promise ← liftM IO.Promise.new pure { promise := promise }

def Lapis.Concurrent.Channel.Oneshot.send {α : Type} (ch : Oneshot α) (value : α) : IO Unit

Send a value (can only be called once)

Equations

• ch.send value = liftM (IO.Promise.resolve value ch.promise)

def Lapis.Concurrent.Channel.Oneshot.recv {α : Type} (ch : Oneshot α) : IO α

Receive the value (blocks until sent)

Equations

• ch.recv = pure ch.promise.result!.get

def Lapis.Concurrent.Channel.Oneshot.tryRecv {α : Type} (ch : Oneshot α) : IO (Option α)

Try to receive without blocking (returns none if not yet resolved)

Equations

• ch.tryRecv = pure ch.promise.result?.get

def Lapis.Concurrent.Channel.Oneshot.isReady {α : Type} (ch : Oneshot α) : IO Bool

Check if the value has been sent

Equations

• ch.isReady = pure ch.promise.result?.get.isSome

Select (Multi-channel waiting)

inductive Lapis.Concurrent.Channel.SelectResult (α β : Type) : Type

Result of a select operation

• first {α β : Type} : α → SelectResult α β
• second {α β : Type} : β → SelectResult α β