a6f394c · streamly · git.iko.soy

git.iko.soy/streamly

Update doc for async streams

commita6f394c49a5163a33061e3f14e2286b57feea17f authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 16 Jan 2020, 23:34 (7 years ago) parent3f5c79e

3 files changed+122−57

compare with master

unifiedsplit

src/Streamly.hs+37−33 src/Streamly/Internal/Data/Stream/Async.hs+71−18 src/Streamly/Internal/Data/Stream/Serial.hs+14−6

src/Streamly.hs+37−33

@@ -511,39 +511,43 @@ forEachWith = P.forEachWith

511511

512512-- $async

513513--

514−-- When a stream consumer demands an element from an asynchronous stream,

515−-- constructed as @a \`consM` b \`consM` ... nil@, the action @a@ along with

516−-- multiple following at the head of the stream sequence are executed

517−-- concurrently and the output of the one that completes first is supplied to

518−-- the consumer. As more actions complete, their results are buffered in the

519−-- order of completion. When the next element is demanded it may be served

520−-- from the buffer and we may initiate execution of more actions in the

521−-- sequence to keep the buffer adequately filled. Thus, the actions are

522−-- executed concurrently and their results are consumed in the order of

523−-- completion. `consM` can be used to fold an infinite lazy container of

524−-- effects, as the number of concurrent executions is limited.

525−--

526−-- Similar to 'consM', the monadic stream generation (e.g. replicateM) and

527−-- transformation operations (e.g. mapM) on asynchronous streams can execute

528−-- multiple effects concurrently in an asynchronous manner.

529−--

530−-- How many effects can be executed concurrently and how many results can be

531−-- buffered are controlled by 'maxThreads' and 'maxBuffer' combinators

532−-- respectively. The actual number of concurrent threads is adjusted according

533−-- to the rate at which the consumer is consuming the stream. It may even

534−-- execute actions serially in a single thread if that is enough to match the

535−-- consumer's speed.

536−--

537−-- Asynchronous streams do not enforce any spatial order on the side effects or

538−-- on the results of the actions. However there is a partial ordering as the

539−-- actions to be executed are picked from the head of stream. The results are

540−-- presented to the consumer in the completion time order. Therefore, the

541−-- semigroup operation for asynchronous streams is commutative i.e. the stream

542−-- is considered unordered.

543−--

544−-- There are two asynchronous stream types 'AsyncT' and 'WAsyncT'. The stream

545−-- evaluation of both the variants works in the same way as described above,

546−-- they differ only in the 'Semigroup' and 'Monad' implementaitons.

514+-- /Scheduling and execution:/ In an asynchronous stream @a \`consM` b \`consM`

515+-- c ...@, the actions @a@, @b@, and @c@ are executed concurrently with the

516+-- consumer of the stream. The actions are /scheduled/ for execution in the

517+-- same order as they are specified in the stream. Multiple scheduled actions

518+-- may be /executed/ concurrently in parallel threads of execution. The

519+-- actions may be executed out of order and they may complete at arbitrary

520+-- times. Therefore, the /effects/ of the actions may be observed out of

521+-- order.

522+--

523+-- /Buffering:/ The /results/ from multiple threads of execution are queued in

524+-- a buffer as soon as they become available. The consumer of the stream is

525+-- served from this buffer. Therefore, the consumer may observe the results to

526+-- be out of order. In other words, an asynchronous stream is an unordered

527+-- stream i.e. order does not matter.

528+--

529+-- /Concurrency control:/ Threads are suspended if the `maxBuffer` limit is

530+-- reached, and resumed when the consumer makes space in the buffer. The

531+-- maximum number of concurrent threads depends on `maxThreads`. Number of

532+-- threads is increased or decreased based on the speed of the consumer.

533+--

534+-- /Generation operations:/ Concurrent stream generation operations e.g.

535+-- 'Streamly.Prelude.replicateM' when used in async style schedule and execute

536+-- the stream generating actions in the manner described above. The generation

537+-- actions run concurrently, effects and results of the actions as observed by

538+-- the consumer of the stream may be out of order.

539+--

540+-- /Transformation operations:/ Concurrent stream transformation operations

541+-- e.g. 'Streamly.Prelude.mapM', when used in async style, schedule and

542+-- execute transformation actions in the manner described above. Transformation

543+-- actions run concurrently, effects and results of the actions may be

544+-- observed by the consumer out of order.

545+--

546+-- /Variants:/ There are two asynchronous stream types 'AsyncT' and 'WAsyncT'.

547+-- They are identical with respect to single stream evaluation behavior. Their

548+-- behaviors differ in how they combine multiple streams using 'Semigroup' or

549+-- 'Monad' composition. Since the order of elements does not matter in

550+-- asynchronous streams the 'Semigroup' operation is effectively commutative.

547551

548552-- $zipping

549553--

src/Streamly/Internal/Data/Stream/Async.hs+71−18

@@ -642,11 +642,16 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)

642642-- AsyncT

643643------------------------------------------------------------------------------

644644

645−-- | The 'Semigroup' operation for 'AsyncT' appends two streams. The combined

646−-- stream behaves like a single stream with the actions from the second stream

647−-- appended to the first stream. The combined stream is evaluated in the

648−-- asynchronous style. This operation can be used to fold an infinite lazy

649−-- container of streams.

645+-- | The 'Semigroup' operation (@<>@) for 'AsyncT' merges two streams

646+-- concurrently with priority given to the first stream. In @s1 <> s2 <> s3

647+-- ...@ the streams s1, s2 and s3 are scheduled for execution in that order.

648+-- Multiple scheduled streams may be executed concurrently and the elements

649+-- generated by them are served to the consumer as and when they become

650+-- available. This behavior is similar to the scheduling and execution behavior

651+-- of actions in a single async stream.

652+--

653+-- Since only a finite number of streams are executed concurrently, this

654+-- operation can be used to fold an infinite lazy container of streams.

650655--

651656-- @

652657-- import "Streamly"

@@ -787,31 +792,79 @@ consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)

787792wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

788793wAsync = joinStreamVarAsync WAsyncVar

789794

790−-- | The 'Semigroup' operation for 'WAsyncT' interleaves the elements from the

791−-- two streams. Therefore, when @a <> b@ is evaluated, one action is picked

792−-- from stream @a@ for evaluation and then the next action is picked from

793−-- stream @b@ and then the next action is again picked from stream @a@, going

794−-- around in a round-robin fashion. Many such actions are executed concurrently

795−-- depending on 'maxThreads' and 'maxBuffer' limits. Results are served to the

796−-- consumer in the order completion of the actions.

795+-- | 'WAsyncT' is similar to 'WSerialT' but with concurrent execution.

796+-- The 'Semigroup' operation (@<>@) for 'WAsyncT' merges two streams

797+-- concurrently interleaving the actions from both the streams. In @s1

798+-- <> s2 <> s3 ...@, the individual actions from streams @s1@, @s2@ and @s3@

799+-- are scheduled for execution in a round-robin fashion. Multiple scheduled

800+-- actions may be executed concurrently, the results from concurrent executions

801+-- are consumed in the order in which they become available.

802+--

797803--

798−-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we

799−-- go in a round-robin fasion across all of them picking one action from each

800−-- up to @z@ and then come back to @a@. Note that this operation cannot be

801−-- used to fold a container of infinite streams as the state that it needs to

802−-- maintain is proportional to the number of streams.

804+-- The @W@ in the name stands for @wide@ or breadth wise scheduling in

805+-- contrast to the depth wise scheduling behavior of 'AsyncT'.

803806--

804807-- @

805808-- import "Streamly"

806809-- import qualified "Streamly.Prelude" as S

807810-- import Control.Concurrent

808811--

809−-- main = (S.toList . 'wAsyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print

812+-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print

810813-- @

811814-- @

812815-- [1,3,2,4]

813816-- @

814817--

818+-- For this example, we are using @maxThreads 1@ so that concurrent thread

819+-- scheduling does not affect the results and make them unpredictable. Let's

820+-- now take a more general example:

821+--

822+-- @

823+-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print

824+-- @

825+-- @

826+-- [1,4,2,7,5,3,8,6,9]

827+-- @

828+--

829+-- This is how the execution of the above stream proceeds:

830+--

831+-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],

832+-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the

833+-- queue is represented by the rightmost item.

834+-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting

835+-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks

836+-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.

837+-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for

838+-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and

839+-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding

840+-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like

841+-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.

842+-- 4. Note that the scheduler queue expands by one more stream component in

843+-- every pass because one more @<>@ is broken down into two components. At this

844+-- point there are no more @<>@ operations to be broken down further and the

845+-- queue has reached its maximum size. Now these streams are scheduled in

846+-- round-robin fashion yielding @[2,7,5,3,8,8,9]@.

847+--

848+-- As we see above, in a right associated expression composed with @<>@, only

849+-- one @<>@ operation is broken down into two components in one execution,

850+-- therefore, if we have @n@ streams composed using @<>@ it will take @n@

851+-- scheduler passes to expand the whole expression. By the time @n-th@

852+-- component is added to the scheduler queue, the first component would have

853+-- received @n@ scheduler passes.

854+--

855+-- Since all streams get interleaved, this operation is not suitable for

856+-- folding an infinite lazy container of infinite size streams. However, if

857+-- the streams are small, the streams on the left may get finished before more

858+-- streams are added to the scheduler queue from the right side of the

859+-- expression, so it may be possible to fold an infinite lazy container of

860+-- streams. For example, if the streams are of size @n@ then at most @n@

861+-- streams would be in the scheduler queue at a time.

862+--

863+-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,

864+-- therefore the output of 'WAsyncT' even with a single thread of execution is

865+-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about

866+-- its scheduling behavior.

867+--

815868-- Any exceptions generated by a constituent stream are propagated to the

816869-- output stream. The output and exceptions from a single stream are guaranteed

817870-- to arrive in the same order in the resulting stream as they were generated

src/Streamly/Internal/Data/Stream/Serial.hs+14−6

@@ -245,12 +245,20 @@ TRAVERSABLE_INSTANCE(SerialT)

245245-- then we go back to evaluating stream @a@ and so on. In other words, the

246246-- elements of stream @a@ are interleaved with the elements of stream @b@.

247247--

248−-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we

249−-- interleave them in a binary fashion i.e. @a@ and @b@ are interleaved with

250−-- each other and the result is interleaved with @c@ and so on. This will not

251−-- act as a true round-robin scheduling across all the streams. Note that this

252−-- operation cannot be used to fold a container of infinite streams as the

253−-- state that it needs to maintain is proportional to the number of streams.

248+-- Note that evaluation of @a <> b <> c@ does not schedule @a@, @b@ and @c@

249+-- with equal priority. This expression is equivalent to @a <> (b <> c)@,

250+-- therefore, it fairly interleaves @a@ with the result of @b <> c@. For

251+-- example, @S.fromList [1,2] <> S.fromList [3,4] <> S.fromList [5,6] ::

252+-- WSerialT Identity Int@ would result in [1,3,2,5,4,6]. In other words, the

253+-- leftmost stream gets the same scheduling priority as the rest of the

254+-- streams taken together. The same is true for each subexpression on the right.

255+--

256+-- Note that this operation cannot be used to fold a container of infinite

257+-- streams as the state that it needs to maintain is proportional to the number

258+-- of streams.

259+--

260+-- The @W@ in the name stands for @wide@ or breadth wise scheduling in

261+-- contrast to the depth wise scheduling behavior of 'SerialT'.

254262--

255263-- @

256264-- import Streamly