on dfecce2 help sign in
streamly/src/Streamly/Tutorial.hs
1522 lines 56 kB history raw
1{-# OPTIONS_GHC -fno-warn-unused-imports #-}
2-- |
3-- Module : Streamly.Tutorial
4-- Copyright : (c) 2017 Harendra Kumar
5--
6-- License : BSD3
7-- Maintainer : harendra.kumar@gmail.com
8--
9-- Streamly is a general computing framework based on streaming IO. The IO
10-- monad and pure lists are a special case of streamly. On one hand, streamly
11-- extends the lists of pure values to lists of monadic actions, on the other
12-- hand it extends the IO monad with concurrrent non-determinism. In simple
13-- imperative terms we can say that streamly extends the IO monad with @for@
14-- loops and nested @for@ loops with concurrency support. You can understand
15-- this analogy better once you can go through this tutorial.
16--
17-- Streaming in general enables writing modular, composable and scalable
18-- applications with ease, and concurrency allows you to make them scale and
19-- perform well. Streamly enables writing scalable concurrent applications
20-- without being aware of threads or synchronization. No explicit thread
21-- control is needed, where applicable the concurrency rate is automatically
22-- controlled based on the demand by the consumer. However, combinators can be
23-- used to fine tune the concurrency control.
24--
25-- Streaming and concurrency together enable expressing reactive applications
26-- conveniently. See the @CirclingSquare@ example in the examples directory for
27-- a simple SDL based FRP example. To summarize, streamly provides a unified
28-- computing framework for streaming, non-determinism and functional reactive
29-- programming in an elegant and simple API that is a natural extension of pure
30-- lists to monadic streams.
31--
32-- In this tutorial we will go over the basic concepts and how to
33-- use the library. See the last section for further reading resources.
35module Streamly.Tutorial
36 (
37 -- * Streams
38 -- $streams
40 -- * Flavors of Streams
41 -- $flavors
43 -- * Imports and Supporting Code
44 -- $imports
46 -- * Generating Streams
47 -- $generating
49 -- * Eliminating Streams
50 -- $eliminating
52 -- * Transforming Streams
53 -- $transformation
55 -- * Merging Streams
57 -- ** Semigroup Style
58 -- $semigroup
60 -- *** Deep Serial Composition ('Serial')
61 -- $serial
63 -- *** Wide Serial Composition ('WSerial')
64 -- $interleaved
66 -- *** Deep Asynchronous Composition ('Async')
67 -- $async
69 -- *** Wide Asynchronous Composition ('WAsync')
70 -- $wasync
72 -- *** Parallel Asynchronous Composition ('Parallel')
73 -- $parallel
75 -- *** Custom composition
76 -- $custom
78 -- ** Monoid Style
79 -- $monoid
81 -- * Nesting Streams
82 -- $nesting
84 -- ** Monad
85 -- $monad
87 -- *** Deep Serial Nesting ('Serial')
88 -- $regularSerial
90 -- *** Wide Serial Nesting ('WSerial')
91 -- $interleavedNesting
93 -- *** Deep Asynchronous Nesting ('Async')
94 -- $concurrentNesting
96 -- *** Wide Asynchronous Nesting ('WAsync')
97 -- $wasyncNesting
99 -- *** Parallel Asynchronous Nesting ('Parallel')
100 -- $parallelNesting
102 -- *** Exercise
103 -- $monadExercise
105 -- ** Applicative
106 -- $applicative
108 -- ** Functor
109 -- $functor
111 -- * Zipping Streams
112 -- $zipping
114 -- ** Serial Zipping
115 -- $serialzip
117 -- ** Parallel Zipping
118 -- $parallelzip
120 -- * Monad transformers
121 -- $monadtransformers
123 -- * Concurrent Programming
124 -- $concurrent
126 -- * Reactive Programming
127 -- $reactive
129 -- * Performance
130 -- $performance
132 -- * Interoperation with Streaming Libraries
133 -- $interop
135 -- * Comparison with Existing Packages
136 -- $comparison
138 -- * Where to go next?
139 -- $furtherReading
140 )
141where
143import Streamly
144import Streamly.Prelude
145import Data.Semigroup
146import Control.Applicative
147import Control.Monad
148import Control.Monad.IO.Class (MonadIO(..))
149import Control.Monad.Trans.Class (MonadTrans (lift))
151-- $streams
152--
153-- The way a list represents a sequence of pure values, a stream represents a
154-- sequence of monadic actions. The monadic stream API offered by Streamly is
155-- very close to the Haskell "Prelude" pure lists' API, it can be considered as a
156-- natural extension of lists to monadic actions. Streamly streams provide
157-- concurrent composition and merging of streams. It can be considered as a
158-- concurrent list transformer. In contrast to the "Prelude" lists, merging or
159-- appending streams of arbitrary length is scalable and inexpensive.
160--
161-- The basic stream type is 'Serial', it represents a sequence of IO actions,
162-- and is a 'Monad'. The 'Serial' monad is almost a drop in replacement for
163-- the 'IO' monad, IO monad is a special case of the 'Serial' monad; IO monad
164-- represents a single IO action whereas the 'Serial' monad represents a series
165-- of IO actions. The only change you need to make to go from 'IO' to 'Serial'
166-- is to use 'runStream' to run the monad and to prefix the IO actions with
167-- either 'once' or 'liftIO'. If you use liftIO you can switch from 'Serial'
168-- to IO monad by simply removing the 'runStream' function; no other changes
169-- are needed unless you have used some stream specific composition or
170-- combinators.
171--
172-- Similarly, the 'Serial' type is almost a drop in replacement for pure lists,
173-- pure lists are a special case of monadic streams. If you use 'nil' in place
174-- of '[]' and '|:' in place ':' you can replace a list with a 'Serial' The
175-- only difference is that the elements must be monadic type and to operate on
176-- the streams we must use the corresponding functions from "Streamly.Prelude"
177-- instead of using the base "Prelude".
179-- $flavors
180--
181-- There are a few more types similar to 'Serial' that all represent streams
182-- and differ only in the 'Semigroup', 'Applicative' and 'Monad' compositions
183-- of the streams.
184--
185-- The composition of two or more streams is distinguished based on three
186-- characterstics, /traversal order/, /execution order/ and
187-- /consumption order/. Traversal of a composition of streams could be @deep@
188-- or @wide@. Deep goes depth first i.e. each stream is traversed fully
189-- before we traverse the next stream. Wide goes breadth first i.e. one element
190-- from each stream is traversed before coming back to the first stream again.
191-- Execution could be serial (i.e. synchronous) or asynchronous. In serial
192-- execution we execute an action in the next stream only after the first has
193-- finished executing. In asynchronous execution actions in both streams can be
194-- executed asynchronously i.e. the next action can start executing even before
195-- the first one has finished. The third parameter is consumption order that is
196-- in what order the output generated by the composition is consumed.
197-- Consumption could be serial or asynchronous. In serial consumption, the
198-- outputs are consumed in the traversal order, in asynchronous consumption the
199-- outputs are consumed as they arrive i.e. first come first serve order.
200--
201-- The following table summarizes different styles of streams based on how they
202-- compose. All these types are monads and they differ in 'Semigroup',
203-- 'Applicative' and 'Monad' compositions:
204--
205-- @
206-- +------------+-----------+--------------+--------------+
207-- | Type | Traversal | Execution | Consumption |
208-- +============+===========+==============+==============+
209-- | 'Serial' | Deep | Serial | Serial |
210-- +------------+-----------+--------------+--------------+
211-- | 'WSerial' | Wide | Serial | Serial |
212-- +------------+-----------+--------------+--------------+
213-- | 'Async' | Deep | Asynchronous | Asynchronous |
214-- +------------+-----------+--------------+--------------+
215-- | 'WAsync' | Wide | Asynchronous | Asynchronous |
216-- +------------+-----------+--------------+--------------+
217-- | 'Parallel' | Parallel | Asynchronous | Asynchronous |
218-- +------------+-----------+--------------+--------------+
219-- @
220--
221-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that
222-- zip streams serially or concurrently using 'Applicative' operation. These
223-- types are not monads they are only applicatives and they do not differ in
224-- 'Semigroup' composition.
225--
226-- All these types can be freely inter-converted using type conversion
227-- combinators or type annotations without any cost, to acheive the desired
228-- composition style. To force a particular type of composition we coerce the
229-- stream type using the corresponding type adapting combinator from
230-- 'serially', 'wSerially', 'asyncly', 'wAsyncly', 'parallely', 'zipSerially'
231-- or 'zipAsyncly'. The default stream type is inferred as 'Serial' unless you
232-- change it by using one of the combinators or using a type annotation.
234-- $monadtransformers
235--
236-- To represent streams in an arbitrary monad use the more general monad
237-- transformer types for example the monad transformer type corresponding to
238-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values
239-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a
240-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for
241-- @SerialT IO@.
242--
243-- Similarly we have monad transformer types for other stream types as well viz.
244-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.
245--
246-- To lift a value from an underlying monad in a monad transformer stack into a
247-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.
248--
249-- @
250-- > 'runStream' $ liftIO $ putStrLn "Hello world!"
251-- Hello world!
252-- > 'runStream' $ lift $ putStrLn "Hello world!"
253-- Hello world!
254-- @
255--
257-- $generating
258--
259-- We will assume the following imports in this tutorial. Go ahead, fire up a
260-- GHCi session and import these lines to start playing.
261--
262-- @
263-- > import "Streamly"
264-- > import "Streamly.Prelude" ((|:))
265-- > import qualified "Streamly.Prelude" as S
266-- @
267--
268-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds
269-- a monadic action at the head of the stream.
270--
271-- @
272-- > S.'toList' S.'nil'
273-- []
274-- > S.'toList' $ 'getLine' |: 'getLine' |: S.'nil'
275-- hello
276-- world
277-- ["hello","world"]
278-- @
279--
280-- To create a singleton stream from a pure value use 'pure' and to create a
281-- singleton stream from a monadic action use 'once'.
282--
283-- @
284-- > S.'toList' $ 'pure' 1
285-- [1]
286-- > S.'toList' $ S.'once' 'getLine'
287-- hello
288-- ["hello"]
289-- @
290--
291-- To create a stream from pure values in a 'Foldable' container use
292-- 'fromFoldable' which is equivalent to a fold using 'cons' and 'nil':
293--
294-- @
295-- > S.'toList' $ S.'fromFoldable' [1..3]
296-- [1,2,3]
297-- > S.'toList' $ 'Prelude.foldr' S.'cons' S.'nil' [1..3]
298-- [1,2,3]
299-- @
300--
301-- To create a stream from monadic actions in a 'Foldable' container just use a
302-- right fold using 'consM' and 'nil':
303--
304-- @
305-- > 'runStream' $ 'Prelude.foldr' ('|:') S.'nil' ['putStr' "Hello ", 'putStrLn' "world!"]
306-- Hello world!
307-- @
308--
309-- For more ways to construct a stream see the module "Streamly.Prelude".
310--
311-- Note that monadic generation functions e.g. 'consM', 'unfoldrM',
312-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' etc work
313-- concurrently when used with appropriate stream type combinator. The pure
314-- versions do not work concurrently, however you can use the monadic versions
315-- even for pure computations to get the concurrent generation capability where
316-- required.
317--
318-- TBD add more details and examples about concurrent generation.
320-- $eliminating
321--
322-- We have already seen 'runStream' and 'toList' to eliminate a stream in the
323-- examples above. 'runStream' runs a stream discarding the results i.e. only
324-- for effects. 'toList' runs the stream and collects the results in a list.
325--
326-- For other ways to eliminate a stream see the @Folding@ section in
327-- "Streamly.Prelude" module.
329-- $transformation
330--
331-- Transformation over a stream is the equivalent of a @for@ loop construct in
332-- imperative paradigm. We iterate over every element in the stream and perform
333-- certain transformations for each element. Transformations may involve
334-- mapping functions over the elements, filtering elements from the stream or
335-- folding all the elements in the stream into a single value. Streamly streams
336-- are exactly like lists and you can perform all the transformations in the
337-- same way as you would on lists.
338--
339-- Here is a simple console echo program that just echoes every input line,
340-- forever:
341--
342-- @
343-- > 'runStream' $ S.'repeatM' getLine & S.'mapM' putStrLn
344-- @
345--
346-- The following code snippet reads lines from standard input, filters blank
347-- lines, drops the first non-blank line, takes the next two, up cases them,
348-- numbers them and prints them:
349--
350-- @
351-- import "Streamly"
352-- import qualified "Streamly.Prelude" as S
353-- import Data.Char (toUpper)
354-- import Data.Function ((&))
355--
356-- main = 'runStream' $
357-- S.'repeatM' getLine
358-- & S.'filter' (not . null)
359-- & S.'drop' 1
360-- & S.'take' 2
361-- & fmap (map toUpper)
362-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])
363-- & S.'mapM' putStrLn
364-- @
365--
366-- Note that monadic transformation functions 'mapM' and 'sequence' work
367-- concurrently when used with appropriate stream type combinator. The pure
368-- versions do not work concurrently, however you can use the monadic versions
369-- even for pure computations to get the concurrent transformation capability
370-- where required.
371--
372-- TBD add more details and examples about concurrent transformation.
374-- $semigroup
375--
376-- We can combine two streams into a single stream using semigroup composition
377-- operation '<>'. Streams can be combined in many different ways as described
378-- in the following sections, the '<>' operation behaves differently depending
379-- on the stream type in effect. The stream type and therefore the composition
380-- style can be changed at any point using one of the type combinators as
381-- discussed earlier.
383-- $imports
384--
385-- In most of example snippets we do not repeat the imports. Where imports are
386-- not explicitly specified use the imports shown below.
387--
388-- @
389-- import "Streamly"
390-- import "Streamly.Prelude" ((|:), nil)
391-- import qualified "Streamly.Prelude" as S
392--
393-- import Control.Concurrent
394-- import Control.Monad (forever)
395-- @
396--
397-- To illustrate concurrent vs serial composition aspects, we will use the
398-- following @delay@ function to introduce a sleep or delay specified in
399-- seconds. After the delay it prints the number of seconds it slept.
400--
401-- @
402-- delay n = S.'once' $ do
403-- threadDelay (n * 1000000)
404-- tid \<- myThreadId
405-- putStrLn (show tid ++ ": Delay " ++ show n)
406-- @
408-- $serial
409--
410-- The 'Semigroup' operation '<>' of the 'Serial' type combines the two streams
411-- in a /serial depth first/ manner. We use the 'serially' type combinator to
412-- effect 'Serial' style of composition. We can also use an explicit 'Serial'
413-- type annotation for the stream to acheive the same effect. However, since
414-- 'Serial' is the default type unless explicitly specified by using a
415-- combinator, we can omit using an explicit combinator or type annotation for
416-- this style of composition.
417--
418-- When two streams with multiple elements are combined in this manner, the
419-- monadic actions in the two streams are performed sequentially i.e. first all
420-- actions in the first stream are performed sequentially and then all actions
421-- in the second stream are performed sequentially. We call it
422-- /serial depth first/ as the full depth of one stream is fully traversed
423-- before we move to the next. The following example prints the sequence 1, 2,
424-- 3, 4:
425--
426-- @
427-- main = 'runStream' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
428-- @
429-- @
430-- 1
431-- 2
432-- 3
433-- 4
434-- @
435--
436-- All actions in both the streams are performed serially in the same thread.
437-- In the following example we can see that all actions are performed in the
438-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
439--
440-- @
441-- main = 'runStream' $ delay 3 <> delay 2 <> delay 1
442-- @
443-- @
444-- ThreadId 36: Delay 3
445-- ThreadId 36: Delay 2
446-- ThreadId 36: Delay 1
447-- @
448--
449-- The polymorphic version of the binary operation '<>' of the 'Serial' type is
450-- 'serial'. We can use 'serial' to join streams in a sequential manner
451-- irrespective of the type of stream:
452--
453-- @
454-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`serial` (print 3 |: print 4 |: nil)
455-- @
457-- $interleaved
458--
459-- The 'Semigroup' operation '<>' of the 'WSerial' type combines the two
460-- streams in a /serial breadth first/ manner. We use the 'wSerially' type
461-- combinator to effect 'WSerial' style of composition. We can also use the
462-- 'WSerial' type annotation for the stream to acheive the same effect.
463--
464-- When two streams with multiple elements are combined in this manner, we
465-- traverse all the streams in a breadth first manner i.e. one action from each
466-- stream is peformed and yielded to the resulting stream before we come back
467-- to the first stream again and so on.
468-- The following example prints the sequence 1, 3, 2, 4
469--
470-- @
471-- main = 'runStream' . 'wSerially' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
472-- @
473-- @
474-- 1
475-- 3
476-- 2
477-- 4
478-- @
479--
480-- Even though the monadic actions of the two streams are performed in an
481-- interleaved manner they are all performed serially in the same thread. In
482-- the following example we can see that all actions are performed in the same
483-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:
484--
485-- @
486-- main = 'runStream' . 'wSerially' $ delay 3 <> delay 2 <> delay 1
487-- @
488-- @
489-- ThreadId 36: Delay 3
490-- ThreadId 36: Delay 2
491-- ThreadId 36: Delay 1
492-- @
493--
494-- The polymorphic version of the 'WSerial' binary operation '<>' is called
495-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner
496-- irrespective of the type, notice that we have not used the 'wSerially'
497-- combinator in the following example:
498--
499-- @
500-- main = 'runStream' $ (print 1 |: print 2 |: nil) \`wSerial` (print 3 |: print 4 |: nil)
501-- @
502-- @
503-- 1
504-- 3
505-- 2
506-- 4
507-- @
508--
509-- Note that this composition cannot be used to fold infinite number of streams
510-- since it requires preserving the state until a stream is finished.
512-- $async
513--
514-- The 'Semigroup' operation '<>' of the 'Async' type combines the two
515-- streams in a depth first manner with parallel look ahead. We use the
516-- 'asyncly' type combinator to effect 'Async' style of composition. We
517-- can also use the 'Async' type annotation for the stream type to acheive
518-- the same effect.
519--
520-- When two streams with multiple elements are combined in this manner, the
521-- streams are traversed in depth first manner just like 'Serial', however it
522-- can execute the next stream concurrently and return the results from it
523-- as they arrive i.e. the results from the next stream may be yielded even
524-- before the results from the first stream. Concurrent execution of the next
525-- stream(s) is performed if the first stream blocks or if it cannot produce
526-- output at the rate that is enough to meet the consumer demand. Multiple
527-- streams can be executed concurrently to meet the demand.
528-- In the following example the first stream does not block,
529-- therefore the first stream is completely exhausted before the second.
530--
531-- @
532-- main = 'runStream' . 'asyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
533-- @
534-- @
535-- 1
536-- 2
537-- 3
538-- 4
539-- @
540--
541-- If the first stream blocks, we can yield from the second. In the example
542-- below each yield in the stream has a constant delay of 1 second therefore 1
543-- and 3 would be yielded first and then 2 and 4 would be yielded.
544--
545-- @
546-- main = 'runStream' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
547-- where p n = threadDelay 1000000 >> print n
548-- @
549-- @
550-- 1
551-- 3
552-- 2
553-- 4
554-- @
555--
556-- In the following example we can see that new threads are started when a
557-- computation blocks. Notice that the output from the stream with the
558-- shortest delay is printed first. The whole computation takes @maximum of
559-- (3, 2, 1) = 3@ seconds:
560--
561-- @
562-- main = 'runStream' . 'asyncly' $ delay 3 '<>' delay 2 '<>' delay 1
563-- @
564-- @
565-- ThreadId 42: Delay 1
566-- ThreadId 41: Delay 2
567-- ThreadId 40: Delay 3
568-- @
569--
570-- When we have a tree of computations composed using this style, the tree is
571-- traversed in DFS style just like the 'Serial' style, the only difference is
572-- that here we can move on to executing the next stream if a stream blocks.
573-- However, we will not start new threads if we have sufficient output to
574-- saturate the consumer. This is why we call it left-biased demand driven or
575-- adaptive concurrency style, the concurrency tends to stay on the left side
576-- of the composition as long as possible. More threads are started based on
577-- the pull rate of the consumer. The following example prints an output every
578-- second as all of the actions are concurrent.
579--
580-- @
581-- main = 'runStream' . 'asyncly' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)
582-- @
583-- @
584-- 1
585-- 2
586-- 3
587-- 4
588-- @
589--
590-- All the computations may even run in a single thread when more threads are
591-- not needed. As you can see, in the following example the computations are
592-- run in a single thread one after another, because none of them blocks.
593-- However, if the thread consuming the stream were faster than the producer
594-- then it would have started parallel threads for each computation to keep up
595-- even if none of them blocks:
596--
597-- @
598-- main = 'runStream' . 'asyncly' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)
599-- where traced m = S.'once' (myThreadId >>= print) >> return m
600-- @
601-- @
602-- ThreadId 40
603-- ThreadId 40
604-- ThreadId 40
605-- @
606--
607-- Note that the order of printing in the above examples may change due to
608-- variations in scheduling latencies for concurrent threads.
609--
610-- The polymorphic version of the 'Async' binary operation '<>' is called
611-- 'async'. We can use 'async' to join streams in a left biased
612-- adaptively concurrent manner irrespective of the type, notice that we have
613-- not used the 'asyncly' combinator in the following example:
614--
615-- @
616-- main = 'runStream' $ delay 3 \`async` delay 2 \`async` delay 1
617-- @
618-- @
619-- ThreadId 42: Delay 1
620-- ThreadId 41: Delay 2
621-- ThreadId 40: Delay 3
622-- @
623--
624-- Since the concurrency provided by this operator is demand driven it cannot
625-- be used when the composed computations start timers that are relative to
626-- each other because all computations may not be started at the same time and
627-- therefore timers in all of them may not start at the same time. When
628-- relative timing among all computations is important or when we need to start
629-- all computations at once for any reason 'Parallel' style must be used
630-- instead.
631--
632-- 'Async' style should be preferred over 'Parallel' or 'WAsync' unless you
633-- really need those. It utilizes the resources optimally. It should be used
634-- when we know that the computations can run in parallel but we do not care if
635-- they actually run in parallel or not, that decision can be left to the
636-- scheduler based on demand. Also, note that this operator can be used to fold
637-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,
638-- because it does not require us to run all of them at the same time in a fair
639-- manner.
641-- $wasync
642--
643-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in
644-- a concurrent manner using /breadth first traversal/. We use the 'wAsyncly'
645-- type combinator to effect 'WAsync' style of composition. We can also use the
646-- 'WAsync' type annotation for the stream to acheive the same effect.
647--
648-- When streams with multiple elements are combined in this manner, we traverse
649-- all the streams concurrently in a breadth first manner i.e. one action from
650-- each stream is peformed and yielded to the resulting stream before we come
651-- back to the first stream again and so on. Even though we execute the actions
652-- in a breadth first order the outputs may be consumed in a different order
653-- because they are consumed on a first come first serve basis.
654--
655-- In the following example we can see that outputs are produced in the breadth
656-- first travresal order but this is not guaranteed.
657--
658-- @
659-- main = 'runStream' . 'wAsyncly' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)
660-- @
661-- @
662-- 1
663-- 3
664-- 2
665-- 4
666-- @
667--
668-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is
669-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first
670-- concurrent traversal irrespective of the type, notice that we have not used
671-- the 'wAsyncly' combinator in the following example:
672--
673-- @
674-- main = 'runStream' $ delay 3 \`wAsync` delay 2 \`wAsync` delay 1
675-- @
676-- @
677-- ThreadId 42: Delay 1
678-- ThreadId 41: Delay 2
679-- ThreadId 40: Delay 3
680-- @
681--
682-- Since the concurrency provided by this style is demand driven it may not
683-- be used when the composed computations start timers that are relative to
684-- each other because all computations may not be started at the same time and
685-- therefore timers in all of them may not start at the same time. When
686-- relative timing among all computations is important or when we need to start
687-- all computations at once for any reason 'Parallel' style must be used
688-- instead.
689--
691-- $parallel
692--
693-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two
694-- streams in a fairly concurrent manner with round robin scheduling. We use
695-- the 'parallely' type combinator to effect 'Parallel' style of composition.
696-- We can also use the 'Parallel' type annotation for the stream type to
697-- acheive the same effect.
698--
699-- When two streams with multiple elements are combined in this manner, the
700-- monadic actions in both the streams are performed concurrently with a fair
701-- round robin scheduling. The outputs are yielded in the order in which the
702-- actions complete. This is pretty similar to the 'WAsync' type, the
703-- difference is that 'WAsync' is adaptive to the consumer demand and may or
704-- may not execute all actions in parallel depending on the demand, whereas
705-- 'Parallel' runs all the streams in parallel irrespective of the demand.
706--
707-- The following example sends a query to all the three search engines in
708-- parallel and prints the name of the search engines in the order in which the
709-- responses arrive:
710--
711-- @
712-- import "Streamly"
713-- import qualified Streamly.Prelude as S
714-- import Network.HTTP.Simple
715--
716-- main = 'runStream' . 'parallely' $ google \<> bing \<> duckduckgo
717-- where
718-- google = get "https://www.google.com/search?q=haskell"
719-- bing = get "https://www.bing.com/search?q=haskell"
720-- duckduckgo = get "https://www.duckduckgo.com/?q=haskell"
721-- get s = S.'once' (httpNoBody (parseRequest_ s) >> putStrLn (show s))
722-- @
723--
724-- The polymorphic version of the binary operation '<>' of the 'Parallel' type
725-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent
726-- manner irrespective of the type, notice that we have not used the
727-- 'parallely' combinator in the following example:
728--
729-- @
730-- main = 'runStream' $ delay 3 \`parallel` delay 2 \`wAsync` delay 1
731-- @
732-- @
733-- ThreadId 42: Delay 1
734-- ThreadId 41: Delay 2
735-- ThreadId 40: Delay 3
736-- @
737--
738-- Note that this style of composition cannot be used to combine infinite
739-- number of streams, as it will lead to an infinite sized scheduling queue.
740--
742-- $custom
743--
744-- The 'mkAsync' API can be used to create references to asynchronously running
745-- stream computations. We can then use 'uncons' to explore the streams
746-- arbitrarily and then recompose individual elements to create a new stream.
747-- This way we can dynamically decide which stream to explore at any given
748-- time. Take an example of a merge sort of two sorted streams. We need to
749-- keep consuming items from the stream which has the lowest item in the sort
750-- order. This can be achieved using async references to streams. See
751-- "MergeSort.hs" in the examples directory.
753-- $monoid
754--
755-- We can use 'Monoid' instances to fold a container of streams in the desired
756-- style using 'fold' or 'foldMap'. We have also provided some fold utilities
757-- to fold streams using the polymorphic combine operations:
758--
759-- * 'foldWith' is like 'fold', it folds a 'Foldable' container of streams
760-- using the given composition operator.
761-- * 'foldMapWith' is like 'foldMap', it folds like @foldWith@ but also maps a
762-- function before folding.
763-- * 'forEachWith' is like @foldMapwith@ but the container argument comes before
764-- the function argument.
765--
766-- All of the following are equivalent and start ten concurrent tasks each with
767-- a delay from 1 to 10 seconds, resulting in the printing of each number every
768-- second:
769--
770-- @
771-- import "Streamly"
772-- import qualified "Streamly.Prelude" as S
773-- import Control.Concurrent
774--
775-- main = do
776-- 'runStream' $ 'asyncly' $ foldMap delay [1..10]
777-- 'runStream' $ 'foldWith' 'async' (map delay [1..10])
778-- 'runStream' $ 'foldMapWith' 'async' delay [1..10]
779-- 'runStream' $ 'forEachWith' 'async' [1..10] delay
780-- where delay n = S.'once' $ threadDelay (n * 1000000) >> print n
781-- @
783-- $nesting
784--
785-- Till now we discussed ways to apply transformations on a stream or to merge
786-- streams together to create another stream. We mentioned earlier that
787-- transforming a stream is similar to a @for@ loop in the imperative paradigm.
788-- We will now discuss the concept of a nested composition of streams which is
789-- analogous to nested @for@ loops in the imperative paradigm. Functional
790-- programmers call this style of composition a list transformer or @ListT@.
791-- Logic programmers call it a logic monad or non-deterministic composition,
792-- but for ordinary imperative minded people like me it is easier to think in
793-- terms of good old nested @for@ loops.
794--
795-- $monad
796--
797-- In functional programmer's parlance the 'Monad' instances of different
798-- 'IsStream' types implement non-determinism, exploring all possible
799-- combination of choices from both the streams. From an imperative
800-- programmer's point of view it behaves like nested loops i.e. for each
801-- element in the first stream and for each element in the second stream
802-- execute the body of the loop.
803--
804-- The 'Monad' instances of 'Serial', 'WSerial', 'Async' and 'WAsync'
805-- stream types support different flavors of nested looping. In other words,
806-- they are all variants of list transformer. The nesting behavior of these
807-- types correspond exactly to the way they merge streams as we discussed in
808-- the previous section.
809--
811-- $regularSerial
812--
813-- The 'Monad' composition of the 'Serial' type behaves like a standard list
814-- transformer. This is the default when we do not use an explicit type
815-- combinator. However, the 'serially' type combinator can be used to switch to
816-- this style of composition. We will see how this style of composition works
817-- in the following examples.
818--
819-- Let's start with an example with a simple @for@ loop without any nesting.
820-- For simplicity of illustration we are using streams of pure values in all
821-- the examples. However, the streams could also be made of monadic actions
822-- instead.
823--
824-- @
825-- import "Streamly"
826-- import qualified "Streamly.Prelude" as S
827--
828-- main = 'runStream' $ do
829-- x <- S.'fromFoldable' [3,2,1]
830-- delay x
831-- @
832-- @
833-- ThreadId 30: Delay 3
834-- ThreadId 30: Delay 2
835-- ThreadId 30: Delay 1
836-- @
837--
838-- As we can see, the code after the @fromFoldable@ statement is run three
839-- times, once for each value of @x@ drawn from the stream. All the three
840-- iterations are serial and run in the same thread one after another. In
841-- imperative terms this is equivalent to a @for@ loop with three iterations.
842--
843-- A console echo loop copying standard input to standard output can simply be
844-- written like this:
845--
846-- @
847-- import "Streamly"
848-- import qualified "Streamly.Prelude" as S
849--
850-- main = 'runStream' $ forever $ S.once getLine >>= S.once . putStrLn
851-- @
852--
853-- When multiple streams are composed using this style they nest in a DFS
854-- manner i.e. nested iterations of a loop are executed before we proceed to
855-- the next iteration of the parent loop. This behaves just like nested @for@
856-- loops in imperative programming.
857--
858-- @
859-- import "Streamly"
860-- import qualified "Streamly.Prelude" as S
861--
862-- main = 'runStream' $ do
863-- x <- S.'fromFoldable' [1,2]
864-- y <- S.'fromFoldable' [3,4]
865-- S.'once' $ putStrLn $ show (x, y)
866-- @
867-- @
868-- (1,3)
869-- (1,4)
870-- (2,3)
871-- (2,4)
872-- @
873--
874-- Notice that this is analogous to merging streams of type 'Serial' or merging
875-- streams using 'serial'.
877-- $concurrentNesting
878--
879-- The 'Monad' composition of 'Async' type can perform the iterations of a
880-- loop concurrently. Concurrency is demand driven i.e. more concurrent
881-- iterations are started only if the previous iterations are not able to
882-- produce enough output for the consumer of the output stream. This works
883-- exactly the same way as the merging of two streams 'asyncly' works.
884-- This is the concurrent analogue of 'Serial' style monadic composition.
885--
886-- The 'asyncly' type combinator can be used to switch to this style of
887-- composition. Alternatively, a type annotation can be used to specify the
888-- type of the stream as 'Async'.
889--
890-- @
891-- import "Streamly"
892-- import "Streamly.Prelude"
893--
894-- main = 'runStream' . 'asyncly' $ do
895-- x <- S.'fromFoldable' [3,2,1]
896-- delay x
897-- @
898-- @
899-- ThreadId 40: Delay 1
900-- ThreadId 39: Delay 2
901-- ThreadId 38: Delay 3
902-- @
903--
904-- As we can see the code after the @fromFoldable@ statement is run three
905-- times, once for each value of @x@. All the three iterations are concurrent
906-- and run in different threads. The iteration with least delay finishes first.
907-- When compared to imperative programming, this can be viewed as a @for@ loop
908-- with three concurrent iterations.
909--
910-- Concurrency is demand driven just as in the case of 'async' merging.
911-- When multiple streams are composed using this style, the iterations are
912-- triggered in a depth first manner just like 'Serial' i.e. nested iterations are
913-- executed before we proceed to the next iteration at higher level. However,
914-- unlike 'Serial' more than one iterations may be started concurrently based
915-- on the demand from the consumer of the stream.
916--
917-- @
918-- import "Streamly"
919-- import qualified "Streamly.Prelude" as S
920--
921-- main = 'runStream' . 'asyncly' $ do
922-- x <- S.'fromFoldable' [1,2]
923-- y <- S.'fromFoldable' [3,4]
924-- S.'once' $ putStrLn $ show (x, y)
925-- @
926-- @
927-- (1,3)
928-- (1,4)
929-- (2,3)
930-- (2,4)
931-- @
933-- $interleavedNesting
934--
935-- The 'Monad' composition of 'WSerial' type interleaves the iterations of
936-- outer and inner loops in a nested loop composition. This works exactly the
937-- same way as the merging of two streams in 'wSerially' fashion works. The
938-- 'wSerially' type combinator can be used to switch to this style of
939-- composition. Alternatively, a type annotation can be used to specify the
940-- type of the stream as 'WSerial'.
941--
942-- @
943-- import "Streamly"
944-- import qualified "Streamly.Prelude" as S
945--
946-- main = 'runStream' . 'wSerially' $ do
947-- x <- S.'fromFoldable' [1,2]
948-- y <- S.'fromFoldable' [3,4]
949-- S.once $ putStrLn $ show (x, y)
950-- @
951-- @
952-- (1,3)
953-- (2,3)
954-- (1,4)
955-- (2,4)
956-- @
957--
959-- $wasyncNesting
960--
961-- Just like 'Async' the 'Monad' composition of 'WAsync' runs the iterations of
962-- a loop concurrently. The difference is in the nested loop behavior. The
963-- nested loops in this type are traversed and executed in a breadth first
964-- manner rather than the depth first manner of 'Async' style.
965-- The loop nesting works exactly the same way as the merging of streams
966-- 'wAsyncly' works. The 'wAsyncly' type combinator can be used to switch to
967-- this style of composition. Alternatively, a type annotation can be used to
968-- specify the type of the stream as 'WAsync'.
969--
970-- @
971-- import "Streamly"
972-- import qualified "Streamly.Prelude" as S
973--
974-- main = 'runStream' . 'wAsyncly' $ do
975-- x <- S.'fromFoldable' [1,2]
976-- y <- S.'fromFoldable' [3,4]
977-- S.'once' $ putStrLn $ show (x, y)
978-- @
979-- @
980-- (1,3)
981-- (2,3)
982-- (1,4)
983-- (2,4)
984-- @
986-- $parallelNesting
987--
988-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the
989-- iterations of a loop concurrently. The difference is in the nested loop
990-- behavior. The streams at each nest level is run fully concurrently
991-- irrespective of the demand. The loop nesting works exactly the same way as
992-- the merging of streams 'parallely' works. The 'parallely' type combinator
993-- can be used to switch to this style of composition. Alternatively, a type
994-- annotation can be used to specify the type of the stream as 'Parallel'.
995--
996-- @
997-- import "Streamly"
998-- import qualified "Streamly.Prelude" as S
999--
1000-- main = 'runStream' . 'parallely' $ do
1001-- x <- S.'fromFoldable' [3,2,1]
1002-- delay x
1003-- @
1004-- @
1005-- ThreadId 40: Delay 1
1006-- ThreadId 39: Delay 2
1007-- ThreadId 38: Delay 3
1008-- @
1010-- $monadExercise
1012-- Streamly code is usually written in a way that is agnostic of the
1013-- specific monadic composition type. We use a polymorphic type with a
1014-- 'IsStream' type class constraint. When running the stream we can choose the
1015-- specific mode of composition. For example take a look at the following code.
1017-- @
1018-- import "Streamly"
1019-- import qualified "Streamly.Prelude" as S
1021-- composed :: (IsStream t, Monad (t IO)) => t IO ()
1022-- composed = do
1023-- sz <- sizes
1024-- cl <- colors
1025-- sh <- shapes
1026-- S.'once' $ putStrLn $ show (sz, cl, sh)
1028-- where
1030-- sizes = S.'fromFoldable' [1, 2, 3]
1031-- colors = S.'fromFoldable' ["red", "green", "blue"]
1032-- shapes = S.'fromFoldable' ["triangle", "square", "circle"]
1033-- @
1035-- Now we can interpret this in whatever way we want:
1037-- @
1038-- main = 'runStream' . 'serially' $ composed
1039-- main = 'runStream' . 'wSerially' $ composed
1040-- main = 'runStream' . 'asyncly' $ composed
1041-- main = 'runStream' . 'wAsyncly' $ composed
1042-- main = 'runStream' . 'parallely' $ composed
1043-- @
1045-- As an exercise try to figure out the output of this code for each mode of
1046-- composition.
1048-- $functor
1050-- 'fmap' transforms a stream by mapping a function on all elements of the
1051-- stream. 'fmap' behaves in the same way for all stream types, it is always
1052-- serial.
1054-- @
1055-- import "Streamly"
1056-- import qualified "Streamly.Prelude" as S
1058-- main = (S.'toList' $ fmap show $ S.'fromFoldable' [1..10]) >>= print
1059-- @
1061-- Also see the 'mapM' and 'sequence' functions for mapping actions, in the
1062-- "Streamly.Prelude" module.
1064-- $applicative
1066-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For
1067-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are
1068-- provided.
1070-- The following example uses the 'Serial' applicative, it runs all iterations
1071-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):
1073-- @
1074-- import "Streamly"
1075-- import qualified "Streamly.Prelude" as S
1076-- import Control.Concurrent
1078-- s1 = d 1 <> d 2
1079-- s2 = d 3 <> d 4
1080-- d n = delay n >> return n
1082-- main = (S.'toList' . 'serially' $ (,) \<$> s1 \<*> s2) >>= print
1083-- @
1084-- @
1085-- ThreadId 36: Delay 1
1086-- ThreadId 36: Delay 3
1087-- ThreadId 36: Delay 4
1088-- ThreadId 36: Delay 2
1089-- ThreadId 36: Delay 3
1090-- ThreadId 36: Delay 4
1091-- [(1,3),(1,4),(2,3),(2,4)]
1092-- @
1094-- Similalrly 'WSerial' applicative runs the iterations in an interleaved
1095-- order but since it is serial it takes a total of 17 seconds:
1097-- @
1098-- main = (S.'toList' . 'wSerially' $ (,) \<$> s1 \<*> s2) >>= print
1099-- @
1100-- @
1101-- ThreadId 36: Delay 1
1102-- ThreadId 36: Delay 3
1103-- ThreadId 36: Delay 2
1104-- ThreadId 36: Delay 3
1105-- ThreadId 36: Delay 4
1106-- ThreadId 36: Delay 4
1107-- [(1,3),(2,3),(1,4),(2,4)]
1108-- @
1110-- 'Async' can run the iterations concurrently and therefore takes a total
1111-- of 10 seconds (1 + 2 + 3 + 4):
1113-- @
1114-- main = (S.'toList' . 'asyncly' $ (,) \<$> s1 \<*> s2) >>= print
1115-- @
1116-- @
1117-- ThreadId 34: Delay 1
1118-- ThreadId 36: Delay 2
1119-- ThreadId 35: Delay 3
1120-- ThreadId 36: Delay 3
1121-- ThreadId 35: Delay 4
1122-- ThreadId 36: Delay 4
1123-- [(1,3),(2,3),(1,4),(2,4)]
1124-- @
1126-- Similalrly 'WAsync' as well can run the iterations concurrently and
1127-- therefore takes a total of 10 seconds (1 + 2 + 3 + 4):
1129-- @
1130-- main = (S.'toList' . 'wAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1131-- @
1132-- @
1133-- ThreadId 34: Delay 1
1134-- ThreadId 36: Delay 2
1135-- ThreadId 35: Delay 3
1136-- ThreadId 36: Delay 3
1137-- ThreadId 35: Delay 4
1138-- ThreadId 36: Delay 4
1139-- [(1,3),(2,3),(1,4),(2,4)]
1140-- @
1142-- $zipping
1144-- Zipping is a special transformation where the corresponding elements of two
1145-- streams are combined together using a zip function producing a new stream of
1146-- outputs. Two different types are provided for serial and concurrent zipping.
1147-- These types provide an applicative instance that can be used to lift
1148-- functions to zip the argument streams.
1149-- Also see the zipping functions in the "Streamly.Prelude" module.
1151-- $serialzip
1153-- The applicative instance of 'ZipSerial' type zips streams serially.
1154-- 'zipSerially' type combinator can be used to switch to serial applicative
1155-- zip composition:
1157-- @
1158-- import "Streamly"
1159-- import qualified "Streamly.Prelude" as S
1160-- import Control.Concurrent
1162-- d n = delay n >> return n
1163-- s1 = 'serially' $ d 1 <> d 2
1164-- s2 = 'serially' $ d 3 <> d 4
1166-- main = (S.'toList' . 'zipSerially' $ (,) \<$> s1 \<*> s2) >>= print
1167-- @
1169-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since
1170-- everything runs serially:
1172-- @
1173-- ThreadId 29: Delay 1
1174-- ThreadId 29: Delay 3
1175-- ThreadId 29: Delay 2
1176-- ThreadId 29: Delay 4
1177-- [(1,3),(2,4)]
1178-- @
1180-- $parallelzip
1182-- The applicative instance of 'ZipAsync' type zips streams concurrently.
1183-- 'zipAsyncly' type combinator can be used to switch to parallel applicative
1184-- zip composition:
1187-- @
1188-- import "Streamly"
1189-- import qualified "Streamly.Prelude" as S
1190-- import Control.Concurrent
1191-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
1193-- d n = delay n >> return n
1194-- s1 = 'serially' $ d 1 <> d 2
1195-- s2 = 'serially' $ d 3 <> d 4
1197-- main = do
1198-- hSetBuffering stdout LineBuffering
1199-- (S.'toList' . 'zipAsyncly' $ (,) \<$> s1 \<*> s2) >>= print
1200-- @
1202-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3
1203-- are produced concurrently, and 2 and 4 are produced concurrently:
1205-- @
1206-- ThreadId 32: Delay 1
1207-- ThreadId 32: Delay 2
1208-- ThreadId 33: Delay 3
1209-- ThreadId 33: Delay 4
1210-- [(1,3),(2,4)]
1211-- @
1213-- $concurrent
1215-- When writing concurrent programs there are two distinct places where the
1216-- programmer can control the concurrency. First, when /composing/ a stream by
1217-- merging multiple streams we can choose an appropriate sum style operators to
1218-- combine them concurrently or serially. Second, when /processing/ a stream in
1219-- a monadic composition we can choose one of the monad composition types to
1220-- choose the desired type of concurrency.
1222-- In the following example the squares of @x@ and @y@ are computed
1223-- concurrently using the 'async' operation and the square roots of their
1224-- sum are computed serially because of the 'streamly' combinator. We can
1225-- choose different combinators for the monadic processing and the stream
1226-- generation, to control the concurrency. We can also use the 'asyncly'
1227-- combinator instead of explicitly folding with 'async'.
1229-- @
1230-- import "Streamly"
1231-- import qualified "Streamly.Prelude" as S
1232-- import Data.List (sum)
1234-- main = do
1235-- z \<- S.'toList'
1236-- $ 'serially' -- Serial monadic processing (sqrt below)
1237-- $ do
1238-- x2 \<- 'forEachWith' 'async' [1..100] $ -- Concurrent @"for"@ loop
1239-- \\x -> return $ x * x -- body of the loop
1240-- y2 \<- 'forEachWith' 'async' [1..100] $
1241-- \\y -> return $ y * y
1242-- return $ sqrt (x2 + y2)
1243-- print $ sum z
1244-- @
1246-- We can see how this directly maps to the imperative style
1247-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators
1248-- and operators instead of the ugly pragmas.
1250-- For more concurrent programming examples see,
1251-- "ListDir.hs", "MergeSort.hs" and "SearchQuery.hs" in the examples directory.
1253-- $reactive
1255-- Reactive programming is nothing but concurrent streaming which is what
1256-- streamly is all about. With streamly we can generate streams of events,
1257-- merge streams that are generated concurrently and process events
1258-- concurrently. We can do all this without any knowledge about the specifics
1259-- of the implementation of concurrency. In the following example you will see
1260-- that the code is just regular Haskell code without much streamly APIs used
1261-- (active hyperlinks are the streamly APIs) and yet it is a reactive
1262-- application.
1264-- This application has two independent and concurrent sources of event
1265-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates
1266-- events that deteriorate the health of the character in the game.
1267-- @userAction@ can be "potion" or "quit". When the user types "potion" the
1268-- health improves and the game continues.
1270-- @
1271-- {-\# LANGUAGE FlexibleContexts #-}
1273-- import "Streamly"
1274-- import Control.Concurrent (threadDelay)
1275-- import Control.Monad (when)
1276-- import Control.Monad.IO.Class (MonadIO(..))
1277-- import Control.Monad.State (MonadState, get, modify, runStateT)
1278-- import Data.Semigroup (cycle1)
1280-- data Event = Harm Int | Heal Int | Quit deriving (Show)
1282-- userAction :: MonadIO m => 'SerialT' m Event
1283-- userAction = cycle1 $ liftIO askUser
1284-- where
1285-- askUser = do
1286-- command <- getLine
1287-- case command of
1288-- "potion" -> return (Heal 10)
1289-- "quit" -> return Quit
1290-- _ -> putStrLn "What?" >> askUser
1292-- acidRain :: MonadIO m => 'SerialT' m Event
1293-- acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
1295-- game :: ('MonadAsync' m, MonadState Int m) => 'SerialT' m ()
1296-- game = do
1297-- event \<- userAction \`parallel` acidRain
1298-- case event of
1299-- Harm n -> modify $ \\h -> h - n
1300-- Heal n -> modify $ \\h -> h + n
1301-- Quit -> fail "quit"
1303-- h <- get
1304-- when (h <= 0) $ fail "You die!"
1305-- liftIO $ putStrLn $ "Health = " ++ show h
1307-- main = do
1308-- putStrLn "Your health is deteriorating due to acid rain,\\
1309-- \\ type \\"potion\\" or \\"quit\\""
1310-- _ <- runStateT ('runStream' game) 60
1311-- return ()
1312-- @
1314-- You can also find the source of this example in the examples directory as
1315-- "AcidRain.hs". It has been adapted from Gabriel's
1316-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>
1317-- package.
1318-- This is much simpler compared to the pipes version because of the builtin
1319-- concurrency in streamly. You can also find a SDL based reactive programming
1320-- example adapted from Yampa in "Streamly.Examples.CirclingSquare".
1322-- $performance
1324-- Streamly is highly optimized for performance, it is designed for serious
1325-- high performing, concurrent and scalable applications. We have created the
1326-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>
1327-- package which is specifically and carefully designed to measure the
1328-- performance of Haskell streaming libraries fairly and squarely in the right
1329-- way. Streamly performs at par or even better than most streaming libraries
1330-- for serial operations even though it needs to deal with the concurrency
1331-- capability.
1333-- $interop
1335-- We can use @unfoldr@ and @uncons@ to convert one streaming type to another.
1337-- Interop with @vector@:
1339-- @
1340-- import Streamly
1341-- import qualified Streamly.Prelude as S
1342-- import qualified Data.Vector.Fusion.Stream.Monadic as V
1344-- -- | vector to streamly
1345-- fromVector :: (IsStream t, Monad m) => V.Stream m a -> t m a
1346-- fromVector = S.unfoldrM unconsV
1347-- where
1348-- unconsV v = do
1349-- r <- V.null v
1350-- if r
1351-- then return Nothing
1352-- else do
1353-- h <- V.head v
1354-- return $ Just (h, V.tail v)
1356-- -- | streamly to vector
1357-- toVector :: Monad m => SerialT m a -> V.Stream m a
1358-- toVector = V.unfoldrM (S.uncons . adapt)
1360-- main = do
1361-- S.toList (fromVector (V.fromList [1..3])) >>= print
1362-- V.toList (toVector (S.fromFoldable [1..3])) >>= print
1363-- @
1365-- Interop with @pipes@:
1367-- @
1368-- import "Streamly"
1369-- import qualified "Streamly.Prelude" as S
1370-- import qualified Pipes as P
1371-- import qualified Pipes.Prelude as P
1373-- -- | pipes to streamly
1374-- fromPipes :: (IsStream t, Monad m) => P.Producer a m r -> t m a
1375-- fromPipes = S.'unfoldrM' unconsP
1376-- where
1377-- -- Adapt P.next to return a Maybe instead of Either
1378-- unconsP p = P.next p >>= either (\\_ -> return Nothing) (return . Just)
1380-- -- | streamly to pipes
1381-- toPipes :: Monad m => SerialT m a -> P.Producer a m ()
1382-- toPipes = P.unfoldr unconsS
1383-- where
1384-- -- Adapt S.uncons to return an Either instead of Maybe
1385-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1387-- main = do
1388-- S.'toList' (fromPipes (P.each [1..3])) >>= print
1389-- P.toListM (toPipes (S.'fromFoldable' [1..3])) >>= print
1390-- @
1392-- Interop with @streaming@:
1394-- @
1395-- import "Streamly"
1396-- import qualified "Streamly.Prelude" as S
1397-- import qualified Streaming as SG
1398-- import qualified Streaming.Prelude as SG
1400-- -- | streaming to streamly
1401-- fromStreaming :: (IsStream t, Monad m) => SG.Stream (SG.Of a) m r -> t m a
1402-- fromStreaming = S.unfoldrM SG.uncons
1404-- -- | streamly to streaming
1405-- toStreaming :: Monad m => SerialT m a -> SG.Stream (SG.Of a) m ()
1406-- toStreaming = SG.unfoldr unconsS
1407-- where
1408-- -- Adapt S.uncons to return an Either instead of Maybe
1409-- unconsS s = S.'uncons' s >>= maybe (return $ Left ()) (return . Right)
1411-- main = do
1412-- S.toList (fromStreaming (SG.each [1..3])) >>= print
1413-- SG.toList (toStreaming (S.fromFoldable [1..3])) >>= print
1414-- @
1416-- Interop with @conduit@:
1418-- @
1419-- import "Streamly"
1420-- import qualified "Streamly.Prelude" as S
1421-- import qualified Data.Conduit as C
1422-- import qualified Data.Conduit.List as C
1423-- import qualified Data.Conduit.Combinators as C
1425-- -- It seems there is no way out of a conduit as it does not provide an
1426-- -- uncons or a tail function. We can convert streamly to conduit though.
1428-- -- | streamly to conduit
1429-- toConduit :: Monad m => SerialT m a -> C.ConduitT i a m ()
1430-- toConduit s = C.unfoldM S.'uncons' s
1432-- main = do
1433-- C.runConduit (toConduit (S.'fromFoldable' [1..3]) C..| C.sinkList) >>= print
1434-- @
1436-- $comparison
1438-- List transformers and logic programming monads also provide a product style
1439-- composition similar to streamly, however streamly generalizes it with the
1440-- time dimension; allowing streams to be composed in an asynchronous and
1441-- concurrent fashion in many different ways. It also provides multiple
1442-- alternative ways of composing streams e.g. serial, interleaved or
1443-- concurrent.
1445-- This seemingly simple addition of asynchronicity and concurrency to product
1446-- style streaming composition unifies a number of disparate abstractions into
1447-- one powerful, concise and elegant abstraction. A wide variety of
1448-- programming problems can be solved elegantly with this abstraction. In
1449-- particular, it unifies three major programming domains namely
1450-- non-deterministic (logic) programming, concurrent programming and functional
1451-- reactive programming. In other words, you can do everything with this one
1452-- abstraction that you could do with the popular libraries listed under these
1453-- categories in the list below.
1455-- @
1456-- +-----------------+----------------+
1457-- | Non-determinism | <https://hackage.haskell.org/package/pipes pipes> |
1458-- | +----------------+
1459-- | | <https://hackage.haskell.org/package/list-t list-t> |
1460-- | +----------------+
1461-- | | <https://hackage.haskell.org/package/logict logict> |
1462-- +-----------------+----------------+
1463-- | Streaming | <https://hackage.haskell.org/package/vector vector> |
1464-- | +----------------+
1465-- | | <https://hackage.haskell.org/package/streaming streaming> |
1466-- | +----------------+
1467-- | | <https://hackage.haskell.org/package/pipes pipes> |
1468-- | +----------------+
1469-- | | <https://hackage.haskell.org/package/conduit conduit> |
1470-- +-----------------+----------------+
1471-- | Concurrency | <https://hackage.haskell.org/package/async async> |
1472-- | +----------------+
1473-- | | <https://hackage.haskell.org/package/transient transient> |
1474-- +-----------------+----------------+
1475-- | FRP | <https://hackage.haskell.org/package/Yampa Yampa> |
1476-- | +----------------+
1477-- | | <https://hackage.haskell.org/package/dunai dunai> |
1478-- | +----------------+
1479-- | | <https://hackage.haskell.org/package/reflex reflex> |
1480-- +-----------------+----------------+
1481-- @
1483-- Streamly is a list-transformer. It provides all the functionality provided
1484-- by any of the list transformer and logic programming packages listed above.
1485-- In addition, Streamly naturally integrates the concurrency dimension to the
1486-- basic list transformer functionality.
1488-- When it comes to streaming, in terms of the streaming API streamly is almost
1489-- identical to the vector package. Streamly, vector and streaming packages all
1490-- represent a stream as data and are therefore similar in the fundamental
1491-- approach to streaming. The fundamental difference is that streamly adds
1492-- concurrency support and the monad instance provides concurrent looping.
1493-- Other streaming libraries like pipes, conduit and machines represent and
1494-- compose stream processors rather than the stream data and therefore fall in
1495-- another class of streaming libraries and have comparatively more complicated
1496-- types.
1498-- When it comes to concurrency, streamly can do everything that the @async@
1499-- package can do and more. async provides applicative concurrency whereas
1500-- streamly provides both applicative and monadic concurrency. The
1501-- 'ZipAsync' type behaves like the applicative instance of async. In
1502-- comparison to transient streamly has a first class streaming interface and
1503-- is a monad transformer that can be used universally in any Haskell monad
1504-- transformer stack. Streamly was in fact originally inspired by the
1505-- concurrency implementation in @transient@ though it has no resemblence with
1506-- that and takes a lazy pull approach versus transient's strict push approach.
1508-- The non-determinism, concurrency and streaming combination make streamly a
1509-- strong FRP capable library as well. FRP is fundamentally stream of events
1510-- that can be processed concurrently. The example in this tutorial as well as
1511-- the "Streamly.Examples.CirclingSquare" example from Yampa demonstrate the
1512-- basic FRP capability of streamly. In core concepts streamly is strikingly
1513-- similar to @dunai@. dunai was designed from a FRP perspective and streamly
1514-- was originally designed from a concurrency perspective. However, both have
1515-- similarity at the core.
1517-- $furtherReading
1519-- * Read the documentation of "Streamly" module
1520-- * Read the documentation of "Streamly.Prelude" module
1521-- * See the examples in the "examples" directory of the package
1522-- * See the tests in the "test" directory of the package