structure IO.Channel.State (α : Type) : Type

Internal state of an Channel.

We maintain the invariant that at all times either consumers or values is empty.

• values : Std.Queue α
• consumers : Std.Queue (IO.Promise (Option α))
• closed : Bool

instance IO.Channel.instInhabitedState : {a : Type} → Inhabited (IO.Channel.State a)

Equations

• IO.Channel.instInhabitedState = { default := { values := default, consumers := default, closed := default } }

def IO.Channel (α : Type) : Type

FIFO channel with unbounded buffer, where recv? returns a Task.

A channel can be closed. Once it is closed, all sends are ignored, and recv? returns none once the queue is empty.

Equations

• IO.Channel α = IO.Mutex (IO.Channel.State α)

instance IO.instNonemptyChannel {α : Type} : Nonempty (IO.Channel α)

Equations

• ⋯ = ⋯

def IO.Channel.new {α : Type} : BaseIO (IO.Channel α)

Creates a new Channel.

Equations

• IO.Channel.new = IO.Mutex.new { values := ∅, consumers := ∅, closed := false }

def IO.Channel.send {α : Type} (ch : IO.Channel α) (v : α) : BaseIO Unit

Sends a message on an Channel.

This function does not block.

Equations

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

def IO.Channel.close {α : Type} (ch : IO.Channel α) : BaseIO Unit

Closes an Channel.

Equations

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

def IO.Channel.recv? {α : Type} (ch : IO.Channel α) : BaseIO (Task (Option α))

Receives a message, without blocking. The returned task waits for the message. Every message is only received once.

Returns none if the channel is closed and the queue is empty.

Equations

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

partial def IO.Channel.forAsync {α : Type} (f : α → BaseIO Unit) (ch : IO.Channel α) (prio : optParam Task.Priority Task.Priority.default) : BaseIO (Task Unit)

ch.forAsync f calls f for every messages received on ch.

Note that if this function is called twice, each forAsync only gets half the messages.

def IO.Channel.recvAllCurrent {α : Type} (ch : IO.Channel α) : BaseIO (Array α)

Receives all currently queued messages from the channel.

Those messages are dequeued and will not be returned by recv?.

Equations

• ch.recvAllCurrent = IO.Mutex.atomically ch (modifyGet fun (st : IO.Channel.State α) => (st.values.toArray, { values := ∅, consumers := st.consumers, closed := st.closed }))

def IO.Channel.Sync (α : Type) : Type

Type tag for synchronous (blocking) operations on a Channel.

Equations

• IO.Channel.Sync = IO.Channel

def IO.Channel.sync {α : Type} (ch : IO.Channel α) : IO.Channel.Sync α

Accesses synchronous (blocking) version of channel operations.

For example, ch.sync.recv? blocks until the next message, and for msg in ch.sync do ... iterates synchronously over the channel. These functions should only be used in dedicated threads.

Equations

• ch.sync = ch

def IO.Channel.Sync.recv? {α : Type} (ch : IO.Channel.Sync α) : BaseIO (Option α)

Synchronously receives a message from the channel.

Every message is only received once. Returns none if the channel is closed and the queue is empty.

Equations

• ch.recv? = do let __do_lift ← IO.Channel.recv? ch IO.wait __do_lift

instance IO.instForInSyncOfMonadLiftTBaseIO {m : Type → Type u_1} {α : Type} [MonadLiftT BaseIO m] : ForIn m (IO.Channel.Sync α) α

for msg in ch.sync do ... receives all messages in the channel until it is closed.

Equations

• IO.instForInSyncOfMonadLiftTBaseIO = { forIn := fun {β : Type} [Monad m] (ch : IO.Channel.Sync α) (b : β) (f : α → β → m (ForInStep β)) => IO.Channel.Sync.forIn ch f b }