Streamly.Data.StreamK

If total iterations are kept the same, each increase in the nesting level increases the cost by roughly 1.5 times.

See bfsConcatFor for detailed documentation.

>>> bfsConcatMap = flip StreamK.bfsConcatFor

See fairConcatFor for detailed documentation.

>>> fairConcatMap = flip StreamK.fairConcatFor

Perform a concatMap using a specified concat strategy. The first argument specifies a merge or concat function that is used to merge the streams generated by the map function.

For example, interleaving n streams in a left biased manner:

>>> lists = mk [[1,5],[2,6],[3,7],[4,8]]>>> un $ StreamK.concatMapWith StreamK.interleave mk lists[1,2,5,3,6,4,7,8]

For a fair interleaving example see bfsConcatMap and mergeMapWith.

Map a stream generating function on each element of a stream and concatenate the results. This is the same as the bind function of the monad instance. It is just a flipped concatMap but more convenient to use for nested use case, feels like an imperative for loop.

>>> concatFor = flip StreamK.concatMap

A concatenating for loop:

>>> :{un $    StreamK.concatFor (mk [1,2,3]) $ \x ->      StreamK.fromPure x:}[1,2,3]

Nested concatenating for loops:

>>> :{un $    StreamK.concatFor (mk [1,2,3]) $ \x ->     StreamK.concatFor (mk [4,5,6]) $ \y ->      StreamK.fromPure (x, y):}[(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]

While concatFor flattens a stream of streams in a depth first manner, bfsConcatFor flattens it in a breadth-first manner. It yields one item from the first stream, then one item from the next stream and so on. In nested loops it has the effect of prioritizing the new outer loop iteration over the inner loops, thus inverting the looping. Given a stream of three streams:

1. [1,2,3]
2. [4,5,6]
3. [7,8,9]

The resulting stream is (1,4,7),(2,5,8),(3,6,9). The parenthesis are added to emphasize the iterations.

For example:

>>> stream = mk [[1,2,3],[4,5,6],[7,8,9]]>>> :{ un $     StreamK.bfsConcatFor stream $ \x ->         StreamK.fromStream $ Stream.fromList x:}[1,4,7,2,5,8,3,6,9]

Compare with concatForWith interleave which explores the depth exponentially more compared to the breadth, such that each stream yields twice as many items compared to the next stream.

See also the equivalent fused version Data.Stream.unfoldEachInterleave.

fairConcatFor is like concatFor but traverses the depth and breadth of nesting equally. Therefore, the outer and the inner loops in a nested loop get equal priority. It can be used to nest infinite streams without starving outer streams due to inner ones.

Given a stream of three streams:

1. [1,2,3]
2. [4,5,6]
3. [7,8,9]

Here, outer loop is the stream of streams and the inner loops are the individual streams. The traversal sweeps the diagonals in the above grid to give equal chance to outer and inner loops. The resulting stream is (1),(2,4),(3,5,7),(6,8),(9), diagonals are parenthesized for emphasis.

Looping

A single stream case is equivalent to concatFor:

>>> un $ StreamK.fairConcatFor (mk [1,2]) $ \x -> StreamK.fromPure x[1,2]

Fair Nested Looping

Multiple streams nest like for loops. The result is a cross product of the streams. However, the ordering of the results of the cross product is such that each stream gets consumed equally. In other words, inner iterations of a nested loop get the same priority as the outer iterations. Inner iterations do not finish completely before the outer iterations start.

>>> :{un $ do    StreamK.fairConcatFor (mk [1,2,3]) $ \x ->     StreamK.fairConcatFor (mk [4,5,6]) $ \y ->      StreamK.fromPure (x, y):}[(1,4),(1,5),(2,4),(1,6),(2,5),(3,4),(2,6),(3,5),(3,6)]

Nesting Infinite Streams

Example with infinite streams. Print all pairs in the cross product with sum less than a specified number.

>>> :{Stream.toList $ Stream.takeWhile (\(x,y) -> x + y < 6) $ StreamK.toStream $ StreamK.fairConcatFor (mk [1..]) $ \x ->    StreamK.fairConcatFor (mk [1..]) $ \y ->     StreamK.fromPure (x, y):}[(1,1),(1,2),(2,1),(1,3),(2,2),(3,1),(1,4),(2,3),(3,2),(4,1)]

How the nesting works?

If we look at the cross product of [1,2,3], [4,5,6], the streams being combined using fairConcatFor are the following sequential loop iterations:

(1,4) (1,5) (1,6) -- first iteration of the outer loop
(2,4) (2,5) (2,6) -- second iteration of the outer loop
(3,4) (3,5) (3,6) -- third iteration of the outer loop

The result is a triangular or diagonal traversal of these iterations:

[(1,4),(1,5),(2,4),(1,6),(2,5),(3,4),(2,6),(3,5),(3,6)]

Non-Termination Cases

If one of the two interleaved streams does not produce an output at all and continues forever then the other stream will never get scheduled. This is because a stream is unscheduled only after it produces an output. This can lead to non-terminating programs, an example is provided below.

>>> :{oddsIf x = mk (if x then [1,3..] else [2,4..])filterEven x = if even x then StreamK.fromPure x else StreamK.nil:}

>>> :{evens =    StreamK.fairConcatFor (mk [True,False]) $ \r ->     StreamK.concatFor (oddsIf r) filterEven:}

The evens function does not terminate because, when r is True, the nested concatFor is a non-productive infinite loop, therefore, the outer loop never gets a chance to generate the False value.

But the following refactoring of the above code works as expected:

>>> :{mixed =     StreamK.fairConcatFor (mk [True,False]) $ \r ->         StreamK.concatFor (oddsIf r) StreamK.fromPure:}

>>> evens = StreamK.fairConcatFor mixed filterEven>>> Stream.toList $ Stream.take 3 $ StreamK.toStream evens[2,4,6]

This works because in mixed both the streams being interleaved are productive.

Care should be taken how you write your program, keep in mind the scheduling implications. To avoid such scheduling problems in serial interleaving, you can use concurrent interleaving instead i.e. parFairConcatFor. Due to concurrent threads the other branch will make progress even if one is an infinite loop producing nothing.

Logic Programming

Streamly provides all operations for logic programming. It provides functionality equivalent to LogicT type from the logict package. The MonadLogic operations can be implemented using the available stream operations. For example, uncons is msplit, interleave corresponds to the interleave operation of MonadLogic, fairConcatFor is the fair bind (>>-) operation.

Related Operations

concatForWith interleave is another way to interleave two serial streams. In this case, the inner loop iterations get exponentially more priority over the outer iterations of the nested loop. This is biased towards the inner loops - this is exactly how the logic-t and list-t implementation of fair bind works.

Like concatFor but maps an effectful function. It allows conveniently mixing monadic effects with streams.

>>> import Control.Monad.IO.Class (liftIO)>>> :{un $    StreamK.concatForM (mk [1,2,3]) $ \x -> do      liftIO $ putStrLn (show x)      pure $ StreamK.fromPure x:}123[1,2,3]

Nested concatentating for loops:

>>> :{un $    StreamK.concatForM (mk [1,2,3]) $ \x -> do      liftIO $ putStrLn (show x)      pure $ StreamK.concatFor (mk [4,5,6]) $ \y ->        StreamK.fromPure (x, y):}123[(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]

Like bfsConcatFor but maps a monadic function.

Like fairConcatFor but maps a monadic function.

Combine streams in pairs using a binary combinator, the resulting streams are then combined again in pairs recursively until we get to a single combined stream. The composition would thus form a binary tree.

For example, 'mergeMapWith interleave' gives the following result:

>>> lists = mk [[1,2,3],[4,5,6],[7,8,9],[10,11,12]]>>> un $ StreamK.mergeMapWith StreamK.interleave mk lists[1,7,4,10,2,8,5,11,3,9,6,12]

The above example is equivalent to the following pairings:

>>> pair1 = mk [1,2,3] `StreamK.interleave` mk [4,5,6]>>> pair2 = mk [7,8,9] `StreamK.interleave` mk [10,11,12]>>> un $ pair1 `StreamK.interleave` pair2[1,7,4,10,2,8,5,11,3,9,6,12]

If the number of streams being combined is not a power of 2, the binary tree composed by mergeMapWith is not balanced, therefore, the output may not look fairly interleaved, it will be biased towards the unpaired streams:

>>> lists = mk [[1,2,3],[4,5,6],[7,8,9]]>>> un $ StreamK.mergeMapWith StreamK.interleave mk lists[1,7,4,8,2,9,5,3,6]

An efficient merge sort can be implemented by using mergeBy as the combining function:

>>> combine = StreamK.mergeBy compare>>> un $ StreamK.mergeMapWith combine StreamK.fromPure (mk [5,1,7,9,2])[1,2,5,7,9]

Caution: the stream of streams must be finite

Pre-release

Sort the input stream using a supplied comparison function.

Sorting can be achieved by simply:

>>> sortBy cmp = StreamK.mergeMapWith (StreamK.mergeBy cmp) StreamK.fromPure

However, this combinator uses a parser to first split the input stream into down and up sorted segments and then merges them to optimize sorting when pre-sorted sequences exist in the input stream.

O(n) space

Like Streamly.Data.Stream.Streamly.Data.Stream.handle but with one significant difference, this function observes exceptions from the consumer of the stream as well.

You can also convert StreamK to Stream and use exception handling from Stream module:

>>> handle f s = StreamK.fromStream $ Stream.handle (\e -> StreamK.toStream (f e)) (StreamK.toStream s)

Like Streamly.Data.Stream.Streamly.Data.Stream.bracketIO but with one significant difference, this function observes exceptions from the consumer of the stream as well. Therefore, it cleans up the resource promptly when the consumer encounters an exception.

You can also convert StreamK to Stream and use resource handling from Stream module:

>>> bracketIO bef aft bet = StreamK.fromStream $ Stream.bracketIO bef aft (StreamK.toStream . bet)

Run a ParserK over a chunked StreamK and return the parse result and the remaining Stream.