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 -- * Generating Streams
43 -- * Eliminating Streams
46 -- * Transforming Streams
55 -- *** Serial Composition ('streamly')
58 -- *** Serial Interleaved Composition ('costreamly')
61 -- *** Left Biased Concurrent Composition ('coparallely')
64 -- *** Round Robin Concurrent Composition ('parallely')
67 -- *** Custom composition
79 -- *** Serial Nesting ('streamly')
82 -- *** Serial Interleaved Nesting ('costreamly')
83 -- $interleavedNesting
85 -- *** Left Biased Concurrent Nesting ('coparallely')
88 -- *** Round Robin Concurrent Nesting ('parallely')
89 -- $fairlyConcurrentNesting
106 -- ** Parallel Zipping
109 -- * Summary of Compositions
110 -- $compositionSummary
112 -- * Monad transformers
113 -- $monadtransformers
115 -- * Concurrent Programming
118 -- * Reactive Programming
124 -- * Interoperation with Streaming Libraries
127 -- * Comparison with Existing Packages
130 -- * Where to go next?
136import Streamly.Prelude
138import Control.Applicative
140import Control.Monad.IO.Class (MonadIO(..))
141import Control.Monad.Trans.Class (MonadTrans (lift))
145-- The way a list represents a sequence of pure values, a stream represents a
146-- sequence of monadic actions. The monadic stream API offered by Streamly is
147-- very close to the Haskell "Prelude" pure lists' API, it can be considered as a
148-- natural extension of lists to monadic actions. Streamly streams provide
149-- concurrent composition and merging of streams. It can be considered as a
150-- concurrent list transformer. In contrast to the "Prelude" lists, merging or
151-- appending streams of arbitrary length is scalable and inexpensive.
153-- The basic stream type is 'Stream', it represents a sequence of IO actions,
154-- and is a 'Monad'. The type 'StreamT' is a monad transformer that can
155-- represent a sequence of actions in an arbitrary monad. The type 'Stream' is
156-- in fact a synonym for @StreamT IO@. There are a few more types similar to
157-- 'StreamT' that all represent streams and differ only in the 'Semigroup',
158-- 'Applicative' and 'Monad' compositions of the streams. 'Stream' and
159-- 'Costream' types compose serially whereas 'Coparallel' and 'Parallel' types
160-- compose concurrently. All these types can be freely inter-converted using
161-- type combinators without any cost. You can freely switch to any type of
162-- composition at any point in the program. When no type annotation or
163-- explicit stream type combinators are used, the default stream type is
164-- inferred as 'Stream'.
166-- The 'Stream' monad is almost a drop in replacement for the 'IO' monad, IO
167-- monad is a special case of the 'Stream' monad; IO monad represents a single
168-- IO action whereas the 'Stream' monad represents a sequence of IO actions.
169-- The only change you need to make to go from 'IO' to 'Stream' is to use
170-- 'runStream' to run the monad and to prefix the IO actions with 'once'. You
171-- can also use 'liftIO' instead of 'once'. In fact, if you use liftIO you can
172-- switch from 'Stream' to IO monad by simply removing the 'runStream'
173-- function; no other changes are needed unless you have used some stream
174-- specific composition or combinators.
176-- Similarly, the 'Stream' type is almost a drop in replacement for pure lists,
177-- pure lists are a special case of monadic streams. If you use 'nil' in place
178-- of '[]' and '|:' in place ':' you can replace a list with a 'Stream'. The
179-- only difference is that the elements must be monadic type and to operate on
180-- the streams we must use the corresponding functions from "Streamly.Prelude"
181-- instead of using the base "Prelude".
185-- To represent streams in an arbitrary monad use the more general monad
186-- transformer types for example the monad transformer type corresponding to
187-- the 'Stream' type is 'StreamT'. @StreamT m a@ represents a stream of values
188-- of type 'a' in some underlying monad 'm'. For example, @StreamT IO Int@ is a
189-- stream of 'Int' in 'IO' monad. In fact, the type 'Stream' is a synonym for
192-- Similarly we have monad transformer types for other stream types as well viz.
193-- 'CostreamT', 'CoparallelT' and 'ParallelT'.
195-- To lift a value from an underlying monad in a monad transformer stack into a
196-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
199-- > 'runStream' $ liftIO $ putStrLn "Hello world!"
201-- > 'runStream' $ lift $ putStrLn "Hello world!"
208-- We will assume the following imports in this tutorial. Go ahead, fire up a
209-- GHCi session and import these lines to start playing.
212-- > import "Streamly"
213-- > import "Streamly.Prelude" (|:)
214-- > import qualified "Streamly.Prelude" as S
217-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
218-- a monadic action at the head of the stream.
221-- > S.'toList' S.'nil'
223-- > S.'toList' $ 'getLine' |: 'getLine' |: S.'nil'
229-- To create a singleton stream from a pure value use 'pure' and to create a
230-- singleton stream from a monadic action use 'once'.
233-- > S.'toList' $ 'pure' 1
235-- > S.'toList' $ S.'once' 'getLine'
240-- To create a stream from pure values in a 'Foldable' container use
241-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
244-- > S.'toList' $ S.'fromFoldable' [1..3]
246-- > S.'toList' $ 'Prelude.foldr' S.'cons' S.'nil' [1..3]
250-- To create a stream from monadic actions in a 'Foldable' container just use a
251-- right fold using 'consM' and 'nil':
254-- > 'runStream' $ 'Prelude.foldr' ('|:') S.'nil' ['putStr' "Hello ", 'putStrLn' "world!"]
258-- For more ways to construct a stream see the module "Streamly.Prelude".
262-- We have already seen 'runStream' and 'toList' to eliminate a stream in the
263-- examples above. 'runStream' runs a stream discarding the results i.e. only
264-- for effects. 'toList' runs the stream and collects the results in a list.
266-- For other ways to eliminate a stream see the @Folding@ section in
267-- "Streamly.Prelude" module.
271-- Transformation over a stream is the equivalent of a @for@ loop construct in
272-- imperative paradigm. We iterate over every element in the stream and perform
273-- certain transformations for each element. Transformations may involve
274-- mapping functions over the elements, filtering elements from the stream or
275-- folding all the elements in the stream into a single value. Streamly streams
276-- are exactly like lists and you can perform all the transformations in the
277-- same way as you would on lists.
279-- Here is a simple console echo program that just echoes every input line,
283-- > 'runStream' $ S.'repeatM' getLine & S.'mapM' putStrLn
286-- The following code snippet reads lines from standard input, filters blank
287-- lines, drops the first non-blank line, takes the next two, up cases them,
288-- numbers them and prints them:
292-- import qualified "Streamly.Prelude" as S
293-- import Data.Char (toUpper)
294-- import Data.Function ((&))
296-- main = 'runStream' $
297-- S.'repeatM' getLine
298-- & S.'filter' (not . null)
301-- & fmap (map toUpper)
302-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
303-- & S.'mapM' putStrLn
308-- We can combine two streams into a single stream using semigroup composition
309-- operation '<>'. Streams can be combined in many different ways as described
310-- in the following sections, the '<>' operation behaves differently depending
311-- on the stream type in effect. Streamly streams can be of many
312-- different types i.e. 'Stream', 'Costream', 'Coparallel', 'Parallel',
313-- 'ZipStream', and 'ZipParallel'. The first four are monads whereas the last
314-- two are applicatives. These types can be freely converted from one to
315-- another to acheive the desired composition style. To force a particular
316-- type of composition we coerce the stream type using the corresponding type
317-- adapting combinator from 'streamly', 'costreamly', 'coparallely',
318-- 'parallely', 'zipStreamly' or 'zipParallely'. The default stream type is
319-- 'Stream' unless you change it by using one of the combinators.
321-- To illustrate the concurrent vs serial composition aspects, we will use the
322-- following @delay@ function to introduce a delay specified in seconds.
326-- import "Streamly.Prelude" (nil)
327-- import qualified "Streamly.Prelude" as S
328-- import Control.Concurrent
330-- delay n = S.'once' $ do
331-- threadDelay (n * 1000000)
333-- putStrLn (show tid ++ ": Delay " ++ show n)
338-- The 'Semigroup' operation '<>' of the 'Stream' type combines the two streams
339-- in a serial manner. We use the 'streamly' type combinator to effect 'Stream'
340-- style of composition. We can also use the 'Stream' type annotation for the
341-- stream type to acheive the same effect. However, since 'Stream' is the
342-- default type unless explicitly specified by using a combinator, we can
343-- omit using an explicit combinator or type annotation for this style of
346-- When two streams with multiple elements are combined in this manner, the
347-- monadic actions in the two streams are performed sequentially i.e. first all
348-- actions in the first stream are performed sequentially and then all actions
349-- in the second stream are performed sequentially. The following example
350-- prints the sequence 1, 2, 3, 4:
353-- main = 'runStream' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
362-- All actions in both the streams are performed serially in the same thread.
363-- In the following example we can see that all actions are performed in the
364-- same thread and take a combined total of 6 seconds:
367-- main = 'runStream' $ delay 3 <> delay 2 <> delay 1
370-- ThreadId 36: Delay 3
371-- ThreadId 36: Delay 2
372-- ThreadId 36: Delay 1
375-- The polymorphic version of the 'Stream' binary operation '<>' is called
376-- 'splice'. We can use 'splice' to join streams in a sequential manner
377-- irrespective of the type of stream:
380-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`splice` (print 3 |: print 4 |: nil)
385-- The 'Semigroup' operation '<>' of the 'Costream' type combines the two
386-- streams in a serial interleaved manner. We use the 'costreamly' type
387-- combinator to effect 'Costream' style of composition. We can also use the
388-- 'Costream' type annotation for the stream type to acheive the same effect.
390-- When two streams with multiple elements are combined in this manner, the
391-- monadic actions in the two streams are performed in an interleaved manner
392-- i.e. one action from the first stream is performed and its output yielded
393-- and then one action from the second stream is performed and its output
394-- yielded, and so on. The following example prints the sequence 1, 3, 2, 4
397-- main = 'runStream' . 'costreamly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
406-- Even though the monadic actions of the two streams are performed in an
407-- interleaved manner they are all performed serially in the same thread. In
408-- the following example we can see that all actions are performed in the same
409-- thread and take a combined total of 6 seconds:
412-- main = 'runStream' . 'costreamly' $ delay 3 <> delay 2 <> delay 1
415-- ThreadId 36: Delay 3
416-- ThreadId 36: Delay 2
417-- ThreadId 36: Delay 1
420-- The polymorphic version of the 'Costream' binary operation '<>' is called
421-- 'cosplice'. We can use 'cosplice' to join streams in an interleaved manner
422-- irrespective of the type, notice that we have not used the 'costreamly'
423-- combinator in the following example:
426-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`cosplice` (print 3 |: print 4 |: nil)
435-- Note that this composition cannot be used to fold infinite number of streams
436-- since it requires preserving the state until a stream is finished.
440-- The 'Semigroup' operation '<>' of the 'Coparallel' type combines the two
441-- streams in a left biased concurrent manner. We use the 'coparallel' type
442-- combinator to effect 'Coparallel' style of composition. We can also use the
443-- 'Coparallel' type annotation for the stream type to acheive the same effect.
445-- When two streams with multiple elements are combined in this manner, the
446-- monadic actions in the two streams can be performed concurrently
447-- /when needed/ with a bias towards the left stream i.e. actions from the left
448-- stream are performed sequentially, if the left stream blocks or cannot
449-- generate enough output to keep the consumer of the stream busy then it can
450-- execute actions from the right stream concurrently. The outputs of the
451-- concurrent actions are yielded by the composite stream in the order in which
452-- they complete. This can be thought of as the concurrent analogue of the
453-- 'Stream' style serial composition. If concurrent execution is not trigerred
454-- by consumer demand then it would execute actions exactly in the same order
455-- as 'Stream'. In the following example the first stream does not block,
456-- therefore the first stream is completely exhausted before the second.
459-- main = 'runStream' . 'coparallely' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
468-- If the first stream blocks, we can yield from the second. In the example
469-- below each yield in the stream has a constant delay of 1 second therefore 1
470-- and 3 would be yielded first and then 2 and 4 would be yielded.
473-- main = 'runStream' . 'coparallely' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
474-- where p n = threadDelay 1000000 >> print n
483-- In the following example we can see that new threads are started when a
484-- computation blocks. Notice that the output from the stream with the
485-- shortest delay is printed first:
488-- main = 'runStream' . 'coparallely' $ delay 3 '<>' delay 2 '<>' delay 1
491-- ThreadId 42: Delay 1
492-- ThreadId 41: Delay 2
493-- ThreadId 40: Delay 3
496-- When we have a tree of computations composed using this style, the tree is
497-- traversed in DFS style just like the serial style, the only difference is
498-- that here we can move on to executing the next stream if a stream blocks.
499-- However, we will not start new threads if we have sufficient output to
500-- saturate the consumer. This is why we call it left-biased demand driven or
501-- adaptive concurrency style, the concurrency tends to stay on the left side
502-- of the composition as long as possible. More threads are started based on
503-- the pull rate of the consumer. The following example prints an output every
504-- second as all of the actions are concurrent.
507-- main = 'runStream' . 'coparallely' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
516-- All the computations may even run in a single thread when more threads are
517-- not needed. As you can see, in the following example the computations are
518-- run in a single thread one after another, because none of them blocks.
519-- However, if the thread consuming the stream were faster than the producer
520-- then it would have started parallel threads for each computation to keep up
521-- even if none of them blocks:
524-- main = 'runStream' . 'coparallely' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
525-- where traced m = S.'once' (myThreadId >>= print) >> return m
533-- Note that the order of printing in the above examples may change due to
534-- variations in scheduling latencies for concurrent threads.
536-- The polymorphic version of the 'Coparallel' binary operation '<>' is called
537-- 'coparallel'. We can use 'coparallel' to join streams in a left biased
538-- adaptively concurrent manner irrespective of the type, notice that we have
539-- not used the 'coparallely' combinator in the following example:
542-- main = 'runStream' $ delay 3 \`coparallel` delay 2 \`coparallel` delay 1
545-- ThreadId 42: Delay 1
546-- ThreadId 41: Delay 2
547-- ThreadId 40: Delay 3
550-- Since the concurrency provided by this operator is demand driven it cannot
551-- be used when the composed computations start timers that are relative to
552-- each other because all computations may not be started at the same time and
553-- therefore timers in all of them may not start at the same time. When
554-- relative timing among all computations is important or when we need to start
555-- all computations at once for any reason 'Parallel' style must be used
558-- 'Coparallel' style should be preferred over 'Parallel' unless you really
559-- need 'Parallel'. It utilizes the resources optimally. It should be used when
560-- we know that the computations can run in parallel but we do not care if they
561-- actually run in parallel or not, that decision can be left to the scheduler
562-- based on demand. Also, note that this operator can be used to fold infinite
563-- number of streams in contrast to the 'Parallel' style, because it does not
564-- require us to run all of them at the same time in a fair manner.
568-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
569-- streams in a fairly concurrent manner with round robin scheduling. We use
570-- the 'parallel' type combinator to effect 'Parallel' style of composition. We
571-- can also use the 'Parallel' type annotation for the stream type to acheive
574-- When two streams with multiple elements are combined in this manner, the
575-- monadic actions in both the streams are performed concurrently with a fair
576-- round robin scheduling. The outputs are yielded in the order in which the
577-- actions complete. This can be thought of as the concurrent analogue of the
578-- 'Costream' style serial composition.
580-- The following example sends a query to all the three search engines in
581-- parallel and prints the name of the search engines in the order in which the
586-- import qualified Streamly.Prelude as S
587-- import Network.HTTP.Simple
589-- main = 'runStream' . 'parallely' $ google \<> bing \<> duckduckgo
591-- google = get "https://www.google.com/search?q=haskell"
592-- bing = get "https://www.bing.com/search?q=haskell"
593-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
594-- get s = S.'once' (httpNoBody (parseRequest_ s) >> putStrLn (show s))
597-- In the following example we can see that the fairly parallel composition
598-- yields outputs in the same order as the 'Costream' composition, though this
599-- is not always guaranteed because of thread scheduling variances.
602-- main = 'runStream' . 'parallely' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
611-- The polymorphic version of the 'Parallel' binary operation '<>' is called
612-- 'parallel'. We can use 'parallel' to join streams in a fairly
613-- concurrent manner irrespective of the type, notice that we have
614-- not used the 'parallely' combinator in the following example:
617-- main = 'runStream' $ delay 3 \`parallel` delay 2 \`parallel` delay 1
620-- ThreadId 42: Delay 1
621-- ThreadId 41: Delay 2
622-- ThreadId 40: Delay 3
625-- Note that this style of composition cannot be used to combine infinite
626-- number of streams, as it will lead to an infinite sized scheduling queue.
631-- The 'async' API can be used to create references to asynchronously running
632-- stream computations. We can then use 'uncons' to explore the streams
633-- arbitrarily and then recompose individual elements to create a new stream.
634-- This way we can dynamically decide which stream to explore at any given
635-- time. Take an example of a merge sort of two sorted streams. We need to
636-- keep consuming items from the stream which has the lowest item in the sort
637-- order. This can be achieved using async references to streams. See
638-- "Streamly.Examples.MergeSortedStreams".
642-- We can use 'Monoid' instances to fold a container of streams in the desired
643-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
644-- to fold streams using the polymorphic combine operations:
646-- * 'foldWith' is like 'fold', it folds a 'Foldable' container of streams
647-- using the given composition operator.
648-- * 'foldMapWith' is like 'foldMap', it folds like @foldWith@ but also maps a
649-- function before folding.
650-- * 'forEachWith' is like @foldMapwith@ but the container argument comes before
651-- the function argument.
653-- All of the following are equivalent and start ten concurrent tasks each with
654-- a delay from 1 to 10 seconds, resulting in the printing of each number every
659-- import qualified "Streamly.Prelude" as S
660-- import Control.Concurrent
663-- 'runStream' $ 'parallely' $ foldMap delay [1..10]
664-- 'runStream' $ 'foldWith' 'parallel' (map delay [1..10])
665-- 'runStream' $ 'foldMapWith' 'parallel' delay [1..10]
666-- 'runStream' $ 'forEachWith' 'parallel' [1..10] delay
667-- where delay n = S.'once' $ threadDelay (n * 1000000) >> print n
672-- Till now we discussed ways to apply transformations on a stream or to merge
673-- streams together to create another stream. We mentioned earlier that
674-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
675-- We will now discuss the concept of a nested composition of streams which is
676-- analogous to nested @for@ loops in the imperative paradigm. Functional
677-- programmers call this style of composition a list transformer or @ListT@.
678-- Logic programmers call it a logic monad or non-deterministic composition,
679-- but for ordinary imperative minded people like me it is easier to think in
680-- terms of good old nested @for@ loops.
684-- In functional programmer's parlance the 'Monad' instances of different
685-- 'IsStream' types implement non-determinism, exploring all possible
686-- combination of choices from both the streams. From an imperative
687-- programmer's point of view it behaves like nested loops i.e. for each
688-- element in the first stream and for each element in the second stream
689-- execute the body of the loop.
691-- The 'Monad' instances of 'Stream', 'Costream', 'Coparallel' and 'Parallel'
692-- stream types support different flavors of nested looping. In other words,
693-- they are all variants of list transformer. The nesting behavior of these
694-- types exactly correspond to the way they merge streams as we discussed in
695-- the previous section.
700-- The 'Monad' composition of 'Stream' type behaves like a standard list
701-- transformer. This is the default when we do not use an explicit type
702-- combinator. However, the 'streamly' type combinator can be used to switch to
703-- this style of composition. We will see how this style of composition works
704-- in the following examples.
706-- Let's start with an example with a simple @for@ loop without any nesting.
707-- For simplicity of illustration we are using streams of pure values in all
708-- the examples. However, the streams could also be made of monadic actions
713-- import qualified "Streamly.Prelude" as S
715-- main = 'runStream' $ do
716-- x <- S.'fromFoldable' [3,2,1]
720-- ThreadId 30: Delay 3
721-- ThreadId 30: Delay 2
722-- ThreadId 30: Delay 1
725-- As you can see, the code after the @fromFoldable@ statement is run three
726-- times, once for each value of @x@ drawn from the stream. All the three
727-- iterations are serial and run in the same thread one after another. In
728-- imperative terms this is equivalent to a @for@ loop with three iterations.
730-- A console echo loop copying standard input to standard output can simply be
735-- import qualified "Streamly.Prelude" as S
737-- main = 'runStream' $ forever $ S.once getLine >>= S.once . putStrLn
740-- When multiple streams are composed using this style they nest in a DFS
741-- manner i.e. nested iterations of a loop are executed before we proceed to
742-- the next iteration of the parent loop. This behaves just like nested @for@
743-- loops in imperative programming.
747-- import qualified "Streamly.Prelude" as S
749-- main = 'runStream' $ do
750-- x <- S.'fromFoldable' [1,2]
751-- y <- S.'fromFoldable' [3,4]
752-- S.'once' $ putStrLn $ show (x, y)
761-- Notice that this is analogous to merging streams of type 'Stream' or merging
762-- streams using 'splice'.
766-- The 'Monad' composition of 'Coparallel' type can perform the iterations of a
767-- loop concurrently. Concurrency is demand driven i.e. more concurrent
768-- iterations are started only if the previous iterations are not able to
769-- produce enough output for the consumer of the output stream. This works
770-- exactly the same way as the merging of two streams 'coparallely' works.
771-- This is the concurrent analogue of 'Stream' style monadic composition.
773-- The 'coparallely' type combinator can be used to switch to this
774-- style of composition. Alternatively, a type annotation can be used to
775-- specify the type of the stream as 'Coparallel'.
779-- import "Streamly.Prelude"
781-- main = 'runStream' . 'coparallely' $ do
782-- x <- 'fromFoldable' [3,2,1]
786-- ThreadId 40: Delay 1
787-- ThreadId 39: Delay 2
788-- ThreadId 38: Delay 3
791-- As you can see the code after the @fromFoldable@ statement is run three
792-- times, once for each value of @x@. All the three iterations are concurrent
793-- and run in different threads. The iteration with least delay finishes first.
794-- When compared to imperative programming, this can be viewed as a @for@ loop
795-- with three concurrent iterations.
797-- Concurrency is demand driven just as in the case of 'coparallel' merging.
798-- When multiple streams are composed using this style, the iterations are
799-- triggered in a DFS manner just like 'Stream' i.e. nested iterations are
800-- executed before we proceed to the next iteration at higher level. However,
801-- unlike 'Stream' more than one iterations may be started concurrently based
802-- on the demand from the consumer.
806-- import qualified "Streamly.Prelude" as S
808-- main = 'runStream' . 'coparallely' $ do
809-- x <- S.'fromFoldable' [1,2]
810-- y <- S.'fromFoldable' [3,4]
811-- S.'once' $ putStrLn $ show (x, y)
820-- $interleavedNesting
822-- The 'Monad' composition of 'Costream' type interleaves the iterations of
823-- outer and inner loops in a nested loop composition. This works exactly the
824-- same way as the merging of two streams in 'costreamly' fashion works. The
825-- 'costreamly' type combinator can be used to switch to this style of
826-- composition. Alternatively, a type annotation can be used to specify the
827-- type of the stream as 'Costream'.
831-- import qualified "Streamly.Prelude" as S
833-- main = 'runStream' . 'costreamly' $ do
834-- x <- S.'fromFoldable' [1,2]
835-- y <- S.'fromFoldable' [3,4]
836-- S.once $ putStrLn $ show (x, y)
846-- $fairlyConcurrentNesting
848-- The 'Monad' composition of 'Parallel' type starts all the iterations of a
849-- loop concurrently instead of demand driven as in the case of 'Coparallel'.
850-- This works exactly the same way as the merging streams 'parallely' works.
851-- The 'parallely' type combinator can be used to switch to this style of
852-- composition. Alternatively, a type annotation can be used to specify the
853-- type of the stream as 'Parallel'.
857-- import qualified "Streamly.Prelude" as S
859-- main = 'runStream' . 'parallely' $ do
860-- x <- S.'fromFoldable' [3,2,1]
864-- ThreadId 40: Delay 1
865-- ThreadId 39: Delay 2
866-- ThreadId 38: Delay 3
871-- The streamly code is usually written in a way that is agnostic of the
872-- specific monadic composition type. We use a polymorphic type with a
873-- 'IsStream' type class constraint. When running the stream we can choose the
874-- specific mode of composition. For example take a look at the following code.
878-- import qualified "Streamly.Prelude" as S
880-- composed :: 'IsStream' t => t m a
885-- S.'once' $ putStrLn $ show (sz, cl, sh)
889-- sizes = S.'fromFoldable' [1, 2, 3]
890-- colors = S.'fromFoldable' ["red", "green", "blue"]
891-- shapes = S.'fromFoldable' ["triangle", "square", "circle"]
894-- Now we can interpret this in whatever way we want:
897-- main = 'runStream' . 'streamly' $ composed
898-- main = 'runStream' . 'costreamly' $ composed
899-- main = 'runStream' . 'coparallely' $ composed
900-- main = 'runStream' . 'parallely' $ composed
903-- As an exercise try to figure out the output of this code for each mode of
908-- 'fmap' transforms a stream by mapping a function on all elements of the
909-- stream. 'fmap' behaves in the same way for all stream types, it is always
914-- import qualified "Streamly.Prelude" as S
916-- main = (S.'toList' $ fmap show $ S.'fromFoldable' [1..10]) >>= print
919-- Also see the 'mapM' and 'sequence' functions for mapping actions, in the
920-- "Streamly.Prelude" module.
924-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
925-- zipping applicatives separate types 'ZipStream' and 'ZipParallel' are
928-- The following example uses the 'Stream' applicative, it runs all iterations
929-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
933-- import qualified "Streamly.Prelude" as S
934-- import Control.Concurrent
938-- d n = delay n >> return n
940-- main = (S.'toList' . 'streamly' $ (,) \<$> s1 \<*> s2) >>= print
943-- ThreadId 36: Delay 1
944-- ThreadId 36: Delay 3
945-- ThreadId 36: Delay 4
946-- ThreadId 36: Delay 2
947-- ThreadId 36: Delay 3
948-- ThreadId 36: Delay 4
949-- [(1,3),(1,4),(2,3),(2,4)]
952-- Similalrly 'Costream' applicative runs the iterations in an interleaved
953-- order but since it is serial it takes a total of 17 seconds:
956-- main = (S.'toList' . 'costreamly' $ (,) \<$> s1 \<*> s2) >>= print
959-- ThreadId 36: Delay 1
960-- ThreadId 36: Delay 3
961-- ThreadId 36: Delay 2
962-- ThreadId 36: Delay 3
963-- ThreadId 36: Delay 4
964-- ThreadId 36: Delay 4
965-- [(1,3),(2,3),(1,4),(2,4)]
968-- 'Coparallel' can run the iterations concurrently and therefore takes a total
969-- of 10 seconds (1 + 2 + 3 + 4):
972-- main = (S.'toList' . 'coparallely' $ (,) \<$> s1 \<*> s2) >>= print
975-- ThreadId 34: Delay 1
976-- ThreadId 36: Delay 2
977-- ThreadId 35: Delay 3
978-- ThreadId 36: Delay 3
979-- ThreadId 35: Delay 4
980-- ThreadId 36: Delay 4
981-- [(1,3),(2,3),(1,4),(2,4)]
984-- Similalrly 'Parallel' as well can run the iterations concurrently and
985-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
988-- main = (S.'toList' . 'parallely' $ (,) \<$> s1 \<*> s2) >>= print
991-- ThreadId 34: Delay 1
992-- ThreadId 36: Delay 2
993-- ThreadId 35: Delay 3
994-- ThreadId 36: Delay 3
995-- ThreadId 35: Delay 4
996-- ThreadId 36: Delay 4
997-- [(1,3),(2,3),(1,4),(2,4)]
1000-- $compositionSummary
1002-- The following table summarizes the types for monadic compositions and the
1003-- operators for sum style compositions. This table captures the essence of
1007-- +-----+--------------+--------------+
1008-- | | Serial | Concurrent |
1009-- +=====+==============+==============+
1010-- | DFS | 'Stream' | 'Coparallel' |
1011-- | +--------------+--------------+
1012-- | | 'splice' | 'coparallel' |
1013-- +-----+--------------+--------------+
1014-- | BFS | 'Costream' | 'Parallel' |
1015-- | +--------------+--------------+
1016-- | | 'cosplice' | 'parallel' |
1017-- +-----+--------------+--------------+
1022-- Zipping is a special transformation where the corresponding elements of two
1023-- streams are combined together using a zip function producing a new stream of
1024-- outputs. Two different types are provided for serial and concurrent zipping.
1025-- These types provide an applicative instance that can be used to lift
1026-- functions to zip the argument streams.
1027-- Also see the zipping functions in the "Streamly.Prelude" module.
1031-- The applicative instance of 'ZipStream' type zips streams serially.
1032-- 'zipStreamly' type combinator can be used to switch to serial applicative
1037-- import qualified "Streamly.Prelude" as S
1038-- import Control.Concurrent
1040-- d n = delay n >> return n
1041-- s1 = 'streamly' $ d 1 <> d 2
1042-- s2 = 'streamly' $ d 3 <> d 4
1044-- main = (S.'toList' . 'zipStreamly' $ (,) \<$> s1 \<*> s2) >>= print
1047-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1048-- everything runs serially:
1051-- ThreadId 29: Delay 1
1052-- ThreadId 29: Delay 3
1053-- ThreadId 29: Delay 2
1054-- ThreadId 29: Delay 4
1060-- The applicative instance of 'ZipParallel' type zips streams concurrently.
1061-- 'zipParallely' type combinator can be used to switch to parallel applicative
1067-- import qualified "Streamly.Prelude" as S
1068-- import Control.Concurrent
1069-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
1071-- d n = delay n >> return n
1072-- s1 = 'streamly' $ d 1 <> d 2
1073-- s2 = 'streamly' $ d 3 <> d 4
1076-- hSetBuffering stdout LineBuffering
1077-- (S.'toList' . 'zipParallely' $ (,) \<$> s1 \<*> s2) >>= print
1080-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1081-- are produced concurrently, and 2 and 4 are produced concurrently:
1084-- ThreadId 32: Delay 1
1085-- ThreadId 32: Delay 2
1086-- ThreadId 33: Delay 3
1087-- ThreadId 33: Delay 4
1093-- When writing concurrent programs there are two distinct places where the
1094-- programmer can control the concurrency. First, when /composing/ a stream by
1095-- merging multiple streams we can choose an appropriate sum style operators to
1096-- combine them concurrently or serially. Second, when /processing/ a stream in
1097-- a monadic composition we can choose one of the monad composition types to
1098-- choose the desired type of concurrency.
1100-- In the following example the squares of @x@ and @y@ are computed
1101-- concurrently using the 'coparallel' operation and the square roots of their
1102-- sum are computed serially because of the 'streamly' combinator. We can
1103-- choose different combinators for the monadic processing and the stream
1104-- generation, to control the concurrency. We can also use the 'coparallely'
1105-- combinator instead of explicitly folding with 'coparallel'.
1109-- import qualified "Streamly.Prelude" as S
1110-- import Data.List (sum)
1114-- $ 'streamly' -- Serial monadic processing (sqrt below)
1116-- x2 \<- 'forEachWith' 'coparallel' [1..100] $ -- Concurrent @"for"@ loop
1117-- \\x -> return $ x * x -- body of the loop
1118-- y2 \<- 'forEachWith' 'coparallel' [1..100] $
1119-- \\y -> return $ y * y
1120-- return $ sqrt (x2 + y2)
1124-- You can see how this directly maps to the imperative style
1125-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1126-- and operators instead of the ugly pragmas.
1128-- For more concurrent programming examples see,
1129-- "Streamly.Examples.ListDirRecursive", "Streamly.Examples.MergeSortedStreams"
1130-- and "Streamly.Examples.SearchEngineQuery".
1134-- Reactive programming is nothing but concurrent streaming which is what
1135-- streamly is all about. With streamly we can generate streams of events,
1136-- merge streams that are generated concurrently and process events
1137-- concurrently. We can do all this without any knowledge about the specifics
1138-- of the implementation of concurrency. In the following example you will see
1139-- that the code is just regular Haskell code without much streamly APIs used
1140-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1143-- This application has two independent and concurrent sources of event
1144-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1145-- events that deteriorate the health of the character in the game.
1146-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1147-- health improves and the game continues.
1150-- {-\# LANGUAGE FlexibleContexts #-}
1153-- import Control.Concurrent (threadDelay)
1154-- import Control.Monad (when)
1155-- import Control.Monad.State (MonadState, get, modify, runStateT)
1156-- import Data.Semigroup (cycle1)
1158-- data Event = Harm Int | Heal Int | Quit deriving (Show)
1160-- userAction :: MonadIO m => 'StreamT' m Event
1161-- userAction = cycle1 $ liftIO askUser
1164-- command <- getLine
1166-- "potion" -> return (Heal 10)
1167-- "quit" -> return Quit
1168-- _ -> putStrLn "What?" >> askUser
1170-- acidRain :: MonadIO m => 'StreamT' m Event
1171-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
1173-- game :: ('MonadParallel' m, MonadState Int m) => 'StreamT' m ()
1175-- event \<- userAction \`parallel` acidRain
1177-- Harm n -> modify $ \\h -> h - n
1178-- Heal n -> modify $ \\h -> h + n
1179-- Quit -> fail "quit"
1182-- when (h <= 0) $ fail "You die!"
1183-- liftIO $ putStrLn $ "Health = " ++ show h
1186-- putStrLn "Your health is deteriorating due to acid rain,\\
1187-- \\ type \\"potion\\" or \\"quit\\""
1188-- _ <- runStateT ('runStream' game) 60
1192-- You can also find the source of this example in
1193-- "Streamly.Examples.AcidRainGame". It has been adapted from Gabriel's
1194-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1196-- This is much simpler compared to the pipes version because of the builtin
1197-- concurrency in streamly. You can also find a SDL based reactive programming
1198-- example adapted from Yampa in "Streamly.Examples.CirclingSquare".
1202-- Streamly is highly optimized for performance, it is designed for serious
1203-- high performing, concurrent and scalable applications. We have created the
1204-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1205-- package which is specifically and carefully designed to measure the
1206-- performance of Haskell streaming libraries fairly and squarely in the right
1207-- way. Streamly performs at par or even better than most streaming libraries
1208-- for serial operations even though it needs to deal with the concurrency
1213-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1215-- Interop with @vector@:
1219-- import qualified Streamly.Prelude as S
1220-- import qualified Data.Vector.Fusion.Stream.Monadic as V
1223-- -- streamly to vector
1224-- V.toList (V.unfoldrM S.uncons (S.fromFoldable [1..3])) >>= print
1226-- -- vector to streamly
1227-- S.toList (S.unfoldrM unconsV (V.fromList [1..3])) >>= print
1233-- then return Nothing
1236-- return $ Just (h, V.tail v)
1239-- Interop with @pipes@:
1243-- import qualified "Streamly.Prelude" as S
1244-- import qualified Pipes as P
1245-- import qualified Pipes.Prelude as P
1248-- -- streamly to pipe
1249-- P.toListM (P.unfoldr unconsS (S.'fromFoldable' [1..3])) >>= print
1251-- -- pipe to streamly
1252-- S.'toList' (S.'unfoldrM' unconsP (P.each [1..3])) >>= print
1255-- -- Adapt P.next to return a Maybe instead of Either
1256-- unconsP p = P.next p >>= either (\\_ -> return Nothing) (return . Just)
1258-- -- Adapt S.uncons to return an Either instead of Maybe
1259-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1262-- Interop with @streaming@:
1266-- import qualified "Streamly.Prelude" as S
1267-- import qualified Streaming as SG
1268-- import qualified Streaming.Prelude as SG
1271-- -- streamly to streaming
1272-- SG.toList (SG.unfoldr unconsS (S.'fromFoldable' [1..3])) >>= print
1274-- -- streaming to streamly
1275-- S.'toList' (S.unfoldrM SG.uncons (SG.each [1..3])) >>= print
1279-- -- Adapt S.uncons to return an Either instead of Maybe
1280-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1283-- Interop with @conduit@:
1287-- import qualified "Streamly.Prelude" as S
1288-- import qualified Data.Conduit as C
1289-- import qualified Data.Conduit.List as C
1290-- import qualified Data.Conduit.Combinators as C
1293-- -- streamly to conduit
1294-- C.runConduit (C.unfoldM S.'uncons' (S.'fromFoldable' [1..3]) C..| C.sinkList) >>= print
1296-- -- It seems there is no way out of a conduit as it does not provide an
1297-- -- uncons or a tail function.
1302-- List transformers and logic programming monads also provide a product style
1303-- composition similar to streamly, however streamly generalizes it with the
1304-- time dimension; allowing streams to be composed in an asynchronous and
1305-- concurrent fashion in many different ways. It also provides multiple
1306-- alternative ways of composing streams e.g. serial, interleaved or
1309-- This seemingly simple addition of asynchronicity and concurrency to product
1310-- style streaming composition unifies a number of disparate abstractions into
1311-- one powerful, concise and elegant abstraction. A wide variety of
1312-- programming problems can be solved elegantly with this abstraction. In
1313-- particular, it unifies three major programming domains namely
1314-- non-deterministic (logic) programming, concurrent programming and functional
1315-- reactive programming. In other words, you can do everything with this one
1316-- abstraction that you could do with the popular libraries listed under these
1317-- categories in the list below.
1320-- +-----------------+----------------+
1321-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1322-- | +----------------+
1323-- | | <https://hackage.haskell.org/package/list-t list-t> |
1324-- | +----------------+
1325-- | | <https://hackage.haskell.org/package/logict logict> |
1326-- +-----------------+----------------+
1327-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1328-- | +----------------+
1329-- | | <https://hackage.haskell.org/package/streaming streaming> |
1330-- | +----------------+
1331-- | | <https://hackage.haskell.org/package/pipes pipes> |
1332-- | +----------------+
1333-- | | <https://hackage.haskell.org/package/conduit conduit> |
1334-- +-----------------+----------------+
1335-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1336-- | +----------------+
1337-- | | <https://hackage.haskell.org/package/transient transient> |
1338-- +-----------------+----------------+
1339-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1340-- | +----------------+
1341-- | | <https://hackage.haskell.org/package/dunai dunai> |
1342-- | +----------------+
1343-- | | <https://hackage.haskell.org/package/reflex reflex> |
1344-- +-----------------+----------------+
1347-- Streamly is a list-transformer. It provides all the functionality provided
1348-- by any of the list transformer and logic programming packages listed above.
1349-- In addition, Streamly naturally integrates the concurrency dimension to the
1350-- basic list transformer functionality.
1352-- When it comes to streaming, in terms of the streaming API streamly is almost
1353-- identical to the vector package. Streamly, vector and streaming packages all
1354-- represent a stream as data and are therefore similar in the fundamental
1355-- approach to streaming. The fundamental difference is that streamly adds
1356-- concurrency support and the monad instance provides concurrent looping.
1357-- Other streaming libraries like pipes, conduit and machines represent and
1358-- compose stream processors rather than the stream data and therefore fall in
1359-- another class of streaming libraries and have comparatively more complicated
1362-- When it comes to concurrency, streamly can do everything that the @async@
1363-- package can do and more. async provides applicative concurrency whereas
1364-- streamly provides both applicative and monadic concurrency. The
1365-- 'ZipParallel' type behaves like the applicative instance of async. In
1366-- comparison to transient streamly has a first class streaming interface and
1367-- is a monad transformer that can be used universally in any Haskell monad
1368-- transformer stack. Streamly was in fact originally inspired by the
1369-- concurrency implementation in @transient@ though it has no resemblence with
1370-- that and takes a lazy pull approach versus transient's strict push approach.
1372-- The non-determinism, concurrency and streaming combination make streamly a
1373-- strong FRP capable library as well. FRP is fundamentally stream of events
1374-- that can be processed concurrently. The example in this tutorial as well as
1375-- the "Streamly.Examples.CirclingSquare" example from Yampa demonstrate the
1376-- basic FRP capability of streamly. In core concepts streamly is strikingly
1377-- similar to @dunai@. dunai was designed from a FRP perspective and streamly
1378-- was originally designed from a concurrency perspective. However, both have
1379-- similarity at the core.
1383-- * Read the documentation of "Streamly" module
1384-- * Read the documentation of "Streamly.Prelude" module
1385-- * See the examples in the "examples" directory of the package
1386-- * 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