Effectful streams in coalgebraic encoding.

A stream consists of an internal state s, and a step function. This step can make use of an effect in m (which is often a monad), alter the state, and return a result value. Its semantics is continuously outputting values of type b, while performing side effects in m.

A coalgebraic encoding was chosen instead of the direct recursion known from e.g. list-transformer, dunai, machines, streaming, ..., because the coalgebraic encoding is much more amenable to compiler optimizations than the coalgebraic encoding, which is:

  data StreamRecursiveT m b = StreamRecursiveT (m (b, StreamRecursiveT m b))

When two streams are composed, GHC can often optimize the combined step function, resulting in a faster streams than what the coalgebraic encoding can ever achieve, because the coalgebraic encoding has to step through every continuation. Put differently, the compiler can perform static analysis on the state types of initially encoded state machines, while the coalgebraic encoding knows its state only at runtime.

This performance gain comes at a peculiar cost: Recursive definitions of streams are not possible, e.g. an equation like: fixA stream = stream * fixA stream This is impossible since the stream under definition itself appears in the definition body, and thus the internal state type would be recursively defined, which GHC doesn't allow: Type level recursion is not supported in existential types. An stream defined thusly will typically hang and/or leak memory, trying to build up an infinite type at runtime.

It is nevertheless possible to define streams recursively, but one needs to first identify the recursive definition of its state type. Then for the greatest generality, fixStream and fixStream' can be used, and some special cases are covered by functions such as fixA, parallely, many and some.

Initialise with an internal state, update the state and produce output without side effects.

Like unfold, but output the current state.

Constantly perform the same effect, without remembering a state.

Like fmap or rmap, but the postcomposed function may have an effect in m.

Translate a coalgebraically encoded stream into a recursive one.

This is usually a performance penalty.

Translate a recursive stream into a coalgebraically encoded one.

The internal state is the stream itself.

Call the monadic action once on the first tick and provide its result indefinitely.

Hoist a stream along a monad morphism, by applying said morphism to the step function.

This is like mmorph's hoist, but it doesn't require a Monad constraint on m2.

Perform one step of a stream, resulting in an updated stream and an output value.

Run a stream with trivial output.

If the output of a stream does not contain information, all of its meaning is in its effects. This function runs the stream indefinitely. Since it will never return with a value, this function also has no output (its output is void). The only way it can return is if m includes some effect of termination, e.g. Maybe or Either could terminate with a Nothing or Left value, or IO can raise an exception.

Run a stream, collecting the outputs in a lazy, infinite list.

Change the output type and effect of a stream without changing its state type.

Buffer the output of a stream, returning one value at a time.

This function lets a stream control the speed at which it produces data, since it can decide to produce any amount of output at every step.

At each step, duplicate the m effect of the current step to the output.

This is useful if m has some means of static analysis, or if you want to re-perform the effects.

Streams with exceptions are Applicative in the exception type.

Run the first stream until it throws a function as an exception, then run the second one. If the second one ever throws an exception, apply the function thrown by the first one to it.

Execute the stream until it throws an exception, then restart it.

One might be tempted to define this function recursively with applyExcept, but this would result in a runtime error, trying to define an infinite state.

Whenever an exception occurs, output it and retry on the next step.

Run the first stream until it throws an exception. If the exception is Right, throw it immediately. If it is Left, run the second stream until it throws a function, which is then applied to the first exception.

Recursively define a stream from a recursive definition of the state, and of the step function.

If you want to define a stream recursively, this is not possible directly. For example, consider this definition: loops :: Monad m => StreamT m [Int] loops = (:) $ unfold_ 0 (+ 1) * loops The defined value loops contains itself in its definition. This means that the internal state type of loops must itself be recursively defined. But GHC cannot do this automatically, because type level and value level are separate. Instead, we need to spell out the type level recursion explicitly with a type constructor, over which we will take the fixpoint.

In this example, we can figure out from the definitions that: 1. unfold_ 0 (+ 1) has 0 :: Int as state 2. (:) does not change the state 3. <*> takes the product of both states

So the internal state s of loops must satisfy the equation s = (Int, s). If the recursion is written as above, it tries to compute the infinite tuple (Int, (Int, (Int, ...))), which hangs. Instead, we need to define a type operator over which we take the fixpoint:

-- You need to write this:
data Loops x = Loops Int x

-- The library supplies:
data Fix f = Fix f (Fix f)
type LoopsState = Fix Loops

We can then use fixStream to define the recursive definition of loops. For this, we have to to tediously inline the definitions of unfold_, (:), and <*>, until we arrive at an explicit recursive definition of the state and the step function of loops, separately. These are the two arguments of fixStream.

loops :: Monad m => StreamT m [Int]
loops = fixStream (Loops 0) $ fixStep (Loops n fixState) -> do
  Result s' a <- fixStep fixState
  return $ Result (Loops (n + 1) s') a

A generalisation of fixStream where the step definition is allowed to depend on the state.

The solution to the equation fixA stream = stream * fixA stream.

Such a fix point operator needs to be used instead of the above direct definition because recursive definitions of streams loop at runtime due to the coalgebraic encoding of the state.

Lift the monad of a stream into a transformer.

Continuously interpret a first order effect.

Several types are relevant here:

• sig: An effect signature functor, that encodes one effect. For example, Either e for raising exceptions of type e, or (w, ) for a logging effect.
• eff: A monad that carries the effect. This can be a monad transformer stack including a transformer corresponding to sig, such as ExceptT for Either. It can also be the Eff monad of an effect library such as polysemy, bluefin, effectful and so on.
• m: The underlying monad in which the interpretation is performed, think "eff without the effects from sig".

This function takes two functions, one to create effects in eff from the signature, and the other to fully interpret them in m, storing the complete effect information in sig again. It then executes the given automaton, extracting the effect by interpretation, and sending it back in. The execution semantics is that of the monad eff, while the pure effect of the whole computation is returned in the output, encoded in sig.

For examples, see handleExceptT, handleWriterT and similar functions below.

Execute a stream until it throws an exception, then output the exception forever.

Return the accumulated log at every step alongside the value.

Execute a stream until it stops, then output Nothing forever.