07b2539 · streamly · git.iko.soy

git.iko.soy/streamly

Implement an eagerEval option

When this option is used dispatching of new workers cannot be blocked by the consumer thread processing. Also, we dispatch as soon as possible unless the maxThreads or maxBuffer limit is breached.

commit07b25394c460cb18639e15e46817bd1004cdbf6c authorHarendra Kumar <harendra@composewell.com> dateThu 6 Oct 2022, 16:23 (4 years ago) parenta6108c9

7 files changed+52−5

compare with master

unifiedsplit

src/Streamly/Internal/Data/Stream/Async.hs+1 src/Streamly/Internal/Data/Stream/Async/Channel.hs+8−1 src/Streamly/Internal/Data/Stream/Async/Channel/Append.hs+8−2 src/Streamly/Internal/Data/Stream/Async/Channel/Dispatcher.hs+13−2 src/Streamly/Internal/Data/Stream/Async/Channel/Interleave.hs+1 src/Streamly/Internal/Data/Stream/Async/Channel/Type.hs+1 src/Streamly/Internal/Data/Stream/Channel/Types.hs+20

src/Streamly/Internal/Data/Stream/Async.hs+1

@@ -28,6 +28,7 @@ module Streamly.Internal.Data.Stream.Async

2828 , Config

2929 , maxThreads

3030 , maxBuffer

31+ , eagerEval

3132 -- maxYields

3233 , Rate(..)

3334 , rate

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

@@ -15,6 +15,7 @@ module Streamly.Internal.Data.Stream.Async.Channel

1515 , Config

1616 , maxThreads

1717 , maxBuffer

18+ , eagerEval

1819 , Rate(..)

1920 , rate

2021 , avgRate

@@ -190,9 +191,15 @@ mkEnqueue :: MonadAsync m =>

190191mkEnqueue chan runner = do

191192 runInIO <- askRunInIO

192193 return

193− $ let q stream =

194+ $ let q stream = do

194195 -- Enqueue the outer loop

195196 liftIO $ enqueue chan False (runInIO, runner q stream)

197+ -- XXX In case of eager dispatch we can just directly dispatch

198+ -- a worker with the tail stream here rather than first queuing

199+ -- and then dispatching a worker which dequeues the work. The

200+ -- older implementation did a direct dispatch here and its perf

201+ -- characterstics looked much better.

202+ eagerDispatch chan

196203 in q

197204

198205-- XXX Can be renamed to concatMapWithK if we move concatMapWithK to higher

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

@@ -21,6 +21,7 @@ where

2121

2222import Control.Concurrent (myThreadId)

2323import Control.Concurrent.MVar (newEmptyMVar)

24+import Control.Monad (when, void)

2425import Control.Monad.IO.Class (MonadIO(liftIO))

2526import Control.Monad.Trans.Control (MonadBaseControl, restoreM)

2627import Data.IORef (IORef, newIORef, readIORef)

@@ -28,12 +29,13 @@ import Streamly.Internal.Control.Concurrent

2829 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)

2930import Streamly.Internal.Data.Atomics

3031 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

31−import Streamly.Internal.Data.Stream.Channel.Dispatcher (delThread)

32+import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)

3233

3334import qualified Data.Set as Set

3435import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

3536

3637import Streamly.Internal.Data.Stream.Async.Channel.Consumer

38+import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

3739import Streamly.Internal.Data.Stream.Async.Channel.Type

3840import Streamly.Internal.Data.Stream.Channel.Types

3941

@@ -251,6 +253,8 @@ getLifoSVar mrun cfg = do

251253 qEmpty <- isWorkFinished sv

252254 return $ qEmpty || yieldsDone

253255

256+ let eager = getEagerDispatch cfg

257+

254258 let getSVar :: Channel m a

255259 -> (Channel m a -> m [ChildEvent a])

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

@@ -272,12 +276,14 @@ getLifoSVar mrun cfg = do

272276 , workerThreads = running

273277 , workLoop = wloop q sv

274278 , enqueue = enqueueLIFO sv q

279+ , eagerDispatch = when eager $ void $ dispatchWorker 0 sv

275280 , isWorkDone = workDone sv

276281 , isQueueDone = workDone sv

277282 , needDoorBell = wfw

278283 , svarMrun = mrun

279284 , workerCount = active

280− , accountThread = delThread running

285+ -- XXX We can use delThread or modThread based on eager flag.

286+ , accountThread = modifyThread running outQMv

281287 , workerStopMVar = undefined

282288 , svarRef = Nothing

283289 , svarInspectMode = getInspectMode cfg

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

@@ -62,11 +62,21 @@ pushWorker yieldMax sv = do

6262 , workerYieldCount = cntRef

6363 , workerLatencyStart = lat

6464 }

65− doFork (workLoop sv winfo) (svarMrun sv) exception

66− >>= addThread (workerThreads sv)

65+ -- In case of lazy dispatch we dispatch workers only from the consumer

66+ -- thread. In that case it is ok to use addThread here as it is guaranteed

67+ -- that the thread will be added to the workerSet before the thread STOP

68+ -- event is processed, because we do both of these actions in the same

69+ -- consumer thread. However, in case of eager dispatch we may dispatch

70+ -- workers from workers, in which case the thread Stop even may get

71+ -- processed before the addThread occurs, so in that case we have to use

72+ -- modifyThread which performs a toggle rather than adding or deleting.

73+ --

74+ -- XXX We can use addThread or modThread based on eager flag.

75+ doFork (workLoop sv winfo) (svarMrun sv) exception >>= modThread

6776

6877 where

6978

79+ modThread = modifyThread (workerThreads sv) (outputDoorBell sv)

7080 exception = handleChildException (outputQueue sv) (outputDoorBell sv)

7181

7282-- | Determine the maximum number of workers required based on 'maxWorkerLimit'

@@ -287,6 +297,7 @@ sendWorkerWait delay dispatch sv = do

287297

288298 liftIO $ delay sv

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

300+ -- XXX or eagerDispatch is true

290301 when (n <= 0) $ do

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

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

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

@@ -198,6 +198,7 @@ getFifoSVar mrun cfg = do

198198 , workerThreads = running

199199 , workLoop = wloop q sv

200200 , enqueue = \_ -> enqueueFIFO sv q

201+ , eagerDispatch = return ()

201202 , isWorkDone = workDone sv

202203 , isQueueDone = workDone sv

203204 , needDoorBell = wfw

src/Streamly/Internal/Data/Stream/Async/Channel/Type.hs+1

@@ -85,6 +85,7 @@ data Channel m a = Channel

8585 , yieldRateInfo :: Maybe YieldRateInfo

8686

8787 , enqueue :: Bool -> (RunInIO m, Stream m a) -> IO ()

88+ , eagerDispatch :: m ()

8889 , isWorkDone :: IO Bool

8990 , isQueueDone :: IO Bool

9091 , needDoorBell :: IORef Bool

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

@@ -71,6 +71,7 @@ module Streamly.Internal.Data.Stream.Channel.Types

7171 , maxBuffer

7272 , maxYields

7373 , inspectMode

74+ , eagerEval

7475

7576 , rate

7677 , avgRate

@@ -86,6 +87,7 @@ module Streamly.Internal.Data.Stream.Channel.Types

8687 , setStreamLatency

8788 , getYieldLimit

8889 , getInspectMode

90+ , getEagerDispatch

8991

9092 -- * Cleanup

9193 , cleanupSVar

@@ -320,6 +322,7 @@ data Config = Config

320322 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency

321323 , _maxStreamRate :: Maybe Rate

322324 , _inspectMode :: Bool

325+ , _eagerDispatch :: Bool

323326 }

324327

325328-------------------------------------------------------------------------------

@@ -349,6 +352,8 @@ defaultConfig = Config

349352 , _maxStreamRate = Nothing

350353 , _streamLatency = Nothing

351354 , _inspectMode = False

355+ -- XXX Set it to True when Rate is not set?

356+ , _eagerDispatch = False

352357 }

353358

354359-------------------------------------------------------------------------------

@@ -458,12 +463,27 @@ setStreamLatency n st =

458463getStreamLatency :: Config -> Maybe NanoSecond64

459464getStreamLatency = _streamLatency

460465

466+-- | Print debug information about the 'Channel' when the stream ends.

467+--

468+-- /Pre-release/

469+--

461470inspectMode :: Config -> Config

462471inspectMode st = st { _inspectMode = True }

463472

464473getInspectMode :: Config -> Bool

465474getInspectMode = _inspectMode

466475

476+-- | By default the stream is evaluated on demand, new evaluation may be

477+-- deferred until the consumer consumes the evaluated stream. When 'eagerEval'

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+eagerEval :: Config -> Config

482+eagerEval st = st { _eagerDispatch = True }

483+

484+getEagerDispatch :: Config -> Bool

485+getEagerDispatch = _eagerDispatch

486+

467487-------------------------------------------------------------------------------

468488-- Initialization

469489-------------------------------------------------------------------------------