2f6f9a2 · streamly · git.iko.soy

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

compare with master

unifiedsplit

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

1380reverse = foldlS (flip cons) nil

1381

1382------------------------------------------------------------------------------

1383−-- Runnining effects

1383+-- Running effects

1384------------------------------------------------------------------------------

1385

1386-- | 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

37

38data WorkerStatus = Continue | Suspend

39

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

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

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

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

42−-- first as long as possible.

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

43+-- order.

43

44

44{-# INLINE enqueueFIFO #-}

45{-# INLINE enqueueFIFO #-}

45enqueueFIFO ::

46enqueueFIFO ::

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

51 pushL q m

52 pushL q m

52 ringDoorBell (needDoorBell sv) (outputDoorBell sv)

53 ringDoorBell (needDoorBell sv) (outputDoorBell sv)

53

54

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

55−

56{-# INLINE workLoopFIFO #-}

55{-# INLINE workLoopFIFO #-}

57workLoopFIFO

56workLoopFIFO

58 :: MonadRunInIO m

57 :: MonadRunInIO m

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

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

263

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

265-- 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

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

268{-# INLINE modifyThread #-}

269modifyThread :: 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 =

417getMaxBuffer :: Config -> Limit

418getMaxBuffer = _bufferHigh

419

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

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

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

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+--

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

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

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

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

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

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

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

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

430+-- a channel is governed by:

426--

431--

427-- * The 'maxThreads' limit

432-- * The 'maxThreads' limit

428-- * The 'maxBuffer' limit

433-- * The 'maxBuffer' limit

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

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

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

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

431--

436--

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+--

432rate :: Maybe Rate -> Config -> Config

444rate :: Maybe Rate -> Config -> Config

433rate r st = st { _maxStreamRate = r }

445rate r st = st { _maxStreamRate = r }

434

446