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 streaming
10-- and concurrency in functional programming. You can write concurrent as well
11-- as non-concurrent applications with streamly. Streaming enables writing
12-- modular, composable and scalable applications with ease and concurrency
13-- allows you to make them scale and perform well.
14-- Streamly enables writing concurrent applications without being aware of
15-- threads or synchronization. No explicit thread control is needed, where
16-- applicable the concurrency rate is automatically controlled based on the
17-- demand by the consumer. However, combinators are provided to fine tune the
18-- concurrency control.
19-- Streaming and concurrency together enable expressing reactive applications
20-- conveniently. See 'Streamly.Examples' for a simple SDL based FRP example.
22-- In this tutorial we will go over the basic concepts and how to use the
23-- library. For examples and other ways to use the library see
24-- the module @Streamly.Examples@ as well.
26module Streamly.Tutorial
31 -- ** Generating Streams
34 -- ** Eliminating Streams
40 -- * Semigroup Style Composition
43 -- ** Serial composition ('<>')
46 -- ** Async composition ('<|')
49 -- ** Interleaved composition ('<=>')
52 -- ** Parallel composition ('<|>')
55 -- ** Custom composition
58 -- * Monoid Style composition
61 -- * Transforming Streams
67 -- *** Serial Iterations ('StreamT')
70 -- *** Concurrent Iterations ('AsyncT')
73 -- *** Interleaved Iterations ('InterleavedT')
74 -- $interleavedNesting
76 -- *** Fairly Concurrent Iterations ('ParallelT')
77 -- $fairlyConcurrentNesting
88 -- * Summary of Compositions
89 -- $compositionSummary
97 -- ** Parallel Zipping
100 -- * Concurrent Programming Examples
103 -- * Reactive Programming
106 -- * State Machine Model
112 -- * Interworking with Streaming Libraries
115 -- * Comparison with Existing Packages
121import Streamly.Prelude
122-- import Streamly.Examples
124import Control.Applicative
126import Control.Monad.IO.Class (MonadIO(..))
127import Control.Monad.Trans.Class (MonadTrans (lift))
131-- Streamly provides many different stream types depending on the desired
132-- composition style. The simplest type is 'StreamT'. 'StreamT' is a monad
133-- transformer, the type @StreamT m a@ represents a stream of values of type
134-- 'a' in some underlying monad 'm'. For example, @StreamT IO Int@ is a stream
135-- of 'Int' in 'IO' monad.
139-- Pure values can be placed into the stream type using 'return' or 'pure'.
140-- Effects in the IO monad can be lifted to the stream type using the 'liftIO'
141-- combinator. In a transformer stack we can lift actions from the lower monad
142-- using the 'lift' combinator. Some examples of streams with a single element:
145-- return 1 :: StreamT IO Int
148-- liftIO $ putStrLn "Hello world!" :: StreamT IO ()
151-- We can combine streams using '<>' to create streams of many elements:
154-- return 1 <> return 2 <> return 3 :: StreamT IO Int
159-- 'runStreamT' runs a composed 'StreamT' computation, lowering the type into
160-- the underlying monad and discarding the result stream:
165-- main = runStreamT $ liftIO $ putStrLn "Hello world!"
168-- 'toList' runs a stream computation and collects the result stream in a list
169-- in the underlying monad. 'toList' is a polymorphic function that works on
170-- multiple stream types belonging to the class 'Streaming'. Therefore, before
171-- you run a stream you need to tell how you want to interpret the stream by
172-- using one of the stream type combinators ('serially', 'asyncly', 'parallely'
173-- etc.). The combinator 'serially' is equivalent to annotating the type as @::
180-- xs \<- toList $ serially $ return 1 <> return 2
184-- For other ways to eliminate a stream see the module @Streamly.Prelude@.
187-- Streams of the same type can be combined into a composite stream in many
188-- different ways using one of the semigroup style binary composition operators
189-- i.e. '<>', '<=>', '<|', '<|>', 'mplus'. These operators work on all stream
190-- types ('StreamT', 'AsyncT' etc.) uniformly.
192-- To illustrate the concurrent aspects, we will use the following @delay@
193-- function to introduce a delay specified in seconds.
197-- import Control.Concurrent
199-- delay n = liftIO $ do
200-- threadDelay (n * 1000000)
202-- putStrLn (show tid ++ ": Delay " ++ show n)
207-- We have already seen, the '<>' operator. It composes two streams in series
208-- i.e. the first stream is completely exhausted and then the second stream is
209-- processed. The following example prints the sequence 3, 2, 1 and takes a
210-- total of 6 seconds because everything is serial:
213-- main = runStreamT $ delay 3 <> delay 2 <> delay 1
216-- ThreadId 36: Delay 3
217-- ThreadId 36: Delay 2
218-- ThreadId 36: Delay 1
222-- The '<=>' operator is serial like '<>' but it interleaves the two streams
223-- i.e. it yields one element from the first stream and then one element from
224-- the second stream, and so on. The following example prints the sequence 1,
225-- 3, 2, 4 and takes a total of 10 seconds because everything is serial:
228-- main = runStreamT $ (delay 1 <> delay 2) \<=> (delay 3 <> delay 4)
231-- ThreadId 36: Delay 1
232-- ThreadId 36: Delay 3
233-- ThreadId 36: Delay 2
234-- ThreadId 36: Delay 4
237-- Note that this operator cannot be used to fold infinite containers since it
238-- requires preserving the state until a stream is finished.
242-- The '<|' operator can run both computations concurrently, /when needed/.
243-- In the following example since the first computation blocks we start the
244-- next one in a separate thread and so on:
247-- main = runStreamT $ delay 3 <| delay 2 <| delay 1
250-- ThreadId 42: Delay 1
251-- ThreadId 41: Delay 2
252-- ThreadId 40: Delay 3
255-- This is the concurrent version of the '<>' operator. The computations are
256-- triggered in the same order as '<>' except that they are concurrent. When
257-- we have a tree of computations composed using this operator, the tree is
258-- traversed in DFS style just like '<>'.
261-- main = runStreamT $ (p 1 <| p 2) <| (p 3 <| p 4)
262-- where p = liftIO . print
271-- Concurrency provided by this operator is demand driven. The second
272-- computation is run concurrently with the first only if the first computation
273-- is not producing enough output to keep the stream consumer busy otherwise
274-- the second computation is run serially after the previous one. The number of
275-- concurrent threads is adapted dynamically based on the pull rate of the
276-- consumer of the stream.
277-- As you can see, in the following example the computations are run in a
278-- single thread one after another, because none of them blocks. However, if
279-- the thread consuming the stream were faster than the producer then it would
280-- have started parallel threads for each computation to keep up even if none
284-- main = runStreamT $ traced (sqrt 9) <| traced (sqrt 16) <| traced (sqrt 25)
292-- Since the concurrency provided by this operator is demand driven it cannot
293-- be used when the composed computations have timers that are relative to each
294-- other because all computations may not be started at the same time and
295-- therefore timers in all of them may not start at the same time. When
296-- relative timing among all computations is important or when we need to start
297-- all computations at once for some reason '<|>' must be used instead.
298-- However, '<|' is useful in situations when we want to optimally utilize the
299-- resources and we know that the computations can run in parallel but we do
300-- not care if they actually run in parallel or not, that decision is left to
301-- the scheduler. Also, note that this operator can be used to fold infinite
302-- containers in contrast to '<|>', because it does not require us to run all
303-- of them at the same time.
305-- The left bias (or the DFS style) of the operator '<|' is suggested by its
306-- shape. You can also think of this as an unbalanced version of the fairly
307-- parallel operator '<|>'.
311-- The 'Alternative' composition operator '<|>', like '<|', runs the composed
312-- computations concurrently. However, unlike '<|' it runs all of the
313-- computations in fairly parallel manner using a round robin scheduling
314-- mechanism. This can be considered as the concurrent version of the fairly
315-- interleaved serial operation '<=>'. Note that this cannot be used on
316-- infinite containers, as it will lead to an infinite sized scheduling queue.
318-- The following example sends a query to three search engines in parallel and
319-- prints the name of the search engine as a response arrives:
323-- import Network.HTTP.Simple
325-- main = runStreamT $ google \<|> bing \<|> duckduckgo
327-- google = get "https://www.google.com/search?q=haskell"
328-- bing = get "https://www.bing.com/search?q=haskell"
329-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
330-- get s = liftIO (httpNoBody (parseRequest_ s) >> putStrLn (show s))
335-- The 'async' API can be used to create references to asynchronously running
336-- stream computations. We can then use 'uncons' to explore the streams
337-- arbitrarily and then recompose individual elements to create a new stream.
338-- This way we can dynamically decide which stream to explore at any given
339-- time. Take an example of a merge sort of two sorted streams. We need to
340-- keep consuming items from the stream which has the lowest item in the sort
341-- order. This can be achieved using async references to streams. See
342-- 'Streamly.Examples.MergeSortedStreams'.
346-- Each of the semigroup compositions described has an identity that can be
347-- used to fold a possibly empty container. An empty stream is represented by
348-- 'nil' which can be represented in various standard forms as 'mempty',
349-- 'empty' or 'mzero'.
350-- Some fold utilities are also provided by the library for convenience:
352-- * 'foldWith' folds a 'Foldable' container of stream computations using the
353-- given composition operator.
354-- * 'foldMapWith' folds like foldWith but also maps a function before folding.
355-- * 'forEachWith' is like foldMapwith but the container argument comes before
356-- the function argument.
357-- * The 'each' primitive from @Streamly.Prelude@ folds a 'Foldable' container
358-- using the '<>' operator:
360-- All of the following are equivalent:
366-- toList . serially $ foldWith (<>) (map return [1..10]) >>= print
367-- toList . serially $ foldMapWith (<>) return [1..10] >>= print
368-- toList . serially $ forEachWith (<>) [1..10] return >>= print
369-- toList . serially $ each [1..10] >>= print
374-- The previous section discussed ways to merge the elements of two streams
375-- without doing any transformation on them. In this section we will explore
376-- how to transform streams using 'Functor', 'Applicative' or 'Monad' style
377-- compositions. The applicative and monad composition of all 'Streaming' types
378-- behave exactly the same way as a list transformer. For simplicity of
379-- illustration we are using streams of pure values in the following examples.
380-- However, the real application of streams arises when these streams are
381-- generated using monadic actions.
385-- In functional programmer's parlance the 'Monad' instance of 'Streaming'
386-- types implement non-determinism, exploring all possible combination of
387-- choices from both the streams. From an imperative programmer's point of view
388-- it behaves like nested loops i.e. for each element in the first stream and
389-- for each element in the second stream apply the body of the loop. If you are
390-- familiar with list transformer this behavior is exactly the same as that of
391-- a list transformer.
393-- Just like we saw in sum style compositions earlier, monadic composition also
394-- has multiple variants each of which exactly corresponds to one of the sum
395-- style composition variant.
399-- When we interpret the monadic composition as 'StreamT' we get a standard
400-- list transformer like serial composition.
403-- main = runStreamT $ do
408-- ThreadId 30: Delay 3
409-- ThreadId 30: Delay 2
410-- ThreadId 30: Delay 1
413-- As you can see the code after the @each@ statement is run three times, once
414-- for each value of @x@. All the three iterations are serial and run in the
415-- same thread one after another. When compared to imperative programming, this
416-- can also be viewed as a @for@ loop with three iterations.
418-- When multiple streams are composed using this style they nest in a DFS
419-- manner i.e. nested iterations of an iteration are executed before we proceed
420-- to the next iteration at higher level. This behaves just like nested @for@
421-- loops in imperative programming.
424-- main = runStreamT $ do
427-- liftIO $ putStrLn $ show (x, y)
436-- You will also notice that this is the monadic equivalent of the sum style
437-- composition using '<>'.
441-- When we interpret the monadic composition as 'AsyncT' we get a /concurrent/
442-- list-transformer like composition. Multiple monadic continuations (or loop
443-- iterations) may be started concurrently. Concurrency is demand driven
444-- i.e. more concurrent iterations are started only if the previous iterations
445-- are not able to produce enough output for the consumer of the output stream.
446-- This is the concurrent version of 'StreamT'.
449-- main = runAsyncT $ do
454-- ThreadId 40: Delay 1
455-- ThreadId 39: Delay 2
456-- ThreadId 38: Delay 3
459-- As you can see the code after the @each@ statement is run three times, once
460-- for each value of @x@. All the three iterations are concurrent and run in
461-- different threads. The iteration with least delay finishes first. When
462-- compared to imperative programming, this can be viewed as a @for@ loop
463-- with three concurrent iterations.
465-- Concurrency is demand driven just as in the case of '<|'. When multiple
466-- streams are composed using this style the iterations are triggered in a DFS
467-- manner just like 'StreamT' i.e. nested iterations are executed before we
468-- proceed to the next iteration at higher level. However, unlike 'StreamT'
469-- more than one iterations may be started concurrently, and based on the
470-- demand from the consumer.
473-- main = runAsyncT $ do
476-- liftIO $ putStrLn $ show (x, y)
485-- You will notice that this is the monadic equivalent of the '<|' style
486-- sum composition. The same caveats apply to this as the '<|' operation.
488-- $interleavedNesting
490-- When we interpret the monadic composition as 'InterleavedT' we get a serial
491-- but fairly interleaved list-transformer like composition. The monadic
492-- continuations or iterations of the outer loop are fairly interleaved with
493-- the continuations or iterations of the inner loop.
496-- main = runInterleavedT $ do
499-- liftIO $ putStrLn $ show (x, y)
508-- You will notice that this is the monadic equivalent of the '<=>' style
509-- sum composition. The same caveats apply to this as the '<=>' operation.
511-- $fairlyConcurrentNesting
513-- When we interpret the monadic composition as 'ParallelT' we get a
514-- /concurrent/ list-transformer like composition just like 'AsyncT'. The
515-- difference is that this is fully parallel with all iterations starting
516-- concurrently instead of the demand driven concurrency of 'AsyncT'.
519-- main = runParallelT $ do
524-- ThreadId 40: Delay 1
525-- ThreadId 39: Delay 2
526-- ThreadId 38: Delay 3
529-- You will notice that this is the monadic equivalent of the '<|>' style
530-- sum composition. The same caveats apply to this as the '<|>' operation.
534-- The streamly code is usually written in a way that is agnostic of the
535-- specific monadic composition type. We use a polymorphic type with a
536-- 'Streaming' type class constraint. When running the stream we can choose the
537-- specific mode of composition. For example look at the following code.
542-- composed :: Streaming t => t m a
547-- liftIO $ putStrLn $ show (sz, cl, sh)
551-- sizes = each [1, 2, 3]
552-- colors = each ["red", "green", "blue"]
553-- shapes = each ["triangle", "square", "circle"]
556-- Now we can interpret this in whatever way we want:
559-- main = runStreamT composed
560-- main = runAsyncT composed
561-- main = runInterleavedT composed
562-- main = runParallelT composed
565-- Equivalently, we can also write it using the type adapter combinators, like
569-- main = runStreaming $ serially $ composed
570-- main = runStreaming $ asyncly $ composed
571-- main = runStreaming $ interleaving $ composed
572-- main = runStreaming $ parallely $ composed
575-- As an exercise try to figure out the output of this code for each mode of
580-- The simplest transformation on a stream is mapping a function on
581-- all elements of the stream using 'fmap'.
586-- main = (toList $ serially $ fmap show $ each [1..10]) >>= print
591-- Applicative is a special case of monad and behaves just like the monadic
592-- composition. The following example runs all iterations serially and takes a
593-- total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
597-- import Streamly.Prelude
598-- import Control.Concurrent
602-- d n = delay n >> return n
604-- main = (toList . serially $ (,) \<$> s1 \<*> s2) >>= print
607-- ThreadId 36: Delay 1
608-- ThreadId 36: Delay 3
609-- ThreadId 36: Delay 4
610-- ThreadId 36: Delay 2
611-- ThreadId 36: Delay 3
612-- ThreadId 36: Delay 4
613-- [(1,3),(1,4),(2,3),(2,4)]
616-- Similalrly interleaving runs the iterations in an interleaved order but
617-- since it is serial it takes a total of 17 seconds:
620-- main = (toList . interleaving $ (,) \<$> s1 \<*> s2) >>= print
623-- ThreadId 36: Delay 1
624-- ThreadId 36: Delay 3
625-- ThreadId 36: Delay 2
626-- ThreadId 36: Delay 3
627-- ThreadId 36: Delay 4
628-- ThreadId 36: Delay 4
629-- [(1,3),(2,3),(1,4),(2,4)]
632-- 'AsyncT' can run the iterations concurrently and therefore takes a total
633-- of 10 seconds (1 + 2 + 3 + 4):
636-- main = (toList . asyncly $ (,) \<$> s1 \<*> s2) >>= print
639-- ThreadId 34: Delay 1
640-- ThreadId 36: Delay 2
641-- ThreadId 35: Delay 3
642-- ThreadId 36: Delay 3
643-- ThreadId 35: Delay 4
644-- ThreadId 36: Delay 4
645-- [(1,3),(2,3),(1,4),(2,4)]
648-- Similalrly 'ParallelT' as well can run the iterations concurrently and
649-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
652-- main = (toList . parallely $ (,) \<$> s1 \<*> s2) >>= print
655-- ThreadId 34: Delay 1
656-- ThreadId 36: Delay 2
657-- ThreadId 35: Delay 3
658-- ThreadId 36: Delay 3
659-- ThreadId 35: Delay 4
660-- ThreadId 36: Delay 4
661-- [(1,3),(2,3),(1,4),(2,4)]
667-- $compositionSummary
669-- The following table summarizes the types for monadic compositions and the
670-- operators for sum style compositions. This table captures the essence of
674-- +-----+--------------+------------+
675-- | | Serial | Concurrent |
676-- +=====+==============+============+
677-- | DFS | 'StreamT' | 'AsyncT' |
678-- | +--------------+------------+
680-- +-----+--------------+------------+
681-- | BFS | 'InterleavedT' | 'ParallelT' |
682-- | +--------------+------------+
683-- | | '<=>' | '<|>' |
684-- +-----+--------------+------------+
689-- Zipping is a special transformation where the corresponding elements of two
690-- streams are combined together using a zip function producing a new
691-- stream of outputs. Two different types are provided for serial and
692-- concurrent zipping. These types provide an applicative instance that zips
693-- the argument streams.
697-- 'StreamZip' zips streams serially:
701-- import Streamly.Prelude
702-- import Control.Concurrent
704-- d n = delay n >> return n
705-- s1 = adapt . serially $ d 1 <> d 2
706-- s2 = adapt . serially $ d 3 <> d 4
708-- main = (toList . zipping $ (,) \<$> s1 \<*> s2) >>= print
711-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
712-- everything runs serially:
715-- ThreadId 29: Delay 1
716-- ThreadId 29: Delay 3
717-- ThreadId 29: Delay 2
718-- ThreadId 29: Delay 4
724-- 'AsyncZip' zips streams concurrently:
728-- import Streamly.Prelude
729-- import Control.Concurrent
730-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
732-- d n = delay n >> return n
733-- s1 = adapt . serially $ d 1 <> d 2
734-- s2 = adapt . serially $ d 3 <> d 4
737-- liftIO $ hSetBuffering stdout LineBuffering
738-- (toList . zippingAsync $ (,) \<$> s1 \<*> s2) >>= print
741-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
742-- are produced concurrently, and 2 and 4 are produced concurrently:
745-- ThreadId 32: Delay 1
746-- ThreadId 32: Delay 2
747-- ThreadId 33: Delay 3
748-- ThreadId 33: Delay 4
754-- There are two ways to achieve concurrency. We can generate individual
755-- elements of a stream concurrently by folding with parallel composition
756-- operators i.e. '<|' or '<|>'. 'forEachWith' can be useful in such cases.
758-- In the following example, we square each number concurrently but then
759-- sum and print them serially:
763-- import Streamly.Prelude (toList)
764-- import Data.List (sum)
767-- squares \<- toList $ serially $ foldMapWith (<|) (\x -\> return $ x * x) [1..100]
768-- print $ sum squares
771-- The following example not just computes the squares concurrently but also
772-- computes the square root of their sums concurrently by using the parallel
777-- import Streamly.Prelude (toList)
778-- import Data.List (sum)
781-- z \<- toList $ asyncly $ do
782-- xsq \<- foldMapWith (\<|) (\\x -> return $ x * x) [1..100]
783-- ysq \<- foldMapWith (\<|) (\\x -> return $ x * x) [1..100]
784-- return $ sqrt (xsq + ysq)
790-- Let us see a reactive programming example:
793-- {-\# LANGUAGE FlexibleContexts #-}
796-- import Control.Concurrent (threadDelay)
797-- import Control.Monad (when)
798-- import Control.Monad.State
799-- import Data.Semigroup (cycle1)
801-- data Event = Harm Int | Heal Int | Quit deriving (Show)
803-- userAction :: MonadIO m => StreamT m Event
804-- userAction = cycle1 $ liftIO askUser
809-- "potion" -> return (Heal 10)
810-- "quit" -> return Quit
811-- _ -> putStrLn "What?" >> askUser
813-- acidRain :: MonadIO m => StreamT m Event
814-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
816-- game :: (MonadAsync m, MonadState Int m) => StreamT m ()
818-- event \<- userAction \<|> acidRain
820-- Harm n -> modify $ \\h -> h - n
821-- Heal n -> modify $ \\h -> h + n
822-- Quit -> fail "quit"
825-- when (h <= 0) $ fail "You die!"
826-- liftIO $ putStrLn $ "Health = " ++ show h
829-- putStrLn "Your health is deteriorating due to acid rain,\\
830-- \\ type \\"potion\\" or \\"quit\\""
831-- _ <- runStateT (runStreamT game) 60
836-- State machine stuff
840-- Even though streamly covers all that is provided by the 'async' package or
841-- most of what is provided by the 'streaming', 'pipes' or 'conduit' packages,
842-- I would not say that it renders those useless. Streamly is like monad if
843-- 'async' is applicative and monads and applicatives both have their use
844-- cases. It can completely repalce 'async', the ZipAsync type is equivalent to
845-- the functionality provided by 'async'.
847-- pipes and conduit are like Arrows and streamly is like monad to them. You
848-- would use pipes and conduit when you do not need the product style
849-- composition and the implicit concurrency.
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