@@ -102,7 +102,8 @@ where
102import Control.Concurrent
102import Control.Concurrent
103 (ThreadId, myThreadId, threadDelay, getNumCapabilities, throwTo)
103 (ThreadId, myThreadId, threadDelay, getNumCapabilities, throwTo)
104import Control.Concurrent.MVar
104import Control.Concurrent.MVar
105− (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)
105+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, readMVar,
106+ modifyMVar, modifyMVar_, withMVar)
106import Control.Exception (SomeException(..), catch, mask, assert, Exception)
107import Control.Exception (SomeException(..), catch, mask, assert, Exception)
107import Control.Monad (when)
108import Control.Monad (when)
108import Control.Monad.Catch (MonadThrow)
109import Control.Monad.Catch (MonadThrow)
@@ -346,6 +347,7 @@ data SVar t m a = SVar
346 -- Used only by bounded SVar types
347 -- Used only by bounded SVar types
347 , enqueue :: t m a -> IO ()
348 , enqueue :: t m a -> IO ()
348 , isWorkDone :: IO Bool
349 , isWorkDone :: IO Bool
350+ , isQueueDone :: IO Bool
349 , needDoorBell :: IORef Bool
351 , needDoorBell :: IORef Bool
350 , workLoop :: Maybe WorkerInfo -> m ()
352 , workLoop :: Maybe WorkerInfo -> m ()
@@ -359,7 +361,7 @@ data SVar t m a = SVar
359 -- to track garbage collection of SVar
361 -- to track garbage collection of SVar
360 , svarRef :: Maybe (IORef ())
362 , svarRef :: Maybe (IORef ())
361#ifdef DIAGNOSTICS
363#ifdef DIAGNOSTICS
362− , svarCreator :: ThreadId
364+ , svarCreator :: ThreadId
363 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
365 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
364 -- Shared work queue (stream, seqNo)
366 -- Shared work queue (stream, seqNo)
365 , aheadWorkQueue :: IORef ([t m a], Int)
367 , aheadWorkQueue :: IORef ([t m a], Int)
@@ -846,19 +848,40 @@ send sv msg = do
846 active <- readIORef (workerCount sv)
848 active <- readIORef (workerCount sv)
847 return $ len < ((fromIntegral lim) - active)
849 return $ len < ((fromIntegral lim) - active)
851+workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecs))
852+workerCollectLatency winfo = do
853+ (cnt0, t0) <- readIORef (workerLatencyStart winfo)
854+ cnt1 <- readIORef (workerYieldCount winfo)
855+ let cnt = cnt1 - cnt0
859+ t1 <- getTime Monotonic
860+ let period = fromInteger $ toNanoSecs (t1 - t0)
861+ writeIORef (workerLatencyStart winfo) (cnt1, t1)
862+ return $ Just (cnt, period)
865+-- XXX There are a number of gotchas in measuring latencies.
849−-- XXX We assume that a worker always yields a value. If we can have
866+-- 1) We measure latencies only when a worker yields a value
867+-- 2) It is possible that a stream calls the stop continuation, in which case
868+-- the worker would not yield a value and we would not account that worker in
869+-- latencies. Even though this case should ideally be accounted we do not
870+-- account it because we cannot or do not distinguish it from the case
850−-- workers that return without yielding anything our computations to
872+-- 3) It is possible that a worker returns without yielding anything because it
851−-- determine the number of workers may be off.
873+-- never got a chance to pick up work.
875+-- We can fix this if we measure the latencies by counting the work items
876+-- picked rather than based on the outputs yielded.
852workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
877workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
853workerUpdateLatency yinfo winfo = do
878workerUpdateLatency yinfo winfo = do
854− cnt1 <- readIORef (workerYieldCount winfo)
879+ r <- workerCollectLatency winfo
855− (cnt0, t0) <- readIORef (workerLatencyStart winfo)
880+ case r of
856− t1 <- getTime Monotonic
881+ Just (cnt, period) -> do
857− writeIORef (workerLatencyStart winfo) (cnt1, t1)
858− let period = fromInteger $ toNanoSecs (t1 - t0)
859− let ref = workerPendingLatency yinfo
882+ let ref = workerPendingLatency yinfo
860− atomicModifyIORefCAS ref $ \(ycnt, ytime) ->
883+ atomicModifyIORefCAS_ ref $ \(n, t) -> (n + cnt, t + period)
861− ((ycnt + cnt1 - cnt0, ytime + period), ())
884+ Nothing -> return ()
863updateYieldCount :: WorkerInfo -> IO Count
886updateYieldCount :: WorkerInfo -> IO Count
864updateYieldCount winfo = do
887updateYieldCount winfo = do
@@ -1262,37 +1285,43 @@ dispatchWorker yieldCount sv = do
1262 done <- liftIO $ isWorkDone sv
1285 done <- liftIO $ isWorkDone sv
1263 if (not done)
1286 if (not done)
1288+ qDone <- liftIO $ isQueueDone sv
1265 -- Note that the worker count is only decremented during event
1289 -- Note that the worker count is only decremented during event
1266 -- processing in fromStreamVar and therefore it is safe to read and
1290 -- processing in fromStreamVar and therefore it is safe to read and
1267 -- use it without a lock.
1291 -- use it without a lock.
1268 active <- liftIO $ readIORef $ workerCount sv
1292 active <- liftIO $ readIORef $ workerCount sv
1269− -- Note that we may deadlock if the previous workers (tasks in the
1295+ -- Note that we may deadlock if the previous workers (tasks in the
1270− -- stream) wait/depend on the future workers (tasks in the stream)
1296+ -- stream) wait/depend on the future workers (tasks in the stream)
1271− -- executing. In that case we should either configure the maxWorker
1297+ -- executing. In that case we should either configure the maxWorker
1272− -- count to higher or use parallel style instead of ahead or async
1298+ -- count to higher or use parallel style instead of ahead or async
1273− -- style.
1299+ -- style.
1274− limit <- case remainingYields sv of
1300+ limit <- case remainingYields sv of
1275− Nothing -> return workerLimit
1301+ Nothing -> return workerLimit
1276− Just ref -> do
1302+ Just ref -> do
1277− n <- liftIO $ readIORef ref
1303+ n <- liftIO $ readIORef ref
1278− return $
1304+ return $
1279− case workerLimit of
1305+ case workerLimit of
1280− Unlimited -> Limited (fromIntegral n)
1306+ Unlimited -> Limited (fromIntegral n)
1281− Limited lim -> Limited $ min lim (fromIntegral n)
1307+ Limited lim -> Limited $ min lim (fromIntegral n)
1283− -- XXX for ahead streams shall we take the heap yields into account for
1309+ -- XXX for ahead streams shall we take the heap yields into account for
1284− -- controlling the dispatch? We should not dispatch if the heap has
1310+ -- controlling the dispatch? We should not dispatch if the heap has
1285− -- already got the limit covered.
1311+ -- already got the limit covered.
1286− let dispatch = pushWorker yieldCount sv >> return True
1312+ let dispatch = pushWorker yieldCount sv >> return True
1287− in case limit of
1313+ in case limit of
1288− Unlimited -> dispatch
1314+ Unlimited -> dispatch
1289− -- Note that the use of remainingYields and workerCount is not
1315+ -- Note that the use of remainingYields and workerCount is not
1290− -- atomic and the counts may even have changed between reading and
1316+ -- atomic and the counts may even have changed between reading and
1291− -- using them here, so this is just approximate logic and we cannot
1317+ -- using them here, so this is just approximate logic and we cannot
1292− -- rely on it for correctness. We may actually dispatch more
1318+ -- rely on it for correctness. We may actually dispatch more
1293− -- workers than required.
1319+ -- workers than required.
1294− Limited lim | active < (fromIntegral lim) -> dispatch
1320+ Limited lim | active < (fromIntegral lim) -> dispatch
1295− _ -> return False
1321+ _ -> return False
1323+ when (active <= 0) $ pushWorker 0 sv
1296 else return False
1325 else return False
1298-- | This is a magic number and it is overloaded, and used at several places to
1327-- | This is a magic number and it is overloaded, and used at several places to
@@ -1646,6 +1675,9 @@ sendWorkerDelay :: SVar t m a -> IO ()
1646sendWorkerDelay sv = do
1675sendWorkerDelay sv = do
1647 -- XXX we need a better way to handle this than hardcoded delays. The
1676 -- XXX we need a better way to handle this than hardcoded delays. The
1648 -- delays may be different for different systems.
1677 -- delays may be different for different systems.
1678+ -- If there is a usecase where this is required we can create a combinator
1679+ -- to set it as a config in the state.
1649 ncpu <- getNumCapabilities
1681 ncpu <- getNumCapabilities
1650 if ncpu <= 1
1682 if ncpu <= 1
@@ -1656,6 +1688,8 @@ sendWorkerDelay sv = do
1656 if (svarStyle sv == AheadVar)
1688 if (svarStyle sv == AheadVar)
1657 then threadDelay 100
1689 then threadDelay 100
1658 else threadDelay 10
1690 else threadDelay 10
1660{-# NOINLINE sendWorkerWait #-}
1694{-# NOINLINE sendWorkerWait #-}
1661sendWorkerWait
1695sendWorkerWait
@@ -1919,6 +1953,7 @@ getAheadSVar st f = do
1919 , workLoop = f q outH st{streamVar = Just sv} sv
1953 , workLoop = f q outH st{streamVar = Just sv} sv
1920 , enqueue = enqueueAhead sv q
1954 , enqueue = enqueueAhead sv q
1921 , isWorkDone = isWorkDoneAhead sv q outH
1955 , isWorkDone = isWorkDoneAhead sv q outH
1956+ , isQueueDone = isQueueDoneAhead sv q
1922 , needDoorBell = wfw
1957 , needDoorBell = wfw
1923 , svarStyle = AheadVar
1958 , svarStyle = AheadVar
1924 , workerCount = active
1959 , workerCount = active
@@ -1951,11 +1986,8 @@ getAheadSVar st f = do
1954− {-# INLINE isWorkDoneAhead #-}
1989+ {-# INLINE isQueueDoneAhead #-}
1955− isWorkDoneAhead sv q ref = do
1990+ isQueueDoneAhead sv q = do
1957− (hp, _) <- readIORef ref
1958− return (H.size hp <= 0)
1959 queueDone <- checkEmpty q
1991 queueDone <- checkEmpty q
1960 yieldsDone <-
1992 yieldsDone <-
1961 case remainingYields sv of
1993 case remainingYields sv of
@@ -1966,7 +1998,15 @@ getAheadSVar st f = do
1966 -- XXX note that yieldsDone can only be authoritative only when there
1998 -- XXX note that yieldsDone can only be authoritative only when there
1967 -- are no workers running. If there are active workers they can
1999 -- are no workers running. If there are active workers they can
1968 -- later increment the yield count and therefore change the result.
2000 -- later increment the yield count and therefore change the result.
1969− return $ (yieldsDone && heapDone) || (queueDone && heapDone)
2001+ return $ yieldsDone || queueDone
2003+ {-# INLINE isWorkDoneAhead #-}
2004+ isWorkDoneAhead sv q ref = do
2006+ (hp, _) <- readIORef ref
2007+ return (H.size hp <= 0)
2008+ queueDone <- isQueueDoneAhead sv q
2009+ return $ heapDone && queueDone
1971 checkEmpty q = do
2011 checkEmpty q = do
1972 (xs, _) <- readIORef q
2012 (xs, _) <- readIORef q
@@ -2010,6 +2050,7 @@ getParallelSVar st = do
2010 , workLoop = undefined
2050 , workLoop = undefined
2011 , enqueue = undefined
2051 , enqueue = undefined
2012 , isWorkDone = undefined
2052 , isWorkDone = undefined
2053+ , isQueueDone = undefined
2013 , needDoorBell = undefined
2054 , needDoorBell = undefined
2014 , svarStyle = ParallelVar
2055 , svarStyle = ParallelVar
2015 , workerCount = active
2056 , workerCount = active