@@ -106,7 +106,7 @@ module Streamly.SVar
106106where
107107
108108import Control.Concurrent
109− (ThreadId, myThreadId, threadDelay, getNumCapabilities, throwTo)
109+ (ThreadId, myThreadId, threadDelay, throwTo)
110110import Control.Concurrent.MVar
111111 (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)
112112import Control.Exception (SomeException(..), catch, mask, assert, Exception)
@@ -342,7 +342,6 @@ data SVar t m a = SVar
342342 , outputDoorBell :: MVar () -- signal the consumer about output
343343 , readOutputQ :: m [ChildEvent a]
344344 , postProcess :: m Bool
345− , hasDelayedOutput :: IO Bool
346345
347346 -- Combined/aggregate parameters
348347 , maxWorkerLimit :: Limit
@@ -1171,20 +1170,6 @@ dequeueFromHeap hpVar =
11711170 ((hp', Nothing), Ready ent)
11721171 _ -> (pair, Waiting n)
11731172
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−
11881173{-# INLINE dequeueFromHeapSeq #-}
11891174dequeueFromHeapSeq
11901175 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
@@ -1743,7 +1728,7 @@ sendWorkerDelayPaced :: SVar t m a -> IO ()
17431728sendWorkerDelayPaced _ = return ()
17441729
17451730sendWorkerDelay :: SVar t m a -> IO ()
1746−sendWorkerDelay sv = do
1731+sendWorkerDelay _sv = do
17471732 -- XXX we need a better way to handle this than hardcoded delays. The
17481733 -- delays may be different for different systems.
17491734 -- If there is a usecase where this is required we can create a combinator
@@ -1776,8 +1761,7 @@ sendWorkerWait delay dispatch sv = do
17761761
17771762 liftIO $ delay sv
17781763 (_, n) <- liftIO $ readIORef (outputQueue sv)
1779− hasOutput <- if (n > 0) then return True else liftIO $ hasDelayedOutput sv
1780− when (not hasOutput) $ do
1764+ when (n <= 0) $ do
17811765 -- The queue may be empty temporarily if the worker has dequeued the
17821766 -- work item but has not enqueued the remaining part yet. For the same
17831767 -- reason, a worker may come back if it tries to dequeue and finds the
@@ -2024,7 +2008,6 @@ getAheadSVar st f = do
20242008 , outputDoorBell = outQMv
20252009 , readOutputQ = readOutput sv
20262010 , postProcess = postProc sv
2027− , hasDelayedOutput = canDequeueFromHeap outH
20282011 , workerThreads = running
20292012 , workLoop = f q outH st{streamVar = Just sv} sv
20302013 , enqueue = enqueueAhead sv q
@@ -2122,7 +2105,6 @@ getParallelSVar st = do
21222105 , outputDoorBell = outQMv
21232106 , readOutputQ = readOutputQPar sv
21242107 , postProcess = allThreadsDone sv
2125− , hasDelayedOutput = return False
21262108 , workerThreads = running
21272109 , workLoop = undefined
21282110 , enqueue = undefined