2f6f9a2 · streamly · git.iko.soy

git.iko.soy/streamly

Edit documentation and comments in concurrent modules

* Fix typo in StreamK * Fix typo in the Dispatcher module * Edit comments in Interleave module * Clarify the documentation of "rate" (Async)

commit2f6f9a2d4e9cb708c210ae964ad9f8502b20a971 authorHarendra Kumar <harendra@composewell.com> dateTue 27 Sep 2022, 19:31 (4 years ago) parent770887d

4 files changed+24−13

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/Stream/StreamK/Type.hs+1−1 src/Streamly/Internal/Data/Stream/Async/Channel/Interleave.hs+4−5 src/Streamly/Internal/Data/Stream/Channel/Dispatcher.hs+1−1 src/Streamly/Internal/Data/Stream/Channel/Types.hs+18−6

core/src/Streamly/Internal/Data/Stream/StreamK/Type.hs+1−1

@@ -1380,7 +1380,7 @@ reverse :: Stream m a -> Stream m a

13801380reverse = foldlS (flip cons) nil

13811381

13821382------------------------------------------------------------------------------

1383−-- Runnining effects

1383+-- Running effects

13841384------------------------------------------------------------------------------

13851385

13861386-- | Run an action before evaluating the stream.

src/Streamly/Internal/Data/Stream/Async/Channel/Interleave.hs+4−5

@@ -37,9 +37,10 @@ import Streamly.Internal.Data.Stream.Channel.Types

3737

3838data WorkerStatus = Continue | Suspend

3939

40−-- XXX we can use the Ahead style sequence/heap mechanism to make the best

41−-- effort to always try to finish the streams on the left side of an expression

42−-- first as long as possible.

40+-- XXX This is not strictly round-robin as the streams that are faster may

41+-- yield more elements than the ones that are slower. Also, when streams

42+-- suspend due to buffer getting full they get added to the queue in a random

43+-- order.

4344

4445{-# INLINE enqueueFIFO #-}

4546enqueueFIFO ::

@@ -51,8 +52,6 @@ enqueueFIFO sv q m = do

5152 pushL q m

5253 ringDoorBell (needDoorBell sv) (outputDoorBell sv)

5354

54−-- XXX we can remove sv as it is derivable from st

55−

5655{-# INLINE workLoopFIFO #-}

5756workLoopFIFO

5857 :: MonadRunInIO m

src/Streamly/Internal/Data/Stream/Channel/Dispatcher.hs+1−1

@@ -263,7 +263,7 @@ delThread workerSet tid =

263263

264264-- If present then delete else add. This takes care of out of order add and

265265-- delete i.e. a delete arriving before we even added a thread.

266−-- This occurs when the forked thRead is done even before the 'addThread' right

266+-- This occurs when the forked thread is done even before the 'addThread' right

267267-- after the fork gets a chance to run.

268268{-# INLINE modifyThread #-}

269269modifyThread :: MonadIO m => IORef (Set ThreadId) -> MVar () -> ThreadId -> m ()

src/Streamly/Internal/Data/Stream/Channel/Types.hs+18−6

@@ -417,18 +417,30 @@ maxBuffer n st =

417417getMaxBuffer :: Config -> Limit

418418getMaxBuffer = _bufferHigh

419419

420−-- | Specify the pull rate of a channel.

421−-- A 'Nothing' value resets the rate to default which is unlimited. When the

422−-- rate is specified, concurrent production may be ramped up or down

423−-- automatically to achieve the specified yield rate. The specific behavior for

424−-- different styles of 'Rate' specifications is documented under 'Rate'. The

425−-- effective maximum production rate achieved by a channel is governed by:

420+-- | Specify the stream evaluation rate of a channel.

421+--

422+-- A 'Nothing' value means there is no smart rate control, concurrent execution

423+-- blocks only if 'maxThreads' or 'maxBuffer' is reached, or there are no more

424+-- concurrent tasks to execute. This is the default.

425+--

426+-- When rate (throughput) is specified, concurrent production may be ramped

427+-- up or down automatically to achieve the specified stream throughput. The

428+-- specific behavior for different styles of 'Rate' specifications is

429+-- documented under 'Rate'. The effective maximum production rate achieved by

430+-- a channel is governed by:

426431--

427432-- * The 'maxThreads' limit

428433-- * The 'maxBuffer' limit

429434-- * The maximum rate that the stream producer can achieve

430435-- * The maximum rate that the stream consumer can achieve

431436--

437+-- Maximum production rate is given by:

438+--

439+-- \(rate = \frac{maxThreads}{latency}\)

440+--

441+-- If we know the average latency of the tasks we can set 'maxThreads'

442+-- accordingly.

443+--

432444rate :: Maybe Rate -> Config -> Config

433445rate r st = st { _maxStreamRate = r }

434446