@@ -139,6 +139,7 @@ import Control.Exception
139139 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
140140 BlockedIndefinitelyOnSTM(..))
141141import System.IO (hPutStrLn, stderr)
142+import Text.Printf (printf)
142143#endif
143144
144145newtype NanoSecs = NanoSecs Word64
@@ -325,6 +326,7 @@ data SVar t m a = SVar
325326 -- to track garbage collection of SVar
326327 , svarRef :: Maybe (IORef ())
327328#ifdef DIAGNOSTICS
329+ , svarCreator :: ThreadId
328330 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)
329331 -- Shared work queue (stream, seqNo)
330332 , aheadWorkQueue :: IORef ([t m a], Int)
@@ -486,12 +488,77 @@ cleanupSVarFromWorker sv = do
486488-------------------------------------------------------------------------------
487489
488490#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
493+-- units.
494+secs :: Double -> String
495+secs k
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"
501+#else
502+ | k >= 1e-6 = (k*1e6) `with` "μs"
503+#endif
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
515+
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
523+
524+ (count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0), v)
525+ (colCount, colTime) <- readIORef col
526+ (lcount, ltime) <- readIORef longTerm
527+ prev <- readIORef measured
528+
529+ let pendingCount = colCount + count
530+ pendingTime = colTime + time
531+
532+ lcount' = lcount + pendingCount
533+ notUpdated = (lcount', ltime, prev)
534+
535+ if (pendingCount > 0)
536+ then do
537+ let new = pendingTime `div` (fromIntegral pendingCount)
538+#ifdef DIAGNOSTICS
539+ minLat <- readIORef (minWorkerLatency _ss)
540+ when (new < minLat || minLat == 0) $
541+ writeIORef (minWorkerLatency _ss) new
542+
543+ maxLat <- readIORef (maxWorkerLatency _ss)
544+ when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
545+#endif
546+ -- To avoid minor fluctuations update in batches
547+ useCollectedBatch yinfo new col measured
548+#ifdef DIAGNOSTICS
549+ modifyIORef (avgWorkerLatency _ss) $
550+ \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
551+#endif
552+ modifyIORef longTerm $ \(_, t) -> (lcount', t)
553+ return (lcount', ltime, new)
554+ else return notUpdated
555+
489556dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
490557dumpSVarStats sv ss style = do
491558 case yieldRateInfo sv of
492559 Nothing -> return ()
493560 Just yinfo -> do
494− _ <- liftIO $ collectLatency (svarStats sv) yinfo
561+ _ <- liftIO $ drainLatency (svarStats sv) yinfo
495562 return ()
496563
497564 dispatches <- readIORef $ totalDispatches ss
@@ -501,8 +568,8 @@ dumpSVarStats sv ss style = do
501568 minLat <- readIORef $ minWorkerLatency ss
502569 maxLat <- readIORef $ maxWorkerLatency ss
503570 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
504− svarLat <- case yieldRateInfo sv of
505− Nothing -> return 0
571+ (svarCnt, svarLat) <- case yieldRateInfo sv of
572+ Nothing -> return (0, 0)
506573 Just yinfo -> do
507574 (cnt, startTime) <- readIORef $ workerLongTermLatency yinfo
508575 if cnt > 0
@@ -512,11 +579,11 @@ dumpSVarStats sv ss style = do
512579 Nothing -> do
513580 now <- getTime Monotonic
514581 let interval = toNanoSecs (now - startTime)
515− return $ interval `div` fromIntegral cnt
582+ return $ (cnt, interval `div` fromIntegral cnt)
516583 Just stopTime -> do
517584 let interval = toNanoSecs (stopTime - startTime)
518− return $ interval `div` fromIntegral cnt
519− else return 0
585+ return $ (cnt, interval `div` fromIntegral cnt)
586+ else return (0, 0)
520587
521588 return $ unlines
522589 [ "total dispatches = " ++ show dispatches
@@ -526,24 +593,30 @@ dumpSVarStats sv ss style = do
526593 then "\nheap max size = " ++ show maxHp
527594 else "")
528595 ++ (if minLat > 0
529− then "\nmin worker latency = " ++ show minLat
596+ then "\nmin worker latency = "
597+ ++ secs (fromIntegral minLat * 1e-9)
530598 else "")
531599 ++ (if maxLat > 0
532− then "\nmax worker latency = " ++ show maxLat
600+ then "\nmax worker latency = "
601+ ++ secs (fromIntegral maxLat * 1e-9)
533602 else "")
534603 ++ (if avgCnt > 0
535− then "\navg worker latency = "
536− ++ show (avgTime `div` fromIntegral avgCnt)
604+ then let lat = avgTime `div` fromIntegral avgCnt
605+ in "\navg worker latency = "
606+ ++ secs (fromIntegral lat * 1e-9)
537607 else "")
538608 ++ (if svarLat > 0
539− then "\nSVar latency = " ++ show svarLat
609+ then "\nSVar latency = "
610+ ++ secs (fromIntegral svarLat * 1e-9)
611+ else "")
612+ ++ (if svarCnt > 0
613+ then "\nSVar yield count = " ++ show svarCnt
540614 else "")
541615 ]
542616
543617{-# NOINLINE dumpSVar #-}
544618dumpSVar :: SVar t m a -> IO String
545619dumpSVar sv = do
546− tid <- myThreadId
547620 (oqList, oqLen) <- readIORef $ outputQueue sv
548621 db <- tryTakeMVar $ outputDoorBell sv
549622 aheadDump <-
@@ -569,7 +642,7 @@ dumpSVar sv = do
569642 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
570643
571644 return $ unlines
572− [ "tid = " ++ show tid
645+ [ "Creator tid = " ++ show (svarCreator sv)
573646 , "style = " ++ show (svarStyle sv)
574647 , "---------CURRENT STATE-----------"
575648 , "outputQueue length computed = " ++ show (length oqList)
@@ -1654,6 +1727,9 @@ getAheadSVar st f = do
16541727 maxLat <- newIORef (NanoSecs 0)
16551728 minLat <- newIORef (NanoSecs 0)
16561729 stpTime <- newIORef Nothing
1730+#ifdef DIAGNOSTICS
1731+ tid <- myThreadId
1732+#endif
16571733
16581734 let getSVar sv readOutput postProc = SVar
16591735 { outputQueue = outQ
@@ -1675,6 +1751,7 @@ getAheadSVar st f = do
16751751 , workerStopMVar = stopMVar
16761752 , svarRef = Nothing
16771753#ifdef DIAGNOSTICS
1754+ , svarCreator = tid
16781755 , aheadWorkQueue = q
16791756 , outputHeap = outH
16801757#endif
@@ -1739,6 +1816,9 @@ getParallelSVar st = do
17391816 maxLat <- newIORef (NanoSecs 0)
17401817 minLat <- newIORef (NanoSecs 0)
17411818 stpTime <- newIORef Nothing
1819+#ifdef DIAGNOSTICS
1820+ tid <- myThreadId
1821+#endif
17421822
17431823 let sv =
17441824 SVar { outputQueue = outQ
@@ -1760,6 +1840,7 @@ getParallelSVar st = do
17601840 , workerStopMVar = undefined
17611841 , svarRef = Nothing
17621842#ifdef DIAGNOSTICS
1843+ , svarCreator = tid
17631844 , aheadWorkQueue = undefined
17641845 , outputHeap = undefined
17651846#endif