1{-# OPTIONS_GHC -fno-warn-unused-imports #-}
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
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))
181-- CAUTION: please keep setup and imports sections in sync
185-- >>> import Data.Function ((&))
186-- >>> import Streamly.Prelude ((|:), (|&))
187-- >>> import qualified Streamly.Prelude as Stream
188-- >>> import qualified Streamly.Data.Fold as Fold
190-- >>> import Control.Concurrent (threadDelay, myThreadId)
192-- delay n = Stream.fromEffect $ do
193-- threadDelay (n * 1000000)
195-- putStrLn (show tid ++ ": Delay " ++ show n)
198-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
199-- >>> hSetBuffering stdout LineBuffering
204-- In most of example snippets we do not repeat the imports. Where imports are
205-- not explicitly specified use the imports shown below.
208-- >>> import Data.Function ((&))
209-- >>> import Streamly.Prelude ((|:), (|&))
210-- >>> import qualified Streamly.Prelude as Stream
211-- >>> import qualified Streamly.Data.Fold as Fold
213-- To illustrate concurrent vs serial composition aspects, we will use the
214-- following @delay@ function to introduce a sleep or delay specified in
215-- seconds. After the delay it prints the number of seconds it slept.
217-- >>> import Control.Concurrent (threadDelay, myThreadId)
219-- delay n = Stream.fromEffect $ do
220-- threadDelay (n * 1000000)
222-- putStrLn (show tid ++ ": Delay " ++ show n)
225-- For concurrent examples, use line buffering, otherwise output from different
226-- threads may get mixed:
228-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
229-- >>> hSetBuffering stdout LineBuffering
234-- The monadic stream API offered by Streamly is very close to the Haskell
235-- "Prelude" pure lists' API, it can be considered as a natural extension of
236-- lists to monadic actions. Streamly streams provide concurrent composition
237-- and merging of streams. It can be considered as a concurrent list
240-- The basic stream type is 'Serial', it represents a sequence of IO actions,
241-- and is a 'Monad'. The 'Serial' monad is almost a drop in replacement for
242-- the 'IO' monad, IO monad is a special case of the 'Serial' monad; IO monad
243-- represents a single IO action whereas the 'Serial' monad represents a series
244-- of IO actions. The only change you need to make to go from 'IO' to 'Serial'
245-- is to use 'drain' to run the monad and to prefix the IO actions with
246-- either 'fromEffect' or 'liftIO'. If you use liftIO you can switch from 'Serial'
247-- to IO monad by simply removing the 'drain' function; no other changes
248-- are needed unless you have used some stream specific composition or
251-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
252-- pure lists are a special case of monadic streams. If you use 'nil' in place
253-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' stream.
254-- The only difference is that the elements must be monadic type and to operate
255-- on the streams we must use the corresponding functions from
256-- "Streamly.Prelude" instead of using the base "Prelude".
260-- Many stream operations can be done concurrently:
262-- * Streams can be generated concurrently.
264-- * Streams can be merged concurrently.
266-- * Multiple stages in a streaming pipeline can run concurrently.
268-- * Streams can be mapped and zipped concurrently.
270-- * In monadic composition they combine like a list transformer,
271-- providing concurrent non-determinism.
273-- There are three basic concurrent stream styles, 'Ahead', 'Async', and
274-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that
275-- they can speculatively execute multiple stream actions concurrently in
276-- advance. 'Ahead' would return exactly the same stream as 'Serial' except
277-- that it may execute the actions concurrently. The 'Async' style streams,
278-- like 'Ahead', speculatively execute multiple stream actions in advance but
279-- return the results in their finishing order rather than in the stream
280-- traversal order. 'Parallel' is like 'Async' except that it provides
281-- unbounded parallelism instead of controlled parallelism.
283-- For easy reference, we can classify the stream types based on /execution order/,
284-- /consumption order/, and /bounded or unbounded/ concurrency.
285-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
286-- execution we execute the next action in the stream only after the previous
287-- one has finished executing. In asynchronous execution multiple actions in
288-- the stream can be executed asynchronously i.e. the next action can start
289-- executing even before the first one has finished. Consumption order
290-- determines the order in which the outputs generated by the composition are
291-- consumed. Consumption could be serial or asynchronous. In serial
292-- consumption, the outputs are consumed in the traversal order, in
293-- asynchronous consumption the outputs are consumed as they arrive i.e. first
294-- come first serve order.
296-- +------------+--------------+--------------+--------------+
297-- | Type | Execution | Consumption | Concurrency |
298-- +============+==============+==============+==============+
299-- | 'Serial' | Serial | Serial | None |
300-- +------------+--------------+--------------+--------------+
301-- | 'Ahead' | Asynchronous | Serial | bounded |
302-- +------------+--------------+--------------+--------------+
303-- | 'Async' | Asynchronous | Asynchronous | bounded |
304-- +------------+--------------+--------------+--------------+
305-- | 'Parallel' | Asynchronous | Asynchronous | unbounded |
306-- +------------+--------------+--------------+--------------+
308-- All these types can be freely inter-converted using type conversion
309-- combinators or type annotations, without any cost, to achieve the desired
310-- composition style. To force a particular type of composition, we coerce the
311-- stream type using the corresponding type adapting combinator from
312-- 'fromSerial', 'fromAhead', 'fromAsync', or 'fromParallel'. The default stream type
313-- is inferred as 'Serial' unless you change it by using one of the combinators
314-- or by using a type annotation.
318-- Streams can be combined using '<>' or 'mappend' to form a
319-- composite. Composite streams can be interpreted in a depth first or
320-- breadth first manner using an appropriate type conversion before
321-- consumption. Deep (e.g. 'Serial') stream type variants traverse a
322-- composite stream in a depth first manner, such that each stream is
323-- traversed fully before traversing the next stream. Wide
324-- (e.g. 'WSerial') stream types traverse it in a breadth first
325-- manner, such that one element from each stream is traversed before
326-- coming back to the first stream again.
328-- Each stream type has a wide traversal variant prefixed by 'W'. The wide
329-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad
330-- compositions of the streams.
331-- The following table summarizes the basic types and the corresponding wide
335-- +------------+-----------+
337-- +============+===========+
338-- | 'Serial' | 'WSerial' |
339-- +------------+-----------+
340-- | 'Ahead' | 'WAhead' |
341-- +------------+-----------+
342-- | 'Async' | 'WAsync' |
343-- +------------+-----------+
346-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
347-- zip streams serially or concurrently using 'Applicative' operation. These
348-- types are not monads they are only applicatives and they do not differ in
349-- 'Semigroup' composition.
354-- When writing concurrent programs it is advised to not use the concurrent
355-- style stream combinators blindly at the top level. That might create too
356-- much concurrency where it is not even required, and can even degrade
357-- performance in some cases. In some cases it can also lead to surprising
358-- behavior because of some code that is supposed to be serial becoming
359-- concurrent. Please be aware that all concurrency capable APIs that you may
360-- have used under the scope of a concurrent stream combinator will become
361-- concurrent. For example if you have a 'repeatM' somewhere in your program
362-- and you use 'fromParallel' on top, the 'repeatM' becomes fully parallel,
363-- resulting into an infinite parallel execution . Instead, use the
364-- /Keep It Serial and Stupid/ principle, start with the default serial
365-- composition and enable concurrent combinators only when and where necessary.
366-- When you use a concurrent combinator you can use an explicit 'fromSerial'
367-- combinator to suppress any unnecessary concurrency under the scope of that
372-- To represent streams in an arbitrary monad use the more general monad
373-- transformer types for example the monad transformer type corresponding to
374-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values
375-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a
376-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for
379-- Similarly we have monad transformer types for other stream types as well viz.
380-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
382-- To lift a value from an underlying monad in a monad transformer stack into a
383-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
385-- >>> import Control.Monad.IO.Class (liftIO)
386-- >>> Stream.drain $ liftIO $ putStrLn "Hello world!"
389-- >>> import Control.Monad.Trans.Class (MonadTrans(lift))
390-- >>> Stream.drain $ lift $ putStrLn "Hello world!"
396-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
397-- a monadic action at the head of the stream.
399-- >>> Stream.toList Stream.nil
402-- Stream.toList $ getLine |: getLine |: Stream.nil
407-- To create a singleton stream from a pure value use 'fromPure' or 'pure' and to
408-- create a singleton stream from a monadic action use 'fromEffect'. Note that in
409-- case of Zip applicative streams "pure" repeats the value to generate an
412-- >>> Stream.toList $ pure 1
415-- >>> Stream.toList $ Stream.fromPure 1
418-- Stream.toList $ Stream.fromEffect getLine
422-- To create a stream from pure values in a 'Foldable' container use
423-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
425-- >>> Stream.toList $ Stream.fromFoldable [1..3]
428-- >>> Stream.toList $ Prelude.foldr Stream.cons Stream.nil [1..3]
431-- To create a stream from monadic actions in a 'Foldable' container just use a
432-- right fold using 'consM' and 'nil':
434-- >>> Stream.drain $ Prelude.foldr (|:) Stream.nil [putStr "Hello ", putStrLn "world!"]
437-- For more ways to construct a stream see the module "Streamly.Prelude".
439-- $generatingConcurrently
441-- Monadic construction and generation functions like 'consM', 'unfoldrM',
442-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' work concurrently
443-- when used with appropriate stream type combinator. The pure versions of
444-- these APIs are not concurrent, however you can use the monadic versions even
445-- for pure computations by wrapping the pure value in a monad to get the
446-- concurrent generation capability where required.
448-- The following code finishes in 3 seconds (6 seconds when serial):
450-- >>> let p n = threadDelay (n * 1000000) >> return n
451-- >>> Stream.toList $ Stream.fromParallel $ p 3 |: p 2 |: p 1 |: Stream.nil
454-- >>> Stream.toList $ Stream.fromAhead $ p 3 |: p 2 |: p 1 |: Stream.nil
457-- The following finishes in 10 seconds (100 seconds when serial):
459-- >>> Stream.drain $ Stream.fromAsync $ Stream.replicateM 10 $ p 10
464-- We have already seen 'drain' and toList to eliminate a stream in the
465-- examples above. 'drain' runs a stream discarding the results i.e. only
466-- for effects. 'toList' runs the stream and collects the results in a list.
468-- For other ways to eliminate a stream see the @Folding@ section in
469-- "Streamly.Prelude" module.
473-- Transformation over a stream is the equivalent of a @for@ loop construct in
474-- imperative paradigm. We iterate over every element in the stream and perform
475-- certain transformations for each element. Transformations may involve
476-- mapping functions over the elements, filtering elements from the stream or
477-- folding all the elements in the stream into a single value. Streamly streams
478-- are exactly like lists and you can perform all the transformations in the
479-- same way as you would on lists.
481-- Here is a simple console echo program that just echoes every input line,
486-- Stream.repeatM getLine
487-- & Stream.mapM putStrLn
491-- The following code snippet reads lines from standard input, filters blank
492-- lines, drops the first non-blank line, takes the next two, up cases them,
493-- numbers them and prints them:
495-- >>> import Data.Char (toUpper)
498-- Stream.repeatM getLine
499-- & Stream.filter (not . null)
502-- & fmap (map toUpper)
503-- & Stream.zipWith (\n s -> show n ++ " " ++ s) (Stream.fromFoldable [1..])
504-- & Stream.mapM putStrLn
509-- $concurrentTransformation
511-- Monadic transformation functions 'mapM' and 'sequence' work concurrently
512-- when used with appropriate stream type combinators. The pure versions do not
513-- work concurrently, however you can use the monadic versions even for pure
514-- computations to get the concurrent transformation capability where required.
516-- This would print a value every second (2 seconds when serial):
518-- >>> let p n = threadDelay (n * 1000000) >> return n
521-- Stream.repeatM (p 1)
522-- & Stream.fromSerial -- repeatM is serial
523-- & Stream.mapM (\x -> p 1 >> print x)
524-- & Stream.fromAhead -- mapM is cocnurrent using Ahead style
529-- $concurrentApplication
531-- The concurrent function application operators '|$' and '|&' apply a stream
532-- argument to a stream function concurrently to compose a concurrent pipeline
533-- of stream processing functions:
535-- Because both the stages run concurrently, we would see a delay of only 1
536-- second instead of 2 seconds in the following:
538-- >>> let p n = threadDelay (n * 1000000) >> return n
541-- Stream.repeatM (p 1)
542-- |& Stream.mapM (\x -> p 1 >> print x)
548-- We can combine two streams into a single stream using semigroup composition
549-- operation '<>'. Streams can be combined in many different ways as described
550-- in the following sections, the '<>' operation behaves differently depending
551-- on the stream type in effect. The stream type and therefore the composition
552-- style can be changed at any point using one of the type combinators as
557-- The 'Semigroup' operation '<>' of the 'Serial' type combines the two streams
558-- in a /serial depth first/ manner. We use the 'fromSerial' type combinator to
559-- effect 'Serial' style of composition. We can also use an explicit 'Serial'
560-- type annotation for the stream to achieve the same effect. However, since
561-- 'Serial' is the default type unless explicitly specified by using a
562-- combinator, we can omit using an explicit combinator or type annotation for
563-- this style of composition.
565-- When two streams with multiple elements are combined in this manner, the
566-- monadic actions in the two streams are performed sequentially i.e. first all
567-- actions in the first stream are performed sequentially and then all actions
568-- in the second stream are performed sequentially. We call it
569-- /serial depth first/ as the full depth of one stream is fully traversed
570-- before we move to the next. The following example prints the sequence 1, 2,
573-- >>> stream1 = print 1 |: print 2 |: Stream.nil
574-- >>> stream2 = print 3 |: print 4 |: Stream.nil
575-- >>> Stream.drain $ stream1 <> stream2
581-- All actions in both the streams are performed serially in the same thread.
582-- In the following example we can see that all actions are performed in the
583-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
585-- >>> Stream.drain $ delay 3 <> delay 2 <> delay 1
586-- ThreadId ... Delay 3
587-- ThreadId ... Delay 2
588-- ThreadId ... Delay 1
590-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
591-- 'serial'. We can use 'serial' to join streams in a sequential manner
592-- irrespective of the type of stream:
594-- >>> Stream.drain $ stream1 `Stream.serial` stream2
603-- The 'Semigroup' operation '<>' of the 'WSerial' type combines the two
604-- streams in a /serial breadth first/ manner. We use the fromWSerial type
605-- combinator to effect 'WSerial' style of composition. We can also use the
606-- 'WSerial' type annotation for the stream to achieve the same effect.
608-- When two streams with multiple elements are combined in this manner, we
609-- traverse all the streams in a breadth first manner i.e. one action from each
610-- stream is performed and yielded to the resulting stream before we come back
611-- to the first stream again and so on.
612-- The following example prints the sequence 1, 3, 2, 4
614-- >>> stream1 = print 1 |: print 2 |: Stream.nil
615-- >>> stream2 = print 3 |: print 4 |: Stream.nil
616-- >>> Stream.drain $ Stream.fromWSerial $ stream1 <> stream2
622-- Even though the monadic actions of the two streams are performed in an
623-- interleaved manner they are all performed serially in the same thread. In
624-- the following example we can see that all actions are performed in the same
625-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
627-- >>> Stream.drain $ Stream.fromWSerial $ delay 3 <> delay 2 <> delay 1
628-- ThreadId ...: Delay 3
629-- ThreadId ...: Delay 2
630-- ThreadId ...: Delay 1
632-- The polymorphic version of the 'WSerial' binary operation '<>' is called
633-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
634-- irrespective of the type, notice that we have not used the fromWSerial
635-- combinator in the following example:
637-- >>> Stream.drain $ stream1 `Stream.wSerial` stream2
643-- Note that this composition cannot be used to fold infinite number of streams
644-- since it requires preserving the state until a stream is finished.
648-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a
649-- /serial depth first/ manner with concurrent lookahead. We use the 'fromAhead'
650-- type combinator to effect 'Ahead' style of composition. We can also use an
651-- explicit 'Ahead' type annotation for the stream to achieve the same effect.
653-- When two streams are combined in this manner, the streams are traversed in
654-- depth first manner just like 'Serial', however it can execute the next
655-- stream concurrently and keep the results ready when its turn arrives.
656-- Concurrent execution of the next stream(s) is performed if the first stream
657-- blocks or if it cannot produce output at the rate that is enough to meet the
658-- consumer demand. Multiple streams can be executed concurrently to meet the
659-- demand. The following example would print the result in a second even
660-- though each action in each stream takes one second:
662-- >>> p n = threadDelay 1000000 >> return n
663-- >>> stream1 = p 1 |: p 2 |: Stream.nil
664-- >>> stream2 = p 3 |: p 4 |: Stream.nil
665-- >>> Stream.toList $ Stream.fromAhead $ stream1 <> stream2
668-- Each stream is constructed 'fromAhead' and then both the streams are merged
669-- 'fromAhead', 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-- 'fromAsync' 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:
695-- >>> p n = threadDelay 1000000 >> return n
696-- >>> stream1 = p 1 |: p 2 |: Stream.nil
697-- >>> stream2 = p 3 |: p 4 |: Stream.nil
698-- >>> Stream.toList $ Stream.fromAsync $ stream1 <> stream2
701-- The constituent streams are also composed in 'Async' manner and the
702-- composition of streams too. We can compose the constituent streams to run
703-- serially, in that case it would take 2 seconds to produce all the results.
704-- The elements in the serial streams would be in serial order in the results:
706-- >>> p n = threadDelay 1000000 >> return n
707-- >>> stream = (Stream.fromSerial stream1) <> (Stream.fromSerial stream2)
708-- >>> Stream.toList $ Stream.fromAsync stream
711-- In the following example we can see that new threads are started when a
712-- computation blocks. Notice that the output from the stream with the
713-- shortest delay is printed first. The whole computation takes @maximum of
714-- (3, 2, 1) = 3@ seconds:
716-- >>> Stream.drain $ Stream.fromAsync $ delay 3 <> delay 2 <> delay 1
717-- ThreadId ...: Delay 1
718-- ThreadId ...: Delay 2
719-- ThreadId ...: Delay 3
721-- When we have a tree of computations composed using this style, the tree is
722-- traversed in DFS style just like the 'Serial' style, the only difference is
723-- that here we can move on to executing the next stream if a stream blocks.
724-- However, we will not start new threads if we have sufficient output to
725-- saturate the consumer. This is why we call it left-biased demand driven or
726-- adaptive concurrency style, the concurrency tends to stay on the left side
727-- of the composition as long as possible. More threads are started based on
728-- the pull rate of the consumer. The following example prints an output every
729-- second as all of the actions are concurrent.
731-- >>> Stream.drain $ Stream.fromAsync $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
732-- ThreadId ...: Delay 1
733-- ThreadId ...: Delay 2
734-- ThreadId ...: Delay 3
735-- ThreadId ...: Delay 4
737-- All the computations may even run in a single thread when more threads are
738-- not needed. As you can see, in the following example the computations are
739-- run in a single thread one after another, because none of them blocks.
740-- However, if the thread consuming the stream were faster than the producer
741-- then it would have started parallel threads for each computation to keep up
742-- even if none of them blocks:
745-- traced m = Stream.fromEffect (myThreadId >>= print) >> return m
746-- stream = traced (sqrt 9) <> traced (sqrt 16) <> traced (sqrt 25)
747-- main = Stream.drain $ Stream.fromAsync stream
750-- Note that the order of printing in the above examples may change due to
751-- variations in scheduling latencies for concurrent threads.
753-- The polymorphic version of the 'Async' binary operation '<>' is called
754-- 'async'. We can use 'async' to join streams in a left biased
755-- adaptively concurrent manner irrespective of the type, notice that we have
756-- not used the 'fromAsync' combinator in the following example:
758-- >>> Stream.drain $ delay 3 `Stream.async` delay 2 `Stream.async` delay 1
759-- ThreadId ...: Delay 1
760-- ThreadId ...: Delay 2
761-- ThreadId ...: Delay 3
763-- Since the concurrency provided by this operator is demand driven it cannot
764-- be used when the composed computations start timers that are relative to
765-- each other because all computations may not be started at the same time and
766-- therefore timers in all of them may not start at the same time. When
767-- relative timing among all computations is important or when we need to start
768-- all computations at once for any reason 'Parallel' style must be used
771-- 'Async' style utilizes resources optimally and should be preferred over
772-- 'Parallel' or 'WAsync' unless you really need those. 'Async' should be used
773-- when we know that the computations can run in parallel but we do not care if
774-- they actually run in parallel or not, that decision can be left to the
775-- scheduler based on demand. Also, note that 'async' operator can be used to fold
776-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
777-- because it does not require us to run all of them at the same time in a fair
782-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
783-- a concurrent manner using /breadth first traversal/. We use the 'fromWAsync'
784-- type combinator to effect 'WAsync' style of composition. We can also use the
785-- 'WAsync' type annotation for the stream to achieve the same effect.
787-- When streams with multiple elements are combined in this manner, we traverse
788-- all the streams concurrently in a breadth first manner i.e. one action from
789-- each stream is performed and yielded to the resulting stream before we come
790-- back to the first stream again and so on. Even though we execute the actions
791-- in a breadth first order the outputs are consumed on a first come first
794-- In the following example we can see that outputs are produced in the breadth
795-- first traversal order but this is not guaranteed.
797-- >>> stream1 = print 1 |: print 2 |: Stream.nil
798-- >>> stream2 = print 3 |: print 4 |: Stream.nil
799-- >>> Stream.drain $ Stream.fromWAsync $ stream1 <> stream2
805-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
806-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
807-- concurrent traversal irrespective of the type, notice that we have not used
808-- the 'fromWAsync' combinator in the following example:
810-- >>> Stream.drain $ delay 3 `Stream.wAsync` delay 2 `Stream.wAsync` delay 1
811-- ThreadId ...: Delay 1
812-- ThreadId ...: Delay 2
813-- ThreadId ...: Delay 3
815-- Since the concurrency provided by this style is demand driven it may not
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
826-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
827-- streams in a fairly concurrent manner with round robin scheduling. We use
828-- the 'fromParallel' type combinator to effect 'Parallel' style of composition.
829-- We can also use the 'Parallel' type annotation for the stream type to
830-- achieve the same effect.
832-- When two streams with multiple elements are combined in this manner, the
833-- monadic actions in both the streams are performed concurrently with a fair
834-- round robin scheduling. The outputs are yielded in the order in which the
835-- actions complete. This is pretty similar to the 'WAsync' type, the
836-- difference is that 'WAsync' is adaptive to the consumer demand and may or
837-- may not execute all actions in parallel depending on the demand, whereas
838-- 'Parallel' runs all the streams in parallel irrespective of the demand.
840-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
841-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
842-- manner irrespective of the type, notice that we have not used the
843-- 'fromParallel' combinator in the following example:
845-- >>> Stream.drain $ delay 3 `Stream.parallel` delay 2 `Stream.wAsync` delay 1
846-- ThreadId ...: Delay 1
847-- ThreadId ...: Delay 2
848-- ThreadId ...: Delay 3
850-- Note that this style of composition cannot be used to combine infinite
851-- number of streams, as it will lead to an infinite sized scheduling queue.
857-- The 'mkAsync' API can be used to create references to asynchronously running
858-- stream computations. We can then use 'uncons' to explore the streams
859-- arbitrarily and then recompose individual elements to create a new stream.
860-- This way we can dynamically decide which stream to explore at any given
861-- time. Take an example of a merge sort of two sorted streams. We need to
862-- keep consuming items from the stream which has the lowest item in the sort
863-- order. This can be achieved using async references to streams. See
864-- "MergeSort.hs" in <https://github.com/composewell/streamly-examples Streamly Examples>.
868-- We can use 'Monoid' instances to fold a container of streams in the desired
869-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
870-- to fold streams using the polymorphic combine operations:
872-- * 'concatFoldableWith' is like 'fold', it folds a 'Foldable' container of
873-- streams using the given composition operator.
874-- * 'concatMapFoldableWith' is like 'foldMap', it folds like
875-- @concatFoldableWith@ but also maps a function before folding.
876-- * 'concatForFoldableWith' is like @concatMapFoldableWith@ but the container
877-- argument comes before the function argument.
879-- All of the following are equivalent and start ten concurrent tasks each with
880-- a delay from 1 to 10 seconds, resulting in the printing of each number every
885-- Stream.drain $ Stream.fromAsync $ foldMap delay [1..10]
886-- Stream.drain $ Stream.concatFoldableWith Stream.async (map delay [1..10])
887-- Stream.drain $ Stream.concatMapFoldableWith Stream.async delay [1..10]
888-- Stream.drain $ Stream.concatForFoldableWith Stream.async [1..10] delay
894-- Till now we discussed ways to apply transformations on a stream or to merge
895-- streams together to create another stream. We mentioned earlier that
896-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
897-- We will now discuss the concept of a nested composition of streams which is
898-- analogous to nested @for@ loops in the imperative paradigm. Functional
899-- programmers call this style of composition a list transformer or @ListT@.
900-- Logic programmers call it a logic monad or non-deterministic composition,
901-- but for ordinary imperative minded people like me it is easier to think in
902-- terms of good old nested @for@ loops.
906-- In functional programmer's parlance the 'Monad' instances of different
907-- 'IsStream' types implement non-determinism, exploring all possible
908-- combination of choices from both the streams. From an imperative
909-- programmer's point of view it behaves like nested loops i.e. for each
910-- element in the first stream and for each element in the second stream
911-- execute the body of the loop.
913-- The 'Monad' instances of 'Serial', 'WSerial', 'Async' and 'WAsync'
914-- stream types support different flavors of nested looping. In other words,
915-- they are all variants of list transformer. The nesting behavior of these
916-- types correspond exactly to the way they merge streams as we discussed in
917-- the previous section.
922-- The 'Monad' composition of the 'Serial' type behaves like a standard list
923-- transformer. This is the default when we do not use an explicit type
924-- combinator. However, the 'fromSerial' type combinator can be used to switch to
925-- this style of composition. We will see how this style of composition works
926-- in the following examples.
928-- Let's start with an example with a simple @for@ loop without any nesting.
929-- For simplicity of illustration we are using streams of pure values in all
930-- the examples. However, the streams could also be made of monadic actions
935-- x <- Stream.fromFoldable [3,2,1]
938-- ThreadId ...: Delay 3
939-- ThreadId ...: Delay 2
940-- ThreadId ...: Delay 1
942-- As we can see, the code after the @fromFoldable@ statement is run three
943-- times, once for each value of @x@ drawn from the stream. All the three
944-- iterations are serial and run in the same thread one after another. In
945-- imperative terms this is equivalent to a @for@ loop with three iterations.
947-- We can write the console echo program that we wrote earlier using the monad
953-- x <- Stream.repeatM getLine
954-- Stream.fromEffect $ putStrLn x
957-- When multiple streams are composed using this style they nest in a DFS
962-- x <- Stream.fromFoldable [1,2]
963-- y <- Stream.fromFoldable [3,4]
964-- Stream.fromEffect $ putStrLn $ show (x, y)
971-- i.e. inner loop iterations ((1,3), (1,4)) are executed before we proceed to
972-- the next iteration of the outer loop ((2,3), (2,4)). This behaves just like
973-- nested @for@ loops in imperative programming.
975-- Notice that this is analogous to merging streams of type 'Serial' or merging
976-- streams using 'serial'.
980-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except
981-- that it can speculatively perform a bounded number of next iterations of a
985-- Stream.toList $ Stream.fromAhead $ do
986-- x <- Stream.fromFoldable [3,2,1]
990-- ThreadId ...: Delay 1
991-- ThreadId ...: Delay 2
992-- ThreadId ...: Delay 3
995-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can
996-- see all the three iterations are concurrent and run in different threads,
997-- however, the results are returned in the serial order.
999-- Concurrency is demand driven, when multiple streams are composed using this
1000-- style, the iterations are executed in a depth first manner just like
1001-- 'Serial' i.e. nested iterations are executed before we proceed to the next
1002-- outer iteration. The only difference is that we may execute multiple future
1003-- iterations concurrently and keep the results ready.
1005-- The 'fromAhead' type combinator can be used to switch to this style of
1006-- composition. Alternatively, a type annotation can be used to specify the
1007-- type of the stream as 'Ahead'.
1010-- $concurrentNesting
1012-- The 'Monad' composition of 'Async' type can perform the iterations of a
1013-- loop concurrently.
1016-- Stream.drain $ Stream.fromAsync $ do
1017-- x <- Stream.fromFoldable [3,2,1]
1020-- ThreadId ...: Delay 1
1021-- ThreadId ...: Delay 2
1022-- ThreadId ...: Delay 3
1024-- As we can see the code after the @fromFoldable@ statement is run three
1025-- times, once for each value of @x@. All the three iterations are concurrent
1026-- and run in different threads. The iteration with least delay finishes first.
1027-- When compared to imperative programming, this can be viewed as a @for@ loop
1028-- with three concurrent iterations.
1030-- Concurrency is demand driven i.e. more concurrent iterations are started
1031-- only if the previous iterations are not able to saturate the consumer of the
1032-- output stream. This works exactly the same way as the merging of two
1033-- streams using 'async' works.
1035-- The 'fromAsync' type combinator can be used to switch to this style of
1036-- composition. Alternatively, a type annotation can be used to specify the
1037-- type of the stream as 'Async'.
1039-- When multiple streams are nested using this style, the iterations are
1040-- concurrently evaluated in a depth first manner:
1044-- Stream.drain $ Stream.fromAsync $ do
1045-- x <- Stream.fromFoldable [1,2]
1046-- y <- Stream.fromFoldable [3,4]
1047-- Stream.fromEffect $ putStrLn $ show (x, y)
1054-- Nested iterations are given preference for concurrent evaluation i.e.
1055-- (1,4) will be scheduled in preference to (2,3).
1057-- $interleavedNesting
1059-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
1060-- outer and inner loops in a nested loop composition.
1063-- Stream.drain $ Stream.fromWSerial $ do
1064-- x <- Stream.fromFoldable [1,2]
1065-- y <- Stream.fromFoldable [3,4]
1066-- Stream.fromEffect $ putStrLn $ show (x, y)
1073-- Note that (2,3) is preferred to (1,4). This works exactly the same way as
1074-- the merging of two streams using 'wSerial' works.
1076-- The fromWSerial type combinator can be used to switch to this style of
1077-- composition. Alternatively, a type annotation can be used to specify the
1078-- type of the stream as 'WSerial'.
1083-- Like 'Async', the 'Monad' composition of 'WAsync' runs the iterations of a
1084-- loop concurrently. It differs from 'Async' in the nested loop behavior. Like
1085-- 'WSerial', the nested loops in this type are traversed and executed in a
1086-- breadth first manner rather than the depth first manner of 'Async' style.
1089-- Stream.drain $ Stream.fromWAsync $ do
1090-- x <- Stream.fromSerial $ Stream.fromFoldable [1,2]
1091-- y <- Stream.fromSerial $ Stream.fromFoldable [3,4]
1092-- Stream.fromEffect $ putStrLn $ show (x, y)
1099-- Note that (2,3) is preferred to (1,4) when evaluating the iterations
1100-- concurrently. This works exactly the same way as the merging of two streams
1101-- using 'wAsync' works.
1103-- The 'fromWAsync' type combinator can be used to switch to this style of
1104-- composition. Alternatively, a type annotation can be used to specify the
1105-- type of the stream as 'WAsync'.
1110-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
1111-- iterations of a loop concurrently.
1114-- Stream.drain $ Stream.fromParallel $ do
1115-- x <- Stream.fromFoldable [3,2,1]
1118-- ThreadId ...: Delay 1
1119-- ThreadId ...: Delay 2
1120-- ThreadId ...: Delay 3
1122-- It differs from 'Async' and 'WAsync' in the nested loop behavior. All
1123-- iterations of the loop are run fully concurrently irrespective of the
1124-- demand. This works exactly the same way as the merging of streams using
1127-- The 'fromParallel' type combinator can be used to switch to this style of
1128-- composition. Alternatively, a type annotation can be used to specify the
1129-- type of the stream as 'Parallel'.
1134-- Streamly code is usually written in a way that is agnostic of the
1135-- specific monadic composition type. We use a polymorphic type with a
1136-- 'IsStream' type class constraint. When running the stream we can choose the
1137-- specific mode of composition. For example take a look at the following code.
1140-- composed :: (Stream.IsStream t, Monad (t IO)) => t IO ()
1145-- Stream.fromEffect $ putStrLn $ show (sz, cl, sh)
1147-- sizes = Stream.fromFoldable [1, 2, 3]
1148-- colors = Stream.fromFoldable ["red", "green", "blue"]
1149-- shapes = Stream.fromFoldable ["triangle", "square", "circle"]
1152-- Now we can interpret this in whatever way we want:
1155-- main = Stream.'drain' $ Stream.'fromSerial' $ composed
1156-- main = Stream.'drain' $ Stream.'fromWSerial' $ composed
1157-- main = Stream.'drain' $ Stream.'fromAsync' $ composed
1158-- main = Stream.'drain' $ Stream.'fromWAsync' $ composed
1159-- main = Stream.'drain' $ Stream.'fromParallel' $ composed
1162-- As an exercise try to figure out the output of this code for each mode of
1167-- 'fmap' transforms a stream by mapping a function on all elements of the
1168-- stream. 'fmap' behaves in the same way for all stream types, it is always
1171-- >>> (Stream.toList $ fmap show $ Stream.fromFoldable [1..10]) >>= print
1172-- ["1","2","3","4","5","6","7","8","9","10"]
1174-- Also see functions 'mapM' and 'sequence' from "Streamly.Prelude" module
1175-- which can map actions concurrently depending on the type of the input stream.
1179-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
1180-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are
1183-- The following is an example of 'Serial' applicative, it runs all iterations
1184-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
1186-- >>> d n = delay n >> return n
1187-- >>> s1 = d 1 <> d 2
1188-- >>> s2 = d 3 <> d 4
1190-- >>> (Stream.toList $ Stream.fromSerial $ (,) <$> s1 <*> s2) >>= print
1191-- ThreadId ...: Delay 1
1192-- ThreadId ...: Delay 3
1193-- ThreadId ...: Delay 4
1194-- ThreadId ...: Delay 2
1195-- ThreadId ...: Delay 3
1196-- ThreadId ...: Delay 4
1197-- [(1,3),(1,4),(2,3),(2,4)]
1199-- Similarly, 'WSerial' applicative runs the iterations in an interleaved
1200-- order but being serial it too takes a total of 17 seconds:
1202-- >>> (Stream.toList $ Stream.fromWSerial $ (,) <$> s1 <*> s2) >>= print
1203-- ThreadId ...: Delay 1
1204-- ThreadId ...: Delay 3
1205-- ThreadId ...: Delay 2
1206-- ThreadId ...: Delay 3
1207-- ThreadId ...: Delay 4
1208-- ThreadId ...: Delay 4
1209-- [(1,3),(2,3),(1,4),(2,4)]
1211-- 'Async' can run the iterations concurrently, therefore, it takes a total
1212-- of 6 seconds which is max (1, 2) + max (3, 4):
1214-- >>> (Stream.toList $ Stream.fromAsync $ (,) <$> s1 <*> s2) >>= print
1218-- ThreadId 34: Delay 1
1219-- ThreadId 36: Delay 2
1220-- ThreadId 35: Delay 3
1221-- ThreadId 36: Delay 3
1222-- ThreadId 35: Delay 4
1223-- ThreadId 36: Delay 4
1224-- [(1,3),(2,3),(1,4),(2,4)]
1227-- Similarly, 'WAsync' as well can run the iterations concurrently, but with a
1228-- different style of scheduling than 'Async' as explained in the Monad
1229-- section, therefore, it too takes a total of 6 seconds (2 + 4):
1231-- >>> (Stream.toList $ Stream.fromWAsync $ (,) <$> s1 <*> s2) >>= print
1235-- ThreadId 34: Delay 1
1236-- ThreadId 36: Delay 2
1237-- ThreadId 35: Delay 3
1238-- ThreadId 36: Delay 3
1239-- ThreadId 35: Delay 4
1240-- ThreadId 36: Delay 4
1241-- [(1,3),(2,3),(1,4),(2,4)]
1246-- Zipping is a special transformation where the corresponding elements of two
1247-- streams are combined together using a zip function producing a new stream of
1248-- outputs. Two different types are provided for serial and concurrent zipping.
1249-- These types provide an applicative instance that can be used to lift
1250-- functions to zip the argument streams.
1251-- Also see the zipping functions in the "Streamly.Prelude" module.
1255-- The applicative instance of 'ZipSerial' type zips streams serially.
1256-- 'fromZipSerial' type combinator can be used to switch to serial applicative
1259-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1260-- everything runs serially:
1262-- >>> d n = delay n >> return n
1263-- >>> s1 = Stream.fromSerial $ d 1 <> d 2
1264-- >>> s2 = Stream.fromSerial $ d 3 <> d 4
1265-- >>> (Stream.toList $ Stream.fromZipSerial $ (,) <$> s1 <*> s2) >>= print
1266-- ThreadId ...: Delay 1
1267-- ThreadId ...: Delay 3
1268-- ThreadId ...: Delay 2
1269-- ThreadId ...: Delay 4
1275-- The applicative instance of 'ZipAsync' type zips streams concurrently.
1276-- 'fromZipAsync' type combinator can be used to switch to parallel applicative
1279-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1280-- are produced concurrently, and 2 and 4 are produced concurrently:
1282-- >>> d n = delay n >> return n
1283-- >>> s1 = Stream.fromSerial $ d 1 <> d 2
1284-- >>> s2 = Stream.fromSerial $ d 3 <> d 4
1285-- >>> (Stream.toList $ Stream.fromZipAsync $ (,) <$> s1 <*> s2) >>= print
1286-- ThreadId ...: Delay 1
1287-- ThreadId ...: Delay 2
1288-- ThreadId ...: Delay 3
1289-- ThreadId ...: Delay 4
1295-- When writing concurrent programs there are two distinct places where the
1296-- programmer can control the concurrency. First, when /composing/ a stream by
1297-- merging multiple streams we can choose an appropriate sum style operators to
1298-- combine them concurrently or serially. Second, when /processing/ a stream in
1299-- a monadic composition we can choose one of the monad composition types to
1300-- choose the desired type of concurrency.
1302-- In the following example the squares of @x@ and @y@ are computed
1303-- concurrently using the 'async' operation and the square roots of their
1304-- sum are computed serially because of the 'fromSerial' combinator. We can
1305-- choose different combinators for the monadic processing and the stream
1306-- generation, to control the concurrency. We can also use the 'fromAsync'
1307-- combinator instead of explicitly folding with 'async'.
1309-- >>> import Data.List (sum)
1312-- z <- Stream.toList
1313-- $ Stream.fromSerial -- Serial monadic processing (sqrt below)
1315-- x2 <- Stream.concatForFoldableWith Stream.async [1..100] $ -- Concurrent @"for"@ loop
1316-- \x -> return $ x * x -- body of the loop
1317-- y2 <- Stream.concatForFoldableWith Stream.async [1..100] $
1318-- \y -> return $ y * y
1319-- return $ sqrt (x2 + y2)
1323-- We can see how this directly maps to the imperative style
1324-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1325-- and operators instead of the ugly pragmas.
1327-- For more concurrent programming examples see,
1328-- <https://github.com/composewell/streamly-examples>.
1332-- Reactive programming is nothing but concurrent streaming which is what
1333-- streamly is all about. With streamly we can generate streams of events,
1334-- merge streams that are generated concurrently and process events
1335-- concurrently. We can do all this without any knowledge about the specifics
1336-- of the implementation of concurrency. In the following example you will see
1337-- that the code is just regular Haskell code without much streamly APIs used
1338-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1341-- This application has two independent and concurrent sources of event
1342-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1343-- events that deteriorate the health of the character in the game.
1344-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1345-- health improves and the game continues.
1348-- {-\# LANGUAGE FlexibleContexts \#-}
1350-- import "Streamly.Prelude" (MonadAsync, SerialT)
1351-- import "Streamly.Prelude" as Stream
1352-- import Control.Monad (void)
1353-- import Control.Monad.IO.Class (MonadIO(liftIO))
1354-- import Control.Monad.State (MonadState, get, modify, runStateT)
1356-- data Event = Quit | Harm Int | Heal Int deriving (Show)
1358-- userAction :: MonadAsync m => 'SerialT' m Event
1359-- userAction = Stream.'repeatM' $ liftIO askUser
1362-- command <- getLine
1364-- "potion" -> return (Heal 10)
1365-- "harm" -> return (Harm 10)
1366-- "quit" -> return Quit
1367-- _ -> putStrLn "Type potion or harm or quit" >> askUser
1369-- acidRain :: MonadAsync m => 'SerialT' m Event
1370-- acidRain = Stream.'fromAsync' $ Stream.'constRate' 1 $ Stream.'repeatM' $ liftIO $ return $ Harm 1
1372-- data Result = Check | Done
1374-- runEvents :: (MonadAsync m, MonadState Int m) => 'SerialT' m Result
1376-- event \<- userAction \`Stream.'parallel'` acidRain
1378-- Harm n -> modify (\\h -> h - n) >> return Check
1379-- Heal n -> modify (\\h -> h + n) >> return Check
1380-- Quit -> return Done
1382-- data Status = Alive | GameOver deriving Eq
1384-- getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status
1385-- getStatus result =
1387-- Done -> liftIO $ putStrLn "You quit!" >> return GameOver
1390-- liftIO $ if (h <= 0)
1391-- then putStrLn "You die!" >> return GameOver
1392-- else putStrLn ("Health = " <> show h) >> return Alive
1396-- putStrLn "Your health is deteriorating due to acid rain, type \\\"potion\\\" or \\\"quit\\\""
1397-- let runGame = Stream.'drainWhile' (== Alive) $ Stream.'mapM' getStatus runEvents
1398-- void $ runStateT runGame 60
1401-- You can also find the source of this example in the streamly-examples repo
1402-- as <https://github.com/composewell/streamly-examples/tree/master/AcidRain.hs AcidRain.hs>.
1403-- It has been adapted from Gabriel's
1404-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1406-- This is much simpler compared to the pipes version because of the builtin
1407-- concurrency in streamly. You can also find a SDL based reactive programming
1408-- example adapted from Yampa in
1409-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare.hs>.
1413-- Streamly is highly optimized for performance, it is designed for serious
1414-- high performing, concurrent and scalable applications. We have created the
1415-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1416-- package which is specifically and carefully designed to measure the
1417-- performance of Haskell streaming libraries fairly and squarely in the right
1418-- way. Streamly performs at par or even better than most streaming libraries
1419-- for serial operations even though it needs to deal with the concurrency
1424-- List transformers and logic programming monads also provide a product style
1425-- composition similar to streamly, however streamly generalizes it with the
1426-- time dimension; allowing streams to be composed in an asynchronous and
1427-- concurrent fashion in many different ways. It also provides multiple
1428-- alternative ways of composing streams e.g. serial, interleaved or
1431-- This seemingly simple addition of asynchronicity and concurrency to product
1432-- style streaming composition unifies a number of disparate abstractions into
1433-- one powerful, concise and elegant abstraction. A wide variety of
1434-- programming problems can be solved elegantly with this abstraction. In
1435-- particular, it unifies three major programming domains namely
1436-- non-deterministic (logic) programming, concurrent programming and functional
1437-- reactive programming. In other words, you can do everything with this one
1438-- abstraction that you could do with the popular libraries listed under these
1439-- categories in the list below.
1442-- +-----------------+----------------+
1443-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1444-- | +----------------+
1445-- | | <https://hackage.haskell.org/package/list-t list-t> |
1446-- | +----------------+
1447-- | | <https://hackage.haskell.org/package/logict logict> |
1448-- +-----------------+----------------+
1449-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1450-- | +----------------+
1451-- | | <https://hackage.haskell.org/package/streaming streaming> |
1452-- | +----------------+
1453-- | | <https://hackage.haskell.org/package/pipes pipes> |
1454-- | +----------------+
1455-- | | <https://hackage.haskell.org/package/conduit conduit> |
1456-- +-----------------+----------------+
1457-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1458-- | +----------------+
1459-- | | <https://hackage.haskell.org/package/transient transient> |
1460-- +-----------------+----------------+
1461-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1462-- | +----------------+
1463-- | | <https://hackage.haskell.org/package/dunai dunai> |
1464-- | +----------------+
1465-- | | <https://hackage.haskell.org/package/reflex reflex> |
1466-- +-----------------+----------------+
1469-- Streamly is a list-transformer. It provides all the functionality provided
1470-- by any of the list transformer and logic programming packages listed above.
1471-- In addition, Streamly naturally integrates the concurrency dimension to the
1472-- basic list transformer functionality.
1474-- When it comes to streaming, in terms of the streaming API streamly is almost
1475-- identical to the vector package. Streamly, vector and streaming packages all
1476-- represent a stream as data and are therefore similar in the fundamental
1477-- approach to streaming. The fundamental difference is that streamly adds
1478-- concurrency support and the monad instance provides concurrent looping.
1479-- Other streaming libraries like pipes, conduit and machines represent and
1480-- compose stream processors rather than the stream data and therefore fall in
1481-- another class of streaming libraries and have comparatively more complicated
1484-- When it comes to concurrency, streamly can do everything that the @async@
1485-- package can do and more. async provides applicative concurrency whereas
1486-- streamly provides both applicative and monadic concurrency. The
1487-- 'ZipAsync' type behaves like the applicative instance of async. In
1488-- comparison to transient streamly has a first class streaming interface and
1489-- is a monad transformer that can be used universally in any Haskell monad
1490-- transformer stack. Streamly was in fact originally inspired by the
1491-- concurrency implementation in @transient@ though it has no resemblance with
1492-- that and takes a lazy pull approach versus transient's strict push approach.
1494-- The non-determinism, concurrency and streaming combination make streamly a
1495-- strong reactive programming library as well. Reactive programming is
1496-- fundamentally stream of events that can be processed concurrently. The
1497-- example in this tutorial as well as the
1498-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare>
1499-- example from Yampa demonstrate the basic reactive capability of streamly.
1500-- In core concepts streamly is strikingly similar to @dunai@. dunai was
1501-- designed from a FRP perspective and streamly was originally designed from a
1502-- concurrency perspective. However, both have similarity at the core.
1506-- * Read the documentation of "Streamly" module
1507-- * Read the documentation of "Streamly.Prelude" module
1508-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.
1509-- * 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