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-- >>> import Prelude hiding (break, map, span, splitAt)
184-- >>> import qualified Streamly.Prelude as Stream
185-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream (parse, foldMany)
186-- >>> import qualified Streamly.Data.Fold as Fold
187-- >>> import qualified Streamly.Internal.Data.Fold as Fold
188-- >>> import qualified Streamly.Internal.Data.Parser as Parser
189-- >>> import Control.Monad.IO.Class (MonadIO(..))
190-- >>> import Control.Monad.Trans.Class (MonadTrans(lift))
191-- >>> import Data.Function ((&))
192-- >>> import Data.Char (toUpper)
193-- >>> import Control.Concurrent
194-- >>> import Control.Monad (forever)
195-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
196-- >>> import Streamly.Prelude
199-- delay n = Stream.fromEffect $ do
200-- threadDelay (n * 1000000)
202-- putStrLn (show tid ++ ": Delay " ++ show n)
208-- The monadic stream API offered by Streamly is very close to the Haskell
209-- "Prelude" pure lists' API, it can be considered as a natural extension of
210-- lists to monadic actions. Streamly streams provide concurrent composition
211-- and merging of streams. It can be considered as a concurrent list
214-- The basic stream type is 'Serial', it represents a sequence of IO actions,
215-- and is a 'Monad'. The 'Serial' monad is almost a drop in replacement for
216-- the 'IO' monad, IO monad is a special case of the 'Serial' monad; IO monad
217-- represents a single IO action whereas the 'Serial' monad represents a series
218-- of IO actions. The only change you need to make to go from 'IO' to 'Serial'
219-- is to use 'drain' to run the monad and to prefix the IO actions with
220-- either 'fromEffect' or 'liftIO'. If you use liftIO you can switch from 'Serial'
221-- to IO monad by simply removing the 'drain' function; no other changes
222-- are needed unless you have used some stream specific composition or
225-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
226-- pure lists are a special case of monadic streams. If you use 'nil' in place
227-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' stream.
228-- The only difference is that the elements must be monadic type and to operate
229-- on the streams we must use the corresponding functions from
230-- "Streamly.Prelude" instead of using the base "Prelude".
234-- Many stream operations can be done concurrently:
236-- * Streams can be generated concurrently.
238-- * Streams can be merged concurrently.
240-- * Multiple stages in a streaming pipeline can run concurrently.
242-- * Streams can be mapped and zipped concurrently.
244-- * In monadic composition they combine like a list transformer,
245-- providing concurrent non-determinism.
247-- There are three basic concurrent stream styles, 'Ahead', 'Async', and
248-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that
249-- they can speculatively execute multiple stream actions concurrently in
250-- advance. 'Ahead' would return exactly the same stream as 'Serial' except
251-- that it may execute the actions concurrently. The 'Async' style streams,
252-- like 'Ahead', speculatively execute multiple stream actions in advance but
253-- return the results in their finishing order rather than in the stream
254-- traversal order. 'Parallel' is like 'Async' except that it provides
255-- unbounded parallelism instead of controlled parallelism.
257-- For easy reference, we can classify the stream types based on /execution order/,
258-- /consumption order/, and /bounded or unbounded/ concurrency.
259-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
260-- execution we execute the next action in the stream only after the previous
261-- one has finished executing. In asynchronous execution multiple actions in
262-- the stream can be executed asynchronously i.e. the next action can start
263-- executing even before the first one has finished. Consumption order
264-- determines the order in which the outputs generated by the composition are
265-- consumed. Consumption could be serial or asynchronous. In serial
266-- consumption, the outputs are consumed in the traversal order, in
267-- asynchronous consumption the outputs are consumed as they arrive i.e. first
268-- come first serve order.
270-- +------------+--------------+--------------+--------------+
271-- | Type | Execution | Consumption | Concurrency |
272-- +============+==============+==============+==============+
273-- | 'Serial' | Serial | Serial | None |
274-- +------------+--------------+--------------+--------------+
275-- | 'Ahead' | Asynchronous | Serial | bounded |
276-- +------------+--------------+--------------+--------------+
277-- | 'Async' | Asynchronous | Asynchronous | bounded |
278-- +------------+--------------+--------------+--------------+
279-- | 'Parallel' | Asynchronous | Asynchronous | unbounded |
280-- +------------+--------------+--------------+--------------+
282-- All these types can be freely inter-converted using type conversion
283-- combinators or type annotations, without any cost, to achieve the desired
284-- composition style. To force a particular type of composition, we coerce the
285-- stream type using the corresponding type adapting combinator from
286-- 'fromSerial', 'fromAhead', 'fromAsync', or 'fromParallel'. The default stream type
287-- is inferred as 'Serial' unless you change it by using one of the combinators
288-- or by using a type annotation.
292-- Streams can be combined using '<>' or 'mappend' to form a
293-- composite. Composite streams can be interpreted in a depth first or
294-- breadth first manner using an appropriate type conversion before
295-- consumption. Deep (e.g. 'Serial') stream type variants traverse a
296-- composite stream in a depth first manner, such that each stream is
297-- traversed fully before traversing the next stream. Wide
298-- (e.g. 'WSerial') stream types traverse it in a breadth first
299-- manner, such that one element from each stream is traversed before
300-- coming back to the first stream again.
302-- Each stream type has a wide traversal variant prefixed by 'W'. The wide
303-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad
304-- compositions of the streams.
305-- The following table summarizes the basic types and the corresponding wide
309-- +------------+-----------+
311-- +============+===========+
312-- | 'Serial' | 'WSerial' |
313-- +------------+-----------+
314-- | 'Ahead' | 'WAhead' |
315-- +------------+-----------+
316-- | 'Async' | 'WAsync' |
317-- +------------+-----------+
320-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
321-- zip streams serially or concurrently using 'Applicative' operation. These
322-- types are not monads they are only applicatives and they do not differ in
323-- 'Semigroup' composition.
328-- When writing concurrent programs it is advised to not use the concurrent
329-- style stream combinators blindly at the top level. That might create too
330-- much concurrency where it is not even required, and can even degrade
331-- performance in some cases. In some cases it can also lead to surprising
332-- behavior because of some code that is supposed to be serial becoming
333-- concurrent. Please be aware that all concurrency capable APIs that you may
334-- have used under the scope of a concurrent stream combinator will become
335-- concurrent. For example if you have a 'repeatM' somewhere in your program
336-- and you use 'fromParallel' on top, the 'repeatM' becomes fully parallel,
337-- resulting into an infinite parallel execution . Instead, use the
338-- /Keep It Serial and Stupid/ principle, start with the default serial
339-- composition and enable concurrent combinators only when and where necessary.
340-- When you use a concurrent combinator you can use an explicit 'fromSerial'
341-- combinator to suppress any unnecessary concurrency under the scope of that
346-- To represent streams in an arbitrary monad use the more general monad
347-- transformer types for example the monad transformer type corresponding to
348-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values
349-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a
350-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for
353-- Similarly we have monad transformer types for other stream types as well viz.
354-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
356-- To lift a value from an underlying monad in a monad transformer stack into a
357-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
360-- >>> Stream.drain $ liftIO $ putStrLn "Hello world!"
363-- >>> Stream.drain $ lift $ putStrLn "Hello world!"
371-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
372-- a monadic action at the head of the stream.
375-- >>> Stream.toList Stream.nil
378-- Stream.toList $ getLine |: getLine |: Stream.nil
385-- To create a singleton stream from a pure value use 'fromPure' or 'pure' and to
386-- create a singleton stream from a monadic action use 'fromEffect'. Note that in
387-- case of Zip applicative streams "pure" repeats the value to generate an
391-- >>> Stream.toList $ pure 1
394-- >>> Stream.toList $ Stream.fromPure 1
397-- Stream.toList $ Stream.fromEffect getLine
403-- To create a stream from pure values in a 'Foldable' container use
404-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
407-- >>> Stream.toList $ Stream.fromFoldable [1..3]
410-- >>> Stream.toList $ Prelude.foldr Stream.cons Stream.nil [1..3]
415-- To create a stream from monadic actions in a 'Foldable' container just use a
416-- right fold using 'consM' and 'nil':
419-- >>> Stream.drain $ Prelude.foldr (|:) Stream.nil [putStr "Hello ", putStrLn "world!"]
424-- For more ways to construct a stream see the module "Streamly.Prelude".
426-- $generatingConcurrently
428-- Monadic construction and generation functions like 'consM', 'unfoldrM',
429-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' work concurrently
430-- when used with appropriate stream type combinator. The pure versions of
431-- these APIs are not concurrent, however you can use the monadic versions even
432-- for pure computations by wrapping the pure value in a monad to get the
433-- concurrent generation capability where required.
435-- The following code finishes in 3 seconds (6 seconds when serial):
438-- >>> let p n = threadDelay (n * 1000000) >> return n
439-- >>> Stream.toList $ Stream.fromParallel $ p 3 |: p 2 |: p 1 |: Stream.nil
442-- (3.01 secs, 2,018,432 bytes)
443-- >>> Stream.toList $ Stream.fromAhead $ p 3 |: p 2 |: p 1 |: Stream.nil
446-- (3.01 secs, 2,055,880 bytes)
448-- The following finishes in 10 seconds (100 seconds when serial):
451-- Stream.'drain' $ Stream.'fromAsync' $ Stream.'replicateM' 10 $ p 10
452-- (10.01 secs, 2,080,320 bytes)
458-- We have already seen 'drain' and toList to eliminate a stream in the
459-- examples above. 'drain' runs a stream discarding the results i.e. only
460-- for effects. 'toList' runs the stream and collects the results in a list.
462-- For other ways to eliminate a stream see the @Folding@ section in
463-- "Streamly.Prelude" module.
467-- Transformation over a stream is the equivalent of a @for@ loop construct in
468-- imperative paradigm. We iterate over every element in the stream and perform
469-- certain transformations for each element. Transformations may involve
470-- mapping functions over the elements, filtering elements from the stream or
471-- folding all the elements in the stream into a single value. Streamly streams
472-- are exactly like lists and you can perform all the transformations in the
473-- same way as you would on lists.
475-- Here is a simple console echo program that just echoes every input line,
479-- Stream.'drain' $ Stream.'repeatM' getLine & Stream.'mapM' putStrLn
482-- The following code snippet reads lines from standard input, filters blank
483-- lines, drops the first non-blank line, takes the next two, up cases them,
484-- numbers them and prints them:
487-- main = Stream.'drain' $
488-- Stream.'repeatM' getLine
489-- & Stream.'filter' (not . null)
492-- & fmap (map toUpper)
493-- & Stream.'zipWith' (\\n s -> show n ++ " " ++ s) (Stream.'fromFoldable' [1..])
494-- & Stream.'mapM' putStrLn
497-- $concurrentTransformation
499-- Monadic transformation functions 'mapM' and 'sequence' work concurrently
500-- when used with appropriate stream type combinators. The pure versions do not
501-- work concurrently, however you can use the monadic versions even for pure
502-- computations to get the concurrent transformation capability where required.
504-- This would print a value every second (2 seconds when serial):
507-- let p n = threadDelay (n * 1000000) >> return n
508-- Stream.'drain' $ Stream.'fromAhead' $ Stream.'mapM' (\\x -> p 1 >> print x) (Stream.'fromSerial' $ Stream.'repeatM' (p 1))
512-- $concurrentApplication
514-- The concurrent function application operators '|$' and '|&' apply a stream
515-- argument to a stream function concurrently to compose a concurrent pipeline
516-- of stream processing functions:
518-- Because both the stages run concurrently, we would see a delay of only 1
519-- second instead of 2 seconds in the following:
522-- let p n = threadDelay (n * 1000000) >> return n
523-- Stream.'drain' $ Stream.'repeatM' (p 1) '|&' Stream.'mapM' (\\x -> p 1 >> print x)
528-- We can combine two streams into a single stream using semigroup composition
529-- operation '<>'. Streams can be combined in many different ways as described
530-- in the following sections, the '<>' operation behaves differently depending
531-- on the stream type in effect. The stream type and therefore the composition
532-- style can be changed at any point using one of the type combinators as
537-- In most of example snippets we do not repeat the imports. Where imports are
538-- not explicitly specified use the imports shown below.
541-- import Control.Concurrent
542-- import Control.Monad (forever)
543-- import Control.Monad.IO.Class (MonadIO(..))
544-- import Control.Monad.Trans.Class (MonadTrans(lift))
545-- import Data.Function ((&))
546-- import Data.Char (toUpper)
547-- import Prelude hiding (break, map, span, splitAt)
548-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
549-- import qualified Streamly.Prelude as Stream
550-- import qualified Streamly.Internal.Data.Stream.IsStream as Stream (parse, foldMany)
551-- import qualified Streamly.Data.Fold as Fold
552-- import qualified Streamly.Internal.Data.Fold as Fold
553-- import qualified Streamly.Internal.Data.Parser as Parser
554-- import Streamly.Prelude
557-- To illustrate concurrent vs serial composition aspects, we will use the
558-- following @delay@ function to introduce a sleep or delay specified in
559-- seconds. After the delay it prints the number of seconds it slept.
563-- delay n = Stream.fromEffect $ do
564-- threadDelay (n * 1000000)
566-- putStrLn (show tid ++ ": Delay " ++ show n)
573-- The 'Semigroup' operation '<>' of the 'Serial' type combines the two streams
574-- in a /serial depth first/ manner. We use the 'fromSerial' type combinator to
575-- effect 'Serial' style of composition. We can also use an explicit 'Serial'
576-- type annotation for the stream to achieve the same effect. However, since
577-- 'Serial' is the default type unless explicitly specified by using a
578-- combinator, we can omit using an explicit combinator or type annotation for
579-- this style of composition.
581-- When two streams with multiple elements are combined in this manner, the
582-- monadic actions in the two streams are performed sequentially i.e. first all
583-- actions in the first stream are performed sequentially and then all actions
584-- in the second stream are performed sequentially. We call it
585-- /serial depth first/ as the full depth of one stream is fully traversed
586-- before we move to the next. The following example prints the sequence 1, 2,
590-- >>> Stream.drain $ (print 1 |: print 2 |: Stream.nil) <> (print 3 |: print 4 |: Stream.nil)
598-- All actions in both the streams are performed serially in the same thread.
599-- In the following example we can see that all actions are performed in the
600-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
603-- >>> Stream.drain $ delay 3 <> delay 2 <> delay 1
604-- ThreadId ... Delay 3
605-- ThreadId ... Delay 2
606-- ThreadId ... Delay 1
610-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
611-- 'serial'. We can use 'serial' to join streams in a sequential manner
612-- irrespective of the type of stream:
615-- >>> Stream.drain $ (print 1 |: print 2 |: Stream.nil) `Stream.serial` (print 3 |: print 4 |: Stream.nil)
625-- The 'Semigroup' operation '<>' of the 'WSerial' type combines the two
626-- streams in a /serial breadth first/ manner. We use the fromWSerial type
627-- combinator to effect 'WSerial' style of composition. We can also use the
628-- 'WSerial' type annotation for the stream to achieve the same effect.
630-- When two streams with multiple elements are combined in this manner, we
631-- traverse all the streams in a breadth first manner i.e. one action from each
632-- stream is performed and yielded to the resulting stream before we come back
633-- to the first stream again and so on.
634-- The following example prints the sequence 1, 3, 2, 4
637-- >>> Stream.drain . Stream.fromWSerial $ (print 1 |: print 2 |: Stream.nil) <> (print 3 |: print 4 |: Stream.nil)
645-- Even though the monadic actions of the two streams are performed in an
646-- interleaved manner they are all performed serially in the same thread. In
647-- the following example we can see that all actions are performed in the same
648-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
651-- >>> Stream.drain . Stream.fromWSerial $ delay 3 <> delay 2 <> delay 1
652-- ThreadId ...: Delay 3
653-- ThreadId ...: Delay 2
654-- ThreadId ...: Delay 1
658-- The polymorphic version of the 'WSerial' binary operation '<>' is called
659-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
660-- irrespective of the type, notice that we have not used the fromWSerial
661-- combinator in the following example:
664-- >>> Stream.drain $ (print 1 |: print 2 |: Stream.nil) `Stream.wSerial` (print 3 |: print 4 |: Stream.nil)
672-- Note that this composition cannot be used to fold infinite number of streams
673-- since it requires preserving the state until a stream is finished.
677-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a
678-- /serial depth first/ manner with concurrent lookahead. We use the 'fromAhead'
679-- type combinator to effect 'Ahead' style of composition. We can also use an
680-- explicit 'Ahead' type annotation for the stream to achieve the same effect.
682-- When two streams are combined in this manner, the streams are traversed in
683-- depth first manner just like 'Serial', however it can execute the next
684-- stream concurrently and keep the results ready when its turn arrives.
685-- Concurrent execution of the next stream(s) is performed if the first stream
686-- blocks or if it cannot produce output at the rate that is enough to meet the
687-- consumer demand. Multiple streams can be executed concurrently to meet the
688-- demand. The following example would print the result in a second even
689-- though each action in each stream takes one second:
692-- >>> p n = threadDelay 1000000 >> return n
695-- xs <- Stream.toList . Stream.fromAhead $ (p 1 |: p 2 |: Stream.nil) <> (p 3 |: p 4 |: Stream.nil)
702-- Each stream is constructed 'aheadly' and then both the streams are merged
703-- 'aheadly', therefore, all the actions can run concurrently but the result is
704-- presented in serial order.
706-- You can also use the polymorphic combinator 'ahead' in place of '<>' to
707-- compose any type of streams in this manner.
711-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
712-- streams in a depth first manner with parallel look ahead. We use the
713-- 'fromAsync' type combinator to effect 'Async' style of composition. We
714-- can also use the 'Async' type annotation for the stream type to achieve
717-- When two streams with multiple elements are combined in this manner, the
718-- streams are traversed in depth first manner just like 'Serial', however it
719-- can execute the next stream concurrently and return the results from it
720-- as they arrive i.e. the results from the next stream may be yielded even
721-- before the results from the first stream. Concurrent execution of the next
722-- stream(s) is performed if the first stream blocks or if it cannot produce
723-- output at the rate that is enough to meet the consumer demand. Multiple
724-- streams can be executed concurrently to meet the demand.
725-- In the example below each element in the stream introduces a constant delay
726-- of 1 second, however, it takes just one second to produce all the results.
727-- The results are not guaranteed to be in any particular order:
730-- p n = threadDelay 1000000 >> return n
732-- xs <- Stream.toList . Stream.fromAsync $ (p 1 |: p 2 |: Stream.nil) <> (p 3 |: p 4 |: Stream.nil)
738-- The constituent streams are also composed in 'Async' manner and the
739-- composition of streams too. We can compose the constituent streams to run
740-- serially, in that case it would take 2 seconds to produce all the results.
741-- The elements in the serial streams would be in serial order in the results:
744-- > p n = threadDelay 1000000 >> return n
745-- > Stream.toList . Stream.fromAsync $ (Stream.fromSerial $ p 1 |: p 2 |: Stream.nil) <> (fromSerial $ p 3 |: p 4 |: Stream.nil)
746-- [1,3,4,2] -- order is not predictable
750-- In the following example we can see that new threads are started when a
751-- computation blocks. Notice that the output from the stream with the
752-- shortest delay is printed first. The whole computation takes @maximum of
753-- (3, 2, 1) = 3@ seconds:
756-- >>> Stream.drain . Stream.fromAsync $ delay 3 <> delay 2 <> delay 1
757-- ThreadId ...: Delay 1
758-- ThreadId ...: Delay 2
759-- ThreadId ...: Delay 3
763-- When we have a tree of computations composed using this style, the tree is
764-- traversed in DFS style just like the 'Serial' style, the only difference is
765-- that here we can move on to executing the next stream if a stream blocks.
766-- However, we will not start new threads if we have sufficient output to
767-- saturate the consumer. This is why we call it left-biased demand driven or
768-- adaptive concurrency style, the concurrency tends to stay on the left side
769-- of the composition as long as possible. More threads are started based on
770-- the pull rate of the consumer. The following example prints an output every
771-- second as all of the actions are concurrent.
774-- >>> Stream.drain . Stream.fromAsync $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
775-- ThreadId ...: Delay 1
776-- ThreadId ...: Delay 2
777-- ThreadId ...: Delay 3
778-- ThreadId ...: Delay 4
782-- All the computations may even run in a single thread when more threads are
783-- not needed. As you can see, in the following example the computations are
784-- run in a single thread one after another, because none of them blocks.
785-- However, if the thread consuming the stream were faster than the producer
786-- then it would have started parallel threads for each computation to keep up
787-- even if none of them blocks:
790-- main = Stream.'drain' . Stream.'fromAsync' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
791-- where traced m = Stream.'fromEffect' (myThreadId >>= print) >> return m
799-- Note that the order of printing in the above examples may change due to
800-- variations in scheduling latencies for concurrent threads.
802-- The polymorphic version of the 'Async' binary operation '<>' is called
803-- 'async'. We can use 'async' to join streams in a left biased
804-- adaptively concurrent manner irrespective of the type, notice that we have
805-- not used the 'fromAsync' combinator in the following example:
808-- >>> Stream.drain $ delay 3 `Stream.async` delay 2 `Stream.async` delay 1
809-- ThreadId ...: Delay 1
810-- ThreadId ...: Delay 2
811-- ThreadId ...: Delay 3
815-- Since the concurrency provided by this operator is demand driven it cannot
816-- be used when the composed computations start timers that are relative to
817-- each other because all computations may not be started at the same time and
818-- therefore timers in all of them may not start at the same time. When
819-- relative timing among all computations is important or when we need to start
820-- all computations at once for any reason 'Parallel' style must be used
823-- 'Async' style utilizes resources optimally and should be preferred over
824-- 'Parallel' or 'WAsync' unless you really need those. 'Async' should be used
825-- when we know that the computations can run in parallel but we do not care if
826-- they actually run in parallel or not, that decision can be left to the
827-- scheduler based on demand. Also, note that 'async' operator can be used to fold
828-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
829-- because it does not require us to run all of them at the same time in a fair
834-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
835-- a concurrent manner using /breadth first traversal/. We use the 'fromWAsync'
836-- type combinator to effect 'WAsync' style of composition. We can also use the
837-- 'WAsync' type annotation for the stream to achieve the same effect.
839-- When streams with multiple elements are combined in this manner, we traverse
840-- all the streams concurrently in a breadth first manner i.e. one action from
841-- each stream is performed and yielded to the resulting stream before we come
842-- back to the first stream again and so on. Even though we execute the actions
843-- in a breadth first order the outputs are consumed on a first come first
846-- In the following example we can see that outputs are produced in the breadth
847-- first traversal order but this is not guaranteed.
852-- hSetBuffering stdout LineBuffering
853-- Stream.drain . Stream.fromWAsync $ (Stream.fromSerial $ print 1 |: print 2 |: Stream.nil) <> (Stream.fromSerial $ print 3 |: print 4 |: Stream.nil)
862-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
863-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
864-- concurrent traversal irrespective of the type, notice that we have not used
865-- the 'fromWAsync' combinator in the following example:
868-- >>> Stream.drain $ delay 3 `Stream.wAsync` delay 2 `Stream.wAsync` delay 1
869-- ThreadId ...: Delay 1
870-- ThreadId ...: Delay 2
871-- ThreadId ...: Delay 3
875-- Since the concurrency provided by this style is demand driven it may not
876-- be used when the composed computations start timers that are relative to
877-- each other because all computations may not be started at the same time and
878-- therefore timers in all of them may not start at the same time. When
879-- relative timing among all computations is important or when we need to start
880-- all computations at once for any reason 'Parallel' style must be used
886-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
887-- streams in a fairly concurrent manner with round robin scheduling. We use
888-- the 'fromParallel' type combinator to effect 'Parallel' style of composition.
889-- We can also use the 'Parallel' type annotation for the stream type to
890-- achieve the same effect.
892-- When two streams with multiple elements are combined in this manner, the
893-- monadic actions in both the streams are performed concurrently with a fair
894-- round robin scheduling. The outputs are yielded in the order in which the
895-- actions complete. This is pretty similar to the 'WAsync' type, the
896-- difference is that 'WAsync' is adaptive to the consumer demand and may or
897-- may not execute all actions in parallel depending on the demand, whereas
898-- 'Parallel' runs all the streams in parallel irrespective of the demand.
900-- The following example sends a query to all the three search engines in
901-- parallel and prints the name of the search engines in the order in which the
902-- responses arrive. You need the [http-conduit](http://hackage.haskell.org/package/http-conduit)
903-- package to run this example:
906-- import qualified Streamly.Prelude as Stream
907-- import Network.HTTP.Simple
909-- main = Stream.'drain' . Stream.'fromParallel' $ google \<> bing \<> duckduckgo
911-- google = get "https://www.google.com/search?q=haskell"
912-- bing = get "https://www.bing.com/search?q=haskell"
913-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
914-- get s = Stream.'fromEffect' (httpNoBody (parseRequest_ s) >> putStrLn (show s))
917-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
918-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
919-- manner irrespective of the type, notice that we have not used the
920-- 'fromParallel' combinator in the following example:
923-- >>> Stream.drain $ delay 3 `Stream.parallel` delay 2 `Stream.wAsync` delay 1
924-- ThreadId ...: Delay 1
925-- ThreadId ...: Delay 2
926-- ThreadId ...: Delay 3
930-- Note that this style of composition cannot be used to combine infinite
931-- number of streams, as it will lead to an infinite sized scheduling queue.
937-- The 'mkAsync' API can be used to create references to asynchronously running
938-- stream computations. We can then use 'uncons' to explore the streams
939-- arbitrarily and then recompose individual elements to create a new stream.
940-- This way we can dynamically decide which stream to explore at any given
941-- time. Take an example of a merge sort of two sorted streams. We need to
942-- keep consuming items from the stream which has the lowest item in the sort
943-- order. This can be achieved using async references to streams. See
944-- "MergeSort.hs" in <https://github.com/composewell/streamly-examples Streamly Examples>.
948-- We can use 'Monoid' instances to fold a container of streams in the desired
949-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
950-- to fold streams using the polymorphic combine operations:
952-- * 'concatFoldableWith' is like 'fold', it folds a 'Foldable' container of
953-- streams using the given composition operator.
954-- * 'concatMapFoldableWith' is like 'foldMap', it folds like
955-- @concatFoldableWith@ but also maps a function before folding.
956-- * 'concatForFoldableWith' is like @concatMapFoldableWith@ but the container
957-- argument comes before the function argument.
959-- All of the following are equivalent and start ten concurrent tasks each with
960-- a delay from 1 to 10 seconds, resulting in the printing of each number every
964-- import qualified "Streamly.Prelude" as Stream
965-- import Control.Concurrent
968-- Stream.'drain' $ Stream.'fromAsync' $ foldMap delay [1..10]
969-- Stream.'drain' $ Stream.'concatFoldableWith' Stream.'async' (map delay [1..10])
970-- Stream.'drain' $ Stream.'concatMapFoldableWith' Stream.'async' delay [1..10]
971-- Stream.'drain' $ Stream.'concatForFoldableWith' Stream.'async' [1..10] delay
972-- where delay n = Stream.'fromEffect' $ threadDelay (n * 1000000) >> print n
977-- Till now we discussed ways to apply transformations on a stream or to merge
978-- streams together to create another stream. We mentioned earlier that
979-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
980-- We will now discuss the concept of a nested composition of streams which is
981-- analogous to nested @for@ loops in the imperative paradigm. Functional
982-- programmers call this style of composition a list transformer or @ListT@.
983-- Logic programmers call it a logic monad or non-deterministic composition,
984-- but for ordinary imperative minded people like me it is easier to think in
985-- terms of good old nested @for@ loops.
989-- In functional programmer's parlance the 'Monad' instances of different
990-- 'IsStream' types implement non-determinism, exploring all possible
991-- combination of choices from both the streams. From an imperative
992-- programmer's point of view it behaves like nested loops i.e. for each
993-- element in the first stream and for each element in the second stream
994-- execute the body of the loop.
996-- The 'Monad' instances of 'Serial', 'WSerial', 'Async' and 'WAsync'
997-- stream types support different flavors of nested looping. In other words,
998-- they are all variants of list transformer. The nesting behavior of these
999-- types correspond exactly to the way they merge streams as we discussed in
1000-- the previous section.
1005-- The 'Monad' composition of the 'Serial' type behaves like a standard list
1006-- transformer. This is the default when we do not use an explicit type
1007-- combinator. However, the 'fromSerial' type combinator can be used to switch to
1008-- this style of composition. We will see how this style of composition works
1009-- in the following examples.
1011-- Let's start with an example with a simple @for@ loop without any nesting.
1012-- For simplicity of illustration we are using streams of pure values in all
1013-- the examples. However, the streams could also be made of monadic actions
1019-- x <- Stream.fromFoldable [3,2,1]
1022-- ThreadId ...: Delay 3
1023-- ThreadId ...: Delay 2
1024-- ThreadId ...: Delay 1
1028-- As we can see, the code after the @fromFoldable@ statement is run three
1029-- times, once for each value of @x@ drawn from the stream. All the three
1030-- iterations are serial and run in the same thread one after another. In
1031-- imperative terms this is equivalent to a @for@ loop with three iterations.
1033-- A console echo loop copying standard input to standard output can simply be
1034-- written like this:
1037-- main = Stream.'drain' $ forever $ Stream.'fromEffect' getLine >>= Stream.'fromEffect' . putStrLn
1040-- When multiple streams are composed using this style they nest in a DFS
1041-- manner i.e. nested iterations of a loop are executed before we proceed to
1042-- the next iteration of the parent loop. This behaves just like nested @for@
1043-- loops in imperative programming.
1048-- x <- Stream.fromFoldable [1,2]
1049-- y <- Stream.fromFoldable [3,4]
1050-- Stream.fromEffect $ putStrLn $ show (x, y)
1059-- Notice that this is analogous to merging streams of type 'Serial' or merging
1060-- streams using 'serial'.
1064-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except
1065-- that it can speculatively perform a bounded number of next iterations of a
1066-- loop concurrently.
1068-- The 'fromAhead' type combinator can be used to switch to this style of
1069-- composition. Alternatively, a type annotation can be used to specify the
1070-- type of the stream as 'Ahead'.
1074-- comp = Stream.toList . Stream.fromAhead $ do
1075-- x <- Stream.fromFoldable [3,2,1]
1076-- delay x >> return x
1079-- >>> comp >>= print
1080-- ThreadId ...: Delay 1
1081-- ThreadId ...: Delay 2
1082-- ThreadId ...: Delay 3
1087-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can
1088-- see all the three iterations are concurrent and run in different threads,
1089-- however, the results are returned in the serial order.
1091-- Concurrency is demand driven, when multiple streams are composed using this
1092-- style, the iterations are executed in a depth first manner just like
1093-- 'Serial' i.e. nested iterations are executed before we proceed to the next
1094-- outer iteration. The only difference is that we may execute multiple future
1095-- iterations concurrently and keep the results ready.
1098-- $concurrentNesting
1100-- The 'Monad' composition of 'Async' type can perform the iterations of a
1101-- loop concurrently. Concurrency is demand driven i.e. more concurrent
1102-- iterations are started only if the previous iterations are not able to
1103-- produce enough output for the consumer of the output stream. This works
1104-- exactly the same way as the merging of two streams 'asyncly' works.
1105-- This is the concurrent analogue of 'Serial' style monadic composition.
1107-- The 'fromAsync' type combinator can be used to switch to this style of
1108-- composition. Alternatively, a type annotation can be used to specify the
1109-- type of the stream as 'Async'.
1113-- Stream.drain . Stream.fromAsync $ do
1114-- x <- Stream.fromFoldable [3,2,1]
1117-- ThreadId ...: Delay 1
1118-- ThreadId ...: Delay 2
1119-- ThreadId ...: Delay 3
1123-- As we can see the code after the @fromFoldable@ statement is run three
1124-- times, once for each value of @x@. All the three iterations are concurrent
1125-- and run in different threads. The iteration with least delay finishes first.
1126-- When compared to imperative programming, this can be viewed as a @for@ loop
1127-- with three concurrent iterations.
1129-- Concurrency is demand driven just as in the case of 'async' merging.
1130-- When multiple streams are composed using this style, the iterations are
1131-- triggered in a depth first manner just like 'Serial' i.e. nested iterations are
1132-- executed before we proceed to the next iteration at higher level. However,
1133-- unlike 'Serial' more than one iterations may be started concurrently based
1134-- on the demand from the consumer of the stream.
1138-- Stream.drain . Stream.fromAsync $ do
1139-- x <- Stream.fromFoldable [1,2]
1140-- y <- Stream.fromFoldable [3,4]
1141-- Stream.fromEffect $ putStrLn $ show (x, y)
1150-- $interleavedNesting
1152-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
1153-- outer and inner loops in a nested loop composition. This works exactly the
1154-- same way as the merging of two streams in 'wSerially' fashion works. The
1155-- fromWSerial type combinator can be used to switch to this style of
1156-- composition. Alternatively, a type annotation can be used to specify the
1157-- type of the stream as 'WSerial'.
1161-- Stream.drain . Stream.fromWSerial $ do
1162-- x <- Stream.fromFoldable [1,2]
1163-- y <- Stream.fromFoldable [3,4]
1164-- Stream.fromEffect $ putStrLn $ show (x, y)
1176-- Just like 'Async' the 'Monad' composition of 'WAsync' runs the iterations of
1177-- a loop concurrently. The difference is in the nested loop behavior. The
1178-- nested loops in this type are traversed and executed in a breadth first
1179-- manner rather than the depth first manner of 'Async' style.
1180-- The loop nesting works exactly the same way as the merging of streams
1181-- 'wAsyncly' works. The 'fromWAsync' type combinator can be used to switch to
1182-- this style of composition. Alternatively, a type annotation can be used to
1183-- specify the type of the stream as 'WAsync'.
1187-- Stream.drain . Stream.fromWAsync $ do
1188-- x <- Stream.delay 1 $ Stream.fromFoldable [1,2]
1189-- y <- Stream.delay 1 $ Stream.fromFoldable [3,4]
1190-- Stream.fromEffect $ putStrLn $ show (x, y)
1201-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
1202-- iterations of a loop concurrently. The difference is in the nested loop
1203-- behavior. The streams at each nest level is run fully concurrently
1204-- irrespective of the demand. The loop nesting works exactly the same way as
1205-- the merging of streams 'parallely' works. The 'fromParallel' type combinator
1206-- can be used to switch to this style of composition. Alternatively, a type
1207-- annotation can be used to specify the type of the stream as 'Parallel'.
1211-- Stream.drain . Stream.fromParallel $ do
1212-- x <- Stream.fromFoldable [3,2,1]
1215-- ThreadId ...: Delay 1
1216-- ThreadId ...: Delay 2
1217-- ThreadId ...: Delay 3
1223-- Streamly code is usually written in a way that is agnostic of the
1224-- specific monadic composition type. We use a polymorphic type with a
1225-- 'IsStream' type class constraint. When running the stream we can choose the
1226-- specific mode of composition. For example take a look at the following code.
1229-- composed :: (Stream.IsStream t, Monad (t IO)) => t IO ()
1234-- Stream.'fromEffect' $ putStrLn $ show (sz, cl, sh)
1238-- sizes = Stream.'fromFoldable' [1, 2, 3]
1239-- colors = Stream.'fromFoldable' ["red", "green", "blue"]
1240-- shapes = Stream.'fromFoldable' ["triangle", "square", "circle"]
1243-- Now we can interpret this in whatever way we want:
1246-- main = Stream.'drain' . Stream.'fromSerial' $ composed
1247-- main = Stream.'drain' . Stream.'fromWSerial' $ composed
1248-- main = Stream.'drain' . Stream.'fromAsync' $ composed
1249-- main = Stream.'drain' . Stream.'fromWAsync' $ composed
1250-- main = Stream.'drain' . Stream.'fromParallel' $ composed
1253-- As an exercise try to figure out the output of this code for each mode of
1258-- 'fmap' transforms a stream by mapping a function on all elements of the
1259-- stream. 'fmap' behaves in the same way for all stream types, it is always
1263-- >>> (Stream.toList $ fmap show $ Stream.fromFoldable [1..10]) >>= print
1264-- ["1","2","3","4","5","6","7","8","9","10"]
1268-- Also see functions 'mapM' and 'sequence' from "Streamly.Prelude" module
1269-- which can map actions concurrently depending on the type of the input stream.
1273-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
1274-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are
1277-- The following example uses the 'Serial' applicative, it runs all iterations
1278-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
1281-- >>> d n = delay n >> return n
1282-- >>> s1 = d 1 <> d 2
1283-- >>> s2 = d 3 <> d 4
1285-- >>> (Stream.toList . Stream.fromSerial $ (,) <$> s1 <*> s2) >>= print
1286-- ThreadId ...: Delay 1
1287-- ThreadId ...: Delay 3
1288-- ThreadId ...: Delay 4
1289-- ThreadId ...: Delay 2
1290-- ThreadId ...: Delay 3
1291-- ThreadId ...: Delay 4
1292-- [(1,3),(1,4),(2,3),(2,4)]
1296-- Similarly 'WSerial' applicative runs the iterations in an interleaved
1297-- order but since it is serial it takes a total of 17 seconds:
1300-- >>> (Stream.toList . Stream.fromWSerial $ (,) <$> s1 <*> s2) >>= print
1301-- ThreadId ...: Delay 1
1302-- ThreadId ...: Delay 3
1303-- ThreadId ...: Delay 2
1304-- ThreadId ...: Delay 3
1305-- ThreadId ...: Delay 4
1306-- ThreadId ...: Delay 4
1307-- [(1,3),(2,3),(1,4),(2,4)]
1311-- 'Async' can run the iterations concurrently and therefore takes a total
1312-- of 6 seconds which is max (1, 2) + max (3, 4):
1315-- main = (Stream.'toList' . Stream.'fromAsync' $ (,) \<$> s1 \<*> s2) >>= print
1318-- ThreadId 34: Delay 1
1319-- ThreadId 36: Delay 2
1320-- ThreadId 35: Delay 3
1321-- ThreadId 36: Delay 3
1322-- ThreadId 35: Delay 4
1323-- ThreadId 36: Delay 4
1324-- [(1,3),(2,3),(1,4),(2,4)]
1327-- Similarly 'WAsync' as well can run the iterations concurrently and
1328-- therefore takes a total of 6 seconds (2 + 4):
1331-- main = (Stream.'toList' . Stream.'fromWAsync' $ (,) \<$> s1 \<*> s2) >>= print
1334-- ThreadId 34: Delay 1
1335-- ThreadId 36: Delay 2
1336-- ThreadId 35: Delay 3
1337-- ThreadId 36: Delay 3
1338-- ThreadId 35: Delay 4
1339-- ThreadId 36: Delay 4
1340-- [(1,3),(2,3),(1,4),(2,4)]
1345-- Zipping is a special transformation where the corresponding elements of two
1346-- streams are combined together using a zip function producing a new stream of
1347-- outputs. Two different types are provided for serial and concurrent zipping.
1348-- These types provide an applicative instance that can be used to lift
1349-- functions to zip the argument streams.
1350-- Also see the zipping functions in the "Streamly.Prelude" module.
1354-- The applicative instance of 'ZipSerial' type zips streams serially.
1355-- 'fromZipSerial' type combinator can be used to switch to serial applicative
1360-- d n = delay n >> return n
1361-- s1 = Stream.'fromSerial' $ d 1 <> d 2
1362-- s2 = Stream.'fromSerial' $ d 3 <> d 4
1364-- main = (Stream.'toList' . Stream.'fromZipSerial' $ (,) \<$> s1 \<*> s2) >>= print
1367-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1368-- everything runs serially:
1370-- >>> d n = delay n >> return n
1371-- >>> s1 = Stream.fromSerial $ d 1 <> d 2
1372-- >>> s2 = Stream.fromSerial $ d 3 <> d 4
1373-- >>> (Stream.toList . Stream.fromZipSerial $ (,) <$> s1 <*> s2) >>= print
1374-- ThreadId ...: Delay 1
1375-- ThreadId ...: Delay 3
1376-- ThreadId ...: Delay 2
1377-- ThreadId ...: Delay 4
1383-- The applicative instance of 'ZipAsync' type zips streams concurrently.
1384-- 'fromZipAsync' type combinator can be used to switch to parallel applicative
1389-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
1391-- d n = delay n >> return n
1392-- s1 = Stream.'fromSerial' $ d 1 <> d 2
1393-- s2 = Stream.'fromSerial' $ d 3 <> d 4
1396-- hSetBuffering stdout LineBuffering
1397-- (Stream.'toList' . Stream.'fromZipAsync' $ (,) \<$> s1 \<*> s2) >>= print
1400-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1401-- are produced concurrently, and 2 and 4 are produced concurrently:
1403-- >>> d n = delay n >> return n
1404-- >>> s1 = Stream.fromSerial $ d 1 <> d 2
1405-- >>> s2 = Stream.fromSerial $ d 3 <> d 4
1409-- hSetBuffering stdout LineBuffering
1410-- (Stream.toList . Stream.fromZipAsync $ (,) <$> s1 <*> s2) >>= print
1412-- ThreadId ...: Delay 1
1413-- ThreadId ...: Delay 2
1414-- ThreadId ...: Delay 3
1415-- ThreadId ...: Delay 4
1421-- When writing concurrent programs there are two distinct places where the
1422-- programmer can control the concurrency. First, when /composing/ a stream by
1423-- merging multiple streams we can choose an appropriate sum style operators to
1424-- combine them concurrently or serially. Second, when /processing/ a stream in
1425-- a monadic composition we can choose one of the monad composition types to
1426-- choose the desired type of concurrency.
1428-- In the following example the squares of @x@ and @y@ are computed
1429-- concurrently using the 'async' operation and the square roots of their
1430-- sum are computed serially because of the 'streamly' combinator. We can
1431-- choose different combinators for the monadic processing and the stream
1432-- generation, to control the concurrency. We can also use the 'fromAsync'
1433-- combinator instead of explicitly folding with 'async'.
1436-- import qualified "Streamly.Prelude" as Stream
1437-- import Data.List (sum)
1440-- z \<- Stream.'toList'
1441-- $ Stream.'fromSerial' -- Serial monadic processing (sqrt below)
1443-- x2 \<- Stream.'concatForFoldableWith' Stream.'async' [1..100] $ -- Concurrent @"for"@ loop
1444-- \\x -> return $ x * x -- body of the loop
1445-- y2 \<- Stream.'concatForFoldableWith' Stream.'async' [1..100] $
1446-- \\y -> return $ y * y
1447-- return $ sqrt (x2 + y2)
1451-- We can see how this directly maps to the imperative style
1452-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1453-- and operators instead of the ugly pragmas.
1455-- For more concurrent programming examples see,
1456-- <https://github.com/composewell/streamly-examples/tree/master/ListDir.hs ListDir.hs>,
1457-- <https://github.com/composewell/streamly-examples/tree/master/MergeSort.hs MergeSort.hs> and
1458-- <https://github.com/composewell/streamly-examples/tree/master/SearchQuery.hs SearchQuery.hs> in the examples directory.
1462-- Reactive programming is nothing but concurrent streaming which is what
1463-- streamly is all about. With streamly we can generate streams of events,
1464-- merge streams that are generated concurrently and process events
1465-- concurrently. We can do all this without any knowledge about the specifics
1466-- of the implementation of concurrency. In the following example you will see
1467-- that the code is just regular Haskell code without much streamly APIs used
1468-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1471-- This application has two independent and concurrent sources of event
1472-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1473-- events that deteriorate the health of the character in the game.
1474-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1475-- health improves and the game continues.
1478-- {-\# LANGUAGE FlexibleContexts \#-}
1480-- import "Streamly.Prelude" (MonadAsync, SerialT)
1481-- import "Streamly.Prelude" as Stream
1482-- import Control.Monad (void)
1483-- import Control.Monad.IO.Class (MonadIO(liftIO))
1484-- import Control.Monad.State (MonadState, get, modify, runStateT)
1486-- data Event = Quit | Harm Int | Heal Int deriving (Show)
1488-- userAction :: MonadAsync m => 'SerialT' m Event
1489-- userAction = Stream.'repeatM' $ liftIO askUser
1492-- command <- getLine
1494-- "potion" -> return (Heal 10)
1495-- "harm" -> return (Harm 10)
1496-- "quit" -> return Quit
1497-- _ -> putStrLn "Type potion or harm or quit" >> askUser
1499-- acidRain :: MonadAsync m => 'SerialT' m Event
1500-- acidRain = Stream.'fromAsync' $ Stream.'constRate' 1 $ Stream.'repeatM' $ liftIO $ return $ Harm 1
1502-- data Result = Check | Done
1504-- runEvents :: (MonadAsync m, MonadState Int m) => 'SerialT' m Result
1506-- event \<- userAction \`Stream.'parallel'` acidRain
1508-- Harm n -> modify (\\h -> h - n) >> return Check
1509-- Heal n -> modify (\\h -> h + n) >> return Check
1510-- Quit -> return Done
1512-- data Status = Alive | GameOver deriving Eq
1514-- getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status
1515-- getStatus result =
1517-- Done -> liftIO $ putStrLn "You quit!" >> return GameOver
1520-- liftIO $ if (h <= 0)
1521-- then putStrLn "You die!" >> return GameOver
1522-- else putStrLn ("Health = " <> show h) >> return Alive
1526-- putStrLn "Your health is deteriorating due to acid rain, type \\\"potion\\\" or \\\"quit\\\""
1527-- let runGame = Stream.'drainWhile' (== Alive) $ Stream.'mapM' getStatus runEvents
1528-- void $ runStateT runGame 60
1531-- You can also find the source of this example in the streamly-examples repo
1532-- as <https://github.com/composewell/streamly-examples/tree/master/AcidRain.hs AcidRain.hs>.
1533-- It has been adapted from Gabriel's
1534-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1536-- This is much simpler compared to the pipes version because of the builtin
1537-- concurrency in streamly. You can also find a SDL based reactive programming
1538-- example adapted from Yampa in
1539-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare.hs>.
1543-- Streamly is highly optimized for performance, it is designed for serious
1544-- high performing, concurrent and scalable applications. We have created the
1545-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1546-- package which is specifically and carefully designed to measure the
1547-- performance of Haskell streaming libraries fairly and squarely in the right
1548-- way. Streamly performs at par or even better than most streaming libraries
1549-- for serial operations even though it needs to deal with the concurrency
1554-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1556-- Interop with @vector@:
1559-- import "Streamly.Prelude" (SerialT)
1560-- import qualified "Streamly.Prelude" as Stream
1561-- import qualified Data.Vector.Fusion.Stream.Monadic as Vector
1563-- -- | vector to streamly
1564-- fromVector :: (Stream.IsStream t, Monad m) => Vector.Stream m a -> t m a
1565-- fromVector = Stream.unfoldrM unconsV
1568-- r <- Vector.null v
1570-- then return Nothing
1572-- h <- Vector.head v
1573-- return $ Just (h, Vector.tail v)
1575-- -- | streamly to vector
1576-- toVector :: Monad m => SerialT m a -> Vector.Stream m a
1577-- toVector = Vector.unfoldrM (Stream.uncons . Stream.adapt)
1580-- Stream.toList (fromVector (Vector.fromList [1..3])) >>= print
1581-- Vector.toList (toVector (Stream.fromFoldable [1..3])) >>= print
1584-- Interop with @pipes@:
1587-- import "Streamly.Prelude" (IsStream, MonadAsync, SerialT)
1588-- import qualified "Streamly.Prelude" as Stream
1589-- import qualified Pipes as Pipe
1590-- import qualified Pipes.Prelude as Pipe
1592-- -- | pipes to streamly
1593-- fromPipes :: (IsStream t, MonadAsync m) => Pipe.Producer a m () -> t m a
1594-- fromPipes = Stream.'unfoldrM' unconsP
1596-- -- Adapt Pipe.next to return a Maybe instead of Either
1597-- unconsP p = either (const Nothing) Just <$> Pipe.next p
1599-- -- | streamly to pipes
1600-- toPipes :: Monad m => SerialT m a -> Pipe.Producer a m ()
1601-- toPipes = Pipe.unfoldr unconsEither
1603-- -- Adapt Stream.uncons to return an Either instead of Maybe
1604-- unconsEither :: Monad m => SerialT m a -> m (Either () (a, SerialT m a))
1605-- unconsEither s = maybe (Left ()) Right <$> Stream.'uncons' s
1608-- Stream.'toList' (fromPipes (Pipe.each [1..3])) >>= print
1609-- Pipe.toListM (toPipes (Stream.'fromFoldable' [1..3])) >>= print
1612-- Interop with @streaming@:
1615-- import "Streamly.Prelude" (IsStream, MonadAsync, SerialT)
1616-- import qualified "Streamly.Prelude" as Stream
1617-- import qualified Streaming as Streaming
1618-- import qualified Streaming.Prelude as Streaming
1620-- -- | streaming to streamly
1621-- fromStreaming :: (IsStream t, MonadAsync m) => Streaming.Stream (Streaming.Of a) m () -> t m a
1622-- fromStreaming = Stream.'unfoldrM' Streaming.uncons
1624-- -- | streamly to streaming
1625-- toStreaming :: Monad m => SerialT m a -> Streaming.Stream (Streaming.Of a) m ()
1626-- toStreaming = Streaming.unfoldr unconsEither
1628-- -- Adapt Stream.uncons to return an Either instead of Maybe
1629-- unconsEither :: Monad m => SerialT m a -> m (Either () (a, SerialT m a))
1630-- unconsEither s = maybe (Left ()) Right <$> Stream.'uncons' s
1633-- Stream.'toList' (fromStreaming (Streaming.each [1..3])) >>= print
1634-- Streaming.toList (toStreaming (Stream.'fromFoldable' [1..3])) >>= print
1637-- Interop with @conduit@:
1640-- import "Streamly.Prelude" (IsStream, MonadAsync, SerialT)
1641-- import qualified "Streamly.Prelude" as Stream
1642-- import qualified Data.Conduit as Conduit
1643-- import qualified Data.Conduit.List as Conduit
1645-- -- | conduit to streamly
1646-- fromConduit :: (IsStream t, MonadAsync m) => Conduit.ConduitT () a m () -> t m a
1647-- fromConduit = Stream.unfoldrM Conduit.unconsM . Conduit.sealConduitT
1649-- -- | streamly to conduit
1650-- toConduit :: Monad m => SerialT m a -> Conduit.ConduitT i a m ()
1651-- toConduit = Conduit.unfoldM Stream.'uncons'
1654-- Stream.'toList' (fromConduit (Conduit.sourceList [1..3])) >>= print
1655-- Conduit.runConduit (toConduit (Stream.'fromFoldable' [1..3]) Conduit..| Conduit.consume) >>= print
1660-- List transformers and logic programming monads also provide a product style
1661-- composition similar to streamly, however streamly generalizes it with the
1662-- time dimension; allowing streams to be composed in an asynchronous and
1663-- concurrent fashion in many different ways. It also provides multiple
1664-- alternative ways of composing streams e.g. serial, interleaved or
1667-- This seemingly simple addition of asynchronicity and concurrency to product
1668-- style streaming composition unifies a number of disparate abstractions into
1669-- one powerful, concise and elegant abstraction. A wide variety of
1670-- programming problems can be solved elegantly with this abstraction. In
1671-- particular, it unifies three major programming domains namely
1672-- non-deterministic (logic) programming, concurrent programming and functional
1673-- reactive programming. In other words, you can do everything with this one
1674-- abstraction that you could do with the popular libraries listed under these
1675-- categories in the list below.
1678-- +-----------------+----------------+
1679-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1680-- | +----------------+
1681-- | | <https://hackage.haskell.org/package/list-t list-t> |
1682-- | +----------------+
1683-- | | <https://hackage.haskell.org/package/logict logict> |
1684-- +-----------------+----------------+
1685-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1686-- | +----------------+
1687-- | | <https://hackage.haskell.org/package/streaming streaming> |
1688-- | +----------------+
1689-- | | <https://hackage.haskell.org/package/pipes pipes> |
1690-- | +----------------+
1691-- | | <https://hackage.haskell.org/package/conduit conduit> |
1692-- +-----------------+----------------+
1693-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1694-- | +----------------+
1695-- | | <https://hackage.haskell.org/package/transient transient> |
1696-- +-----------------+----------------+
1697-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1698-- | +----------------+
1699-- | | <https://hackage.haskell.org/package/dunai dunai> |
1700-- | +----------------+
1701-- | | <https://hackage.haskell.org/package/reflex reflex> |
1702-- +-----------------+----------------+
1705-- Streamly is a list-transformer. It provides all the functionality provided
1706-- by any of the list transformer and logic programming packages listed above.
1707-- In addition, Streamly naturally integrates the concurrency dimension to the
1708-- basic list transformer functionality.
1710-- When it comes to streaming, in terms of the streaming API streamly is almost
1711-- identical to the vector package. Streamly, vector and streaming packages all
1712-- represent a stream as data and are therefore similar in the fundamental
1713-- approach to streaming. The fundamental difference is that streamly adds
1714-- concurrency support and the monad instance provides concurrent looping.
1715-- Other streaming libraries like pipes, conduit and machines represent and
1716-- compose stream processors rather than the stream data and therefore fall in
1717-- another class of streaming libraries and have comparatively more complicated
1720-- When it comes to concurrency, streamly can do everything that the @async@
1721-- package can do and more. async provides applicative concurrency whereas
1722-- streamly provides both applicative and monadic concurrency. The
1723-- 'ZipAsync' type behaves like the applicative instance of async. In
1724-- comparison to transient streamly has a first class streaming interface and
1725-- is a monad transformer that can be used universally in any Haskell monad
1726-- transformer stack. Streamly was in fact originally inspired by the
1727-- concurrency implementation in @transient@ though it has no resemblance with
1728-- that and takes a lazy pull approach versus transient's strict push approach.
1730-- The non-determinism, concurrency and streaming combination make streamly a
1731-- strong reactive programming library as well. Reactive programming is
1732-- fundamentally stream of events that can be processed concurrently. The
1733-- example in this tutorial as well as the
1734-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare>
1735-- example from Yampa demonstrate the basic reactive capability of streamly.
1736-- In core concepts streamly is strikingly similar to @dunai@. dunai was
1737-- designed from a FRP perspective and streamly was originally designed from a
1738-- concurrency perspective. However, both have similarity at the core.
1742-- * Read the documentation of "Streamly" module
1743-- * Read the documentation of "Streamly.Prelude" module
1744-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.
1745-- * 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