@@ -37,7 +37,10 @@ module Streamly.Tutorial
3737 -- * Streams
3838 -- $streams
3939
40− -- * Flavors of Streams
40+ -- * Concurrent Streams
41+ -- $concurrentStreams
42+
43+ -- * Combining Streams
4144 -- $flavors
4245
4346 -- * Imports and Supporting Code
@@ -46,12 +49,21 @@ module Streamly.Tutorial
4649 -- * Generating Streams
4750 -- $generating
4851
52+ -- * Generating Streams Concurrently
53+ -- $generatingConcurrently
54+
4955 -- * Eliminating Streams
5056 -- $eliminating
5157
58+ -- * Concurrent Pipeline Stages
59+ -- $concurrentApplication
60+
5261 -- * Transforming Streams
5362 -- $transformation
5463
64+ -- * Mapping Concurrently
65+ -- $concurrentTransformation
66+
5567 -- * Merging Streams
5668
5769 -- ** Semigroup Style
@@ -63,6 +75,9 @@ module Streamly.Tutorial
6375 -- *** Wide Serial Composition ('WSerial')
6476 -- $interleaved
6577
78+ -- *** Deep Ahead Composition ('Ahead')
79+ -- $ahead
80+
6681 -- *** Deep Asynchronous Composition ('Async')
6782 -- $async
6883
@@ -90,6 +105,9 @@ module Streamly.Tutorial
90105 -- *** Wide Serial Nesting ('WSerial')
91106 -- $interleavedNesting
92107
108+ -- *** Deep Ahead Nesting ('Ahead')
109+ -- $aheadNesting
110+
93111 -- *** Deep Asynchronous Nesting ('Async')
94112 -- $concurrentNesting
95113
@@ -174,51 +192,89 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
174192--
175193-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
176194-- pure lists are a special case of monadic streams. If you use 'nil' in place
177−-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' The
178−-- only difference is that the elements must be monadic type and to operate on
179−-- the streams we must use the corresponding functions from "Streamly.Prelude"
180−-- instead of using the base "Prelude".
195+-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' stream.
196+-- The only difference is that the elements must be monadic type and to operate
197+-- on the streams we must use the corresponding functions from
198+-- "Streamly.Prelude" instead of using the base "Prelude".
199+
200+-- $concurrentStreams
201+--
202+-- Streams can be generated concurrently, even infnite streams can be generated
203+-- concurrently using controlled concurrency, streams can be merged
204+-- concurrently, multiple stages in a streaming pipeline can run concurrently,
205+-- streams can be mapped concurrently, they can be zipped concurrently, and in
206+-- a monadic composition they combine like a list transformer providing
207+-- concurrent non-determinism.
208+--
209+-- There are three basic concurrent stream styles, 'Ahead', 'Async', and
210+-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that
211+-- they can speculatively execute multiple stream actions concurrently in
212+-- advance. 'Ahead' would return exactly the same stream as 'Serial' except
213+-- that it may execute the actions concurrently. The 'Async' style streams,
214+-- like 'Ahead', speculatively execute multiple stream actions in advance but
215+-- return the results in their finishing order rather than in the stream
216+-- traversal order. 'Parallel' is like 'Async' except that it provides
217+-- unbounded parallelism instead of controlled parallelism.
218+--
219+-- For easy reference, we can classify the stream types based on /execution order/,
220+-- /consumption order/, and /bounded or unbounded/ concurrency.
221+-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
222+-- execution we execute the next action in the stream only after the previous
223+-- one has finished executing. In asynchronous execution multiple actions in
224+-- the stream can be executed asynchronously i.e. the next action can start
225+-- executing even before the first one has finished. Consumption order
226+-- determines the order in which the outputs generated by the composition are
227+-- consumed. Consumption could be serial or asynchronous. In serial
228+-- consumption, the outputs are consumed in the traversal order, in
229+-- asynchronous consumption the outputs are consumed as they arrive i.e. first
230+-- come first serve order.
231+--
232+-- @
233+-- +------------+--------------+--------------+--------------+
234+-- | Type | Execution | Consumption | Concurrency |
235+-- +============+==============+==============+==============+
236+-- | 'Serial' | Serial | Serial | None |
237+-- +------------+--------------+--------------+--------------+
238+-- | 'Ahead' | Asynchronous | Serial | bounded |
239+-- +------------+--------------+--------------+--------------+
240+-- | 'Async' | Asynchronous | Asynchronous | bounded |
241+-- +------------+--------------+--------------+--------------+
242+-- | 'Parallel' | Asynchronous | Asynchronous | unbounded |
243+-- +------------+--------------+--------------+--------------+
244+-- @
245+--
246+-- All these types can be freely inter-converted using type conversion
247+-- combinators or type annotations without any cost, to acheive the desired
248+-- composition style. To force a particular type of composition we coerce the
249+-- stream type using the corresponding type adapting combinator from
250+-- 'serially', 'aheadly', 'asyncly', or 'parallely'. The default stream type
251+-- is inferred as 'Serial' unless you change it by using one of the combinators
252+-- or using a type annotation.
181253
182254-- $flavors
183255--
184−-- There are a few more types similar to 'Serial' that all represent streams
185−-- and differ only in the 'Semigroup', 'Applicative' and 'Monad' compositions
186−-- of the streams.
256+-- Streams can be combined using semigroup or monoid composition to form a
257+-- composite stream. Traversal of a composition of streams could be @deep@ or
258+-- @wide@. Deep goes depth first i.e. each stream is traversed fully before
259+-- we traverse the next stream. Wide goes breadth first i.e. one element from
260+-- each stream is traversed before coming back to the first stream again.
187261--
188−-- The composition of two or more streams is distinguished based on three
189−-- characterstics, /traversal order/, /execution order/ and
190−-- /consumption order/. Traversal of a composition of streams could be @deep@
191−-- or @wide@. Deep goes depth first i.e. each stream is traversed fully
192−-- before we traverse the next stream. Wide goes breadth first i.e. one element
193−-- from each stream is traversed before coming back to the first stream again.
194−-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
195−-- execution we execute an action in the next stream only after the first has
196−-- finished executing. In asynchronous execution actions in both streams can be
197−-- executed asynchronously i.e. the next action can start executing even before
198−-- the first one has finished. The third parameter is consumption order that is
199−-- in what order the output generated by the composition is consumed.
200−-- Consumption could be serial or asynchronous. In serial consumption, the
201−-- outputs are consumed in the traversal order, in asynchronous consumption the
202−-- outputs are consumed as they arrive i.e. first come first serve order.
203−--
204−-- The following table summarizes different styles of streams based on how they
205−-- compose. All these types are monads and they differ in 'Semigroup',
206−-- 'Applicative' and 'Monad' compositions:
207−--
208−-- @
209−-- +------------+-----------+--------------+--------------+
210−-- | Type | Traversal | Execution | Consumption |
211−-- +============+===========+==============+==============+
212−-- | 'Serial' | Deep | Serial | Serial |
213−-- +------------+-----------+--------------+--------------+
214−-- | 'WSerial' | Wide | Serial | Serial |
215−-- +------------+-----------+--------------+--------------+
216−-- | 'Async' | Deep | Asynchronous | Asynchronous |
217−-- +------------+-----------+--------------+--------------+
218−-- | 'WAsync' | Wide | Asynchronous | Asynchronous |
219−-- +------------+-----------+--------------+--------------+
220−-- | 'Parallel' | Parallel | Asynchronous | Asynchronous |
221−-- +------------+-----------+--------------+--------------+
262+-- Each stream type has a wide traversal variant prefixed by 'W'. The wide
263+-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad
264+-- compositions of the streams.
265+-- The following table summarizes the basic types and the corresponding wide
266+-- variants:
267+--
268+-- @
269+-- +------------+-----------+
270+-- | Deep | Wide |
271+-- +============+===========+
272+-- | 'Serial' | 'WSerial' |
273+-- +------------+-----------+
274+-- | 'Ahead' | 'WAhead' |
275+-- +------------+-----------+
276+-- | 'Async' | 'WAsync' |
277+-- +------------+-----------+
222278-- @
223279--
224280-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
@@ -226,13 +282,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
226282-- types are not monads they are only applicatives and they do not differ in
227283-- 'Semigroup' composition.
228284--
229−-- All these types can be freely inter-converted using type conversion
230−-- combinators or type annotations without any cost, to acheive the desired
231−-- composition style. To force a particular type of composition we coerce the
232−-- stream type using the corresponding type adapting combinator from
233−-- 'serially', 'wSerially', 'asyncly', 'wAsyncly', 'parallely', 'zipSerially'
234−-- or 'zipAsyncly'. The default stream type is inferred as 'Serial' unless you
235−-- change it by using one of the combinators or using a type annotation.
236285
237286-- $programs
238287--
@@ -328,15 +377,31 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
328377-- @
329378--
330379-- For more ways to construct a stream see the module "Streamly.Prelude".
380+
381+-- $generatingConcurrently
331382--
332−-- Note that monadic generation functions e.g. 'consM', 'unfoldrM',
383+-- Monadic construction and generation functions e.g. 'consM', 'unfoldrM',
333384-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' etc work
334385-- concurrently when used with appropriate stream type combinator. The pure
335−-- versions do not work concurrently, however you can use the monadic versions
336−-- even for pure computations to get the concurrent generation capability where
337−-- required.
386+-- versions of these APIs are not concurrent, however you can use the monadic
387+-- versions even for pure computations by wrapping the pure value in a monad to
388+-- get the concurrent generation capability where required.
389+--
390+-- The following code finishes in 3 seconds (6 seconds when serial):
391+--
392+-- @
393+-- > let p n = threadDelay (n * 1000000) >> return n
394+-- > S.'toList' $ 'parallely' $ p 3 |: p 2 |: p 1 |: S.'nil'
395+-- [1,2,3]
396+-- > S.'toList' $ 'aheadly' $ p 3 |: p 2 |: p 1 |: S.'nil'
397+-- [3,2,1]
398+-- @
399+-- The following finishes in 10 seconds (100 seconds when serial):
400+--
401+-- @
402+-- > 'runStream' $ 'asyncly' $ S.'replicateM' 10 $ p 10
403+-- @
338404--
339−-- TBD add more details and examples about concurrent generation.
340405
341406-- $eliminating
342407--
@@ -383,15 +448,35 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
383448-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
384449-- & S.'mapM' putStrLn
385450-- @
451+
452+-- $concurrentTransformation
386453--
387−-- Note that monadic transformation functions 'mapM' and 'sequence' work
388−-- concurrently when used with appropriate stream type combinator. The pure
389−-- versions do not work concurrently, however you can use the monadic versions
390−-- even for pure computations to get the concurrent transformation capability
391−-- where required.
454+-- Monadic transformation functions 'mapM' and 'sequence' work concurrently
455+-- when used with appropriate stream type combinators. The pure versions do not
456+-- work concurrently, however you can use the monadic versions even for pure
457+-- computations to get the concurrent transformation capability where required.
392458--
393−-- TBD add more details and examples about concurrent transformation.
394−-- TBD add details about concurrent function application.
459+-- This would print a value every second (2 seconds when serial):
460+--
461+-- @
462+-- > let p n = threadDelay (n * 1000000) >> return n
463+-- > runStream $ aheadly $ S.'mapM' (\\x -> p 1 >> print x) (serially $ repeatM (p 1))
464+-- @
465+--
466+
467+-- $concurrentApplication
468+--
469+-- The concurrent function application operators '|$' and '|&' apply a stream
470+-- argument to a stream function concurrently to compose a concurrent pipeline
471+-- of stream processing functions:
472+--
473+-- Because both the stages run concurrently, we would see a delay of only 1
474+-- second instead of 2 seconds in the following:
475+--
476+-- @
477+-- > let p n = threadDelay (n * 1000000) >> return n
478+-- > runStream $ S.'repeatM' (p 1) '|&' S.'mapM' (\\x -> p 1 >> print x)
479+-- @
395480
396481-- $semigroup
397482--
@@ -531,6 +616,39 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
531616-- Note that this composition cannot be used to fold infinite number of streams
532617-- since it requires preserving the state until a stream is finished.
533618
619+-- $ahead
620+--
621+-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a
622+-- /serial depth first/ manner with concurrent lookahead. We use the 'aheadly'
623+-- type combinator to effect 'Ahead' style of composition. We can also use an
624+-- explicit 'Ahead' type annotation for the stream to acheive the same effect.
625+--
626+-- When two streams are combined in this manner, the streams are traversed in
627+-- depth first manner just like 'Serial', however it can execute the next
628+-- stream concurrently and keep the results ready when its turn arrives.
629+-- Concurrent execution of the next stream(s) is performed if the first stream
630+-- blocks or if it cannot produce output at the rate that is enough to meet the
631+-- consumer demand. Multiple streams can be executed concurrently to meet the
632+-- demand. The following example would print the result in a second even
633+-- though each action in each stream takes one second:
634+--
635+-- @
636+-- main = do
637+-- xs \<- 'toList' . 'aheadly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
638+-- print xs
639+-- where p n = threadDelay 1000000 >> return n
640+-- @
641+-- @
642+-- [1,2,3,4]
643+-- @
644+--
645+-- Each stream is constructed 'aheadly' and then both the streams are merged
646+-- 'aheadly', therefore, all the actions can run concurrently but the result is
647+-- presented in serial order.
648+--
649+-- You can also use the polymorphic combinator 'ahead' in place of '<>' to
650+-- compose any type of streams in this manner.
651+
534652-- $async
535653--
536654-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
@@ -547,32 +665,33 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
547665-- stream(s) is performed if the first stream blocks or if it cannot produce
548666-- output at the rate that is enough to meet the consumer demand. Multiple
549667-- streams can be executed concurrently to meet the demand.
550−-- In the following example the first stream does not block,
551−-- therefore the first stream is completely exhausted before the second.
668+-- In the example below each element in the stream introduces a constant delay
669+-- of 1 second, however, it takes just one second to produce all the results.
670+-- The results are not guaranteed to be in any particular order:
552671--
553672-- @
554−-- main = 'runStream' . 'asyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
673+-- main = do
674+-- xs \<- 'runStream' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
675+-- print xs
676+-- where p n = threadDelay 1000000 >> return n
555677-- @
556678-- @
557−-- 1
558−-- 2
559−-- 3
560−-- 4
679+-- [4,2,1,3]
561680-- @
562681--
563−-- If the first stream blocks, we can yield from the second. In the example
564−-- below each yield in the stream has a constant delay of 1 second therefore 1
565−-- and 3 would be yielded first and then 2 and 4 would be yielded.
682+-- The constituent streams are also composed in 'Async' manner and the
683+-- composition of streams too. We can compose the constituent streams to run
684+-- serially, in that case it would take 2 seconds to produce all the results.
685+-- The elements in the serial streams would be in serial order in the results:
566686--
567687-- @
568−-- main = 'runStream' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
569−-- where p n = threadDelay 1000000 >> print n
688+-- main = do
689+-- xs \<- 'runStream' . 'asyncly' $ (serially $ p 1 |: p 2 |: nil) <> (serially $ p 3 |: p 4 |: nil)
690+-- print xs
691+-- where p n = threadDelay 1000000 >> return n
570692-- @
571693-- @
572−-- 1
573−-- 3
574−-- 2
575−-- 4
694+-- [3,1,2,4]
576695-- @
577696--
578697-- In the following example we can see that new threads are started when a
@@ -665,20 +784,20 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
665784-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
666785-- a concurrent manner using /breadth first traversal/. We use the 'wAsyncly'
667786-- type combinator to effect 'WAsync' style of composition. We can also use the
668−-- 'WAsync' type annotation for the stream to acheive the same effect.
787+-- 'WAsync' type annotation for the stream to achieve the same effect.
669788--
670789-- When streams with multiple elements are combined in this manner, we traverse
671790-- all the streams concurrently in a breadth first manner i.e. one action from
672−-- each stream is peformed and yielded to the resulting stream before we come
791+-- each stream is performed and yielded to the resulting stream before we come
673792-- back to the first stream again and so on. Even though we execute the actions
674−-- in a breadth first order the outputs may be consumed in a different order
675−-- because they are consumed on a first come first serve basis.
793+-- in a breadth first order the outputs are consumed on a first come first
794+-- serve basis.
676795--
677796-- In the following example we can see that outputs are produced in the breadth
678−-- first travresal order but this is not guaranteed.
797+-- first traversal order but this is not guaranteed.
679798--
680799-- @
681−-- main = 'runStream' . 'wAsyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
800+-- main = 'runStream' . 'wAsyncly' $ (serially $ print 1 |: print 2 |: nil) <> (serially $ print 3 |: print 4 |: nil)
682801-- @
683802-- @
684803-- 1
@@ -896,6 +1015,44 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
8961015-- Notice that this is analogous to merging streams of type 'Serial' or merging
8971016-- streams using 'serial'.
8981017
1018+-- $aheadNesting
1019+--
1020+-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except
1021+-- that it can speculatively perform a bounded number of next iterations of a
1022+-- loop concurrently.
1023+--
1024+-- The 'aheadly' type combinator can be used to switch to this style of
1025+-- composition. Alternatively, a type annotation can be used to specify the
1026+-- type of the stream as 'Ahead'.
1027+--
1028+-- @
1029+-- import "Streamly"
1030+-- import "Streamly.Prelude" as S
1031+--
1032+-- comp = 'toList' . 'aheadly' $ do
1033+-- x <- S.'fromFoldable' [3,2,1]
1034+-- delay x >> return x
1035+--
1036+-- main = comp >> print
1037+-- @
1038+-- @
1039+-- ThreadId 40: Delay 1
1040+-- ThreadId 39: Delay 2
1041+-- ThreadId 38: Delay 3
1042+-- [3,2,1]
1043+-- @
1044+--
1045+-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can
1046+-- see all the three iterations are concurrent and run in different threads,
1047+-- however, the results are returned in the serial order.
1048+--
1049+-- Concurrency is demand driven, when multiple streams are composed using this
1050+-- style, the iterations are executed in a depth first manner just like
1051+-- 'Serial' i.e. nested iterations are executed before we proceed to the next
1052+-- outer iteration. The only difference is that we may execute multiple future
1053+-- iterations concurrently and keep the results ready.
1054+--
1055+
8991056-- $concurrentNesting
9001057--
9011058-- The 'Monad' composition of 'Async' type can perform the iterations of a