1{-# OPTIONS_GHC -fno-warn-unused-imports #-}
3-- Module : Streamly.Tutorial
4-- Copyright : (c) 2017 Composewell Technologies
7-- Maintainer : streamly@composewell.com
9-- Streamly is a general computing framework based on concurrent data flow
10-- programming. The IO monad and pure lists are a special case of streamly. On
11-- one hand, streamly extends the lists of pure values to lists of monadic
12-- actions, on the other hand it extends the IO monad with concurrent
13-- non-determinism. In simple imperative terms we can say that streamly extends
14-- the IO monad with @for@ loops and nested @for@ loops with concurrency
15-- support. Hopefully, this analogy becomes clearer once you go through this
18-- Streaming in general enables writing modular, composable and scalable
19-- applications with ease, and concurrency allows you to make them scale and
20-- perform well. Streamly enables writing scalable concurrent applications
21-- without being aware of threads or synchronization. No explicit thread
22-- control is needed. Where applicable, concurrency rate is automatically
23-- controlled based on the demand by the consumer. However, combinators can be
24-- used to fine tune the concurrency control.
26-- Streaming and concurrency together enable expressing reactive applications
27-- conveniently. See the @CirclingSquare@ example in
28-- <https://github.com/composewell/streamly-examples Streamly Examples> for
29-- a simple SDL based FRP example. To summarize, streamly provides a unified
30-- computing framework for streaming, non-determinism and functional reactive
31-- programming in an elegant and simple API that is a natural extension of pure
32-- lists to monadic streams.
34-- In this tutorial we will go over the basic concepts and how to use the
35-- library. Before you go through this tutorial we recommend that you take a
38-- * The quick overview in the package <README.md README file>.
39-- * The overview of streams and folds in the "Streamly" module.
41-- Once you finish this tutorial, see the last section for further reading
44module Streamly.Tutorial
49 -- * Concurrent Streams
52 -- * Combining Streams
55 -- * Imports and Supporting Code
58 -- * Generating Streams
61 -- * Generating Streams Concurrently
62 -- $generatingConcurrently
64 -- * Eliminating Streams
67 -- * Concurrent Pipeline Stages
68 -- $concurrentApplication
70 -- * Transforming Streams
73 -- * Mapping Concurrently
74 -- $concurrentTransformation
81 -- *** Deep Serial Composition ('Serial')
84 -- *** Wide Serial Composition ('WSerial')
87 -- *** Deep Speculative Composition ('Ahead')
90 -- *** Deep Asynchronous Composition ('Async')
93 -- *** Wide Asynchronous Composition ('WAsync')
96 -- *** Parallel Asynchronous Composition ('Parallel')
99 -- XXX we should deprecate and remove the mkAsync API
100 -- Custom composition
112 -- *** Deep Serial Nesting ('Serial')
115 -- *** Wide Serial Nesting ('WSerial')
116 -- $interleavedNesting
118 -- *** Deep Speculative Nesting ('Ahead')
121 -- *** Deep Asynchronous Nesting ('Async')
122 -- $concurrentNesting
124 -- *** Wide Asynchronous Nesting ('WAsync')
127 -- *** Parallel Asynchronous Nesting ('Parallel')
145 -- ** Parallel Zipping
148 -- * Monad transformers
149 -- $monadtransformers
151 -- * Concurrent Programming
154 -- * Reactive Programming
157 -- * Writing Concurrent Programs
163 -- * Interoperation with Streaming Libraries
166 -- * Comparison with Existing Packages
169 -- * Where to go next?
174import Streamly.Prelude
176import Control.Applicative
178import Control.Monad.IO.Class (MonadIO(..))
179import Control.Monad.Trans.Class (MonadTrans (lift))
183-- The monadic stream API offered by Streamly is very close to the Haskell
184-- "Prelude" pure lists' API, it can be considered as a natural extension of
185-- lists to monadic actions. Streamly streams provide concurrent composition
186-- and merging of streams. It can be considered as a concurrent list
189-- The basic stream type is 'Serial', it represents a sequence of IO actions,
190-- and is a 'Monad'. The 'Serial' monad is almost a drop in replacement for
191-- the 'IO' monad, IO monad is a special case of the 'Serial' monad; IO monad
192-- represents a single IO action whereas the 'Serial' monad represents a series
193-- of IO actions. The only change you need to make to go from 'IO' to 'Serial'
194-- is to use 'drain' to run the monad and to prefix the IO actions with
195-- either 'yieldM' or 'liftIO'. If you use liftIO you can switch from 'Serial'
196-- to IO monad by simply removing the 'drain' function; no other changes
197-- are needed unless you have used some stream specific composition or
200-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
201-- pure lists are a special case of monadic streams. If you use 'nil' in place
202-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' stream.
203-- The only difference is that the elements must be monadic type and to operate
204-- on the streams we must use the corresponding functions from
205-- "Streamly.Prelude" instead of using the base "Prelude".
209-- Many stream operations can be done concurrently:
211-- * Streams can be generated concurrently.
213-- * Streams can be merged concurrently.
215-- * Multiple stages in a streaming pipeline can run concurrently.
217-- * Streams can be mapped and zipped concurrently.
219-- * In monadic composition they combine like a list transformer,
220-- providing concurrent non-determinism.
222-- There are three basic concurrent stream styles, 'Ahead', 'Async', and
223-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that
224-- they can speculatively execute multiple stream actions concurrently in
225-- advance. 'Ahead' would return exactly the same stream as 'Serial' except
226-- that it may execute the actions concurrently. The 'Async' style streams,
227-- like 'Ahead', speculatively execute multiple stream actions in advance but
228-- return the results in their finishing order rather than in the stream
229-- traversal order. 'Parallel' is like 'Async' except that it provides
230-- unbounded parallelism instead of controlled parallelism.
232-- For easy reference, we can classify the stream types based on /execution order/,
233-- /consumption order/, and /bounded or unbounded/ concurrency.
234-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
235-- execution we execute the next action in the stream only after the previous
236-- one has finished executing. In asynchronous execution multiple actions in
237-- the stream can be executed asynchronously i.e. the next action can start
238-- executing even before the first one has finished. Consumption order
239-- determines the order in which the outputs generated by the composition are
240-- consumed. Consumption could be serial or asynchronous. In serial
241-- consumption, the outputs are consumed in the traversal order, in
242-- asynchronous consumption the outputs are consumed as they arrive i.e. first
243-- come first serve order.
245-- +------------+--------------+--------------+--------------+
246-- | Type | Execution | Consumption | Concurrency |
247-- +============+==============+==============+==============+
248-- | 'Serial' | Serial | Serial | None |
249-- +------------+--------------+--------------+--------------+
250-- | 'Ahead' | Asynchronous | Serial | bounded |
251-- +------------+--------------+--------------+--------------+
252-- | 'Async' | Asynchronous | Asynchronous | bounded |
253-- +------------+--------------+--------------+--------------+
254-- | 'Parallel' | Asynchronous | Asynchronous | unbounded |
255-- +------------+--------------+--------------+--------------+
257-- All these types can be freely inter-converted using type conversion
258-- combinators or type annotations, without any cost, to achieve the desired
259-- composition style. To force a particular type of composition, we coerce the
260-- stream type using the corresponding type adapting combinator from
261-- 'serially', 'aheadly', 'asyncly', or 'parallely'. The default stream type
262-- is inferred as 'Serial' unless you change it by using one of the combinators
263-- or by using a type annotation.
267-- Streams can be combined using '<>' or 'mappend' to form a
268-- composite. Composite streams can be interpreted in a depth first or
269-- breadth first manner using an appropriate type conversion before
270-- consumption. Deep (e.g. 'Serial') stream type variants traverse a
271-- composite stream in a depth first manner, such that each stream is
272-- traversed fully before traversing the next stream. Wide
273-- (e.g. 'WSerial') stream types traverse it in a breadth first
274-- manner, such that one element from each stream is traversed before
275-- coming back to the first stream again.
277-- Each stream type has a wide traversal variant prefixed by 'W'. The wide
278-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad
279-- compositions of the streams.
280-- The following table summarizes the basic types and the corresponding wide
284-- +------------+-----------+
286-- +============+===========+
287-- | 'Serial' | 'WSerial' |
288-- +------------+-----------+
289-- | 'Ahead' | 'WAhead' |
290-- +------------+-----------+
291-- | 'Async' | 'WAsync' |
292-- +------------+-----------+
295-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
296-- zip streams serially or concurrently using 'Applicative' operation. These
297-- types are not monads they are only applicatives and they do not differ in
298-- 'Semigroup' composition.
303-- When writing concurrent programs it is advised to not use the concurrent
304-- style stream combinators blindly at the top level. That might create too
305-- much concurrency where it is not even required, and can even degrade
306-- performance in some cases. In some cases it can also lead to surprising
307-- behavior because of some code that is supposed to be serial becoming
308-- concurrent. Please be aware that all concurrency capable APIs that you may
309-- have used under the scope of a concurrent stream combinator will become
310-- concurrent. For example if you have a 'repeatM' somewhere in your program
311-- and you use 'parallely' on top, the 'repeatM' becomes fully parallel,
312-- resulting into an infinite parallel execution . Instead, use the
313-- /Keep It Serial and Stupid/ principle, start with the default serial
314-- composition and enable concurrent combinators only when and where necessary.
315-- When you use a concurrent combinator you can use an explicit 'serially'
316-- combinator to suppress any unnecessary concurrency under the scope of that
321-- To represent streams in an arbitrary monad use the more general monad
322-- transformer types for example the monad transformer type corresponding to
323-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values
324-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a
325-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for
328-- Similarly we have monad transformer types for other stream types as well viz.
329-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
331-- To lift a value from an underlying monad in a monad transformer stack into a
332-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
335-- > S.'drain' $ liftIO $ putStrLn "Hello world!"
337-- > S.'drain' $ lift $ putStrLn "Hello world!"
344-- We will assume the following imports in this tutorial. Go ahead, fire up a
345-- GHCi session and import these lines to start playing.
348-- > import "Streamly.Prelude" ((|:), (|&))
349-- > import qualified "Streamly.Prelude" as S
351-- > import Control.Concurrent
354-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
355-- a monadic action at the head of the stream.
358-- > S.'toList' S.'nil'
360-- > S.'toList' $ 'getLine' |: 'getLine' |: S.'nil'
366-- To create a singleton stream from a pure value use 'yield' or 'pure' and to
367-- create a singleton stream from a monadic action use 'yieldM'. Note that in
368-- case of Zip applicative streams "pure" repeats the value to generate an
372-- > S.'toList' $ 'pure' 1
374-- > S.'toList' $ S.'yield' 1
376-- > S.'toList' $ S.'yieldM' 'getLine'
381-- To create a stream from pure values in a 'Foldable' container use
382-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
385-- > S.'toList' $ S.'fromFoldable' [1..3]
387-- > S.'toList' $ 'Prelude.foldr' S.'cons' S.'nil' [1..3]
391-- To create a stream from monadic actions in a 'Foldable' container just use a
392-- right fold using 'consM' and 'nil':
395-- > S.'drain' $ 'Prelude.foldr' ('|:') S.'nil' ['putStr' "Hello ", 'putStrLn' "world!"]
399-- For more ways to construct a stream see the module "Streamly.Prelude".
401-- $generatingConcurrently
403-- Monadic construction and generation functions like 'consM', 'unfoldrM',
404-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' work concurrently
405-- when used with appropriate stream type combinator. The pure versions of
406-- these APIs are not concurrent, however you can use the monadic versions even
407-- for pure computations by wrapping the pure value in a monad to get the
408-- concurrent generation capability where required.
410-- The following code finishes in 3 seconds (6 seconds when serial):
413-- > let p n = threadDelay (n * 1000000) >> return n
415-- > S.'toList' $ S.'parallely' $ p 3 |: p 2 |: p 1 |: S.'nil'
417-- (3.01 secs, 2,018,432 bytes)
418-- > S.'toList' $ S.'aheadly' $ p 3 |: p 2 |: p 1 |: S.'nil'
420-- (3.01 secs, 2,055,880 bytes)
422-- The following finishes in 10 seconds (100 seconds when serial):
425-- > S.'drain' $ S.'asyncly' $ S.'replicateM' 10 $ p 10
426-- (10.01 secs, 2,080,320 bytes)
432-- We have already seen 'drain' and 'toList' to eliminate a stream in the
433-- examples above. 'drain' runs a stream discarding the results i.e. only
434-- for effects. 'toList' runs the stream and collects the results in a list.
436-- For other ways to eliminate a stream see the @Folding@ section in
437-- "Streamly.Prelude" module.
441-- Transformation over a stream is the equivalent of a @for@ loop construct in
442-- imperative paradigm. We iterate over every element in the stream and perform
443-- certain transformations for each element. Transformations may involve
444-- mapping functions over the elements, filtering elements from the stream or
445-- folding all the elements in the stream into a single value. Streamly streams
446-- are exactly like lists and you can perform all the transformations in the
447-- same way as you would on lists.
449-- Here is a simple console echo program that just echoes every input line,
453-- > import Data.Function ((&))
454-- > S.'drain' $ S.'repeatM' getLine & S.'mapM' putStrLn
457-- The following code snippet reads lines from standard input, filters blank
458-- lines, drops the first non-blank line, takes the next two, up cases them,
459-- numbers them and prints them:
462-- import qualified "Streamly.Prelude" as S
463-- import Data.Char (toUpper)
464-- import Data.Function ((&))
467-- S.'repeatM' getLine
468-- & S.'filter' (not . null)
471-- & fmap (map toUpper)
472-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
473-- & S.'mapM' putStrLn
476-- $concurrentTransformation
478-- Monadic transformation functions 'mapM' and 'sequence' work concurrently
479-- when used with appropriate stream type combinators. The pure versions do not
480-- work concurrently, however you can use the monadic versions even for pure
481-- computations to get the concurrent transformation capability where required.
483-- This would print a value every second (2 seconds when serial):
486-- > let p n = threadDelay (n * 1000000) >> return n
487-- > S.'drain' $ S.'aheadly' $ S.'mapM' (\\x -> p 1 >> print x) (S.'serially' $ S.'repeatM' (p 1))
491-- $concurrentApplication
493-- The concurrent function application operators '|$' and '|&' apply a stream
494-- argument to a stream function concurrently to compose a concurrent pipeline
495-- of stream processing functions:
497-- Because both the stages run concurrently, we would see a delay of only 1
498-- second instead of 2 seconds in the following:
501-- > let p n = threadDelay (n * 1000000) >> return n
502-- > S.'drain' $ S.'repeatM' (p 1) '|&' S.'mapM' (\\x -> p 1 >> print x)
507-- We can combine two streams into a single stream using semigroup composition
508-- operation '<>'. Streams can be combined in many different ways as described
509-- in the following sections, the '<>' operation behaves differently depending
510-- on the stream type in effect. The stream type and therefore the composition
511-- style can be changed at any point using one of the type combinators as
516-- In most of example snippets we do not repeat the imports. Where imports are
517-- not explicitly specified use the imports shown below.
520-- import "Streamly.Prelude" ((|:), nil)
521-- import qualified "Streamly.Prelude" as S
523-- import Control.Concurrent
524-- import Control.Monad (forever)
527-- To illustrate concurrent vs serial composition aspects, we will use the
528-- following @delay@ function to introduce a sleep or delay specified in
529-- seconds. After the delay it prints the number of seconds it slept.
532-- delay n = S.'yieldM' $ do
533-- threadDelay (n * 1000000)
535-- putStrLn (show tid ++ ": Delay " ++ show n)
540-- The 'Semigroup' operation '<>' of the 'Serial' type combines the two streams
541-- in a /serial depth first/ manner. We use the 'serially' type combinator to
542-- effect 'Serial' style of composition. We can also use an explicit 'Serial'
543-- type annotation for the stream to achieve the same effect. However, since
544-- 'Serial' is the default type unless explicitly specified by using a
545-- combinator, we can omit using an explicit combinator or type annotation for
546-- this style of composition.
548-- When two streams with multiple elements are combined in this manner, the
549-- monadic actions in the two streams are performed sequentially i.e. first all
550-- actions in the first stream are performed sequentially and then all actions
551-- in the second stream are performed sequentially. We call it
552-- /serial depth first/ as the full depth of one stream is fully traversed
553-- before we move to the next. The following example prints the sequence 1, 2,
557-- main = S.'drain' $ (print 1 |: print 2 |: S.nil) <> (print 3 |: print 4 |: S.nil)
566-- All actions in both the streams are performed serially in the same thread.
567-- In the following example we can see that all actions are performed in the
568-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
571-- main = S.'drain' $ delay 3 <> delay 2 <> delay 1
574-- ThreadId 36: Delay 3
575-- ThreadId 36: Delay 2
576-- ThreadId 36: Delay 1
579-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
580-- 'serial'. We can use 'serial' to join streams in a sequential manner
581-- irrespective of the type of stream:
584-- main = S.'drain' $ (print 1 |: print 2 |: S.nil) \`S.serial` (print 3 |: print 4 |: S.nil)
589-- The 'Semigroup' operation '<>' of the 'WSerial' type combines the two
590-- streams in a /serial breadth first/ manner. We use the 'wSerially' type
591-- combinator to effect 'WSerial' style of composition. We can also use the
592-- 'WSerial' type annotation for the stream to achieve the same effect.
594-- When two streams with multiple elements are combined in this manner, we
595-- traverse all the streams in a breadth first manner i.e. one action from each
596-- stream is performed and yielded to the resulting stream before we come back
597-- to the first stream again and so on.
598-- The following example prints the sequence 1, 3, 2, 4
601-- main = S.'drain' . S.'wSerially' $ (print 1 |: print 2 |: S.'nil') <> (print 3 |: print 4 |: S.'nil')
610-- Even though the monadic actions of the two streams are performed in an
611-- interleaved manner they are all performed serially in the same thread. In
612-- the following example we can see that all actions are performed in the same
613-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
616-- main = S.'drain' . S.'wSerially' $ delay 3 <> delay 2 <> delay 1
619-- ThreadId 36: Delay 3
620-- ThreadId 36: Delay 2
621-- ThreadId 36: Delay 1
624-- The polymorphic version of the 'WSerial' binary operation '<>' is called
625-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
626-- irrespective of the type, notice that we have not used the 'wSerially'
627-- combinator in the following example:
630-- main = S.'drain' $ (print 1 |: print 2 |: S.'nil') \`S.wSerial` (print 3 |: print 4 |: S.'nil')
639-- Note that this composition cannot be used to fold infinite number of streams
640-- since it requires preserving the state until a stream is finished.
644-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a
645-- /serial depth first/ manner with concurrent lookahead. We use the 'aheadly'
646-- type combinator to effect 'Ahead' style of composition. We can also use an
647-- explicit 'Ahead' type annotation for the stream to achieve the same effect.
649-- When two streams are combined in this manner, the streams are traversed in
650-- depth first manner just like 'Serial', however it can execute the next
651-- stream concurrently and keep the results ready when its turn arrives.
652-- Concurrent execution of the next stream(s) is performed if the first stream
653-- blocks or if it cannot produce output at the rate that is enough to meet the
654-- consumer demand. Multiple streams can be executed concurrently to meet the
655-- demand. The following example would print the result in a second even
656-- though each action in each stream takes one second:
660-- xs \<- S.'toList' . S.'aheadly' $ (p 1 |: p 2 |: S.nil) <> (p 3 |: p 4 |: S.nil)
662-- where p n = threadDelay 1000000 >> return n
668-- Each stream is constructed 'aheadly' and then both the streams are merged
669-- 'aheadly', therefore, all the actions can run concurrently but the result is
670-- presented in serial order.
672-- You can also use the polymorphic combinator 'ahead' in place of '<>' to
673-- compose any type of streams in this manner.
677-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
678-- streams in a depth first manner with parallel look ahead. We use the
679-- 'asyncly' type combinator to effect 'Async' style of composition. We
680-- can also use the 'Async' type annotation for the stream type to achieve
683-- When two streams with multiple elements are combined in this manner, the
684-- streams are traversed in depth first manner just like 'Serial', however it
685-- can execute the next stream concurrently and return the results from it
686-- as they arrive i.e. the results from the next stream may be yielded even
687-- before the results from the first stream. Concurrent execution of the next
688-- stream(s) is performed if the first stream blocks or if it cannot produce
689-- output at the rate that is enough to meet the consumer demand. Multiple
690-- streams can be executed concurrently to meet the demand.
691-- In the example below each element in the stream introduces a constant delay
692-- of 1 second, however, it takes just one second to produce all the results.
693-- The results are not guaranteed to be in any particular order:
697-- xs \<- S.'toList' . S.'asyncly' $ (p 1 |: p 2 |: S.'nil') <> (p 3 |: p 4 |: S.'nil')
699-- where p n = threadDelay 1000000 >> return n
705-- The constituent streams are also composed in 'Async' manner and the
706-- composition of streams too. We can compose the constituent streams to run
707-- serially, in that case it would take 2 seconds to produce all the results.
708-- The elements in the serial streams would be in serial order in the results:
712-- xs \<- S.'toList' . S.'asyncly' $ (S.'serially' $ p 1 |: p 2 |: S.'nil') <> (serially $ p 3 |: p 4 |: S.'nil')
714-- where p n = threadDelay 1000000 >> return n
720-- In the following example we can see that new threads are started when a
721-- computation blocks. Notice that the output from the stream with the
722-- shortest delay is printed first. The whole computation takes @maximum of
723-- (3, 2, 1) = 3@ seconds:
726-- main = S.'drain' . S.'asyncly' $ delay 3 '<>' delay 2 '<>' delay 1
729-- ThreadId 42: Delay 1
730-- ThreadId 41: Delay 2
731-- ThreadId 40: Delay 3
734-- When we have a tree of computations composed using this style, the tree is
735-- traversed in DFS style just like the 'Serial' style, the only difference is
736-- that here we can move on to executing the next stream if a stream blocks.
737-- However, we will not start new threads if we have sufficient output to
738-- saturate the consumer. This is why we call it left-biased demand driven or
739-- adaptive concurrency style, the concurrency tends to stay on the left side
740-- of the composition as long as possible. More threads are started based on
741-- the pull rate of the consumer. The following example prints an output every
742-- second as all of the actions are concurrent.
745-- main = S.'drain' . S.'asyncly' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
754-- All the computations may even run in a single thread when more threads are
755-- not needed. As you can see, in the following example the computations are
756-- run in a single thread one after another, because none of them blocks.
757-- However, if the thread consuming the stream were faster than the producer
758-- then it would have started parallel threads for each computation to keep up
759-- even if none of them blocks:
762-- main = S.'drain' . S.'asyncly' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
763-- where traced m = S.'yieldM' (myThreadId >>= print) >> return m
771-- Note that the order of printing in the above examples may change due to
772-- variations in scheduling latencies for concurrent threads.
774-- The polymorphic version of the 'Async' binary operation '<>' is called
775-- 'async'. We can use 'async' to join streams in a left biased
776-- adaptively concurrent manner irrespective of the type, notice that we have
777-- not used the 'asyncly' combinator in the following example:
780-- main = S.'drain' $ delay 3 \`S.'async'` delay 2 \`S.'async'` delay 1
783-- ThreadId 42: Delay 1
784-- ThreadId 41: Delay 2
785-- ThreadId 40: Delay 3
788-- Since the concurrency provided by this operator is demand driven it cannot
789-- be used when the composed computations start timers that are relative to
790-- each other because all computations may not be started at the same time and
791-- therefore timers in all of them may not start at the same time. When
792-- relative timing among all computations is important or when we need to start
793-- all computations at once for any reason 'Parallel' style must be used
796-- 'Async' style utilizes resources optimally and should be preferred over
797-- 'Parallel' or 'WAsync' unless you really need those. 'Async' should be used
798-- when we know that the computations can run in parallel but we do not care if
799-- they actually run in parallel or not, that decision can be left to the
800-- scheduler based on demand. Also, note that 'async' operator can be used to fold
801-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
802-- because it does not require us to run all of them at the same time in a fair
807-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
808-- a concurrent manner using /breadth first traversal/. We use the 'wAsyncly'
809-- type combinator to effect 'WAsync' style of composition. We can also use the
810-- 'WAsync' type annotation for the stream to achieve the same effect.
812-- When streams with multiple elements are combined in this manner, we traverse
813-- all the streams concurrently in a breadth first manner i.e. one action from
814-- each stream is performed and yielded to the resulting stream before we come
815-- back to the first stream again and so on. Even though we execute the actions
816-- in a breadth first order the outputs are consumed on a first come first
819-- In the following example we can see that outputs are produced in the breadth
820-- first traversal order but this is not guaranteed.
823-- main = S.'drain' . S.'wAsyncly' $ (S.'serially' $ print 1 |: print 2 |: S.'nil') <> (S.'serially' $ print 3 |: print 4 |: S.'nil')
832-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
833-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
834-- concurrent traversal irrespective of the type, notice that we have not used
835-- the 'wAsyncly' combinator in the following example:
838-- main = S.'drain' $ delay 3 \S.`wAsync` delay 2 \S.`wAsync` delay 1
841-- ThreadId 42: Delay 1
842-- ThreadId 41: Delay 2
843-- ThreadId 40: Delay 3
846-- Since the concurrency provided by this style is demand driven it may not
847-- be used when the composed computations start timers that are relative to
848-- each other because all computations may not be started at the same time and
849-- therefore timers in all of them may not start at the same time. When
850-- relative timing among all computations is important or when we need to start
851-- all computations at once for any reason 'Parallel' style must be used
857-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
858-- streams in a fairly concurrent manner with round robin scheduling. We use
859-- the 'parallely' type combinator to effect 'Parallel' style of composition.
860-- We can also use the 'Parallel' type annotation for the stream type to
861-- achieve the same effect.
863-- When two streams with multiple elements are combined in this manner, the
864-- monadic actions in both the streams are performed concurrently with a fair
865-- round robin scheduling. The outputs are yielded in the order in which the
866-- actions complete. This is pretty similar to the 'WAsync' type, the
867-- difference is that 'WAsync' is adaptive to the consumer demand and may or
868-- may not execute all actions in parallel depending on the demand, whereas
869-- 'Parallel' runs all the streams in parallel irrespective of the demand.
871-- The following example sends a query to all the three search engines in
872-- parallel and prints the name of the search engines in the order in which the
873-- responses arrive. You need the [http-conduit](http://hackage.haskell.org/package/http-conduit)
874-- package to run this example:
877-- import qualified Streamly.Prelude as S
878-- import Network.HTTP.Simple
880-- main = S.'drain' . S.'parallely' $ google \<> bing \<> duckduckgo
882-- google = get "https://www.google.com/search?q=haskell"
883-- bing = get "https://www.bing.com/search?q=haskell"
884-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
885-- get s = S.'yieldM' (httpNoBody (parseRequest_ s) >> putStrLn (show s))
888-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
889-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
890-- manner irrespective of the type, notice that we have not used the
891-- 'parallely' combinator in the following example:
894-- main = S.'drain' $ delay 3 \`S.'parallel'` delay 2 \`S.'wAsync'` delay 1
897-- ThreadId 42: Delay 1
898-- ThreadId 41: Delay 2
899-- ThreadId 40: Delay 3
902-- Note that this style of composition cannot be used to combine infinite
903-- number of streams, as it will lead to an infinite sized scheduling queue.
909-- The 'mkAsync' API can be used to create references to asynchronously running
910-- stream computations. We can then use 'uncons' to explore the streams
911-- arbitrarily and then recompose individual elements to create a new stream.
912-- This way we can dynamically decide which stream to explore at any given
913-- time. Take an example of a merge sort of two sorted streams. We need to
914-- keep consuming items from the stream which has the lowest item in the sort
915-- order. This can be achieved using async references to streams. See
916-- "MergeSort.hs" in <https://github.com/composewell/streamly-examples Streamly Examples>.
920-- We can use 'Monoid' instances to fold a container of streams in the desired
921-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
922-- to fold streams using the polymorphic combine operations:
924-- * 'concatFoldableWith' is like 'fold', it folds a 'Foldable' container of
925-- streams using the given composition operator.
926-- * 'concatMapFoldableWith' is like 'foldMap', it folds like
927-- @concatFoldableWith@ but also maps a function before folding.
928-- * 'concatForFoldableWith' is like @concatMapFoldableWith@ but the container
929-- argument comes before the function argument.
931-- All of the following are equivalent and start ten concurrent tasks each with
932-- a delay from 1 to 10 seconds, resulting in the printing of each number every
936-- import qualified "Streamly.Prelude" as S
937-- import Control.Concurrent
940-- S.'drain' $ S.'asyncly' $ foldMap delay [1..10]
941-- S.'drain' $ S.'concatFoldableWith' S.'async' (map delay [1..10])
942-- S.'drain' $ S.'concatMapFoldableWith' S.'async' delay [1..10]
943-- S.'drain' $ S.'concatForFoldableWith' S.'async' [1..10] delay
944-- where delay n = S.'yieldM' $ threadDelay (n * 1000000) >> print n
949-- Till now we discussed ways to apply transformations on a stream or to merge
950-- streams together to create another stream. We mentioned earlier that
951-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
952-- We will now discuss the concept of a nested composition of streams which is
953-- analogous to nested @for@ loops in the imperative paradigm. Functional
954-- programmers call this style of composition a list transformer or @ListT@.
955-- Logic programmers call it a logic monad or non-deterministic composition,
956-- but for ordinary imperative minded people like me it is easier to think in
957-- terms of good old nested @for@ loops.
961-- In functional programmer's parlance the 'Monad' instances of different
962-- 'IsStream' types implement non-determinism, exploring all possible
963-- combination of choices from both the streams. From an imperative
964-- programmer's point of view it behaves like nested loops i.e. for each
965-- element in the first stream and for each element in the second stream
966-- execute the body of the loop.
968-- The 'Monad' instances of 'Serial', 'WSerial', 'Async' and 'WAsync'
969-- stream types support different flavors of nested looping. In other words,
970-- they are all variants of list transformer. The nesting behavior of these
971-- types correspond exactly to the way they merge streams as we discussed in
972-- the previous section.
977-- The 'Monad' composition of the 'Serial' type behaves like a standard list
978-- transformer. This is the default when we do not use an explicit type
979-- combinator. However, the 'serially' type combinator can be used to switch to
980-- this style of composition. We will see how this style of composition works
981-- in the following examples.
983-- Let's start with an example with a simple @for@ loop without any nesting.
984-- For simplicity of illustration we are using streams of pure values in all
985-- the examples. However, the streams could also be made of monadic actions
989-- import qualified "Streamly.Prelude" as S
991-- main = S.'drain' $ do
992-- x <- S.'fromFoldable' [3,2,1]
996-- ThreadId 30: Delay 3
997-- ThreadId 30: Delay 2
998-- ThreadId 30: Delay 1
1001-- As we can see, the code after the @fromFoldable@ statement is run three
1002-- times, once for each value of @x@ drawn from the stream. All the three
1003-- iterations are serial and run in the same thread one after another. In
1004-- imperative terms this is equivalent to a @for@ loop with three iterations.
1006-- A console echo loop copying standard input to standard output can simply be
1007-- written like this:
1010-- import qualified "Streamly.Prelude" as S
1012-- import Control.Monad (forever)
1014-- main = S.'drain' $ forever $ S.'yieldM' getLine >>= S.'yieldM' . putStrLn
1017-- When multiple streams are composed using this style they nest in a DFS
1018-- manner i.e. nested iterations of a loop are executed before we proceed to
1019-- the next iteration of the parent loop. This behaves just like nested @for@
1020-- loops in imperative programming.
1023-- import qualified "Streamly.Prelude" as S
1025-- main = S.'drain' $ do
1026-- x <- S.'fromFoldable' [1,2]
1027-- y <- S.'fromFoldable' [3,4]
1028-- S.'yieldM' $ putStrLn $ show (x, y)
1037-- Notice that this is analogous to merging streams of type 'Serial' or merging
1038-- streams using 'serial'.
1042-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except
1043-- that it can speculatively perform a bounded number of next iterations of a
1044-- loop concurrently.
1046-- The 'aheadly' type combinator can be used to switch to this style of
1047-- composition. Alternatively, a type annotation can be used to specify the
1048-- type of the stream as 'Ahead'.
1051-- import qualified "Streamly.Prelude" as S
1053-- comp = S.'toList' . S.'aheadly' $ do
1054-- x <- S.'fromFoldable' [3,2,1]
1055-- delay x >> return x
1057-- main = comp >>= print
1060-- ThreadId 40: Delay 1
1061-- ThreadId 39: Delay 2
1062-- ThreadId 38: Delay 3
1066-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can
1067-- see all the three iterations are concurrent and run in different threads,
1068-- however, the results are returned in the serial order.
1070-- Concurrency is demand driven, when multiple streams are composed using this
1071-- style, the iterations are executed in a depth first manner just like
1072-- 'Serial' i.e. nested iterations are executed before we proceed to the next
1073-- outer iteration. The only difference is that we may execute multiple future
1074-- iterations concurrently and keep the results ready.
1077-- $concurrentNesting
1079-- The 'Monad' composition of 'Async' type can perform the iterations of a
1080-- loop concurrently. Concurrency is demand driven i.e. more concurrent
1081-- iterations are started only if the previous iterations are not able to
1082-- produce enough output for the consumer of the output stream. This works
1083-- exactly the same way as the merging of two streams 'asyncly' works.
1084-- This is the concurrent analogue of 'Serial' style monadic composition.
1086-- The 'asyncly' type combinator can be used to switch to this style of
1087-- composition. Alternatively, a type annotation can be used to specify the
1088-- type of the stream as 'Async'.
1091-- import qualified "Streamly.Prelude" as S
1093-- main = S.'drain' . S.'asyncly' $ do
1094-- x <- S.'fromFoldable' [3,2,1]
1098-- ThreadId 40: Delay 1
1099-- ThreadId 39: Delay 2
1100-- ThreadId 38: Delay 3
1103-- As we can see the code after the @fromFoldable@ statement is run three
1104-- times, once for each value of @x@. All the three iterations are concurrent
1105-- and run in different threads. The iteration with least delay finishes first.
1106-- When compared to imperative programming, this can be viewed as a @for@ loop
1107-- with three concurrent iterations.
1109-- Concurrency is demand driven just as in the case of 'async' merging.
1110-- When multiple streams are composed using this style, the iterations are
1111-- triggered in a depth first manner just like 'Serial' i.e. nested iterations are
1112-- executed before we proceed to the next iteration at higher level. However,
1113-- unlike 'Serial' more than one iterations may be started concurrently based
1114-- on the demand from the consumer of the stream.
1117-- import qualified "Streamly.Prelude" as S
1119-- main = S.'drain' . S.'asyncly' $ do
1120-- x <- S.'fromFoldable' [1,2]
1121-- y <- S.'fromFoldable' [3,4]
1122-- S.'yieldM' $ putStrLn $ show (x, y)
1131-- $interleavedNesting
1133-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
1134-- outer and inner loops in a nested loop composition. This works exactly the
1135-- same way as the merging of two streams in 'wSerially' fashion works. The
1136-- 'wSerially' type combinator can be used to switch to this style of
1137-- composition. Alternatively, a type annotation can be used to specify the
1138-- type of the stream as 'WSerial'.
1141-- import qualified "Streamly.Prelude" as S
1143-- main = S.'drain' . S.'wSerially' $ do
1144-- x <- S.'fromFoldable' [1,2]
1145-- y <- S.'fromFoldable' [3,4]
1146-- S.yieldM $ putStrLn $ show (x, y)
1158-- Just like 'Async' the 'Monad' composition of 'WAsync' runs the iterations of
1159-- a loop concurrently. The difference is in the nested loop behavior. The
1160-- nested loops in this type are traversed and executed in a breadth first
1161-- manner rather than the depth first manner of 'Async' style.
1162-- The loop nesting works exactly the same way as the merging of streams
1163-- 'wAsyncly' works. The 'wAsyncly' type combinator can be used to switch to
1164-- this style of composition. Alternatively, a type annotation can be used to
1165-- specify the type of the stream as 'WAsync'.
1168-- import qualified "Streamly.Prelude" as S
1170-- main = S.'drain' . S.'wAsyncly' $ do
1171-- x <- S.'fromFoldable' [1,2]
1172-- y <- S.'fromFoldable' [3,4]
1173-- S.'yieldM' $ putStrLn $ show (x, y)
1184-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
1185-- iterations of a loop concurrently. The difference is in the nested loop
1186-- behavior. The streams at each nest level is run fully concurrently
1187-- irrespective of the demand. The loop nesting works exactly the same way as
1188-- the merging of streams 'parallely' works. The 'parallely' type combinator
1189-- can be used to switch to this style of composition. Alternatively, a type
1190-- annotation can be used to specify the type of the stream as 'Parallel'.
1193-- import qualified "Streamly.Prelude" as S
1195-- main = S.'drain' . S.'parallely' $ do
1196-- x <- S.'fromFoldable' [3,2,1]
1200-- ThreadId 40: Delay 1
1201-- ThreadId 39: Delay 2
1202-- ThreadId 38: Delay 3
1207-- Streamly code is usually written in a way that is agnostic of the
1208-- specific monadic composition type. We use a polymorphic type with a
1209-- 'IsStream' type class constraint. When running the stream we can choose the
1210-- specific mode of composition. For example take a look at the following code.
1213-- import qualified "Streamly.Prelude" as S
1215-- composed :: (S.IsStream t, Monad (t IO)) => t IO ()
1220-- S.'yieldM' $ putStrLn $ show (sz, cl, sh)
1224-- sizes = S.'fromFoldable' [1, 2, 3]
1225-- colors = S.'fromFoldable' ["red", "green", "blue"]
1226-- shapes = S.'fromFoldable' ["triangle", "square", "circle"]
1229-- Now we can interpret this in whatever way we want:
1232-- main = S.'drain' . S.'serially' $ composed
1233-- main = S.'drain' . S.'wSerially' $ composed
1234-- main = S.'drain' . S.'asyncly' $ composed
1235-- main = S.'drain' . S.'wAsyncly' $ composed
1236-- main = S.'drain' . S.'parallely' $ composed
1239-- As an exercise try to figure out the output of this code for each mode of
1244-- 'fmap' transforms a stream by mapping a function on all elements of the
1245-- stream. 'fmap' behaves in the same way for all stream types, it is always
1249-- import qualified "Streamly.Prelude" as S
1251-- main = (S.'toList' $ fmap show $ S.'fromFoldable' [1..10]) >>= print
1254-- Also see functions 'mapM' and 'sequence' from "Streamly.Prelude" module
1255-- which can map actions concurrently depending on the type of the input stream.
1259-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
1260-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are
1263-- The following example uses the 'Serial' applicative, it runs all iterations
1264-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
1267-- import qualified "Streamly.Prelude" as S
1268-- import Control.Concurrent
1272-- d n = delay n >> return n
1274-- main = (S.'toList' . S.'serially' $ (,) \<$> s1 \<*> s2) >>= print
1277-- ThreadId 36: Delay 1
1278-- ThreadId 36: Delay 3
1279-- ThreadId 36: Delay 4
1280-- ThreadId 36: Delay 2
1281-- ThreadId 36: Delay 3
1282-- ThreadId 36: Delay 4
1283-- [(1,3),(1,4),(2,3),(2,4)]
1286-- Similarly 'WSerial' applicative runs the iterations in an interleaved
1287-- order but since it is serial it takes a total of 17 seconds:
1290-- main = (S.'toList' . S.'wSerially' $ (,) \<$> s1 \<*> s2) >>= print
1293-- ThreadId 36: Delay 1
1294-- ThreadId 36: Delay 3
1295-- ThreadId 36: Delay 2
1296-- ThreadId 36: Delay 3
1297-- ThreadId 36: Delay 4
1298-- ThreadId 36: Delay 4
1299-- [(1,3),(2,3),(1,4),(2,4)]
1302-- 'Async' can run the iterations concurrently and therefore takes a total
1303-- of 6 seconds which is max (1, 2) + max (3, 4):
1306-- main = (S.'toList' . S.'asyncly' $ (,) \<$> s1 \<*> s2) >>= print
1309-- ThreadId 34: Delay 1
1310-- ThreadId 36: Delay 2
1311-- ThreadId 35: Delay 3
1312-- ThreadId 36: Delay 3
1313-- ThreadId 35: Delay 4
1314-- ThreadId 36: Delay 4
1315-- [(1,3),(2,3),(1,4),(2,4)]
1318-- Similarly 'WAsync' as well can run the iterations concurrently and
1319-- therefore takes a total of 6 seconds (2 + 4):
1322-- main = (S.'toList' . S.'wAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1325-- ThreadId 34: Delay 1
1326-- ThreadId 36: Delay 2
1327-- ThreadId 35: Delay 3
1328-- ThreadId 36: Delay 3
1329-- ThreadId 35: Delay 4
1330-- ThreadId 36: Delay 4
1331-- [(1,3),(2,3),(1,4),(2,4)]
1336-- Zipping is a special transformation where the corresponding elements of two
1337-- streams are combined together using a zip function producing a new stream of
1338-- outputs. Two different types are provided for serial and concurrent zipping.
1339-- These types provide an applicative instance that can be used to lift
1340-- functions to zip the argument streams.
1341-- Also see the zipping functions in the "Streamly.Prelude" module.
1345-- The applicative instance of 'ZipSerial' type zips streams serially.
1346-- 'zipSerially' type combinator can be used to switch to serial applicative
1350-- import qualified "Streamly.Prelude" as S
1351-- import Control.Concurrent
1353-- d n = delay n >> return n
1354-- s1 = S.'serially' $ d 1 <> d 2
1355-- s2 = S.'serially' $ d 3 <> d 4
1357-- main = (S.'toList' . S.'zipSerially' $ (,) \<$> s1 \<*> s2) >>= print
1360-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1361-- everything runs serially:
1364-- ThreadId 29: Delay 1
1365-- ThreadId 29: Delay 3
1366-- ThreadId 29: Delay 2
1367-- ThreadId 29: Delay 4
1373-- The applicative instance of 'ZipAsync' type zips streams concurrently.
1374-- 'zipAsyncly' type combinator can be used to switch to parallel applicative
1379-- import qualified "Streamly.Prelude" as S
1380-- import Control.Concurrent
1381-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
1383-- d n = delay n >> return n
1384-- s1 = S.'serially' $ d 1 <> d 2
1385-- s2 = S.'serially' $ d 3 <> d 4
1388-- hSetBuffering stdout LineBuffering
1389-- (S.'toList' . S.'zipAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1392-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1393-- are produced concurrently, and 2 and 4 are produced concurrently:
1396-- ThreadId 32: Delay 1
1397-- ThreadId 32: Delay 2
1398-- ThreadId 33: Delay 3
1399-- ThreadId 33: Delay 4
1405-- When writing concurrent programs there are two distinct places where the
1406-- programmer can control the concurrency. First, when /composing/ a stream by
1407-- merging multiple streams we can choose an appropriate sum style operators to
1408-- combine them concurrently or serially. Second, when /processing/ a stream in
1409-- a monadic composition we can choose one of the monad composition types to
1410-- choose the desired type of concurrency.
1412-- In the following example the squares of @x@ and @y@ are computed
1413-- concurrently using the 'async' operation and the square roots of their
1414-- sum are computed serially because of the 'streamly' combinator. We can
1415-- choose different combinators for the monadic processing and the stream
1416-- generation, to control the concurrency. We can also use the 'asyncly'
1417-- combinator instead of explicitly folding with 'async'.
1420-- import qualified "Streamly.Prelude" as S
1421-- import Data.List (sum)
1425-- $ S.'serially' -- Serial monadic processing (sqrt below)
1427-- x2 \<- S.'concatForFoldableWith' S.'async' [1..100] $ -- Concurrent @"for"@ loop
1428-- \\x -> return $ x * x -- body of the loop
1429-- y2 \<- S.'concatForFoldableWith' S.'async' [1..100] $
1430-- \\y -> return $ y * y
1431-- return $ sqrt (x2 + y2)
1435-- We can see how this directly maps to the imperative style
1436-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1437-- and operators instead of the ugly pragmas.
1439-- For more concurrent programming examples see,
1440-- <https://github.com/composewell/streamly-examples/tree/master/ListDir.hs ListDir.hs>,
1441-- <https://github.com/composewell/streamly-examples/tree/master/MergeSort.hs MergeSort.hs> and
1442-- <https://github.com/composewell/streamly-examples/tree/master/SearchQuery.hs SearchQuery.hs> in the examples directory.
1446-- Reactive programming is nothing but concurrent streaming which is what
1447-- streamly is all about. With streamly we can generate streams of events,
1448-- merge streams that are generated concurrently and process events
1449-- concurrently. We can do all this without any knowledge about the specifics
1450-- of the implementation of concurrency. In the following example you will see
1451-- that the code is just regular Haskell code without much streamly APIs used
1452-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1455-- This application has two independent and concurrent sources of event
1456-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1457-- events that deteriorate the health of the character in the game.
1458-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1459-- health improves and the game continues.
1462-- {-\# LANGUAGE FlexibleContexts #-}
1464-- import "Streamly.Prelude" (MonadAsync, SerialT)
1465-- import "Streamly.Prelude" as S
1466-- import Control.Monad (void)
1467-- import Control.Monad.IO.Class (MonadIO(liftIO))
1468-- import Control.Monad.State (MonadState, get, modify, runStateT)
1470-- data Event = Quit | Harm Int | Heal Int deriving (Show)
1472-- userAction :: MonadAsync m => 'SerialT' m Event
1473-- userAction = S.'repeatM' $ liftIO askUser
1476-- command <- getLine
1478-- "potion" -> return (Heal 10)
1479-- "harm" -> return (Harm 10)
1480-- "quit" -> return Quit
1481-- _ -> putStrLn "Type potion or harm or quit" >> askUser
1483-- acidRain :: MonadAsync m => 'SerialT' m Event
1484-- acidRain = S.'asyncly' $ S.'constRate' 1 $ S.'repeatM' $ liftIO $ return $ Harm 1
1486-- data Result = Check | Done
1488-- runEvents :: (MonadAsync m, MonadState Int m) => 'SerialT' m Result
1490-- event \<- userAction \`S.'parallel'` acidRain
1492-- Harm n -> modify (\\h -> h - n) >> return Check
1493-- Heal n -> modify (\\h -> h + n) >> return Check
1494-- Quit -> return Done
1496-- data Status = Alive | GameOver deriving Eq
1498-- getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status
1499-- getStatus result =
1501-- Done -> liftIO $ putStrLn "You quit!" >> return GameOver
1504-- liftIO $ if (h <= 0)
1505-- then putStrLn "You die!" >> return GameOver
1506-- else putStrLn ("Health = " <> show h) >> return Alive
1510-- putStrLn "Your health is deteriorating due to acid rain,\\
1511-- \\ type \\"potion\\" or \\"quit\\""
1512-- let runGame = S.'drainWhile' (== Alive) $ S.'mapM' getStatus runEvents
1513-- void $ runStateT runGame 60
1516-- You can also find the source of this example in the streamly-examples repo
1517-- as <https://github.com/composewell/streamly-examples/tree/master/AcidRain.hs AcidRain.hs>.
1518-- It has been adapted from Gabriel's
1519-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1521-- This is much simpler compared to the pipes version because of the builtin
1522-- concurrency in streamly. You can also find a SDL based reactive programming
1523-- example adapted from Yampa in
1524-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare.hs>.
1528-- Streamly is highly optimized for performance, it is designed for serious
1529-- high performing, concurrent and scalable applications. We have created the
1530-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1531-- package which is specifically and carefully designed to measure the
1532-- performance of Haskell streaming libraries fairly and squarely in the right
1533-- way. Streamly performs at par or even better than most streaming libraries
1534-- for serial operations even though it needs to deal with the concurrency
1539-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1541-- Interop with @vector@:
1544-- import "Streamly.Prelude" (SerialT)
1545-- import qualified "Streamly.Prelude" as S
1546-- import qualified Data.Vector.Fusion.Stream.Monadic as V
1548-- -- | vector to streamly
1549-- fromVector :: (S.IsStream t, Monad m) => V.Stream m a -> t m a
1550-- fromVector = S.unfoldrM unconsV
1555-- then return Nothing
1558-- return $ Just (h, V.tail v)
1560-- -- | streamly to vector
1561-- toVector :: Monad m => SerialT m a -> V.Stream m a
1562-- toVector = V.unfoldrM (S.uncons . S.adapt)
1565-- S.toList (fromVector (V.fromList [1..3])) >>= print
1566-- V.toList (toVector (S.fromFoldable [1..3])) >>= print
1569-- Interop with @pipes@:
1572-- import "Streamly.Prelude" (IsStream, MonadAsync, SerialT)
1573-- import qualified "Streamly.Prelude" as S
1574-- import qualified Pipes as P
1575-- import qualified Pipes.Prelude as P
1577-- -- | pipes to streamly
1578-- fromPipes :: (IsStream t, MonadAsync m) => P.Producer a m () -> t m a
1579-- fromPipes = S.'unfoldrM' unconsP
1581-- -- Adapt P.next to return a Maybe instead of Either
1582-- unconsP p = either (const Nothing) Just <$> P.next p
1584-- -- | streamly to pipes
1585-- toPipes :: Monad m => SerialT m a -> P.Producer a m ()
1586-- toPipes = P.unfoldr unconsEither
1588-- -- Adapt S.uncons to return an Either instead of Maybe
1589-- unconsEither :: Monad m => SerialT m a -> m (Either () (a, SerialT m a))
1590-- unconsEither s = maybe (Left ()) Right <$> S.'uncons' s
1593-- S.'toList' (fromPipes (P.each [1..3])) >>= print
1594-- P.toListM (toPipes (S.'fromFoldable' [1..3])) >>= print
1597-- Interop with @streaming@:
1600-- import "Streamly.Prelude" (IsStream, MonadAsync, SerialT)
1601-- import qualified "Streamly.Prelude" as S
1602-- import qualified Streaming as SG
1603-- import qualified Streaming.Prelude as SG
1605-- -- | streaming to streamly
1606-- fromStreaming :: (IsStream t, MonadAsync m) => SG.Stream (SG.Of a) m () -> t m a
1607-- fromStreaming = S.'unfoldrM' SG.uncons
1609-- -- | streamly to streaming
1610-- toStreaming :: Monad m => SerialT m a -> SG.Stream (SG.Of a) m ()
1611-- toStreaming = SG.unfoldr unconsEither
1613-- -- Adapt S.uncons to return an Either instead of Maybe
1614-- unconsEither :: Monad m => SerialT m a -> m (Either () (a, SerialT m a))
1615-- unconsEither s = maybe (Left ()) Right <$> S.'uncons' s
1618-- S.'toList' (fromStreaming (SG.each [1..3])) >>= print
1619-- SG.toList (toStreaming (S.'fromFoldable' [1..3])) >>= print
1622-- Interop with @conduit@:
1625-- import "Streamly.Prelude" (IsStream, MonadAsync, SerialT)
1626-- import qualified "Streamly.Prelude" as S
1627-- import qualified Data.Conduit as C
1628-- import qualified Data.Conduit.List as C
1630-- -- | conduit to streamly
1631-- fromConduit :: (IsStream t, MonadAsync m) => C.ConduitT () a m () -> t m a
1632-- fromConduit = S.unfoldrM C.unconsM . C.sealConduitT
1634-- -- | streamly to conduit
1635-- toConduit :: Monad m => SerialT m a -> C.ConduitT i a m ()
1636-- toConduit = C.unfoldM S.'uncons'
1639-- S.'toList' (fromConduit (C.sourceList [1..3])) >>= print
1640-- C.runConduit (toConduit (S.'fromFoldable' [1..3]) C..| C.consume) >>= print
1645-- List transformers and logic programming monads also provide a product style
1646-- composition similar to streamly, however streamly generalizes it with the
1647-- time dimension; allowing streams to be composed in an asynchronous and
1648-- concurrent fashion in many different ways. It also provides multiple
1649-- alternative ways of composing streams e.g. serial, interleaved or
1652-- This seemingly simple addition of asynchronicity and concurrency to product
1653-- style streaming composition unifies a number of disparate abstractions into
1654-- one powerful, concise and elegant abstraction. A wide variety of
1655-- programming problems can be solved elegantly with this abstraction. In
1656-- particular, it unifies three major programming domains namely
1657-- non-deterministic (logic) programming, concurrent programming and functional
1658-- reactive programming. In other words, you can do everything with this one
1659-- abstraction that you could do with the popular libraries listed under these
1660-- categories in the list below.
1663-- +-----------------+----------------+
1664-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1665-- | +----------------+
1666-- | | <https://hackage.haskell.org/package/list-t list-t> |
1667-- | +----------------+
1668-- | | <https://hackage.haskell.org/package/logict logict> |
1669-- +-----------------+----------------+
1670-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1671-- | +----------------+
1672-- | | <https://hackage.haskell.org/package/streaming streaming> |
1673-- | +----------------+
1674-- | | <https://hackage.haskell.org/package/pipes pipes> |
1675-- | +----------------+
1676-- | | <https://hackage.haskell.org/package/conduit conduit> |
1677-- +-----------------+----------------+
1678-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1679-- | +----------------+
1680-- | | <https://hackage.haskell.org/package/transient transient> |
1681-- +-----------------+----------------+
1682-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1683-- | +----------------+
1684-- | | <https://hackage.haskell.org/package/dunai dunai> |
1685-- | +----------------+
1686-- | | <https://hackage.haskell.org/package/reflex reflex> |
1687-- +-----------------+----------------+
1690-- Streamly is a list-transformer. It provides all the functionality provided
1691-- by any of the list transformer and logic programming packages listed above.
1692-- In addition, Streamly naturally integrates the concurrency dimension to the
1693-- basic list transformer functionality.
1695-- When it comes to streaming, in terms of the streaming API streamly is almost
1696-- identical to the vector package. Streamly, vector and streaming packages all
1697-- represent a stream as data and are therefore similar in the fundamental
1698-- approach to streaming. The fundamental difference is that streamly adds
1699-- concurrency support and the monad instance provides concurrent looping.
1700-- Other streaming libraries like pipes, conduit and machines represent and
1701-- compose stream processors rather than the stream data and therefore fall in
1702-- another class of streaming libraries and have comparatively more complicated
1705-- When it comes to concurrency, streamly can do everything that the @async@
1706-- package can do and more. async provides applicative concurrency whereas
1707-- streamly provides both applicative and monadic concurrency. The
1708-- 'ZipAsync' type behaves like the applicative instance of async. In
1709-- comparison to transient streamly has a first class streaming interface and
1710-- is a monad transformer that can be used universally in any Haskell monad
1711-- transformer stack. Streamly was in fact originally inspired by the
1712-- concurrency implementation in @transient@ though it has no resemblance with
1713-- that and takes a lazy pull approach versus transient's strict push approach.
1715-- The non-determinism, concurrency and streaming combination make streamly a
1716-- strong reactive programming library as well. Reactive programming is
1717-- fundamentally stream of events that can be processed concurrently. The
1718-- example in this tutorial as well as the
1719-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare>
1720-- example from Yampa demonstrate the basic reactive capability of streamly.
1721-- In core concepts streamly is strikingly similar to @dunai@. dunai was
1722-- designed from a FRP perspective and streamly was originally designed from a
1723-- concurrency perspective. However, both have similarity at the core.
1727-- * Read the documentation of "Streamly" module
1728-- * Read the documentation of "Streamly.Prelude" module
1729-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.
1730-- * See the tests in the "test" directory of the package
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list