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, short for stream concurrently, combines the essence of
10-- non-determinism, streaming and concurrency in functional programming.
11-- Concurrent and non-concurrent applications are almost indistinguishable,
12-- concurrency capability does not at all impact the performance of
13-- non-concurrent case.
14-- Streaming enables writing modular, composable and scalable applications with
15-- ease and concurrency allows you to make them scale and perform well.
16-- Streamly enables writing concurrent applications without being aware of
17-- threads or synchronization. No explicit thread control is needed, where
18-- applicable the concurrency rate is automatically controlled based on the
19-- demand by the consumer. However, combinators are provided to fine tune the
20-- concurrency control.
21-- Streaming and concurrency together enable expressing reactive applications
22-- conveniently. See "Streamly.Examples" for a simple SDL based FRP example.
24-- Streamly streams are very much like the Haskell lists and most of the
25-- functions that work on lists have a counterpart that works on streams.
26-- However, streamly streams can be generated, consumed or combined
27-- concurrently. In this tutorial we will go over the basic concepts and how to
28-- use the library. The documentation of @Streamly@ module has more details on
29-- core APIs. For more APIs for constructing, folding, filtering, mapping and
30-- zipping etc. see the documentation of "Streamly.Prelude" module. For
31-- examples and other ways to use the library see the module
32-- "Streamly.Examples" as well.
34module Streamly.Tutorial
39 -- ** Generating Streams
42 -- ** Eliminating Streams
45 -- * Transforming Streams
54 -- *** Serial Composition ('streamly')
57 -- *** Serial Interleaved Composition ('costreamly')
60 -- *** Left Biased Concurrent Composition ('coparallely')
63 -- *** Round Robin Concurrent Composition ('parallely')
66 -- *** Custom composition
78 -- *** Serial Nesting ('streamly')
81 -- *** Serial Interleaved Nesting ('costreamly')
82 -- $interleavedNesting
84 -- *** Left Biased Concurrent Nesting ('coparallely')
87 -- *** Round Robin Concurrent Nesting ('parallely')
88 -- $fairlyConcurrentNesting
105 -- ** Parallel Zipping
108 -- * Summary of Compositions
109 -- $compositionSummary
111 -- * Monad transformers
112 -- $monadtransformers
114 -- * Concurrent Programming
117 -- * Reactive Programming
123 -- * Interoperation with Streaming Libraries
126 -- * Comparison with Existing Packages
132import Streamly.Prelude
134import Control.Applicative
136import Control.Monad.IO.Class (MonadIO(..))
137import Control.Monad.Trans.Class (MonadTrans (lift))
141-- Streamly provides many different stream types depending on the desired
142-- composition style. The simplest type is 'Stream', @Stream a@ represents a
143-- stream of elements of type @a@ in IO monad. Streams are very much like lists
144-- the only difference is that they are lists of monadic actions instead of
145-- pure values. We can call them monadic lists. You can apply all the
146-- constructs related to lists to monadic streams as well.
150-- To represent streams in an arbitrary monad use the more general monad
151-- transformer types for example the monad transformer type corresponding to
152-- the 'Stream' type is 'StreamT'. @StreamT m a@ represents a stream of values
153-- of type 'a' in some underlying monad 'm'. For example, @StreamT IO Int@ is a
154-- stream of 'Int' in 'IO' monad. In fact, the type 'Stream' is a synonym for
157-- Similarly we have monad transformer types for other stream types as well viz.
158-- 'CostreamT', 'CoparallelT' and 'ParallelT'.
160-- To lift a value from an underlying monad in a monad transformer stack into a
161-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
164-- > runStream $ liftIO $ putStrLn "Hello world!"
166-- > runStream $ lift $ putStrLn "Hello world!"
173-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
174-- a monadic action at the head of the stream.
179-- > toList $ getLine |: getLine |: nil
185-- To create a singleton stream from a pure value use 'pure' and to create a
186-- singleton stream from a monadic action use 'once'.
191-- > toList $ once getLine
196-- To create a stream from pure values in a 'Foldable' container use
197-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
200-- > toList $ fromFoldable [1..3]
202-- > toList $ Prelude.foldr cons nil [1..3]
206-- To create a stream from monadic actions in a 'Foldable' container just use a
207-- right fold using 'consM' and 'nil':
210-- > runStream $ Prelude.foldr (|:) nil [putStr "Hello ", putStrLn "world!"]
214-- For more ways to construct a stream see the module "Streamly.Prelude".
218-- We have already seen 'runStream' and 'toList' to eliminate a stream in the
219-- examples above. 'runStream' runs a stream discarding the results i.e. only
220-- for effects. 'toList' runs the stream and collects the results in a list.
222-- For other ways to eliminate a stream see the @Folding@ section in
223-- "Streamly.Prelude" module.
227-- Transformation over a stream is the equivalent of a @for@ loop construct in
228-- imperative paradigm. We iterate over every element in the stream and perform
229-- certain transformations for each element. Transformations may involve
230-- mapping functions over the elements, filtering elements from the stream or
231-- folding all the elements in the stream into a single value. Streamly streams
232-- are exactly like lists and you can perform all the transformations in the
233-- same way as you would on lists.
235-- Here is a simple console echo program that just echoes every input line,
239-- > runStream $ S.repeatM getLine & S.mapM putStrLn
242-- The following code snippet reads lines from standard input, filters blank
243-- lines, drops the first non-blank line, takes the next two, up cases them,
244-- numbers them and prints them:
248-- import qualified Streamly.Prelude as S
249-- import Data.Char (toUpper)
250-- import Data.Function ((&))
254-- & S.filter (not . null)
257-- & fmap (map toUpper)
258-- & S.zipWith (\\n s -> show n ++ " " ++ s) (S.fromFoldable [1..])
264-- We can combine two streams into a single stream using semigroup composition
265-- operation '<>'. Streams can be combined in many different ways as described
266-- in the following sections, the '<>' operation behaves differently depending
267-- on the stream type in effect. Streamly streams can be of many
268-- different types i.e. 'Stream', 'Costream', 'Coparallel', 'Parallel',
269-- 'ZipStream', and 'ZipParallel'. The first four are monads whereas the last
270-- two are applicatives. These types can be freely converted from one to
271-- another to acheive the desired composition style. To force a particular
272-- type of composition we coerce the stream type using the corresponding type
273-- adapting combinator from 'streamly', 'costreamly', 'coparallely',
274-- 'parallely', 'zipStreamly' or 'zipParallely'. The default stream type is
275-- 'Stream' unless you change it by using one of the combinators.
277-- To illustrate the concurrent vs serial composition aspects, we will use the
278-- following @delay@ function to introduce a delay specified in seconds.
282-- import "Streamly.Prelude"(once)
283-- import Control.Concurrent
285-- delay n = once $ do
286-- threadDelay (n * 1000000)
288-- putStrLn (show tid ++ ": Delay " ++ show n)
293-- The 'Semigroup' operation '<>' of the 'Stream' type combines the two streams
294-- in a serial manner. We use the 'streamly' type combinator to effect 'Stream'
295-- style of composition. We can also use the 'Stream' type annotation for the
296-- stream type to acheive the same effect. However, since 'Stream' is the
297-- default type unless explicitly specified by using a combinator, we can
298-- omit using an explicit combinator or type annotation for this style of
301-- When two streams with multiple elements are combined in this manner, the
302-- monadic actions in the two streams are performed sequentially i.e. first all
303-- actions in the first stream are performed sequentially and then all actions
304-- in the second stream are performed sequentially. The following example
305-- prints the sequence 1, 2, 3, 4:
308-- main = 'runStream' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
317-- All actions in both the streams are performed serially in the same thread.
318-- In the following example we can see that all actions are performed in the
319-- same thread and take a combined total of 6 seconds:
322-- main = 'runStream' $ delay 3 <> delay 2 <> delay 1
325-- ThreadId 36: Delay 3
326-- ThreadId 36: Delay 2
327-- ThreadId 36: Delay 1
330-- The polymorphic version of the 'Stream' binary operation '<>' is called
331-- 'splice'. We can use 'splice' to join streams in a sequential manner
332-- irrespective of the type of stream:
335-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`splice` (print 3 |: print 4 |: nil)
340-- The 'Semigroup' operation '<>' of the 'Costream' type combines the two
341-- streams in a serial interleaved manner. We use the 'costreamly' type
342-- combinator to effect 'Costream' style of composition. We can also use the
343-- 'Costream' type annotation for the stream type to acheive the same effect.
345-- When two streams with multiple elements are combined in this manner, the
346-- monadic actions in the two streams are performed in an interleaved manner
347-- i.e. one action from the first stream is performed and its output yielded
348-- and then one action from the second stream is performed and its output
349-- yielded, and so on. The following example prints the sequence 1, 3, 2, 4
352-- main = 'runStream' . 'costreamly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
361-- Even though the monadic actions of the two streams are performed in an
362-- interleaved manner they are all performed serially in the same thread. In
363-- the following example we can see that all actions are performed in the same
364-- thread and take a combined total of 6 seconds:
367-- main = 'runStream' . 'costreamly' $ 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 'Costream' binary operation '<>' is called
376-- 'cosplice'. We can use 'cosplice' to join streams in an interleaved manner
377-- irrespective of the type, notice that we have not used the 'costreamly'
378-- combinator in the following example:
381-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`cosplice` (print 3 |: print 4 |: nil)
390-- Note that this composition cannot be used to fold infinite number of streams
391-- since it requires preserving the state until a stream is finished.
395-- The 'Semigroup' operation '<>' of the 'Coparallel' type combines the two
396-- streams in a left biased concurrent manner. We use the 'coparallel' type
397-- combinator to effect 'Coparallel' style of composition. We can also use the
398-- 'Coparallel' type annotation for the stream type to acheive the same effect.
400-- When two streams with multiple elements are combined in this manner, the
401-- monadic actions in the two streams can be performed concurrently
402-- /when needed/ with a bias towards the left stream i.e. actions from the left
403-- stream are performed sequentially, if the left stream blocks or cannot
404-- generate enough output to keep the consumer of the stream busy then it can
405-- execute actions from the right stream concurrently. The outputs of the
406-- concurrent actions are yielded by the composite stream in the order in which
407-- they complete. This can be thought of as the concurrent analogue of the
408-- 'Stream' style serial composition. If concurrent execution is not trigerred
409-- by consumer demand then it would execute actions exactly in the same order
410-- as 'Stream'. In the following example the first stream does not block,
411-- therefore the first stream is completely exhausted before the second.
414-- main = 'runStream' . 'coparallely' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
423-- If the first stream blocks, we can yield from the second. In the example
424-- below each yield in the stream has a constant delay of 1 second therefore 1
425-- and 3 would be yielded first and then 2 and 4 would be yielded.
428-- main = 'runStream' . 'coparallely' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
429-- where p n = threadDelay 1000000 >> print n
438-- In the following example we can see that new threads are started when a
439-- computation blocks. Notice that the output from the stream with the
440-- shortest delay is printed first:
443-- main = 'runStream' . 'coparallely' $ delay 3 '<>' delay 2 '<>' delay 1
446-- ThreadId 42: Delay 1
447-- ThreadId 41: Delay 2
448-- ThreadId 40: Delay 3
451-- When we have a tree of computations composed using this style, the tree is
452-- traversed in DFS style just like the serial style, the only difference is
453-- that here we can move on to executing the next stream if a stream blocks.
454-- However, we will not start new threads if we have sufficient output to
455-- saturate the consumer. This is why we call it left-biased demand driven or
456-- adaptive concurrency style, the concurrency tends to stay on the left side
457-- of the composition as long as possible. More threads are started based on
458-- the pull rate of the consumer. The following example prints an output every
459-- second as all of the actions are concurrent.
462-- main = 'runStream' . 'coparallely' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
471-- All the computations may even run in a single thread when more threads are
472-- not needed. As you can see, in the following example the computations are
473-- run in a single thread one after another, because none of them blocks.
474-- However, if the thread consuming the stream were faster than the producer
475-- then it would have started parallel threads for each computation to keep up
476-- even if none of them blocks:
479-- main = 'runStream' . 'coparallely' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
480-- where traced m = once (myThreadId >>= print) >> return m
488-- Note that the order of printing in the above examples may change due to
489-- variations in scheduling latencies for concurrent threads.
491-- The polymorphic version of the 'Coparallel' binary operation '<>' is called
492-- 'coparallel'. We can use 'coparallel' to join streams in a left biased
493-- adaptively concurrent manner irrespective of the type, notice that we have
494-- not used the 'coparallely' combinator in the following example:
497-- main = 'runStream' $ delay 3 \`coparallel` delay 2 \`coparallel` delay 1
500-- ThreadId 42: Delay 1
501-- ThreadId 41: Delay 2
502-- ThreadId 40: Delay 3
505-- Since the concurrency provided by this operator is demand driven it cannot
506-- be used when the composed computations start timers that are relative to
507-- each other because all computations may not be started at the same time and
508-- therefore timers in all of them may not start at the same time. When
509-- relative timing among all computations is important or when we need to start
510-- all computations at once for any reason 'Parallel' style must be used
513-- 'Coparallel' style should be preferred over 'Parallel' unless you really
514-- need 'Parallel'. It utilizes the resources optimally. It should be used when
515-- we know that the computations can run in parallel but we do not care if they
516-- actually run in parallel or not, that decision can be left to the scheduler
517-- based on demand. Also, note that this operator can be used to fold infinite
518-- number of streams in contrast to the 'Parallel' style, because it does not
519-- require us to run all of them at the same time in a fair manner.
523-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
524-- streams in a fairly concurrent manner with round robin scheduling. We use
525-- the 'parallel' type combinator to effect 'Parallel' style of composition. We
526-- can also use the 'Parallel' type annotation for the stream type to acheive
529-- When two streams with multiple elements are combined in this manner, the
530-- monadic actions in both the streams are performed concurrently with a fair
531-- round robin scheduling. The outputs are yielded in the order in which the
532-- actions complete. This can be thought of as the concurrent analogue of the
533-- 'Costream' style serial composition.
535-- The following example sends a query to all the three search engines in
536-- parallel and prints the name of the search engines in the order in which the
541-- import Streamly.Prelude (once)
542-- import Network.HTTP.Simple
544-- main = 'runStream' . 'parallely' $ google \<> bing \<> duckduckgo
546-- google = get "https://www.google.com/search?q=haskell"
547-- bing = get "https://www.bing.com/search?q=haskell"
548-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
549-- get s = once (httpNoBody (parseRequest_ s) >> putStrLn (show s))
552-- In the following example we can see that the fairly parallel composition
553-- yields outputs in the same order as the 'Costream' composition, though this
554-- is not always guaranteed because of thread scheduling variances.
557-- main = 'runStream' . 'parallely' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
566-- The polymorphic version of the 'Parallel' binary operation '<>' is called
567-- 'parallel'. We can use 'parallel' to join streams in a fairly
568-- concurrent manner irrespective of the type, notice that we have
569-- not used the 'parallely' combinator in the following example:
572-- main = 'runStream' $ delay 3 \`parallel` delay 2 \`parallel` delay 1
575-- ThreadId 42: Delay 1
576-- ThreadId 41: Delay 2
577-- ThreadId 40: Delay 3
580-- Note that this style of composition cannot be used to combine infinite
581-- number of streams, as it will lead to an infinite sized scheduling queue.
586-- The 'async' API can be used to create references to asynchronously running
587-- stream computations. We can then use 'uncons' to explore the streams
588-- arbitrarily and then recompose individual elements to create a new stream.
589-- This way we can dynamically decide which stream to explore at any given
590-- time. Take an example of a merge sort of two sorted streams. We need to
591-- keep consuming items from the stream which has the lowest item in the sort
592-- order. This can be achieved using async references to streams. See
593-- "Streamly.Examples.MergeSortedStreams".
597-- We can use 'Monoid' instances to fold a container of streams in the desired
598-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
599-- to fold streams using the polymorphic combine operations:
601-- * 'foldWith' is like 'fold', it folds a 'Foldable' container of streams
602-- using the given composition operator.
603-- * 'foldMapWith' is like 'foldMap', it folds like @foldWith@ but also maps a
604-- function before folding.
605-- * 'forEachWith' is like @foldMapwith@ but the container argument comes before
606-- the function argument.
608-- All of the following are equivalent and start ten concurrent tasks each with
609-- a delay from 1 to 10 seconds, resulting in the printing of each number every
614-- import "Streamly.Prelude"
615-- import Control.Concurrent
618-- runStream $ parallely $ foldMap delay [1..10]
619-- runStream $ 'foldWith' parallel (map delay [1..10])
620-- runStream $ 'foldMapWith' parallel delay [1..10]
621-- runStream $ 'forEachWith' parallel [1..10] delay
622-- where delay n = once $ threadDelay (n * 1000000) >> print n
627-- Till now we discussed ways to apply transformations on a stream or to merge
628-- streams together to create another stream. We mentioned earlier that
629-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
630-- We will now discuss the concept of a nested composition of streams which is
631-- analogous to nested @for@ loops in the imperative paradigm. Functional
632-- programmers call this style of composition a list transformer or @ListT@.
633-- Logic programmers call it a logic monad or non-deterministic composition,
634-- but for ordinary imperative minded people like me it is easier to think in
635-- terms of good old nested @for@ loops.
639-- In functional programmer's parlance the 'Monad' instances of different
640-- 'IsStream' types implement non-determinism, exploring all possible
641-- combination of choices from both the streams. From an imperative
642-- programmer's point of view it behaves like nested loops i.e. for each
643-- element in the first stream and for each element in the second stream
644-- execute the body of the loop.
646-- The 'Monad' instances of 'Stream', 'Costream', 'Coparallel' and 'Parallel'
647-- stream types support different flavors of nested looping. In other words,
648-- they are all variants of list transformer. The nesting behavior of these
649-- types exactly correspond to the way they merge streams as we discussed in
650-- the previous section.
655-- The 'Monad' composition of 'Stream' type behaves like a standard list
656-- transformer. This is the default when we do not use an explicit type
657-- combinator. However, the 'streamly' type combinator can be used to switch to
658-- this style of composition. We will see how this style of composition works
659-- in the following examples.
661-- Let's start with an example with a simple @for@ loop without any nesting.
662-- For simplicity of illustration we are using streams of pure values in all
663-- the examples. However, the streams could also be made of monadic actions
668-- import "Streamly.Prelude"
670-- main = 'runStream' $ do
671-- x <- 'fromFoldable' [3,2,1]
675-- ThreadId 30: Delay 3
676-- ThreadId 30: Delay 2
677-- ThreadId 30: Delay 1
680-- As you can see, the code after the @fromFoldable@ statement is run three
681-- times, once for each value of @x@ drawn from the stream. All the three
682-- iterations are serial and run in the same thread one after another. In
683-- imperative terms this is equivalent to a @for@ loop with three iterations.
685-- A console echo loop copying standard input to standard output can simply be
691-- main = 'runStream' $ forever $ once getLine >>= once . putStrLn
694-- When multiple streams are composed using this style they nest in a DFS
695-- manner i.e. nested iterations of a loop are executed before we proceed to
696-- the next iteration of the parent loop. This behaves just like nested @for@
697-- loops in imperative programming.
701-- import "Streamly.Prelude"
703-- main = 'runStream' $ do
704-- x <- 'fromFoldable' [1,2]
705-- y <- 'fromFoldable' [3,4]
706-- once $ putStrLn $ show (x, y)
715-- Notice that this is analogous to merging streams of type 'Stream' or merging
716-- streams using 'splice'.
720-- The 'Monad' composition of 'Coparallel' type can perform the iterations of a
721-- loop concurrently. Concurrency is demand driven i.e. more concurrent
722-- iterations are started only if the previous iterations are not able to
723-- produce enough output for the consumer of the output stream. This works
724-- exactly the same way as the merging of two streams 'coparallely' works.
725-- This is the concurrent analogue of 'Stream' style monadic composition.
727-- The 'coparallely' type combinator can be used to switch to this
728-- style of composition. Alternatively, a type annotation can be used to
729-- specify the type of the stream as 'Coparallel'.
733-- import "Streamly.Prelude"
735-- main = 'runStream' . 'coparallely' $ do
736-- x <- 'fromFoldable' [3,2,1]
740-- ThreadId 40: Delay 1
741-- ThreadId 39: Delay 2
742-- ThreadId 38: Delay 3
745-- As you can see the code after the @fromFoldable@ statement is run three
746-- times, once for each value of @x@. All the three iterations are concurrent
747-- and run in different threads. The iteration with least delay finishes first.
748-- When compared to imperative programming, this can be viewed as a @for@ loop
749-- with three concurrent iterations.
751-- Concurrency is demand driven just as in the case of 'coparallel' merging.
752-- When multiple streams are composed using this style, the iterations are
753-- triggered in a DFS manner just like 'Stream' i.e. nested iterations are
754-- executed before we proceed to the next iteration at higher level. However,
755-- unlike 'Stream' more than one iterations may be started concurrently based
756-- on the demand from the consumer.
760-- import "Streamly.Prelude"
762-- main = 'runStream' . 'coparallely' $ do
763-- x <- 'fromFoldable' [1,2]
764-- y <- 'fromFoldable' [3,4]
765-- once $ putStrLn $ show (x, y)
774-- $interleavedNesting
776-- The 'Monad' composition of 'Costream' type interleaves the iterations of
777-- outer and inner loops in a nested loop composition. This works exactly the
778-- same way as the merging of two streams in 'costreamly' fashion works. The
779-- 'costreamly' type combinator can be used to switch to this style of
780-- composition. Alternatively, a type annotation can be used to specify the
781-- type of the stream as 'Costream'.
785-- import "Streamly.Prelude"
787-- main = 'runStream' . 'costreamly' $ do
788-- x <- 'fromFoldable' [1,2]
789-- y <- 'fromFoldable' [3,4]
790-- once $ putStrLn $ show (x, y)
800-- $fairlyConcurrentNesting
802-- The 'Monad' composition of 'Parallel' type starts all the iterations of a
803-- loop concurrently instead of demand driven as in the case of 'Coparallel'.
804-- This works exactly the same way as the merging streams 'parallely' works.
805-- The 'parallely' type combinator can be used to switch to this style of
806-- composition. Alternatively, a type annotation can be used to specify the
807-- type of the stream as 'Parallel'.
811-- import "Streamly.Prelude"
813-- main = 'runStream' . 'parallely' $ do
814-- x <- 'fromFoldable' [3,2,1]
818-- ThreadId 40: Delay 1
819-- ThreadId 39: Delay 2
820-- ThreadId 38: Delay 3
825-- The streamly code is usually written in a way that is agnostic of the
826-- specific monadic composition type. We use a polymorphic type with a
827-- 'IsStream' type class constraint. When running the stream we can choose the
828-- specific mode of composition. For example take a look at the following code.
832-- import "Streamly.Prelude"
834-- composed :: 'IsStream' t => t m a
839-- once $ putStrLn $ show (sz, cl, sh)
843-- sizes = 'fromFoldable' [1, 2, 3]
844-- colors = 'fromFoldable' ["red", "green", "blue"]
845-- shapes = 'fromFoldable' ["triangle", "square", "circle"]
848-- Now we can interpret this in whatever way we want:
851-- main = 'runStream' . 'streamly' $ composed
852-- main = 'runStream' . 'costreamly' $ composed
853-- main = 'runStream' . 'coparallely' $ composed
854-- main = 'runStream' . 'parallely' $ composed
857-- As an exercise try to figure out the output of this code for each mode of
862-- 'fmap' transforms a stream by mapping a function on all elements of the
863-- stream. 'fmap' behaves in the same way for all stream types, it is always
869-- main = ('toList' $ fmap show $ 'fromFoldable' [1..10]) >>= print
872-- Also see the 'mapM' and 'sequence' functions for mapping actions, in the
873-- "Streamly.Prelude" module.
877-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
878-- zipping applicatives separate types 'ZipStream' and 'ZipParallel' are
881-- The following example uses the 'Stream' applicative, it runs all iterations
882-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
886-- import "Streamly.Prelude"
887-- import Control.Concurrent
891-- d n = delay n >> return n
893-- main = ('toList' . 'streamly' $ (,) \<$> s1 \<*> s2) >>= print
896-- ThreadId 36: Delay 1
897-- ThreadId 36: Delay 3
898-- ThreadId 36: Delay 4
899-- ThreadId 36: Delay 2
900-- ThreadId 36: Delay 3
901-- ThreadId 36: Delay 4
902-- [(1,3),(1,4),(2,3),(2,4)]
905-- Similalrly 'Costream' applicative runs the iterations in an interleaved
906-- order but since it is serial it takes a total of 17 seconds:
909-- main = ('toList' . 'costreamly' $ (,) \<$> s1 \<*> s2) >>= print
912-- ThreadId 36: Delay 1
913-- ThreadId 36: Delay 3
914-- ThreadId 36: Delay 2
915-- ThreadId 36: Delay 3
916-- ThreadId 36: Delay 4
917-- ThreadId 36: Delay 4
918-- [(1,3),(2,3),(1,4),(2,4)]
921-- 'Coparallel' can run the iterations concurrently and therefore takes a total
922-- of 10 seconds (1 + 2 + 3 + 4):
925-- main = ('toList' . 'coparallely' $ (,) \<$> s1 \<*> s2) >>= print
928-- ThreadId 34: Delay 1
929-- ThreadId 36: Delay 2
930-- ThreadId 35: Delay 3
931-- ThreadId 36: Delay 3
932-- ThreadId 35: Delay 4
933-- ThreadId 36: Delay 4
934-- [(1,3),(2,3),(1,4),(2,4)]
937-- Similalrly 'Parallel' as well can run the iterations concurrently and
938-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
941-- main = ('toList' . 'parallely' $ (,) \<$> s1 \<*> s2) >>= print
944-- ThreadId 34: Delay 1
945-- ThreadId 36: Delay 2
946-- ThreadId 35: Delay 3
947-- ThreadId 36: Delay 3
948-- ThreadId 35: Delay 4
949-- ThreadId 36: Delay 4
950-- [(1,3),(2,3),(1,4),(2,4)]
953-- $compositionSummary
955-- The following table summarizes the types for monadic compositions and the
956-- operators for sum style compositions. This table captures the essence of
960-- +-----+--------------+--------------+
961-- | | Serial | Concurrent |
962-- +=====+==============+==============+
963-- | DFS | 'Stream' | 'Coparallel' |
964-- | +--------------+--------------+
965-- | | 'splice' | 'coparallel' |
966-- +-----+--------------+--------------+
967-- | BFS | 'Costream' | 'Parallel' |
968-- | +--------------+--------------+
969-- | | 'cosplice' | 'parallel' |
970-- +-----+--------------+--------------+
975-- Zipping is a special transformation where the corresponding elements of two
976-- streams are combined together using a zip function producing a new stream of
977-- outputs. Two different types are provided for serial and concurrent zipping.
978-- These types provide an applicative instance that can be used to lift
979-- functions to zip the argument streams.
980-- Also see the zipping functions in the "Streamly.Prelude" module.
984-- The applicative instance of 'ZipStream' type zips streams serially.
985-- 'zipStreamly' type combinator can be used to switch to serial applicative
990-- import "Streamly.Prelude"
991-- import Control.Concurrent
993-- d n = delay n >> return n
994-- s1 = 'streamly' $ d 1 <> d 2
995-- s2 = 'streamly' $ d 3 <> d 4
997-- main = ('toList' . 'zipStreamly' $ (,) \<$> s1 \<*> s2) >>= print
1000-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1001-- everything runs serially:
1004-- ThreadId 29: Delay 1
1005-- ThreadId 29: Delay 3
1006-- ThreadId 29: Delay 2
1007-- ThreadId 29: Delay 4
1013-- The applicative instance of 'ZipParallel' type zips streams concurrently.
1014-- 'zipParallely' type combinator can be used to switch to parallel applicative
1020-- import "Streamly.Prelude"
1021-- import Control.Concurrent
1022-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
1024-- d n = delay n >> return n
1025-- s1 = 'streamly' $ d 1 <> d 2
1026-- s2 = 'streamly' $ d 3 <> d 4
1029-- hSetBuffering stdout LineBuffering
1030-- ('toList' . 'zipParallely' $ (,) \<$> s1 \<*> s2) >>= print
1033-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1034-- are produced concurrently, and 2 and 4 are produced concurrently:
1037-- ThreadId 32: Delay 1
1038-- ThreadId 32: Delay 2
1039-- ThreadId 33: Delay 3
1040-- ThreadId 33: Delay 4
1046-- When writing concurrent programs there are two distinct places where the
1047-- programmer can control the concurrency. First, when /composing/ a stream by
1048-- merging multiple streams we can choose an appropriate sum style operators to
1049-- combine them concurrently or serially. Second, when /processing/ a stream in
1050-- a monadic composition we can choose one of the monad composition types to
1051-- choose the desired type of concurrency.
1053-- In the following example the squares of @x@ and @y@ are computed
1054-- concurrently using the 'coparallel' operation and the square roots of their
1055-- sum are computed serially because of the 'streamly' combinator. We can
1056-- choose different combinators for the monadic processing and the stream
1057-- generation, to control the concurrency. We can also use the 'coparallely'
1058-- combinator instead of explicitly folding with 'coparallel'.
1062-- import "Streamly.Prelude" (toList)
1063-- import Data.List (sum)
1067-- $ 'streamly' -- Serial monadic processing (sqrt below)
1069-- x2 \<- 'forEachWith' 'coparallel' [1..100] $ -- Concurrent @"for"@ loop
1070-- \\x -> return $ x * x -- body of the loop
1071-- y2 \<- 'forEachWith' 'coparallel' [1..100] $
1072-- \\y -> return $ y * y
1073-- return $ sqrt (x2 + y2)
1077-- You can see how this directly maps to the imperative style
1078-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1079-- and operators instead of the ugly pragmas.
1081-- For more concurrent programming examples see,
1082-- "Streamly.Examples.ListDirRecursive", "Streamly.Examples.MergeSortedStreams"
1083-- and "Streamly.Examples.SearchEngineQuery".
1087-- Reactive programming is nothing but concurrent streaming which is what
1088-- streamly is all about. With streamly we can generate streams of events,
1089-- merge streams that are generated concurrently and process events
1090-- concurrently. We can do all this without any knowledge about the specifics
1091-- of the implementation of concurrency. In the following example you will see
1092-- that the code is just regular Haskell code without much streamly APIs used
1093-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1096-- This application has two independent and concurrent sources of event
1097-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1098-- events that deteriorate the health of the character in the game.
1099-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1100-- health improves and the game continues.
1103-- {-\# LANGUAGE FlexibleContexts #-}
1106-- import Control.Concurrent (threadDelay)
1107-- import Control.Monad (when)
1108-- import Control.Monad.State (MonadState, get, modify, runStateT)
1109-- import Data.Semigroup (cycle1)
1111-- data Event = Harm Int | Heal Int | Quit deriving (Show)
1113-- userAction :: MonadIO m => 'StreamT' m Event
1114-- userAction = cycle1 $ liftIO askUser
1117-- command <- getLine
1119-- "potion" -> return (Heal 10)
1120-- "quit" -> return Quit
1121-- _ -> putStrLn "What?" >> askUser
1123-- acidRain :: MonadIO m => 'StreamT' m Event
1124-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
1126-- game :: ('MonadParallel' m, MonadState Int m) => 'StreamT' m ()
1128-- event \<- userAction \`parallel` acidRain
1130-- Harm n -> modify $ \\h -> h - n
1131-- Heal n -> modify $ \\h -> h + n
1132-- Quit -> fail "quit"
1135-- when (h <= 0) $ fail "You die!"
1136-- liftIO $ putStrLn $ "Health = " ++ show h
1139-- putStrLn "Your health is deteriorating due to acid rain,\\
1140-- \\ type \\"potion\\" or \\"quit\\""
1141-- _ <- runStateT ('runStream' game) 60
1145-- You can also find the source of this example in
1146-- "Streamly.Examples.AcidRainGame". It has been adapted from Gabriel's
1147-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1149-- This is much simpler compared to the pipes version because of the builtin
1150-- concurrency in streamly. You can also find a SDL based reactive programming
1151-- example adapted from Yampa in "Streamly.Examples.CirclingSquare".
1155-- Streamly is highly optimized for performance, it is designed for serious
1156-- high performing, concurrent and scalable applications. We have created the
1157-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1158-- package which is specifically and carefully designed to measure the
1159-- performance of Haskell streaming libraries fairly and squarely in the right
1160-- way. Streamly performs at par or even better than most streaming libraries
1161-- for serial operations even though it needs to deal with the concurrency
1166-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1168-- Interop with @vector@:
1172-- import qualified Streamly.Prelude as S
1173-- import qualified Data.Vector.Fusion.Stream.Monadic as V
1176-- -- streamly to vector
1177-- V.toList (V.unfoldrM S.uncons (S.fromFoldable [1..3])) >>= print
1179-- -- vector to streamly
1180-- S.toList (S.unfoldrM unconsV (V.fromList [1..3])) >>= print
1186-- then return Nothing
1189-- return $ Just (h, V.tail v)
1192-- Interop with @pipes@:
1196-- import qualified "Streamly.Prelude" as S
1197-- import qualified Pipes as P
1198-- import qualified Pipes.Prelude as P
1201-- -- streamly to pipe
1202-- P.toListM (P.unfoldr unconsS (S.'fromFoldable' [1..3])) >>= print
1204-- -- pipe to streamly
1205-- S.'toList' (S.'unfoldrM' unconsP (P.each [1..3])) >>= print
1208-- -- Adapt P.next to return a Maybe instead of Either
1209-- unconsP p = P.next p >>= either (\\_ -> return Nothing) (return . Just)
1211-- -- Adapt S.uncons to return an Either instead of Maybe
1212-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1215-- Interop with @streaming@:
1219-- import qualified "Streamly.Prelude" as S
1220-- import qualified Streaming as SG
1221-- import qualified Streaming.Prelude as SG
1224-- -- streamly to streaming
1225-- SG.toList (SG.unfoldr unconsS (S.'fromFoldable' [1..3])) >>= print
1227-- -- streaming to streamly
1228-- S.'toList' (S.unfoldrM SG.uncons (SG.each [1..3])) >>= print
1232-- -- Adapt S.uncons to return an Either instead of Maybe
1233-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1236-- Interop with @conduit@:
1240-- import qualified "Streamly.Prelude" as S
1241-- import qualified Data.Conduit as C
1242-- import qualified Data.Conduit.List as C
1243-- import qualified Data.Conduit.Combinators as C
1246-- -- streamly to conduit
1247-- C.runConduit (C.unfoldM S.'uncons' (S.'fromFoldable' [1..3]) C..| C.sinkList) >>= print
1249-- -- It seems there is no way out of a conduit as it does not provide an
1250-- -- uncons or a tail function.
1255-- Streamly unifies non-determinism, streaming, concurrency and FRP
1256-- functionality that is otherwise covered by several disparate packages, and
1257-- it does that with a surprisingly concise API. Here is a list of popular and
1258-- well-known packages in all these areas:
1261-- +-----------------+----------------+
1262-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1263-- | +----------------+
1264-- | | <https://hackage.haskell.org/package/list-t list-t> |
1265-- | +----------------+
1266-- | | <https://hackage.haskell.org/package/logict logict> |
1267-- +-----------------+----------------+
1268-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1269-- | +----------------+
1270-- | | <https://hackage.haskell.org/package/streaming streaming> |
1271-- | +----------------+
1272-- | | <https://hackage.haskell.org/package/pipes pipes> |
1273-- | +----------------+
1274-- | | <https://hackage.haskell.org/package/conduit conduit> |
1275-- +-----------------+----------------+
1276-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1277-- | +----------------+
1278-- | | <https://hackage.haskell.org/package/transient transient> |
1279-- +-----------------+----------------+
1280-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1281-- | +----------------+
1282-- | | <https://hackage.haskell.org/package/dunai dunai> |
1283-- | +----------------+
1284-- | | <https://hackage.haskell.org/package/reflex reflex> |
1285-- +-----------------+----------------+
1288-- Streamly is a list-transformer. It provides all the functionality provided
1289-- by any of the list transformer and logic programming packages listed above.
1290-- In addition, Streamly naturally integrates the concurrency dimension to the
1291-- basic list transformer functionality.
1293-- When it comes to streaming, in terms of the streaming API streamly is almost
1294-- identical to the vector package. Streamly, vector and streaming packages all
1295-- represent a stream as data and are therefore similar in the fundamental
1296-- approach to streaming. The fundamental difference is that streamly adds
1297-- concurrency support and the monad instance provides concurrent looping.
1298-- Other streaming libraries like pipes, conduit and machines represent and
1299-- compose stream processors rather than the stream data and therefore fall in
1300-- another class of streaming libraries and have comparatively more complicated
1303-- When it comes to concurrency, streamly can do everything that the @async@
1304-- package can do and more. async provides applicative concurrency whereas
1305-- streamly provides both applicative and monadic concurrency. The
1306-- 'ZipParallel' type behaves like the applicative instance of async. In
1307-- comparison to transient streamly has a first class streaming interface and
1308-- is a monad transformer that can be used universally in any Haskell monad
1309-- transformer stack. Streamly was in fact originally inspired by the
1310-- concurrency implementation in @transient@ though it has no resemblence with
1311-- that and takes a lazy pull approach versus transient's strict push approach.
1313-- The non-determinism, concurrency and streaming combination make streamly a
1314-- strong FRP capable library as well. FRP is fundamentally stream of events
1315-- that can be processed concurrently. The example in this tutorial as well as
1316-- the "Streamly.Examples.CirclingSquare" example from Yampa demonstrate the
1317-- basic FRP capability of streamly. In core concepts streamly is strikingly
1318-- similar to @dunai@. dunai was designed from a FRP perspective and streamly
1319-- was originally designed from a concurrency perspective. However, both have
1320-- similarity at the core.
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