07b2539 · streamly · git.iko.soy

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

compare with master

unifiedsplit

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

28 , Config

29 , maxThreads

30 , maxBuffer

31+ , eagerEval

31 -- maxYields

32 -- maxYields

32 , Rate(..)

33 , Rate(..)

33 , rate

34 , rate

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

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

15 , Config

16 , maxThreads

17 , maxBuffer

18+ , eagerEval

18 , Rate(..)

19 , Rate(..)

19 , rate

20 , rate

20 , avgRate

21 , avgRate

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

190mkEnqueue chan runner = do

191mkEnqueue chan runner = do

191 runInIO <- askRunInIO

192 runInIO <- askRunInIO

192 return

193 return

193− $ let q stream =

194+ $ let q stream = do

194 -- Enqueue the outer loop

195 -- Enqueue the outer loop

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

196 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

196 in q

203 in q

197

204

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

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

21

22import Control.Concurrent (myThreadId)

23import Control.Concurrent.MVar (newEmptyMVar)

24+import Control.Monad (when, void)

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

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

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

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

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

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

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

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

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

29import Streamly.Internal.Data.Atomics

30import Streamly.Internal.Data.Atomics

30 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

31 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

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

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

32

33

33import qualified Data.Set as Set

34import qualified Data.Set as Set

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

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

35

36

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

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

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

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

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

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

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

39

41

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

251 qEmpty <- isWorkFinished sv

253 qEmpty <- isWorkFinished sv

252 return $ qEmpty || yieldsDone

254 return $ qEmpty || yieldsDone

253

255

256+ let eager = getEagerDispatch cfg

257+

254 let getSVar :: Channel m a

258 let getSVar :: Channel m a

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

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

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

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

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

272 , workerThreads = running

276 , workerThreads = running

273 , workLoop = wloop q sv

277 , workLoop = wloop q sv

274 , enqueue = enqueueLIFO sv q

278 , enqueue = enqueueLIFO sv q

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

275 , isWorkDone = workDone sv

280 , isWorkDone = workDone sv

276 , isQueueDone = workDone sv

281 , isQueueDone = workDone sv

277 , needDoorBell = wfw

282 , needDoorBell = wfw

278 , svarMrun = mrun

283 , svarMrun = mrun

279 , workerCount = active

284 , workerCount = active

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

280− , accountThread = delThread running

286+ , accountThread = modifyThread running outQMv

281 , workerStopMVar = undefined

287 , workerStopMVar = undefined

282 , svarRef = Nothing

288 , svarRef = Nothing

283 , svarInspectMode = getInspectMode cfg

289 , svarInspectMode = getInspectMode cfg

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

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

62 , workerYieldCount = cntRef

63 , workerLatencyStart = lat

64 }

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.

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

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

66− >>= addThread (workerThreads sv)

67

76

68 where

77 where

69

78

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

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

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

71

81

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

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

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

287

297

288 liftIO $ delay sv

298 liftIO $ delay sv

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

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

300+ -- XXX or eagerDispatch is true

290 when (n <= 0) $ do

301 when (n <= 0) $ do

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

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

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

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

198 , workerThreads = running

199 , workLoop = wloop q sv

200 , enqueue = \_ -> enqueueFIFO sv q

201+ , eagerDispatch = return ()

201 , isWorkDone = workDone sv

202 , isWorkDone = workDone sv

202 , isQueueDone = workDone sv

203 , isQueueDone = workDone sv

203 , needDoorBell = wfw

204 , needDoorBell = wfw

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

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

85 , yieldRateInfo :: Maybe YieldRateInfo

86

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

88+ , eagerDispatch :: m ()

88 , isWorkDone :: IO Bool

89 , isWorkDone :: IO Bool

89 , isQueueDone :: IO Bool

90 , isQueueDone :: IO Bool

90 , needDoorBell :: IORef Bool

91 , needDoorBell :: IORef Bool

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

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

71 , maxBuffer

72 , maxYields

73 , inspectMode

74+ , eagerEval

74

75

75 , rate

76 , rate

76 , avgRate

77 , avgRate

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

86 , setStreamLatency

87 , setStreamLatency

87 , getYieldLimit

88 , getYieldLimit

88 , getInspectMode

89 , getInspectMode

90+ , getEagerDispatch

89

91

90 -- * Cleanup

92 -- * Cleanup

91 , cleanupSVar

93 , cleanupSVar

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

320 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency

322 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency

321 , _maxStreamRate :: Maybe Rate

323 , _maxStreamRate :: Maybe Rate

322 , _inspectMode :: Bool

324 , _inspectMode :: Bool

325+ , _eagerDispatch :: Bool

323 }

326 }

324

327

325-------------------------------------------------------------------------------

328-------------------------------------------------------------------------------

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

349 , _maxStreamRate = Nothing

352 , _maxStreamRate = Nothing

350 , _streamLatency = Nothing

353 , _streamLatency = Nothing

351 , _inspectMode = False

354 , _inspectMode = False

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

356+ , _eagerDispatch = False

352 }

357 }

353

358

354-------------------------------------------------------------------------------

359-------------------------------------------------------------------------------

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

458getStreamLatency :: Config -> Maybe NanoSecond64

463getStreamLatency :: Config -> Maybe NanoSecond64

459getStreamLatency = _streamLatency

464getStreamLatency = _streamLatency

460

465

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

467+--

468+-- /Pre-release/

469+--

461inspectMode :: Config -> Config

470inspectMode :: Config -> Config

462inspectMode st = st { _inspectMode = True }

471inspectMode st = st { _inspectMode = True }

463

472

464getInspectMode :: Config -> Bool

473getInspectMode :: Config -> Bool

465getInspectMode = _inspectMode

474getInspectMode = _inspectMode

466

475

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+

467-------------------------------------------------------------------------------

487-------------------------------------------------------------------------------

468-- Initialization

488-- Initialization

469-------------------------------------------------------------------------------

489-------------------------------------------------------------------------------