on master help sign in
Merge pull request #261 from composewell/streamly-module-doc

update documentation for different stream types

commita014617f5e7277e9371cdac14f7814c7f9452d92 authorHarendra Kumar <harendra.kumar@gmail.com> dateFri 30 Aug 2019, 12:36 (7 years ago) parent379732dc1ba040
4 files changed+187−88
src/Streamly.hs+112−13 src/Streamly/Streams/Ahead.hs+17−19 src/Streamly/Streams/Async.hs+28−32 src/Streamly/Streams/Serial.hs+30−24
src/Streamly.hs+112−13
@@ -79,13 +79,40 @@ module Streamly
7979 MonadAsync
8080
8181 -- * Stream transformers
82+ -- | A stream represents a sequence of pure or effectful actions. The
83+ -- `cons` and `consM` operations and the corresponding operators '.:' and
84+ -- '|:' can be used to join pure values or effectful actions in a sequence.
85+ -- The effects in the stream can be executed in many different ways
86+ -- depending on the type of stream. In other words, the behavior of 'consM'
87+ -- depends on the type of the stream.
88+ --
89+ -- There are three high level categories of streams, /spatially ordered/
90+ -- streams, /speculative/ streams and /time ordered/ streams. Spatially
91+ -- ordered streams, 'SerialT' and 'WSerialT', execute the effects in serial
92+ -- order i.e. one at a time and present the outputs of those effects to the
93+ -- consumer in the same order. Speculative streams, 'AheadT', may execute
94+ -- many effects concurrently but present the outputs to the consumer in the
95+ -- specified spatial order. Time ordered streams, 'AsyncT', 'WAsyncT' and
96+ -- 'ParallelT', may execute many effects concurrently and present the
97+ -- outputs of those effects to the consumer in time order i.e. as soon as
98+ -- the output is generated.
99+ --
100+ -- We described above how the effects in a sequence are executed for
101+ -- different types of streams. The behvavior of the 'Semigroup' and 'Monad'
102+ -- instances follow the behavior of 'consM'. Stream generation operations
103+ -- like 'repeatM' also execute the effects differently for different
104+ -- streams, providing a concurrent generation capability when used with
105+ -- stream types that execute effects concurrently. Similarly, effectful
106+ -- transformation operations like 'mapM' also execute the transforming
107+ -- effects differently for different types of streams.
108+
82109 -- ** Serial Streams
83110 -- $serial
84111 , SerialT
85112 , WSerialT
86113
87114 -- ** Speculative Streams
88− -- $lookahead
115+ -- $ahead
89116 , AheadT
90117
91118 -- ** Asynchronous Streams
@@ -415,21 +442,93 @@ forEachWith = P.forEachWith
415442
416443-- $serial
417444--
418−-- Serial streams compose serially or non-concurrently. In a composed stream,
419−-- each action is executed only after the previous action has finished. The two
420−-- serial stream types 'SerialT' and 'WSerialT' differ in how they traverse the
421−-- streams in a 'Semigroup' or 'Monad' composition.
445+-- When a stream consumer demands an element from a serial stream constructed
446+-- as @a \`consM` b \`consM` ... nil@, the action @a@ at the head of the stream
447+-- sequence is executed and the result is supplied to the consumer. When the
448+-- next element is demanded, the action @b@ is executed and its result is
449+-- supplied. Thus, the effects are performed and results are consumed strictly
450+-- in a serial order. Serial streams can be considered as /spatially ordered/
451+-- streams as the order of execution and consumption is the same as the spatial
452+-- order in which the actions are composed by the programmer.
453+--
454+-- Serial streams enforce the side effects as well as the results of the
455+-- actions to be in the same order in which the actions are added to the
456+-- stream. Therefore, the semigroup operation for serial streams is not
457+-- commutative:
458+--
459+-- @
460+-- a <> b is not the same as b <> a
461+-- @
462+--
463+-- There are two serial stream types 'SerialT' and 'WSerialT'. The stream
464+-- evaluation of both the variants works in the same way as described above,
465+-- they differ only in the 'Semigroup' and 'Monad' implementaitons.
466+
467+-- $ahead
468+--
469+-- When a stream consumer demands an element from a speculative stream
470+-- constructed as @a \`consM` b \`consM` ... nil@, the action @a@ at the head
471+-- of the stream is executed and the output of the action is supplied to the
472+-- consumer. However, in addition to the action at the head multiple actions
473+-- following it may also be executed concurrently and the results buffered.
474+-- When the next element is demanded it may be served from the buffer and we
475+-- may execute the next action in the sequence to keep the buffer adequately
476+-- filled. Thus, the actions are executed concurrently but results consumed in
477+-- serial order just like serial streams. `consM` can be used to fold an
478+-- infinite lazy container of effects, as the number of concurrent executions
479+-- is limited.
480+--
481+-- Similar to 'consM', the monadic stream generation (e.g. replicateM) and
482+-- transformation operations (e.g. mapM) on speculative streams can execute
483+-- multiple effects concurrently in a speculative manner.
484+--
485+-- How many effects can be executed concurrently and how many results can be
486+-- buffered are controlled by 'maxThreads' and 'maxBuffer' combinators
487+-- respectively. The actual number of concurrent threads is adjusted according
488+-- to the rate at which the consumer is consuming the stream. It may even
489+-- execute actions serially in a single thread if that is enough to match the
490+-- consumer's speed.
491+--
492+-- Speculative streams enforce ordering of the results of actions in the stream
493+-- but the side effects are only partially ordered. Therefore, the semigroup
494+-- operation for speculative streams is not commutative from the pure outputs
495+-- perspective but commutative from side effects perspective.
422496
423497-- $async
424498--
425−-- The async style streams execute actions asynchronously and consume the
426−-- outputs as well asynchronously. In a composed stream, at any point of time
427−-- more than one stream can run concurrently and yield elements. The elements
428−-- are yielded by the composed stream as they are generated by the constituent
429−-- streams on a first come first serve basis. Therefore, on each run the
430−-- stream may yield elements in a different sequence depending on the delays
431−-- introduced by scheduling. The two async types 'AsyncT' and 'WAsyncT' differ
432−-- in how they traverse streams in 'Semigroup' or 'Monad' compositions.
499+-- When a stream consumer demands an element from an asynchronous stream,
500+-- constructed as @a \`consM` b \`consM` ... nil@, the action @a@ along with
501+-- multiple following at the head of the stream sequence are executed
502+-- concurrently and the output of the one that completes first is supplied to
503+-- the consumer. As more actions complete, their results are buffered in the
504+-- order of completion. When the next element is demanded it may be served
505+-- from the buffer and we may initiate execution of more actions in the
506+-- sequence to keep the buffer adequately filled. Thus, the actions are
507+-- executed concurrently and their results are consumed in the order of
508+-- completion. `consM` can be used to fold an infinite lazy container of
509+-- effects, as the number of concurrent executions is limited.
510+--
511+-- Similar to 'consM', the monadic stream generation (e.g. replicateM) and
512+-- transformation operations (e.g. mapM) on asynchronous streams can execute
513+-- multiple effects concurrently in an asynchronous manner.
514+--
515+-- How many effects can be executed concurrently and how many results can be
516+-- buffered are controlled by 'maxThreads' and 'maxBuffer' combinators
517+-- respectively. The actual number of concurrent threads is adjusted according
518+-- to the rate at which the consumer is consuming the stream. It may even
519+-- execute actions serially in a single thread if that is enough to match the
520+-- consumer's speed.
521+--
522+-- Asynchronous streams do not enforce any spatial order on the side effects or
523+-- on the results of the actions. However there is a partial ordering as the
524+-- actions to be executed are picked from the head of stream. The results are
525+-- presented to the consumer in the completion time order. Therefore, the
526+-- semigroup operation for asynchronous streams is commutative i.e. the stream
527+-- is considered unordered.
528+--
529+-- There are two asynchronous stream types 'AsyncT' and 'WAsyncT'. The stream
530+-- evaluation of both the variants works in the same way as described above,
531+-- they differ only in the 'Semigroup' and 'Monad' implementaitons.
433532
434533-- $zipping
435534--
src/Streamly/Streams/Ahead.hs+17−19
@@ -590,14 +590,21 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)
590590-- AheadT
591591------------------------------------------------------------------------------
592592
593−-- | Deep ahead composition or ahead composition with depth first traversal.
594−-- The semigroup composition of 'AheadT' appends streams in a depth first
595−-- manner just like 'SerialT' except that it can produce elements concurrently
596−-- ahead of time. It is like 'AsyncT' except that 'AsyncT' produces the output
597−-- as it arrives whereas 'AheadT' orders the output in the traversal order.
593+-- | The 'Semigroup' operation for 'AheadT' appends two streams. The combined
594+-- stream behaves like a single stream with the actions from the second stream
595+-- appended to the first stream. The combined stream is evaluated in the
596+-- speculative style. This operation can be used to fold an infinite lazy
597+-- container of streams.
598598--
599599-- @
600−-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
600+-- import "Streamly"
601+-- import qualified "Streamly.Prelude" as S
602+-- import Control.Concurrent
603+--
604+-- main = do
605+-- xs \<- S.'toList' . 'aheadly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
606+-- print xs
607+-- where p n = threadDelay 1000000 >> return n
601608-- @
602609-- @
603610-- [1,2,3,4]
@@ -606,16 +613,13 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)
606613-- Any exceptions generated by a constituent stream are propagated to the
607614-- output stream.
608615--
609−-- Similarly, the monad instance of 'AheadT' may run each iteration
610−-- concurrently ahead of time but presents the results in the same order as
616+-- The monad instance of 'AheadT' may run each monadic continuation (bind)
617+-- concurrently in a speculative manner, performing side effects in a partially
618+-- ordered manner but producing the outputs in an ordered manner like
611619-- 'SerialT'.
612620--
613621-- @
614−-- import "Streamly"
615−-- import qualified "Streamly.Prelude" as S
616−-- import Control.Concurrent
617−--
618−-- main = 'drain' . 'aheadly' $ do
622+-- main = S.drain . 'aheadly' $ do
619623-- n <- return 3 \<\> return 2 \<\> return 1
620624-- S.yieldM $ do
621625-- threadDelay (n * 1000000)
@@ -627,12 +631,6 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)
627631-- ThreadId 38: Delay 3
628632-- @
629633--
630−-- All iterations may run in the same thread if they do not block.
631−--
632−-- Note that ahead composition with depth first traversal can be used to
633−-- combine infinite number of streams as it explores only a bounded number of
634−-- streams at a time.
635−--
636634-- @since 0.3.0
637635newtype AheadT m a = AheadT {getAheadT :: Stream m a}
638636 deriving (MonadTrans)
src/Streamly/Streams/Async.hs+28−32
@@ -600,14 +600,18 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
600600-- AsyncT
601601------------------------------------------------------------------------------
602602
603−-- | Deep async composition or async composition with depth first traversal. In
604−-- a left to right 'Semigroup' composition it tries to yield elements from the
605−-- left stream as long as it can, but it can run the right stream in parallel
606−-- if it needs to, based on demand. The right stream can be run if the left
607−-- stream blocks on IO or cannot produce elements fast enough for the consumer.
603+-- | The 'Semigroup' operation for 'AsyncT' appends two streams. The combined
604+-- stream behaves like a single stream with the actions from the second stream
605+-- appended to the first stream. The combined stream is evaluated in the
606+-- asynchronous style. This operation can be used to fold an infinite lazy
607+-- container of streams.
608608--
609609-- @
610−-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
610+-- import "Streamly"
611+-- import qualified "Streamly.Prelude" as S
612+-- import Control.Concurrent
613+--
614+-- main = (S.toList . 'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
611615-- @
612616-- @
613617-- [1,2,3,4]
@@ -626,10 +630,6 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
626630-- consumer.
627631--
628632-- @
629−-- import "Streamly"
630−-- import qualified "Streamly.Prelude" as S
631−-- import Control.Concurrent
632−--
633633-- main = 'drain' . 'asyncly' $ do
634634-- n <- return 3 \<\> return 2 \<\> return 1
635635-- S.yieldM $ do
@@ -642,12 +642,6 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
642642-- ThreadId 38: Delay 3
643643-- @
644644--
645−-- All iterations may run in the same thread if they do not block.
646−--
647−-- Note that async composition with depth first traversal can be used to
648−-- combine infinite number of streams as it explores only a bounded number of
649−-- streams at a time.
650−--
651645-- @since 0.1.0
652646newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
653647 deriving (MonadTrans)
@@ -743,13 +737,26 @@ consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
743737wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
744738wAsync = joinStreamVarAsync WAsyncVar
745739
746−-- | Wide async composition or async composition with breadth first traversal.
747−-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
748−-- streams using a depth first travesal or in a round robin fashion, yielding
749−-- elements from both streams alternately.
740+-- | The 'Semigroup' operation for 'WAsyncT' interleaves the elements from the
741+-- two streams. Therefore, when @a <> b@ is evaluated, one action is picked
742+-- from stream @a@ for evaluation and then the next action is picked from
743+-- stream @b@ and then the next action is again picked from stream @a@, going
744+-- around in a round-robin fashion. Many such actions are executed concurrently
745+-- depending on 'maxThreads' and 'maxBuffer' limits. Results are served to the
746+-- consumer in the order completion of the actions.
747+--
748+-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we
749+-- go in a round-robin fasion across all of them picking one action from each
750+-- up to @z@ and then come back to @a@. Note that this operation cannot be
751+-- used to fold a container of infinite streams as the state that it needs to
752+-- maintain is proportional to the number of streams.
750753--
751754-- @
752−-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
755+-- import "Streamly"
756+-- import qualified "Streamly.Prelude" as S
757+-- import Control.Concurrent
758+--
759+-- main = (S.toList . 'wAsyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
753760-- @
754761-- @
755762-- [1,3,2,4]
@@ -766,10 +773,6 @@ wAsync = joinStreamVarAsync WAsyncVar
766773-- concurrently using a round robin scheduling.
767774--
768775-- @
769−-- import "Streamly"
770−-- import qualified "Streamly.Prelude" as S
771−-- import Control.Concurrent
772−--
773776-- main = 'drain' . 'wAsyncly' $ do
774777-- n <- return 3 \<\> return 2 \<\> return 1
775778-- S.yieldM $ do
@@ -782,13 +785,6 @@ wAsync = joinStreamVarAsync WAsyncVar
782785-- ThreadId 38: Delay 3
783786-- @
784787--
785−-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
786−-- concurrently, unconditionally.
787−--
788−-- Note that async composition with breadth first traversal can only combine a
789−-- finite number of streams as it needs to retain state for each unfinished
790−-- stream.
791−--
792788-- @since 0.2.0
793789newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
794790 deriving (MonadTrans)
src/Streamly/Streams/Serial.hs+30−24
@@ -80,16 +80,17 @@ import qualified Streamly.Streams.StreamD as D
8080-- SerialT
8181------------------------------------------------------------------------------
8282
83−-- | Deep serial composition or serial composition with depth first traversal.
84−-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
85−-- depth first manner, yielding all elements from the first stream, and then
86−-- all elements from the second stream.
83+-- | The 'Semigroup' operation for 'SerialT' behaves like a regular append
84+-- operation. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
85+-- first until it exhausts and then stream @b@ is evaluated. In other words,
86+-- the elements of stream @b@ are appended to the elements of stream @a@. This
87+-- operation can be used to fold an infinite lazy container of streams.
8788--
8889-- @
8990-- import Streamly
9091-- import qualified "Streamly.Prelude" as S
9192--
92−-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
93+-- main = (S.toList . 'serially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
9394-- @
9495-- @
9596-- [1,2,3,4]
@@ -99,7 +100,7 @@ import qualified Streamly.Streams.StreamD as D
99100-- element of the stream, serially.
100101--
101102-- @
102−-- main = 'drain' . 'serially' $ do
103+-- main = S.drain . 'serially' $ do
103104-- x <- return 1 \<\> return 2
104105-- S.yieldM $ print x
105106-- @
@@ -111,7 +112,7 @@ import qualified Streamly.Streams.StreamD as D
111112-- 'SerialT' nests streams serially in a depth first manner.
112113--
113114-- @
114−-- main = 'drain' . 'serially' $ do
115+-- main = S.drain . 'serially' $ do
115116-- x <- return 1 \<\> return 2
116117-- y <- return 3 \<\> return 4
117118-- S.yieldM $ print (x, y)
@@ -123,17 +124,17 @@ import qualified Streamly.Streams.StreamD as D
123124-- (2,4)
124125-- @
125126--
126−-- This behavior of 'SerialT' is exactly like a list transformer. We call the
127−-- monadic code being run for each element of the stream a monadic
127+-- We call the monadic code being run for each element of the stream a monadic
128128-- continuation. In imperative paradigm we can think of this composition as
129129-- nested @for@ loops and the monadic continuation is the body of the loop. The
130130-- loop iterates for all elements of the stream.
131131--
132−-- The 'serially' combinator can be omitted as the default stream type is
133−-- 'SerialT'.
134−-- Note that serial composition with depth first traversal can be used to
135−-- combine an infinite number of streams as it explores only one stream at a
136−-- time.
132+-- Note that the behavior and semantics of 'SerialT', including 'Semigroup'
133+-- and 'Monad' instances are exactly like Haskell lists except that 'SerialT'
134+-- can contain effectful actions while lists are pure.
135+--
136+-- In the code above, the 'serially' combinator can be omitted as the default
137+-- stream type is 'SerialT'.
137138--
138139-- @since 0.2.0
139140newtype SerialT m a = SerialT {getSerialT :: Stream m a}
@@ -216,16 +217,25 @@ TRAVERSABLE_INSTANCE(SerialT)
216217-- WSerialT
217218------------------------------------------------------------------------------
218219
219−-- | Wide serial composition or serial composition with a breadth first
220−-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
221−-- the two streams in a breadth first manner. In other words, it interleaves
222−-- two streams, yielding one element from each stream alternately.
220+-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
221+-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
222+-- first to produce the first element of the combined stream and then stream
223+-- @b@ is evaluated to produce the next element of the combined stream, and
224+-- then we go back to evaluating stream @a@ and so on. In other words, the
225+-- elements of stream @a@ are interleaved with the elements of stream @b@.
226+--
227+-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we
228+-- interleave them in a binary fashion i.e. @a@ and @b@ are interleaved with
229+-- each other and the result is interleaved with @c@ and so on. This will not
230+-- act as a true round-robin scheduling across all the streams. Note that this
231+-- operation cannot be used to fold a container of infinite streams as the
232+-- state that it needs to maintain is proportional to the number of streams.
223233--
224234-- @
225235-- import Streamly
226236-- import qualified "Streamly.Prelude" as S
227237--
228−-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
238+-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
229239-- @
230240-- @
231241-- [1,3,2,4]
@@ -236,7 +246,7 @@ TRAVERSABLE_INSTANCE(SerialT)
236246--
237247--
238248-- @
239−-- main = 'drain' . 'wSerially' $ do
249+-- main = S.drain . 'wSerially' $ do
240250-- x <- return 1 \<\> return 2
241251-- y <- return 3 \<\> return 4
242252-- S.yieldM $ print (x, y)
@@ -248,10 +258,6 @@ TRAVERSABLE_INSTANCE(SerialT)
248258-- (2,4)
249259-- @
250260--
251−-- Note that a serial composition with breadth first traversal can only combine
252−-- a finite number of streams as it needs to retain state for each unfinished
253−-- stream.
254−--
255261-- @since 0.2.0
256262newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
257263 deriving (MonadTrans)