Stream Consumers

We can classify stream consumers in the following categories in order of increasing complexity and power:

• Accumulators: Tee/Zip is simple, cannot be appended, good for scanning.
• Terminating folds: Tee/Zip varies based on termination, can be appended, good for scanning, nesting (many) is easy.
• Non-failing (backtracking only) parsers: cannot be used as scans because of backtracking, nesting is complicated because of backtracking, appending is efficient because of no Alternative, Alternative does not make sense because it cannot fail.
• Parsers: Alternative on failure, appending is not as efficient because of buffering for Alternative.

First two are represented by the Fold type and the last two by the Parser type.

Folds that never terminate (Accumulators)

An Accumulator is the simplest type of fold, it never fails and never terminates. It can always accept more inputs (never terminates) and the accumulator is always valid. For example sum. Traditional Haskell left folds like foldl are accumulators.

Accumulators can be composed in parallel where we distribute the input stream to all accumulators. Since accumulators never terminate they cannot be appended.

An accumulator can be represented as:

data Fold0 m a b =
  forall s. Fold0
     (s -> a -> m s) -- step
     (m s)           -- initial
     (s -> m b)      -- extract

This is just a traditional left fold, compare with foldl. The driver of the fold would call initial at the beginning and then keep accumulating inputs into its result using step and finally extract the result using extract.

Folds that terminate after one or more input

Terminating folds are accumulators that can terminate, like accumulators they do not fail. Once a fold terminates it no longer accepts any more inputs. Terminating folds can be appended, the next fold can be applied after the first one terminates. Because they cannot fail, they do not need backtracking.

The take operation is an example of a terminating fold. It terminates after consuming n items. Coupled with an accumulator (e.g. sum) it can be used to process the stream into chunks of fixed size.

A terminating fold can be represented as:

data Step s b
    = Partial !s -- the fold can accept more input
    | Done !b    -- the fold is done

data Fold1 m a b =
  forall s. Fold1
     (s -> a -> m (Step s b)) -- step
     (m s)                    -- initial
     (s -> m b)               -- extract

The fold driver stops driving the fold as soon as the fold returns a Done. extract is required only if the fold has not stopped yet and the input ends. extract can never be called if the fold is Done.

Notice that the initial of Fold1 type does not return a Step type, therefore, it cannot say Done in initial. It always has to consume at least one element before it can say Done for termination, via the step function.

Folds that terminate after 0 or more input

The Fold1 type makes combinators like take 0 impossible to implement because they need to terminate even before they can consume any elements at all. Implementing this requires the initial function to be able to return Done.

data Fold m a b =
  forall s. Fold
     (s -> a -> m (Step s b)) -- step
     (m (Step s b))           -- initial
     (s -> m b)               -- extract

This is also required if we want to compose terminating folds using an Applicative or Monadic composition. pure needs to yield an output without having to consume an input.

initial now has the ability to terminate the fold without consuming any input based on the state of the monad.

In some cases it does not make sense to use a fold that does not consume any items at all, and it may even lead to an infinite loop. It might make sense to use a Fold1 type for such cases because it guarantees to consume at least one input, therefore, guarantees progress. For example, in classifySessionsBy or any other splitting operations it may not make sense to pass a fold that never consumes an input. However, we do not have a separate Fold1 type for the sake of simplicity of types/API.

Adding this capability adds a certain amount of complexity in the implementation of fold combinators. initial has to always handle two cases now. We could potentially not implement this in folds to keep fold implementation simpler, and these use cases can be transferred to the parser type. However, it would be a bit inconvenient to not have a take operation or to not be able to use `take 0` if we have it. Also, applicative and monadic composition of folds would not be possible.

Terminating Folds with backtracking

Consider the example of takeWhile operation, it needs to inspect an element for termination decision. However, it does not consume the element on which it terminates. To implement takeWhile a terminating fold will have to implement a way to return the unconsumed input to the fold driver.

Single element leftover case is quite common and its easy to implement it in terminating folds by adding a Done1 constructor in the Step type which indicates that the last element was not consumed by the fold. The following additional operations can be implemented as terminating folds if we do that.

takeWhile
groupBy
wordBy

However, it creates several complications. The most important one is that we cannot use such folds for scanning. We cannot backtrack after producing an output in a scan.

Nested backtracking

Nesting of backtracking folds increases the amount of backtracking required exponentially.

For example, the combinator many inner outer applies the outer fold on the input stream and applies the inner fold on the results of the outer fold.

many :: Monad m => Fold m b c -> Fold m a b -> Fold m a c

If the inner fold itself returns a Done1 then we need to backtrack all the elements that have been consumed by the outer fold to generate that value. We need backtracking of more than one element.

Arbitrary backtracking requires arbitrary buffering. However, we do not want to buffer unconditionally, only if the buffer is needed. One way to do this is to use a Continue constructor like parsers. When we have nested folds, the top level fold always returns a Continue to the driver until an output is generated by it, this means the top level driver keeps buffering until an output is generated via Partial or Done. Intermediate level Continue keep propagating up to the top level.

Parallel backtracking

In compositions like Alternative and Distributive we may have several branches. Each branch can backtrack independently. We need to keep the input as long as any of the branches need it. We can use a single copy of the buffer and maintain it based on all the branches, or we can make each branch have its own buffer. The latter approach may be simpler to implement. Whenever we branch we can introduce an independent buffer for backtracking. Or we can use a newtype that allows branched composition to handle backtracking.

Implementation Approach

To avoid these issues we can enforce, by using types, that the collecting folds can never return a leftover. This leads us to define a type that can never return a leftover. The use cases of single leftover can be transferred to parsers where we have general backtracking mechanism and single leftover is just a special case of backtracking.

This means: takeWhile, groupBy, wordBy would be implemented as parsers.

A proposed design is to use the same Step type with Error in Folds as well as Parsers. Folds won't use the Error constructor and even if they use, it will be equivalent to just throwing an error. They won't have an alternative.

Because of the complexity of implementing a distributive composition in presence of backtracking we could possibly have a type without backtracking but with the Continue constructor, and use either the Parser type or another type for backtracking.

Folds with an additional input

The Fold type does not allow a dynamic input to be used to generate the initial value of the fold accumulator. We can extend the type further to allow that:

data Refold m i a b =
  forall s. Refold
     (s -> a -> m (Step s b)) -- step
     (i -> m (Step s b))      -- initial
     (s -> m b)               -- extract

Parsers

The next upgrade after terminating folds with a leftover are parsers. Parsers are terminating folds that can fail and backtrack. Parsers can be composed using an alternative style composition where they can backtrack and apply another parser if one parser fails. satisfy is a simple example of a parser, it would succeed if the condition is satisfied and it would fail otherwise, on failure an alternative parser can be used on the same input.

We add Error and Continue to the Step type of fold. Continue is to skip producing an output or to backtrack. We also add the ability to backtrack in Partial and Done.:

Also extract now needs to be able to express an error. We could have it return the Step type as well but that makes the implementation more complicated.

data Step s b =
      Partial Int s   -- partial result and how much to backtrack
    | Done Int b      -- final result and how much to backtrack
    | Continue Int s  -- no result and how much to backtrack
    | Error String    -- error

data Parser a m b =
  forall s. Fold
     (s -> a -> m (Step s b))   -- step
     (m (Step s b))             -- initial
     (s -> m (Either String b)) -- extract

Types for Stream Consumers

We do not have a separate type for accumulators. Terminating folds are a superset of accumulators and to avoid too many types we represent both using the same type, Fold.

We do not club the leftovers functionality with terminating folds because of the reasons explained earlier. Instead combinators that require leftovers are implemented as the Parser type. This is a sweet spot to balance ease of use, type safety and performance. Using separate Accumulator and terminating fold types would encode more information in types but it would make ease of use, implementation, maintenance effort worse. Combining Accumulator, terminating folds and Parser into a single Parser type would make ease of use even better but type safety and performance worse.

One of the design requirements that we have placed for better ease of use and code reuse is that Parser type should be a strict superset of the Fold type i.e. it can do everything that a Fold can do and more. Therefore, folds can be easily upgraded to parsers and we can use parser combinators on folds as well when needed.

Fold Design

A fold is represented by a collection of "initial", "step" and "extract" functions. The "initial" action generates the initial state of the fold. The state is internal to the fold and maintains the accumulated output. The "step" function is invoked using the current state and the next input value and results in a Partial or Done. A Partial returns the next intermediate state of the fold, a Done indicates that the fold has terminated and returns the final value of the accumulator.

Every Partial indicates that a new accumulated output is available. The accumulated output can be extracted from the state at any point using "extract". "extract" can never fail. A fold returns a valid output even without any input i.e. even if you call "extract" on "initial" state it provides an output. This is not true for parsers.

In general, "extract" is used in two cases:

• When the fold is used as a scan extract is called on the intermediate state every time it is yielded by the fold, the resulting value is yielded as a stream.
• When the fold is used as a regular fold, extract is called once when we are done feeding input to the fold.

Alternate Designs

An alternate and simpler design would be to return the intermediate output via Partial along with the state, instead of using "extract" on the yielded state and remove the extract function altogether.

This may even facilitate more efficient implementation. Extract from the intermediate state after each yield may be more costly compared to the fold step itself yielding the output. The fold may have more efficient ways to retrieve the output rather than stuffing it in the state and using extract on the state.

However, removing extract altogether may lead to less optimal code in some cases because the driver of the fold needs to thread around the intermediate output to return it if the stream stops before the fold could return Done. When using this approach, the parseMany (FL.take filesize) benchmark shows a 2x worse performance even after ensuring everything fuses. So we keep the "extract" approach to ensure better perf in all cases.

But we could still yield both state and the output in Partial, the output can be used for the scan use case, instead of using extract. Extract would then be used only for the case when the stream stops before the fold completes.

Monoids

Monoids allow generalized, modular folding. The accumulators in this module can be expressed using mconcat and a suitable Monoid. Instead of writing folds we can write Monoids and turn them into folds.

>>> :m
>>> :set -XFlexibleContexts
>>> import Control.Monad (void)
>>> import qualified Data.Foldable as Foldable
>>> import Data.Function ((&))
>>> import Data.Functor.Identity (Identity, runIdentity)
>>> import Data.IORef (newIORef, readIORef, writeIORef)
>>> import Data.Maybe (fromJust, isJust)
>>> import Data.Monoid (Endo(..), Last(..), Sum(..))

>>> import Streamly.Data.Array (Array)
>>> import Streamly.Data.Fold (Fold, Tee(..))
>>> import Streamly.Data.Stream (Stream)

>>> import qualified Streamly.Data.Array as Array
>>> import qualified Streamly.Data.Fold as Fold
>>> import qualified Streamly.Data.MutArray as MutArray
>>> import qualified Streamly.Data.Parser as Parser
>>> import qualified Streamly.Data.Stream as Stream
>>> import qualified Streamly.Data.StreamK as StreamK
>>> import qualified Streamly.Data.Unfold as Unfold

For APIs that have not been released yet.

>>> import qualified Streamly.Internal.Data.Fold as Fold
>>> import qualified Streamly.Internal.Data.Fold.Window as FoldW