@@ -139,6 +139,7 @@ import Control.Exception
139 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
139 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
140 BlockedIndefinitelyOnSTM(..))
140 BlockedIndefinitelyOnSTM(..))
141import System.IO (hPutStrLn, stderr)
141import System.IO (hPutStrLn, stderr)
142+import Text.Printf (printf)
144newtype NanoSecs = NanoSecs Word64
145newtype NanoSecs = NanoSecs Word64
@@ -325,6 +326,7 @@ data SVar t m a = SVar
325 -- to track garbage collection of SVar
326 -- to track garbage collection of SVar
326 , svarRef :: Maybe (IORef ())
327 , svarRef :: Maybe (IORef ())
327#ifdef DIAGNOSTICS
328#ifdef DIAGNOSTICS
329+ , svarCreator :: ThreadId
328 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
330 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
329 -- Shared work queue (stream, seqNo)
331 -- Shared work queue (stream, seqNo)
330 , aheadWorkQueue :: IORef ([t m a], Int)
332 , aheadWorkQueue :: IORef ([t m a], Int)
@@ -486,12 +488,77 @@ cleanupSVarFromWorker sv = do
486-------------------------------------------------------------------------------
488-------------------------------------------------------------------------------
488#ifdef DIAGNOSTICS
490#ifdef DIAGNOSTICS
491+-- | Convert a number of seconds to a string. The string will consist
492+-- of four decimal places, followed by a short description of the time
494+secs :: Double -> String
496+ | k < 0 = '-' : secs (-k)
497+ | k >= 1 = k `with` "s"
498+ | k >= 1e-3 = (k*1e3) `with` "ms"
499+#ifdef mingw32_HOST_OS
500+ | k >= 1e-6 = (k*1e6) `with` "us"
502+ | k >= 1e-6 = (k*1e6) `with` "μs"
504+ | k >= 1e-9 = (k*1e9) `with` "ns"
505+ | k >= 1e-12 = (k*1e12) `with` "ps"
506+ | k >= 1e-15 = (k*1e15) `with` "fs"
507+ | k >= 1e-18 = (k*1e18) `with` "as"
508+ | otherwise = printf "%g s" k
509+ where with (t :: Double) (u :: String)
510+ | t >= 1e9 = printf "%.4g %s" t u
511+ | t >= 1e3 = printf "%.0f %s" t u
512+ | t >= 1e2 = printf "%.1f %s" t u
513+ | t >= 1e1 = printf "%.2f %s" t u
514+ | otherwise = printf "%.3f %s" t u
516+-- XXX Code duplicated from collectLatency
517+drainLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
518+drainLatency _ss yinfo = do
519+ let cur = workerCurrentLatency yinfo
520+ col = workerCollectedLatency yinfo
521+ longTerm = workerLongTermLatency yinfo
522+ measured = workerMeasuredLatency yinfo
524+ (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
525+ (colCount, colTime) <- readIORef col
526+ (lcount, ltime) <- readIORef longTerm
527+ prev <- readIORef measured
529+ let pendingCount = colCount + count
530+ pendingTime = colTime + time
532+ lcount' = lcount + pendingCount
533+ notUpdated = (lcount', ltime, prev)
535+ if (pendingCount > 0)
537+ let new = pendingTime `div` (fromIntegral pendingCount)
539+ minLat <- readIORef (minWorkerLatency _ss)
540+ when (new < minLat || minLat == 0) $
541+ writeIORef (minWorkerLatency _ss) new
543+ maxLat <- readIORef (maxWorkerLatency _ss)
544+ when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
546+ -- To avoid minor fluctuations update in batches
547+ useCollectedBatch yinfo new col measured
549+ modifyIORef (avgWorkerLatency _ss) $
550+ \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
552+ modifyIORef longTerm $ \(_, t) -> (lcount', t)
553+ return (lcount', ltime, new)
554+ else return notUpdated
489dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
556dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
490dumpSVarStats sv ss style = do
557dumpSVarStats sv ss style = do
491 case yieldRateInfo sv of
558 case yieldRateInfo sv of
492 Nothing -> return ()
559 Nothing -> return ()
493 Just yinfo -> do
560 Just yinfo -> do
494− _ <- liftIO $ collectLatency (svarStats sv) yinfo
561+ _ <- liftIO $ drainLatency (svarStats sv) yinfo
495 return ()
562 return ()
497 dispatches <- readIORef $ totalDispatches ss
564 dispatches <- readIORef $ totalDispatches ss
@@ -501,8 +568,8 @@ dumpSVarStats sv ss style = do
501 minLat <- readIORef $ minWorkerLatency ss
568 minLat <- readIORef $ minWorkerLatency ss
502 maxLat <- readIORef $ maxWorkerLatency ss
569 maxLat <- readIORef $ maxWorkerLatency ss
503 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
570 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
504− svarLat <- case yieldRateInfo sv of
571+ (svarCnt, svarLat) <- case yieldRateInfo sv of
505− Nothing -> return 0
572+ Nothing -> return (0, 0)
506 Just yinfo -> do
573 Just yinfo -> do
507 (cnt, startTime) <- readIORef $ workerLongTermLatency yinfo
574 (cnt, startTime) <- readIORef $ workerLongTermLatency yinfo
508 if cnt > 0
575 if cnt > 0
@@ -512,11 +579,11 @@ dumpSVarStats sv ss style = do
512 Nothing -> do
579 Nothing -> do
513 now <- getTime Monotonic
580 now <- getTime Monotonic
514 let interval = toNanoSecs (now - startTime)
581 let interval = toNanoSecs (now - startTime)
515− return $ interval `div` fromIntegral cnt
582+ return $ (cnt, interval `div` fromIntegral cnt)
516 Just stopTime -> do
583 Just stopTime -> do
517 let interval = toNanoSecs (stopTime - startTime)
584 let interval = toNanoSecs (stopTime - startTime)
518− return $ interval `div` fromIntegral cnt
585+ return $ (cnt, interval `div` fromIntegral cnt)
519− else return 0
586+ else return (0, 0)
521 return $ unlines
588 return $ unlines
522 [ "total dispatches = " ++ show dispatches
589 [ "total dispatches = " ++ show dispatches
@@ -526,24 +593,30 @@ dumpSVarStats sv ss style = do
526 then "\nheap max size = " ++ show maxHp
593 then "\nheap max size = " ++ show maxHp
528 ++ (if minLat > 0
595 ++ (if minLat > 0
529− then "\nmin worker latency = " ++ show minLat
596+ then "\nmin worker latency = "
597+ ++ secs (fromIntegral minLat * 1e-9)
531 ++ (if maxLat > 0
599 ++ (if maxLat > 0
532− then "\nmax worker latency = " ++ show maxLat
600+ then "\nmax worker latency = "
601+ ++ secs (fromIntegral maxLat * 1e-9)
534 ++ (if avgCnt > 0
603 ++ (if avgCnt > 0
604+ then let lat = avgTime `div` fromIntegral avgCnt
535− then "\navg worker latency = "
605+ in "\navg worker latency = "
536− ++ show (avgTime `div` fromIntegral avgCnt)
606+ ++ secs (fromIntegral lat * 1e-9)
538 ++ (if svarLat > 0
608 ++ (if svarLat > 0
539− then "\nSVar latency = " ++ show svarLat
609+ then "\nSVar latency = "
610+ ++ secs (fromIntegral svarLat * 1e-9)
613+ then "\nSVar yield count = " ++ show svarCnt
543{-# NOINLINE dumpSVar #-}
617{-# NOINLINE dumpSVar #-}
544dumpSVar :: SVar t m a -> IO String
618dumpSVar :: SVar t m a -> IO String
545dumpSVar sv = do
619dumpSVar sv = do
547 (oqList, oqLen) <- readIORef $ outputQueue sv
620 (oqList, oqLen) <- readIORef $ outputQueue sv
548 db <- tryTakeMVar $ outputDoorBell sv
621 db <- tryTakeMVar $ outputDoorBell sv
549 aheadDump <-
622 aheadDump <-
@@ -569,7 +642,7 @@ dumpSVar sv = do
569 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
642 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
571 return $ unlines
644 return $ unlines
572− [ "tid = " ++ show tid
645+ [ "Creator tid = " ++ show (svarCreator sv)
573 , "style = " ++ show (svarStyle sv)
646 , "style = " ++ show (svarStyle sv)
574 , "---------CURRENT STATE-----------"
647 , "---------CURRENT STATE-----------"
575 , "outputQueue length computed = " ++ show (length oqList)
648 , "outputQueue length computed = " ++ show (length oqList)
@@ -1654,6 +1727,9 @@ getAheadSVar st f = do
1654 maxLat <- newIORef (NanoSecs 0)
1727 maxLat <- newIORef (NanoSecs 0)
1655 minLat <- newIORef (NanoSecs 0)
1728 minLat <- newIORef (NanoSecs 0)
1656 stpTime <- newIORef Nothing
1729 stpTime <- newIORef Nothing
1658 let getSVar sv readOutput postProc = SVar
1734 let getSVar sv readOutput postProc = SVar
1659 { outputQueue = outQ
1735 { outputQueue = outQ
@@ -1675,6 +1751,7 @@ getAheadSVar st f = do
1675 , workerStopMVar = stopMVar
1751 , workerStopMVar = stopMVar
1676 , svarRef = Nothing
1752 , svarRef = Nothing
1677#ifdef DIAGNOSTICS
1753#ifdef DIAGNOSTICS
1754+ , svarCreator = tid
1678 , aheadWorkQueue = q
1755 , aheadWorkQueue = q
1679 , outputHeap = outH
1756 , outputHeap = outH
@@ -1739,6 +1816,9 @@ getParallelSVar st = do
1739 maxLat <- newIORef (NanoSecs 0)
1816 maxLat <- newIORef (NanoSecs 0)
1740 minLat <- newIORef (NanoSecs 0)
1817 minLat <- newIORef (NanoSecs 0)
1741 stpTime <- newIORef Nothing
1818 stpTime <- newIORef Nothing
1743 let sv =
1823 let sv =
1744 SVar { outputQueue = outQ
1824 SVar { outputQueue = outQ
@@ -1760,6 +1840,7 @@ getParallelSVar st = do
1760 , workerStopMVar = undefined
1840 , workerStopMVar = undefined
1761 , svarRef = Nothing
1841 , svarRef = Nothing
1762#ifdef DIAGNOSTICS
1842#ifdef DIAGNOSTICS
1843+ , svarCreator = tid
1763 , aheadWorkQueue = undefined
1844 , aheadWorkQueue = undefined
1764 , outputHeap = undefined
1845 , outputHeap = undefined