362ceb1 · streamly · git.iko.soy

git.iko.soy/streamly

Rename eagerEval to eager and inspectMode to inspect

commit362ceb1e397e42d9e0ef3511e6e1f83ca914c465 authorHarendra Kumar <harendra@composewell.com> dateSat 15 Oct 2022, 18:30 (4 years ago) parentc60b704

11 files changed+49−49

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Data/Stream/ConcurrentEager.hs+1−1

@@ -20,4 +20,4 @@ moduleName = "Data.Stream.ConcurrentEager"

2020

2121main :: IO ()

2222main =

23− runWithCLIOpts defaultStreamSize (allBenchmarks moduleName Async.eagerEval)

23+ runWithCLIOpts defaultStreamSize (allBenchmarks moduleName Async.eager)

src/Streamly/Data/Stream/Concurrent.hs+2−2

@@ -14,8 +14,8 @@ module Streamly.Data.Stream.Concurrent

1414 -- ** Set config

1515 , maxThreads

1616 , maxBuffer

17− , inspectMode

18− , eagerEval

17+ , inspect

18+ , eager

1919 , stopWhen

2020 , ordered

2121 , interleaved

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

@@ -138,7 +138,7 @@ collectLatency ::

138138 -> YieldRateInfo

139139 -> Bool

140140 -> IO (Count, AbsTime, NanoSecond64)

141−collectLatency inspect ss yinfo drain = do

141+collectLatency inspecting ss yinfo drain = do

142142 let cur = workerPendingLatency yinfo

143143 col = workerCollectedLatency yinfo

144144 longTerm = svarAllTimeLatency yinfo

@@ -155,7 +155,7 @@ collectLatency inspect ss yinfo drain = do

155155 Nothing -> retWith prevLat

156156 Just (count, time) -> do

157157 let newLat = time `div` fromIntegral count

158− when inspect $ recordMinMaxLatency ss newLat

158+ when inspecting $ recordMinMaxLatency ss newLat

159159 -- When we have collected a significant sized batch we compute the

160160 -- new latency using that batch and return the new latency,

161161 -- otherwise we return the previous latency derived from the

@@ -164,7 +164,7 @@ collectLatency inspect ss yinfo drain = do

164164 then do

165165 -- XXX make this NOINLINE?

166166 updateWorkerPollingInterval yinfo (max newLat prevLat)

167− when inspect $ recordAvgLatency ss (count, time)

167+ when inspecting $ recordAvgLatency ss (count, time)

168168 writeIORef col (0, 0, 0)

169169 writeIORef measured ((prevLat + newLat) `div` 2)

170170 modifyIORef longTerm $ \(_, t) -> (newLcount, t)

@@ -176,11 +176,11 @@ collectLatency inspect ss yinfo drain = do

176176-------------------------------------------------------------------------------

177177

178178dumpSVarStats :: Bool -> Maybe YieldRateInfo -> SVarStats -> IO String

179−dumpSVarStats inspect rateInfo ss = do

179+dumpSVarStats inspecting rateInfo ss = do

180180 case rateInfo of

181181 Nothing -> return ()

182182 Just yinfo -> do

183− _ <- liftIO $ collectLatency inspect ss yinfo True

183+ _ <- liftIO $ collectLatency inspecting ss yinfo True

184184 return ()

185185

186186 dispatches <- readIORef $ totalDispatches ss

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

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

7171 , maxThreads

7272 , maxBuffer

7373 , maxYields

74− , inspectMode

75− , eagerEval

74+ , inspect

75+ , eager

7676 , stopWhen

7777 , ordered

7878 , interleaved

@@ -340,7 +340,7 @@ data Config = Config

340340 -- XXX these two can be collapsed into a single type

341341 , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency

342342 , _maxStreamRate :: Maybe Rate

343− , _inspectMode :: Bool

343+ , _inspect :: Bool

344344 , _eagerDispatch :: Bool

345345 , _stopWhen :: StopWhen

346346 , _ordered :: Bool

@@ -373,7 +373,7 @@ defaultConfig = Config

373373 , _bufferHigh = defaultMaxBuffer

374374 , _maxStreamRate = Nothing

375375 , _streamLatency = Nothing

376− , _inspectMode = False

376+ , _inspect = False

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

378378 , _eagerDispatch = False

379379 , _stopWhen = AllStop

@@ -494,24 +494,24 @@ getStreamLatency = _streamLatency

494494--

495495-- /Pre-release/

496496--

497−inspectMode :: Config -> Config

498−inspectMode st = st { _inspectMode = True }

497+inspect :: Config -> Config

498+inspect st = st { _inspect = True }

499499

500500getInspectMode :: Config -> Bool

501−getInspectMode = _inspectMode

501+getInspectMode = _inspect

502502

503−-- XXX rename to "eager"

504−

505−-- | By default the stream is evaluated on demand, new evaluation may be

506−-- deferred until the consumer consumes the evaluated stream. When 'eagerEval'

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

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

509−-- 'maxBuffer' limitation.

503+-- | By default, processing of output from the worker threads is given priority

504+-- over dispatching new workers. More workers are dispatched only when there is

505+-- no output to process. With 'eager' on workers are dispatched aggresively as

506+-- long as there is more work to do irrespective of whether there is output

507+-- pending to be processed. However, dispatching may stop if 'maxThreads' or

508+-- 'maxBuffer' is reached.

509+--

510+-- /Note:/ This option has no effect when rate has been specified.

511+-- /Note:/ Not supported with 'interleaved'.

510512--

511−-- Note that this option does nothing for interleave operations or when rate

512−-- has been specified.

513−eagerEval :: Config -> Config

514−eagerEval st = st { _eagerDispatch = True }

513+eager :: Config -> Config

514+eager st = st { _eagerDispatch = True }

515515

516516getEagerDispatch :: Config -> Bool

517517getEagerDispatch = _eagerDispatch

@@ -783,8 +783,8 @@ stmExcHandler dump label e@BlockedIndefinitelyOnSTM = do

783783-- happen, but it may result in unexpected output when threads are left hanging

784784-- until they are GCed because the consumer went away.

785785withDiagMVar :: Bool -> IO String -> String -> IO () -> IO ()

786−withDiagMVar inspect dump label action =

787− if inspect

786+withDiagMVar inspecting dump label action =

787+ if inspecting

788788 then

789789 action `catches` [ Handler (mvarExcHandler dump label)

790790 , Handler (stmExcHandler dump label)

src/Streamly/Internal/Data/Stream/Concurrent.hs+9−9

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

2828 , Config

2929 , maxThreads

3030 , maxBuffer

31− , eagerEval

31+ , eager

3232 , StopWhen (..)

3333 , stopWhen

3434 , ordered

@@ -40,7 +40,7 @@ module Streamly.Internal.Data.Stream.Concurrent

4040 , minRate

4141 , maxRate

4242 , constRate

43− , inspectMode

43+ , inspect

4444

4545 -- * Combinators

4646 -- | Stream combinators using a concurrent channel

@@ -272,32 +272,32 @@ interleave = combineWith interleaved

272272ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

273273ahead = combineWith ordered

274274

275−-- | Like 'append' but with 'eagerEval' on.

275+-- | Like 'append' but with 'eager' on.

276276--

277−-- >>> parallel = Async.combineWith Async.eagerEval

277+-- >>> parallel = Async.combineWith Async.eager

278278--

279279{-# INLINE parallel #-}

280280parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

281−parallel = combineWith eagerEval

281+parallel = combineWith eager

282282

283283-- | Like 'parallel' but stops the output as soon as the first stream stops.

284284--

285−-- >>> parallelFst = Async.combineWith (Async.eagerEval . Async.stopWhen Async.FirstStops)

285+-- >>> parallelFst = Async.combineWith (Async.eager . Async.stopWhen Async.FirstStops)

286286--

287287-- /Pre-release/

288288{-# INLINE parallelFst #-}

289289parallelFst :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

290−parallelFst = combineWith (eagerEval . stopWhen FirstStops)

290+parallelFst = combineWith (eager . stopWhen FirstStops)

291291

292292-- | Like 'parallel' but stops the output as soon as any of the two streams

293293-- stops.

294294--

295−-- >>> parallelMin = Async.combineWith (Async.eagerEval . Async.stopWhen Async.AnyStops)

295+-- >>> parallelMin = Async.combineWith (Async.eager . Async.stopWhen Async.AnyStops)

296296--

297297-- /Pre-release/

298298{-# INLINE parallelMin #-}

299299parallelMin :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

300−parallelMin = combineWith (eagerEval . stopWhen AnyStops)

300+parallelMin = combineWith (eager . stopWhen AnyStops)

301301

302302-------------------------------------------------------------------------------

303303-- concat

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

@@ -40,12 +40,12 @@ module Streamly.Internal.Data.Stream.Concurrent.Channel

4040 , getStopWhen

4141

4242 -- ** Scheduling behavior

43− , eagerEval

43+ , eager

4444 , ordered

4545 , interleaved

4646

4747 -- ** Diagnostics

48− , inspectMode

48+ , inspect

4949 )

5050where

5151

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Append.hs+4−4

@@ -941,7 +941,7 @@ getLifoSVar mrun cfg = do

941941 qEmpty <- isWorkFinished sv

942942 return $ qEmpty || yieldsDone

943943

944− let eager = getEagerDispatch cfg

944+ let eagerEval = getEagerDispatch cfg

945945 inOrder = getOrdered cfg

946946

947947 let getSVar :: Channel m a

@@ -973,7 +973,7 @@ getLifoSVar mrun cfg = do

973973 if inOrder

974974 then enqueueAhead sv aheadQ

975975 else enqueueLIFO sv q inner

976− , eagerDispatch = when eager $ void $ dispatchWorker 0 sv

976+ , eagerDispatch = when eagerEval $ void $ dispatchWorker 0 sv

977977 , isWorkDone =

978978 if inOrder

979979 then isWorkDoneAhead sv aheadQ outH

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

999999 case getStreamRate cfg of

10001000 Nothing ->

10011001 case getYieldLimit cfg of

1002− Nothing -> getSVar sv (readOutputQBounded eager)

1002+ Nothing -> getSVar sv (readOutputQBounded eagerEval)

10031003 postProcessBounded

10041004 isWorkFinished

10051005 workLoopLIFO

1006− Just _ -> getSVar sv (readOutputQBounded eager)

1006+ Just _ -> getSVar sv (readOutputQBounded eagerEval)

10071007 postProcessBounded

10081008 isWorkFinishedLimited

10091009 workLoopLIFOLimited

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Consumer.hs+2−2

@@ -40,7 +40,7 @@ readOutputQChan sv = do

4040

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

4242 Bool -> Channel m a -> m [ChildEvent a]

43−readOutputQBounded eager sv = do

43+readOutputQBounded eagerEval sv = do

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

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

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

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

6464

6565 {-# INLINE blockingRead #-}

6666 blockingRead = do

67− sendWorkerWait eager sendWorkerDelay (dispatchWorker 0) sv

67+ sendWorkerWait eagerEval sendWorkerDelay (dispatchWorker 0) sv

6868 liftIO (fst `fmap` readOutputQChan sv)

6969

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

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

@@ -291,7 +291,7 @@ sendWorkerWait

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

292292 -> Channel m a

293293 -> m ()

294−sendWorkerWait eager delay dispatch sv = go

294+sendWorkerWait eagerEval delay dispatch sv = go

295295

296296 where

297297

@@ -303,7 +303,7 @@ sendWorkerWait eager delay dispatch sv = go

303303

304304 liftIO $ delay sv

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

306− when (n <= 0 || eager) $ do

306+ when (n <= 0 || eagerEval) $ do

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

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

309309 -- the same reason, a worker may come back if it tries to dequeue

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

@@ -44,7 +44,7 @@ import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

4444

4545import Streamly.Internal.Data.Stream.Concurrent.Channel.Dispatcher

4646import Streamly.Internal.Data.Stream.Concurrent.Channel.Type

47−import Streamly.Internal.Data.Stream.Channel.Types

47+import Streamly.Internal.Data.Stream.Channel.Types hiding (inspect)

4848

4949import Prelude hiding (map, concat, concatMap)

5050

test/Streamly/Test/Data/Stream/Concurrent.hs+1−1

@@ -82,7 +82,7 @@ commonOpConfigs =

8282 , ("maxBuffer 0", Async.maxBuffer 0)

8383 , ("maxThreads 0", Async.maxThreads 0)

8484 , ("maxThreads 1", Async.maxThreads 1)

85− , ("eagerEval", Async.eagerEval)

85+ , ("eager", Async.eager)

8686 -- XXX Need to use an unsorted eq operation for ahead

8787 , ("ordered", Async.ordered)

8888#ifdef USE_LARGE_MEMORY