13563a0 · streamly · git.iko.soy

src/Streamly/Internal/Data/Stream/Async/Channel/Append.hs+2−2

compare with master

unifiedsplit

@@ -295,11 +295,11 @@ getLifoSVar mrun cfg = do

295 case getStreamRate cfg of

296 Nothing ->

297 case getYieldLimit cfg of

298− Nothing -> getSVar sv readOutputQBounded

298+ Nothing -> getSVar sv (readOutputQBounded eager)

299 postProcessBounded

300 isWorkFinished

301 workLoopLIFO

302− Just _ -> getSVar sv readOutputQBounded

302+ Just _ -> getSVar sv (readOutputQBounded eager)

303 postProcessBounded

304 isWorkFinishedLimited

305 workLoopLIFOLimited

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

@@ -39,8 +39,8 @@ readOutputQChan sv = do

39 in readOutputQRaw (outputQueue sv) ss

40

41readOutputQBounded :: (MonadIO m, MonadBaseControl IO m) =>

42− Channel m a -> m [ChildEvent a]

42+ Bool -> Channel m a -> m [ChildEvent a]

43−readOutputQBounded sv = do

43+readOutputQBounded eager sv = do

44 (list, len) <- liftIO $ readOutputQChan sv

45 -- When there is no output seen we dispatch more workers to help

46 -- out if there is work pending in the work queue.

@@ -64,7 +64,7 @@ readOutputQBounded sv = do

64

65 {-# INLINE blockingRead #-}

66 blockingRead = do

67− sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv

67+ sendWorkerWait eager sendWorkerDelay (dispatchWorker 0) sv

68 liftIO (fst `fmap` readOutputQChan sv)

69

70readOutputQPaced :: (MonadIO m, MonadBaseControl IO m) =>

@@ -83,7 +83,7 @@ readOutputQPaced sv = do

83

84 {-# INLINE blockingRead #-}

85 blockingRead = do

86− sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv

86+ sendWorkerWait False sendWorkerDelayPaced dispatchWorkerPaced sv

87 liftIO (fst `fmap` readOutputQChan sv)

88

89postProcessPaced :: (MonadIO m, MonadBaseControl IO m) => Channel m a -> m Bool

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

@@ -286,19 +286,19 @@ dispatchWorkerPaced sv = do

286{-# NOINLINE sendWorkerWait #-}

287sendWorkerWait

288 :: MonadIO m

289+ => Bool

289− => (Channel m a -> IO ())

290+ -> (Channel m a -> IO ())

290 -> (Channel m a -> m Bool)

291 -> (Channel m a -> m Bool)

291 -> Channel m a

292 -> Channel m a

292 -> m ()

293 -> m ()

293−sendWorkerWait delay dispatch sv = do

294+sendWorkerWait eager delay dispatch sv = do

294 -- Note that we are guaranteed to have at least one outstanding worker when

295 -- Note that we are guaranteed to have at least one outstanding worker when

295 -- we enter this function. So if we sleep we are guaranteed to be woken up

296 -- we enter this function. So if we sleep we are guaranteed to be woken up

296 -- by an outputDoorBell, when the worker exits.

297 -- by an outputDoorBell, when the worker exits.

297

298

298 liftIO $ delay sv

299 liftIO $ delay sv

299 (_, n) <- liftIO $ readIORef (outputQueue sv)

300 (_, n) <- liftIO $ readIORef (outputQueue sv)

300− -- XXX or eagerDispatch is true

301− when (n <= 0) $ do

301+ when (n <= 0 || eager) $ do

302 -- The queue may be empty temporarily if the worker has dequeued the

303 -- work item but has not enqueued the remaining part yet. For the same

304 -- reason, a worker may come back if it tries to dequeue and finds the

@@ -356,7 +356,7 @@ sendWorkerWait delay dispatch sv = do

356 -- forever.

357

358 if canDoMore

359− then sendWorkerWait delay dispatch sv

359+ then sendWorkerWait eager delay dispatch sv

360 else do

361 liftIO

362 $ withDiagMVar

@@ -365,7 +365,7 @@ sendWorkerWait delay dispatch sv = do

365 "sendWorkerWait: nothing to do"

366 $ takeMVar (outputDoorBell sv)

367 (_, len) <- liftIO $ readIORef (outputQueue sv)

368− when (len <= 0) $ sendWorkerWait delay dispatch sv

368+ when (len <= 0) $ sendWorkerWait eager delay dispatch sv

369

370-- | Start the evaluation of the channel's work queue by kicking off a worker.

371-- Note: Work queue must not be empty otherwise the worker will exit without

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

@@ -216,11 +216,11 @@ getFifoSVar mrun cfg = do

216 case getStreamRate cfg of

217 Nothing ->

218 case getYieldLimit cfg of

219− Nothing -> getSVar sv readOutputQBounded

219+ Nothing -> getSVar sv (readOutputQBounded False)

220 postProcessBounded

221 isWorkFinished

222 workLoopFIFO

223− Just _ -> getSVar sv readOutputQBounded

223+ Just _ -> getSVar sv (readOutputQBounded False)

224 postProcessBounded

225 isWorkFinishedLimited

226 workLoopFIFOLimited

src/Streamly/Internal/Data/Stream/Channel/Types.hs+3

@@ -478,6 +478,9 @@ getInspectMode = _inspectMode

478-- option is on the evaluation is aggressive and does not wait for consumption

479-- to proceed, the only reason it might block is because of 'maxThreads' or

480-- 'maxBuffer' limitation.

481+--

482+-- Note that this option does nothing for interleave operations or when rate

483+-- has been specified.

481eagerEval :: Config -> Config

484eagerEval :: Config -> Config

482eagerEval st = st { _eagerDispatch = True }

485eagerEval st = st { _eagerDispatch = True }

483

486