cef0c59 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Reset doorbell flag after wakeup in sendWorkerWait

cef0c59Harendra Kumar4 years ago

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

compare with master

unifiedsplit

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

63enqueueLIFO sv q inner m = do

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

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

66− ringDoorBell (needDoorBell sv) (outputDoorBell sv)

66+ ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)

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

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

239 atomicModifyIORefCAS_ q $ \ case

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

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

242− ringDoorBell (needDoorBell sv) (outputDoorBell sv)

242+ ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)

243

244-- enqueue without incrementing the sequence number

245{-# INLINE reEnqueueAhead #-}

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

249 atomicModifyIORefCAS_ q $ \ case

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

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

252− ringDoorBell (needDoorBell sv) (outputDoorBell sv)

252+ ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)

253

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

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

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

983 then isQueueDoneAhead sv aheadQ

984 else workDone sv

985

986− , needDoorBell = wfw

986+ , doorBellOnWorkQ = wfw

987 , svarMrun = mrun

988 , workerCount = active

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

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

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

353

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

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

355 liftIO storeLoadBarrier

356 canDoMore <- dispatch sv

357

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

372 "sendWorkerWait: nothing to do"

373 $ takeMVar (outputDoorBell sv)

374 (_, 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

376

381

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

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

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

383-- 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 ::

50 -> IO ()

51enqueueFIFO sv q m = do

52 pushL q m

53− ringDoorBell (needDoorBell sv) (outputDoorBell sv)

53+ ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)

55{-# INLINE workLoopFIFO #-}

56workLoopFIFO

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

201 , eagerDispatch = return ()

202 , isWorkDone = workDone sv

203 , isQueueDone = workDone sv

204− , needDoorBell = wfw

204+ , doorBellOnWorkQ = wfw

205 , svarMrun = mrun

206 , workerCount = active

207 , accountThread = delThread running

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

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

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

69 -- 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.

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

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

73 , readOutputQ :: m [ChildEvent a]

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

92 , eagerDispatch :: m ()

93 , isWorkDone :: IO Bool

94 , 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.

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

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

98 -- Shared, thread tracking

99 -- Shared, thread tracking

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

160 , dumpOutputQ (outputQueue sv)

161 , dumpOutputQ (outputQueue sv)

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

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

162 , dumpDoorBell (outputDoorBell sv)

163 , dumpDoorBell (outputDoorBell sv)

163− , dumpNeedDoorBell (needDoorBell sv)

164+ , dumpNeedDoorBell (doorBellOnWorkQ sv)

164 , dumpRunningThreads (workerThreads sv)

165 , dumpRunningThreads (workerThreads sv)

165 -- XXX print the status of first 5 threads

166 -- XXX print the status of first 5 threads

166 , dumpWorkerCount (workerCount sv)

167 , dumpWorkerCount (workerCount sv)