@@ -282,17 +282,20 @@ data YieldRateInfo = YieldRateInfo
282282
283283 -- This is in progress latency stats maintained by the workers which we
284284 -- empty into workerCollectedLatency stats at certain intervals - whenever
285− -- we process the stream elements yielded in this period.
286− -- (yieldCount, timeTaken)
287− , workerPendingLatency :: IORef (Count, NanoSecond64)
285+ -- we process the stream elements yielded in this period. The first count
286+ -- is all yields, the second count is only those yields for which the
287+ -- latency was measured to be non-zero (note that if the timer resolution
288+ -- is low the measured latency may be zero e.g. on JS platform).
289+ -- (allYieldCount, yieldCount, timeTaken)
290+ , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
288291
289292 -- This is the second level stat which is an accmulation from
290293 -- workerPendingLatency stats. We keep accumulating latencies in this
291294 -- bucket until we have stats for a sufficient period and then we reset it
292295 -- to start collecting for the next period and retain the computed average
293296 -- latency for the last period in workerMeasuredLatency.
294− -- (yieldCount, timeTaken)
295− , workerCollectedLatency :: IORef (Count, NanoSecond64)
297+ -- (allYieldCount, yieldCount, timeTaken)
298+ , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
296299
297300 -- Latency as measured by workers, aggregated for the last period.
298301 , workerMeasuredLatency :: IORef NanoSecond64
@@ -518,6 +521,124 @@ cleanupSVarFromWorker sv = do
518521 mapM_ (`throwTo` ThreadAbort)
519522 (S.toList workers \\ [self])
520523
524+-------------------------------------------------------------------------------
525+-- Worker latency data collection
526+-------------------------------------------------------------------------------
527+
528+-- Every once in a while workers update the latencies and check the yield rate.
529+-- They return if we are above the expected yield rate. If we check too often
530+-- it may impact performance, if we check less often we may have a stale
531+-- picture. We update every minThreadDelay but we translate that into a yield
532+-- count based on latency so that the checking overhead is little.
533+--
534+-- XXX use a generation count to indicate that the value is updated. If the
535+-- value is updated an existing worker must check it again on the next yield.
536+-- Otherwise it is possible that we may keep updating it and because of the mod
537+-- worker keeps skipping it.
538+updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
539+updateWorkerPollingInterval yinfo latency = do
540+ let periodRef = workerPollingInterval yinfo
541+ cnt = max 1 $ minThreadDelay `div` latency
542+ period = min cnt (fromIntegral magicMaxBuffer)
543+
544+ writeIORef periodRef (fromIntegral period)
545+
546+{-# INLINE recordMinMaxLatency #-}
547+recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
548+recordMinMaxLatency sv new = do
549+ let ss = svarStats sv
550+ minLat <- readIORef (minWorkerLatency ss)
551+ when (new < minLat || minLat == 0) $
552+ writeIORef (minWorkerLatency ss) new
553+
554+ maxLat <- readIORef (maxWorkerLatency ss)
555+ when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
556+
557+recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
558+recordAvgLatency sv (count, time) = do
559+ let ss = svarStats sv
560+ modifyIORef (avgWorkerLatency ss) $
561+ \(cnt, t) -> (cnt + count, t + time)
562+
563+-- Pour the pending latency stats into a collection bucket
564+{-# INLINE collectWorkerPendingLatency #-}
565+collectWorkerPendingLatency
566+ :: IORef (Count, Count, NanoSecond64)
567+ -> IORef (Count, Count, NanoSecond64)
568+ -> IO (Count, Maybe (Count, NanoSecond64))
569+collectWorkerPendingLatency cur col = do
570+ (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
571+
572+ (fcnt, cnt, t) <- readIORef col
573+ let totalCount = fcnt + fcount
574+ latCount = cnt + count
575+ latTime = t + time
576+ writeIORef col (totalCount, latCount, latTime)
577+
578+ assert (latCount == 0 || latTime /= 0) (return ())
579+ let latPair =
580+ if latCount > 0 && latTime > 0
581+ then Just $ (latCount, latTime)
582+ else Nothing
583+ return (totalCount, latPair)
584+
585+{-# INLINE shouldUseCollectedBatch #-}
586+shouldUseCollectedBatch
587+ :: Count
588+ -> NanoSecond64
589+ -> NanoSecond64
590+ -> NanoSecond64
591+ -> Bool
592+shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
593+ let r = fromIntegral newLat / fromIntegral prevLat :: Double
594+ in (collectedYields > fromIntegral magicMaxBuffer)
595+ || (collectedTime > minThreadDelay)
596+ || (prevLat > 0 && (r > 2 || r < 0.5))
597+ || (prevLat == 0)
598+
599+-- Returns a triple, (1) yield count since last collection, (2) the base time
600+-- when we started counting, (3) average latency in the last measurement
601+-- period. The former two are used for accurate measurement of the going rate
602+-- whereas the average is used for future estimates e.g. how many workers
603+-- should be maintained to maintain the rate.
604+-- CAUTION! keep it in sync with getWorkerLatency
605+collectLatency :: SVar t m a
606+ -> YieldRateInfo
607+ -> Bool
608+ -> IO (Count, AbsTime, NanoSecond64)
609+collectLatency sv yinfo drain = do
610+ let cur = workerPendingLatency yinfo
611+ col = workerCollectedLatency yinfo
612+ longTerm = svarAllTimeLatency yinfo
613+ measured = workerMeasuredLatency yinfo
614+
615+ (newCount, newLatPair) <- collectWorkerPendingLatency cur col
616+ (lcount, ltime) <- readIORef longTerm
617+ prevLat <- readIORef measured
618+
619+ let newLcount = lcount + newCount
620+ retWith lat = return (newLcount, ltime, lat)
621+
622+ case newLatPair of
623+ Nothing -> retWith prevLat
624+ Just (count, time) -> do
625+ let newLat = time `div` (fromIntegral count)
626+ when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
627+ -- When we have collected a significant sized batch we compute the
628+ -- new latency using that batch and return the new latency,
629+ -- otherwise we return the previous latency derived from the
630+ -- previous batch.
631+ if shouldUseCollectedBatch newCount time newLat prevLat || drain
632+ then do
633+ -- XXX make this NOINLINE?
634+ updateWorkerPollingInterval yinfo (max newLat prevLat)
635+ when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
636+ writeIORef col (0, 0, 0)
637+ writeIORef measured ((prevLat + newLat) `div` 2)
638+ modifyIORef longTerm $ \(_, t) -> (newLcount, t)
639+ retWith newLat
640+ else retWith prevLat
641+
521642-------------------------------------------------------------------------------
522643-- Dumping the SVar for debug/diag
523644-------------------------------------------------------------------------------
@@ -547,51 +668,12 @@ secs k
547668 | t >= 1e1 = printf "%.2f %s" t u
548669 | otherwise = printf "%.3f %s" t u
549670
550−-- XXX Code duplicated from collectLatency
551−drainLatency :: SVar t m a -> YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
552−drainLatency sv yinfo = do
553− let cur = workerPendingLatency yinfo
554− col = workerCollectedLatency yinfo
555− longTerm = svarAllTimeLatency yinfo
556− measured = workerMeasuredLatency yinfo
557−
558− (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
559− (colCount, colTime) <- readIORef col
560− (lcount, ltime) <- readIORef longTerm
561− prev <- readIORef measured
562−
563− let pendingCount = colCount + count
564− pendingTime = colTime + time
565−
566− lcount' = lcount + pendingCount
567− notUpdated = (lcount', ltime, prev)
568−
569− if (pendingCount > 0)
570− then do
571− let new = pendingTime `div` (fromIntegral pendingCount)
572− when (svarInspectMode sv) $ do
573− let ss = svarStats sv
574− minLat <- readIORef (minWorkerLatency ss)
575− when (new < minLat || minLat == 0) $
576− writeIORef (minWorkerLatency ss) new
577−
578− maxLat <- readIORef (maxWorkerLatency ss)
579− when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
580− modifyIORef (avgWorkerLatency ss) $
581− \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
582− -- To avoid minor fluctuations update in batches
583− writeIORef col (0, 0)
584− writeIORef measured new
585− modifyIORef longTerm $ \(_, t) -> (lcount', t)
586− return (lcount', ltime, new)
587− else return notUpdated
588−
589671dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
590672dumpSVarStats sv ss style = do
591673 case yieldRateInfo sv of
592674 Nothing -> return ()
593675 Just yinfo -> do
594− _ <- liftIO $ drainLatency sv yinfo
676+ _ <- liftIO $ collectLatency sv yinfo True
595677 return ()
596678
597679 dispatches <- readIORef $ totalDispatches ss
@@ -889,6 +971,8 @@ workerCollectLatency winfo = do
889971-- described next.
890972-- 3) It is possible that a worker returns without yielding anything because it
891973-- never got a chance to pick up work.
974+-- 4) If the system timer resolution is lower than the latency, the latency
975+-- computation turns out to be zero.
892976--
893977-- We can fix this if we measure the latencies by counting the work items
894978-- picked rather than based on the outputs yielded.
@@ -897,8 +981,16 @@ workerUpdateLatency yinfo winfo = do
897981 r <- workerCollectLatency winfo
898982 case r of
899983 Just (cnt, period) -> do
984+ -- NOTE: On JS platform the timer resolution could be pretty low. When
985+ -- the timer resolution is low, measurement of latencies could be
986+ -- tricky. All the worker latencies will turn out to be zero if they
987+ -- are lower than the resolution. We only take into account those
988+ -- measurements which are more than the timer resolution.
989+
900990 let ref = workerPendingLatency yinfo
901− atomicModifyIORefCAS_ ref $ \(n, t) -> (n + cnt, t + period)
991+ (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
992+ atomicModifyIORefCAS_ ref $
993+ \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
902994 Nothing -> return ()
903995
904996updateYieldCount :: WorkerInfo -> IO Count
@@ -1499,20 +1591,21 @@ getWorkerLatency yinfo = do
14991591 longTerm = svarAllTimeLatency yinfo
15001592 measured = workerMeasuredLatency yinfo
15011593
1502− (count, time) <- readIORef cur
1503− (colCount, colTime) <- readIORef col
1594+ (curTotalCount, curCount, curTime) <- readIORef cur
1595+ (colTotalCount, colCount, colTime) <- readIORef col
15041596 (lcount, ltime) <- readIORef longTerm
1505− prev <- readIORef measured
1506−
1507− let pendingCount = colCount + count
1508− pendingTime = colTime + time
1509− new =
1510− if pendingCount > 0
1511− then let lat = pendingTime `div` fromIntegral pendingCount
1597+ prevLat <- readIORef measured
1598+
1599+ let latCount = colCount + curCount
1600+ latTime = colTime + curTime
1601+ totalCount = colTotalCount + curTotalCount
1602+ newLat =
1603+ if latCount > 0 && latTime > 0
1604+ then let lat = latTime `div` fromIntegral latCount
15121605 -- XXX Give more weight to new?
1513− in (lat + prev) `div` 2
1514− else prev
1515− return (lcount + pendingCount, ltime, new)
1606+ in (lat + prevLat) `div` 2
1607+ else prevLat
1608+ return (lcount + totalCount, ltime, newLat)
15161609
15171610isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
15181611isBeyondMaxRate sv yinfo = do
@@ -1530,82 +1623,6 @@ isBeyondMaxRate sv yinfo = do
15301623 ManyWorkers n _ -> cnt > n
15311624 BlockWait _ -> True
15321625
1533−-- Every once in a while workers update the latencies and check the yield rate.
1534−-- They return if we are above the expected yield rate. If we check too often
1535−-- it may impact performance, if we check less often we may have a stale
1536−-- picture. We update every minThreadDelay but we translate that into a yield
1537−-- count based on latency so that the checking overhead is little.
1538−--
1539−-- XXX use a generation count to indicate that the value is updated. If the
1540−-- value is updated an existing worker must check it again on the next yield.
1541−-- Otherwise it is possible that we may keep updating it and because of the mod
1542−-- worker keeps skipping it.
1543−updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
1544−updateWorkerPollingInterval yinfo latency = do
1545− let periodRef = workerPollingInterval yinfo
1546− cnt = max 1 $ minThreadDelay `div` latency
1547− period = min cnt (fromIntegral magicMaxBuffer)
1548−
1549− writeIORef periodRef (fromIntegral period)
1550−
1551−-- Returns a triple, (1) yield count since last collection, (2) the base time
1552−-- when we started counting, (3) average latency in the last measurement
1553−-- period. The former two are used for accurate measurement of the going rate
1554−-- whereas the average is used for future estimates e.g. how many workers
1555−-- should be maintained to maintain the rate.
1556−-- CAUTION! keep it in sync with getWorkerLatency
1557−collectLatency :: SVar t m a -> YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1558−collectLatency sv yinfo = do
1559− let cur = workerPendingLatency yinfo
1560− col = workerCollectedLatency yinfo
1561− longTerm = svarAllTimeLatency yinfo
1562− measured = workerMeasuredLatency yinfo
1563−
1564− (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
1565− (colCount, colTime) <- readIORef col
1566− (lcount, ltime) <- readIORef longTerm
1567− prev <- readIORef measured
1568−
1569− let pendingCount = colCount + count
1570− pendingTime = colTime + time
1571−
1572− lcount' = lcount + pendingCount
1573− tripleWith lat = (lcount', ltime, lat)
1574−
1575− if pendingCount > 0
1576− then do
1577− let new = pendingTime `div` (fromIntegral pendingCount)
1578− when (svarInspectMode sv) $ do
1579− let ss = svarStats sv
1580− minLat <- readIORef (minWorkerLatency ss)
1581− when (new < minLat || minLat == 0) $
1582− writeIORef (minWorkerLatency ss) new
1583−
1584− maxLat <- readIORef (maxWorkerLatency ss)
1585− when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
1586− -- When we have collected a significant sized batch we compute the new
1587− -- latency using that batch and return the new latency, otherwise we
1588− -- return the previous latency derived from the previous batch.
1589− if (pendingCount > fromIntegral magicMaxBuffer)
1590− || (pendingTime > minThreadDelay)
1591− || (let r = fromIntegral new / fromIntegral prev :: Double
1592− in prev > 0 && (r > 2 || r < 0.5))
1593− || (prev == 0)
1594− then do
1595− when (svarInspectMode sv) $ do
1596− let ss = svarStats sv
1597− modifyIORef (avgWorkerLatency ss) $
1598− \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1599− updateWorkerPollingInterval yinfo (max new prev)
1600− writeIORef col (0, 0)
1601− writeIORef measured ((prev + new) `div` 2)
1602− modifyIORef longTerm $ \(_, t) -> (lcount', t)
1603− return $ tripleWith new
1604− else do
1605− writeIORef col (pendingCount, pendingTime)
1606− return $ tripleWith prev
1607− else return $ tripleWith prev
1608−
16091626-- XXX in case of ahead style stream we need to take the heap size into account
16101627-- because we return the workers on the basis of that which causes a condition
16111628-- where we keep dispatching and they keep returning. So we must have exactly
@@ -1620,7 +1637,7 @@ dispatchWorkerPaced sv = do
16201637 (svarYields, svarElapsed, wLatency) <- do
16211638 now <- liftIO $ getTime Monotonic
16221639 (yieldCount, baseTime, lat) <-
1623− liftIO $ collectLatency sv yinfo
1640+ liftIO $ collectLatency sv yinfo False
16241641 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
16251642 let latency =
16261643 if lat == 0
@@ -1938,8 +1955,8 @@ getYieldRateInfo st = do
19381955
19391956 mkYieldRateInfo latency latRange buf = do
19401957 measured <- newIORef 0
1941− wcur <- newIORef (0,0)
1942− wcol <- newIORef (0,0)
1958+ wcur <- newIORef (0,0,0)
1959+ wcol <- newIORef (0,0,0)
19431960 now <- getTime Monotonic
19441961 wlong <- newIORef (0,now)
19451962 period <- newIORef 1
@@ -2128,7 +2145,7 @@ getParallelSVar st mrun = do
21282145 $ takeMVar (outputDoorBell sv)
21292146 case yieldRateInfo sv of
21302147 Nothing -> return ()
2131− Just yinfo -> void $ collectLatency sv yinfo
2148+ Just yinfo -> void $ collectLatency sv yinfo False
21322149 fst `fmap` readOutputQRaw sv
21332150
21342151sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)