on ff475e7 help sign in
streamly/src/Streamly/Tutorial.hs
1544 lines 57 kB history raw
1{-# OPTIONS_GHC -fno-warn-unused-imports #-}
2-- |
3-- Module : Streamly.Tutorial
4-- Copyright : (c) 2017 Harendra Kumar
5--
6-- License : BSD3
7-- Maintainer : harendra.kumar@gmail.com
8--
9-- Streamly is a general computing framework based on streaming IO. The IO
10-- monad and pure lists are a special case of streamly. On one hand, streamly
11-- extends the lists of pure values to lists of monadic actions, on the other
12-- hand it extends the IO monad with concurrrent non-determinism. In simple
13-- imperative terms we can say that streamly extends the IO monad with @for@
14-- loops and nested @for@ loops with concurrency support. You can understand
15-- this analogy better once you can go through this tutorial.
16--
17-- Streaming in general enables writing modular, composable and scalable
18-- applications with ease, and concurrency allows you to make them scale and
19-- perform well. Streamly enables writing scalable concurrent applications
20-- without being aware of threads or synchronization. No explicit thread
21-- control is needed, where applicable the concurrency rate is automatically
22-- controlled based on the demand by the consumer. However, combinators can be
23-- used to fine tune the concurrency control.
24--
25-- Streaming and concurrency together enable expressing reactive applications
26-- conveniently. See the @CirclingSquare@ example in the examples directory for
27-- a simple SDL based FRP example. To summarize, streamly provides a unified
28-- computing framework for streaming, non-determinism and functional reactive
29-- programming in an elegant and simple API that is a natural extension of pure
30-- lists to monadic streams.
31--
32-- In this tutorial we will go over the basic concepts and how to
33-- use the library. See the last section for further reading resources.
35module Streamly.Tutorial
36 (
37 -- * Streams
38 -- $streams
40 -- * Flavors of Streams
41 -- $flavors
43 -- * Imports and Supporting Code
44 -- $imports
46 -- * Generating Streams
47 -- $generating
49 -- * Eliminating Streams
50 -- $eliminating
52 -- * Transforming Streams
53 -- $transformation
55 -- * Merging Streams
57 -- ** Semigroup Style
58 -- $semigroup
60 -- *** Deep Serial Composition ('Serial')
61 -- $serial
63 -- *** Wide Serial Composition ('WSerial')
64 -- $interleaved
66 -- *** Deep Asynchronous Composition ('Async')
67 -- $async
69 -- *** Wide Asynchronous Composition ('WAsync')
70 -- $wasync
72 -- *** Parallel Asynchronous Composition ('Parallel')
73 -- $parallel
75 -- *** Custom composition
76 -- $custom
78 -- ** Monoid Style
79 -- $monoid
81 -- * Nesting Streams
82 -- $nesting
84 -- ** Monad
85 -- $monad
87 -- *** Deep Serial Nesting ('Serial')
88 -- $regularSerial
90 -- *** Wide Serial Nesting ('WSerial')
91 -- $interleavedNesting
93 -- *** Deep Asynchronous Nesting ('Async')
94 -- $concurrentNesting
96 -- *** Wide Asynchronous Nesting ('WAsync')
97 -- $wasyncNesting
99 -- *** Parallel Asynchronous Nesting ('Parallel')
100 -- $parallelNesting
102 -- *** Exercise
103 -- $monadExercise
105 -- ** Applicative
106 -- $applicative
108 -- ** Functor
109 -- $functor
111 -- * Zipping Streams
112 -- $zipping
114 -- ** Serial Zipping
115 -- $serialzip
117 -- ** Parallel Zipping
118 -- $parallelzip
120 -- * Monad transformers
121 -- $monadtransformers
123 -- * Concurrent Programming
124 -- $concurrent
126 -- * Reactive Programming
127 -- $reactive
129 -- * Writing Concurrent Programs
130 -- $programs
132 -- * Performance
133 -- $performance
135 -- * Interoperation with Streaming Libraries
136 -- $interop
138 -- * Comparison with Existing Packages
139 -- $comparison
141 -- * Where to go next?
142 -- $furtherReading
143 )
144where
146import Streamly
147import Streamly.Prelude
148import Data.Semigroup
149import Control.Applicative
150import Control.Monad
151import Control.Monad.IO.Class (MonadIO(..))
152import Control.Monad.Trans.Class (MonadTrans (lift))
154-- $streams
155--
156-- The way a list represents a sequence of pure values, a stream represents a
157-- sequence of monadic actions. The monadic stream API offered by Streamly is
158-- very close to the Haskell "Prelude" pure lists' API, it can be considered as a
159-- natural extension of lists to monadic actions. Streamly streams provide
160-- concurrent composition and merging of streams. It can be considered as a
161-- concurrent list transformer. In contrast to the "Prelude" lists, merging or
162-- appending streams of arbitrary length is scalable and inexpensive.
163--
164-- The basic stream type is 'Serial', it represents a sequence of IO actions,
165-- and is a 'Monad'. The 'Serial' monad is almost a drop in replacement for
166-- the 'IO' monad, IO monad is a special case of the 'Serial' monad; IO monad
167-- represents a single IO action whereas the 'Serial' monad represents a series
168-- of IO actions. The only change you need to make to go from 'IO' to 'Serial'
169-- is to use 'runStream' to run the monad and to prefix the IO actions with
170-- either 'once' or 'liftIO'. If you use liftIO you can switch from 'Serial'
171-- to IO monad by simply removing the 'runStream' function; no other changes
172-- are needed unless you have used some stream specific composition or
173-- combinators.
174--
175-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
176-- pure lists are a special case of monadic streams. If you use 'nil' in place
177-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' The
178-- only difference is that the elements must be monadic type and to operate on
179-- the streams we must use the corresponding functions from "Streamly.Prelude"
180-- instead of using the base "Prelude".
182-- $flavors
183--
184-- There are a few more types similar to 'Serial' that all represent streams
185-- and differ only in the 'Semigroup', 'Applicative' and 'Monad' compositions
186-- of the streams.
187--
188-- The composition of two or more streams is distinguished based on three
189-- characterstics, /traversal order/, /execution order/ and
190-- /consumption order/. Traversal of a composition of streams could be @deep@
191-- or @wide@. Deep goes depth first i.e. each stream is traversed fully
192-- before we traverse the next stream. Wide goes breadth first i.e. one element
193-- from each stream is traversed before coming back to the first stream again.
194-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
195-- execution we execute an action in the next stream only after the first has
196-- finished executing. In asynchronous execution actions in both streams can be
197-- executed asynchronously i.e. the next action can start executing even before
198-- the first one has finished. The third parameter is consumption order that is
199-- in what order the output generated by the composition is consumed.
200-- Consumption could be serial or asynchronous. In serial consumption, the
201-- outputs are consumed in the traversal order, in asynchronous consumption the
202-- outputs are consumed as they arrive i.e. first come first serve order.
203--
204-- The following table summarizes different styles of streams based on how they
205-- compose. All these types are monads and they differ in 'Semigroup',
206-- 'Applicative' and 'Monad' compositions:
207--
208-- @
209-- +------------+-----------+--------------+--------------+
210-- | Type | Traversal | Execution | Consumption |
211-- +============+===========+==============+==============+
212-- | 'Serial' | Deep | Serial | Serial |
213-- +------------+-----------+--------------+--------------+
214-- | 'WSerial' | Wide | Serial | Serial |
215-- +------------+-----------+--------------+--------------+
216-- | 'Async' | Deep | Asynchronous | Asynchronous |
217-- +------------+-----------+--------------+--------------+
218-- | 'WAsync' | Wide | Asynchronous | Asynchronous |
219-- +------------+-----------+--------------+--------------+
220-- | 'Parallel' | Parallel | Asynchronous | Asynchronous |
221-- +------------+-----------+--------------+--------------+
222-- @
223--
224-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
225-- zip streams serially or concurrently using 'Applicative' operation. These
226-- types are not monads they are only applicatives and they do not differ in
227-- 'Semigroup' composition.
228--
229-- All these types can be freely inter-converted using type conversion
230-- combinators or type annotations without any cost, to acheive the desired
231-- composition style. To force a particular type of composition we coerce the
232-- stream type using the corresponding type adapting combinator from
233-- 'serially', 'wSerially', 'asyncly', 'wAsyncly', 'parallely', 'zipSerially'
234-- or 'zipAsyncly'. The default stream type is inferred as 'Serial' unless you
235-- change it by using one of the combinators or using a type annotation.
237-- $programs
238--
239-- When writing concurrent programs it is advised to not use the concurrent
240-- style stream combinators blindly at the top level. That might create too
241-- much concurrency where it is not even required, and can even degrade
242-- performance in some cases. In some cases it can also lead to surprising
243-- behavior because of some code that is supposed to be serial becoming
244-- concurrent. Please be aware that all concurrency capable APIs that you may
245-- have used under the scope of a concurrent stream combinator will become
246-- concurrent. For example if you have a 'repeatM' somewhere in your program
247-- and you use 'parallely' on top, the 'repeatM' becomes fully parallel,
248-- resulting into an infinite parallel execution . Instead, use the
249-- /Keep It Serial and Stupid/ principle, start with the default serial
250-- composition and enable concurrent combinators only when and where necessary.
251-- When you use a concurrent combinator you can use an explicit 'serially'
252-- combinator to suppress any unnecessary concurrency under the scope of that
253-- combinator.
255-- $monadtransformers
256--
257-- To represent streams in an arbitrary monad use the more general monad
258-- transformer types for example the monad transformer type corresponding to
259-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values
260-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a
261-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for
262-- @SerialT IO@.
263--
264-- Similarly we have monad transformer types for other stream types as well viz.
265-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
266--
267-- To lift a value from an underlying monad in a monad transformer stack into a
268-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
269--
270-- @
271-- > 'runStream' $ liftIO $ putStrLn "Hello world!"
272-- Hello world!
273-- > 'runStream' $ lift $ putStrLn "Hello world!"
274-- Hello world!
275-- @
276--
278-- $generating
279--
280-- We will assume the following imports in this tutorial. Go ahead, fire up a
281-- GHCi session and import these lines to start playing.
282--
283-- @
284-- > import "Streamly"
285-- > import "Streamly.Prelude" ((|:))
286-- > import qualified "Streamly.Prelude" as S
287-- @
288--
289-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
290-- a monadic action at the head of the stream.
291--
292-- @
293-- > S.'toList' S.'nil'
294-- []
295-- > S.'toList' $ 'getLine' |: 'getLine' |: S.'nil'
296-- hello
297-- world
298-- ["hello","world"]
299-- @
300--
301-- To create a singleton stream from a pure value use 'pure' and to create a
302-- singleton stream from a monadic action use 'once'.
303--
304-- @
305-- > S.'toList' $ 'pure' 1
306-- [1]
307-- > S.'toList' $ S.'once' 'getLine'
308-- hello
309-- ["hello"]
310-- @
311--
312-- To create a stream from pure values in a 'Foldable' container use
313-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
314--
315-- @
316-- > S.'toList' $ S.'fromFoldable' [1..3]
317-- [1,2,3]
318-- > S.'toList' $ 'Prelude.foldr' S.'cons' S.'nil' [1..3]
319-- [1,2,3]
320-- @
321--
322-- To create a stream from monadic actions in a 'Foldable' container just use a
323-- right fold using 'consM' and 'nil':
324--
325-- @
326-- > 'runStream' $ 'Prelude.foldr' ('|:') S.'nil' ['putStr' "Hello ", 'putStrLn' "world!"]
327-- Hello world!
328-- @
329--
330-- For more ways to construct a stream see the module "Streamly.Prelude".
331--
332-- Note that monadic generation functions e.g. 'consM', 'unfoldrM',
333-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' etc work
334-- concurrently when used with appropriate stream type combinator. The pure
335-- versions do not work concurrently, however you can use the monadic versions
336-- even for pure computations to get the concurrent generation capability where
337-- required.
338--
339-- TBD add more details and examples about concurrent generation.
341-- $eliminating
342--
343-- We have already seen 'runStream' and 'toList' to eliminate a stream in the
344-- examples above. 'runStream' runs a stream discarding the results i.e. only
345-- for effects. 'toList' runs the stream and collects the results in a list.
346--
347-- For other ways to eliminate a stream see the @Folding@ section in
348-- "Streamly.Prelude" module.
350-- $transformation
351--
352-- Transformation over a stream is the equivalent of a @for@ loop construct in
353-- imperative paradigm. We iterate over every element in the stream and perform
354-- certain transformations for each element. Transformations may involve
355-- mapping functions over the elements, filtering elements from the stream or
356-- folding all the elements in the stream into a single value. Streamly streams
357-- are exactly like lists and you can perform all the transformations in the
358-- same way as you would on lists.
359--
360-- Here is a simple console echo program that just echoes every input line,
361-- forever:
362--
363-- @
364-- > 'runStream' $ S.'repeatM' getLine & S.'mapM' putStrLn
365-- @
366--
367-- The following code snippet reads lines from standard input, filters blank
368-- lines, drops the first non-blank line, takes the next two, up cases them,
369-- numbers them and prints them:
370--
371-- @
372-- import "Streamly"
373-- import qualified "Streamly.Prelude" as S
374-- import Data.Char (toUpper)
375-- import Data.Function ((&))
376--
377-- main = 'runStream' $
378-- S.'repeatM' getLine
379-- & S.'filter' (not . null)
380-- & S.'drop' 1
381-- & S.'take' 2
382-- & fmap (map toUpper)
383-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
384-- & S.'mapM' putStrLn
385-- @
386--
387-- Note that monadic transformation functions 'mapM' and 'sequence' work
388-- concurrently when used with appropriate stream type combinator. The pure
389-- versions do not work concurrently, however you can use the monadic versions
390-- even for pure computations to get the concurrent transformation capability
391-- where required.
392--
393-- TBD add more details and examples about concurrent transformation.
394-- TBD add details about concurrent function application.
396-- $semigroup
397--
398-- We can combine two streams into a single stream using semigroup composition
399-- operation '<>'. Streams can be combined in many different ways as described
400-- in the following sections, the '<>' operation behaves differently depending
401-- on the stream type in effect. The stream type and therefore the composition
402-- style can be changed at any point using one of the type combinators as
403-- discussed earlier.
405-- $imports
406--
407-- In most of example snippets we do not repeat the imports. Where imports are
408-- not explicitly specified use the imports shown below.
409--
410-- @
411-- import "Streamly"
412-- import "Streamly.Prelude" ((|:), nil)
413-- import qualified "Streamly.Prelude" as S
414--
415-- import Control.Concurrent
416-- import Control.Monad (forever)
417-- @
418--
419-- To illustrate concurrent vs serial composition aspects, we will use the
420-- following @delay@ function to introduce a sleep or delay specified in
421-- seconds. After the delay it prints the number of seconds it slept.
422--
423-- @
424-- delay n = S.'once' $ do
425-- threadDelay (n * 1000000)
426-- tid \<- myThreadId
427-- putStrLn (show tid ++ ": Delay " ++ show n)
428-- @
430-- $serial
431--
432-- The 'Semigroup' operation '<>' of the 'Serial' type combines the two streams
433-- in a /serial depth first/ manner. We use the 'serially' type combinator to
434-- effect 'Serial' style of composition. We can also use an explicit 'Serial'
435-- type annotation for the stream to acheive the same effect. However, since
436-- 'Serial' is the default type unless explicitly specified by using a
437-- combinator, we can omit using an explicit combinator or type annotation for
438-- this style of composition.
439--
440-- When two streams with multiple elements are combined in this manner, the
441-- monadic actions in the two streams are performed sequentially i.e. first all
442-- actions in the first stream are performed sequentially and then all actions
443-- in the second stream are performed sequentially. We call it
444-- /serial depth first/ as the full depth of one stream is fully traversed
445-- before we move to the next. The following example prints the sequence 1, 2,
446-- 3, 4:
447--
448-- @
449-- main = 'runStream' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
450-- @
451-- @
452-- 1
453-- 2
454-- 3
455-- 4
456-- @
457--
458-- All actions in both the streams are performed serially in the same thread.
459-- In the following example we can see that all actions are performed in the
460-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
461--
462-- @
463-- main = 'runStream' $ delay 3 <> delay 2 <> delay 1
464-- @
465-- @
466-- ThreadId 36: Delay 3
467-- ThreadId 36: Delay 2
468-- ThreadId 36: Delay 1
469-- @
470--
471-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
472-- 'serial'. We can use 'serial' to join streams in a sequential manner
473-- irrespective of the type of stream:
474--
475-- @
476-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`serial` (print 3 |: print 4 |: nil)
477-- @
479-- $interleaved
480--
481-- The 'Semigroup' operation '<>' of the 'WSerial' type combines the two
482-- streams in a /serial breadth first/ manner. We use the 'wSerially' type
483-- combinator to effect 'WSerial' style of composition. We can also use the
484-- 'WSerial' type annotation for the stream to acheive the same effect.
485--
486-- When two streams with multiple elements are combined in this manner, we
487-- traverse all the streams in a breadth first manner i.e. one action from each
488-- stream is peformed and yielded to the resulting stream before we come back
489-- to the first stream again and so on.
490-- The following example prints the sequence 1, 3, 2, 4
491--
492-- @
493-- main = 'runStream' . 'wSerially' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
494-- @
495-- @
496-- 1
497-- 3
498-- 2
499-- 4
500-- @
501--
502-- Even though the monadic actions of the two streams are performed in an
503-- interleaved manner they are all performed serially in the same thread. In
504-- the following example we can see that all actions are performed in the same
505-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
506--
507-- @
508-- main = 'runStream' . 'wSerially' $ delay 3 <> delay 2 <> delay 1
509-- @
510-- @
511-- ThreadId 36: Delay 3
512-- ThreadId 36: Delay 2
513-- ThreadId 36: Delay 1
514-- @
515--
516-- The polymorphic version of the 'WSerial' binary operation '<>' is called
517-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
518-- irrespective of the type, notice that we have not used the 'wSerially'
519-- combinator in the following example:
520--
521-- @
522-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`wSerial` (print 3 |: print 4 |: nil)
523-- @
524-- @
525-- 1
526-- 3
527-- 2
528-- 4
529-- @
530--
531-- Note that this composition cannot be used to fold infinite number of streams
532-- since it requires preserving the state until a stream is finished.
534-- $async
535--
536-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
537-- streams in a depth first manner with parallel look ahead. We use the
538-- 'asyncly' type combinator to effect 'Async' style of composition. We
539-- can also use the 'Async' type annotation for the stream type to acheive
540-- the same effect.
541--
542-- When two streams with multiple elements are combined in this manner, the
543-- streams are traversed in depth first manner just like 'Serial', however it
544-- can execute the next stream concurrently and return the results from it
545-- as they arrive i.e. the results from the next stream may be yielded even
546-- before the results from the first stream. Concurrent execution of the next
547-- stream(s) is performed if the first stream blocks or if it cannot produce
548-- output at the rate that is enough to meet the consumer demand. Multiple
549-- streams can be executed concurrently to meet the demand.
550-- In the following example the first stream does not block,
551-- therefore the first stream is completely exhausted before the second.
552--
553-- @
554-- main = 'runStream' . 'asyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
555-- @
556-- @
557-- 1
558-- 2
559-- 3
560-- 4
561-- @
562--
563-- If the first stream blocks, we can yield from the second. In the example
564-- below each yield in the stream has a constant delay of 1 second therefore 1
565-- and 3 would be yielded first and then 2 and 4 would be yielded.
566--
567-- @
568-- main = 'runStream' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
569-- where p n = threadDelay 1000000 >> print n
570-- @
571-- @
572-- 1
573-- 3
574-- 2
575-- 4
576-- @
577--
578-- In the following example we can see that new threads are started when a
579-- computation blocks. Notice that the output from the stream with the
580-- shortest delay is printed first. The whole computation takes @maximum of
581-- (3, 2, 1) = 3@ seconds:
582--
583-- @
584-- main = 'runStream' . 'asyncly' $ delay 3 '<>' delay 2 '<>' delay 1
585-- @
586-- @
587-- ThreadId 42: Delay 1
588-- ThreadId 41: Delay 2
589-- ThreadId 40: Delay 3
590-- @
591--
592-- When we have a tree of computations composed using this style, the tree is
593-- traversed in DFS style just like the 'Serial' style, the only difference is
594-- that here we can move on to executing the next stream if a stream blocks.
595-- However, we will not start new threads if we have sufficient output to
596-- saturate the consumer. This is why we call it left-biased demand driven or
597-- adaptive concurrency style, the concurrency tends to stay on the left side
598-- of the composition as long as possible. More threads are started based on
599-- the pull rate of the consumer. The following example prints an output every
600-- second as all of the actions are concurrent.
601--
602-- @
603-- main = 'runStream' . 'asyncly' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
604-- @
605-- @
606-- 1
607-- 2
608-- 3
609-- 4
610-- @
611--
612-- All the computations may even run in a single thread when more threads are
613-- not needed. As you can see, in the following example the computations are
614-- run in a single thread one after another, because none of them blocks.
615-- However, if the thread consuming the stream were faster than the producer
616-- then it would have started parallel threads for each computation to keep up
617-- even if none of them blocks:
618--
619-- @
620-- main = 'runStream' . 'asyncly' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
621-- where traced m = S.'once' (myThreadId >>= print) >> return m
622-- @
623-- @
624-- ThreadId 40
625-- ThreadId 40
626-- ThreadId 40
627-- @
628--
629-- Note that the order of printing in the above examples may change due to
630-- variations in scheduling latencies for concurrent threads.
631--
632-- The polymorphic version of the 'Async' binary operation '<>' is called
633-- 'async'. We can use 'async' to join streams in a left biased
634-- adaptively concurrent manner irrespective of the type, notice that we have
635-- not used the 'asyncly' combinator in the following example:
636--
637-- @
638-- main = 'runStream' $ delay 3 \`async` delay 2 \`async` delay 1
639-- @
640-- @
641-- ThreadId 42: Delay 1
642-- ThreadId 41: Delay 2
643-- ThreadId 40: Delay 3
644-- @
645--
646-- Since the concurrency provided by this operator is demand driven it cannot
647-- be used when the composed computations start timers that are relative to
648-- each other because all computations may not be started at the same time and
649-- therefore timers in all of them may not start at the same time. When
650-- relative timing among all computations is important or when we need to start
651-- all computations at once for any reason 'Parallel' style must be used
652-- instead.
653--
654-- 'Async' style should be preferred over 'Parallel' or 'WAsync' unless you
655-- really need those. It utilizes the resources optimally. It should be used
656-- when we know that the computations can run in parallel but we do not care if
657-- they actually run in parallel or not, that decision can be left to the
658-- scheduler based on demand. Also, note that this operator can be used to fold
659-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
660-- because it does not require us to run all of them at the same time in a fair
661-- manner.
663-- $wasync
664--
665-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
666-- a concurrent manner using /breadth first traversal/. We use the 'wAsyncly'
667-- type combinator to effect 'WAsync' style of composition. We can also use the
668-- 'WAsync' type annotation for the stream to acheive the same effect.
669--
670-- When streams with multiple elements are combined in this manner, we traverse
671-- all the streams concurrently in a breadth first manner i.e. one action from
672-- each stream is peformed and yielded to the resulting stream before we come
673-- back to the first stream again and so on. Even though we execute the actions
674-- in a breadth first order the outputs may be consumed in a different order
675-- because they are consumed on a first come first serve basis.
676--
677-- In the following example we can see that outputs are produced in the breadth
678-- first travresal order but this is not guaranteed.
679--
680-- @
681-- main = 'runStream' . 'wAsyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
682-- @
683-- @
684-- 1
685-- 3
686-- 2
687-- 4
688-- @
689--
690-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
691-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
692-- concurrent traversal irrespective of the type, notice that we have not used
693-- the 'wAsyncly' combinator in the following example:
694--
695-- @
696-- main = 'runStream' $ delay 3 \`wAsync` delay 2 \`wAsync` delay 1
697-- @
698-- @
699-- ThreadId 42: Delay 1
700-- ThreadId 41: Delay 2
701-- ThreadId 40: Delay 3
702-- @
703--
704-- Since the concurrency provided by this style is demand driven it may not
705-- be used when the composed computations start timers that are relative to
706-- each other because all computations may not be started at the same time and
707-- therefore timers in all of them may not start at the same time. When
708-- relative timing among all computations is important or when we need to start
709-- all computations at once for any reason 'Parallel' style must be used
710-- instead.
711--
713-- $parallel
714--
715-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
716-- streams in a fairly concurrent manner with round robin scheduling. We use
717-- the 'parallely' type combinator to effect 'Parallel' style of composition.
718-- We can also use the 'Parallel' type annotation for the stream type to
719-- acheive the same effect.
720--
721-- When two streams with multiple elements are combined in this manner, the
722-- monadic actions in both the streams are performed concurrently with a fair
723-- round robin scheduling. The outputs are yielded in the order in which the
724-- actions complete. This is pretty similar to the 'WAsync' type, the
725-- difference is that 'WAsync' is adaptive to the consumer demand and may or
726-- may not execute all actions in parallel depending on the demand, whereas
727-- 'Parallel' runs all the streams in parallel irrespective of the demand.
728--
729-- The following example sends a query to all the three search engines in
730-- parallel and prints the name of the search engines in the order in which the
731-- responses arrive:
732--
733-- @
734-- import "Streamly"
735-- import qualified Streamly.Prelude as S
736-- import Network.HTTP.Simple
737--
738-- main = 'runStream' . 'parallely' $ google \<> bing \<> duckduckgo
739-- where
740-- google = get "https://www.google.com/search?q=haskell"
741-- bing = get "https://www.bing.com/search?q=haskell"
742-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
743-- get s = S.'once' (httpNoBody (parseRequest_ s) >> putStrLn (show s))
744-- @
745--
746-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
747-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
748-- manner irrespective of the type, notice that we have not used the
749-- 'parallely' combinator in the following example:
750--
751-- @
752-- main = 'runStream' $ delay 3 \`parallel` delay 2 \`wAsync` delay 1
753-- @
754-- @
755-- ThreadId 42: Delay 1
756-- ThreadId 41: Delay 2
757-- ThreadId 40: Delay 3
758-- @
759--
760-- Note that this style of composition cannot be used to combine infinite
761-- number of streams, as it will lead to an infinite sized scheduling queue.
762--
764-- $custom
765--
766-- The 'mkAsync' API can be used to create references to asynchronously running
767-- stream computations. We can then use 'uncons' to explore the streams
768-- arbitrarily and then recompose individual elements to create a new stream.
769-- This way we can dynamically decide which stream to explore at any given
770-- time. Take an example of a merge sort of two sorted streams. We need to
771-- keep consuming items from the stream which has the lowest item in the sort
772-- order. This can be achieved using async references to streams. See
773-- "MergeSort.hs" in the examples directory.
775-- $monoid
776--
777-- We can use 'Monoid' instances to fold a container of streams in the desired
778-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
779-- to fold streams using the polymorphic combine operations:
780--
781-- * 'foldWith' is like 'fold', it folds a 'Foldable' container of streams
782-- using the given composition operator.
783-- * 'foldMapWith' is like 'foldMap', it folds like @foldWith@ but also maps a
784-- function before folding.
785-- * 'forEachWith' is like @foldMapwith@ but the container argument comes before
786-- the function argument.
787--
788-- All of the following are equivalent and start ten concurrent tasks each with
789-- a delay from 1 to 10 seconds, resulting in the printing of each number every
790-- second:
791--
792-- @
793-- import "Streamly"
794-- import qualified "Streamly.Prelude" as S
795-- import Control.Concurrent
796--
797-- main = do
798-- 'runStream' $ 'asyncly' $ foldMap delay [1..10]
799-- 'runStream' $ 'foldWith' 'async' (map delay [1..10])
800-- 'runStream' $ 'foldMapWith' 'async' delay [1..10]
801-- 'runStream' $ 'forEachWith' 'async' [1..10] delay
802-- where delay n = S.'once' $ threadDelay (n * 1000000) >> print n
803-- @
805-- $nesting
806--
807-- Till now we discussed ways to apply transformations on a stream or to merge
808-- streams together to create another stream. We mentioned earlier that
809-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
810-- We will now discuss the concept of a nested composition of streams which is
811-- analogous to nested @for@ loops in the imperative paradigm. Functional
812-- programmers call this style of composition a list transformer or @ListT@.
813-- Logic programmers call it a logic monad or non-deterministic composition,
814-- but for ordinary imperative minded people like me it is easier to think in
815-- terms of good old nested @for@ loops.
816--
817-- $monad
818--
819-- In functional programmer's parlance the 'Monad' instances of different
820-- 'IsStream' types implement non-determinism, exploring all possible
821-- combination of choices from both the streams. From an imperative
822-- programmer's point of view it behaves like nested loops i.e. for each
823-- element in the first stream and for each element in the second stream
824-- execute the body of the loop.
825--
826-- The 'Monad' instances of 'Serial', 'WSerial', 'Async' and 'WAsync'
827-- stream types support different flavors of nested looping. In other words,
828-- they are all variants of list transformer. The nesting behavior of these
829-- types correspond exactly to the way they merge streams as we discussed in
830-- the previous section.
831--
833-- $regularSerial
834--
835-- The 'Monad' composition of the 'Serial' type behaves like a standard list
836-- transformer. This is the default when we do not use an explicit type
837-- combinator. However, the 'serially' type combinator can be used to switch to
838-- this style of composition. We will see how this style of composition works
839-- in the following examples.
840--
841-- Let's start with an example with a simple @for@ loop without any nesting.
842-- For simplicity of illustration we are using streams of pure values in all
843-- the examples. However, the streams could also be made of monadic actions
844-- instead.
845--
846-- @
847-- import "Streamly"
848-- import qualified "Streamly.Prelude" as S
849--
850-- main = 'runStream' $ do
851-- x <- S.'fromFoldable' [3,2,1]
852-- delay x
853-- @
854-- @
855-- ThreadId 30: Delay 3
856-- ThreadId 30: Delay 2
857-- ThreadId 30: Delay 1
858-- @
859--
860-- As we can see, the code after the @fromFoldable@ statement is run three
861-- times, once for each value of @x@ drawn from the stream. All the three
862-- iterations are serial and run in the same thread one after another. In
863-- imperative terms this is equivalent to a @for@ loop with three iterations.
864--
865-- A console echo loop copying standard input to standard output can simply be
866-- written like this:
867--
868-- @
869-- import "Streamly"
870-- import qualified "Streamly.Prelude" as S
871--
872-- main = 'runStream' $ forever $ S.once getLine >>= S.once . putStrLn
873-- @
874--
875-- When multiple streams are composed using this style they nest in a DFS
876-- manner i.e. nested iterations of a loop are executed before we proceed to
877-- the next iteration of the parent loop. This behaves just like nested @for@
878-- loops in imperative programming.
879--
880-- @
881-- import "Streamly"
882-- import qualified "Streamly.Prelude" as S
883--
884-- main = 'runStream' $ do
885-- x <- S.'fromFoldable' [1,2]
886-- y <- S.'fromFoldable' [3,4]
887-- S.'once' $ putStrLn $ show (x, y)
888-- @
889-- @
890-- (1,3)
891-- (1,4)
892-- (2,3)
893-- (2,4)
894-- @
895--
896-- Notice that this is analogous to merging streams of type 'Serial' or merging
897-- streams using 'serial'.
899-- $concurrentNesting
900--
901-- The 'Monad' composition of 'Async' type can perform the iterations of a
902-- loop concurrently. Concurrency is demand driven i.e. more concurrent
903-- iterations are started only if the previous iterations are not able to
904-- produce enough output for the consumer of the output stream. This works
905-- exactly the same way as the merging of two streams 'asyncly' works.
906-- This is the concurrent analogue of 'Serial' style monadic composition.
907--
908-- The 'asyncly' type combinator can be used to switch to this style of
909-- composition. Alternatively, a type annotation can be used to specify the
910-- type of the stream as 'Async'.
911--
912-- @
913-- import "Streamly"
914-- import "Streamly.Prelude"
915--
916-- main = 'runStream' . 'asyncly' $ do
917-- x <- S.'fromFoldable' [3,2,1]
918-- delay x
919-- @
920-- @
921-- ThreadId 40: Delay 1
922-- ThreadId 39: Delay 2
923-- ThreadId 38: Delay 3
924-- @
925--
926-- As we can see the code after the @fromFoldable@ statement is run three
927-- times, once for each value of @x@. All the three iterations are concurrent
928-- and run in different threads. The iteration with least delay finishes first.
929-- When compared to imperative programming, this can be viewed as a @for@ loop
930-- with three concurrent iterations.
931--
932-- Concurrency is demand driven just as in the case of 'async' merging.
933-- When multiple streams are composed using this style, the iterations are
934-- triggered in a depth first manner just like 'Serial' i.e. nested iterations are
935-- executed before we proceed to the next iteration at higher level. However,
936-- unlike 'Serial' more than one iterations may be started concurrently based
937-- on the demand from the consumer of the stream.
938--
939-- @
940-- import "Streamly"
941-- import qualified "Streamly.Prelude" as S
942--
943-- main = 'runStream' . 'asyncly' $ do
944-- x <- S.'fromFoldable' [1,2]
945-- y <- S.'fromFoldable' [3,4]
946-- S.'once' $ putStrLn $ show (x, y)
947-- @
948-- @
949-- (1,3)
950-- (1,4)
951-- (2,3)
952-- (2,4)
953-- @
955-- $interleavedNesting
956--
957-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
958-- outer and inner loops in a nested loop composition. This works exactly the
959-- same way as the merging of two streams in 'wSerially' fashion works. The
960-- 'wSerially' type combinator can be used to switch to this style of
961-- composition. Alternatively, a type annotation can be used to specify the
962-- type of the stream as 'WSerial'.
963--
964-- @
965-- import "Streamly"
966-- import qualified "Streamly.Prelude" as S
967--
968-- main = 'runStream' . 'wSerially' $ do
969-- x <- S.'fromFoldable' [1,2]
970-- y <- S.'fromFoldable' [3,4]
971-- S.once $ putStrLn $ show (x, y)
972-- @
973-- @
974-- (1,3)
975-- (2,3)
976-- (1,4)
977-- (2,4)
978-- @
979--
981-- $wasyncNesting
982--
983-- Just like 'Async' the 'Monad' composition of 'WAsync' runs the iterations of
984-- a loop concurrently. The difference is in the nested loop behavior. The
985-- nested loops in this type are traversed and executed in a breadth first
986-- manner rather than the depth first manner of 'Async' style.
987-- The loop nesting works exactly the same way as the merging of streams
988-- 'wAsyncly' works. The 'wAsyncly' type combinator can be used to switch to
989-- this style of composition. Alternatively, a type annotation can be used to
990-- specify the type of the stream as 'WAsync'.
991--
992-- @
993-- import "Streamly"
994-- import qualified "Streamly.Prelude" as S
995--
996-- main = 'runStream' . 'wAsyncly' $ do
997-- x <- S.'fromFoldable' [1,2]
998-- y <- S.'fromFoldable' [3,4]
999-- S.'once' $ putStrLn $ show (x, y)
1000-- @
1001-- @
1002-- (1,3)
1003-- (2,3)
1004-- (1,4)
1005-- (2,4)
1006-- @
1008-- $parallelNesting
1010-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
1011-- iterations of a loop concurrently. The difference is in the nested loop
1012-- behavior. The streams at each nest level is run fully concurrently
1013-- irrespective of the demand. The loop nesting works exactly the same way as
1014-- the merging of streams 'parallely' works. The 'parallely' type combinator
1015-- can be used to switch to this style of composition. Alternatively, a type
1016-- annotation can be used to specify the type of the stream as 'Parallel'.
1018-- @
1019-- import "Streamly"
1020-- import qualified "Streamly.Prelude" as S
1022-- main = 'runStream' . 'parallely' $ do
1023-- x <- S.'fromFoldable' [3,2,1]
1024-- delay x
1025-- @
1026-- @
1027-- ThreadId 40: Delay 1
1028-- ThreadId 39: Delay 2
1029-- ThreadId 38: Delay 3
1030-- @
1032-- $monadExercise
1034-- Streamly code is usually written in a way that is agnostic of the
1035-- specific monadic composition type. We use a polymorphic type with a
1036-- 'IsStream' type class constraint. When running the stream we can choose the
1037-- specific mode of composition. For example take a look at the following code.
1039-- @
1040-- import "Streamly"
1041-- import qualified "Streamly.Prelude" as S
1043-- composed :: (IsStream t, Monad (t IO)) => t IO ()
1044-- composed = do
1045-- sz <- sizes
1046-- cl <- colors
1047-- sh <- shapes
1048-- S.'once' $ putStrLn $ show (sz, cl, sh)
1050-- where
1052-- sizes = S.'fromFoldable' [1, 2, 3]
1053-- colors = S.'fromFoldable' ["red", "green", "blue"]
1054-- shapes = S.'fromFoldable' ["triangle", "square", "circle"]
1055-- @
1057-- Now we can interpret this in whatever way we want:
1059-- @
1060-- main = 'runStream' . 'serially' $ composed
1061-- main = 'runStream' . 'wSerially' $ composed
1062-- main = 'runStream' . 'asyncly' $ composed
1063-- main = 'runStream' . 'wAsyncly' $ composed
1064-- main = 'runStream' . 'parallely' $ composed
1065-- @
1067-- As an exercise try to figure out the output of this code for each mode of
1068-- composition.
1070-- $functor
1072-- 'fmap' transforms a stream by mapping a function on all elements of the
1073-- stream. 'fmap' behaves in the same way for all stream types, it is always
1074-- serial.
1076-- @
1077-- import "Streamly"
1078-- import qualified "Streamly.Prelude" as S
1080-- main = (S.'toList' $ fmap show $ S.'fromFoldable' [1..10]) >>= print
1081-- @
1083-- Also see the 'mapM' and 'sequence' functions for mapping actions, in the
1084-- "Streamly.Prelude" module.
1086-- $applicative
1088-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
1089-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are
1090-- provided.
1092-- The following example uses the 'Serial' applicative, it runs all iterations
1093-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
1095-- @
1096-- import "Streamly"
1097-- import qualified "Streamly.Prelude" as S
1098-- import Control.Concurrent
1100-- s1 = d 1 <> d 2
1101-- s2 = d 3 <> d 4
1102-- d n = delay n >> return n
1104-- main = (S.'toList' . 'serially' $ (,) \<$> s1 \<*> s2) >>= print
1105-- @
1106-- @
1107-- ThreadId 36: Delay 1
1108-- ThreadId 36: Delay 3
1109-- ThreadId 36: Delay 4
1110-- ThreadId 36: Delay 2
1111-- ThreadId 36: Delay 3
1112-- ThreadId 36: Delay 4
1113-- [(1,3),(1,4),(2,3),(2,4)]
1114-- @
1116-- Similalrly 'WSerial' applicative runs the iterations in an interleaved
1117-- order but since it is serial it takes a total of 17 seconds:
1119-- @
1120-- main = (S.'toList' . 'wSerially' $ (,) \<$> s1 \<*> s2) >>= print
1121-- @
1122-- @
1123-- ThreadId 36: Delay 1
1124-- ThreadId 36: Delay 3
1125-- ThreadId 36: Delay 2
1126-- ThreadId 36: Delay 3
1127-- ThreadId 36: Delay 4
1128-- ThreadId 36: Delay 4
1129-- [(1,3),(2,3),(1,4),(2,4)]
1130-- @
1132-- 'Async' can run the iterations concurrently and therefore takes a total
1133-- of 10 seconds (1 + 2 + 3 + 4):
1135-- @
1136-- main = (S.'toList' . 'asyncly' $ (,) \<$> s1 \<*> s2) >>= print
1137-- @
1138-- @
1139-- ThreadId 34: Delay 1
1140-- ThreadId 36: Delay 2
1141-- ThreadId 35: Delay 3
1142-- ThreadId 36: Delay 3
1143-- ThreadId 35: Delay 4
1144-- ThreadId 36: Delay 4
1145-- [(1,3),(2,3),(1,4),(2,4)]
1146-- @
1148-- Similalrly 'WAsync' as well can run the iterations concurrently and
1149-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
1151-- @
1152-- main = (S.'toList' . 'wAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1153-- @
1154-- @
1155-- ThreadId 34: Delay 1
1156-- ThreadId 36: Delay 2
1157-- ThreadId 35: Delay 3
1158-- ThreadId 36: Delay 3
1159-- ThreadId 35: Delay 4
1160-- ThreadId 36: Delay 4
1161-- [(1,3),(2,3),(1,4),(2,4)]
1162-- @
1164-- $zipping
1166-- Zipping is a special transformation where the corresponding elements of two
1167-- streams are combined together using a zip function producing a new stream of
1168-- outputs. Two different types are provided for serial and concurrent zipping.
1169-- These types provide an applicative instance that can be used to lift
1170-- functions to zip the argument streams.
1171-- Also see the zipping functions in the "Streamly.Prelude" module.
1173-- $serialzip
1175-- The applicative instance of 'ZipSerial' type zips streams serially.
1176-- 'zipSerially' type combinator can be used to switch to serial applicative
1177-- zip composition:
1179-- @
1180-- import "Streamly"
1181-- import qualified "Streamly.Prelude" as S
1182-- import Control.Concurrent
1184-- d n = delay n >> return n
1185-- s1 = 'serially' $ d 1 <> d 2
1186-- s2 = 'serially' $ d 3 <> d 4
1188-- main = (S.'toList' . 'zipSerially' $ (,) \<$> s1 \<*> s2) >>= print
1189-- @
1191-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1192-- everything runs serially:
1194-- @
1195-- ThreadId 29: Delay 1
1196-- ThreadId 29: Delay 3
1197-- ThreadId 29: Delay 2
1198-- ThreadId 29: Delay 4
1199-- [(1,3),(2,4)]
1200-- @
1202-- $parallelzip
1204-- The applicative instance of 'ZipAsync' type zips streams concurrently.
1205-- 'zipAsyncly' type combinator can be used to switch to parallel applicative
1206-- zip composition:
1209-- @
1210-- import "Streamly"
1211-- import qualified "Streamly.Prelude" as S
1212-- import Control.Concurrent
1213-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
1215-- d n = delay n >> return n
1216-- s1 = 'serially' $ d 1 <> d 2
1217-- s2 = 'serially' $ d 3 <> d 4
1219-- main = do
1220-- hSetBuffering stdout LineBuffering
1221-- (S.'toList' . 'zipAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1222-- @
1224-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1225-- are produced concurrently, and 2 and 4 are produced concurrently:
1227-- @
1228-- ThreadId 32: Delay 1
1229-- ThreadId 32: Delay 2
1230-- ThreadId 33: Delay 3
1231-- ThreadId 33: Delay 4
1232-- [(1,3),(2,4)]
1233-- @
1235-- $concurrent
1237-- When writing concurrent programs there are two distinct places where the
1238-- programmer can control the concurrency. First, when /composing/ a stream by
1239-- merging multiple streams we can choose an appropriate sum style operators to
1240-- combine them concurrently or serially. Second, when /processing/ a stream in
1241-- a monadic composition we can choose one of the monad composition types to
1242-- choose the desired type of concurrency.
1244-- In the following example the squares of @x@ and @y@ are computed
1245-- concurrently using the 'async' operation and the square roots of their
1246-- sum are computed serially because of the 'streamly' combinator. We can
1247-- choose different combinators for the monadic processing and the stream
1248-- generation, to control the concurrency. We can also use the 'asyncly'
1249-- combinator instead of explicitly folding with 'async'.
1251-- @
1252-- import "Streamly"
1253-- import qualified "Streamly.Prelude" as S
1254-- import Data.List (sum)
1256-- main = do
1257-- z \<- S.'toList'
1258-- $ 'serially' -- Serial monadic processing (sqrt below)
1259-- $ do
1260-- x2 \<- 'forEachWith' 'async' [1..100] $ -- Concurrent @"for"@ loop
1261-- \\x -> return $ x * x -- body of the loop
1262-- y2 \<- 'forEachWith' 'async' [1..100] $
1263-- \\y -> return $ y * y
1264-- return $ sqrt (x2 + y2)
1265-- print $ sum z
1266-- @
1268-- We can see how this directly maps to the imperative style
1269-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1270-- and operators instead of the ugly pragmas.
1272-- For more concurrent programming examples see,
1273-- "ListDir.hs", "MergeSort.hs" and "SearchQuery.hs" in the examples directory.
1275-- $reactive
1277-- Reactive programming is nothing but concurrent streaming which is what
1278-- streamly is all about. With streamly we can generate streams of events,
1279-- merge streams that are generated concurrently and process events
1280-- concurrently. We can do all this without any knowledge about the specifics
1281-- of the implementation of concurrency. In the following example you will see
1282-- that the code is just regular Haskell code without much streamly APIs used
1283-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1284-- application.
1286-- This application has two independent and concurrent sources of event
1287-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1288-- events that deteriorate the health of the character in the game.
1289-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1290-- health improves and the game continues.
1292-- @
1293-- {-\# LANGUAGE FlexibleContexts #-}
1295-- import "Streamly"
1296-- import Control.Concurrent (threadDelay)
1297-- import Control.Monad (when)
1298-- import Control.Monad.IO.Class (MonadIO(..))
1299-- import Control.Monad.State (MonadState, get, modify, runStateT)
1300-- import Data.Semigroup (cycle1)
1302-- data Event = Harm Int | Heal Int | Quit deriving (Show)
1304-- userAction :: MonadIO m => 'SerialT' m Event
1305-- userAction = cycle1 $ liftIO askUser
1306-- where
1307-- askUser = do
1308-- command <- getLine
1309-- case command of
1310-- "potion" -> return (Heal 10)
1311-- "quit" -> return Quit
1312-- _ -> putStrLn "What?" >> askUser
1314-- acidRain :: MonadIO m => 'SerialT' m Event
1315-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
1317-- game :: ('MonadAsync' m, MonadState Int m) => 'SerialT' m ()
1318-- game = do
1319-- event \<- userAction \`parallel` acidRain
1320-- case event of
1321-- Harm n -> modify $ \\h -> h - n
1322-- Heal n -> modify $ \\h -> h + n
1323-- Quit -> fail "quit"
1325-- h <- get
1326-- when (h <= 0) $ fail "You die!"
1327-- liftIO $ putStrLn $ "Health = " ++ show h
1329-- main = do
1330-- putStrLn "Your health is deteriorating due to acid rain,\\
1331-- \\ type \\"potion\\" or \\"quit\\""
1332-- _ <- runStateT ('runStream' game) 60
1333-- return ()
1334-- @
1336-- You can also find the source of this example in the examples directory as
1337-- "AcidRain.hs". It has been adapted from Gabriel's
1338-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1339-- package.
1340-- This is much simpler compared to the pipes version because of the builtin
1341-- concurrency in streamly. You can also find a SDL based reactive programming
1342-- example adapted from Yampa in "Streamly.Examples.CirclingSquare".
1344-- $performance
1346-- Streamly is highly optimized for performance, it is designed for serious
1347-- high performing, concurrent and scalable applications. We have created the
1348-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1349-- package which is specifically and carefully designed to measure the
1350-- performance of Haskell streaming libraries fairly and squarely in the right
1351-- way. Streamly performs at par or even better than most streaming libraries
1352-- for serial operations even though it needs to deal with the concurrency
1353-- capability.
1355-- $interop
1357-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1359-- Interop with @vector@:
1361-- @
1362-- import Streamly
1363-- import qualified Streamly.Prelude as S
1364-- import qualified Data.Vector.Fusion.Stream.Monadic as V
1366-- -- | vector to streamly
1367-- fromVector :: (IsStream t, Monad m) => V.Stream m a -> t m a
1368-- fromVector = S.unfoldrM unconsV
1369-- where
1370-- unconsV v = do
1371-- r <- V.null v
1372-- if r
1373-- then return Nothing
1374-- else do
1375-- h <- V.head v
1376-- return $ Just (h, V.tail v)
1378-- -- | streamly to vector
1379-- toVector :: Monad m => SerialT m a -> V.Stream m a
1380-- toVector = V.unfoldrM (S.uncons . adapt)
1382-- main = do
1383-- S.toList (fromVector (V.fromList [1..3])) >>= print
1384-- V.toList (toVector (S.fromFoldable [1..3])) >>= print
1385-- @
1387-- Interop with @pipes@:
1389-- @
1390-- import "Streamly"
1391-- import qualified "Streamly.Prelude" as S
1392-- import qualified Pipes as P
1393-- import qualified Pipes.Prelude as P
1395-- -- | pipes to streamly
1396-- fromPipes :: (IsStream t, Monad m) => P.Producer a m r -> t m a
1397-- fromPipes = S.'unfoldrM' unconsP
1398-- where
1399-- -- Adapt P.next to return a Maybe instead of Either
1400-- unconsP p = P.next p >>= either (\\_ -> return Nothing) (return . Just)
1402-- -- | streamly to pipes
1403-- toPipes :: Monad m => SerialT m a -> P.Producer a m ()
1404-- toPipes = P.unfoldr unconsS
1405-- where
1406-- -- Adapt S.uncons to return an Either instead of Maybe
1407-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1409-- main = do
1410-- S.'toList' (fromPipes (P.each [1..3])) >>= print
1411-- P.toListM (toPipes (S.'fromFoldable' [1..3])) >>= print
1412-- @
1414-- Interop with @streaming@:
1416-- @
1417-- import "Streamly"
1418-- import qualified "Streamly.Prelude" as S
1419-- import qualified Streaming as SG
1420-- import qualified Streaming.Prelude as SG
1422-- -- | streaming to streamly
1423-- fromStreaming :: (IsStream t, Monad m) => SG.Stream (SG.Of a) m r -> t m a
1424-- fromStreaming = S.unfoldrM SG.uncons
1426-- -- | streamly to streaming
1427-- toStreaming :: Monad m => SerialT m a -> SG.Stream (SG.Of a) m ()
1428-- toStreaming = SG.unfoldr unconsS
1429-- where
1430-- -- Adapt S.uncons to return an Either instead of Maybe
1431-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1433-- main = do
1434-- S.toList (fromStreaming (SG.each [1..3])) >>= print
1435-- SG.toList (toStreaming (S.fromFoldable [1..3])) >>= print
1436-- @
1438-- Interop with @conduit@:
1440-- @
1441-- import "Streamly"
1442-- import qualified "Streamly.Prelude" as S
1443-- import qualified Data.Conduit as C
1444-- import qualified Data.Conduit.List as C
1445-- import qualified Data.Conduit.Combinators as C
1447-- -- It seems there is no way out of a conduit as it does not provide an
1448-- -- uncons or a tail function. We can convert streamly to conduit though.
1450-- -- | streamly to conduit
1451-- toConduit :: Monad m => SerialT m a -> C.ConduitT i a m ()
1452-- toConduit s = C.unfoldM S.'uncons' s
1454-- main = do
1455-- C.runConduit (toConduit (S.'fromFoldable' [1..3]) C..| C.sinkList) >>= print
1456-- @
1458-- $comparison
1460-- List transformers and logic programming monads also provide a product style
1461-- composition similar to streamly, however streamly generalizes it with the
1462-- time dimension; allowing streams to be composed in an asynchronous and
1463-- concurrent fashion in many different ways. It also provides multiple
1464-- alternative ways of composing streams e.g. serial, interleaved or
1465-- concurrent.
1467-- This seemingly simple addition of asynchronicity and concurrency to product
1468-- style streaming composition unifies a number of disparate abstractions into
1469-- one powerful, concise and elegant abstraction. A wide variety of
1470-- programming problems can be solved elegantly with this abstraction. In
1471-- particular, it unifies three major programming domains namely
1472-- non-deterministic (logic) programming, concurrent programming and functional
1473-- reactive programming. In other words, you can do everything with this one
1474-- abstraction that you could do with the popular libraries listed under these
1475-- categories in the list below.
1477-- @
1478-- +-----------------+----------------+
1479-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1480-- | +----------------+
1481-- | | <https://hackage.haskell.org/package/list-t list-t> |
1482-- | +----------------+
1483-- | | <https://hackage.haskell.org/package/logict logict> |
1484-- +-----------------+----------------+
1485-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1486-- | +----------------+
1487-- | | <https://hackage.haskell.org/package/streaming streaming> |
1488-- | +----------------+
1489-- | | <https://hackage.haskell.org/package/pipes pipes> |
1490-- | +----------------+
1491-- | | <https://hackage.haskell.org/package/conduit conduit> |
1492-- +-----------------+----------------+
1493-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1494-- | +----------------+
1495-- | | <https://hackage.haskell.org/package/transient transient> |
1496-- +-----------------+----------------+
1497-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1498-- | +----------------+
1499-- | | <https://hackage.haskell.org/package/dunai dunai> |
1500-- | +----------------+
1501-- | | <https://hackage.haskell.org/package/reflex reflex> |
1502-- +-----------------+----------------+
1503-- @
1505-- Streamly is a list-transformer. It provides all the functionality provided
1506-- by any of the list transformer and logic programming packages listed above.
1507-- In addition, Streamly naturally integrates the concurrency dimension to the
1508-- basic list transformer functionality.
1510-- When it comes to streaming, in terms of the streaming API streamly is almost
1511-- identical to the vector package. Streamly, vector and streaming packages all
1512-- represent a stream as data and are therefore similar in the fundamental
1513-- approach to streaming. The fundamental difference is that streamly adds
1514-- concurrency support and the monad instance provides concurrent looping.
1515-- Other streaming libraries like pipes, conduit and machines represent and
1516-- compose stream processors rather than the stream data and therefore fall in
1517-- another class of streaming libraries and have comparatively more complicated
1518-- types.
1520-- When it comes to concurrency, streamly can do everything that the @async@
1521-- package can do and more. async provides applicative concurrency whereas
1522-- streamly provides both applicative and monadic concurrency. The
1523-- 'ZipAsync' type behaves like the applicative instance of async. In
1524-- comparison to transient streamly has a first class streaming interface and
1525-- is a monad transformer that can be used universally in any Haskell monad
1526-- transformer stack. Streamly was in fact originally inspired by the
1527-- concurrency implementation in @transient@ though it has no resemblence with
1528-- that and takes a lazy pull approach versus transient's strict push approach.
1530-- The non-determinism, concurrency and streaming combination make streamly a
1531-- strong FRP capable library as well. FRP is fundamentally stream of events
1532-- that can be processed concurrently. The example in this tutorial as well as
1533-- the "Streamly.Examples.CirclingSquare" example from Yampa demonstrate the
1534-- basic FRP capability of streamly. In core concepts streamly is strikingly
1535-- similar to @dunai@. dunai was designed from a FRP perspective and streamly
1536-- was originally designed from a concurrency perspective. However, both have
1537-- similarity at the core.
1539-- $furtherReading
1541-- * Read the documentation of "Streamly" module
1542-- * Read the documentation of "Streamly.Prelude" module
1543-- * See the examples in the "examples" directory of the package
1544-- * See the tests in the "test" directory of the package