e583f56 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

remove heap delayed output check

e583f56Harendra Kumar8 years ago

src/Streamly/SVar.hs+3−21

compare with master

unifiedsplit

@@ -106,7 +106,7 @@ module Streamly.SVar

106where

107

108import Control.Concurrent

109− (ThreadId, myThreadId, threadDelay, getNumCapabilities, throwTo)

109+ (ThreadId, myThreadId, threadDelay, throwTo)

110import Control.Concurrent.MVar

111 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)

112import Control.Exception (SomeException(..), catch, mask, assert, Exception)

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

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

343 , readOutputQ :: m [ChildEvent a]

344 , postProcess :: m Bool

345− , hasDelayedOutput :: IO Bool

346

345

347 -- Combined/aggregate parameters

346 -- Combined/aggregate parameters

348 , maxWorkerLimit :: Limit

347 , maxWorkerLimit :: Limit

@@ -1171,20 +1170,6 @@ dequeueFromHeap hpVar =

1171 ((hp', Nothing), Ready ent)

1170 ((hp', Nothing), Ready ent)

1172 _ -> (pair, Waiting n)

1171 _ -> (pair, Waiting n)

1173

1172

1174−{-# INLINE canDequeueFromHeap #-}

1175−canDequeueFromHeap

1176− :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

1177− -> IO Bool

1178−canDequeueFromHeap hpVar =

1179− withIORef hpVar $ \(hp, snum) ->

1180− case snum of

1181− Nothing -> return True

1182− Just n -> do

1183− let r = H.uncons hp

1184− case r of

1185− Just ((Entry seqNo _ev), _) -> return $ seqNo == n

1186− _ -> return False

1187−

1188{-# INLINE dequeueFromHeapSeq #-}

1173{-# INLINE dequeueFromHeapSeq #-}

1189dequeueFromHeapSeq

1174dequeueFromHeapSeq

1190 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

1175 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

@@ -1743,7 +1728,7 @@ sendWorkerDelayPaced :: SVar t m a -> IO ()

1743sendWorkerDelayPaced _ = return ()

1728sendWorkerDelayPaced _ = return ()

1744

1729

1745sendWorkerDelay :: SVar t m a -> IO ()

1730sendWorkerDelay :: SVar t m a -> IO ()

1746−sendWorkerDelay sv = do

1731+sendWorkerDelay _sv = do

1747 -- XXX we need a better way to handle this than hardcoded delays. The

1732 -- XXX we need a better way to handle this than hardcoded delays. The

1748 -- delays may be different for different systems.

1733 -- delays may be different for different systems.

1749 -- If there is a usecase where this is required we can create a combinator

1734 -- If there is a usecase where this is required we can create a combinator

@@ -1776,8 +1761,7 @@ sendWorkerWait delay dispatch sv = do

1776

1761

1777 liftIO $ delay sv

1762 liftIO $ delay sv

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

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

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

1764+ when (n <= 0) $ do

1780− when (not hasOutput) $ do

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

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

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

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

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

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

@@ -2024,7 +2008,6 @@ getAheadSVar st f = do

2024 , outputDoorBell = outQMv

2008 , outputDoorBell = outQMv

2025 , readOutputQ = readOutput sv

2009 , readOutputQ = readOutput sv

2026 , postProcess = postProc sv

2010 , postProcess = postProc sv

2027− , hasDelayedOutput = canDequeueFromHeap outH

2028 , workerThreads = running

2011 , workerThreads = running

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

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

2030 , enqueue = enqueueAhead sv q

2013 , enqueue = enqueueAhead sv q

@@ -2122,7 +2105,6 @@ getParallelSVar st = do

2122 , outputDoorBell = outQMv

2105 , outputDoorBell = outQMv

2123 , readOutputQ = readOutputQPar sv

2106 , readOutputQ = readOutputQPar sv

2124 , postProcess = allThreadsDone sv

2107 , postProcess = allThreadsDone sv

2125− , hasDelayedOutput = return False

2126 , workerThreads = running

2108 , workerThreads = running

2127 , workLoop = undefined

2109 , workLoop = undefined

2128 , enqueue = undefined

2110 , enqueue = undefined

src/Streamly/Streams/Async.hs−2

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

300 , outputDoorBell = outQMv

301 , readOutputQ = readOutput sv

302 , postProcess = postProc sv

303− , hasDelayedOutput = return False

304 , workerThreads = running

303 , workerThreads = running

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

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

306 , enqueue = enqueueLIFO sv q

305 , enqueue = enqueueLIFO sv q

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

400 , outputDoorBell = outQMv

399 , outputDoorBell = outQMv

401 , readOutputQ = readOutput sv

400 , readOutputQ = readOutput sv

402 , postProcess = postProc sv

401 , postProcess = postProc sv

403− , hasDelayedOutput = return False

404 , workerThreads = running

402 , workerThreads = running

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

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

406 , enqueue = enqueueFIFO sv q

404 , enqueue = enqueueFIFO sv q