@@ -20,6 +20,10 @@
2020-- Portability : GHC
2121--
2222--
23+#ifdef DIAGNOSTICS_VERBOSE
24+#define DIAGNOSTICS
25+#endif
26+
2327module Streamly.SVar
2428 (
2529 MonadAsync
@@ -80,8 +84,9 @@ module Streamly.SVar
8084 , delThread
8185
8286 , toStreamVar
83−#ifdef DIAGNOSTICS
8487 , SVarStats (..)
88+ , NanoSecs (..)
89+#ifdef DIAGNOSTICS
8590 , dumpSVar
8691#endif
8792 )
@@ -127,7 +132,6 @@ import Control.Concurrent.MVar (tryTakeMVar)
127132import Control.Exception
128133 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
129134 BlockedIndefinitelyOnSTM(..))
130−import Data.Maybe (isJust)
131135import System.IO (hPutStrLn, stderr)
132136#endif
133137
@@ -264,18 +268,17 @@ data YieldRateInfo = YieldRateInfo
264268 , workerStopTimeStamp :: IORef TimeSpec
265269 }
266270
267−#ifdef DIAGNOSTICS
268271data SVarStats = SVarStats {
269− totalDispatches :: IORef Int
270− , maxWorkers :: IORef Int
271− , maxOutQSize :: IORef Int
272− , maxHeapSize :: IORef Int
273− , maxWorkQSize :: IORef Int
274− -- , avgLatency :: IORef NanoSecs
275− -- , minLatency :: IORef NanoSecs
276− -- , maxLatency :: IORef NanoSecs
272+ totalDispatches :: IORef Int
273+ , maxWorkers :: IORef Int
274+ , maxOutQSize :: IORef Int
275+ , maxHeapSize :: IORef Int
276+ , maxWorkQSize :: IORef Int
277+ , avgWorkerLatency :: IORef (Count, NanoSecs)
278+ , minWorkerLatency :: IORef NanoSecs
279+ , maxWorkerLatency :: IORef NanoSecs
280+ , svarStopTime :: IORef (Maybe TimeSpec)
277281}
278−#endif
279282
280283data Limit = Unlimited | Limited Word deriving Show
281284
@@ -308,13 +311,13 @@ data SVar t m a = SVar
308311 , accountThread :: ThreadId -> m ()
309312 , workerStopMVar :: MVar ()
310313
314+ , svarStats :: SVarStats
311315#ifdef DIAGNOSTICS
312316 -- to track garbage collection of SVar
313317 , svarRef :: Maybe (IORef ())
314318 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
315319 -- Shared work queue (stream, seqNo)
316320 , aheadWorkQueue :: IORef ([t m a], Int)
317− , svarStats :: SVarStats
318321#endif
319322 }
320323
@@ -456,20 +459,58 @@ getStreamLatency = _streamLatency
456459-------------------------------------------------------------------------------
457460
458461#ifdef DIAGNOSTICS
459−dumpSVarStats :: SVarStats -> SVarStyle -> Bool -> IO String
460−dumpSVarStats ss style _paced = do
462+dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
463+dumpSVarStats sv ss style = do
464+ case yieldRateInfo sv of
465+ Nothing -> return ()
466+ Just yinfo -> do
467+ _ <- liftIO $ collectLatency (svarStats sv) yinfo
468+ return ()
469+
461470 dispatches <- readIORef $ totalDispatches ss
462471 maxWrk <- readIORef $ maxWorkers ss
463472 maxOq <- readIORef $ maxOutQSize ss
464473 maxHp <- readIORef $ maxHeapSize ss
474+ minLat <- readIORef $ minWorkerLatency ss
475+ maxLat <- readIORef $ maxWorkerLatency ss
476+ (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
477+ svarLat <- case yieldRateInfo sv of
478+ Nothing -> return 0
479+ Just yinfo -> do
480+ (cnt, startTime) <- readIORef $ workerLongTermLatency yinfo
481+ if cnt > 0
482+ then do
483+ t <- readIORef (svarStopTime ss)
484+ case t of
485+ Nothing -> do
486+ now <- getTime Monotonic
487+ let interval = toNanoSecs (now - startTime)
488+ return $ interval `div` fromIntegral cnt
489+ Just stopTime -> do
490+ let interval = toNanoSecs (stopTime - startTime)
491+ return $ interval `div` fromIntegral cnt
492+ else return 0
465493
466494 return $ unlines
467495 [ "total dispatches = " ++ show dispatches
468496 , "max workers = " ++ show maxWrk
469497 , "max outQSize = " ++ show maxOq
470− , if style == AheadVar
471− then "heap max size = " ++ show maxHp
472− else ""
498+ ++ (if style == AheadVar
499+ then "\nheap max size = " ++ show maxHp
500+ else "")
501+ ++ (if minLat > 0
502+ then "\nmin worker latency = " ++ show minLat
503+ else "")
504+ ++ (if maxLat > 0
505+ then "\nmax worker latency = " ++ show maxLat
506+ else "")
507+ ++ (if avgCnt > 0
508+ then "\navg worker latency = "
509+ ++ show (avgTime `div` fromIntegral avgCnt)
510+ else "")
511+ ++ (if svarLat > 0
512+ then "\nSVar latency = " ++ show svarLat
513+ else "")
473514 ]
474515
475516{-# NOINLINE dumpSVar #-}
@@ -498,8 +539,7 @@ dumpSVar sv = do
498539 else return False
499540 rthread <- readIORef $ workerThreads sv
500541 workers <- readIORef $ workerCount sv
501− stats <- dumpSVarStats (svarStats sv) (svarStyle sv)
502− (isJust (yieldRateInfo sv))
542+ stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
503543
504544 return $ unlines
505545 [ "tid = " ++ show tid
@@ -630,6 +670,9 @@ incrementYieldLimit sv =
630670-- TBD Each worker can have their own queue and the consumer can empty one
631671-- queue at a time, that way contention can be reduced.
632672
673+-- XXX Only yields should be counted in the buffer limit and not the Stop
674+-- events.
675+
633676-- | This function is used by the producer threads to queue output for the
634677-- consumer thread to consume. Returns whether the queue has more space.
635678send :: SVar t m a -> ChildEvent a -> IO Bool
@@ -1126,7 +1169,7 @@ countDispatchWorkers workerLimit count duration wLatency expectedLat =
11261169 -- XXX we can have a maxEfficiency combinator as well which runs the
11271170 -- producer at the maximal efficiency i.e. the number of workers are chosen
11281171 -- such that the latency is minimum or within a range. Or we can call it
1129− -- maxLatency.
1172+ -- maxWorkerLatency.
11301173 --
11311174 let nWorkers = (fromIntegral wLatency) / (fromIntegral expectedLat)
11321175 -- Calculate how many yields are we ahead or behind to match the exact
@@ -1207,8 +1250,8 @@ useCollectedBatch yinfo latency colRef measRef = do
12071250-- period. The former two are used for accurate measurement of the going rate
12081251-- whereas the average is used for future estimates e.g. how many workers
12091252-- should be maintained to maintain the rate.
1210−collectLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1211−collectLatency yinfo = do
1253+collectLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1254+collectLatency _ss yinfo = do
12121255 let cur = workerCurrentLatency yinfo
12131256 col = workerCollectedLatency yinfo
12141257 longTerm = workerLongTermLatency yinfo
@@ -1228,6 +1271,14 @@ collectLatency yinfo = do
12281271 if (pendingCount > 0)
12291272 then do
12301273 let new = pendingTime `div` (fromIntegral pendingCount)
1274+#ifdef DIAGNOSTICS
1275+ minLat <- readIORef (minWorkerLatency _ss)
1276+ when (new < minLat || minLat == 0) $
1277+ writeIORef (minWorkerLatency _ss) new
1278+
1279+ maxLat <- readIORef (maxWorkerLatency _ss)
1280+ when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
1281+#endif
12311282 -- To avoid minor fluctuations update in batches
12321283 if (pendingCount > fromIntegral magicMaxBuffer)
12331284 || (pendingTime > minThreadDelay)
@@ -1236,6 +1287,10 @@ collectLatency yinfo = do
12361287 || (prev == 0)
12371288 then do
12381289 useCollectedBatch yinfo new col measured
1290+#ifdef DIAGNOSTICS
1291+ modifyIORef (avgWorkerLatency _ss) $
1292+ \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1293+#endif
12391294 modifyIORef longTerm $ \(_, t) -> (lcount', t)
12401295 return (lcount', ltime, new)
12411296 else do
@@ -1263,7 +1318,7 @@ dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
12631318dispatchWorkerPaced sv = do
12641319 let workerLimit = maxWorkerLimit sv
12651320 let yinfo = fromJust $ yieldRateInfo sv
1266− (count, tstamp, wLatency) <- liftIO $ collectLatency yinfo
1321+ (count, tstamp, wLatency) <- liftIO $ collectLatency (svarStats sv) yinfo
12671322 -- XXX use workerBootstrapLatency
12681323 if wLatency == 0
12691324 -- Need to measure the latency with a single worker before we can perform
@@ -1547,13 +1602,15 @@ getAheadSVar st f = do
15471602 Just x -> Just <$> newIORef x
15481603 rateInfo <- getYieldRateInfo st
15491604
1550−#ifdef DIAGNOSTICS
15511605 disp <- newIORef 0
15521606 maxWrk <- newIORef 0
15531607 maxOq <- newIORef 0
15541608 maxHs <- newIORef 0
15551609 maxWq <- newIORef 0
1556−#endif
1610+ avgLat <- newIORef (0, NanoSecs 0)
1611+ maxLat <- newIORef (NanoSecs 0)
1612+ minLat <- newIORef (NanoSecs 0)
1613+ stpTime <- newIORef Nothing
15571614
15581615 let getSVar sv readOutput postProc = SVar
15591616 { outputQueue = outQ
@@ -1577,14 +1634,18 @@ getAheadSVar st f = do
15771634 , svarRef = Nothing
15781635 , aheadWorkQueue = q
15791636 , outputHeap = outH
1637+#endif
15801638 , svarStats = SVarStats
15811639 { totalDispatches = disp
15821640 , maxWorkers = maxWrk
15831641 , maxOutQSize = maxOq
15841642 , maxHeapSize = maxHs
15851643 , maxWorkQSize = maxWq
1644+ , avgWorkerLatency = avgLat
1645+ , minWorkerLatency = minLat
1646+ , maxWorkerLatency = maxLat
1647+ , svarStopTime = stpTime
15861648 }
1587−#endif
15881649 }
15891650
15901651 let sv =
@@ -1622,13 +1683,17 @@ getParallelSVar = do
16221683 outQMv <- newEmptyMVar
16231684 active <- newIORef 0
16241685 running <- newIORef S.empty
1625−#ifdef DIAGNOSTICS
1686+
16261687 disp <- newIORef 0
16271688 maxWrk <- newIORef 0
16281689 maxOq <- newIORef 0
16291690 maxHs <- newIORef 0
16301691 maxWq <- newIORef 0
1631−#endif
1692+ avgLat <- newIORef (0, NanoSecs 0)
1693+ maxLat <- newIORef (NanoSecs 0)
1694+ minLat <- newIORef (NanoSecs 0)
1695+ stpTime <- newIORef Nothing
1696+
16321697 let sv =
16331698 SVar { outputQueue = outQ
16341699 , remainingYields = Nothing
@@ -1651,14 +1716,18 @@ getParallelSVar = do
16511716 , svarRef = Nothing
16521717 , aheadWorkQueue = undefined
16531718 , outputHeap = undefined
1719+#endif
16541720 , svarStats = SVarStats
16551721 { totalDispatches = disp
16561722 , maxWorkers = maxWrk
16571723 , maxOutQSize = maxOq
16581724 , maxHeapSize = maxHs
16591725 , maxWorkQSize = maxWq
1726+ , avgWorkerLatency = avgLat
1727+ , minWorkerLatency = minLat
1728+ , maxWorkerLatency = maxLat
1729+ , svarStopTime = stpTime
16601730 }
1661−#endif
16621731 }
16631732 in return sv
16641733