@@ -37,7 +37,10 @@ module Streamly.Tutorial
37 -- * Streams
37 -- * Streams
38 -- $streams
38 -- $streams
40+ -- * Concurrent Streams
41+ -- $concurrentStreams
40− -- * Flavors of Streams
43+ -- * Combining Streams
41 -- $flavors
44 -- $flavors
43 -- * Imports and Supporting Code
46 -- * Imports and Supporting Code
@@ -46,12 +49,21 @@ module Streamly.Tutorial
46 -- * Generating Streams
49 -- * Generating Streams
47 -- $generating
50 -- $generating
52+ -- * Generating Streams Concurrently
53+ -- $generatingConcurrently
49 -- * Eliminating Streams
55 -- * Eliminating Streams
50 -- $eliminating
56 -- $eliminating
58+ -- * Concurrent Pipeline Stages
59+ -- $concurrentApplication
52 -- * Transforming Streams
61 -- * Transforming Streams
53 -- $transformation
62 -- $transformation
64+ -- * Mapping Concurrently
65+ -- $concurrentTransformation
55 -- * Merging Streams
67 -- * Merging Streams
57 -- ** Semigroup Style
69 -- ** Semigroup Style
@@ -63,6 +75,9 @@ module Streamly.Tutorial
63 -- *** Wide Serial Composition ('WSerial')
75 -- *** Wide Serial Composition ('WSerial')
64 -- $interleaved
76 -- $interleaved
78+ -- *** Deep Ahead Composition ('Ahead')
66 -- *** Deep Asynchronous Composition ('Async')
81 -- *** Deep Asynchronous Composition ('Async')
@@ -90,6 +105,9 @@ module Streamly.Tutorial
90 -- *** Wide Serial Nesting ('WSerial')
105 -- *** Wide Serial Nesting ('WSerial')
91 -- $interleavedNesting
106 -- $interleavedNesting
108+ -- *** Deep Ahead Nesting ('Ahead')
93 -- *** Deep Asynchronous Nesting ('Async')
111 -- *** Deep Asynchronous Nesting ('Async')
94 -- $concurrentNesting
112 -- $concurrentNesting
@@ -174,51 +192,89 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
175-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
193-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
176-- pure lists are a special case of monadic streams. If you use 'nil' in place
194-- 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
195+-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' stream.
178−-- only difference is that the elements must be monadic type and to operate on
196+-- The only difference is that the elements must be monadic type and to operate
179−-- the streams we must use the corresponding functions from "Streamly.Prelude"
197+-- on the streams we must use the corresponding functions from
180−-- instead of using the base "Prelude".
198+-- "Streamly.Prelude" instead of using the base "Prelude".
200+-- $concurrentStreams
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.
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.
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.
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+-- +------------+--------------+--------------+--------------+
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.
182-- $flavors
254-- $flavors
184−-- There are a few more types similar to 'Serial' that all represent streams
256+-- Streams can be combined using semigroup or monoid composition to form a
185−-- and differ only in the 'Semigroup', 'Applicative' and 'Monad' compositions
257+-- composite stream. Traversal of a composition of streams could be @deep@ or
186−-- of the streams.
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.
188−-- The composition of two or more streams is distinguished based on three
262+-- Each stream type has a wide traversal variant prefixed by 'W'. The wide
189−-- characterstics, /traversal order/, /execution order/ and
263+-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad
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.
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:
264+-- compositions of the streams.
265+-- The following table summarizes the basic types and the corresponding wide
209−-- +------------+-----------+--------------+--------------+
269+-- +------------+-----------+
210−-- | Type | Traversal | Execution | Consumption |
270+-- | Deep | Wide |
211−-- +============+===========+==============+==============+
271+-- +============+===========+
212−-- | 'Serial' | Deep | Serial | Serial |
272+-- | 'Serial' | 'WSerial' |
213−-- +------------+-----------+--------------+--------------+
273+-- +------------+-----------+
214−-- | 'WSerial' | Wide | Serial | Serial |
274+-- | 'Ahead' | 'WAhead' |
215−-- +------------+-----------+--------------+--------------+
275+-- +------------+-----------+
216−-- | 'Async' | Deep | Asynchronous | Asynchronous |
276+-- | 'Async' | 'WAsync' |
217−-- +------------+-----------+--------------+--------------+
277+-- +------------+-----------+
218−-- | 'WAsync' | Wide | Asynchronous | Asynchronous |
219−-- +------------+-----------+--------------+--------------+
220−-- | 'Parallel' | Parallel | Asynchronous | Asynchronous |
221−-- +------------+-----------+--------------+--------------+
224-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
280-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
@@ -226,13 +282,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
226-- types are not monads they are only applicatives and they do not differ in
282-- types are not monads they are only applicatives and they do not differ in
227-- 'Semigroup' composition.
283-- 'Semigroup' composition.
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.
237-- $programs
286-- $programs
@@ -328,15 +377,31 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
330-- For more ways to construct a stream see the module "Streamly.Prelude".
379-- For more ways to construct a stream see the module "Streamly.Prelude".
381+-- $generatingConcurrently
332−-- Note that monadic generation functions e.g. 'consM', 'unfoldrM',
383+-- Monadic construction and generation functions e.g. 'consM', 'unfoldrM',
333-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' etc work
384-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' etc work
334-- concurrently when used with appropriate stream type combinator. The pure
385-- concurrently when used with appropriate stream type combinator. The pure
335−-- versions do not work concurrently, however you can use the monadic versions
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
336−-- even for pure computations to get the concurrent generation capability where
388+-- get the concurrent generation capability where required.
390+-- The following code finishes in 3 seconds (6 seconds when serial):
393+-- > let p n = threadDelay (n * 1000000) >> return n
394+-- > S.'toList' $ 'parallely' $ p 3 |: p 2 |: p 1 |: S.'nil'
396+-- > S.'toList' $ 'aheadly' $ p 3 |: p 2 |: p 1 |: S.'nil'
399+-- The following finishes in 10 seconds (100 seconds when serial):
402+-- > 'runStream' $ 'asyncly' $ S.'replicateM' 10 $ p 10
339−-- TBD add more details and examples about concurrent generation.
341-- $eliminating
406-- $eliminating
@@ -383,15 +448,35 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
383-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
448-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
384-- & S.'mapM' putStrLn
449-- & S.'mapM' putStrLn
452+-- $concurrentTransformation
387−-- Note that monadic transformation functions 'mapM' and 'sequence' work
454+-- Monadic transformation functions 'mapM' and 'sequence' work concurrently
388−-- concurrently when used with appropriate stream type combinator. The pure
455+-- when used with appropriate stream type combinators. The pure versions do not
389−-- versions do not work concurrently, however you can use the monadic versions
456+-- work concurrently, however you can use the monadic versions even for pure
390−-- even for pure computations to get the concurrent transformation capability
457+-- computations to get the concurrent transformation capability where required.
393−-- TBD add more details and examples about concurrent transformation.
459+-- This would print a value every second (2 seconds when serial):
394−-- TBD add details about concurrent function application.
460+--
462+-- > let p n = threadDelay (n * 1000000) >> return n
463+-- > runStream $ aheadly $ S.'mapM' (\\x -> p 1 >> print x) (serially $ repeatM (p 1))
467+-- $concurrentApplication
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:
473+-- Because both the stages run concurrently, we would see a delay of only 1
474+-- second instead of 2 seconds in the following:
477+-- > let p n = threadDelay (n * 1000000) >> return n
478+-- > runStream $ S.'repeatM' (p 1) '|&' S.'mapM' (\\x -> p 1 >> print x)
396-- $semigroup
481-- $semigroup
@@ -531,6 +616,39 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
531-- Note that this composition cannot be used to fold infinite number of streams
616-- Note that this composition cannot be used to fold infinite number of streams
532-- since it requires preserving the state until a stream is finished.
617-- since it requires preserving the state until a stream is finished.
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.
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:
637+-- xs \<- 'toList' . 'aheadly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
639+-- where p n = threadDelay 1000000 >> return n
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.
649+-- You can also use the polymorphic combinator 'ahead' in place of '<>' to
650+-- compose any type of streams in this manner.
536-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
654-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
@@ -547,32 +665,33 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
547-- stream(s) is performed if the first stream blocks or if it cannot produce
665-- stream(s) is performed if the first stream blocks or if it cannot produce
548-- output at the rate that is enough to meet the consumer demand. Multiple
666-- output at the rate that is enough to meet the consumer demand. Multiple
549-- streams can be executed concurrently to meet the demand.
667-- streams can be executed concurrently to meet the demand.
550−-- In the following example the first stream does not block,
668+-- In the example below each element in the stream introduces a constant delay
551−-- therefore the first stream is completely exhausted before the second.
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:
554−-- main = 'runStream' . 'asyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
674+-- xs \<- 'runStream' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
676+-- where p n = threadDelay 1000000 >> return n
563−-- If the first stream blocks, we can yield from the second. In the example
682+-- The constituent streams are also composed in 'Async' manner and the
564−-- below each yield in the stream has a constant delay of 1 second therefore 1
683+-- composition of streams too. We can compose the constituent streams to run
565−-- and 3 would be yielded first and then 2 and 4 would be yielded.
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:
568−-- main = 'runStream' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
689+-- xs \<- 'runStream' . 'asyncly' $ (serially $ p 1 |: p 2 |: nil) <> (serially $ p 3 |: p 4 |: nil)
569−-- where p n = threadDelay 1000000 >> print n
691+-- where p n = threadDelay 1000000 >> return n
578-- In the following example we can see that new threads are started when a
697-- 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))
665-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
784-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
666-- a concurrent manner using /breadth first traversal/. We use the 'wAsyncly'
785-- a concurrent manner using /breadth first traversal/. We use the 'wAsyncly'
667-- type combinator to effect 'WAsync' style of composition. We can also use the
786-- 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.
670-- When streams with multiple elements are combined in this manner, we traverse
789-- When streams with multiple elements are combined in this manner, we traverse
671-- all the streams concurrently in a breadth first manner i.e. one action from
790-- 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
673-- back to the first stream again and so on. Even though we execute the actions
792-- 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
677-- In the following example we can see that outputs are produced in the breadth
796-- 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.
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)
@@ -896,6 +1015,44 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
896-- Notice that this is analogous to merging streams of type 'Serial' or merging
1015-- Notice that this is analogous to merging streams of type 'Serial' or merging
897-- streams using 'serial'.
1016-- streams using 'serial'.
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.
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'.
1029+-- import "Streamly"
1030+-- import "Streamly.Prelude" as S
1032+-- comp = 'toList' . 'aheadly' $ do
1033+-- x <- S.'fromFoldable' [3,2,1]
1034+-- delay x >> return x
1036+-- main = comp >> print
1039+-- ThreadId 40: Delay 1
1040+-- ThreadId 39: Delay 2
1041+-- ThreadId 38: Delay 3
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.
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.
899-- $concurrentNesting
1056-- $concurrentNesting
901-- The 'Monad' composition of 'Async' type can perform the iterations of a
1058-- The 'Monad' composition of 'Async' type can perform the iterations of a