on master help sign in
update documents for different stream types

Have a more detailed and clear document. Remove the depth first/breadth first terminology as it may be confusing to many.

commitc1ba0404e39bce0e0e045bbada396f83030c7f3d authorHarendra Kumar <harendra.kumar@gmail.com> dateFri 16 Aug 2019, 14:00 (7 years ago) parentcb6e9f2
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
@@ -588,14 +588,21 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)
588588-- AheadT
589589------------------------------------------------------------------------------
590590
591−-- | Deep ahead composition or ahead composition with depth first traversal.
592−-- The semigroup composition of 'AheadT' appends streams in a depth first
593−-- manner just like 'SerialT' except that it can produce elements concurrently
594−-- ahead of time. It is like 'AsyncT' except that 'AsyncT' produces the output
595−-- as it arrives whereas 'AheadT' orders the output in the traversal order.
591+-- | The 'Semigroup' operation for 'AheadT' appends two streams. The combined
592+-- stream behaves like a single stream with the actions from the second stream
593+-- appended to the first stream. The combined stream is evaluated in the
594+-- speculative style. This operation can be used to fold an infinite lazy
595+-- container of streams.
596596--
597597-- @
598−-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
598+-- import "Streamly"
599+-- import qualified "Streamly.Prelude" as S
600+-- import Control.Concurrent
601+--
602+-- main = do
603+-- xs \<- S.'toList' . 'aheadly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
604+-- print xs
605+-- where p n = threadDelay 1000000 >> return n
599606-- @
600607-- @
601608-- [1,2,3,4]
@@ -604,16 +611,13 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)
604611-- Any exceptions generated by a constituent stream are propagated to the
605612-- output stream.
606613--
607−-- Similarly, the monad instance of 'AheadT' may run each iteration
608−-- concurrently ahead of time but presents the results in the same order as
614+-- The monad instance of 'AheadT' may run each monadic continuation (bind)
615+-- concurrently in a speculative manner, performing side effects in a partially
616+-- ordered manner but producing the outputs in an ordered manner like
609617-- 'SerialT'.
610618--
611619-- @
612−-- import "Streamly"
613−-- import qualified "Streamly.Prelude" as S
614−-- import Control.Concurrent
615−--
616−-- main = 'drain' . 'aheadly' $ do
620+-- main = S.drain . 'aheadly' $ do
617621-- n <- return 3 \<\> return 2 \<\> return 1
618622-- S.yieldM $ do
619623-- threadDelay (n * 1000000)
@@ -625,12 +629,6 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)
625629-- ThreadId 38: Delay 3
626630-- @
627631--
628−-- All iterations may run in the same thread if they do not block.
629−--
630−-- Note that ahead composition with depth first traversal can be used to
631−-- combine infinite number of streams as it explores only a bounded number of
632−-- streams at a time.
633−--
634632-- @since 0.3.0
635633newtype AheadT m a = AheadT {getAheadT :: Stream m a}
636634 deriving (MonadTrans)
src/Streamly/Streams/Async.hs+28−32
@@ -598,14 +598,18 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
598598-- AsyncT
599599------------------------------------------------------------------------------
600600
601−-- | Deep async composition or async composition with depth first traversal. In
602−-- a left to right 'Semigroup' composition it tries to yield elements from the
603−-- left stream as long as it can, but it can run the right stream in parallel
604−-- if it needs to, based on demand. The right stream can be run if the left
605−-- stream blocks on IO or cannot produce elements fast enough for the consumer.
601+-- | The 'Semigroup' operation for 'AsyncT' appends two streams. The combined
602+-- stream behaves like a single stream with the actions from the second stream
603+-- appended to the first stream. The combined stream is evaluated in the
604+-- asynchronous style. This operation can be used to fold an infinite lazy
605+-- container of streams.
606606--
607607-- @
608−-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
608+-- import "Streamly"
609+-- import qualified "Streamly.Prelude" as S
610+-- import Control.Concurrent
611+--
612+-- main = (S.toList . 'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
609613-- @
610614-- @
611615-- [1,2,3,4]
@@ -624,10 +628,6 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
624628-- consumer.
625629--
626630-- @
627−-- import "Streamly"
628−-- import qualified "Streamly.Prelude" as S
629−-- import Control.Concurrent
630−--
631631-- main = 'drain' . 'asyncly' $ do
632632-- n <- return 3 \<\> return 2 \<\> return 1
633633-- S.yieldM $ do
@@ -640,12 +640,6 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
640640-- ThreadId 38: Delay 3
641641-- @
642642--
643−-- All iterations may run in the same thread if they do not block.
644−--
645−-- Note that async composition with depth first traversal can be used to
646−-- combine infinite number of streams as it explores only a bounded number of
647−-- streams at a time.
648−--
649643-- @since 0.1.0
650644newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
651645 deriving (MonadTrans)
@@ -741,13 +735,26 @@ consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
741735wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
742736wAsync = joinStreamVarAsync WAsyncVar
743737
744−-- | Wide async composition or async composition with breadth first traversal.
745−-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
746−-- streams using a depth first travesal or in a round robin fashion, yielding
747−-- elements from both streams alternately.
738+-- | The 'Semigroup' operation for 'WAsyncT' interleaves the elements from the
739+-- two streams. Therefore, when @a <> b@ is evaluated, one action is picked
740+-- from stream @a@ for evaluation and then the next action is picked from
741+-- stream @b@ and then the next action is again picked from stream @a@, going
742+-- around in a round-robin fashion. Many such actions are executed concurrently
743+-- depending on 'maxThreads' and 'maxBuffer' limits. Results are served to the
744+-- consumer in the order completion of the actions.
745+--
746+-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we
747+-- go in a round-robin fasion across all of them picking one action from each
748+-- up to @z@ and then come back to @a@. Note that this operation cannot be
749+-- used to fold a container of infinite streams as the state that it needs to
750+-- maintain is proportional to the number of streams.
748751--
749752-- @
750−-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
753+-- import "Streamly"
754+-- import qualified "Streamly.Prelude" as S
755+-- import Control.Concurrent
756+--
757+-- main = (S.toList . 'wAsyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
751758-- @
752759-- @
753760-- [1,3,2,4]
@@ -764,10 +771,6 @@ wAsync = joinStreamVarAsync WAsyncVar
764771-- concurrently using a round robin scheduling.
765772--
766773-- @
767−-- import "Streamly"
768−-- import qualified "Streamly.Prelude" as S
769−-- import Control.Concurrent
770−--
771774-- main = 'drain' . 'wAsyncly' $ do
772775-- n <- return 3 \<\> return 2 \<\> return 1
773776-- S.yieldM $ do
@@ -780,13 +783,6 @@ wAsync = joinStreamVarAsync WAsyncVar
780783-- ThreadId 38: Delay 3
781784-- @
782785--
783−-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
784−-- concurrently, unconditionally.
785−--
786−-- Note that async composition with breadth first traversal can only combine a
787−-- finite number of streams as it needs to retain state for each unfinished
788−-- stream.
789−--
790786-- @since 0.2.0
791787newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
792788 deriving (MonadTrans)
src/Streamly/Streams/Serial.hs+30−24
@@ -78,16 +78,17 @@ import qualified Streamly.Streams.StreamD as D
7878-- SerialT
7979------------------------------------------------------------------------------
8080
81−-- | Deep serial composition or serial composition with depth first traversal.
82−-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
83−-- depth first manner, yielding all elements from the first stream, and then
84−-- all elements from the second stream.
81+-- | The 'Semigroup' operation for 'SerialT' behaves like a regular append
82+-- operation. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
83+-- first until it exhausts and then stream @b@ is evaluated. In other words,
84+-- the elements of stream @b@ are appended to the elements of stream @a@. This
85+-- operation can be used to fold an infinite lazy container of streams.
8586--
8687-- @
8788-- import Streamly
8889-- import qualified "Streamly.Prelude" as S
8990--
90−-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
91+-- main = (S.toList . 'serially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
9192-- @
9293-- @
9394-- [1,2,3,4]
@@ -97,7 +98,7 @@ import qualified Streamly.Streams.StreamD as D
9798-- element of the stream, serially.
9899--
99100-- @
100−-- main = 'drain' . 'serially' $ do
101+-- main = S.drain . 'serially' $ do
101102-- x <- return 1 \<\> return 2
102103-- S.yieldM $ print x
103104-- @
@@ -109,7 +110,7 @@ import qualified Streamly.Streams.StreamD as D
109110-- 'SerialT' nests streams serially in a depth first manner.
110111--
111112-- @
112−-- main = 'drain' . 'serially' $ do
113+-- main = S.drain . 'serially' $ do
113114-- x <- return 1 \<\> return 2
114115-- y <- return 3 \<\> return 4
115116-- S.yieldM $ print (x, y)
@@ -121,17 +122,17 @@ import qualified Streamly.Streams.StreamD as D
121122-- (2,4)
122123-- @
123124--
124−-- This behavior of 'SerialT' is exactly like a list transformer. We call the
125−-- monadic code being run for each element of the stream a monadic
125+-- We call the monadic code being run for each element of the stream a monadic
126126-- continuation. In imperative paradigm we can think of this composition as
127127-- nested @for@ loops and the monadic continuation is the body of the loop. The
128128-- loop iterates for all elements of the stream.
129129--
130−-- The 'serially' combinator can be omitted as the default stream type is
131−-- 'SerialT'.
132−-- Note that serial composition with depth first traversal can be used to
133−-- combine an infinite number of streams as it explores only one stream at a
134−-- time.
130+-- Note that the behavior and semantics of 'SerialT', including 'Semigroup'
131+-- and 'Monad' instances are exactly like Haskell lists except that 'SerialT'
132+-- can contain effectful actions while lists are pure.
133+--
134+-- In the code above, the 'serially' combinator can be omitted as the default
135+-- stream type is 'SerialT'.
135136--
136137-- @since 0.2.0
137138newtype SerialT m a = SerialT {getSerialT :: Stream m a}
@@ -214,16 +215,25 @@ TRAVERSABLE_INSTANCE(SerialT)
214215-- WSerialT
215216------------------------------------------------------------------------------
216217
217−-- | Wide serial composition or serial composition with a breadth first
218−-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
219−-- the two streams in a breadth first manner. In other words, it interleaves
220−-- two streams, yielding one element from each stream alternately.
218+-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
219+-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
220+-- first to produce the first element of the combined stream and then stream
221+-- @b@ is evaluated to produce the next element of the combined stream, and
222+-- then we go back to evaluating stream @a@ and so on. In other words, the
223+-- elements of stream @a@ are interleaved with the elements of stream @b@.
224+--
225+-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we
226+-- interleave them in a binary fashion i.e. @a@ and @b@ are interleaved with
227+-- each other and the result is interleaved with @c@ and so on. This will not
228+-- act as a true round-robin scheduling across all the streams. Note that this
229+-- operation cannot be used to fold a container of infinite streams as the
230+-- state that it needs to maintain is proportional to the number of streams.
221231--
222232-- @
223233-- import Streamly
224234-- import qualified "Streamly.Prelude" as S
225235--
226−-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
236+-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
227237-- @
228238-- @
229239-- [1,3,2,4]
@@ -234,7 +244,7 @@ TRAVERSABLE_INSTANCE(SerialT)
234244--
235245--
236246-- @
237−-- main = 'drain' . 'wSerially' $ do
247+-- main = S.drain . 'wSerially' $ do
238248-- x <- return 1 \<\> return 2
239249-- y <- return 3 \<\> return 4
240250-- S.yieldM $ print (x, y)
@@ -246,10 +256,6 @@ TRAVERSABLE_INSTANCE(SerialT)
246256-- (2,4)
247257-- @
248258--
249−-- Note that a serial composition with breadth first traversal can only combine
250−-- a finite number of streams as it needs to retain state for each unfinished
251−-- stream.
252−--
253259-- @since 0.2.0
254260newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
255261 deriving (MonadTrans)