7f2bf64 · streamly · git.iko.soy

git.iko.soy/streamly

Restrict maxThreads to maxBuffer

Currently maxThreads are independent of maxBuffer. If maxThreads are larger than maxBuffer we may dispatch as many workers. Since we check the buffer limit only after the worker yields a value, it may be too late and we may exceed the buffer already. To make sure that we never exceed the buffer we restrict the maxWorkers to maxBuffer value. The invariant maxThreads <= maxBuffer holds.

commit7f2bf64d3d86a6a955303ae26fdd78fcc72f9b30 authorHarendra Kumar <harendra.kumar@gmail.com> dateSun 23 Jun 2019, 20:40 (7 years ago) parenta7556e8

2 files changed+22−5

compare with master

unifiedsplit

src/Streamly/SVar.hs+20−3 src/Streamly/Streams/Async.hs+2−2

src/Streamly/SVar.hs+20−3

@@ -331,6 +331,18 @@ data SVarStats = SVarStats {

331331-- This is essentially a 'Maybe Word' type

332332data Limit = Unlimited | Limited Word deriving Show

333333

334+instance Eq Limit where

335+ Unlimited == Unlimited = True

336+ Unlimited == Limited _ = False

337+ Limited _ == Unlimited = False

338+ Limited x == Limited y = x == y

339+

340+instance Ord Limit where

341+ Unlimited <= Unlimited = True

342+ Unlimited <= Limited _ = False

343+ Limited _ <= Unlimited = True

344+ Limited x <= Limited y = x <= y

345+

334346-- When to stop the composed stream.

335347data SVarStopStyle =

336348 StopNone -- stops only when all streams are finished

@@ -375,8 +387,15 @@ data SVar t m a = SVar

375387 , postProcess :: m Bool

376388

377389 -- Combined/aggregate parameters

390+ -- This is truncated to maxBufferLimit if set to more than that. Otherwise

391+ -- potentially each worker may yield one value to the buffer in the worst

392+ -- case exceeding the requested buffer size.

378393 , maxWorkerLimit :: Limit

379394 , maxBufferLimit :: Limit

395+

396+ -- [LOCKING] Read only access by consumer when dispatching a worker.

397+ -- Decremented by workers when picking work and undo decrement if the

398+ -- worker does not yield a value.

380399 , remainingWork :: Maybe (IORef Count)

381400 , yieldRateInfo :: Maybe YieldRateInfo

382401

@@ -947,8 +966,6 @@ send sv msg = do

947966 -- doorbell if something was added to the queue after it empties it.

948967 void $ tryPutMVar (outputDoorBell sv) ()

949968

950− -- XXX we should reserve the buffer when we pick up the work from the

951− -- queue, instead of checking it here when it is too late.

952969 let limit = maxBufferLimit sv

953970 case limit of

954971 Unlimited -> return True

@@ -2085,7 +2102,7 @@ getAheadSVar st f mrun = do

20852102 { outputQueue = outQ

20862103 , remainingWork = yl

20872104 , maxBufferLimit = getMaxBuffer st

2088− , maxWorkerLimit = getMaxThreads st

2105+ , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

20892106 , yieldRateInfo = rateInfo

20902107 , outputDoorBell = outQMv

20912108 , readOutputQ = readOutput sv

src/Streamly/Streams/Async.hs+2−2

@@ -294,7 +294,7 @@ getLifoSVar st mrun = do

294294 { outputQueue = outQ

295295 , remainingWork = yl

296296 , maxBufferLimit = getMaxBuffer st

297− , maxWorkerLimit = getMaxThreads st

297+ , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

298298 , yieldRateInfo = rateInfo

299299 , outputDoorBell = outQMv

300300 , readOutputQ = readOutput sv

@@ -386,7 +386,7 @@ getFifoSVar st mrun = do

386386 { outputQueue = outQ

387387 , remainingWork = yl

388388 , maxBufferLimit = getMaxBuffer st

389− , maxWorkerLimit = getMaxThreads st

389+ , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

390390 , yieldRateInfo = rateInfo

391391 , outputDoorBell = outQMv

392392 , readOutputQ = readOutput sv