| Safe Haskell | None |
|---|---|
| Language | GHC2024 |
Circuit.Agent.Ends
Description
Effectful Poles constructors and queueing strategies.
This module lives in circuits-agent because it needs STM. Core
Poles is pure in the base arrow; the queue implementations here
are one possible effectful instantiation, placed beside their consumers
rather than forcing the core library to depend on stm.
Synopsis
- data Queue a
- data ChannelPolicy a
- = Linear
- | SingleSlot
- | SwapOne
- | LatestValue a
- | BoundedN Int
- | NewestN Int
- openChannel :: ChannelPolicy a -> IO (Poles (K IO) a a)
- openChannelSTM :: ChannelPolicy a -> STM (Poles (K STM) a a)
- openLinearChannel :: IO (Poles (K IO) a a)
- openLinearChannelSTM :: STM (Poles (K STM) a a)
- data HaltChannel (p :: ChannelPolicy a) where
- HaltChannel :: forall a (p :: ChannelPolicy a). IsLinear p => Poles (K STM) a a -> HaltChannel p
- type family IsLinear (p :: ChannelPolicy a) where ...
- openHaltChannel :: STM (HaltChannel ('Linear :: ChannelPolicy a))
- writeHaltChannel :: forall a (p :: ChannelPolicy a). HaltChannel p -> a -> STM ()
- readHaltChannel :: forall a (p :: ChannelPolicy a). HaltChannel p -> STM a
- openSTM :: Queue a -> STM (Poles (K STM) a a)
- openIO :: Queue a -> IO (Poles (K IO) a a)
- pipeEnds :: Poles (K IO) a b -> (TQueue b -> IO (Poles (K IO) b c)) -> IO (Poles (K IO) a c, IO ())
Queue strategies
How messages are queued between producer and consumer.
Constructors
| Unbounded | Unbounded FIFO queue. |
| Bounded Int | Bounded FIFO with backpressure (write blocks when full). |
| Single | Single-slot buffer (write blocks when full). |
| SwapQ | Single-slot buffer, overwrite-on-full. Write always succeeds; read empties. |
| Latest a | Always holds the latest value (overwrites, never blocks). |
| Newest Int | Like |
Channel policies (mediator-configured buffering)
data ChannelPolicy a Source #
A channel policy names the residual mediator that governs an effectful
channel. This is the Track-B relocation of the old Queue annotation:
the policy is a value passed at allocation time, not a field of the
channel type. The constructors match the ?-modality vocabulary from the
B0 spike; Linear is the empty-residual default and the only policy on
which halt marks are safe.
Constructors
| Linear | Unbounded FIFO: empty residual, preserves every token in order. This is the effectful face of a linear process. |
| SingleSlot | Single-slot buffer with backpressure (write blocks when full). |
| SwapOne | Single-slot overwrite: write always succeeds, read empties. A weakening policy that can drop a halt mark. |
| LatestValue a | Always holds the latest value; requires a seed for the first read. A weakening policy suitable for diagnostics, not for halt marks. |
| BoundedN Int | Bounded FIFO with backpressure. |
| NewestN Int | Bounded FIFO dropping oldest when full. A weakening policy suitable for bounded diagnostics. |
Instances
| Eq a => Eq (ChannelPolicy a) Source # | |
Defined in Circuit.Agent.Ends Methods (==) :: ChannelPolicy a -> ChannelPolicy a -> Bool # (/=) :: ChannelPolicy a -> ChannelPolicy a -> Bool # | |
| Show a => Show (ChannelPolicy a) Source # | |
Defined in Circuit.Agent.Ends Methods showsPrec :: Int -> ChannelPolicy a -> ShowS # show :: ChannelPolicy a -> String # showList :: [ChannelPolicy a] -> ShowS # | |
openChannel :: ChannelPolicy a -> IO (Poles (K IO) a a) Source #
Open a channel policy as IO Poles.
openChannelSTM :: ChannelPolicy a -> STM (Poles (K STM) a a) Source #
Open a channel policy as STM Poles.
Linear default channels
openLinearChannel :: IO (Poles (K IO) a a) Source #
Open a linear channel as IO Poles.
Linear is the default policy: unbounded FIFO, empty residual, preserves
every token in order. This is the effectful face of a linear process.
Halt-mark channels (compile-time linearity witness)
data HaltChannel (p :: ChannelPolicy a) where Source #
A channel statically known to be linear.
The index p :: ChannelPolicy a is checked by IsLinear at construction
time. Attempting to build a HaltChannel with a non-linear policy fails
to typecheck.
Constructors
| HaltChannel :: forall a (p :: ChannelPolicy a). IsLinear p => Poles (K STM) a a -> HaltChannel p |
type family IsLinear (p :: ChannelPolicy a) where ... Source #
Type-level witness that a channel policy is linear.
Only Linear is allowed to carry halt marks; any other policy produces a
compile-time type error.
Equations
| IsLinear ('Linear :: ChannelPolicy a) = () | |
| IsLinear (p :: ChannelPolicy a) = TypeError ('Text "only 'Linear' channels can carry halt marks") :: Constraint |
openHaltChannel :: STM (HaltChannel ('Linear :: ChannelPolicy a)) Source #
Open a halt-mark channel. This is openLinearChannelSTM with a
type-level certificate.
writeHaltChannel :: forall a (p :: ChannelPolicy a). HaltChannel p -> a -> STM () Source #
Write a token to a halt-mark channel.
readHaltChannel :: forall a (p :: ChannelPolicy a). HaltChannel p -> STM a Source #
Read a token from a halt-mark channel.
STM Ends
openSTM :: Queue a -> STM (Poles (K STM) a a) Source #
Open a queue strategy as STM Poles.
Allocates STM primitives and returns a matched pair of ends sharing
the same mutable channel. Both ends live in STM, so you can compose
operations across channels in a single atomically block.
IO Ends
openIO :: Queue a -> IO (Poles (K IO) a a) Source #
Open a queue strategy as IO Poles.
Like openSTM, but each primitive operation is wrapped in its own
atomically. You cannot batch multiple writes or a write-plus-read
into a single STM transaction; for that use openSTM and wrap in
atomically yourself.
Honest composition
pipeEnds :: Poles (K IO) a b -> (TQueue b -> IO (Poles (K IO) b c)) -> IO (Poles (K IO) a c, IO ()) Source #
Honest sequential composition of two allocated ends via an intermediate queue and a pump.
pipeEnds e1 makeE2 allocates a queue of b values, builds the right end
around that queue with makeE2, and starts a pump that moves values from
e1 into the right end. The returned Poles uses e1 for input and the
built right end for output; the close action cancels the pump.
This is the coend-style composition that composePoles cannot express: the
intermediate carrier is a real queue (the residual's home) rather than the
unit type, so a multi-read consumer can accumulate inputs before emitting.