477a6de · streamly · git.iko.soy

git.iko.soy/streamly

ahead stream: do not send extra workers

when there is output available in the heap but not yet transferred to the outputQueue. This change may not be necessary if the output is promptly transferred from heap to outputQueue, which is not done at this momement, but future changes may do that.

commit477a6dea13296a3b09c33ffa6bd8b58caf13f824 authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 10 Sep 2018, 13:56 (8 years ago) parent2eec2cd

2 files changed+22−1

compare with master

unifiedsplit

src/Streamly/SVar.hs+19−1 src/Streamly/Streams/Async.hs+3

src/Streamly/SVar.hs+19−1

@@ -337,6 +337,7 @@ data SVar t m a = SVar

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

338338 , readOutputQ :: m [ChildEvent a]

339339 , postProcess :: m Bool

340+ , hasDelayedOutput :: IO Bool

340341

341342 -- Combined/aggregate parameters

342343 , maxWorkerLimit :: Limit

@@ -1147,6 +1148,20 @@ dequeueFromHeap hpRef = do

11471148 then ((hp', seqNo), Just ent)

11481149 else (hp, Nothing)

11491150

1151+{-# INLINE canDequeueFromHeap #-}

1152+canDequeueFromHeap

1153+ :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1154+ -> IO Bool

1155+canDequeueFromHeap hpVar =

1156+ withIORef hpVar $ \pair@(hp, snum) -> do

1157+ let r = H.uncons hp

1158+ case r of

1159+ Just (ent@(Entry seqNo _ev), hp') -> return $ seqNo == snum

1160+ _ -> return False

1161+ where

1162+

1163+ withIORef ref f = readIORef ref >>= f

1164+

11501165-------------------------------------------------------------------------------

11511166-- WAhead

11521167-------------------------------------------------------------------------------

@@ -1705,7 +1720,8 @@ sendWorkerWait delay dispatch sv = do

17051720

17061721 liftIO $ delay sv

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

1708− when (n <= 0) $ do

1723+ hasOutput <- if (n > 0) then return True else liftIO $ hasDelayedOutput sv

1724+ when (not hasOutput) $ do

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

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

17111727 -- reason, a worker may come back if it tries to dequeue and finds the

@@ -1949,6 +1965,7 @@ getAheadSVar st f = do

19491965 , outputDoorBell = outQMv

19501966 , readOutputQ = readOutput sv

19511967 , postProcess = postProc sv

1968+ , hasDelayedOutput = canDequeueFromHeap outH

19521969 , workerThreads = running

19531970 , workLoop = f q outH st{streamVar = Just sv} sv

19541971 , enqueue = enqueueAhead sv q

@@ -2046,6 +2063,7 @@ getParallelSVar st = do

20462063 , outputDoorBell = outQMv

20472064 , readOutputQ = readOutputQPar sv

20482065 , postProcess = allThreadsDone sv

2066+ , hasDelayedOutput = return False

20492067 , workerThreads = running

20502068 , workLoop = undefined

20512069 , enqueue = undefined

src/Streamly/Streams/Async.hs+3

@@ -300,6 +300,7 @@ getLifoSVar st = do

300300 , outputDoorBell = outQMv

301301 , readOutputQ = readOutput sv

302302 , postProcess = postProc sv

303+ , hasDelayedOutput = return False

303304 , workerThreads = running

304305 , workLoop = wloop q st{streamVar = Just sv} sv

305306 , enqueue = enqueueLIFO sv q

@@ -399,6 +400,7 @@ getFifoSVar st = do

399400 , outputDoorBell = outQMv

400401 , readOutputQ = readOutput sv

401402 , postProcess = postProc sv

403+ , hasDelayedOutput = return False

402404 , workerThreads = running

403405 , workLoop = wloop q st{streamVar = Just sv} sv

404406 , enqueue = enqueueFIFO sv q

@@ -408,6 +410,7 @@ getFifoSVar st = do

408410 , svarStyle = WAsyncVar

409411 , workerCount = active

410412 , accountThread = delThread sv

413+ , workerStopMVar = undefined

411414 , svarRef = Nothing

412415#ifdef DIAGNOSTICS

413416 , svarCreator = tid