This module contains the implementation of Std.Channel. Std.Channel is a multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering as well as synchronous and asynchronous APIs.

Additionally Std.CloseableChannel is provided in case closing the channel is of interest. The two are distinct as the non closable Std.Channel can never throw errors which makes for cleaner code.

inductive Std.CloseableChannel.Error : Type

Errors that may be thrown while interacting with the channel API.

• closed : Error

  Tried to send to a closed channel.

• alreadyClosed : Error

  Tried to close an already closed channel.

Instances For

• Std.CloseableChannel.instHashableError
• Std.CloseableChannel.instReprError
• Std.CloseableChannel.instToStringError

instance Std.CloseableChannel.instReprError : Repr Error

Equations

• Std.CloseableChannel.instReprError = { reprPrec := Std.CloseableChannel.reprError✝ }

instance Std.CloseableChannel.instDecidableEqError : DecidableEq Error

Equations

• Std.CloseableChannel.instDecidableEqError x✝ y✝ = if h : x✝.toCtorIdx = y✝.toCtorIdx then isTrue ⋯ else isFalse ⋯

instance Std.CloseableChannel.instHashableError : Hashable Error

Equations

• Std.CloseableChannel.instHashableError = { hash := Std.CloseableChannel.hashError✝ }

instance Std.CloseableChannel.instToStringError : ToString Error

Equations

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

instance Std.CloseableChannel.instMonadLiftEIOErrorIO : MonadLift (EIO Error) IO

Equations

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

instance Std.CloseableChannel.Unbounded.instNonemptyState {α✝ : Type} : Nonempty (Std.CloseableChannel.Unbounded.State✝ α✝)

instance Std.CloseableChannel.instNonemptyUnbounded {α✝ : Type} : Nonempty (Std.CloseableChannel.Unbounded✝ α✝)

instance Std.CloseableChannel.instNonemptyFlavors {α✝ : Type} : Nonempty (Std.CloseableChannel.Flavors✝ α✝)

def Std.CloseableChannel (α : Type) : Type

A multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering and an asynchronous API, to switch into synchronous mode use CloseableChannel.sync.

Additionally Std.CloseableChannel can be closed if necessary, unlike Std.Channel. This introduces a need for error handling in some cases, thus it is usually easier to use Std.Channel if applicable.

Equations

• Std.CloseableChannel α = Std.CloseableChannel.Flavors✝ α

Instances For

• Std.instNonemptyCloseableChannel

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

A multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering and a synchronous API. This type acts as a convenient layer to use a channel in a blocking fashion and is not actually different from the original channel.

Additionally Std.CloseableChannel.Sync can be closed if necessary, unlike Std.Channel.Sync. This introduces the need to handle errors in some cases, thus it is usually easier to use Std.Channel if applicable.

Equations

• Std.CloseableChannel.Sync α = Std.CloseableChannel α

Instances For

• Std.CloseableChannel.Sync.instForInOfMonadLiftTBaseIO
• Std.instNonemptySync

instance Std.instNonemptyCloseableChannel {α : Type} : Nonempty (CloseableChannel α)

instance Std.instNonemptySync {α : Type} : Nonempty (CloseableChannel.Sync α)

def Std.CloseableChannel.new {α : Type} (capacity : Option Nat := none) : BaseIO (CloseableChannel α)

Create a new channel, if:

• capacity is none it will be unbounded (the default)
• capacity is some 0 it will always force a rendezvous between sender and receiver
• capacity is some n with n > 0 it will use a buffer of size n and begin blocking once it is filled

Equations

• Std.CloseableChannel.new = do let __do_lift ← Std.CloseableChannel.Unbounded.new✝ pure (Std.CloseableChannel.Flavors.unbounded✝ __do_lift)
• Std.CloseableChannel.new (some 0) = do let __do_lift ← Std.CloseableChannel.Zero.new✝ pure (Std.CloseableChannel.Flavors.zero✝ __do_lift)
• Std.CloseableChannel.new (some n.succ) = do let __do_lift ← Std.CloseableChannel.Bounded.new✝ (n + 1) ⋯ pure (Std.CloseableChannel.Flavors.bounded✝ __do_lift)

def Std.CloseableChannel.trySend {α : Type} (ch : CloseableChannel α) (v : α) : BaseIO Bool

Try to send a value to the channel, if this can be completed right away without blocking return true, otherwise don't send the value and return false.

Equations

• Std.CloseableChannel.trySend (Std.CloseableChannel.Flavors.unbounded✝ ch_2) v = ch_2.trySend v
• Std.CloseableChannel.trySend (Std.CloseableChannel.Flavors.zero✝ ch_2) v = ch_2.trySend v
• Std.CloseableChannel.trySend (Std.CloseableChannel.Flavors.bounded✝ ch_2) v = ch_2.trySend v

def Std.CloseableChannel.send {α : Type} (ch : CloseableChannel α) (v : α) : BaseIO (Task (Except Error Unit))

Send a value through the channel, returning a task that will resolve once the transmission could be completed. Note that the task may resolve to Except.error if the channel was closed before it could be completed.

Equations

• Std.CloseableChannel.send (Std.CloseableChannel.Flavors.unbounded✝ ch_2) v = ch_2.send v
• Std.CloseableChannel.send (Std.CloseableChannel.Flavors.zero✝ ch_2) v = ch_2.send v
• Std.CloseableChannel.send (Std.CloseableChannel.Flavors.bounded✝ ch_2) v = ch_2.send v

def Std.CloseableChannel.close {α : Type} (ch : CloseableChannel α) : EIO Error Unit

Closes the channel, returns Except.ok when called the first time, otherwise Except.error. When a channel is closed:

• no new values can be sent successfully anymore
• all blocked consumers are resolved to none (as no new messages can be sent they will never resolve)
• if there are already values waiting to be received they can still be received by subsequent recv calls

Equations

• Std.CloseableChannel.close (Std.CloseableChannel.Flavors.unbounded✝ ch_2) = ch_2.close
• Std.CloseableChannel.close (Std.CloseableChannel.Flavors.zero✝ ch_2) = ch_2.close
• Std.CloseableChannel.close (Std.CloseableChannel.Flavors.bounded✝ ch_2) = ch_2.close

def Std.CloseableChannel.isClosed {α : Type} (ch : CloseableChannel α) : BaseIO Bool

Return true if the channel is closed.

Equations

• Std.CloseableChannel.isClosed (Std.CloseableChannel.Flavors.unbounded✝ ch_2) = ch_2.isClosed
• Std.CloseableChannel.isClosed (Std.CloseableChannel.Flavors.zero✝ ch_2) = ch_2.isClosed
• Std.CloseableChannel.isClosed (Std.CloseableChannel.Flavors.bounded✝ ch_2) = ch_2.isClosed

def Std.CloseableChannel.tryRecv {α : Type} (ch : CloseableChannel α) : BaseIO (Option α)

Try to receive a value from the channel, if this can be completed right away without blocking return some value, otherwise return none.

Equations

• Std.CloseableChannel.tryRecv (Std.CloseableChannel.Flavors.unbounded✝ ch_2) = ch_2.tryRecv
• Std.CloseableChannel.tryRecv (Std.CloseableChannel.Flavors.zero✝ ch_2) = ch_2.tryRecv
• Std.CloseableChannel.tryRecv (Std.CloseableChannel.Flavors.bounded✝ ch_2) = ch_2.tryRecv

def Std.CloseableChannel.recv {α : Type} (ch : CloseableChannel α) : BaseIO (Task (Option α))

Receive a value from the channel, returning a task that will resolve once the transmission could be completed. Note that the task may resolve to none if the channel was closed before it could be completed.

Equations

• Std.CloseableChannel.recv (Std.CloseableChannel.Flavors.unbounded✝ ch_2) = ch_2.recv
• Std.CloseableChannel.recv (Std.CloseableChannel.Flavors.zero✝ ch_2) = ch_2.recv
• Std.CloseableChannel.recv (Std.CloseableChannel.Flavors.bounded✝ ch_2) = ch_2.recv

def Std.CloseableChannel.recvSelector {α : Type} (ch : CloseableChannel α) : Internal.IO.Async.Selector (Option α)

Creates a Selector that resolves once ch has data available and provides that that data. In particular if ch is closed while waiting on this Selector and no data is available already this will resolve to none.

Equations

• Std.CloseableChannel.recvSelector (Std.CloseableChannel.Flavors.unbounded✝ ch_2) = ch_2.recvSelector
• Std.CloseableChannel.recvSelector (Std.CloseableChannel.Flavors.zero✝ ch_2) = ch_2.recvSelector
• Std.CloseableChannel.recvSelector (Std.CloseableChannel.Flavors.bounded✝ ch_2) = ch_2.recvSelector

partial def Std.CloseableChannel.forAsync {α : Type} (f : α → BaseIO Unit) (ch : CloseableChannel α) (prio : Task.Priority := Task.Priority.default) : BaseIO (Task Unit)

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

Note that if this function is called twice, each message will only arrive at exactly one invocation.

@[inline]

def Std.CloseableChannel.sync {α : Type} (ch : CloseableChannel α) : Sync α

This function is a no-op and just a convenient way to expose the synchronous API of the channel.

Equations

• ch.sync = ch

@[inline]

def Std.CloseableChannel.Sync.new {α : Type} (capacity : Option Nat := none) : BaseIO (Sync α)

Create a new channel, if:

• capacity is none it will be unbounded (the default)
• capacity is some 0 it will always force a rendezvous between sender and receiver
• capacity is some n with n > 0 it will use a buffer of size n and begin blocking once it is filled

Equations

• Std.CloseableChannel.Sync.new capacity = Std.CloseableChannel.new capacity

@[inline]

def Std.CloseableChannel.Sync.trySend {α : Type} (ch : Sync α) (v : α) : BaseIO Bool

Try to send a value to the channel, if this can be completed right away without blocking return true, otherwise don't send the value and return false.

Equations

• ch.trySend v = Std.CloseableChannel.trySend ch v

def Std.CloseableChannel.Sync.send {α : Type} (ch : Sync α) (v : α) : EIO Error Unit

Send a value through the channel, blocking until the transmission could be completed. Note that this function may throw an error when trying to send to an already closed channel.

Equations

• ch.send v = do let __do_lift ← liftM (Std.CloseableChannel.send ch v) let __do_lift ← liftM (IO.wait __do_lift) EIO.ofExcept __do_lift

@[inline]

def Std.CloseableChannel.Sync.close {α : Type} (ch : Sync α) : EIO Error Unit

Closes the channel, returns Except.ok when called the first time, otherwise Except.error. When a channel is closed:

• no new values can be sent successfully anymore
• all blocked consumers are resolved to none (as no new messages can be sent they will never resolve)
• if there are already values waiting to be received they can still be received by subsequent recv calls

Equations

• ch.close = Std.CloseableChannel.close ch

@[inline]

def Std.CloseableChannel.Sync.isClosed {α : Type} (ch : Sync α) : BaseIO Bool

Return true if the channel is closed.

Equations

• ch.isClosed = Std.CloseableChannel.isClosed ch

@[inline]

def Std.CloseableChannel.Sync.tryRecv {α : Type} (ch : Sync α) : BaseIO (Option α)

Try to receive a value from the channel, if this can be completed right away without blocking return some value, otherwise return none.

Equations

• ch.tryRecv = Std.CloseableChannel.tryRecv ch

def Std.CloseableChannel.Sync.recv {α : Type} (ch : Sync α) : BaseIO (Option α)

Receive a value from the channel, blocking until the transmission could be completed. Note that the return value may be none if the channel was closed before it could be completed.

Equations

• ch.recv = do let __do_lift ← Std.CloseableChannel.recv ch IO.wait __do_lift

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

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

Equations

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

structure Std.Channel (α : Type) : Type

A multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering and an asynchronous API, to switch into synchronous mode use Channel.sync.

If a channel needs to be closed to indicate some sort of completion event use Std.CloseableChannel instead. Note that Std.CloseableChannel introduces a need for error handling in some cases, thus Std.Channel is usually easier to use if applicable.

• inner : Std.CloseableChannel α

Instances For

• Std.instNonemptyChannel

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

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

A multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering and a synchronous API. This type acts as a convenient layer to use a channel in a blocking fashion and is not actually different from the original channel.

If a channel needs to be closed to indicate some sort of completion event use Std.CloseableChannel.Sync instead. Note that Std.CloseableChannel.Sync introduces a need for error handling in some cases, thus Std.Channel.Sync is usually easier to use if applicable.

Equations

• Std.Channel.Sync α = Std.Channel α

Instances For

• Std.Channel.Sync.instForInOfInhabitedOfMonadLiftTBaseIO
• Std.instNonemptySync_1

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

@[inline]

def Std.Channel.new {α : Type} (capacity : Option Nat := none) : BaseIO (Channel α)

Create a new channel, if:

• capacity is none it will be unbounded (the default)
• capacity is some 0 it will always force a rendezvous between sender and receiver
• capacity is some n with n > 0 it will use a buffer of size n and begin blocking once it is filled

Equations

• Std.Channel.new capacity = do let __do_lift ← Std.CloseableChannel.new capacity pure { inner := __do_lift }

@[inline]

def Std.Channel.trySend {α : Type} (ch : Channel α) (v : α) : BaseIO Bool

Try to send a value to the channel, if this can be completed right away without blocking return true, otherwise don't send the value and return false.

Equations

• ch.trySend v = (Std.Channel.inner✝ ch).trySend v

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

Send a value through the channel, returning a task that will resolve once the transmission could be completed.

Equations

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

@[inline]

def Std.Channel.tryRecv {α : Type} (ch : Channel α) : BaseIO (Option α)

Try to receive a value from the channel, if this can be completed right away without blocking return some value, otherwise return none.

Equations

• ch.tryRecv = (Std.Channel.inner✝ ch).tryRecv

def Std.Channel.recv {α : Type} [Inhabited α] (ch : Channel α) : BaseIO (Task α)

Receive a value from the channel, returning a task that will resolve once the transmission could be completed. Note that the task may resolve to none if the channel was closed before it could be completed.

Equations

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

def Std.Channel.recvSelector {α : Type} [Inhabited α] (ch : Channel α) : Internal.IO.Async.Selector α

Creates a Selector that resolves once ch has data available and provides that that data.

Equations

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

partial def Std.Channel.forAsync {α : Type} [Inhabited α] (f : α → BaseIO Unit) (ch : Channel α) (prio : Task.Priority := Task.Priority.default) : BaseIO (Task Unit)

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

Note that if this function is called twice, each message will only arrive at exactly one invocation.

@[inline]

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

This function is a no-op and just a convenient way to expose the synchronous API of the channel.

Equations

• ch.sync = ch

@[inline]

def Std.Channel.Sync.new {α : Type} (capacity : Option Nat := none) : BaseIO (Sync α)

Create a new channel, if:

• capacity is none it will be unbounded (the default)
• capacity is some 0 it will always force a rendezvous between sender and receiver
• capacity is some n with n > 0 it will use a buffer of size n and begin blocking once it is filled

Equations

• Std.Channel.Sync.new capacity = Std.Channel.new capacity

@[inline]

def Std.Channel.Sync.trySend {α : Type} (ch : Sync α) (v : α) : BaseIO Bool

Try to send a value to the channel, if this can be completed right away without blocking return true, otherwise don't send the value and return false.

Equations

• ch.trySend v = Std.Channel.trySend ch v

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

Send a value through the channel, blocking until the transmission could be completed.

Equations

• ch.send v = do let __do_lift ← Std.Channel.send ch v IO.wait __do_lift

@[inline]

def Std.Channel.Sync.tryRecv {α : Type} (ch : Sync α) : BaseIO (Option α)

Try to receive a value from the channel, if this can be completed right away without blocking return some value, otherwise return none.

Equations

• ch.tryRecv = Std.Channel.tryRecv ch

def Std.Channel.Sync.recv {α : Type} [Inhabited α] (ch : Sync α) : BaseIO α

Receive a value from the channel, blocking until the transmission could be completed.

Equations

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

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

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

Equations

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