@@ -102,7 +102,8 @@ where
102102import Control.Concurrent
103103 (ThreadId, myThreadId, threadDelay, getNumCapabilities, throwTo)
104104import Control.Concurrent.MVar
105− (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)
105+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, readMVar,
106+ modifyMVar, modifyMVar_, withMVar)
106107import Control.Exception (SomeException(..), catch, mask, assert, Exception)
107108import Control.Monad (when)
108109import Control.Monad.Catch (MonadThrow)
@@ -346,6 +347,7 @@ data SVar t m a = SVar
346347 -- Used only by bounded SVar types
347348 , enqueue :: t m a -> IO ()
348349 , isWorkDone :: IO Bool
350+ , isQueueDone :: IO Bool
349351 , needDoorBell :: IORef Bool
350352 , workLoop :: Maybe WorkerInfo -> m ()
351353
@@ -359,7 +361,7 @@ data SVar t m a = SVar
359361 -- to track garbage collection of SVar
360362 , svarRef :: Maybe (IORef ())
361363#ifdef DIAGNOSTICS
362− , svarCreator :: ThreadId
364+ , svarCreator :: ThreadId
363365 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
364366 -- Shared work queue (stream, seqNo)
365367 , aheadWorkQueue :: IORef ([t m a], Int)
@@ -846,19 +848,40 @@ send sv msg = do
846848 active <- readIORef (workerCount sv)
847849 return $ len < ((fromIntegral lim) - active)
848850
849−-- XXX We assume that a worker always yields a value. If we can have
850−-- workers that return without yielding anything our computations to
851−-- determine the number of workers may be off.
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
856+
857+ if (cnt > 0)
858+ then do
859+ t1 <- getTime Monotonic
860+ let period = fromInteger $ toNanoSecs (t1 - t0)
861+ writeIORef (workerLatencyStart winfo) (cnt1, t1)
862+ return $ Just (cnt, period)
863+ else return Nothing
864+
865+-- XXX There are a number of gotchas in measuring latencies.
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
871+-- described next.
872+-- 3) It is possible that a worker returns without yielding anything because it
873+-- never got a chance to pick up work.
874+--
875+-- We can fix this if we measure the latencies by counting the work items
876+-- picked rather than based on the outputs yielded.
852877workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
853878workerUpdateLatency yinfo winfo = do
854− cnt1 <- readIORef (workerYieldCount winfo)
855− (cnt0, t0) <- readIORef (workerLatencyStart winfo)
856− t1 <- getTime Monotonic
857− writeIORef (workerLatencyStart winfo) (cnt1, t1)
858− let period = fromInteger $ toNanoSecs (t1 - t0)
859− let ref = workerPendingLatency yinfo
860− atomicModifyIORefCAS ref $ \(ycnt, ytime) ->
861− ((ycnt + cnt1 - cnt0, ytime + period), ())
879+ r <- workerCollectLatency winfo
880+ case r of
881+ Just (cnt, period) -> do
882+ let ref = workerPendingLatency yinfo
883+ atomicModifyIORefCAS_ ref $ \(n, t) -> (n + cnt, t + period)
884+ Nothing -> return ()
862885
863886updateYieldCount :: WorkerInfo -> IO Count
864887updateYieldCount winfo = do
@@ -1262,37 +1285,43 @@ dispatchWorker yieldCount sv = do
12621285 done <- liftIO $ isWorkDone sv
12631286 if (not done)
12641287 then do
1288+ qDone <- liftIO $ isQueueDone sv
12651289 -- Note that the worker count is only decremented during event
12661290 -- processing in fromStreamVar and therefore it is safe to read and
12671291 -- use it without a lock.
12681292 active <- liftIO $ readIORef $ workerCount sv
1269− -- Note that we may deadlock if the previous workers (tasks in the
1270− -- stream) wait/depend on the future workers (tasks in the stream)
1271− -- executing. In that case we should either configure the maxWorker
1272− -- count to higher or use parallel style instead of ahead or async
1273− -- style.
1274− limit <- case remainingYields sv of
1275− Nothing -> return workerLimit
1276− Just ref -> do
1277− n <- liftIO $ readIORef ref
1278− return $
1279− case workerLimit of
1280− Unlimited -> Limited (fromIntegral n)
1281− Limited lim -> Limited $ min lim (fromIntegral n)
1282−
1283− -- 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
1285− -- already got the limit covered.
1286− let dispatch = pushWorker yieldCount sv >> return True
1287− in case limit of
1288− Unlimited -> dispatch
1289− -- Note that the use of remainingYields and workerCount is not
1290− -- atomic and the counts may even have changed between reading and
1291− -- using them here, so this is just approximate logic and we cannot
1292− -- rely on it for correctness. We may actually dispatch more
1293− -- workers than required.
1294− Limited lim | active < (fromIntegral lim) -> dispatch
1295− _ -> return False
1293+ if (not qDone)
1294+ then do
1295+ -- Note that we may deadlock if the previous workers (tasks in the
1296+ -- stream) wait/depend on the future workers (tasks in the stream)
1297+ -- executing. In that case we should either configure the maxWorker
1298+ -- count to higher or use parallel style instead of ahead or async
1299+ -- style.
1300+ limit <- case remainingYields sv of
1301+ Nothing -> return workerLimit
1302+ Just ref -> do
1303+ n <- liftIO $ readIORef ref
1304+ return $
1305+ case workerLimit of
1306+ Unlimited -> Limited (fromIntegral n)
1307+ Limited lim -> Limited $ min lim (fromIntegral n)
1308+
1309+ -- XXX for ahead streams shall we take the heap yields into account for
1310+ -- controlling the dispatch? We should not dispatch if the heap has
1311+ -- already got the limit covered.
1312+ let dispatch = pushWorker yieldCount sv >> return True
1313+ in case limit of
1314+ Unlimited -> dispatch
1315+ -- Note that the use of remainingYields and workerCount is not
1316+ -- atomic and the counts may even have changed between reading and
1317+ -- using them here, so this is just approximate logic and we cannot
1318+ -- rely on it for correctness. We may actually dispatch more
1319+ -- workers than required.
1320+ Limited lim | active < (fromIntegral lim) -> dispatch
1321+ _ -> return False
1322+ else do
1323+ when (active <= 0) $ pushWorker 0 sv
1324+ return False
12961325 else return False
12971326
12981327-- | 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 ()
16461675sendWorkerDelay sv = do
16471676 -- XXX we need a better way to handle this than hardcoded delays. The
16481677 -- 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.
1680+ {-
16491681 ncpu <- getNumCapabilities
16501682 if ncpu <= 1
16511683 then
@@ -1656,6 +1688,8 @@ sendWorkerDelay sv = do
16561688 if (svarStyle sv == AheadVar)
16571689 then threadDelay 100
16581690 else threadDelay 10
1691+ -}
1692+ return ()
16591693
16601694{-# NOINLINE sendWorkerWait #-}
16611695sendWorkerWait
@@ -1919,6 +1953,7 @@ getAheadSVar st f = do
19191953 , workLoop = f q outH st{streamVar = Just sv} sv
19201954 , enqueue = enqueueAhead sv q
19211955 , isWorkDone = isWorkDoneAhead sv q outH
1956+ , isQueueDone = isQueueDoneAhead sv q
19221957 , needDoorBell = wfw
19231958 , svarStyle = AheadVar
19241959 , workerCount = active
@@ -1951,11 +1986,8 @@ getAheadSVar st f = do
19511986
19521987 where
19531988
1954− {-# INLINE isWorkDoneAhead #-}
1955− isWorkDoneAhead sv q ref = do
1956− heapDone <- do
1957− (hp, _) <- readIORef ref
1958− return (H.size hp <= 0)
1989+ {-# INLINE isQueueDoneAhead #-}
1990+ isQueueDoneAhead sv q = do
19591991 queueDone <- checkEmpty q
19601992 yieldsDone <-
19611993 case remainingYields sv of
@@ -1966,7 +1998,15 @@ getAheadSVar st f = do
19661998 -- XXX note that yieldsDone can only be authoritative only when there
19671999 -- are no workers running. If there are active workers they can
19682000 -- later increment the yield count and therefore change the result.
1969− return $ (yieldsDone && heapDone) || (queueDone && heapDone)
2001+ return $ yieldsDone || queueDone
2002+
2003+ {-# INLINE isWorkDoneAhead #-}
2004+ isWorkDoneAhead sv q ref = do
2005+ heapDone <- do
2006+ (hp, _) <- readIORef ref
2007+ return (H.size hp <= 0)
2008+ queueDone <- isQueueDoneAhead sv q
2009+ return $ heapDone && queueDone
19702010
19712011 checkEmpty q = do
19722012 (xs, _) <- readIORef q
@@ -2010,6 +2050,7 @@ getParallelSVar st = do
20102050 , workLoop = undefined
20112051 , enqueue = undefined
20122052 , isWorkDone = undefined
2053+ , isQueueDone = undefined
20132054 , needDoorBell = undefined
20142055 , svarStyle = ParallelVar
20152056 , workerCount = active