git.iko.soy/streamly
on master help sign in
Break Tutorial module into multiple small modules
54815c5Harendra Kumar
docs/ConcurrentStreams.hsnew file+780
compare with master
unifiedsplit
@@ -0,0 +1,780 @@
1+{-# OPTIONS_GHC -fno-warn-unused-imports #-}
2+-- |
3+-- Module : ConcurrentStreams
4+-- Copyright : (c) 2017 Composewell Technologies
5+--
6+-- License : BSD3
7+-- Maintainer : streamly@composewell.com
8+--
9+-- In this tutorial we will show how streamly can be used for idiomatic and
10+-- declarative concurrent programming. Before you go through this tutorial we
11+-- recommend that you take a look at the Streamly serial streams tutorial.
12+
13+module ConcurrentStreams
14+ (
15+ -- * Concurrent Streams
16+ -- $concurrentStreams
17+
18+ -- * Combining Streams
19+ -- $flavors
20+
21+ -- * Imports and Supporting Code
22+ -- $imports
23+
24+ -- * Generating Streams Concurrently
25+ -- $generatingConcurrently
26+
27+ -- * Concurrent Pipeline Stages
28+ -- $concurrentApplication
29+
30+ -- * Mapping Concurrently
31+ -- $concurrentTransformation
32+
33+ -- * Merging Streams
34+
35+ -- ** Semigroup Style
36+
37+ -- *** Deep Speculative Composition ('Ahead')
38+ -- $ahead
39+
40+ -- *** Deep Asynchronous Composition ('Async')
41+ -- $async
42+
43+ -- *** Wide Asynchronous Composition ('WAsync')
44+ -- $wasync
45+
46+ -- *** Parallel Asynchronous Composition ('Parallel')
47+ -- $parallel
48+
49+ -- XXX we should deprecate and remove the mkAsync API
50+ -- Custom composition
51+ -- custom
52+
53+ -- ** Monoid Style
54+ -- $monoid
55+
56+ -- * Nesting Streams
57+
58+ -- ** Monad
59+ -- *** Deep Speculative Nesting ('Ahead')
60+ -- $aheadNesting
61+
62+ -- *** Deep Asynchronous Nesting ('Async')
63+ -- $concurrentNesting
64+
65+ -- *** Wide Asynchronous Nesting ('WAsync')
66+ -- $wasyncNesting
67+
68+ -- *** Parallel Asynchronous Nesting ('Parallel')
69+ -- $parallelNesting
70+
71+ -- ** Applicative
72+ -- $applicative
73+
74+ -- * Zipping Streams
75+
76+ -- ** Parallel Zipping
77+ -- $parallelzip
78+
79+ -- * Concurrent Programming
80+ -- $concurrent
81+
82+ -- * Writing Concurrent Programs
83+ -- $programs
84+
85+ -- * Where to go next?
86+ -- $furtherReading
87+ )
88+where
89+
90+import Streamly.Prelude
91+import Data.Semigroup
92+import Control.Applicative
93+import Control.Monad
94+import Control.Monad.IO.Class (MonadIO(..))
95+import Control.Monad.Trans.Class (MonadTrans (lift))
96+
97+-- CAUTION: please keep setup and imports sections in sync
98+--
99+-- $setup
100+-- >>> :m
101+-- >>> import Data.Function ((&))
102+-- >>> import Streamly.Prelude ((|:), (|&))
103+-- >>> import qualified Streamly.Prelude as Stream
104+-- >>> import qualified Streamly.Data.Fold as Fold
105+--
106+-- >>> import Control.Concurrent (threadDelay, myThreadId)
107+-- >>> :{
108+-- delay n = Stream.fromEffect $ do
109+-- threadDelay (n * 1000000)
110+-- tid <- myThreadId
111+-- putStrLn (show tid ++ ": Delay " ++ show n)
112+-- :}
113+--
114+-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
115+-- >>> hSetBuffering stdout LineBuffering
116+--
117+
118+-- $imports
119+--
120+-- In most of example snippets we do not repeat the imports. Where imports are
121+-- not explicitly specified use the imports shown below.
122+--
123+-- >>> :m
124+-- >>> import Data.Function ((&))
125+-- >>> import Streamly.Prelude ((|:), (|&))
126+-- >>> import qualified Streamly.Prelude as Stream
127+-- >>> import qualified Streamly.Data.Fold as Fold
128+--
129+-- To illustrate concurrent vs serial composition aspects, we will use the
130+-- following @delay@ function to introduce a sleep or delay specified in
131+-- seconds. After the delay it prints the number of seconds it slept.
132+--
133+-- >>> import Control.Concurrent (threadDelay, myThreadId)
134+-- >>> :{
135+-- delay n = Stream.fromEffect $ do
136+-- threadDelay (n * 1000000)
137+-- tid <- myThreadId
138+-- putStrLn (show tid ++ ": Delay " ++ show n)
139+-- :}
140+--
141+-- For concurrent examples, use line buffering, otherwise output from different
142+-- threads may get mixed:
143+--
144+-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
145+-- >>> hSetBuffering stdout LineBuffering
146+--
147+
148+-- $concurrentStreams
149+--
150+-- Many stream operations can be done concurrently:
151+--
152+-- * Streams can be generated concurrently.
153+--
154+-- * Streams can be merged concurrently.
155+--
156+-- * Multiple stages in a streaming pipeline can run concurrently.
157+--
158+-- * Streams can be mapped and zipped concurrently.
159+--
160+-- * In monadic composition they combine like a list transformer,
161+-- providing concurrent non-determinism.
162+--
163+-- There are three basic concurrent stream styles, 'Ahead', 'Async', and
164+-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that
165+-- they can speculatively execute multiple stream actions concurrently in
166+-- advance. 'Ahead' would return exactly the same stream as 'Serial' except
167+-- that it may execute the actions concurrently. The 'Async' style streams,
168+-- like 'Ahead', speculatively execute multiple stream actions in advance but
169+-- return the results in their finishing order rather than in the stream
170+-- traversal order. 'Parallel' is like 'Async' except that it provides
171+-- unbounded parallelism instead of controlled parallelism.
172+--
173+-- For easy reference, we can classify the stream types based on /execution order/,
174+-- /consumption order/, and /bounded or unbounded/ concurrency.
175+-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
176+-- execution we execute the next action in the stream only after the previous
177+-- one has finished executing. In asynchronous execution multiple actions in
178+-- the stream can be executed asynchronously i.e. the next action can start
179+-- executing even before the first one has finished. Consumption order
180+-- determines the order in which the outputs generated by the composition are
181+-- consumed. Consumption could be serial or asynchronous. In serial
182+-- consumption, the outputs are consumed in the traversal order, in
183+-- asynchronous consumption the outputs are consumed as they arrive i.e. first
184+-- come first serve order.
185+--
186+-- +------------+--------------+--------------+--------------+
187+-- | Type | Execution | Consumption | Concurrency |
188+-- +============+==============+==============+==============+
189+-- | 'Serial' | Serial | Serial | None |
190+-- +------------+--------------+--------------+--------------+
191+-- | 'Ahead' | Asynchronous | Serial | bounded |
192+-- +------------+--------------+--------------+--------------+
193+-- | 'Async' | Asynchronous | Asynchronous | bounded |
194+-- +------------+--------------+--------------+--------------+
195+-- | 'Parallel' | Asynchronous | Asynchronous | unbounded |
196+-- +------------+--------------+--------------+--------------+
197+--
198+-- All these types can be freely inter-converted using type conversion
199+-- combinators or type annotations, without any cost, to achieve the desired
200+-- composition style. To force a particular type of composition, we coerce the
201+-- stream type using the corresponding type adapting combinator from
202+-- 'fromSerial', 'fromAhead', 'fromAsync', or 'fromParallel'. The default stream type
203+-- is inferred as 'Serial' unless you change it by using one of the combinators
204+-- or by using a type annotation.
205+
206+-- $flavors
207+--
208+-- Streams can be combined using '<>' or 'mappend' to form a
209+-- composite. Composite streams can be interpreted in a depth first or
210+-- breadth first manner using an appropriate type conversion before
211+-- consumption. Deep (e.g. 'Serial') stream type variants traverse a
212+-- composite stream in a depth first manner, such that each stream is
213+-- traversed fully before traversing the next stream. Wide
214+-- (e.g. 'WSerial') stream types traverse it in a breadth first
215+-- manner, such that one element from each stream is traversed before
216+-- coming back to the first stream again.
217+--
218+-- Each stream type has a wide traversal variant prefixed by 'W'. The wide
219+-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad
220+-- compositions of the streams.
221+-- The following table summarizes the basic types and the corresponding wide
222+-- variants:
223+--
224+-- @
225+-- +------------+-----------+
226+-- | Deep | Wide |
227+-- +============+===========+
228+-- | 'Serial' | 'WSerial' |
229+-- +------------+-----------+
230+-- | 'Ahead' | 'WAhead' |
231+-- +------------+-----------+
232+-- | 'Async' | 'WAsync' |
233+-- +------------+-----------+
234+-- @
235+--
236+-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
237+-- zip streams serially or concurrently using 'Applicative' operation. These
238+-- types are not monads they are only applicatives and they do not differ in
239+-- 'Semigroup' composition.
240+--
241+
242+-- $programs
243+--
244+-- When writing concurrent programs it is advised to not use the concurrent
245+-- style stream combinators blindly at the top level. That might create too
246+-- much concurrency where it is not even required, and can even degrade
247+-- performance in some cases. In some cases it can also lead to surprising
248+-- behavior because of some code that is supposed to be serial becoming
249+-- concurrent. Please be aware that all concurrency capable APIs that you may
250+-- have used under the scope of a concurrent stream combinator will become
251+-- concurrent. For example if you have a 'repeatM' somewhere in your program
252+-- and you use 'fromParallel' on top, the 'repeatM' becomes fully parallel,
253+-- resulting into an infinite parallel execution . Instead, use the
254+-- /Keep It Serial and Stupid/ principle, start with the default serial
255+-- composition and enable concurrent combinators only when and where necessary.
256+-- When you use a concurrent combinator you can use an explicit 'fromSerial'
257+-- combinator to suppress any unnecessary concurrency under the scope of that
258+-- combinator.
259+
260+-- $generatingConcurrently
261+--
262+-- Monadic construction and generation functions like 'consM', 'unfoldrM',
263+-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' work concurrently
264+-- when used with appropriate stream type combinator. The pure versions of
265+-- these APIs are not concurrent, however you can use the monadic versions even
266+-- for pure computations by wrapping the pure value in a monad to get the
267+-- concurrent generation capability where required.
268+--
269+-- The following code finishes in 3 seconds (6 seconds when serial):
270+--
271+-- >>> let p n = threadDelay (n * 1000000) >> return n
272+-- >>> Stream.toList $ Stream.fromParallel $ p 3 |: p 2 |: p 1 |: Stream.nil
273+-- [1,2,3]
274+--
275+-- >>> Stream.toList $ Stream.fromAhead $ p 3 |: p 2 |: p 1 |: Stream.nil
276+-- [3,2,1]
277+--
278+-- The following finishes in 10 seconds (100 seconds when serial):
279+--
280+-- >>> Stream.drain $ Stream.fromAsync $ Stream.replicateM 10 $ p 10
281+--
282+
283+-- $concurrentTransformation
284+--
285+-- Monadic transformation functions 'mapM' and 'sequence' work concurrently
286+-- when used with appropriate stream type combinators. The pure versions do not
287+-- work concurrently, however you can use the monadic versions even for pure
288+-- computations to get the concurrent transformation capability where required.
289+--
290+-- This would print a value every second (2 seconds when serial):
291+--
292+-- >>> let p n = threadDelay (n * 1000000) >> return n
293+-- >>> :{
294+-- parMap =
295+-- Stream.repeatM (p 1)
296+-- & Stream.fromSerial -- repeatM is serial
297+-- & Stream.mapM (\x -> p 1 >> print x)
298+-- & Stream.fromAhead -- mapM is cocnurrent using Ahead style
299+-- & Stream.drain
300+-- :}
301+--
302+
303+-- $concurrentApplication
304+--
305+-- The concurrent function application operators '|$' and '|&' apply a stream
306+-- argument to a stream function concurrently to compose a concurrent pipeline
307+-- of stream processing functions:
308+--
309+-- Because both the stages run concurrently, we would see a delay of only 1
310+-- second instead of 2 seconds in the following:
311+--
312+-- >>> let p n = threadDelay (n * 1000000) >> return n
313+-- >>> :{
314+-- parApp =
315+-- Stream.repeatM (p 1)
316+-- |& Stream.mapM (\x -> p 1 >> print x)
317+-- & Stream.drain
318+-- :}
319+
320+-- $ahead
321+--
322+-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a
323+-- /serial depth first/ manner with concurrent lookahead. We use the 'fromAhead'
324+-- type combinator to effect 'Ahead' style of composition. We can also use an
325+-- explicit 'Ahead' type annotation for the stream to achieve the same effect.
326+--
327+-- When two streams are combined in this manner, the streams are traversed in
328+-- depth first manner just like 'Serial', however it can execute the next
329+-- stream concurrently and keep the results ready when its turn arrives.
330+-- Concurrent execution of the next stream(s) is performed if the first stream
331+-- blocks or if it cannot produce output at the rate that is enough to meet the
332+-- consumer demand. Multiple streams can be executed concurrently to meet the
333+-- demand. The following example would print the result in a second even
334+-- though each action in each stream takes one second:
335+--
336+-- >>> p n = threadDelay 1000000 >> return n
337+-- >>> stream1 = p 1 |: p 2 |: Stream.nil
338+-- >>> stream2 = p 3 |: p 4 |: Stream.nil
339+-- >>> Stream.toList $ Stream.fromAhead $ stream1 <> stream2
340+-- [1,2,3,4]
341+--
342+-- Each stream is constructed 'fromAhead' and then both the streams are merged
343+-- 'fromAhead', therefore, all the actions can run concurrently but the result is
344+-- presented in serial order.
345+--
346+-- You can also use the polymorphic combinator 'ahead' in place of '<>' to
347+-- compose any type of streams in this manner.
348+
349+-- $async
350+--
351+-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
352+-- streams in a depth first manner with parallel look ahead. We use the
353+-- 'fromAsync' type combinator to effect 'Async' style of composition. We
354+-- can also use the 'Async' type annotation for the stream type to achieve
355+-- the same effect.
356+--
357+-- When two streams with multiple elements are combined in this manner, the
358+-- streams are traversed in depth first manner just like 'Serial', however it
359+-- can execute the next stream concurrently and return the results from it
360+-- as they arrive i.e. the results from the next stream may be yielded even
361+-- before the results from the first stream. Concurrent execution of the next
362+-- stream(s) is performed if the first stream blocks or if it cannot produce
363+-- output at the rate that is enough to meet the consumer demand. Multiple
364+-- streams can be executed concurrently to meet the demand.
365+-- In the example below each element in the stream introduces a constant delay
366+-- of 1 second, however, it takes just one second to produce all the results.
367+-- The results are not guaranteed to be in any particular order:
368+--
369+-- >>> p n = threadDelay 1000000 >> return n
370+-- >>> stream1 = p 1 |: p 2 |: Stream.nil
371+-- >>> stream2 = p 3 |: p 4 |: Stream.nil
372+-- >>> Stream.toList $ Stream.fromAsync $ stream1 <> stream2
373+-- ...
374+--
375+-- The constituent streams are also composed in 'Async' manner and the
376+-- composition of streams too. We can compose the constituent streams to run
377+-- serially, in that case it would take 2 seconds to produce all the results.
378+-- The elements in the serial streams would be in serial order in the results:
379+--
380+-- >>> p n = threadDelay 1000000 >> return n
381+-- >>> stream = (Stream.fromSerial stream1) <> (Stream.fromSerial stream2)
382+-- >>> Stream.toList $ Stream.fromAsync stream
383+-- ...
384+--
385+-- In the following example we can see that new threads are started when a
386+-- computation blocks. Notice that the output from the stream with the
387+-- shortest delay is printed first. The whole computation takes @maximum of
388+-- (3, 2, 1) = 3@ seconds:
389+--
390+-- >>> Stream.drain $ Stream.fromAsync $ delay 3 <> delay 2 <> delay 1
391+-- ThreadId ...: Delay 1
392+-- ThreadId ...: Delay 2
393+-- ThreadId ...: Delay 3
394+--
395+-- When we have a tree of computations composed using this style, the tree is
396+-- traversed in DFS style just like the 'Serial' style, the only difference is
397+-- that here we can move on to executing the next stream if a stream blocks.
398+-- However, we will not start new threads if we have sufficient output to
399+-- saturate the consumer. This is why we call it left-biased demand driven or
400+-- adaptive concurrency style, the concurrency tends to stay on the left side
401+-- of the composition as long as possible. More threads are started based on
402+-- the pull rate of the consumer. The following example prints an output every
403+-- second as all of the actions are concurrent.
404+--
405+-- >>> Stream.drain $ Stream.fromAsync $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
406+-- ThreadId ...: Delay 1
407+-- ThreadId ...: Delay 2
408+-- ThreadId ...: Delay 3
409+-- ThreadId ...: Delay 4
410+--
411+-- All the computations may even run in a single thread when more threads are
412+-- not needed. As you can see, in the following example the computations are
413+-- run in a single thread one after another, because none of them blocks.
414+-- However, if the thread consuming the stream were faster than the producer
415+-- then it would have started parallel threads for each computation to keep up
416+-- even if none of them blocks:
417+--
418+-- >>> :{
419+-- traced m = Stream.fromEffect (myThreadId >>= print) >> return m
420+-- stream = traced (sqrt 9) <> traced (sqrt 16) <> traced (sqrt 25)
421+-- main = Stream.drain $ Stream.fromAsync stream
422+-- :}
423+--
424+-- Note that the order of printing in the above examples may change due to
425+-- variations in scheduling latencies for concurrent threads.
426+--
427+-- The polymorphic version of the 'Async' binary operation '<>' is called
428+-- 'async'. We can use 'async' to join streams in a left biased
429+-- adaptively concurrent manner irrespective of the type, notice that we have
430+-- not used the 'fromAsync' combinator in the following example:
431+--
432+-- >>> Stream.drain $ delay 3 `Stream.async` delay 2 `Stream.async` delay 1
433+-- ThreadId ...: Delay 1
434+-- ThreadId ...: Delay 2
435+-- ThreadId ...: Delay 3
436+--
437+-- Since the concurrency provided by this operator is demand driven it cannot
438+-- be used when the composed computations start timers that are relative to
439+-- each other because all computations may not be started at the same time and
440+-- therefore timers in all of them may not start at the same time. When
441+-- relative timing among all computations is important or when we need to start
442+-- all computations at once for any reason 'Parallel' style must be used
443+-- instead.
444+--
445+-- 'Async' style utilizes resources optimally and should be preferred over
446+-- 'Parallel' or 'WAsync' unless you really need those. 'Async' should be used
447+-- when we know that the computations can run in parallel but we do not care if
448+-- they actually run in parallel or not, that decision can be left to the
449+-- scheduler based on demand. Also, note that 'async' operator can be used to fold
450+-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
451+-- because it does not require us to run all of them at the same time in a fair
452+-- manner.
453+
454+-- $wasync
455+--
456+-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
457+-- a concurrent manner using /breadth first traversal/. We use the 'fromWAsync'
458+-- type combinator to effect 'WAsync' style of composition. We can also use the
459+-- 'WAsync' type annotation for the stream to achieve the same effect.
460+--
461+-- When streams with multiple elements are combined in this manner, we traverse
462+-- all the streams concurrently in a breadth first manner i.e. one action from
463+-- each stream is performed and yielded to the resulting stream before we come
464+-- back to the first stream again and so on. Even though we execute the actions
465+-- in a breadth first order the outputs are consumed on a first come first
466+-- serve basis.
467+--
468+-- In the following example we can see that outputs are produced in the breadth
469+-- first traversal order but this is not guaranteed.
470+--
471+-- >>> stream1 = print 1 |: print 2 |: Stream.nil
472+-- >>> stream2 = print 3 |: print 4 |: Stream.nil
473+-- >>> Stream.drain $ Stream.fromWAsync $ stream1 <> stream2
474+-- 1
475+-- 3
476+-- 2
477+-- 4
478+--
479+-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
480+-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
481+-- concurrent traversal irrespective of the type, notice that we have not used
482+-- the 'fromWAsync' combinator in the following example:
483+--
484+-- >>> Stream.drain $ delay 3 `Stream.wAsync` delay 2 `Stream.wAsync` delay 1
485+-- ThreadId ...: Delay 1
486+-- ThreadId ...: Delay 2
487+-- ThreadId ...: Delay 3
488+--
489+-- Since the concurrency provided by this style is demand driven it may not
490+-- be used when the composed computations start timers that are relative to
491+-- each other because all computations may not be started at the same time and
492+-- therefore timers in all of them may not start at the same time. When
493+-- relative timing among all computations is important or when we need to start
494+-- all computations at once for any reason 'Parallel' style must be used
495+-- instead.
496+--
497+
498+-- $parallel
499+--
500+-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
501+-- streams in a fairly concurrent manner with round robin scheduling. We use
502+-- the 'fromParallel' type combinator to effect 'Parallel' style of composition.
503+-- We can also use the 'Parallel' type annotation for the stream type to
504+-- achieve the same effect.
505+--
506+-- When two streams with multiple elements are combined in this manner, the
507+-- monadic actions in both the streams are performed concurrently with a fair
508+-- round robin scheduling. The outputs are yielded in the order in which the
509+-- actions complete. This is pretty similar to the 'WAsync' type, the
510+-- difference is that 'WAsync' is adaptive to the consumer demand and may or
511+-- may not execute all actions in parallel depending on the demand, whereas
512+-- 'Parallel' runs all the streams in parallel irrespective of the demand.
513+--
514+-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
515+-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
516+-- manner irrespective of the type, notice that we have not used the
517+-- 'fromParallel' combinator in the following example:
518+--
519+-- >>> Stream.drain $ delay 3 `Stream.parallel` delay 2 `Stream.wAsync` delay 1
520+-- ThreadId ...: Delay 1
521+-- ThreadId ...: Delay 2
522+-- ThreadId ...: Delay 3
523+--
524+-- Note that this style of composition cannot be used to combine infinite
525+-- number of streams, as it will lead to an infinite sized scheduling queue.
526+--
527+
528+-- XXX to be removed
529+-- $custom
530+--
531+-- The 'mkAsync' API can be used to create references to asynchronously running
532+-- stream computations. We can then use 'uncons' to explore the streams
533+-- arbitrarily and then recompose individual elements to create a new stream.
534+-- This way we can dynamically decide which stream to explore at any given
535+-- time. Take an example of a merge sort of two sorted streams. We need to
536+-- keep consuming items from the stream which has the lowest item in the sort
537+-- order. This can be achieved using async references to streams. See
538+-- "MergeSort.hs" in <https://github.com/composewell/streamly-examples Streamly Examples>.
539+
540+-- $monoid
541+--
542+-- All of the following are equivalent and start ten concurrent tasks each with
543+-- a delay from 1 to 10 seconds, resulting in the printing of each number every
544+-- second:
545+--
546+-- >>> :{
547+-- main = do
548+-- Stream.drain $ Stream.fromAsync $ foldMap delay [1..10]
549+-- Stream.drain $ Stream.concatFoldableWith Stream.async (map delay [1..10])
550+-- Stream.drain $ Stream.concatMapFoldableWith Stream.async delay [1..10]
551+-- Stream.drain $ Stream.concatForFoldableWith Stream.async [1..10] delay
552+-- :}
553+--
554+
555+-- $aheadNesting
556+--
557+-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except
558+-- that it can speculatively perform a bounded number of next iterations of a
559+-- loop concurrently.
560+--
561+-- >>> :{
562+-- Stream.toList $ Stream.fromAhead $ do
563+-- x <- Stream.fromFoldable [3,2,1]
564+-- delay x
565+-- return x
566+-- :}
567+-- ThreadId ...: Delay 1
568+-- ThreadId ...: Delay 2
569+-- ThreadId ...: Delay 3
570+-- [3,2,1]
571+--
572+-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can
573+-- see all the three iterations are concurrent and run in different threads,
574+-- however, the results are returned in the serial order.
575+--
576+-- Concurrency is demand driven, when multiple streams are composed using this
577+-- style, the iterations are executed in a depth first manner just like
578+-- 'Serial' i.e. nested iterations are executed before we proceed to the next
579+-- outer iteration. The only difference is that we may execute multiple future
580+-- iterations concurrently and keep the results ready.
581+--
582+-- The 'fromAhead' type combinator can be used to switch to this style of
583+-- composition. Alternatively, a type annotation can be used to specify the
584+-- type of the stream as 'Ahead'.
585+--
586+
587+-- $concurrentNesting
588+--
589+-- The 'Monad' composition of 'Async' type can perform the iterations of a
590+-- loop concurrently.
591+--
592+-- >>> :{
593+-- Stream.drain $ Stream.fromAsync $ do
594+-- x <- Stream.fromFoldable [3,2,1]
595+-- delay x
596+-- :}
597+-- ThreadId ...: Delay 1
598+-- ThreadId ...: Delay 2
599+-- ThreadId ...: Delay 3
600+--
601+-- As we can see the code after the @fromFoldable@ statement is run three
602+-- times, once for each value of @x@. All the three iterations are concurrent
603+-- and run in different threads. The iteration with least delay finishes first.
604+-- When compared to imperative programming, this can be viewed as a @for@ loop
605+-- with three concurrent iterations.
606+--
607+-- Concurrency is demand driven i.e. more concurrent iterations are started
608+-- only if the previous iterations are not able to saturate the consumer of the
609+-- output stream. This works exactly the same way as the merging of two
610+-- streams using 'async' works.
611+--
612+-- The 'fromAsync' type combinator can be used to switch to this style of
613+-- composition. Alternatively, a type annotation can be used to specify the
614+-- type of the stream as 'Async'.
615+--
616+-- When multiple streams are nested using this style, the iterations are
617+-- concurrently evaluated in a depth first manner:
618+--
619+--
620+-- >>> :{
621+-- Stream.drain $ Stream.fromAsync $ do
622+-- x <- Stream.fromFoldable [1,2]
623+-- y <- Stream.fromFoldable [3,4]
624+-- Stream.fromEffect $ putStrLn $ show (x, y)
625+-- :}
626+-- (1,3)
627+-- (1,4)
628+-- (2,3)
629+-- (2,4)
630+--
631+-- Nested iterations are given preference for concurrent evaluation i.e.
632+-- (1,4) will be scheduled in preference to (2,3).
633+
634+-- $wasyncNesting
635+--
636+-- Like 'Async', the 'Monad' composition of 'WAsync' runs the iterations of a
637+-- loop concurrently. It differs from 'Async' in the nested loop behavior. Like
638+-- 'WSerial', the nested loops in this type are traversed and executed in a
639+-- breadth first manner rather than the depth first manner of 'Async' style.
640+--
641+-- >>> :{
642+-- Stream.drain $ Stream.fromWAsync $ do
643+-- x <- Stream.fromSerial $ Stream.fromFoldable [1,2]
644+-- y <- Stream.fromSerial $ Stream.fromFoldable [3,4]
645+-- Stream.fromEffect $ putStrLn $ show (x, y)
646+-- :}
647+-- (1,3)
648+-- (1,4)
649+-- (2,3)
650+-- (2,4)
651+--
652+-- Note that (2,3) is preferred to (1,4) when evaluating the iterations
653+-- concurrently. This works exactly the same way as the merging of two streams
654+-- using 'wAsync' works.
655+--
656+-- The 'fromWAsync' type combinator can be used to switch to this style of
657+-- composition. Alternatively, a type annotation can be used to specify the
658+-- type of the stream as 'WAsync'.
659+--
660+
661+-- $parallelNesting
662+--
663+-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
664+-- iterations of a loop concurrently.
665+--
666+-- >>> :{
667+-- Stream.drain $ Stream.fromParallel $ do
668+-- x <- Stream.fromFoldable [3,2,1]
669+-- delay x
670+-- :}
671+-- ThreadId ...: Delay 1
672+-- ThreadId ...: Delay 2
673+-- ThreadId ...: Delay 3
674+--
675+-- It differs from 'Async' and 'WAsync' in the nested loop behavior. All
676+-- iterations of the loop are run fully concurrently irrespective of the
677+-- demand. This works exactly the same way as the merging of streams using
678+-- 'parallel' works.
679+--
680+-- The 'fromParallel' type combinator can be used to switch to this style of
681+-- composition. Alternatively, a type annotation can be used to specify the
682+-- type of the stream as 'Parallel'.
683+--
684+
685+-- $applicative
686+--
687+-- 'Async' can run the iterations concurrently, therefore, it takes a total
688+-- of 6 seconds which is max (1, 2) + max (3, 4):
689+--
690+-- >>> (Stream.toList $ Stream.fromAsync $ (,) <$> s1 <*> s2) >>= print
691+-- ...
692+--
693+-- @
694+-- ThreadId 34: Delay 1
695+-- ThreadId 36: Delay 2
696+-- ThreadId 35: Delay 3
697+-- ThreadId 36: Delay 3
698+-- ThreadId 35: Delay 4
699+-- ThreadId 36: Delay 4
700+-- [(1,3),(2,3),(1,4),(2,4)]
701+-- @
702+--
703+-- Similarly, 'WAsync' as well can run the iterations concurrently, but with a
704+-- different style of scheduling than 'Async' as explained in the Monad
705+-- section, therefore, it too takes a total of 6 seconds (2 + 4):
706+--
707+-- >>> (Stream.toList $ Stream.fromWAsync $ (,) <$> s1 <*> s2) >>= print
708+-- ...
709+--
710+-- @
711+-- ThreadId 34: Delay 1
712+-- ThreadId 36: Delay 2
713+-- ThreadId 35: Delay 3
714+-- ThreadId 36: Delay 3
715+-- ThreadId 35: Delay 4
716+-- ThreadId 36: Delay 4
717+-- [(1,3),(2,3),(1,4),(2,4)]
718+-- @
719+
720+-- $parallelzip
721+--
722+-- The applicative instance of 'ZipAsync' type zips streams concurrently.
723+-- 'fromZipAsync' type combinator can be used to switch to parallel applicative
724+-- zip composition:
725+--
726+-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
727+-- are produced concurrently, and 2 and 4 are produced concurrently:
728+--
729+-- >>> d n = delay n >> return n
730+-- >>> s1 = Stream.fromSerial $ d 1 <> d 2
731+-- >>> s2 = Stream.fromSerial $ d 3 <> d 4
732+-- >>> (Stream.toList $ Stream.fromZipAsync $ (,) <$> s1 <*> s2) >>= print
733+-- ThreadId ...: Delay 1
734+-- ThreadId ...: Delay 2
735+-- ThreadId ...: Delay 3
736+-- ThreadId ...: Delay 4
737+-- [(1,3),(2,4)]
738+--
739+
740+-- $concurrent
741+--
742+-- When writing concurrent programs there are two distinct places where the
743+-- programmer can control the concurrency. First, when /composing/ a stream by
744+-- merging multiple streams we can choose an appropriate sum style operators to
745+-- combine them concurrently or serially. Second, when /processing/ a stream in
746+-- a monadic composition we can choose one of the monad composition types to
747+-- choose the desired type of concurrency.
748+--
749+-- In the following example the squares of @x@ and @y@ are computed
750+-- concurrently using the 'async' operation and the square roots of their
751+-- sum are computed serially because of the 'fromSerial' combinator. We can
752+-- choose different combinators for the monadic processing and the stream
753+-- generation, to control the concurrency. We can also use the 'fromAsync'
754+-- combinator instead of explicitly folding with 'async'.
755+--
756+-- >>> import Data.List (sum)
757+-- >>> :{
758+-- main = do
759+-- z <- Stream.toList
760+-- $ Stream.fromSerial -- Serial monadic processing (sqrt below)
761+-- $ do
762+-- x2 <- Stream.concatForFoldableWith Stream.async [1..100] $ -- Concurrent @"for"@ loop
763+-- \x -> return $ x * x -- body of the loop
764+-- y2 <- Stream.concatForFoldableWith Stream.async [1..100] $
765+-- \y -> return $ y * y
766+-- return $ sqrt (x2 + y2)
767+-- print $ sum z
768+-- :}
769+--
770+-- We can see how this directly maps to the imperative style
771+-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
772+-- and operators instead of the ugly pragmas.
773+--
774+-- For more concurrent programming examples see,
775+-- <https://github.com/composewell/streamly-examples>.
776+
777+-- $furtherReading
778+--
779+-- * Read the reactive programming tutorial
780+-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.
docs/ReactiveProgramming.hsnew file+124
@@ -0,0 +1,124 @@
1+{-# OPTIONS_GHC -fno-warn-unused-imports #-}
2+-- |
3+-- Module : ReactiveProgramming
4+-- Copyright : (c) 2017 Composewell Technologies
5+--
6+-- License : BSD3
7+-- Maintainer : streamly@composewell.com
8+--
9+-- In this tutorial we will show how Streamly can be used for reactive
10+-- programming. Before you go through this tutorial we recommend that you take
11+-- a look at the Streamly concurrent programming tutorial.
12+
13+module ReactiveProgramming
14+ (
15+ -- * Reactive Programming
16+ -- $reactive
17+
18+ -- * Where to go next?
19+ -- $furtherReading
20+ )
21+where
22+
23+import Streamly.Prelude
24+import Data.Semigroup
25+import Control.Applicative
26+import Control.Monad
27+import Control.Monad.IO.Class (MonadIO(..))
28+import Control.Monad.Trans.Class (MonadTrans (lift))
29+
30+-- $reactive
31+--
32+-- Reactive programming is nothing but concurrent streaming which is what
33+-- streamly is all about. With streamly we can generate streams of events,
34+-- merge streams that are generated concurrently and process events
35+-- concurrently. We can do all this without any knowledge about the specifics
36+-- of the implementation of concurrency. In the following example you will see
37+-- that the code is just regular Haskell code without much streamly APIs used
38+-- (active hyperlinks are the streamly APIs) and yet it is a reactive
39+-- application.
40+--
41+-- This application has two independent and concurrent sources of event
42+-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
43+-- events that deteriorate the health of the character in the game.
44+-- @userAction@ can be "potion" or "quit". When the user types "potion" the
45+-- health improves and the game continues.
46+--
47+-- @
48+-- {-\# LANGUAGE FlexibleContexts \#-}
49+--
50+-- import "Streamly.Prelude" (MonadAsync, SerialT)
51+-- import "Streamly.Prelude" as Stream
52+-- import Control.Monad (void)
53+-- import Control.Monad.IO.Class (MonadIO(liftIO))
54+-- import Control.Monad.State (MonadState, get, modify, runStateT)
55+--
56+-- data Event = Quit | Harm Int | Heal Int deriving (Show)
57+--
58+-- userAction :: MonadAsync m => 'SerialT' m Event
59+-- userAction = Stream.'repeatM' $ liftIO askUser
60+-- where
61+-- askUser = do
62+-- command <- getLine
63+-- case command of
64+-- "potion" -> return (Heal 10)
65+-- "harm" -> return (Harm 10)
66+-- "quit" -> return Quit
67+-- _ -> putStrLn "Type potion or harm or quit" >> askUser
68+--
69+-- acidRain :: MonadAsync m => 'SerialT' m Event
70+-- acidRain = Stream.'fromAsync' $ Stream.'constRate' 1 $ Stream.'repeatM' $ liftIO $ return $ Harm 1
71+--
72+-- data Result = Check | Done
73+--
74+-- runEvents :: (MonadAsync m, MonadState Int m) => 'SerialT' m Result
75+-- runEvents = do
76+-- event \<- userAction \`Stream.'parallel'` acidRain
77+-- case event of
78+-- Harm n -> modify (\\h -> h - n) >> return Check
79+-- Heal n -> modify (\\h -> h + n) >> return Check
80+-- Quit -> return Done
81+--
82+-- data Status = Alive | GameOver deriving Eq
83+--
84+-- getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status
85+-- getStatus result =
86+-- case result of
87+-- Done -> liftIO $ putStrLn "You quit!" >> return GameOver
88+-- Check -> do
89+-- h <- get
90+-- liftIO $ if (h <= 0)
91+-- then putStrLn "You die!" >> return GameOver
92+-- else putStrLn ("Health = " <> show h) >> return Alive
93+--
94+-- main :: IO ()
95+-- main = do
96+-- putStrLn "Your health is deteriorating due to acid rain, type \\\"potion\\\" or \\\"quit\\\""
97+-- let runGame = Stream.'drainWhile' (== Alive) $ Stream.'mapM' getStatus runEvents
98+-- void $ runStateT runGame 60
99+-- @
100+--
101+-- You can also find the source of this example in the streamly-examples repo
102+-- as <https://github.com/composewell/streamly-examples/tree/master/AcidRain.hs AcidRain.hs>.
103+-- It has been adapted from Gabriel's
104+-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
105+-- package.
106+-- This is much simpler compared to the pipes version because of the builtin
107+-- concurrency in streamly. You can also find a SDL based reactive programming
108+-- example adapted from Yampa in
109+-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare.hs>.
110+
111+-- $performance
112+--
113+-- Streamly is highly optimized for performance, it is designed for serious
114+-- high performing, concurrent and scalable applications. We have created the
115+-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
116+-- package which is specifically and carefully designed to measure the
117+-- performance of Haskell streaming libraries fairly and squarely in the right
118+-- way. Streamly performs at par or even better than most streaming libraries
119+-- for serial operations even though it needs to deal with the concurrency
120+-- capability.
121+
122+-- $furtherReading
123+--
124+-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.
docs/Tutorial.hs+42−888
@@ -6,45 +6,28 @@
6-- License : BSD3
6-- License : BSD3
7-- Maintainer : streamly@composewell.com
7-- Maintainer : streamly@composewell.com
8--
8--
9−-- In this tutorial we will go over the basic concepts and how to use the
9+-- In this tutorial we will show how Streamly can be used for idiomatic
10−-- Streamly library. Before you go through this tutorial we recommend that you
10+-- dataflow programming. Before you go through this tutorial we recommend that
11−-- take a look at:
11+-- you take a look at the Streamly Getting Started guide so that you are ready
12−--
12+-- to try out the examples.
13−-- * The quick overview of streamly in the <README.md README file>.
14−-- * The overview of streams in "Streamly.Prelude" module.
15
13
16module Tutorial
14module Tutorial
17 (
15 (
18 -- * Stream Types
16 -- * Stream Types
19 -- $streams
17 -- $streams
20
18
21− -- * Concurrent Streams
22− -- $concurrentStreams
23−
24− -- * Combining Streams
25− -- $flavors
26−
27 -- * Imports and Supporting Code
19 -- * Imports and Supporting Code
28 -- $imports
20 -- $imports
29
21
30 -- * Generating Streams
22 -- * Generating Streams
31 -- $generating
23 -- $generating
32
24
33− -- * Generating Streams Concurrently
34− -- $generatingConcurrently
35−
36 -- * Eliminating Streams
25 -- * Eliminating Streams
37 -- $eliminating
26 -- $eliminating
38
27
39− -- * Concurrent Pipeline Stages
40− -- $concurrentApplication
41−
42 -- * Transforming Streams
28 -- * Transforming Streams
43 -- $transformation
29 -- $transformation
44
30
45− -- * Mapping Concurrently
46− -- $concurrentTransformation
47−
48 -- * Merging Streams
31 -- * Merging Streams
49
32
50 -- ** Semigroup Style
33 -- ** Semigroup Style
@@ -56,22 +39,6 @@ module Tutorial
56 -- *** Wide Serial Composition ('WSerial')
39 -- *** Wide Serial Composition ('WSerial')
57 -- $interleaved
40 -- $interleaved
58
41
59− -- *** Deep Speculative Composition ('Ahead')
60− -- $ahead
61−
62− -- *** Deep Asynchronous Composition ('Async')
63− -- $async
64−
65− -- *** Wide Asynchronous Composition ('WAsync')
66− -- $wasync
67−
68− -- *** Parallel Asynchronous Composition ('Parallel')
69− -- $parallel
70−
71− -- XXX we should deprecate and remove the mkAsync API
72− -- Custom composition
73− -- custom
74−
75 -- ** Monoid Style
42 -- ** Monoid Style
76 -- $monoid
43 -- $monoid
77
44
@@ -87,18 +54,6 @@ module Tutorial
87 -- *** Wide Serial Nesting ('WSerial')
54 -- *** Wide Serial Nesting ('WSerial')
88 -- $interleavedNesting
55 -- $interleavedNesting
89
56
90− -- *** Deep Speculative Nesting ('Ahead')
91− -- $aheadNesting
92−
93− -- *** Deep Asynchronous Nesting ('Async')
94− -- $concurrentNesting
95−
96− -- *** Wide Asynchronous Nesting ('WAsync')
97− -- $wasyncNesting
98−
99− -- *** Parallel Asynchronous Nesting ('Parallel')
100− -- $parallelNesting
101−
102 -- *** Exercise
57 -- *** Exercise
103 -- $monadExercise
58 -- $monadExercise
104
59
@@ -114,30 +69,6 @@ module Tutorial
114 -- ** Serial Zipping
69 -- ** Serial Zipping
115 -- $serialzip
70 -- $serialzip
116
71
117− -- ** Parallel Zipping
118− -- $parallelzip
119−
120− -- * Monad transformers
121− -- $monadtransformers
122−
123− -- * Concurrent Programming
124− -- $concurrent
125−
126− -- * Reactive Programming
127− -- $reactive
128−
129− -- * Writing Concurrent Programs
130− -- $programs
131−
132− -- * Performance
133− -- $performance
134−
135− -- * Interoperation with Streaming Libraries
136− -- $interop
137−
138− -- * Comparison with Existing Packages
139− -- $comparison
140−
141 -- * Where to go next?
72 -- * Where to go next?
142 -- $furtherReading
73 -- $furtherReading
143 )
74 )
@@ -159,17 +90,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
159-- >>> import qualified Streamly.Prelude as Stream
90-- >>> import qualified Streamly.Prelude as Stream
160-- >>> import qualified Streamly.Data.Fold as Fold
91-- >>> import qualified Streamly.Data.Fold as Fold
161--
92--
162−-- >>> import Control.Concurrent (threadDelay, myThreadId)
163−-- >>> :{
164−-- delay n = Stream.fromEffect $ do
165−-- threadDelay (n * 1000000)
166−-- tid <- myThreadId
167−-- putStrLn (show tid ++ ": Delay " ++ show n)
168−-- :}
169−--
170−-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
171−-- >>> hSetBuffering stdout LineBuffering
172−--
173
93
174-- $imports
94-- $imports
175--
95--
@@ -182,24 +102,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
182-- >>> import qualified Streamly.Prelude as Stream
102-- >>> import qualified Streamly.Prelude as Stream
183-- >>> import qualified Streamly.Data.Fold as Fold
103-- >>> import qualified Streamly.Data.Fold as Fold
184--
104--
185−-- To illustrate concurrent vs serial composition aspects, we will use the
186−-- following @delay@ function to introduce a sleep or delay specified in
187−-- seconds. After the delay it prints the number of seconds it slept.
188−--
189−-- >>> import Control.Concurrent (threadDelay, myThreadId)
190−-- >>> :{
191−-- delay n = Stream.fromEffect $ do
192−-- threadDelay (n * 1000000)
193−-- tid <- myThreadId
194−-- putStrLn (show tid ++ ": Delay " ++ show n)
195−-- :}
196−--
197−-- For concurrent examples, use line buffering, otherwise output from different
198−-- threads may get mixed:
199−--
200−-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
201−-- >>> hSetBuffering stdout LineBuffering
202−--
203
105
204-- $streams
106-- $streams
205--
107--
@@ -227,142 +129,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
227-- on the streams we must use the corresponding functions from
129-- on the streams we must use the corresponding functions from
228-- "Streamly.Prelude" instead of using the base "Prelude".
130-- "Streamly.Prelude" instead of using the base "Prelude".
229
131
230−-- $concurrentStreams
231−--
232−-- Many stream operations can be done concurrently:
233−--
234−-- * Streams can be generated concurrently.
235−--
236−-- * Streams can be merged concurrently.
237−--
238−-- * Multiple stages in a streaming pipeline can run concurrently.
239−--
240−-- * Streams can be mapped and zipped concurrently.
241−--
242−-- * In monadic composition they combine like a list transformer,
243−-- providing concurrent non-determinism.
244−--
245−-- There are three basic concurrent stream styles, 'Ahead', 'Async', and
246−-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that
247−-- they can speculatively execute multiple stream actions concurrently in
248−-- advance. 'Ahead' would return exactly the same stream as 'Serial' except
249−-- that it may execute the actions concurrently. The 'Async' style streams,
250−-- like 'Ahead', speculatively execute multiple stream actions in advance but
251−-- return the results in their finishing order rather than in the stream
252−-- traversal order. 'Parallel' is like 'Async' except that it provides
253−-- unbounded parallelism instead of controlled parallelism.
254−--
255−-- For easy reference, we can classify the stream types based on /execution order/,
256−-- /consumption order/, and /bounded or unbounded/ concurrency.
257−-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
258−-- execution we execute the next action in the stream only after the previous
259−-- one has finished executing. In asynchronous execution multiple actions in
260−-- the stream can be executed asynchronously i.e. the next action can start
261−-- executing even before the first one has finished. Consumption order
262−-- determines the order in which the outputs generated by the composition are
263−-- consumed. Consumption could be serial or asynchronous. In serial
264−-- consumption, the outputs are consumed in the traversal order, in
265−-- asynchronous consumption the outputs are consumed as they arrive i.e. first
266−-- come first serve order.
267−--
268−-- +------------+--------------+--------------+--------------+
269−-- | Type | Execution | Consumption | Concurrency |
270−-- +============+==============+==============+==============+
271−-- | 'Serial' | Serial | Serial | None |
272−-- +------------+--------------+--------------+--------------+
273−-- | 'Ahead' | Asynchronous | Serial | bounded |
274−-- +------------+--------------+--------------+--------------+
275−-- | 'Async' | Asynchronous | Asynchronous | bounded |
276−-- +------------+--------------+--------------+--------------+
277−-- | 'Parallel' | Asynchronous | Asynchronous | unbounded |
278−-- +------------+--------------+--------------+--------------+
279−--
280−-- All these types can be freely inter-converted using type conversion
281−-- combinators or type annotations, without any cost, to achieve the desired
282−-- composition style. To force a particular type of composition, we coerce the
283−-- stream type using the corresponding type adapting combinator from
284−-- 'fromSerial', 'fromAhead', 'fromAsync', or 'fromParallel'. The default stream type
285−-- is inferred as 'Serial' unless you change it by using one of the combinators
286−-- or by using a type annotation.
287−
288−-- $flavors
289−--
290−-- Streams can be combined using '<>' or 'mappend' to form a
291−-- composite. Composite streams can be interpreted in a depth first or
292−-- breadth first manner using an appropriate type conversion before
293−-- consumption. Deep (e.g. 'Serial') stream type variants traverse a
294−-- composite stream in a depth first manner, such that each stream is
295−-- traversed fully before traversing the next stream. Wide
296−-- (e.g. 'WSerial') stream types traverse it in a breadth first
297−-- manner, such that one element from each stream is traversed before
298−-- coming back to the first stream again.
299−--
300−-- Each stream type has a wide traversal variant prefixed by 'W'. The wide
301−-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad
302−-- compositions of the streams.
303−-- The following table summarizes the basic types and the corresponding wide
304−-- variants:
305−--
306−-- @
307−-- +------------+-----------+
308−-- | Deep | Wide |
309−-- +============+===========+
310−-- | 'Serial' | 'WSerial' |
311−-- +------------+-----------+
312−-- | 'Ahead' | 'WAhead' |
313−-- +------------+-----------+
314−-- | 'Async' | 'WAsync' |
315−-- +------------+-----------+
316−-- @
317−--
318−-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
319−-- zip streams serially or concurrently using 'Applicative' operation. These
320−-- types are not monads they are only applicatives and they do not differ in
321−-- 'Semigroup' composition.
322−--
323−
324−-- $programs
325−--
326−-- When writing concurrent programs it is advised to not use the concurrent
327−-- style stream combinators blindly at the top level. That might create too
328−-- much concurrency where it is not even required, and can even degrade
329−-- performance in some cases. In some cases it can also lead to surprising
330−-- behavior because of some code that is supposed to be serial becoming
331−-- concurrent. Please be aware that all concurrency capable APIs that you may
332−-- have used under the scope of a concurrent stream combinator will become
333−-- concurrent. For example if you have a 'repeatM' somewhere in your program
334−-- and you use 'fromParallel' on top, the 'repeatM' becomes fully parallel,
335−-- resulting into an infinite parallel execution . Instead, use the
336−-- /Keep It Serial and Stupid/ principle, start with the default serial
337−-- composition and enable concurrent combinators only when and where necessary.
338−-- When you use a concurrent combinator you can use an explicit 'fromSerial'
339−-- combinator to suppress any unnecessary concurrency under the scope of that
340−-- combinator.
341−
342−-- $monadtransformers
343−--
344−-- To represent streams in an arbitrary monad use the more general monad
345−-- transformer types for example the monad transformer type corresponding to
346−-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values
347−-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a
348−-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for
349−-- @SerialT IO@.
350−--
351−-- Similarly we have monad transformer types for other stream types as well viz.
352−-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
353−--
354−-- To lift a value from an underlying monad in a monad transformer stack into a
355−-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
356−--
357−-- >>> import Control.Monad.IO.Class (liftIO)
358−-- >>> Stream.drain $ liftIO $ putStrLn "Hello world!"
359−-- Hello world!
360−--
361−-- >>> import Control.Monad.Trans.Class (MonadTrans(lift))
362−-- >>> Stream.drain $ lift $ putStrLn "Hello world!"
363−-- Hello world!
364−--
365−
366-- $generating
132-- $generating
367--
133--
368-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
134-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
@@ -408,29 +174,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
408--
174--
409-- For more ways to construct a stream see the module "Streamly.Prelude".
175-- For more ways to construct a stream see the module "Streamly.Prelude".
410
176
411−-- $generatingConcurrently
412−--
413−-- Monadic construction and generation functions like 'consM', 'unfoldrM',
414−-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' work concurrently
415−-- when used with appropriate stream type combinator. The pure versions of
416−-- these APIs are not concurrent, however you can use the monadic versions even
417−-- for pure computations by wrapping the pure value in a monad to get the
418−-- concurrent generation capability where required.
419−--
420−-- The following code finishes in 3 seconds (6 seconds when serial):
421−--
422−-- >>> let p n = threadDelay (n * 1000000) >> return n
423−-- >>> Stream.toList $ Stream.fromParallel $ p 3 |: p 2 |: p 1 |: Stream.nil
424−-- [1,2,3]
425−--
426−-- >>> Stream.toList $ Stream.fromAhead $ p 3 |: p 2 |: p 1 |: Stream.nil
427−-- [3,2,1]
428−--
429−-- The following finishes in 10 seconds (100 seconds when serial):
430−--
431−-- >>> Stream.drain $ Stream.fromAsync $ Stream.replicateM 10 $ p 10
432−--
433−
434-- $eliminating
177-- $eliminating
435--
178--
436-- We have already seen 'drain' and toList to eliminate a stream in the
179-- We have already seen 'drain' and toList to eliminate a stream in the
@@ -478,43 +221,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
478-- :}
221-- :}
479--
222--
480
223
481−-- $concurrentTransformation
482−--
483−-- Monadic transformation functions 'mapM' and 'sequence' work concurrently
484−-- when used with appropriate stream type combinators. The pure versions do not
485−-- work concurrently, however you can use the monadic versions even for pure
486−-- computations to get the concurrent transformation capability where required.
487−--
488−-- This would print a value every second (2 seconds when serial):
489−--
490−-- >>> let p n = threadDelay (n * 1000000) >> return n
491−-- >>> :{
492−-- parMap =
493−-- Stream.repeatM (p 1)
494−-- & Stream.fromSerial -- repeatM is serial
495−-- & Stream.mapM (\x -> p 1 >> print x)
496−-- & Stream.fromAhead -- mapM is cocnurrent using Ahead style
497−-- & Stream.drain
498−-- :}
499−--
500−
501−-- $concurrentApplication
502−--
503−-- The concurrent function application operators '|$' and '|&' apply a stream
504−-- argument to a stream function concurrently to compose a concurrent pipeline
505−-- of stream processing functions:
506−--
507−-- Because both the stages run concurrently, we would see a delay of only 1
508−-- second instead of 2 seconds in the following:
509−--
510−-- >>> let p n = threadDelay (n * 1000000) >> return n
511−-- >>> :{
512−-- parApp =
513−-- Stream.repeatM (p 1)
514−-- |& Stream.mapM (\x -> p 1 >> print x)
515−-- & Stream.drain
516−-- :}
517−
518-- $semigroup
224-- $semigroup
519--
225--
520-- We can combine two streams into a single stream using semigroup composition
226-- We can combine two streams into a single stream using semigroup composition
@@ -551,13 +257,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
551-- 4
257-- 4
552--
258--
553-- All actions in both the streams are performed serially in the same thread.
259-- All actions in both the streams are performed serially in the same thread.
554−-- In the following example we can see that all actions are performed in the
555−-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
556−--
557−-- >>> Stream.drain $ delay 3 <> delay 2 <> delay 1
558−-- ThreadId ... Delay 3
559−-- ThreadId ... Delay 2
560−-- ThreadId ... Delay 1
561--
260--
562-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
261-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
563-- 'serial'. We can use 'serial' to join streams in a sequential manner
262-- 'serial'. We can use 'serial' to join streams in a sequential manner
@@ -592,14 +291,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
592-- 4
291-- 4
593--
292--
594-- Even though the monadic actions of the two streams are performed in an
293-- Even though the monadic actions of the two streams are performed in an
595−-- interleaved manner they are all performed serially in the same thread. In
294+-- interleaved manner they are all performed serially in the same thread.
596−-- the following example we can see that all actions are performed in the same
597−-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
598−--
599−-- >>> Stream.drain $ Stream.fromWSerial $ delay 3 <> delay 2 <> delay 1
600−-- ThreadId ...: Delay 3
601−-- ThreadId ...: Delay 2
602−-- ThreadId ...: Delay 1
603--
295--
604-- The polymorphic version of the 'WSerial' binary operation '<>' is called
296-- The polymorphic version of the 'WSerial' binary operation '<>' is called
605-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
297-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
@@ -615,226 +307,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
615-- Note that this composition cannot be used to fold infinite number of streams
307-- Note that this composition cannot be used to fold infinite number of streams
616-- since it requires preserving the state until a stream is finished.
308-- since it requires preserving the state until a stream is finished.
617
309
618−-- $ahead
619−--
620−-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a
621−-- /serial depth first/ manner with concurrent lookahead. We use the 'fromAhead'
622−-- type combinator to effect 'Ahead' style of composition. We can also use an
623−-- explicit 'Ahead' type annotation for the stream to achieve the same effect.
624−--
625−-- When two streams are combined in this manner, the streams are traversed in
626−-- depth first manner just like 'Serial', however it can execute the next
627−-- stream concurrently and keep the results ready when its turn arrives.
628−-- Concurrent execution of the next stream(s) is performed if the first stream
629−-- blocks or if it cannot produce output at the rate that is enough to meet the
630−-- consumer demand. Multiple streams can be executed concurrently to meet the
631−-- demand. The following example would print the result in a second even
632−-- though each action in each stream takes one second:
633−--
634−-- >>> p n = threadDelay 1000000 >> return n
635−-- >>> stream1 = p 1 |: p 2 |: Stream.nil
636−-- >>> stream2 = p 3 |: p 4 |: Stream.nil
637−-- >>> Stream.toList $ Stream.fromAhead $ stream1 <> stream2
638−-- [1,2,3,4]
639−--
640−-- Each stream is constructed 'fromAhead' and then both the streams are merged
641−-- 'fromAhead', therefore, all the actions can run concurrently but the result is
642−-- presented in serial order.
643−--
644−-- You can also use the polymorphic combinator 'ahead' in place of '<>' to
645−-- compose any type of streams in this manner.
646−
647−-- $async
648−--
649−-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
650−-- streams in a depth first manner with parallel look ahead. We use the
651−-- 'fromAsync' type combinator to effect 'Async' style of composition. We
652−-- can also use the 'Async' type annotation for the stream type to achieve
653−-- the same effect.
654−--
655−-- When two streams with multiple elements are combined in this manner, the
656−-- streams are traversed in depth first manner just like 'Serial', however it
657−-- can execute the next stream concurrently and return the results from it
658−-- as they arrive i.e. the results from the next stream may be yielded even
659−-- before the results from the first stream. Concurrent execution of the next
660−-- stream(s) is performed if the first stream blocks or if it cannot produce
661−-- output at the rate that is enough to meet the consumer demand. Multiple
662−-- streams can be executed concurrently to meet the demand.
663−-- In the example below each element in the stream introduces a constant delay
664−-- of 1 second, however, it takes just one second to produce all the results.
665−-- The results are not guaranteed to be in any particular order:
666−--
667−-- >>> p n = threadDelay 1000000 >> return n
668−-- >>> stream1 = p 1 |: p 2 |: Stream.nil
669−-- >>> stream2 = p 3 |: p 4 |: Stream.nil
670−-- >>> Stream.toList $ Stream.fromAsync $ stream1 <> stream2
671−-- ...
672−--
673−-- The constituent streams are also composed in 'Async' manner and the
674−-- composition of streams too. We can compose the constituent streams to run
675−-- serially, in that case it would take 2 seconds to produce all the results.
676−-- The elements in the serial streams would be in serial order in the results:
677−--
678−-- >>> p n = threadDelay 1000000 >> return n
679−-- >>> stream = (Stream.fromSerial stream1) <> (Stream.fromSerial stream2)
680−-- >>> Stream.toList $ Stream.fromAsync stream
681−-- ...
682−--
683−-- In the following example we can see that new threads are started when a
684−-- computation blocks. Notice that the output from the stream with the
685−-- shortest delay is printed first. The whole computation takes @maximum of
686−-- (3, 2, 1) = 3@ seconds:
687−--
688−-- >>> Stream.drain $ Stream.fromAsync $ delay 3 <> delay 2 <> delay 1
689−-- ThreadId ...: Delay 1
690−-- ThreadId ...: Delay 2
691−-- ThreadId ...: Delay 3
692−--
693−-- When we have a tree of computations composed using this style, the tree is
694−-- traversed in DFS style just like the 'Serial' style, the only difference is
695−-- that here we can move on to executing the next stream if a stream blocks.
696−-- However, we will not start new threads if we have sufficient output to
697−-- saturate the consumer. This is why we call it left-biased demand driven or
698−-- adaptive concurrency style, the concurrency tends to stay on the left side
699−-- of the composition as long as possible. More threads are started based on
700−-- the pull rate of the consumer. The following example prints an output every
701−-- second as all of the actions are concurrent.
702−--
703−-- >>> Stream.drain $ Stream.fromAsync $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
704−-- ThreadId ...: Delay 1
705−-- ThreadId ...: Delay 2
706−-- ThreadId ...: Delay 3
707−-- ThreadId ...: Delay 4
708−--
709−-- All the computations may even run in a single thread when more threads are
710−-- not needed. As you can see, in the following example the computations are
711−-- run in a single thread one after another, because none of them blocks.
712−-- However, if the thread consuming the stream were faster than the producer
713−-- then it would have started parallel threads for each computation to keep up
714−-- even if none of them blocks:
715−--
716−-- >>> :{
717−-- traced m = Stream.fromEffect (myThreadId >>= print) >> return m
718−-- stream = traced (sqrt 9) <> traced (sqrt 16) <> traced (sqrt 25)
719−-- main = Stream.drain $ Stream.fromAsync stream
720−-- :}
721−--
722−-- Note that the order of printing in the above examples may change due to
723−-- variations in scheduling latencies for concurrent threads.
724−--
725−-- The polymorphic version of the 'Async' binary operation '<>' is called
726−-- 'async'. We can use 'async' to join streams in a left biased
727−-- adaptively concurrent manner irrespective of the type, notice that we have
728−-- not used the 'fromAsync' combinator in the following example:
729−--
730−-- >>> Stream.drain $ delay 3 `Stream.async` delay 2 `Stream.async` delay 1
731−-- ThreadId ...: Delay 1
732−-- ThreadId ...: Delay 2
733−-- ThreadId ...: Delay 3
734−--
735−-- Since the concurrency provided by this operator is demand driven it cannot
736−-- be used when the composed computations start timers that are relative to
737−-- each other because all computations may not be started at the same time and
738−-- therefore timers in all of them may not start at the same time. When
739−-- relative timing among all computations is important or when we need to start
740−-- all computations at once for any reason 'Parallel' style must be used
741−-- instead.
742−--
743−-- 'Async' style utilizes resources optimally and should be preferred over
744−-- 'Parallel' or 'WAsync' unless you really need those. 'Async' should be used
745−-- when we know that the computations can run in parallel but we do not care if
746−-- they actually run in parallel or not, that decision can be left to the
747−-- scheduler based on demand. Also, note that 'async' operator can be used to fold
748−-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
749−-- because it does not require us to run all of them at the same time in a fair
750−-- manner.
751−
752−-- $wasync
753−--
754−-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
755−-- a concurrent manner using /breadth first traversal/. We use the 'fromWAsync'
756−-- type combinator to effect 'WAsync' style of composition. We can also use the
757−-- 'WAsync' type annotation for the stream to achieve the same effect.
758−--
759−-- When streams with multiple elements are combined in this manner, we traverse
760−-- all the streams concurrently in a breadth first manner i.e. one action from
761−-- each stream is performed and yielded to the resulting stream before we come
762−-- back to the first stream again and so on. Even though we execute the actions
763−-- in a breadth first order the outputs are consumed on a first come first
764−-- serve basis.
765−--
766−-- In the following example we can see that outputs are produced in the breadth
767−-- first traversal order but this is not guaranteed.
768−--
769−-- >>> stream1 = print 1 |: print 2 |: Stream.nil
770−-- >>> stream2 = print 3 |: print 4 |: Stream.nil
771−-- >>> Stream.drain $ Stream.fromWAsync $ stream1 <> stream2
772−-- 1
773−-- 3
774−-- 2
775−-- 4
776−--
777−-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
778−-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
779−-- concurrent traversal irrespective of the type, notice that we have not used
780−-- the 'fromWAsync' combinator in the following example:
781−--
782−-- >>> Stream.drain $ delay 3 `Stream.wAsync` delay 2 `Stream.wAsync` delay 1
783−-- ThreadId ...: Delay 1
784−-- ThreadId ...: Delay 2
785−-- ThreadId ...: Delay 3
786−--
787−-- Since the concurrency provided by this style is demand driven it may not
788−-- be used when the composed computations start timers that are relative to
789−-- each other because all computations may not be started at the same time and
790−-- therefore timers in all of them may not start at the same time. When
791−-- relative timing among all computations is important or when we need to start
792−-- all computations at once for any reason 'Parallel' style must be used
793−-- instead.
794−--
795−
796−-- $parallel
797−--
798−-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
799−-- streams in a fairly concurrent manner with round robin scheduling. We use
800−-- the 'fromParallel' type combinator to effect 'Parallel' style of composition.
801−-- We can also use the 'Parallel' type annotation for the stream type to
802−-- achieve the same effect.
803−--
804−-- When two streams with multiple elements are combined in this manner, the
805−-- monadic actions in both the streams are performed concurrently with a fair
806−-- round robin scheduling. The outputs are yielded in the order in which the
807−-- actions complete. This is pretty similar to the 'WAsync' type, the
808−-- difference is that 'WAsync' is adaptive to the consumer demand and may or
809−-- may not execute all actions in parallel depending on the demand, whereas
810−-- 'Parallel' runs all the streams in parallel irrespective of the demand.
811−--
812−-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
813−-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
814−-- manner irrespective of the type, notice that we have not used the
815−-- 'fromParallel' combinator in the following example:
816−--
817−-- >>> Stream.drain $ delay 3 `Stream.parallel` delay 2 `Stream.wAsync` delay 1
818−-- ThreadId ...: Delay 1
819−-- ThreadId ...: Delay 2
820−-- ThreadId ...: Delay 3
821−--
822−-- Note that this style of composition cannot be used to combine infinite
823−-- number of streams, as it will lead to an infinite sized scheduling queue.
824−--
825−
826−-- XXX to be removed
827−-- $custom
828−--
829−-- The 'mkAsync' API can be used to create references to asynchronously running
830−-- stream computations. We can then use 'uncons' to explore the streams
831−-- arbitrarily and then recompose individual elements to create a new stream.
832−-- This way we can dynamically decide which stream to explore at any given
833−-- time. Take an example of a merge sort of two sorted streams. We need to
834−-- keep consuming items from the stream which has the lowest item in the sort
835−-- order. This can be achieved using async references to streams. See
836−-- "MergeSort.hs" in <https://github.com/composewell/streamly-examples Streamly Examples>.
837−
838-- $monoid
310-- $monoid
839--
311--
840-- We can use 'Monoid' instances to fold a container of streams in the desired
312-- We can use 'Monoid' instances to fold a container of streams in the desired
@@ -848,16 +320,15 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
848-- * 'concatForFoldableWith' is like @concatMapFoldableWith@ but the container
320-- * 'concatForFoldableWith' is like @concatMapFoldableWith@ but the container
849-- argument comes before the function argument.
321-- argument comes before the function argument.
850--
322--
851−-- All of the following are equivalent and start ten concurrent tasks each with
323+-- All of the following are equivalent:
852−-- a delay from 1 to 10 seconds, resulting in the printing of each number every
853−-- second:
854--
324--
855-- >>> :{
325-- >>> :{
326+-- traced = Stream.fromEffect . print
856-- main = do
327-- main = do
857−-- Stream.drain $ Stream.fromAsync $ foldMap delay [1..10]
328+-- Stream.drain $ foldMap traced [1..10]
858−-- Stream.drain $ Stream.concatFoldableWith Stream.async (map delay [1..10])
329+-- Stream.drain $ Stream.concatFoldableWith Stream.serial (map traced [1..10])
859−-- Stream.drain $ Stream.concatMapFoldableWith Stream.async delay [1..10]
330+-- Stream.drain $ Stream.concatMapFoldableWith Stream.serial traced [1..10]
860−-- Stream.drain $ Stream.concatForFoldableWith Stream.async [1..10] delay
331+-- Stream.drain $ Stream.concatForFoldableWith Stream.serial [1..10] traced
861-- :}
332-- :}
862--
333--
863
334
@@ -905,11 +376,11 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
905-- >>> :{
376-- >>> :{
906-- Stream.drain $ do
377-- Stream.drain $ do
907-- x <- Stream.fromFoldable [3,2,1]
378-- x <- Stream.fromFoldable [3,2,1]
908−-- delay x
379+-- Stream.fromEffect $ print x
909-- :}
380-- :}
910−-- ThreadId ...: Delay 3
381+-- 3
911−-- ThreadId ...: Delay 2
382+-- 2
912−-- ThreadId ...: Delay 1
383+-- 1
913--
384--
914-- As we can see, the code after the @fromFoldable@ statement is run three
385-- As we can see, the code after the @fromFoldable@ statement is run three
915-- times, once for each value of @x@ drawn from the stream. All the three
386-- times, once for each value of @x@ drawn from the stream. All the three
@@ -933,7 +404,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
933-- Stream.drain $ do
404-- Stream.drain $ do
934-- x <- Stream.fromFoldable [1,2]
405-- x <- Stream.fromFoldable [1,2]
935-- y <- Stream.fromFoldable [3,4]
406-- y <- Stream.fromFoldable [3,4]
936−-- Stream.fromEffect $ putStrLn $ show (x, y)
407+-- Stream.fromEffect $ print (x, y)
937-- :}
408-- :}
938-- (1,3)
409-- (1,3)
939-- (1,4)
410-- (1,4)
@@ -947,85 +418,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
947-- Notice that this is analogous to merging streams of type 'Serial' or merging
418-- Notice that this is analogous to merging streams of type 'Serial' or merging
948-- streams using 'serial'.
419-- streams using 'serial'.
949
420
950−-- $aheadNesting
951−--
952−-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except
953−-- that it can speculatively perform a bounded number of next iterations of a
954−-- loop concurrently.
955−--
956−-- >>> :{
957−-- Stream.toList $ Stream.fromAhead $ do
958−-- x <- Stream.fromFoldable [3,2,1]
959−-- delay x
960−-- return x
961−-- :}
962−-- ThreadId ...: Delay 1
963−-- ThreadId ...: Delay 2
964−-- ThreadId ...: Delay 3
965−-- [3,2,1]
966−--
967−-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can
968−-- see all the three iterations are concurrent and run in different threads,
969−-- however, the results are returned in the serial order.
970−--
971−-- Concurrency is demand driven, when multiple streams are composed using this
972−-- style, the iterations are executed in a depth first manner just like
973−-- 'Serial' i.e. nested iterations are executed before we proceed to the next
974−-- outer iteration. The only difference is that we may execute multiple future
975−-- iterations concurrently and keep the results ready.
976−--
977−-- The 'fromAhead' type combinator can be used to switch to this style of
978−-- composition. Alternatively, a type annotation can be used to specify the
979−-- type of the stream as 'Ahead'.
980−--
981−
982−-- $concurrentNesting
983−--
984−-- The 'Monad' composition of 'Async' type can perform the iterations of a
985−-- loop concurrently.
986−--
987−-- >>> :{
988−-- Stream.drain $ Stream.fromAsync $ do
989−-- x <- Stream.fromFoldable [3,2,1]
990−-- delay x
991−-- :}
992−-- ThreadId ...: Delay 1
993−-- ThreadId ...: Delay 2
994−-- ThreadId ...: Delay 3
995−--
996−-- As we can see the code after the @fromFoldable@ statement is run three
997−-- times, once for each value of @x@. All the three iterations are concurrent
998−-- and run in different threads. The iteration with least delay finishes first.
999−-- When compared to imperative programming, this can be viewed as a @for@ loop
1000−-- with three concurrent iterations.
1001−--
1002−-- Concurrency is demand driven i.e. more concurrent iterations are started
1003−-- only if the previous iterations are not able to saturate the consumer of the
1004−-- output stream. This works exactly the same way as the merging of two
1005−-- streams using 'async' works.
1006−--
1007−-- The 'fromAsync' type combinator can be used to switch to this style of
1008−-- composition. Alternatively, a type annotation can be used to specify the
1009−-- type of the stream as 'Async'.
1010−--
1011−-- When multiple streams are nested using this style, the iterations are
1012−-- concurrently evaluated in a depth first manner:
1013−--
1014−--
1015−-- >>> :{
1016−-- Stream.drain $ Stream.fromAsync $ do
1017−-- x <- Stream.fromFoldable [1,2]
1018−-- y <- Stream.fromFoldable [3,4]
1019−-- Stream.fromEffect $ putStrLn $ show (x, y)
1020−-- :}
1021−-- (1,3)
1022−-- (1,4)
1023−-- (2,3)
1024−-- (2,4)
1025−--
1026−-- Nested iterations are given preference for concurrent evaluation i.e.
1027−-- (1,4) will be scheduled in preference to (2,3).
1028−
1029-- $interleavedNesting
421-- $interleavedNesting
1030--
422--
1031-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
423-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
@@ -1035,7 +427,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
1035-- Stream.drain $ Stream.fromWSerial $ do
427-- Stream.drain $ Stream.fromWSerial $ do
1036-- x <- Stream.fromFoldable [1,2]
428-- x <- Stream.fromFoldable [1,2]
1037-- y <- Stream.fromFoldable [3,4]
429-- y <- Stream.fromFoldable [3,4]
1038−-- Stream.fromEffect $ putStrLn $ show (x, y)
430+-- Stream.fromEffect $ print (x, y)
1039-- :}
431-- :}
1040-- (1,3)
432-- (1,3)
1041-- (2,3)
433-- (2,3)
@@ -1050,57 +442,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
1050-- type of the stream as 'WSerial'.
442-- type of the stream as 'WSerial'.
1051--
443--
1052
444
1053−-- $wasyncNesting
1054−--
1055−-- Like 'Async', the 'Monad' composition of 'WAsync' runs the iterations of a
1056−-- loop concurrently. It differs from 'Async' in the nested loop behavior. Like
1057−-- 'WSerial', the nested loops in this type are traversed and executed in a
1058−-- breadth first manner rather than the depth first manner of 'Async' style.
1059−--
1060−-- >>> :{
1061−-- Stream.drain $ Stream.fromWAsync $ do
1062−-- x <- Stream.fromSerial $ Stream.fromFoldable [1,2]
1063−-- y <- Stream.fromSerial $ Stream.fromFoldable [3,4]
1064−-- Stream.fromEffect $ putStrLn $ show (x, y)
1065−-- :}
1066−-- (1,3)
1067−-- (1,4)
1068−-- (2,3)
1069−-- (2,4)
1070−--
1071−-- Note that (2,3) is preferred to (1,4) when evaluating the iterations
1072−-- concurrently. This works exactly the same way as the merging of two streams
1073−-- using 'wAsync' works.
1074−--
1075−-- The 'fromWAsync' type combinator can be used to switch to this style of
1076−-- composition. Alternatively, a type annotation can be used to specify the
1077−-- type of the stream as 'WAsync'.
1078−--
1079−
1080−-- $parallelNesting
1081−--
1082−-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
1083−-- iterations of a loop concurrently.
1084−--
1085−-- >>> :{
1086−-- Stream.drain $ Stream.fromParallel $ do
1087−-- x <- Stream.fromFoldable [3,2,1]
1088−-- delay x
1089−-- :}
1090−-- ThreadId ...: Delay 1
1091−-- ThreadId ...: Delay 2
1092−-- ThreadId ...: Delay 3
1093−--
1094−-- It differs from 'Async' and 'WAsync' in the nested loop behavior. All
1095−-- iterations of the loop are run fully concurrently irrespective of the
1096−-- demand. This works exactly the same way as the merging of streams using
1097−-- 'parallel' works.
1098−--
1099−-- The 'fromParallel' type combinator can be used to switch to this style of
1100−-- composition. Alternatively, a type annotation can be used to specify the
1101−-- type of the stream as 'Parallel'.
1102−--
1103−
1104-- $monadExercise
445-- $monadExercise
1105--
446--
1106-- Streamly code is usually written in a way that is agnostic of the
447-- Streamly code is usually written in a way that is agnostic of the
@@ -1114,7 +455,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
1114-- sz <- sizes
455-- sz <- sizes
1115-- cl <- colors
456-- cl <- colors
1116-- sh <- shapes
457-- sh <- shapes
1117−-- Stream.fromEffect $ putStrLn $ show (sz, cl, sh)
458+-- Stream.fromEffect $ print (sz, cl, sh)
1118-- where
459-- where
1119-- sizes = Stream.fromFoldable [1, 2, 3]
460-- sizes = Stream.fromFoldable [1, 2, 3]
1120-- colors = Stream.fromFoldable ["red", "green", "blue"]
461-- colors = Stream.fromFoldable ["red", "green", "blue"]
@@ -1153,65 +494,32 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
1153-- provided.
494-- provided.
1154--
495--
1155-- The following is an example of 'Serial' applicative, it runs all iterations
496-- The following is an example of 'Serial' applicative, it runs all iterations
1156−-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
497+-- serially:
1157--
498--
1158−-- >>> d n = delay n >> return n
499+-- >>> p n = Stream.fromEffect (print n) >> return n
1159−-- >>> s1 = d 1 <> d 2
500+-- >>> s1 = p 1 <> p 2
1160−-- >>> s2 = d 3 <> d 4
501+-- >>> s2 = p 3 <> p 4
1161--
502--
1162-- >>> (Stream.toList $ Stream.fromSerial $ (,) <$> s1 <*> s2) >>= print
503-- >>> (Stream.toList $ Stream.fromSerial $ (,) <$> s1 <*> s2) >>= print
1163−-- ThreadId ...: Delay 1
504+-- 1
1164−-- ThreadId ...: Delay 3
505+-- 3
1165−-- ThreadId ...: Delay 4
506+-- 4
1166−-- ThreadId ...: Delay 2
507+-- 2
1167−-- ThreadId ...: Delay 3
508+-- 3
1168−-- ThreadId ...: Delay 4
509+-- 4
1169-- [(1,3),(1,4),(2,3),(2,4)]
510-- [(1,3),(1,4),(2,3),(2,4)]
1170--
511--
1171-- Similarly, 'WSerial' applicative runs the iterations in an interleaved
512-- Similarly, 'WSerial' applicative runs the iterations in an interleaved
1172-- order but being serial it too takes a total of 17 seconds:
513-- order but being serial it too takes a total of 17 seconds:
1173--
514--
1174-- >>> (Stream.toList $ Stream.fromWSerial $ (,) <$> s1 <*> s2) >>= print
515-- >>> (Stream.toList $ Stream.fromWSerial $ (,) <$> s1 <*> s2) >>= print
1175−-- ThreadId ...: Delay 1
516+-- 1
1176−-- ThreadId ...: Delay 3
517+-- 3
1177−-- ThreadId ...: Delay 2
518+-- 2
1178−-- ThreadId ...: Delay 3
519+-- 3
1179−-- ThreadId ...: Delay 4
520+-- 4
1180−-- ThreadId ...: Delay 4
521+-- 4
1181−-- [(1,3),(2,3),(1,4),(2,4)]
1182−--
1183−-- 'Async' can run the iterations concurrently, therefore, it takes a total
1184−-- of 6 seconds which is max (1, 2) + max (3, 4):
1185−--
1186−-- >>> (Stream.toList $ Stream.fromAsync $ (,) <$> s1 <*> s2) >>= print
1187−-- ...
1188−--
1189−-- @
1190−-- ThreadId 34: Delay 1
1191−-- ThreadId 36: Delay 2
1192−-- ThreadId 35: Delay 3
1193−-- ThreadId 36: Delay 3
1194−-- ThreadId 35: Delay 4
1195−-- ThreadId 36: Delay 4
1196−-- [(1,3),(2,3),(1,4),(2,4)]
1197−-- @
1198−--
1199−-- Similarly, 'WAsync' as well can run the iterations concurrently, but with a
1200−-- different style of scheduling than 'Async' as explained in the Monad
1201−-- section, therefore, it too takes a total of 6 seconds (2 + 4):
1202−--
1203−-- >>> (Stream.toList $ Stream.fromWAsync $ (,) <$> s1 <*> s2) >>= print
1204−-- ...
1205−--
1206−-- @
1207−-- ThreadId 34: Delay 1
1208−-- ThreadId 36: Delay 2
1209−-- ThreadId 35: Delay 3
1210−-- ThreadId 36: Delay 3
1211−-- ThreadId 35: Delay 4
1212−-- ThreadId 36: Delay 4
1213-- [(1,3),(2,3),(1,4),(2,4)]
522-- [(1,3),(2,3),(1,4),(2,4)]
1214−-- @
1215
523
1216-- $zipping
524-- $zipping
1217--
525--
@@ -1228,172 +536,18 @@ import Control.Monad.Trans.Class (MonadTrans (lift))
1228-- 'fromZipSerial' type combinator can be used to switch to serial applicative
536-- 'fromZipSerial' type combinator can be used to switch to serial applicative
1229-- zip composition:
537-- zip composition:
1230--
538--
1231−-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
539+-- >>> p n = Stream.fromEffect (print n) >> return n
1232−-- everything runs serially:
1233−--
1234−-- >>> d n = delay n >> return n
1235−-- >>> s1 = Stream.fromSerial $ d 1 <> d 2
540+-- >>> s1 = Stream.fromSerial $ p 1 <> p 2
1236−-- >>> s2 = Stream.fromSerial $ d 3 <> d 4
541+-- >>> s2 = Stream.fromSerial $ p 3 <> p 4
1237-- >>> (Stream.toList $ Stream.fromZipSerial $ (,) <$> s1 <*> s2) >>= print
542-- >>> (Stream.toList $ Stream.fromZipSerial $ (,) <$> s1 <*> s2) >>= print
1238−-- ThreadId ...: Delay 1
543+-- 1
1239−-- ThreadId ...: Delay 3
544+-- 3
1240−-- ThreadId ...: Delay 2
545+-- 2
1241−-- ThreadId ...: Delay 4
546+-- 4
1242−-- [(1,3),(2,4)]
1243−--
1244−
1245−-- $parallelzip
1246−--
1247−-- The applicative instance of 'ZipAsync' type zips streams concurrently.
1248−-- 'fromZipAsync' type combinator can be used to switch to parallel applicative
1249−-- zip composition:
1250−--
1251−-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1252−-- are produced concurrently, and 2 and 4 are produced concurrently:
1253−--
1254−-- >>> d n = delay n >> return n
1255−-- >>> s1 = Stream.fromSerial $ d 1 <> d 2
1256−-- >>> s2 = Stream.fromSerial $ d 3 <> d 4
1257−-- >>> (Stream.toList $ Stream.fromZipAsync $ (,) <$> s1 <*> s2) >>= print
1258−-- ThreadId ...: Delay 1
1259−-- ThreadId ...: Delay 2
1260−-- ThreadId ...: Delay 3
1261−-- ThreadId ...: Delay 4
1262-- [(1,3),(2,4)]
547-- [(1,3),(2,4)]
1263--
548--
1264
549
1265−-- $concurrent
1266−--
1267−-- When writing concurrent programs there are two distinct places where the
1268−-- programmer can control the concurrency. First, when /composing/ a stream by
1269−-- merging multiple streams we can choose an appropriate sum style operators to
1270−-- combine them concurrently or serially. Second, when /processing/ a stream in
1271−-- a monadic composition we can choose one of the monad composition types to
1272−-- choose the desired type of concurrency.
1273−--
1274−-- In the following example the squares of @x@ and @y@ are computed
1275−-- concurrently using the 'async' operation and the square roots of their
1276−-- sum are computed serially because of the 'fromSerial' combinator. We can
1277−-- choose different combinators for the monadic processing and the stream
1278−-- generation, to control the concurrency. We can also use the 'fromAsync'
1279−-- combinator instead of explicitly folding with 'async'.
1280−--
1281−-- >>> import Data.List (sum)
1282−-- >>> :{
1283−-- main = do
1284−-- z <- Stream.toList
1285−-- $ Stream.fromSerial -- Serial monadic processing (sqrt below)
1286−-- $ do
1287−-- x2 <- Stream.concatForFoldableWith Stream.async [1..100] $ -- Concurrent @"for"@ loop
1288−-- \x -> return $ x * x -- body of the loop
1289−-- y2 <- Stream.concatForFoldableWith Stream.async [1..100] $
1290−-- \y -> return $ y * y
1291−-- return $ sqrt (x2 + y2)
1292−-- print $ sum z
1293−-- :}
1294−--
1295−-- We can see how this directly maps to the imperative style
1296−-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1297−-- and operators instead of the ugly pragmas.
1298−--
1299−-- For more concurrent programming examples see,
1300−-- <https://github.com/composewell/streamly-examples>.
1301−
1302−-- $reactive
1303−--
1304−-- Reactive programming is nothing but concurrent streaming which is what
1305−-- streamly is all about. With streamly we can generate streams of events,
1306−-- merge streams that are generated concurrently and process events
1307−-- concurrently. We can do all this without any knowledge about the specifics
1308−-- of the implementation of concurrency. In the following example you will see
1309−-- that the code is just regular Haskell code without much streamly APIs used
1310−-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1311−-- application.
1312−--
1313−-- This application has two independent and concurrent sources of event
1314−-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1315−-- events that deteriorate the health of the character in the game.
1316−-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1317−-- health improves and the game continues.
1318−--
1319−-- @
1320−-- {-\# LANGUAGE FlexibleContexts \#-}
1321−--
1322−-- import "Streamly.Prelude" (MonadAsync, SerialT)
1323−-- import "Streamly.Prelude" as Stream
1324−-- import Control.Monad (void)
1325−-- import Control.Monad.IO.Class (MonadIO(liftIO))
1326−-- import Control.Monad.State (MonadState, get, modify, runStateT)
1327−--
1328−-- data Event = Quit | Harm Int | Heal Int deriving (Show)
1329−--
1330−-- userAction :: MonadAsync m => 'SerialT' m Event
1331−-- userAction = Stream.'repeatM' $ liftIO askUser
1332−-- where
1333−-- askUser = do
1334−-- command <- getLine
1335−-- case command of
1336−-- "potion" -> return (Heal 10)
1337−-- "harm" -> return (Harm 10)
1338−-- "quit" -> return Quit
1339−-- _ -> putStrLn "Type potion or harm or quit" >> askUser
1340−--
1341−-- acidRain :: MonadAsync m => 'SerialT' m Event
1342−-- acidRain = Stream.'fromAsync' $ Stream.'constRate' 1 $ Stream.'repeatM' $ liftIO $ return $ Harm 1
1343−--
1344−-- data Result = Check | Done
1345−--
1346−-- runEvents :: (MonadAsync m, MonadState Int m) => 'SerialT' m Result
1347−-- runEvents = do
1348−-- event \<- userAction \`Stream.'parallel'` acidRain
1349−-- case event of
1350−-- Harm n -> modify (\\h -> h - n) >> return Check
1351−-- Heal n -> modify (\\h -> h + n) >> return Check
1352−-- Quit -> return Done
1353−--
1354−-- data Status = Alive | GameOver deriving Eq
1355−--
1356−-- getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status
1357−-- getStatus result =
1358−-- case result of
1359−-- Done -> liftIO $ putStrLn "You quit!" >> return GameOver
1360−-- Check -> do
1361−-- h <- get
1362−-- liftIO $ if (h <= 0)
1363−-- then putStrLn "You die!" >> return GameOver
1364−-- else putStrLn ("Health = " <> show h) >> return Alive
1365−--
1366−-- main :: IO ()
1367−-- main = do
1368−-- putStrLn "Your health is deteriorating due to acid rain, type \\\"potion\\\" or \\\"quit\\\""
1369−-- let runGame = Stream.'drainWhile' (== Alive) $ Stream.'mapM' getStatus runEvents
1370−-- void $ runStateT runGame 60
1371−-- @
1372−--
1373−-- You can also find the source of this example in the streamly-examples repo
1374−-- as <https://github.com/composewell/streamly-examples/tree/master/AcidRain.hs AcidRain.hs>.
1375−-- It has been adapted from Gabriel's
1376−-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1377−-- package.
1378−-- This is much simpler compared to the pipes version because of the builtin
1379−-- concurrency in streamly. You can also find a SDL based reactive programming
1380−-- example adapted from Yampa in
1381−-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare.hs>.
1382−
1383−-- $performance
1384−--
1385−-- Streamly is highly optimized for performance, it is designed for serious
1386−-- high performing, concurrent and scalable applications. We have created the
1387−-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1388−-- package which is specifically and carefully designed to measure the
1389−-- performance of Haskell streaming libraries fairly and squarely in the right
1390−-- way. Streamly performs at par or even better than most streaming libraries
1391−-- for serial operations even though it needs to deal with the concurrency
1392−-- capability.
1393−
1394-- $furtherReading
550-- $furtherReading
1395--
551--
1396−-- * Read the documentation of "Streamly" module
552+-- * Read the concurrent streams tutorial
1397−-- * Read the documentation of "Streamly.Prelude" module
1398-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.
553-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.
1399−-- * See the tests in the "test" directory of the package
docs/performance.mdnew file+7
@@ -0,0 +1,7 @@
1+# Performance
2+
3+Streamly is highly optimized for performance, it is designed for
4+high performing, concurrent and scalable applications. Please see the
5+[streaming-benchmarks](https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks)
6+package for microbencmarks measuring the performance of streamly and comparing
7+it with other streaming libraries.
docs/streamly-docs.cabal+2
@@ -26,6 +26,8 @@ library
26 hs-source-dirs: .
26 hs-source-dirs: .
27 exposed-modules:
27 exposed-modules:
28 Tutorial
28 Tutorial
29+ ConcurrentStreams
30+ ReactiveProgramming
29
31
30 build-depends:
32 build-depends:
31 base >= 4.9 && < 5
33 base >= 4.9 && < 5
docs/transformers.md+34−1
@@ -1,3 +1,28 @@
1+# Monad Transformers
2+
3+In the stream tutorials we mostly used streams in the IO monad. In
4+general, the type `SerialT` is a monad transformer, @SerialT m a@
5+represents a stream of values of type 'a' in some underlying monad
6+'m'. For example, @SerialT IO Int@ is a stream of 'Int' in 'IO'
7+monad. Similarly, `SerialT Identity Int` would be a pure stream
8+equivalent to `[a]`.
9+
10+Similarly we have monad transformer types for other stream types as well
11+viz. 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
12+
13+To lift a value from an underlying monad in a monad transformer stack into a
14+singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
15+
16+```
17+>>> import Control.Monad.IO.Class (liftIO)
18+>>> Stream.drain $ liftIO $ putStrLn "Hello world!"
19+Hello world!
20+
21+>>> import Control.Monad.Trans.Class (MonadTrans(lift))
22+>>> Stream.drain $ lift $ putStrLn "Hello world!"
23+Hello world!
24+```
25+
1## Using Monad Transformers
26## Using Monad Transformers
2
27
3Common monad transformers can be used with streamly serial streams, without any
28Common monad transformers can be used with streamly serial streams, without any
@@ -5,12 +30,16 @@ issues. `ReaderT` can be used with concurrent streams as well without any
5issues.
30issues.
6
31
7The semantics of monads other than `ReaderT` with concurrent streams are
32The semantics of monads other than `ReaderT` with concurrent streams are
8−not yet finalized and will change in future, therefore as of now they are not
33+not yet finalized and will change in future, therefore, as of now they are not
9recommended to be used with concurrent streams.
34recommended to be used with concurrent streams.
10
35
11## Ordering of Monad Transformers
36## Ordering of Monad Transformers
12
37
13In most cases it is a good idea to keep streamly as the top level monad.
38In most cases it is a good idea to keep streamly as the top level monad.
39+[This
40+example](https://github.com/composewell/streamly-examples/blob/master/examples/ControlFlow.hs)
41+demonstrates how various control flow modifying monads can be combined
42+with streamly stream monads.
14
43
15## State Sharing
44## State Sharing
16### Serial Applications
45### Serial Applications
@@ -30,3 +59,7 @@ monad or just share the `IORef` by passing it to the required functions. The
30`IORef` can be updated atomically using `atomicModifyIORef`.
59`IORef` can be updated atomically using `atomicModifyIORef`.
31
60
32The `CirclingSquare.hs` example shares an `IORef` across parallel tasks.
61The `CirclingSquare.hs` example shares an `IORef` across parallel tasks.
62+
63+## See also
64+
65+* [Examples of control flow monads with Streamly](https://github.com/composewell/streamly-examples/blob/master/examples/ControlFlow.hs)
streamly.cabal+1−3
@@ -103,9 +103,7 @@ extra-source-files:
103 design/*.png
103 design/*.png
104 design/*.rst
104 design/*.rst
105 docs/*.md
105 docs/*.md
106− docs/Setup.hs
106+ docs/*.hs
107− docs/Tutorial.hs
108− docs/doctests.hs
109 docs/streamly-docs.cabal
107 docs/streamly-docs.cabal
110 examples/README.md
108 examples/README.md
111 src/Streamly/Internal/Data/Stream/Instances.hs
109 src/Streamly/Internal/Data/Stream/Instances.hs