cef0c59 · streamly · git.iko.soy

git.iko.soy/streamly

Reset doorbell flag after wakeup in sendWorkerWait

commitcef0c5935b94ae6e8d7bde016fb59e49eecda0d5 authorHarendra Kumar <harendra@composewell.com> dateFri 14 Oct 2022, 04:03 (4 years ago) parent36953a1

4 files changed+17−11

compare with master

unifiedsplit

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Append.hs+4−4 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Dispatcher.hs+7−2 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Interleave.hs+2−2 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Type.hs+4−3

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

@@ -63,7 +63,7 @@ enqueueLIFO ::

6363enqueueLIFO sv q inner m = do

6464 atomicModifyIORefCAS_ q $ \(xs, ys) ->

6565 if inner then (xs, m : ys) else (m : xs, ys)

66− ringDoorBell (needDoorBell sv) (outputDoorBell sv)

66+ ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)

6767

6868data QResult a = QEmpty | QOuter a | QInner a

6969

@@ -239,7 +239,7 @@ enqueueAhead sv q m = do

239239 atomicModifyIORefCAS_ q $ \ case

240240 ([], n) -> ([snd m], n + 1) -- increment sequence

241241 _ -> error "enqueueAhead: queue is not empty"

242− ringDoorBell (needDoorBell sv) (outputDoorBell sv)

242+ ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)

243243

244244-- enqueue without incrementing the sequence number

245245{-# INLINE reEnqueueAhead #-}

@@ -249,7 +249,7 @@ reEnqueueAhead sv q m = do

249249 atomicModifyIORefCAS_ q $ \ case

250250 ([], n) -> ([m], n) -- DO NOT increment sequence

251251 _ -> error "reEnqueueAhead: queue is not empty"

252− ringDoorBell (needDoorBell sv) (outputDoorBell sv)

252+ ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)

253253

254254-- Normally the thread that has the token should never go away. The token gets

255255-- handed over to another thread, but someone or the other has the token at any

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

983983 then isQueueDoneAhead sv aheadQ

984984 else workDone sv

985985

986− , needDoorBell = wfw

986+ , doorBellOnWorkQ = wfw

987987 , svarMrun = mrun

988988 , workerCount = active

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

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

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

351351 -- that if we sleep because the queue was empty we are guaranteed

352352 -- to get a doorbell on the next enqueue.

353353

354− liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True

354+ liftIO $ atomicModifyIORefCAS_ (doorBellOnWorkQ sv) $ const True

355355 liftIO storeLoadBarrier

356356 canDoMore <- dispatch sv

357357

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

372372 "sendWorkerWait: nothing to do"

373373 $ takeMVar (outputDoorBell sv)

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

375− when (len <= 0) go

375+ if len <= 0

376+ then go

377+ else

378+ liftIO

379+ $ atomicModifyIORefCAS_ (doorBellOnWorkQ sv)

380+ $ const False

376381

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

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

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

@@ -50,7 +50,7 @@ enqueueFIFO ::

5050 -> IO ()

5151enqueueFIFO sv q m = do

5252 pushL q m

53− ringDoorBell (needDoorBell sv) (outputDoorBell sv)

53+ ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)

5454

5555{-# INLINE workLoopFIFO #-}

5656workLoopFIFO

@@ -201,7 +201,7 @@ getFifoSVar mrun cfg = do

201201 , eagerDispatch = return ()

202202 , isWorkDone = workDone sv

203203 , isQueueDone = workDone sv

204− , needDoorBell = wfw

204+ , doorBellOnWorkQ = wfw

205205 , svarMrun = mrun

206206 , workerCount = active

207207 , accountThread = delThread running

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

@@ -67,7 +67,7 @@ data Channel m a = Channel

6767

6868 -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty

6969 -- to non-empty, or a work item is queued by a worker to the work queue and

70− -- needDoorBell is set by the consumer.

70+ -- doorBellOnWorkQ is set by the consumer.

7171 , outputDoorBell :: MVar () -- signal the consumer about output

7272 -- XXX Can we use IO instead of m here?

7373 , readOutputQ :: m [ChildEvent a]

@@ -92,7 +92,8 @@ data Channel m a = Channel

9292 , eagerDispatch :: m ()

9393 , isWorkDone :: IO Bool

9494 , isQueueDone :: IO Bool

95− , needDoorBell :: IORef Bool

95+ , doorBellOnWorkQ :: IORef Bool -- ring outputDoorBell on enqueue so that a

96+ -- sleeping consumer thread can wake up and send more workers.

9697 , workLoop :: Maybe WorkerInfo -> m ()

9798

9899 -- Shared, thread tracking

@@ -160,7 +161,7 @@ dumpSVar sv = do

160161 , dumpOutputQ (outputQueue sv)

161162 -- XXX print the types of events in the outputQueue, first 5

162163 , dumpDoorBell (outputDoorBell sv)

163− , dumpNeedDoorBell (needDoorBell sv)

164+ , dumpNeedDoorBell (doorBellOnWorkQ sv)

164165 , dumpRunningThreads (workerThreads sv)

165166 -- XXX print the status of first 5 threads

166167 , dumpWorkerCount (workerCount sv)