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 Composition ('StreamT')
81 -- *** Async Composition ('AsyncT')
84 -- *** Interleaved Composition ('InterleavedT')
85 -- $interleavedNesting
87 -- *** Fair Concurrent Composition ('ParallelT')
88 -- $fairlyConcurrentNesting
105 -- ** Parallel Zipping
108 -- * Summary of Compositions
109 -- $compositionSummary
111 -- * Concurrent Programming
114 -- * Reactive Programming
120 -- * Interoperation with Streaming Libraries
123 -- * Comparison with Existing Packages
129import Streamly.Prelude
131import Control.Applicative
133import Control.Monad.IO.Class (MonadIO(..))
134import Control.Monad.Trans.Class (MonadTrans (lift))
138-- Streamly provides many different stream types depending on the desired
139-- composition style. The simplest type is 'Stream', @Stream a@ represents a
140-- stream of elements of type @a@ in IO monad. To represent streams in an
141-- arbitrary monad use the more general monad transformer type 'StreamT'.
142-- @StreamT m a@ represents a stream of values of type 'a' in some underlying
143-- monad 'm'. For example, @StreamT IO Int@ is a stream of 'Int' in 'IO' monad.
144-- In fact, the type 'Stream' is a synonym for @StreamT IO@.
148-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
149-- a monadic action at the head of the stream.
154-- > toList $ getLine |: getLine |: nil
160-- To create a singleton stream from a pure value use 'pure' and to create a
161-- singleton stream from a monadic action use 'once'.
166-- > toList $ once getLine
171-- To lift a value from an underlying monad in a monad transformer stack into a
172-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
175-- > runStream $ liftIO $ putStrLn "Hello world!"
177-- > runStream $ lift $ putStrLn "Hello world!"
181-- To create a stream of pure values from a 'Foldable' container use
185-- > toList $ fromFoldable [1..3]
189-- To create a stream from monadic actions in a 'Foldable' container just use a
193-- > runStream $ Prelude.foldr (|:) nil [putStr "Hello ", putStrLn "world!"]
197-- For more ways to construct a stream see the module "Streamly.Prelude".
201-- We have already seen 'runStream' and 'toList' to eliminate a stream in the
202-- examples above. 'runStream' runs a stream discarding the results i.e. only
203-- for effects. 'toList' runs the stream and collects the results in a list.
205-- For other ways to eliminate a stream see the @Folding@ section in
206-- "Streamly.Prelude" module.
210-- Transformation over a stream is the equivalent of a @for@ loop construct in
211-- imperative paradigm. We iterate over every element in the stream and perform
212-- certain transformations for each element. Transformations may involve
213-- mapping functions over the elements, filtering elements from the stream or
214-- folding all the elements in the stream into a single value. Streamly streams
215-- are exactly like lists and you can perform all the transformations in the
216-- same way as you would on lists. For example the following snippet reads
217-- lines from standard input, filters blank lines, drops the first non-blank
218-- line, takes the next two, up cases them, numbers them and prints them:
222-- import qualified Streamly.Prelude as S
223-- import Data.Char (toUpper)
224-- import Data.Function ((&))
225-- import Data.Semigroup (cycle1)
227-- main = runStream $ cycle1 (S.once getLine)
228-- & S.filter (not . null)
231-- & fmap (map toUpper)
232-- & S.zipWith (\\n s -> show n ++ " " ++ s) (S.fromFoldable [1..])
238-- We can combine two streams into a single stream using semigroup composition
239-- operation '<>'. Streams can be combined in many different ways as described
240-- in the following sections, the '<>' operation behaves differently depending
241-- on the stream type in effect. Streamly streams can be of many
242-- different types i.e. 'Stream', 'Costream', 'Coparallel', 'Parallel',
243-- 'ZipStream', and 'ZipParallel'. The first four are monads whereas the last
244-- two are applicatives. These types can be freely converted from one to
245-- another to acheive the desired composition style. To force a particular
246-- type of composition we coerce the stream type using the corresponding type
247-- adapting combinator from 'streamly', 'costreamly', 'coparallely',
248-- 'parallely', 'zipStreamly' or 'zipParallely'. The default stream type is
249-- 'Stream' unless you change it by using one of the combinators.
251-- To illustrate the concurrent vs serial composition aspects, we will use the
252-- following @delay@ function to introduce a delay specified in seconds.
256-- import Control.Concurrent
258-- delay n = liftIO $ do
259-- threadDelay (n * 1000000)
261-- putStrLn (show tid ++ ": Delay " ++ show n)
266-- The 'Semigroup' operation '<>' of the 'Stream' type combines the two streams
267-- in a serial manner. We use the 'streamly' type combinator to effect 'Stream'
268-- style of composition. We can also use the 'Stream' type annotation for the
269-- stream type to acheive the same effect. However, since 'Stream' is the
270-- default type unless explicitly specified by using a combinator, we can
271-- omit using an explicit combinator or type annotation for this style of
274-- When two streams with multiple elements are combined in this manner, the
275-- monadic actions in the two streams are performed sequentially i.e. first all
276-- actions in the first stream are performed sequentially and then all actions
277-- in the second stream are performed sequentially. The following example
278-- prints the sequence 1, 2, 3, 4:
281-- main = 'runStream' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
290-- All actions in both the streams are performed serially in the same thread.
291-- In the following example we can see that all actions are performed in the
292-- same thread and take a combined total of 6 seconds:
295-- main = 'runStream' $ delay 3 <> delay 2 <> delay 1
298-- ThreadId 36: Delay 3
299-- ThreadId 36: Delay 2
300-- ThreadId 36: Delay 1
303-- The polymorphic version of the 'Stream' binary operation '<>' is called
304-- 'splice'. We can use 'splice' to join streams in a sequential manner
305-- irrespective of the type of stream:
308-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`splice` (print 3 |: print 4 |: nil)
313-- The 'Semigroup' operation '<>' of the 'Costream' type combines the two
314-- streams in a serial interleaved manner. We use the 'costreamly' type
315-- combinator to effect 'Costream' style of composition. We can also use the
316-- 'Costream' type annotation for the stream type to acheive the same effect.
318-- When two streams with multiple elements are combined in this manner, the
319-- monadic actions in the two streams are performed in an interleaved manner
320-- i.e. one action from the first stream is performed and its output yielded
321-- and then one action from the second stream is performed and its output
322-- yielded, and so on. The following example prints the sequence 1, 3, 2, 4
325-- main = 'runStream' . 'costreamly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
334-- Even though the monadic actions of the two streams are performed in an
335-- interleaved manner they are all performed serially in the same thread. In
336-- the following example we can see that all actions are performed in the same
337-- thread and take a combined total of 6 seconds:
340-- main = 'runStream' . 'costreamly' $ delay 3 <> delay 2 <> delay 1
343-- ThreadId 36: Delay 3
344-- ThreadId 36: Delay 2
345-- ThreadId 36: Delay 1
348-- The polymorphic version of the 'Costream' binary operation '<>' is called
349-- 'cosplice'. We can use 'cosplice' to join streams in an interleaved manner
350-- irrespective of the type, notice that we have not used the 'costreamly'
351-- combinator in the following example:
354-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`cosplice` (print 3 |: print 4 |: nil)
363-- Note that this composition cannot be used to fold infinite number of streams
364-- since it requires preserving the state until a stream is finished.
368-- The 'Semigroup' operation '<>' of the 'Coparallel' type combines the two
369-- streams in a left biased concurrent manner. We use the 'coparallel' type
370-- combinator to effect 'Coparallel' style of composition. We can also use the
371-- 'Coparallel' type annotation for the stream type to acheive the same effect.
373-- When two streams with multiple elements are combined in this manner, the
374-- monadic actions in the two streams can be performed concurrently /when
375-- needed/ with a bias towards the left stream i.e. actions from the left
376-- stream are performed sequentially, if the left stream blocks or cannot
377-- generate enough output to keep the consumer of the stream busy then it can
378-- execute actions from the right stream concurrently. The outputs of the
379-- concurrent actions are yielded by the composite stream in the order in which
380-- they complete. This can be thought of as the concurrent analogue of the
381-- 'Stream' style serial composition. If concurrent execution is not trigerred
382-- by consumer demand then it would execute actions exactly in the same order
383-- as 'Stream'. In the following example the first stream does not block,
384-- therefore the first stream is completely exhausted before the second.
387-- main = 'runStream' . 'coparallely' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
396-- If the first stream blocks, we can yield from the second. In the example
397-- below each yield in the stream has a constant delay of 1 second therefore 1
398-- and 3 would be yielded first and then 2 and 4 would be yielded.
401-- main = 'runStream' . 'coparallely' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
402-- where p n = threadDelay 1000000 >> print n
411-- In the following example we can see that new threads are started when a
412-- computation blocks. Notice that the output from the stream with the
413-- shortest delay is printed first:
416-- main = 'runStream' . 'coparallely' $ delay 3 '<>' delay 2 '<>' delay 1
419-- ThreadId 42: Delay 1
420-- ThreadId 41: Delay 2
421-- ThreadId 40: Delay 3
424-- When we have a tree of computations composed using this style, the tree is
425-- traversed in DFS style just like the serial style, the only difference is
426-- that here we can move on to executing the next stream if a stream blocks.
427-- However, we will not start new threads if we have sufficient output to
428-- saturate the consumer. This is why we call it left-biased demand driven or
429-- adaptive concurrency style, the concurrency tends to stay on the left side
430-- of the composition as long as possible. More threads are started based on
431-- the pull rate of the consumer. The following example prints an output every
432-- second as all of the actions are concurrent.
435-- main = 'runStream' . 'coparallely' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
444-- All the computations may even run in a single thread when more threads are
445-- not needed. As you can see, in the following example the computations are
446-- run in a single thread one after another, because none of them blocks.
447-- However, if the thread consuming the stream were faster than the producer
448-- then it would have started parallel threads for each computation to keep up
449-- even if none of them blocks:
452-- main = 'runStream' . 'coparallely' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
453-- where traced m = once (myThreadId >>= print) >> return m
461-- Note that the order of printing in the above examples may change due to
462-- variations in scheduling latencies for concurrent threads.
464-- The polymorphic version of the 'Coparallel' binary operation '<>' is called
465-- 'coparallel'. We can use 'coparallel' to join streams in a left biased
466-- adaptively concurrent manner irrespective of the type, notice that we have
467-- not used the 'coparallely' combinator in the following example:
470-- main = 'runStream' $ delay 3 \`coparallel` delay 2 \`coparallel` delay 1
473-- ThreadId 42: Delay 1
474-- ThreadId 41: Delay 2
475-- ThreadId 40: Delay 3
478-- Since the concurrency provided by this operator is demand driven it cannot
479-- be used when the composed computations start timers that are relative to
480-- each other because all computations may not be started at the same time and
481-- therefore timers in all of them may not start at the same time. When
482-- relative timing among all computations is important or when we need to start
483-- all computations at once for any reason 'Parallel' style must be used
486-- 'Coparallel' style should be preferred over 'Parallel' unless you really
487-- need 'Parallel'. It utilizes the resources optimally. It should be used when
488-- we know that the computations can run in parallel but we do not care if they
489-- actually run in parallel or not, that decision can be left to the scheduler
490-- based on demand. Also, note that this operator can be used to fold infinite
491-- number of streams in contrast to the 'Parallel' style, because it does not
492-- require us to run all of them at the same time in a fair manner.
496-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
497-- streams in a fairly concurrent manner with round robin scheduling. We use
498-- the 'parallel' type combinator to effect 'Parallel' style of composition. We
499-- can also use the 'Parallel' type annotation for the stream type to acheive
502-- When two streams with multiple elements are combined in this manner, the
503-- monadic actions in both the streams are performed concurrently with a fair
504-- round robin scheduling. The outputs are yielded in the order in which the
505-- actions complete. This can be thought of as the concurrent analogue of the
506-- 'Costream' style serial composition.
508-- The following example sends a query to all the three search engines in
509-- parallel and prints the name of the search engines in the order in which the
514-- import Streamly.Prelude (once)
515-- import Network.HTTP.Simple
517-- main = 'runStream' . 'parallely' $ google \<> bing \<> duckduckgo
519-- google = get "https://www.google.com/search?q=haskell"
520-- bing = get "https://www.bing.com/search?q=haskell"
521-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
522-- get s = once (httpNoBody (parseRequest_ s) >> putStrLn (show s))
525-- In the following example we can see that the fairly parallel composition
526-- yields outputs in the same order as the 'Costream' composition, though this
527-- is not always guaranteed because of thread scheduling variances.
530-- main = 'runStream' . 'parallely' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
539-- The polymorphic version of the 'Parallel' binary operation '<>' is called
540-- 'parallel'. We can use 'parallel' to join streams in a fairly
541-- concurrent manner irrespective of the type, notice that we have
542-- not used the 'parallely' combinator in the following example:
545-- main = 'runStream' $ delay 3 \`parallel` delay 2 \`parallel` delay 1
548-- ThreadId 42: Delay 1
549-- ThreadId 41: Delay 2
550-- ThreadId 40: Delay 3
553-- Note that this style of composition cannot be used to combine infinite
554-- number of streams, as it will lead to an infinite sized scheduling queue.
559-- The 'async' API can be used to create references to asynchronously running
560-- stream computations. We can then use 'uncons' to explore the streams
561-- arbitrarily and then recompose individual elements to create a new stream.
562-- This way we can dynamically decide which stream to explore at any given
563-- time. Take an example of a merge sort of two sorted streams. We need to
564-- keep consuming items from the stream which has the lowest item in the sort
565-- order. This can be achieved using async references to streams. See
566-- "Streamly.Examples.MergeSortedStreams".
570-- We can use 'Monoid' instances to fold a container of streams in the desired
571-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
572-- to fold streams using the polymorphic combine operations:
574-- * 'foldWith' is like 'fold', it folds a 'Foldable' container of streams
575-- using the given composition operator.
576-- * 'foldMapWith' is like 'foldMap', it folds like @foldWith@ but also maps a
577-- function before folding.
578-- * 'forEachWith' is like @foldMapwith@ but the container argument comes before
579-- the function argument.
581-- All of the following are equivalent and start ten concurrent tasks each with
582-- a delay from 1 to 10 seconds, resulting in the printing of each number every
587-- import "Streamly.Prelude"
588-- import Control.Concurrent
591-- runStream $ parallely $ foldMap delay [1..10]
592-- runStream $ 'foldWith' parallel (map delay [1..10])
593-- runStream $ 'foldMapWith' parallel delay [1..10]
594-- runStream $ 'forEachWith' parallel [1..10] delay
595-- where delay n = once $ threadDelay (n * 1000000) >> print n
600-- The previous section discussed ways to merge the elements of two streams
601-- without doing any transformation on them. In this section we will explore
602-- how to transform streams using 'Functor', 'Applicative' or 'Monad' style
603-- compositions. The applicative and monad compositions of all 'IsStream' types
604-- behave exactly the same way as a list transformer. For simplicity of
605-- illustration we are using streams of pure values in the following examples.
606-- However, the real application of streams arises when these streams are
607-- generated using monadic actions.
611-- In functional programmer's parlance the 'Monad' instance of 'IsStream'
612-- types implement non-determinism, exploring all possible combination of
613-- choices from both the streams. From an imperative programmer's point of view
614-- it behaves like nested loops i.e. for each element in the first stream and
615-- for each element in the second stream apply the body of the loop. If you are
616-- familiar with list transformer this behavior is exactly the same as that of
617-- a list transformer.
619-- Just like we saw in sum style compositions earlier, monadic composition also
620-- has multiple variants each of which exactly corresponds to one of the sum
621-- style composition variant.
625-- When we interpret the monadic composition as 'StreamT' we get a standard
626-- list transformer like serial composition.
630-- import "Streamly.Prelude"
632-- main = 'runStreamT' $ do
633-- x <- 'fromFoldable' [3,2,1]
637-- ThreadId 30: Delay 3
638-- ThreadId 30: Delay 2
639-- ThreadId 30: Delay 1
642-- As you can see the code after the @fromFoldable@ statement is run three
643-- times, once for each value of @x@. All the three iterations are serial and
644-- run in the same thread one after another. When compared to imperative
645-- programming, this can also be viewed as a @for@ loop with three iterations.
647-- A console echo loop copying standard input to standard output can simply be
652-- import Data.Semigroup (cycle1)
654-- main = 'runStreamT' $ cycle1 (liftIO getLine) >>= liftIO . putStrLn
657-- When multiple streams are composed using this style they nest in a DFS
658-- manner i.e. nested iterations of an iteration are executed before we proceed
659-- to the next iteration at higher level. This behaves just like nested @for@
660-- loops in imperative programming.
664-- import "Streamly.Prelude"
666-- main = 'runStreamT' $ do
667-- x <- 'fromFoldable' [1,2]
668-- y <- 'fromFoldable' [3,4]
669-- liftIO $ putStrLn $ show (x, y)
678-- You will also notice that this is the monadic equivalent of the sum style
679-- composition using '<>'.
683-- When we interpret the monadic composition as 'AsyncT' we get a /concurrent/
684-- list-transformer like composition. Multiple monadic continuations (or loop
685-- iterations) may be started concurrently. Concurrency is demand driven
686-- i.e. more concurrent iterations are started only if the previous iterations
687-- are not able to produce enough output for the consumer of the output stream.
688-- This is the concurrent version of 'StreamT'.
692-- import "Streamly.Prelude"
694-- main = 'runAsyncT' $ do
695-- x <- 'fromFoldable' [3,2,1]
699-- ThreadId 40: Delay 1
700-- ThreadId 39: Delay 2
701-- ThreadId 38: Delay 3
704-- As you can see the code after the @fromFoldable'@ statement is run three
705-- times, once for each value of @x@. All the three iterations are concurrent
706-- and run in different threads. The iteration with least delay finishes first.
707-- When compared to imperative programming, this can be viewed as a @for@ loop
708-- with three concurrent iterations.
710-- Concurrency is demand driven just as in the case of '<|'. When multiple
711-- streams are composed using this style the iterations are triggered in a DFS
712-- manner just like 'StreamT' i.e. nested iterations are executed before we
713-- proceed to the next iteration at higher level. However, unlike 'StreamT'
714-- more than one iterations may be started concurrently, and based on the
715-- demand from the consumer.
719-- import "Streamly.Prelude"
721-- main = 'runAsyncT' $ do
722-- x <- 'fromFoldable' [1,2]
723-- y <- 'fromFoldable' [3,4]
724-- liftIO $ putStrLn $ show (x, y)
733-- You will notice that this is the monadic equivalent of the '<|' style
734-- sum composition. The same caveats apply to this as the '<|' operation.
736-- $interleavedNesting
738-- When we interpret the monadic composition as 'InterleavedT' we get a serial
739-- but fairly interleaved list-transformer like composition. The monadic
740-- continuations or iterations of the outer loop are fairly interleaved with
741-- the continuations or iterations of the inner loop.
745-- import "Streamly.Prelude"
747-- main = 'runInterleavedT' $ do
748-- x <- 'fromFoldable' [1,2]
749-- y <- 'fromFoldable' [3,4]
750-- liftIO $ putStrLn $ show (x, y)
759-- You will notice that this is the monadic equivalent of the '<=>' style
760-- sum composition. The same caveats apply to this as the '<=>' operation.
762-- $fairlyConcurrentNesting
764-- When we interpret the monadic composition as 'ParallelT' we get a
765-- /concurrent/ list-transformer like composition just like 'AsyncT'. The
766-- difference is that this is fully parallel with all iterations starting
767-- concurrently instead of the demand driven concurrency of 'AsyncT'.
771-- import "Streamly.Prelude"
773-- main = 'runParallelT' $ do
774-- x <- 'fromFoldable' [3,2,1]
778-- ThreadId 40: Delay 1
779-- ThreadId 39: Delay 2
780-- ThreadId 38: Delay 3
783-- You will notice that this is the monadic equivalent of the '<|>' style
784-- sum composition. The same caveats apply to this as the '<|>' operation.
788-- The streamly code is usually written in a way that is agnostic of the
789-- specific monadic composition type. We use a polymorphic type with a
790-- 'IsStream' type class constraint. When running the stream we can choose the
791-- specific mode of composition. For example look at the following code.
795-- import "Streamly.Prelude"
798-- composed :: 'IsStream' t => t m a
803-- liftIO $ putStrLn $ show (sz, cl, sh)
807-- sizes = 'fromFoldable' [1, 2, 3]
808-- colors = 'fromFoldable' ["red", "green", "blue"]
809-- shapes = 'fromFoldable' ["triangle", "square", "circle"]
812-- Now we can interpret this in whatever way we want:
815-- main = 'runStreamT' composed
816-- main = 'runAsyncT' composed
817-- main = 'runInterleavedT' composed
818-- main = 'runParallelT' composed
821-- Equivalently, we can also write it using the type adapter combinators, like
825-- main = 'runStream' $ 'serially' $ composed
826-- main = 'runStream' $ 'asyncly' $ composed
827-- main = 'runStream' $ 'interleaving' $ composed
828-- main = 'runStream' $ 'parallely' $ composed
831-- As an exercise try to figure out the output of this code for each mode of
836-- 'fmap' transforms a stream by mapping a function on all elements of the
837-- stream. The functor instance of each stream type defines 'fmap' to be
838-- precisely the same as 'liftM', and therefore 'fmap' is always serial
839-- irrespective of the type. For concurrent mapping, alternative versions of
840-- 'fmap', namely, 'asyncMap' and 'parMap' are provided.
845-- main = ('toList' $ 'serially' $ fmap show $ 'fromFoldable' [1..10]) >>= print
848-- Also see the 'mapM' and 'sequence' functions for mapping actions, in the
849-- "Streamly.Prelude" module.
853-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
854-- zipping and parallel applicatives separate types 'ZipSerial' and 'ZipAsync'
857-- The following example runs all iterations serially and takes a total 17
858-- seconds (1 + 3 + 4 + 2 + 3 + 4):
862-- import "Streamly.Prelude"
863-- import Control.Concurrent
867-- d n = delay n >> return n
869-- main = ('toList' . 'serially' $ (,) \<$> s1 \<*> s2) >>= print
872-- ThreadId 36: Delay 1
873-- ThreadId 36: Delay 3
874-- ThreadId 36: Delay 4
875-- ThreadId 36: Delay 2
876-- ThreadId 36: Delay 3
877-- ThreadId 36: Delay 4
878-- [(1,3),(1,4),(2,3),(2,4)]
881-- Similalrly interleaving runs the iterations in an interleaved order but
882-- since it is serial it takes a total of 17 seconds:
885-- main = ('toList' . 'interleaving' $ (,) \<$> s1 \<*> s2) >>= print
888-- ThreadId 36: Delay 1
889-- ThreadId 36: Delay 3
890-- ThreadId 36: Delay 2
891-- ThreadId 36: Delay 3
892-- ThreadId 36: Delay 4
893-- ThreadId 36: Delay 4
894-- [(1,3),(2,3),(1,4),(2,4)]
897-- 'AsyncT' can run the iterations concurrently and therefore takes a total
898-- of 10 seconds (1 + 2 + 3 + 4):
901-- main = ('toList' . 'asyncly' $ (,) \<$> s1 \<*> s2) >>= print
904-- ThreadId 34: Delay 1
905-- ThreadId 36: Delay 2
906-- ThreadId 35: Delay 3
907-- ThreadId 36: Delay 3
908-- ThreadId 35: Delay 4
909-- ThreadId 36: Delay 4
910-- [(1,3),(2,3),(1,4),(2,4)]
913-- Similalrly 'ParallelT' as well can run the iterations concurrently and
914-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
917-- main = ('toList' . 'parallely' $ (,) \<$> s1 \<*> s2) >>= print
920-- ThreadId 34: Delay 1
921-- ThreadId 36: Delay 2
922-- ThreadId 35: Delay 3
923-- ThreadId 36: Delay 3
924-- ThreadId 35: Delay 4
925-- ThreadId 36: Delay 4
926-- [(1,3),(2,3),(1,4),(2,4)]
929-- $compositionSummary
931-- The following table summarizes the types for monadic compositions and the
932-- operators for sum style compositions. This table captures the essence of
936-- +-----+--------------+------------+
937-- | | Serial | Concurrent |
938-- +=====+==============+============+
939-- | DFS | 'StreamT' | 'AsyncT' |
940-- | +--------------+------------+
942-- +-----+--------------+------------+
943-- | BFS | 'InterleavedT' | 'ParallelT' |
944-- | +--------------+------------+
945-- | | '<=>' | '<|>' |
946-- +-----+--------------+------------+
951-- Zipping is a special transformation where the corresponding elements of two
952-- streams are combined together using a zip function producing a new stream of
953-- outputs. Two different types are provided for serial and concurrent zipping.
954-- These types provide an applicative instance that zips the argument streams.
955-- Also see the zipping function in the "Streamly.Prelude" module.
959-- 'ZipSerial' zips streams serially:
963-- import "Streamly.Prelude"
964-- import Control.Concurrent
966-- d n = delay n >> return n
967-- s1 = 'adapt' . 'serially' $ d 1 <> d 2
968-- s2 = 'adapt' . 'serially' $ d 3 <> d 4
970-- main = ('toList' . 'zipping' $ (,) \<$> s1 \<*> s2) >>= print
973-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
974-- everything runs serially:
977-- ThreadId 29: Delay 1
978-- ThreadId 29: Delay 3
979-- ThreadId 29: Delay 2
980-- ThreadId 29: Delay 4
986-- 'ZipAsync' zips streams concurrently:
990-- import "Streamly.Prelude"
991-- import Control.Concurrent
992-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
994-- d n = delay n >> return n
995-- s1 = 'adapt' . 'serially' $ d 1 <> d 2
996-- s2 = 'adapt' . 'serially' $ d 3 <> d 4
999-- liftIO $ hSetBuffering stdout LineBuffering
1000-- ('toList' . 'zippingAsync' $ (,) \<$> s1 \<*> s2) >>= print
1003-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1004-- are produced concurrently, and 2 and 4 are produced concurrently:
1007-- ThreadId 32: Delay 1
1008-- ThreadId 32: Delay 2
1009-- ThreadId 33: Delay 3
1010-- ThreadId 33: Delay 4
1016-- When writing concurrent programs there are two distinct places where the
1017-- programmer chooses the type of concurrency. First, when /generating/ a
1018-- stream by combining other streams we can use one of the sum style operators
1019-- to combine them concurrently or serially. Second, when /processing/ a stream
1020-- in a monadic composition we can choose one of the monad composition types to
1021-- choose the desired type of concurrency.
1023-- In the following example the squares of @x@ and @y@ are computed
1024-- concurrently using the '<|' operator and the square roots of their sum are
1025-- also computed concurrently by using the 'asyncly' combinator. We can choose
1026-- different combinators e.g. '<>' and 'serially', to control the concurrency.
1030-- import "Streamly.Prelude" (toList)
1031-- import Data.List (sum)
1035-- $ 'asyncly' -- Concurrent monadic processing (sqrt below)
1037-- x2 \<- 'forEachWith' ('<|') [1..100] $ -- Concurrent @"for"@ loop
1038-- \\x -> return $ x * x -- body of the loop
1039-- y2 \<- 'forEachWith' ('<|') [1..100] $
1040-- \\y -> return $ y * y
1041-- return $ sqrt (x2 + y2)
1045-- You can see how this directly maps to the imperative style
1046-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1047-- and operators instead of the ugly pragmas.
1049-- For more concurrent programming examples see,
1050-- "Streamly.Examples.ListDirRecursive", "Streamly.Examples.MergeSortedStreams"
1051-- and "Streamly.Examples.SearchEngineQuery".
1055-- Reactive programming is nothing but concurrent streaming which is what
1056-- streamly is all about. With streamly we can generate streams of events,
1057-- merge streams that are generated concurrently and process events
1058-- concurrently. We can do all this without any knowledge about the specifics
1059-- of the implementation of concurrency. In the following example you will see
1060-- that the code is just regular Haskell code without much streamly APIs used
1061-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1065-- This application has two independent and concurrent sources of event
1066-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1067-- events that deteriorate the health of the game character. @userAction@ can
1068-- be "potion" or "quit". When the user types "potion" the health improves and
1069-- the game continues.
1072-- {-\# LANGUAGE FlexibleContexts #-}
1075-- import Control.Concurrent (threadDelay)
1076-- import Control.Monad (when)
1077-- import Control.Monad.State
1078-- import Data.Semigroup (cycle1)
1080-- data Event = Harm Int | Heal Int | Quit deriving (Show)
1082-- userAction :: MonadIO m => 'StreamT' m Event
1083-- userAction = cycle1 $ liftIO askUser
1086-- command <- getLine
1088-- "potion" -> return (Heal 10)
1089-- "quit" -> return Quit
1090-- _ -> putStrLn "What?" >> askUser
1092-- acidRain :: MonadIO m => 'StreamT' m Event
1093-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
1095-- game :: ('MonadAsync' m, MonadState Int m) => 'StreamT' m ()
1097-- event \<- userAction \<|> acidRain
1099-- Harm n -> modify $ \\h -> h - n
1100-- Heal n -> modify $ \\h -> h + n
1101-- Quit -> fail "quit"
1104-- when (h <= 0) $ fail "You die!"
1105-- liftIO $ putStrLn $ "Health = " ++ show h
1108-- putStrLn "Your health is deteriorating due to acid rain,\\
1109-- \\ type \\"potion\\" or \\"quit\\""
1110-- _ <- runStateT ('runStreamT' game) 60
1114-- You can also find the source of this example in
1115-- "Streamly.Examples.AcidRainGame". It has been adapted from Gabriel's
1116-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1118-- This is much simpler compared to the pipes version because of the builtin
1119-- concurrency in streamly. You can also find a SDL based reactive programming
1120-- example adapted from Yampa in "Streamly.Examples.CirclingSquare".
1124-- Streamly is highly optimized for performance, it is designed for serious
1125-- high performing, concurrent and scalable applications. We have created the
1126-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1127-- package which is specifically and carefully designed to measure the
1128-- performance of Haskell streaming libraries fairly and squarely in the right
1129-- way. Streamly performs at par or even better than most streaming libraries
1130-- for common operations even though it needs to deal with the concurrency
1135-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1136-- We will assume the following common code to be available in the examples
1137-- demonstrated below.
1141-- import "Streamly.Prelude"
1142-- import System.IO (stdin)
1144-- -- Adapt uncons to return an Either instead of Maybe
1145-- unconsE s = 'uncons' s >>= maybe (return $ Left ()) (return . Right)
1146-- stdinLn = 'serially' $ 'fromHandle' stdin
1149-- Interop with @pipes@:
1152-- import qualified Pipes as P
1153-- import qualified Pipes.Prelude as P
1156-- -- streamly to pipe
1157-- P.runEffect $ P.for (P.unfoldr unconsE stdinLn) (lift . putStrLn)
1159-- -- pipe to streamly
1160-- -- Adapt P.next to return a Maybe instead of Either
1161-- let nextM p = P.next p >>= either (\\_ -> return Nothing) (return . Just)
1162-- 'runStreamT' $ 'unfoldrM' nextM P.stdinLn >>= lift . putStrLn
1165-- Interop with @streaming@:
1168-- import qualified Streaming as S
1169-- import qualified Streaming.Prelude as S
1172-- -- streamly to streaming
1173-- S.stdoutLn $ S.unfoldr unconsE stdinLn
1175-- -- streaming to streamly
1176-- 'runStreamT' $ unfoldrM S.uncons S.stdinLn >>= lift . putStrLn
1180-- Interop with @conduit@:
1183-- import qualified Data.Conduit as C
1184-- import qualified Data.Conduit.List as C
1185-- import qualified Data.Conduit.Combinators as C
1187-- -- streamly to conduit
1188-- main = (C.unfoldM 'uncons' stdinLn) C.$$ C.print
1193-- Streamly unifies non-determinism, streaming, concurrency and FRP
1194-- functionality that is otherwise covered by several disparate packages, and
1195-- it does that with a surprisingly concise API. Here is a list of popular and
1196-- well-known packages in all these areas:
1199-- +-----------------+----------------+
1200-- | Non-determinism | <https://hackage.haskell.org/package/list-t list-t> |
1201-- | +----------------+
1202-- | | <https://hackage.haskell.org/package/logict logict> |
1203-- +-----------------+----------------+
1204-- | Streaming | <https://hackage.haskell.org/package/streaming streaming> |
1205-- | +----------------+
1206-- | | <https://hackage.haskell.org/package/conduit conduit> |
1207-- | +----------------+
1208-- | | <https://hackage.haskell.org/package/pipes pipes> |
1209-- | +----------------+
1210-- | | <https://hackage.haskell.org/package/simple-conduit simple-conduit> |
1211-- +-----------------+----------------+
1212-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1213-- | +----------------+
1214-- | | <https://hackage.haskell.org/package/transient transient> |
1215-- +-----------------+----------------+
1216-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1217-- | +----------------+
1218-- | | <https://hackage.haskell.org/package/dunai dunai> |
1219-- | +----------------+
1220-- | | <https://hackage.haskell.org/package/reflex reflex> |
1221-- +-----------------+----------------+
1224-- Streamly covers all the functionality provided by both the non-determinism
1225-- packages listed above and provides better performance in comparison to
1226-- those. In fact, at the core streamly is a list transformer but it naturally
1227-- integrates the concurrency dimension to the basic list transformer
1230-- When it comes to streaming, in terms of core concepts, @simple-conduit@ is
1231-- the package that is closest to streamly if we set aside the concurrency
1232-- dimension, both are streaming packages with list transformer like monad
1233-- composition. However, in terms of API @streamly@ is more like the @streaming@
1234-- package. Streamly can be used to achieve more or less the functionality
1235-- provided by any of the streaming packages listed above. The types and API of
1236-- streamly are much simpler in comparison to conduit and pipes. It is more or
1237-- less like the standard Haskell list APIs.
1239-- When it comes to concurrency, streamly can do everything that the @async@
1240-- package can do and more. async provides applicative concurrency whereas
1241-- streamly provides both applicative and monadic concurrency. The 'ZipAsync'
1242-- type behaves like the applicative instance of async. This work was
1243-- originally inspired by the concurrency implementation in @transient@ though
1244-- it has no resemblence with that. Streamly provides concurrency as transient
1245-- does but in a sort of dual manner, it can lazily stream the output. In
1246-- comparison to transient streamly has a first class streaming interface and
1247-- is a monad transformer that can be used universally in any Haskell monad
1248-- transformer stack.
1250-- The non-determinism, concurrency and streaming combination make streamly a
1251-- strong FRP capable library as well. FRP is fundamentally stream of events
1252-- that can be processed concurrently. The example in this tutorial as well as
1253-- the "Streamly.Examples.CirclingSquare" example from Yampa demonstrate the
1254-- basic FRP capability of streamly. In core concepts streamly is strikingly
1255-- similar to @dunai@. dunai was designed from a FRP perspective and streamly
1256-- was originally designed from a concurrency perspective. However, both have
1257-- 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