1{-# OPTIONS_GHC -fno-warn-unused-imports #-}
3-- Module : Streamly.Tutorial
4-- Copyright : (c) 2017 Harendra Kumar
7-- Maintainer : harendra.kumar@gmail.com
9-- Streamly is a general computing framework based on streaming IO. The IO
10-- monad and pure lists are a special case of streamly. On one hand, streamly
11-- extends the lists of pure values to lists of monadic actions, on the other
12-- hand it extends the IO monad with concurrrent non-determinism. In simple
13-- imperative terms we can say that streamly extends the IO monad with @for@
14-- loops and nested @for@ loops with concurrency support. You can understand
15-- this analogy better once you can go through this tutorial.
17-- Streaming in general enables writing modular, composable and scalable
18-- applications with ease, and concurrency allows you to make them scale and
19-- perform well. Streamly enables writing scalable concurrent applications
20-- without being aware of threads or synchronization. No explicit thread
21-- control is needed, where applicable the concurrency rate is automatically
22-- controlled based on the demand by the consumer. However, combinators can be
23-- used to fine tune the concurrency control.
25-- Streaming and concurrency together enable expressing reactive applications
26-- conveniently. See the @CirclingSquare@ example in the examples directory for
27-- a simple SDL based FRP example. To summarize, streamly provides a unified
28-- computing framework for streaming, non-determinism and functional reactive
29-- programming in an elegant and simple API that is a natural extension of pure
30-- lists to monadic streams.
32-- In this tutorial we will go over the basic concepts and how to
33-- use the library. See the last section for further reading resources.
35module Streamly.Tutorial
40 -- * Flavors of Streams
43 -- * Generating Streams
46 -- * Eliminating Streams
49 -- * Transforming Streams
57 -- *** Deep Serial Composition ('Serial')
60 -- *** Wide Serial Composition ('WSerial')
63 -- *** Deep Asynchronous Composition ('Async')
66 -- *** Wide Asynchronous Composition ('WAsync')
69 -- *** Parallel Asynchronous Composition ('Parallel')
72 -- *** Custom composition
84 -- *** Deep Serial Nesting ('Serial')
87 -- *** Wide Serial Nesting ('WSerial')
88 -- $interleavedNesting
90 -- *** Deep Asynchronous Nesting ('Async')
93 -- *** Wide Asynchronous Nesting ('WAsync')
96 -- *** Parallel Asynchronous Nesting ('Parallel')
114 -- ** Parallel Zipping
117 -- * Monad transformers
118 -- $monadtransformers
120 -- * Concurrent Programming
123 -- * Reactive Programming
129 -- * Interoperation with Streaming Libraries
132 -- * Comparison with Existing Packages
135 -- * Where to go next?
141import Streamly.Prelude
143import Control.Applicative
145import Control.Monad.IO.Class (MonadIO(..))
146import Control.Monad.Trans.Class (MonadTrans (lift))
150-- The way a list represents a sequence of pure values, a stream represents a
151-- sequence of monadic actions. The monadic stream API offered by Streamly is
152-- very close to the Haskell "Prelude" pure lists' API, it can be considered as a
153-- natural extension of lists to monadic actions. Streamly streams provide
154-- concurrent composition and merging of streams. It can be considered as a
155-- concurrent list transformer. In contrast to the "Prelude" lists, merging or
156-- appending streams of arbitrary length is scalable and inexpensive.
158-- The basic stream type is 'Serial', it represents a sequence of IO actions,
159-- and is a 'Monad'. The 'Serial' monad is almost a drop in replacement for
160-- the 'IO' monad, IO monad is a special case of the 'Serial' monad; IO monad
161-- represents a single IO action whereas the 'Serial' monad represents a series
162-- of IO actions. The only change you need to make to go from 'IO' to 'Serial'
163-- is to use 'runStream' to run the monad and to prefix the IO actions with
164-- either 'once' or 'liftIO'. If you use liftIO you can switch from 'Serial'
165-- to IO monad by simply removing the 'runStream' function; no other changes
166-- are needed unless you have used some stream specific composition or
169-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
170-- pure lists are a special case of monadic streams. If you use 'nil' in place
171-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' The
172-- only difference is that the elements must be monadic type and to operate on
173-- the streams we must use the corresponding functions from "Streamly.Prelude"
174-- instead of using the base "Prelude".
178-- There are a few more types similar to 'Serial' that all represent streams
179-- and differ only in the 'Semigroup', 'Applicative' and 'Monad' compositions
182-- The composition of two or more streams is distinguished based on three
183-- characterstics, /traversal order/, /execution order/ and
184-- /consumption order/. Traversal of a composition of streams could be @deep@
185-- or @wide@. Deep goes depth first i.e. each stream is traversed fully
186-- before we traverse the next stream. Wide goes breadth first i.e. one element
187-- from each stream is traversed before coming back to the first stream again.
188-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
189-- execution we execute an action in the next stream only after the first has
190-- finished executing. In asynchronous execution actions in both streams can be
191-- executed asynchronously i.e. the next action can start executing even before
192-- the first one has finished. The third parameter is consumption order that is
193-- in what order the output generated by the composition is consumed.
194-- Consumption could be serial or asynchronous. In serial consumption, the
195-- outputs are consumed in the traversal order, in asynchronous consumption the
196-- outputs are consumed as they arrive i.e. first come first serve order.
198-- The following table summarizes different styles of streams based on how they
199-- compose. All these types are monads and they differ in 'Semigroup',
200-- 'Applicative' and 'Monad' compositions:
203-- +------------+-----------+--------------+--------------+
204-- | Type | Traversal | Execution | Consumption |
205-- +============+===========+==============+==============+
206-- | 'Serial' | Deep | Serial | Serial |
207-- +------------+-----------+--------------+--------------+
208-- | 'WSerial' | Wide | Serial | Serial |
209-- +------------+-----------+--------------+--------------+
210-- | 'Async' | Deep | Asynchronous | Asynchronous |
211-- +------------+-----------+--------------+--------------+
212-- | 'WAsync' | Wide | Asynchronous | Asynchronous |
213-- +------------+-----------+--------------+--------------+
214-- | 'Parallel' | Parallel | Asynchronous | Asynchronous |
215-- +------------+-----------+--------------+--------------+
218-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
219-- zip streams serially or concurrently using 'Applicative' operation. These
220-- types are not monads they are only applicatives and they do not differ in
221-- 'Semigroup' composition.
223-- All these types can be freely inter-converted using type conversion
224-- combinators or type annotations without any cost, to acheive the desired
225-- composition style. To force a particular type of composition we coerce the
226-- stream type using the corresponding type adapting combinator from
227-- 'serially', 'wSerially', 'asyncly', 'wAsyncly', 'parallely', 'zipSerially'
228-- or 'zipAsyncly'. The default stream type is inferred as 'Serial' unless you
229-- change it by using one of the combinators or using a type annotation.
233-- To represent streams in an arbitrary monad use the more general monad
234-- transformer types for example the monad transformer type corresponding to
235-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values
236-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a
237-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for
240-- Similarly we have monad transformer types for other stream types as well viz.
241-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
243-- To lift a value from an underlying monad in a monad transformer stack into a
244-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
247-- > 'runStream' $ liftIO $ putStrLn "Hello world!"
249-- > 'runStream' $ lift $ putStrLn "Hello world!"
256-- We will assume the following imports in this tutorial. Go ahead, fire up a
257-- GHCi session and import these lines to start playing.
260-- > import "Streamly"
261-- > import "Streamly.Prelude" (|:)
262-- > import qualified "Streamly.Prelude" as S
265-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
266-- a monadic action at the head of the stream.
269-- > S.'toList' S.'nil'
271-- > S.'toList' $ 'getLine' |: 'getLine' |: S.'nil'
277-- To create a singleton stream from a pure value use 'pure' and to create a
278-- singleton stream from a monadic action use 'once'.
281-- > S.'toList' $ 'pure' 1
283-- > S.'toList' $ S.'once' 'getLine'
288-- To create a stream from pure values in a 'Foldable' container use
289-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
292-- > S.'toList' $ S.'fromFoldable' [1..3]
294-- > S.'toList' $ 'Prelude.foldr' S.'cons' S.'nil' [1..3]
298-- To create a stream from monadic actions in a 'Foldable' container just use a
299-- right fold using 'consM' and 'nil':
302-- > 'runStream' $ 'Prelude.foldr' ('|:') S.'nil' ['putStr' "Hello ", 'putStrLn' "world!"]
306-- For more ways to construct a stream see the module "Streamly.Prelude".
310-- We have already seen 'runStream' and 'toList' to eliminate a stream in the
311-- examples above. 'runStream' runs a stream discarding the results i.e. only
312-- for effects. 'toList' runs the stream and collects the results in a list.
314-- For other ways to eliminate a stream see the @Folding@ section in
315-- "Streamly.Prelude" module.
319-- Transformation over a stream is the equivalent of a @for@ loop construct in
320-- imperative paradigm. We iterate over every element in the stream and perform
321-- certain transformations for each element. Transformations may involve
322-- mapping functions over the elements, filtering elements from the stream or
323-- folding all the elements in the stream into a single value. Streamly streams
324-- are exactly like lists and you can perform all the transformations in the
325-- same way as you would on lists.
327-- Here is a simple console echo program that just echoes every input line,
331-- > 'runStream' $ S.'repeatM' getLine & S.'mapM' putStrLn
334-- The following code snippet reads lines from standard input, filters blank
335-- lines, drops the first non-blank line, takes the next two, up cases them,
336-- numbers them and prints them:
340-- import qualified "Streamly.Prelude" as S
341-- import Data.Char (toUpper)
342-- import Data.Function ((&))
344-- main = 'runStream' $
345-- S.'repeatM' getLine
346-- & S.'filter' (not . null)
349-- & fmap (map toUpper)
350-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
351-- & S.'mapM' putStrLn
356-- We can combine two streams into a single stream using semigroup composition
357-- operation '<>'. Streams can be combined in many different ways as described
358-- in the following sections, the '<>' operation behaves differently depending
359-- on the stream type in effect. The stream type and therefore the composition
360-- style can be changed at any point using one of the type combinators as
363-- To illustrate concurrent vs serial composition aspects, we will use the
364-- following @delay@ function to introduce a delay specified in seconds.
368-- import "Streamly.Prelude" (nil)
369-- import qualified "Streamly.Prelude" as S
370-- import Control.Concurrent
372-- delay n = S.'once' $ do
373-- threadDelay (n * 1000000)
375-- putStrLn (show tid ++ ": Delay " ++ show n)
380-- The 'Semigroup' operation '<>' of the 'Serial' type combines the two streams
381-- in a /serial depth first/ manner. We use the 'serially' type combinator to
382-- effect 'Serial' style of composition. We can also use an explicit 'Serial'
383-- type annotation for the stream to acheive the same effect. However, since
384-- 'Serial' is the default type unless explicitly specified by using a
385-- combinator, we can omit using an explicit combinator or type annotation for
386-- this style of composition.
388-- When two streams with multiple elements are combined in this manner, the
389-- monadic actions in the two streams are performed sequentially i.e. first all
390-- actions in the first stream are performed sequentially and then all actions
391-- in the second stream are performed sequentially. We call it
392-- /serial depth first/ as the full depth of one stream is fully traversed
393-- before we move to the next. The following example prints the sequence 1, 2,
397-- main = 'runStream' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
406-- All actions in both the streams are performed serially in the same thread.
407-- In the following example we can see that all actions are performed in the
408-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
411-- main = 'runStream' $ delay 3 <> delay 2 <> delay 1
414-- ThreadId 36: Delay 3
415-- ThreadId 36: Delay 2
416-- ThreadId 36: Delay 1
419-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
420-- 'serial'. We can use 'serial' to join streams in a sequential manner
421-- irrespective of the type of stream:
424-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`serial` (print 3 |: print 4 |: nil)
429-- The 'Semigroup' operation '<>' of the 'WSerial' type combines the two
430-- streams in a /serial breadth first/ manner. We use the 'wSerially' type
431-- combinator to effect 'WSerial' style of composition. We can also use the
432-- 'WSerial' type annotation for the stream to acheive the same effect.
434-- When two streams with multiple elements are combined in this manner, we
435-- traverse all the streams in a breadth first manner i.e. one action from each
436-- stream is peformed and yielded to the resulting stream before we come back
437-- to the first stream again and so on.
438-- The following example prints the sequence 1, 3, 2, 4
441-- main = 'runStream' . 'wSerially' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
450-- Even though the monadic actions of the two streams are performed in an
451-- interleaved manner they are all performed serially in the same thread. In
452-- the following example we can see that all actions are performed in the same
453-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
456-- main = 'runStream' . 'wSerially' $ delay 3 <> delay 2 <> delay 1
459-- ThreadId 36: Delay 3
460-- ThreadId 36: Delay 2
461-- ThreadId 36: Delay 1
464-- The polymorphic version of the 'WSerial' binary operation '<>' is called
465-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
466-- irrespective of the type, notice that we have not used the 'wSerially'
467-- combinator in the following example:
470-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`wSerial` (print 3 |: print 4 |: nil)
479-- Note that this composition cannot be used to fold infinite number of streams
480-- since it requires preserving the state until a stream is finished.
484-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
485-- streams in a depth first manner with parallel look ahead. We use the
486-- 'asyncly' type combinator to effect 'Async' style of composition. We
487-- can also use the 'Async' type annotation for the stream type to acheive
490-- When two streams with multiple elements are combined in this manner, the
491-- streams are traversed in depth first manner just like 'Serial', however it
492-- can execute the next stream concurrently and return the results from it
493-- as they arrive i.e. the results from the next stream may be yielded even
494-- before the results from the first stream. Concurrent execution of the next
495-- stream(s) is performed if the first stream blocks or if it cannot produce
496-- output at the rate that is enough to meet the consumer demand. Multiple
497-- streams can be executed concurrently to meet the demand.
498-- In the following example the first stream does not block,
499-- therefore the first stream is completely exhausted before the second.
502-- main = 'runStream' . 'asyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
511-- If the first stream blocks, we can yield from the second. In the example
512-- below each yield in the stream has a constant delay of 1 second therefore 1
513-- and 3 would be yielded first and then 2 and 4 would be yielded.
516-- main = 'runStream' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
517-- where p n = threadDelay 1000000 >> print n
526-- In the following example we can see that new threads are started when a
527-- computation blocks. Notice that the output from the stream with the
528-- shortest delay is printed first. The whole computation takes @maximum of
529-- (3, 2, 1) = 3@ seconds:
532-- main = 'runStream' . 'asyncly' $ delay 3 '<>' delay 2 '<>' delay 1
535-- ThreadId 42: Delay 1
536-- ThreadId 41: Delay 2
537-- ThreadId 40: Delay 3
540-- When we have a tree of computations composed using this style, the tree is
541-- traversed in DFS style just like the 'Serial' style, the only difference is
542-- that here we can move on to executing the next stream if a stream blocks.
543-- However, we will not start new threads if we have sufficient output to
544-- saturate the consumer. This is why we call it left-biased demand driven or
545-- adaptive concurrency style, the concurrency tends to stay on the left side
546-- of the composition as long as possible. More threads are started based on
547-- the pull rate of the consumer. The following example prints an output every
548-- second as all of the actions are concurrent.
551-- main = 'runStream' . 'asyncly' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
560-- All the computations may even run in a single thread when more threads are
561-- not needed. As you can see, in the following example the computations are
562-- run in a single thread one after another, because none of them blocks.
563-- However, if the thread consuming the stream were faster than the producer
564-- then it would have started parallel threads for each computation to keep up
565-- even if none of them blocks:
568-- main = 'runStream' . 'asyncly' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
569-- where traced m = S.'once' (myThreadId >>= print) >> return m
577-- Note that the order of printing in the above examples may change due to
578-- variations in scheduling latencies for concurrent threads.
580-- The polymorphic version of the 'Async' binary operation '<>' is called
581-- 'async'. We can use 'async' to join streams in a left biased
582-- adaptively concurrent manner irrespective of the type, notice that we have
583-- not used the 'asyncly' combinator in the following example:
586-- main = 'runStream' $ delay 3 \`async` delay 2 \`async` delay 1
589-- ThreadId 42: Delay 1
590-- ThreadId 41: Delay 2
591-- ThreadId 40: Delay 3
594-- Since the concurrency provided by this operator is demand driven it cannot
595-- be used when the composed computations start timers that are relative to
596-- each other because all computations may not be started at the same time and
597-- therefore timers in all of them may not start at the same time. When
598-- relative timing among all computations is important or when we need to start
599-- all computations at once for any reason 'Parallel' style must be used
602-- 'Async' style should be preferred over 'Parallel' or 'WAsync' unless you
603-- really need those. It utilizes the resources optimally. It should be used
604-- when we know that the computations can run in parallel but we do not care if
605-- they actually run in parallel or not, that decision can be left to the
606-- scheduler based on demand. Also, note that this operator can be used to fold
607-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
608-- because it does not require us to run all of them at the same time in a fair
613-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
614-- a concurrent manner using /breadth first traversal/. We use the 'wAsyncly'
615-- type combinator to effect 'WAsync' style of composition. We can also use the
616-- 'WAsync' type annotation for the stream to acheive the same effect.
618-- When streams with multiple elements are combined in this manner, we traverse
619-- all the streams concurrently in a breadth first manner i.e. one action from
620-- each stream is peformed and yielded to the resulting stream before we come
621-- back to the first stream again and so on. Even though we execute the actions
622-- in a breadth first order the outputs may be consumed in a different order
623-- because they are consumed on a first come first serve basis.
625-- In the following example we can see that outputs are produced in the breadth
626-- first travresal order but this is not guaranteed.
629-- main = 'runStream' . 'wAsyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
638-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
639-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
640-- concurrent traversal irrespective of the type, notice that we have not used
641-- the 'wAsyncly' combinator in the following example:
644-- main = 'runStream' $ delay 3 \`wAsync` delay 2 \`wAsync` delay 1
647-- ThreadId 42: Delay 1
648-- ThreadId 41: Delay 2
649-- ThreadId 40: Delay 3
652-- Since the concurrency provided by this style is demand driven it may not
653-- be used when the composed computations start timers that are relative to
654-- each other because all computations may not be started at the same time and
655-- therefore timers in all of them may not start at the same time. When
656-- relative timing among all computations is important or when we need to start
657-- all computations at once for any reason 'Parallel' style must be used
663-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
664-- streams in a fairly concurrent manner with round robin scheduling. We use
665-- the 'parallely' type combinator to effect 'Parallel' style of composition.
666-- We can also use the 'Parallel' type annotation for the stream type to
667-- acheive the same effect.
669-- When two streams with multiple elements are combined in this manner, the
670-- monadic actions in both the streams are performed concurrently with a fair
671-- round robin scheduling. The outputs are yielded in the order in which the
672-- actions complete. This is pretty similar to the 'WAsync' type, the
673-- difference is that 'WAsync' is adaptive to the consumer demand and may or
674-- may not execute all actions in parallel depending on the demand, whereas
675-- 'Parallel' runs all the streams in parallel irrespective of the demand.
677-- The following example sends a query to all the three search engines in
678-- parallel and prints the name of the search engines in the order in which the
683-- import qualified Streamly.Prelude as S
684-- import Network.HTTP.Simple
686-- main = 'runStream' . 'parallely' $ google \<> bing \<> duckduckgo
688-- google = get "https://www.google.com/search?q=haskell"
689-- bing = get "https://www.bing.com/search?q=haskell"
690-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
691-- get s = S.'once' (httpNoBody (parseRequest_ s) >> putStrLn (show s))
694-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
695-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
696-- manner irrespective of the type, notice that we have not used the
697-- 'parallely' combinator in the following example:
700-- main = 'runStream' $ delay 3 \`parallel` delay 2 \`wAsync` delay 1
703-- ThreadId 42: Delay 1
704-- ThreadId 41: Delay 2
705-- ThreadId 40: Delay 3
708-- Note that this style of composition cannot be used to combine infinite
709-- number of streams, as it will lead to an infinite sized scheduling queue.
714-- The 'mkAsync' API can be used to create references to asynchronously running
715-- stream computations. We can then use 'uncons' to explore the streams
716-- arbitrarily and then recompose individual elements to create a new stream.
717-- This way we can dynamically decide which stream to explore at any given
718-- time. Take an example of a merge sort of two sorted streams. We need to
719-- keep consuming items from the stream which has the lowest item in the sort
720-- order. This can be achieved using async references to streams. See
721-- "MergeSort.hs" in the examples directory.
725-- We can use 'Monoid' instances to fold a container of streams in the desired
726-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
727-- to fold streams using the polymorphic combine operations:
729-- * 'foldWith' is like 'fold', it folds a 'Foldable' container of streams
730-- using the given composition operator.
731-- * 'foldMapWith' is like 'foldMap', it folds like @foldWith@ but also maps a
732-- function before folding.
733-- * 'forEachWith' is like @foldMapwith@ but the container argument comes before
734-- the function argument.
736-- All of the following are equivalent and start ten concurrent tasks each with
737-- a delay from 1 to 10 seconds, resulting in the printing of each number every
742-- import qualified "Streamly.Prelude" as S
743-- import Control.Concurrent
746-- 'runStream' $ 'asyncly' $ foldMap delay [1..10]
747-- 'runStream' $ 'foldWith' 'async' (map delay [1..10])
748-- 'runStream' $ 'foldMapWith' 'async' delay [1..10]
749-- 'runStream' $ 'forEachWith' 'async' [1..10] delay
750-- where delay n = S.'once' $ threadDelay (n * 1000000) >> print n
755-- Till now we discussed ways to apply transformations on a stream or to merge
756-- streams together to create another stream. We mentioned earlier that
757-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
758-- We will now discuss the concept of a nested composition of streams which is
759-- analogous to nested @for@ loops in the imperative paradigm. Functional
760-- programmers call this style of composition a list transformer or @ListT@.
761-- Logic programmers call it a logic monad or non-deterministic composition,
762-- but for ordinary imperative minded people like me it is easier to think in
763-- terms of good old nested @for@ loops.
767-- In functional programmer's parlance the 'Monad' instances of different
768-- 'IsStream' types implement non-determinism, exploring all possible
769-- combination of choices from both the streams. From an imperative
770-- programmer's point of view it behaves like nested loops i.e. for each
771-- element in the first stream and for each element in the second stream
772-- execute the body of the loop.
774-- The 'Monad' instances of 'Serial', 'WSerial', 'Async' and 'WAsync'
775-- stream types support different flavors of nested looping. In other words,
776-- they are all variants of list transformer. The nesting behavior of these
777-- types correspond exactly to the way they merge streams as we discussed in
778-- the previous section.
783-- The 'Monad' composition of the 'Serial' type behaves like a standard list
784-- transformer. This is the default when we do not use an explicit type
785-- combinator. However, the 'serially' type combinator can be used to switch to
786-- this style of composition. We will see how this style of composition works
787-- in the following examples.
789-- Let's start with an example with a simple @for@ loop without any nesting.
790-- For simplicity of illustration we are using streams of pure values in all
791-- the examples. However, the streams could also be made of monadic actions
796-- import qualified "Streamly.Prelude" as S
798-- main = 'runStream' $ do
799-- x <- S.'fromFoldable' [3,2,1]
803-- ThreadId 30: Delay 3
804-- ThreadId 30: Delay 2
805-- ThreadId 30: Delay 1
808-- As we can see, the code after the @fromFoldable@ statement is run three
809-- times, once for each value of @x@ drawn from the stream. All the three
810-- iterations are serial and run in the same thread one after another. In
811-- imperative terms this is equivalent to a @for@ loop with three iterations.
813-- A console echo loop copying standard input to standard output can simply be
818-- import qualified "Streamly.Prelude" as S
820-- main = 'runStream' $ forever $ S.once getLine >>= S.once . putStrLn
823-- When multiple streams are composed using this style they nest in a DFS
824-- manner i.e. nested iterations of a loop are executed before we proceed to
825-- the next iteration of the parent loop. This behaves just like nested @for@
826-- loops in imperative programming.
830-- import qualified "Streamly.Prelude" as S
832-- main = 'runStream' $ do
833-- x <- S.'fromFoldable' [1,2]
834-- y <- S.'fromFoldable' [3,4]
835-- S.'once' $ putStrLn $ show (x, y)
844-- Notice that this is analogous to merging streams of type 'Serial' or merging
845-- streams using 'serial'.
849-- The 'Monad' composition of 'Async' type can perform the iterations of a
850-- loop concurrently. Concurrency is demand driven i.e. more concurrent
851-- iterations are started only if the previous iterations are not able to
852-- produce enough output for the consumer of the output stream. This works
853-- exactly the same way as the merging of two streams 'asyncly' works.
854-- This is the concurrent analogue of 'Serial' style monadic composition.
856-- The 'asyncly' type combinator can be used to switch to this style of
857-- composition. Alternatively, a type annotation can be used to specify the
858-- type of the stream as 'Async'.
862-- import "Streamly.Prelude"
864-- main = 'runStream' . 'asyncly' $ do
865-- x <- 'fromFoldable' [3,2,1]
869-- ThreadId 40: Delay 1
870-- ThreadId 39: Delay 2
871-- ThreadId 38: Delay 3
874-- As we can see the code after the @fromFoldable@ statement is run three
875-- times, once for each value of @x@. All the three iterations are concurrent
876-- and run in different threads. The iteration with least delay finishes first.
877-- When compared to imperative programming, this can be viewed as a @for@ loop
878-- with three concurrent iterations.
880-- Concurrency is demand driven just as in the case of 'async' merging.
881-- When multiple streams are composed using this style, the iterations are
882-- triggered in a depth first manner just like 'Serial' i.e. nested iterations are
883-- executed before we proceed to the next iteration at higher level. However,
884-- unlike 'Serial' more than one iterations may be started concurrently based
885-- on the demand from the consumer of the stream.
889-- import qualified "Streamly.Prelude" as S
891-- main = 'runStream' . 'asyncly' $ do
892-- x <- S.'fromFoldable' [1,2]
893-- y <- S.'fromFoldable' [3,4]
894-- S.'once' $ putStrLn $ show (x, y)
903-- $interleavedNesting
905-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
906-- outer and inner loops in a nested loop composition. This works exactly the
907-- same way as the merging of two streams in 'wSerially' fashion works. The
908-- 'wSerially' type combinator can be used to switch to this style of
909-- composition. Alternatively, a type annotation can be used to specify the
910-- type of the stream as 'WSerial'.
914-- import qualified "Streamly.Prelude" as S
916-- main = 'runStream' . 'wSerially' $ do
917-- x <- S.'fromFoldable' [1,2]
918-- y <- S.'fromFoldable' [3,4]
919-- S.once $ putStrLn $ show (x, y)
931-- Just like 'Async' the 'Monad' composition of 'WAsync' runs the iterations of
932-- a loop concurrently. The difference is in the nested loop behavior. The
933-- nested loops in this type are traversed and executed in a breadth first
934-- manner rather than the depth first manner of 'Async' style.
935-- The loop nesting works exactly the same way as the merging of streams
936-- 'wAsyncly' works. The 'wAsyncly' type combinator can be used to switch to
937-- this style of composition. Alternatively, a type annotation can be used to
938-- specify the type of the stream as 'WAsync'.
942-- import qualified "Streamly.Prelude" as S
944-- main = 'runStream' . 'wAsyncly' $ do
945-- x <- S.'fromFoldable' [1,2]
946-- y <- S.'fromFoldable' [3,4]
947-- S.'once' $ putStrLn $ show (x, y)
958-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
959-- iterations of a loop concurrently. The difference is in the nested loop
960-- behavior. The streams at each nest level is run fully concurrently
961-- irrespective of the demand. The loop nesting works exactly the same way as
962-- the merging of streams 'parallely' works. The 'parallely' type combinator
963-- can be used to switch to this style of composition. Alternatively, a type
964-- annotation can be used to specify the type of the stream as 'Parallel'.
968-- import qualified "Streamly.Prelude" as S
970-- main = 'runStream' . 'parallely' $ do
971-- x <- S.'fromFoldable' [3,2,1]
975-- ThreadId 40: Delay 1
976-- ThreadId 39: Delay 2
977-- ThreadId 38: Delay 3
982-- Streamly code is usually written in a way that is agnostic of the
983-- specific monadic composition type. We use a polymorphic type with a
984-- 'IsStream' type class constraint. When running the stream we can choose the
985-- specific mode of composition. For example take a look at the following code.
989-- import qualified "Streamly.Prelude" as S
991-- composed :: 'IsStream' t => t m a
996-- S.'once' $ putStrLn $ show (sz, cl, sh)
1000-- sizes = S.'fromFoldable' [1, 2, 3]
1001-- colors = S.'fromFoldable' ["red", "green", "blue"]
1002-- shapes = S.'fromFoldable' ["triangle", "square", "circle"]
1005-- Now we can interpret this in whatever way we want:
1008-- main = 'runStream' . 'serially' $ composed
1009-- main = 'runStream' . 'wSerially' $ composed
1010-- main = 'runStream' . 'asyncly' $ composed
1011-- main = 'runStream' . 'wAsyncly' $ composed
1012-- main = 'runStream' . 'parallely' $ composed
1015-- As an exercise try to figure out the output of this code for each mode of
1020-- 'fmap' transforms a stream by mapping a function on all elements of the
1021-- stream. 'fmap' behaves in the same way for all stream types, it is always
1026-- import qualified "Streamly.Prelude" as S
1028-- main = (S.'toList' $ fmap show $ S.'fromFoldable' [1..10]) >>= print
1031-- Also see the 'mapM' and 'sequence' functions for mapping actions, in the
1032-- "Streamly.Prelude" module.
1036-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
1037-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are
1040-- The following example uses the 'Serial' applicative, it runs all iterations
1041-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
1045-- import qualified "Streamly.Prelude" as S
1046-- import Control.Concurrent
1050-- d n = delay n >> return n
1052-- main = (S.'toList' . 'serially' $ (,) \<$> s1 \<*> s2) >>= print
1055-- ThreadId 36: Delay 1
1056-- ThreadId 36: Delay 3
1057-- ThreadId 36: Delay 4
1058-- ThreadId 36: Delay 2
1059-- ThreadId 36: Delay 3
1060-- ThreadId 36: Delay 4
1061-- [(1,3),(1,4),(2,3),(2,4)]
1064-- Similalrly 'WSerial' applicative runs the iterations in an interleaved
1065-- order but since it is serial it takes a total of 17 seconds:
1068-- main = (S.'toList' . 'wSerially' $ (,) \<$> s1 \<*> s2) >>= print
1071-- ThreadId 36: Delay 1
1072-- ThreadId 36: Delay 3
1073-- ThreadId 36: Delay 2
1074-- ThreadId 36: Delay 3
1075-- ThreadId 36: Delay 4
1076-- ThreadId 36: Delay 4
1077-- [(1,3),(2,3),(1,4),(2,4)]
1080-- 'Async' can run the iterations concurrently and therefore takes a total
1081-- of 10 seconds (1 + 2 + 3 + 4):
1084-- main = (S.'toList' . 'asyncly' $ (,) \<$> s1 \<*> s2) >>= print
1087-- ThreadId 34: Delay 1
1088-- ThreadId 36: Delay 2
1089-- ThreadId 35: Delay 3
1090-- ThreadId 36: Delay 3
1091-- ThreadId 35: Delay 4
1092-- ThreadId 36: Delay 4
1093-- [(1,3),(2,3),(1,4),(2,4)]
1096-- Similalrly 'WAsync' as well can run the iterations concurrently and
1097-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
1100-- main = (S.'toList' . 'wAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1103-- ThreadId 34: Delay 1
1104-- ThreadId 36: Delay 2
1105-- ThreadId 35: Delay 3
1106-- ThreadId 36: Delay 3
1107-- ThreadId 35: Delay 4
1108-- ThreadId 36: Delay 4
1109-- [(1,3),(2,3),(1,4),(2,4)]
1114-- Zipping is a special transformation where the corresponding elements of two
1115-- streams are combined together using a zip function producing a new stream of
1116-- outputs. Two different types are provided for serial and concurrent zipping.
1117-- These types provide an applicative instance that can be used to lift
1118-- functions to zip the argument streams.
1119-- Also see the zipping functions in the "Streamly.Prelude" module.
1123-- The applicative instance of 'ZipSerial' type zips streams serially.
1124-- 'zipSerially' type combinator can be used to switch to serial applicative
1129-- import qualified "Streamly.Prelude" as S
1130-- import Control.Concurrent
1132-- d n = delay n >> return n
1133-- s1 = 'streamly' $ d 1 <> d 2
1134-- s2 = 'streamly' $ d 3 <> d 4
1136-- main = (S.'toList' . 'zipSerially' $ (,) \<$> s1 \<*> s2) >>= print
1139-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1140-- everything runs serially:
1143-- ThreadId 29: Delay 1
1144-- ThreadId 29: Delay 3
1145-- ThreadId 29: Delay 2
1146-- ThreadId 29: Delay 4
1152-- The applicative instance of 'ZipAsync' type zips streams concurrently.
1153-- 'zipAsyncly' type combinator can be used to switch to parallel applicative
1159-- import qualified "Streamly.Prelude" as S
1160-- import Control.Concurrent
1161-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
1163-- d n = delay n >> return n
1164-- s1 = 'streamly' $ d 1 <> d 2
1165-- s2 = 'streamly' $ d 3 <> d 4
1168-- hSetBuffering stdout LineBuffering
1169-- (S.'toList' . 'zipAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1172-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1173-- are produced concurrently, and 2 and 4 are produced concurrently:
1176-- ThreadId 32: Delay 1
1177-- ThreadId 32: Delay 2
1178-- ThreadId 33: Delay 3
1179-- ThreadId 33: Delay 4
1185-- When writing concurrent programs there are two distinct places where the
1186-- programmer can control the concurrency. First, when /composing/ a stream by
1187-- merging multiple streams we can choose an appropriate sum style operators to
1188-- combine them concurrently or serially. Second, when /processing/ a stream in
1189-- a monadic composition we can choose one of the monad composition types to
1190-- choose the desired type of concurrency.
1192-- In the following example the squares of @x@ and @y@ are computed
1193-- concurrently using the 'async' operation and the square roots of their
1194-- sum are computed serially because of the 'streamly' combinator. We can
1195-- choose different combinators for the monadic processing and the stream
1196-- generation, to control the concurrency. We can also use the 'asyncly'
1197-- combinator instead of explicitly folding with 'async'.
1201-- import qualified "Streamly.Prelude" as S
1202-- import Data.List (sum)
1206-- $ 'serially' -- Serial monadic processing (sqrt below)
1208-- x2 \<- 'forEachWith' 'async' [1..100] $ -- Concurrent @"for"@ loop
1209-- \\x -> return $ x * x -- body of the loop
1210-- y2 \<- 'forEachWith' 'async' [1..100] $
1211-- \\y -> return $ y * y
1212-- return $ sqrt (x2 + y2)
1216-- We can see how this directly maps to the imperative style
1217-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1218-- and operators instead of the ugly pragmas.
1220-- For more concurrent programming examples see,
1221-- "ListDir.hs", "MergeSort.hs" and "SearchQuery.hs" in the examples directory.
1225-- Reactive programming is nothing but concurrent streaming which is what
1226-- streamly is all about. With streamly we can generate streams of events,
1227-- merge streams that are generated concurrently and process events
1228-- concurrently. We can do all this without any knowledge about the specifics
1229-- of the implementation of concurrency. In the following example you will see
1230-- that the code is just regular Haskell code without much streamly APIs used
1231-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1234-- This application has two independent and concurrent sources of event
1235-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1236-- events that deteriorate the health of the character in the game.
1237-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1238-- health improves and the game continues.
1241-- {-\# LANGUAGE FlexibleContexts #-}
1244-- import Control.Concurrent (threadDelay)
1245-- import Control.Monad (when)
1246-- import Control.Monad.State (MonadState, get, modify, runStateT)
1247-- import Data.Semigroup (cycle1)
1249-- data Event = Harm Int | Heal Int | Quit deriving (Show)
1251-- userAction :: MonadIO m => 'SerialT' m Event
1252-- userAction = cycle1 $ liftIO askUser
1255-- command <- getLine
1257-- "potion" -> return (Heal 10)
1258-- "quit" -> return Quit
1259-- _ -> putStrLn "What?" >> askUser
1261-- acidRain :: MonadIO m => 'SerialT' m Event
1262-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
1264-- game :: ('MonadAsync' m, MonadState Int m) => 'SerialT' m ()
1266-- event \<- userAction \`parallel` acidRain
1268-- Harm n -> modify $ \\h -> h - n
1269-- Heal n -> modify $ \\h -> h + n
1270-- Quit -> fail "quit"
1273-- when (h <= 0) $ fail "You die!"
1274-- liftIO $ putStrLn $ "Health = " ++ show h
1277-- putStrLn "Your health is deteriorating due to acid rain,\\
1278-- \\ type \\"potion\\" or \\"quit\\""
1279-- _ <- runStateT ('runStream' game) 60
1283-- You can also find the source of this example in the examples directory as
1284-- "AcidRain.hs". It has been adapted from Gabriel's
1285-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1287-- This is much simpler compared to the pipes version because of the builtin
1288-- concurrency in streamly. You can also find a SDL based reactive programming
1289-- example adapted from Yampa in "Streamly.Examples.CirclingSquare".
1293-- Streamly is highly optimized for performance, it is designed for serious
1294-- high performing, concurrent and scalable applications. We have created the
1295-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1296-- package which is specifically and carefully designed to measure the
1297-- performance of Haskell streaming libraries fairly and squarely in the right
1298-- way. Streamly performs at par or even better than most streaming libraries
1299-- for serial operations even though it needs to deal with the concurrency
1304-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1306-- Interop with @vector@:
1310-- import qualified Streamly.Prelude as S
1311-- import qualified Data.Vector.Fusion.Stream.Monadic as V
1314-- -- streamly to vector
1315-- V.toList (V.unfoldrM S.uncons (S.fromFoldable [1..3])) >>= print
1317-- -- vector to streamly
1318-- S.toList (S.unfoldrM unconsV (V.fromList [1..3])) >>= print
1324-- then return Nothing
1327-- return $ Just (h, V.tail v)
1330-- Interop with @pipes@:
1334-- import qualified "Streamly.Prelude" as S
1335-- import qualified Pipes as P
1336-- import qualified Pipes.Prelude as P
1339-- -- streamly to pipe
1340-- P.toListM (P.unfoldr unconsS (S.'fromFoldable' [1..3])) >>= print
1342-- -- pipe to streamly
1343-- S.'toList' (S.'unfoldrM' unconsP (P.each [1..3])) >>= print
1346-- -- Adapt P.next to return a Maybe instead of Either
1347-- unconsP p = P.next p >>= either (\\_ -> return Nothing) (return . Just)
1349-- -- Adapt S.uncons to return an Either instead of Maybe
1350-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1353-- Interop with @streaming@:
1357-- import qualified "Streamly.Prelude" as S
1358-- import qualified Streaming as SG
1359-- import qualified Streaming.Prelude as SG
1362-- -- streamly to streaming
1363-- SG.toList (SG.unfoldr unconsS (S.'fromFoldable' [1..3])) >>= print
1365-- -- streaming to streamly
1366-- S.'toList' (S.unfoldrM SG.uncons (SG.each [1..3])) >>= print
1370-- -- Adapt S.uncons to return an Either instead of Maybe
1371-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1374-- Interop with @conduit@:
1378-- import qualified "Streamly.Prelude" as S
1379-- import qualified Data.Conduit as C
1380-- import qualified Data.Conduit.List as C
1381-- import qualified Data.Conduit.Combinators as C
1384-- -- streamly to conduit
1385-- C.runConduit (C.unfoldM S.'uncons' (S.'fromFoldable' [1..3]) C..| C.sinkList) >>= print
1387-- -- It seems there is no way out of a conduit as it does not provide an
1388-- -- uncons or a tail function.
1393-- List transformers and logic programming monads also provide a product style
1394-- composition similar to streamly, however streamly generalizes it with the
1395-- time dimension; allowing streams to be composed in an asynchronous and
1396-- concurrent fashion in many different ways. It also provides multiple
1397-- alternative ways of composing streams e.g. serial, interleaved or
1400-- This seemingly simple addition of asynchronicity and concurrency to product
1401-- style streaming composition unifies a number of disparate abstractions into
1402-- one powerful, concise and elegant abstraction. A wide variety of
1403-- programming problems can be solved elegantly with this abstraction. In
1404-- particular, it unifies three major programming domains namely
1405-- non-deterministic (logic) programming, concurrent programming and functional
1406-- reactive programming. In other words, you can do everything with this one
1407-- abstraction that you could do with the popular libraries listed under these
1408-- categories in the list below.
1411-- +-----------------+----------------+
1412-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1413-- | +----------------+
1414-- | | <https://hackage.haskell.org/package/list-t list-t> |
1415-- | +----------------+
1416-- | | <https://hackage.haskell.org/package/logict logict> |
1417-- +-----------------+----------------+
1418-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1419-- | +----------------+
1420-- | | <https://hackage.haskell.org/package/streaming streaming> |
1421-- | +----------------+
1422-- | | <https://hackage.haskell.org/package/pipes pipes> |
1423-- | +----------------+
1424-- | | <https://hackage.haskell.org/package/conduit conduit> |
1425-- +-----------------+----------------+
1426-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1427-- | +----------------+
1428-- | | <https://hackage.haskell.org/package/transient transient> |
1429-- +-----------------+----------------+
1430-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1431-- | +----------------+
1432-- | | <https://hackage.haskell.org/package/dunai dunai> |
1433-- | +----------------+
1434-- | | <https://hackage.haskell.org/package/reflex reflex> |
1435-- +-----------------+----------------+
1438-- Streamly is a list-transformer. It provides all the functionality provided
1439-- by any of the list transformer and logic programming packages listed above.
1440-- In addition, Streamly naturally integrates the concurrency dimension to the
1441-- basic list transformer functionality.
1443-- When it comes to streaming, in terms of the streaming API streamly is almost
1444-- identical to the vector package. Streamly, vector and streaming packages all
1445-- represent a stream as data and are therefore similar in the fundamental
1446-- approach to streaming. The fundamental difference is that streamly adds
1447-- concurrency support and the monad instance provides concurrent looping.
1448-- Other streaming libraries like pipes, conduit and machines represent and
1449-- compose stream processors rather than the stream data and therefore fall in
1450-- another class of streaming libraries and have comparatively more complicated
1453-- When it comes to concurrency, streamly can do everything that the @async@
1454-- package can do and more. async provides applicative concurrency whereas
1455-- streamly provides both applicative and monadic concurrency. The
1456-- 'ZipAsync' type behaves like the applicative instance of async. In
1457-- comparison to transient streamly has a first class streaming interface and
1458-- is a monad transformer that can be used universally in any Haskell monad
1459-- transformer stack. Streamly was in fact originally inspired by the
1460-- concurrency implementation in @transient@ though it has no resemblence with
1461-- that and takes a lazy pull approach versus transient's strict push approach.
1463-- The non-determinism, concurrency and streaming combination make streamly a
1464-- strong FRP capable library as well. FRP is fundamentally stream of events
1465-- that can be processed concurrently. The example in this tutorial as well as
1466-- the "Streamly.Examples.CirclingSquare" example from Yampa demonstrate the
1467-- basic FRP capability of streamly. In core concepts streamly is strikingly
1468-- similar to @dunai@. dunai was designed from a FRP perspective and streamly
1469-- was originally designed from a concurrency perspective. However, both have
1470-- similarity at the core.
1474-- * Read the documentation of "Streamly" module
1475-- * Read the documentation of "Streamly.Prelude" module
1476-- * See the examples in the "examples" directory of the package
1477-- * 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