@@ -34,18 +34,21 @@ module Streamly.Tutorial
34 -- ** Eliminating Streams
34 -- ** Eliminating Streams
35 -- $eliminating
35 -- $eliminating
37 -- * Semigroup Style Composition
40 -- * Semigroup Style Composition
38 -- $semigroup
41 -- $semigroup
40 -- ** Serial composition ('<>')
43 -- ** Serial composition ('<>')
41 -- $serial
44 -- $serial
43− -- ** Interleaved composition ('<=>')
46 -- ** Async composition ('<|')
46 -- ** Async composition ('<|')
47 -- $parallel
47 -- $parallel
49+ -- ** Interleaved composition ('<=>')
49 -- ** Parallel composition ('<|>')
52 -- ** Parallel composition ('<|>')
50 -- $fairParallel
53 -- $fairParallel
@@ -58,32 +61,32 @@ module Streamly.Tutorial
58 -- * Transforming Streams
61 -- * Transforming Streams
59 -- $transforming
62 -- $transforming
67 -- ** Monad
64 -- ** Monad
68− -- $listt
65+ -- $monad
70− -- * Alternate Monadic Compositions
71− -- $monadicAlternatives
73− -- ** Serial Iterations
67+ -- *** Serial Iterations ('StreamT')
74 -- $regularSerial
68 -- $regularSerial
76− -- ** Interleaved Iterations
77− -- $interleavedNesting
79− -- ** Concurrent Iterations
70+ -- *** Concurrent Iterations ('AsyncT')
80 -- $concurrentNesting
71 -- $concurrentNesting
73+ -- *** Interleaved Iterations ('InterleavedT')
74+ -- $interleavedNesting
82− -- ** Fairly Concurrent Iterations
76+ -- *** Fairly Concurrent Iterations ('ParallelT')
83 -- $fairlyConcurrentNesting
77 -- $fairlyConcurrentNesting
85− -- * Folding Streams
79+ -- *** Exercise
86− -- $folding
80+ -- $monadExercise
88+ -- * Summary of Compositions
89+ -- $compositionSummary
88 -- * Zipping Streams
91 -- * Zipping Streams
89 -- $zipping
92 -- $zipping
@@ -194,10 +197,11 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
196-- $serial
199-- $serial
197−-- As we have already seen, the '<>' operator composes multiple streams in
201+-- We have already seen, the '<>' operator. It composes two streams in series
198−-- series i.e. the next element is yielded only after the previous one has been
202+-- i.e. the first stream is completely exhausted and then the second stream is
199−-- processed completely. The following example prints the sequence 3, 2, 1 and
203+-- processed. The following example prints the sequence 3, 2, 1 and takes a
200−-- takes a total of 6 seconds because everything is serial:
204+-- total of 6 seconds because everything is serial:
203-- main = runStreamT $ delay 3 <> delay 2 <> delay 1
207-- main = runStreamT $ delay 3 <> delay 2 <> delay 1
@@ -209,10 +213,10 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
211-- $interleaved
215-- $interleaved
212−-- The '<=>' operator interleaves the two computations i.e. it yields one
216+-- The '<=>' operator is serial like '<>' but it interleaves the two streams
213−-- element from the first stream and then one element from the second stream,
217+-- i.e. it yields one element from the first stream and then one element from
214−-- and so on. The following example prints the sequence 1, 3, 2, 4 and takes a
218+-- the second stream, and so on. The following example prints the sequence 1,
215−-- total of 10 seconds because everything is serial:
219+-- 3, 2, 4 and takes a total of 10 seconds because everything is serial:
218-- main = runStreamT $ (delay 1 <> delay 2) \<=> (delay 3 <> delay 4)
222-- main = runStreamT $ (delay 1 <> delay 2) \<=> (delay 3 <> delay 4)
@@ -223,33 +227,13 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
223-- ThreadId 36: Delay 2
227-- ThreadId 36: Delay 2
224-- ThreadId 36: Delay 4
228-- ThreadId 36: Delay 4
231+-- Note that this operator cannot be used to fold infinite containers since it
232+-- requires preserving the state until a stream is finished.
227-- $parallel
234-- $parallel
229−-- The '<|' operator can run computations concurrently, preferring the first
236+-- The '<|' operator can run both computations concurrently, /when needed/.
230−-- computation over the second. The second computation is run concurrently with
231−-- the first only if the first computation is not producing enough output to
232−-- keep the consumer busy otherwise the second computation is run serially
233−-- after the previous one. The number of concurrent threads is adapted
234−-- dynamically based on the pull rate of the consumer of the stream.
236−-- Note that the left bias of the operator '<|' is suggested by its shape.
237−-- It is also an unbalanced version of the fairly parallel operator '<|>'.
239−-- The following example runs all the parallel computations in a single thread
240−-- one after another, because none of them blocks. Note that if the consumer
241−-- were faster than the producer this would start new threads for each
242−-- computation to keep up even if none of them blocks:
245−-- main = runStreamT $ traced (sqrt 9) <| traced (sqrt 16) <| traced (sqrt 25)
253-- In the following example since the first computation blocks we start the
237-- In the following example since the first computation blocks we start the
254-- next one in a separate thread and so on:
238-- next one in a separate thread and so on:
@@ -262,8 +246,10 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
262-- ThreadId 40: Delay 3
246-- ThreadId 40: Delay 3
249+-- This is the concurrent version of the '<>' operator. The computations are
250+-- triggered in the same order as '<>' except that they are concurrent. When
265−-- When we have a tree of computations composed using this operator, the tree
251+-- we have a tree of computations composed using this operator, the tree is
266−-- is traversed in DFS style because it always prefers the left computation:
252+-- traversed in DFS style just like '<>'.
269-- main = runStreamT $ (p 1 <| p 2) <| (p 3 <| p 4)
255-- main = runStreamT $ (p 1 <| p 2) <| (p 3 <| p 4)
@@ -276,28 +262,55 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
279−-- Note that since this operator is left biased it cannot be used when the
265+-- Concurrency provided by this operator is demand driven. The second
266+-- computation is run concurrently with the first only if the first computation
267+-- is not producing enough output to keep the stream consumer busy otherwise
268+-- the second computation is run serially after the previous one. The number of
269+-- concurrent threads is adapted dynamically based on the pull rate of the
270+-- consumer of the stream.
271+-- As you can see, in the following example the computations are run in a
272+-- single thread one after another, because none of them blocks. However, if
273+-- the thread consuming the stream were faster than the producer then it would
274+-- have started parallel threads for each computation to keep up even if none
278+-- main = runStreamT $ traced (sqrt 9) <| traced (sqrt 16) <| traced (sqrt 25)
286+-- Since the concurrency provided by this operator is demand driven it cannot
280−-- composed computations have timers that are relative to each other because
287+-- be used when the composed computations have timers that are relative to each
281−-- all computations may not be started at the same time and therefore timers in
288+-- other because all computations may not be started at the same time and
282−-- all of them may not start at the same time. When relative timing among all
289+-- therefore timers in all of them may not start at the same time. When
283−-- computations is important or when we need to start all computations at once
290+-- relative timing among all computations is important or when we need to start
284−-- for some reason '<|>' must be used instead. However, '<|' is useful in
291+-- all computations at once for some reason '<|>' must be used instead.
285−-- situations when we want to optimally utilize the resources and we know that
292+-- However, '<|' is useful in situations when we want to optimally utilize the
286−-- the computations can run in parallel but we do not care if they actually run
293+-- resources and we know that the computations can run in parallel but we do
287−-- in parallel or not, that decision is left to the scheduler. Also, note that
294+-- not care if they actually run in parallel or not, that decision is left to
288−-- this operator can be used to fold infinite containers in contrast to '<|>',
295+-- the scheduler. Also, note that this operator can be used to fold infinite
289−-- because it does not require us to run all of them at the same time.
296+-- containers in contrast to '<|>', because it does not require us to run all
297+-- of them at the same time.
299+-- The left bias (or the DFS style) of the operator '<|' is suggested by its
300+-- shape. You can also think of this as an unbalanced version of the fairly
301+-- parallel operator '<|>'.
291-- $fairParallel
303-- $fairParallel
293-- The 'Alternative' composition operator '<|>', like '<|', runs the composed
305-- The 'Alternative' composition operator '<|>', like '<|', runs the composed
294−-- computations in parallel. However, unlike '<|' it runs all of the
306+-- computations concurrently. However, unlike '<|' it runs all of the
295-- computations in fairly parallel manner using a round robin scheduling
307-- computations in fairly parallel manner using a round robin scheduling
296−-- mechanism. Note that this should not be used on infinite containers, as it
308+-- mechanism. This can be considered as the concurrent version of the fairly
309+-- interleaved serial operation '<=>'. Note that this cannot be used on
297−-- will lead to an infinite sized scheduling queue.
310+-- infinite containers, as it will lead to an infinite sized scheduling queue.
299−-- The following example sends a query to three search engines and prints the
312+-- The following example sends a query to three search engines in parallel and
300−-- name of the search engine as a response arrives:
313+-- prints the name of the search engine as a response arrives:
303-- import Streamly
316-- import Streamly
@@ -335,16 +348,19 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
335-- * 'foldMapWith' folds like foldWith but also maps a function before folding.
348-- * 'foldMapWith' folds like foldWith but also maps a function before folding.
336-- * 'forEachWith' is like foldMapwith but the container argument comes before
349-- * 'forEachWith' is like foldMapwith but the container argument comes before
337-- the function argument.
350-- the function argument.
351+-- * The 'each' primitive from @Streamly.Prelude@ folds a 'Foldable' container
352+-- using the '<>' operator:
354+-- All of the following are equivalent:
340-- import Streamly
357-- import Streamly
342-- main = do
359-- main = do
343−-- runStreaming . serially $ liftIO $ putStrLn "Hello world!"
344−-- xs \<- toList . serially $ return 1 <> return 2; print xs
345−-- xs \<- toList . serially $ foldWith (<>) (map return [1..10]); print xs
360+-- toList . serially $ foldWith (<>) (map return [1..10]) >>= print
346−-- xs \<- toList . serially $ foldMapWith (<>) return [1..10]; print xs
361+-- toList . serially $ foldMapWith (<>) return [1..10] >>= print
347−-- xs \<- toList . serially $ forEachWith (<>) [1..10] return; print xs
362+-- toList . serially $ forEachWith (<>) [1..10] return >>= print
363+-- toList . serially $ each [1..10] >>= print
350-- $transforming
366-- $transforming
@@ -356,76 +372,163 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
356-- behave exactly the same way as a list transformer. For simplicity of
372-- behave exactly the same way as a list transformer. For simplicity of
357-- illustration we are using streams of pure values in the following examples.
373-- illustration we are using streams of pure values in the following examples.
358-- However, the real application of streams arises when these streams are
374-- However, the real application of streams arises when these streams are
359−-- generated using monadic actions. In the pure stream cases 'StreamT' is
375+-- generated using monadic actions.
360−-- equivalent to the list monad.
362−-- $functor
377+-- $monad
364−-- The simplest transformation on a stream is mapping a function on
379+-- In functional programmer's parlance the 'Monad' instance of 'Streaming'
365−-- all elements of the stream using 'fmap'.
380+-- types implement non-determinism, exploring all possible combination of
381+-- choices from both the streams. From an imperative programmer's point of view
382+-- it behaves like nested loops i.e. for each element in the first stream and
383+-- for each element in the second stream apply the body of the loop. If you are
384+-- familiar with list transformer this behavior is exactly the same as that of
385+-- a list transformer.
387+-- Just like we saw in sum style compositions earlier, monadic composition also
388+-- has multiple variants each of which exactly corresponds to one of the sum
389+-- style composition variant.
393+-- When we interpret the monadic composition as 'StreamT' we get a standard
394+-- list transformer like serial composition.
368−-- import Streamly
397+-- main = runStreamT $ do
402+-- ThreadId 30: Delay 3
403+-- ThreadId 30: Delay 2
404+-- ThreadId 30: Delay 1
407+-- As you can see the code after the @each@ statement is run three times, once
408+-- for each value of @x@. All the three iterations are serial and run in the
409+-- same thread one after another. When compared to imperative programming, this
410+-- can also be viewed as a @for@ loop with three iterations.
412+-- When multiple streams are composed using this style they nest in a DFS
413+-- manner i.e. nested iterations of an iteration are executed before we proceed
414+-- to the next iteration at higher level. This behaves just like nested @for@
415+-- loops in imperative programming.
371−-- let nums = each [1..10]
372−-- let strings = fmap show nums
373−-- (toList $ serially $ fmap putStrLn strings) >>= sequence
375−
418+-- main = runStreamT $ do
376−-- $applicative
419+-- x <- each [1,2]
421+-- liftIO $ putStrLn $ show (x, y)
378−-- In functional programmer's parlance the 'Applicative' instance of
430+-- You will also notice that this is the monadic equivalent of the sum style
379−-- 'Streaming' types implement non-determinism, exploring all possible
431+-- composition using '<>'.
380−-- combination of choices from both the streams. From an imperative
432+
381−-- programmer's point of view the applicative behaves like nested loops i.e.
433+-- $concurrentNesting
382−-- for each element in the first argument and for each element in the second
383−-- argument apply the body of the loop. Note that this behavior of applicative
384−-- instance of all 'Streaming' types is exactly the same as that of a list
387−-- The following example prints all combinations of size, color and shape:
435+-- When we interpret the monadic composition as 'AsyncT' we get a /concurrent/
436+-- list-transformer like composition. Multiple monadic continuations (or loop
437+-- iterations) may be started concurrently. Concurrency is demand driven
438+-- i.e. more concurrent iterations are started only if the previous iterations
439+-- are not able to produce enough output for the consumer of the output stream.
440+-- This is the concurrent version of 'StreamT'.
390−-- main = (toList . serially $ (,,) \<$> sizes \<*> colors \<*> shapes) >>= print
443+-- main = runAsyncT $ do
448+-- ThreadId 40: Delay 1
449+-- ThreadId 39: Delay 2
450+-- ThreadId 38: Delay 3
392−-- where
453+-- As you can see the code after the @each@ statement is run three times, once
454+-- for each value of @x@. All the three iterations are concurrent and run in
455+-- different threads. The iteration with least delay finishes first. When
456+-- compared to imperative programming, this can be viewed as a @for@ loop
457+-- with three concurrent iterations.
459+-- Concurrency is demand driven just as in the case of '<|'. When multiple
460+-- streams are composed using this style the iterations are triggered in a DFS
461+-- manner just like 'StreamT' i.e. nested iterations are executed before we
462+-- proceed to the next iteration at higher level. However, unlike 'StreamT'
463+-- more than one iterations may be started concurrently, and based on the
464+-- demand from the consumer.
467+-- main = runAsyncT $ do
394−-- sizes = each [1, 2, 3]
468+-- x <- each [1,2]
395−-- colors = each ["red", "green", "blue"]
469+-- y <- each [3,4]
396−-- shapes = each ["triangle", "square", "circle"]
470+-- liftIO $ putStrLn $ show (x, y)
479+-- You will notice that this is the monadic equivalent of the '<|' style
480+-- sum composition. The same caveats apply to this as the '<|' operation.
399−-- $listt
482+-- $interleavedNesting
401−-- The monadic composition of 'Streaming' types is similar to 'Applicative'
484+-- When we interpret the monadic composition as 'InterleavedT' we get a serial
402−-- composition and just the same as that of list transformer. It is like nested
485+-- but fairly interleaved list-transformer like composition. The monadic
403−-- loops in the imperative parlance. The monadic continuation is executed for
486+-- continuations or iterations of the outer loop are fairly interleaved with
404−-- each element in a stream.
487+-- the continuations or iterations of the inner loop.
405−-- For example the following program repeatedly reads strings from standard
406−-- input and echoes them to the standard output. This is like one infinite
410−-- import Streamly
490+-- main = runInterleavedT $ do
411−-- import Data.Semigroup (cycle1)
491+-- x <- each [1,2]
412−--
492+-- y <- each [3,4]
413−-- main = runStreamT $ do
414−-- str <- cycle1 (liftIO getLine)
415−-- liftIO $ putStrLn str
493+-- liftIO $ putStrLn $ show (x, y)
502+-- You will notice that this is the monadic equivalent of the '<=>' style
503+-- sum composition. The same caveats apply to this as the '<=>' operation.
418−-- $monadicAlternatives
505+-- $fairlyConcurrentNesting
507+-- When we interpret the monadic composition as 'ParallelT' we get a
508+-- /concurrent/ list-transformer like composition just like 'AsyncT'. The
509+-- difference is that this is fully parallel with all iterations starting
510+-- concurrently instead of the demand driven concurrency of 'AsyncT'.
513+-- main = runParallelT $ do
518+-- ThreadId 40: Delay 1
519+-- ThreadId 39: Delay 2
520+-- ThreadId 38: Delay 3
420−-- Just like we saw multiple non-concurrent and concurrent variants of sum
523+-- You will notice that this is the monadic equivalent of the '<|>' style
421−-- style compositions earlier, monadic composition also has multiple variants
524+-- sum composition. The same caveats apply to this as the '<|>' operation.
422−-- each of which exactly corresponds to one of the sum style composition
525+
423−-- variant.
526+-- $monadExercise
425−-- We will use the following polymorphic code which is agnostic of a specific
528+-- The streamly code is usually written in a way that is agnostic of the
426−-- 'Streaming' type to illustrate various variants of monadic stream
529+-- specific monadic composition type. We use a polymorphic type with a
427−-- composition. This code can be interpreted in many different ways depending
530+-- 'Streaming' type class constraint. When running the stream we can choose the
428−-- on the actual type used.
531+-- specific mode of composition. For example look at the following code.
431-- import Streamly
534-- import Streamly
@@ -443,85 +546,119 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
443-- colors = each ["red", "green", "blue"]
546-- colors = each ["red", "green", "blue"]
444-- shapes = each ["triangle", "square", "circle"]
547-- shapes = each ["triangle", "square", "circle"]
449−-- When we interpret the code as 'StreamT' we get the regular list transformer
550+-- Now we can interpret this in whatever way we want:
450−-- like serial composition. The following is equivalent to three nested loops,
451−-- the first one for sizes, the second one for colors and the third one for
452−-- shapes. For each size we iterate each color and for each color we iterate
456−-- main = runStreaming $ serially $ composed
553+-- main = runStreamT composed
554+-- main = runAsyncT composed
555+-- main = runInterleavedT composed
556+-- main = runParallelT composed
459−
559+-- Equivalently, we can also write it using the type adapter combinators, like
460−-- $interleavedNesting
560+-- this:
463−-- main = runStreaming $ interleaved $ composed
563+-- main = runStreaming $ serially $ composed
564+-- main = runStreaming $ asyncly $ composed
565+-- main = runStreaming $ interleaving $ composed
566+-- main = runStreaming $ parallely $ composed
569+-- As an exercise try to figure out the output of this code for each mode of
466−-- $concurrentNesting
572+-- $functor
574+-- The simplest transformation on a stream is mapping a function on
575+-- all elements of the stream using 'fmap'.
469-- import Streamly
578-- import Streamly
470−-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
471−-- import Control.Concurrent
473−-- printWithTid s = do
580+-- main = (toList $ serially $ fmap show $ each [1..10]) >>= print
474−-- threadDelay 500
581+-- @
475−-- tid <- myThreadId
582+
476−-- putStrLn (show tid ++ ": " ++ s)
583+-- $applicative
478−-- main = runStreaming $ asyncly $ do
585+-- Applicative is a special case of monad and behaves just like the monadic
479−-- liftIO $ hSetBuffering stdout LineBuffering
586+-- composition. The following example runs all iterations serially and takes a
480−-- sz <- sizes
587+-- total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
483−-- liftIO $ putStrLn $ show (sz, cl, sh)
592+-- d n = delay n >> return n
487−-- toStream xs = foldMapWith (<>) return xs
594+-- main = (toList . serially $ (,) \<$> s1 \<*> s2) >>= print
488−-- sizes = toStream [1, 2, 3]
489−-- colors = toStream ["red", "green", "blue"]
490−-- shapes = toStream ["triangle", "square", "circle"]
493−-- $fairlyConcurrentNesting
597+-- ThreadId 36: Delay 1
598+-- ThreadId 36: Delay 3
599+-- ThreadId 36: Delay 4
600+-- ThreadId 36: Delay 2
601+-- ThreadId 36: Delay 3
602+-- ThreadId 36: Delay 4
603+-- [(1,3),(1,4),(2,3),(2,4)]
606+-- Similalrly interleaving runs the iterations in an interleaved order but
607+-- since it is serial it takes a total of 17 seconds:
496−-- import Streamly
610+-- main = (toList . interleaving $ (,) \<$> s1 \<*> s2) >>= print
497−-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
611+-- @
498−-- import Control.Concurrent
612+-- @
613+-- ThreadId 36: Delay 1
614+-- ThreadId 36: Delay 3
615+-- ThreadId 36: Delay 2
616+-- ThreadId 36: Delay 3
617+-- ThreadId 36: Delay 4
618+-- ThreadId 36: Delay 4
619+-- [(1,3),(2,3),(1,4),(2,4)]
500−-- printWithTid s = do
622+-- 'AsyncT' can run the iterations concurrently and therefore takes a total
501−-- threadDelay 500
623+-- of 10 seconds (1 + 2 + 3 + 4):
503−-- putStrLn (show tid ++ ": " ++ s)
505−-- main = runStreaming $ parallely $ do
625+-- @
506−-- liftIO $ hSetBuffering stdout LineBuffering
626+-- main = (toList . asyncly $ (,) \<$> s1 \<*> s2) >>= print
507−-- sz <- sizes
627+-- @
508−-- cl <- colors
628+-- @
509−-- sh <- shapes
629+-- ThreadId 34: Delay 1
510−-- liftIO $ putStrLn $ show (sz, cl, sh)
630+-- ThreadId 36: Delay 2
631+-- ThreadId 35: Delay 3
632+-- ThreadId 36: Delay 3
633+-- ThreadId 35: Delay 4
634+-- ThreadId 36: Delay 4
635+-- [(1,3),(2,3),(1,4),(2,4)]
512−-- where
638+-- Similalrly 'ParallelT' as well can run the iterations concurrently and
639+-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
514−-- toStream xs = foldMapWith (<>) return xs
641+-- @
515−-- sizes = toStream [1, 2, 3]
642+-- main = (toList . parallely $ (,) \<$> s1 \<*> s2) >>= print
516−-- colors = toStream ["red", "green", "blue"]
643+-- @
517−-- shapes = toStream ["triangle", "square", "circle"]
644+-- @
645+-- ThreadId 34: Delay 1
646+-- ThreadId 36: Delay 2
647+-- ThreadId 35: Delay 3
648+-- ThreadId 36: Delay 3
649+-- ThreadId 35: Delay 4
650+-- ThreadId 36: Delay 4
651+-- [(1,3),(2,3),(1,4),(2,4)]
520-- $folding
654-- $folding
657+-- $compositionSummary
658+-- Summary of compositions
523-- $zipping
660-- $zipping
524−-- Applicative composition
661+-- Applicative composition, Parallel applicative
526-- $concurrent
663-- $concurrent