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

Internal state of an Channel.

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

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

Instances For

• Std.Channel.instInhabitedState

instance Std.Channel.instInhabitedState {a✝ : Type} : Inhabited (State a✝)

Equations

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

def Std.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

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

Instances For

• Std.instNonemptyChannel

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

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

Creates a new Channel.

Equations

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

def Std.Channel.send {α : Type} (ch : 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 Std.Channel.close {α : Type} (ch : Channel α) : BaseIO Unit

Closes an Channel.

Equations

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

def Std.Channel.recv? {α : Type} (ch : 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 Std.Channel.forAsync {α : Type} (f : α → BaseIO Unit) (ch : Channel α) (prio : 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 Std.Channel.recvAllCurrent {α : Type} (ch : 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 = Std.Mutex.atomically ch (modifyGet fun (st : Std.Channel.State α) => (st.values.toArray, { values := ∅, consumers := st.consumers, closed := st.closed }))

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

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

Equations

• Std.Channel.Sync = Std.Channel

Instances For

• Std.instForInSyncOfMonadLiftTBaseIO

def Std.Channel.sync {α : Type} (ch : 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 Std.Channel.Sync.recv? {α : Type} (ch : 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 ← Std.Channel.recv? ch IO.wait __do_lift

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

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

Equations

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