@@ -19,11 +19,6 @@
1919-- Maintainer : harendra.kumar@gmail.com
2020-- Stability : experimental
2121-- Portability : GHC
22−--
23−--
24−#ifdef DIAGNOSTICS_VERBOSE
25−#define DIAGNOSTICS
26−#endif
2722
2823module Streamly.SVar
2924 (
@@ -45,6 +40,8 @@ module Streamly.SVar
4540 , setStreamLatency
4641 , getYieldLimit
4742 , setYieldLimit
43+ , getInspectMode
44+ , setInspectMode
4845
4946 , cleanupSVar
5047 , cleanupSVarFromWorker
@@ -99,17 +96,18 @@ module Streamly.SVar
9996 , toStreamVar
10097 , SVarStats (..)
10198 , NanoSecs (..)
102−#ifdef DIAGNOSTICS
10399 , dumpSVar
104−#endif
105100 )
106101where
107102
108103import Control.Concurrent
109104 (ThreadId, myThreadId, threadDelay, throwTo)
110105import Control.Concurrent.MVar
111− (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)
112−import Control.Exception (SomeException(..), catch, mask, assert, Exception)
106+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, tryReadMVar)
107+import Control.Exception
108+ (SomeException(..), catch, mask, assert, Exception, catches,
109+ throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
110+ BlockedIndefinitelyOnSTM(..))
113111import Control.Monad (when)
114112import Control.Monad.Catch (MonadThrow)
115113import Control.Monad.IO.Class (MonadIO(..))
@@ -130,24 +128,12 @@ import GHC.Conc (ThreadId(..))
130128import GHC.Exts
131129import GHC.IO (IO(..))
132130import System.Clock (TimeSpec, Clock(Monotonic), getTime, toNanoSecs)
131+import System.IO (hPutStrLn, stderr)
132+import Text.Printf (printf)
133133
134134import qualified Data.Heap as H
135135import qualified Data.Set as S
136136
137−-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
138−-- can keep it on in production to debug problems quickly if and when they
139−-- happen, but it may result in unexpected output when threads are left hanging
140−-- until they are GCed because the consumer went away.
141−
142−#ifdef DIAGNOSTICS
143−import Control.Concurrent.MVar (tryTakeMVar)
144−import Control.Exception
145− (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
146− BlockedIndefinitelyOnSTM(..))
147−import System.IO (hPutStrLn, stderr)
148−import Text.Printf (printf)
149−#endif
150−
151137-- Always use signed arithmetic to avoid inadvertant overflows of signed values
152138-- on conversion when comparing unsigned quantities with signed.
153139newtype NanoSecs = NanoSecs Int64
@@ -371,13 +357,14 @@ data SVar t m a = SVar
371357 , svarStats :: SVarStats
372358 -- to track garbage collection of SVar
373359 , svarRef :: Maybe (IORef ())
374−#ifdef DIAGNOSTICS
360+
361+ -- Only for diagnostics
362+ , svarInspectMode :: Bool
375363 , svarCreator :: ThreadId
376364 , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
377365 , Maybe Int)
378366 -- Shared work queue (stream, seqNo)
379367 , aheadWorkQueue :: IORef ([t m a], Int)
380−#endif
381368 }
382369
383370-------------------------------------------------------------------------------
@@ -401,6 +388,7 @@ data State t m a = State
401388 -- XXX these two can be collapsed into a single type
402389 , _streamLatency :: Maybe NanoSecs -- bootstrap latency
403390 , _maxStreamRate :: Maybe Rate
391+ , _inspectMode :: Bool
404392 }
405393
406394-------------------------------------------------------------------------------
@@ -428,6 +416,7 @@ defState = State
428416 , _bufferHigh = defaultMaxBuffer
429417 , _maxStreamRate = Nothing
430418 , _streamLatency = Nothing
419+ , _inspectMode = False
431420 }
432421
433422-- XXX if perf gets affected we can have all the Nothing params in a single
@@ -505,6 +494,12 @@ setStreamLatency n st =
505494getStreamLatency :: State t m a -> Maybe NanoSecs
506495getStreamLatency = _streamLatency
507496
497+setInspectMode :: State t m a -> State t m a
498+setInspectMode st = st { _inspectMode = True }
499+
500+getInspectMode :: State t m a -> Bool
501+getInspectMode = _inspectMode
502+
508503-------------------------------------------------------------------------------
509504-- Cleanup
510505-------------------------------------------------------------------------------
@@ -526,7 +521,6 @@ cleanupSVarFromWorker sv = do
526521-- Dumping the SVar for debug/diag
527522-------------------------------------------------------------------------------
528523
529−#ifdef DIAGNOSTICS
530524-- | Convert a number of seconds to a string. The string will consist
531525-- of four decimal places, followed by a short description of the time
532526-- units.
@@ -553,8 +547,8 @@ secs k
553547 | otherwise = printf "%.3f %s" t u
554548
555549-- XXX Code duplicated from collectLatency
556−drainLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
557−drainLatency _ss yinfo = do
550+drainLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
551+drainLatency sv yinfo = do
558552 let cur = workerPendingLatency yinfo
559553 col = workerCollectedLatency yinfo
560554 longTerm = svarAllTimeLatency yinfo
@@ -574,21 +568,19 @@ drainLatency _ss yinfo = do
574568 if (pendingCount > 0)
575569 then do
576570 let new = pendingTime `div` (fromIntegral pendingCount)
577−#ifdef DIAGNOSTICS
578− minLat <- readIORef (minWorkerLatency _ss)
579− when (new < minLat || minLat == 0) $
580− writeIORef (minWorkerLatency _ss) new
581−
582− maxLat <- readIORef (maxWorkerLatency _ss)
583− when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
584−#endif
571+ when (svarInspectMode sv) $ do
572+ let ss = svarStats sv
573+ minLat <- readIORef (minWorkerLatency ss)
574+ when (new < minLat || minLat == 0) $
575+ writeIORef (minWorkerLatency ss) new
576+
577+ maxLat <- readIORef (maxWorkerLatency ss)
578+ when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
579+ modifyIORef (avgWorkerLatency ss) $
580+ \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
585581 -- To avoid minor fluctuations update in batches
586582 writeIORef col (0, 0)
587583 writeIORef measured new
588−#ifdef DIAGNOSTICS
589− modifyIORef (avgWorkerLatency _ss) $
590− \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
591−#endif
592584 modifyIORef longTerm $ \(_, t) -> (lcount', t)
593585 return (lcount', ltime, new)
594586 else return notUpdated
@@ -598,7 +590,7 @@ dumpSVarStats sv ss style = do
598590 case yieldRateInfo sv of
599591 Nothing -> return ()
600592 Just yinfo -> do
601− _ <- liftIO $ drainLatency (svarStats sv) yinfo
593+ _ <- liftIO $ drainLatency sv yinfo
602594 return ()
603595
604596 dispatches <- readIORef $ totalDispatches ss
@@ -662,7 +654,7 @@ dumpSVarStats sv ss style = do
662654dumpSVar :: SVar t m a -> IO String
663655dumpSVar sv = do
664656 (oqList, oqLen) <- readIORef $ outputQueue sv
665− db <- tryTakeMVar $ outputDoorBell sv
657+ db <- tryReadMVar $ outputDoorBell sv
666658 aheadDump <-
667659 if svarStyle sv == AheadVar
668660 then do
@@ -686,15 +678,17 @@ dumpSVar sv = do
686678 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
687679
688680 return $ unlines
689− [ "Creator tid = " ++ show (svarCreator sv)
690− , "style = " ++ show (svarStyle sv)
681+ [
682+ "Creator tid = " ++ show (svarCreator sv),
683+ "style = " ++ show (svarStyle sv)
691684 , "---------CURRENT STATE-----------"
692685 , "outputQueue length computed = " ++ show (length oqList)
693686 , "outputQueue length maintained = " ++ show oqLen
694687 -- XXX print the types of events in the outputQueue, first 5
695688 , "outputDoorBell = " ++ show db
696689 ]
697− ++ aheadDump ++ unlines
690+ ++ aheadDump
691+ ++ unlines
698692 [ "needDoorBell = " ++ show waiting
699693 , "running threads = " ++ show rthread
700694 -- XXX print the status of first 5 threads
@@ -703,6 +697,11 @@ dumpSVar sv = do
703697 ++ "---------STATS-----------\n"
704698 ++ stats
705699
700+-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
701+-- can keep it on in production to debug problems quickly if and when they
702+-- happen, but it may result in unexpected output when threads are left hanging
703+-- until they are GCed because the consumer went away.
704+
706705{-# NOINLINE mvarExcHandler #-}
707706mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
708707mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
@@ -717,15 +716,14 @@ stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
717716 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
718717 throwIO e
719718
720−withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
721−withDBGMVar sv label action =
722− action `catches` [ Handler (mvarExcHandler sv label)
723− , Handler (stmExcHandler sv label)
724− ]
725−#else
726−withDBGMVar :: SVar t m a -> String -> IO () -> IO ()
727−withDBGMVar _ _ action = action
728−#endif
719+withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
720+withDiagMVar sv label action =
721+ if svarInspectMode sv
722+ then
723+ action `catches` [ Handler (mvarExcHandler sv label)
724+ , Handler (stmExcHandler sv label)
725+ ]
726+ else action
729727
730728-------------------------------------------------------------------------------
731729-- CAS
@@ -1248,22 +1246,19 @@ handleChildException sv e = do
12481246 tid <- myThreadId
12491247 void $ send sv (ChildStop tid (Just e))
12501248
1251−#ifdef DIAGNOSTICS
1249+{-# NOINLINE recordMaxWorkers #-}
12521250recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
12531251recordMaxWorkers sv = liftIO $ do
12541252 active <- readIORef (workerCount sv)
12551253 maxWrk <- readIORef (maxWorkers $ svarStats sv)
12561254 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
12571255 modifyIORef (totalDispatches $ svarStats sv) (+1)
1258−#endif
12591256
12601257{-# NOINLINE pushWorker #-}
12611258pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
12621259pushWorker yieldMax sv = do
12631260 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1264−#ifdef DIAGNOSTICS
1265− recordMaxWorkers sv
1266−#endif
1261+ when (svarInspectMode sv) $ recordMaxWorkers sv
12671262 -- This allocation matters when significant number of workers are being
12681263 -- sent. We allocate it only when needed.
12691264 winfo <-
@@ -1288,34 +1283,39 @@ pushWorker yieldMax sv = do
12881283-- producer side. So we need to use a thread safe modification of
12891284-- workerThreads. Alternatively, we can use a CreateThread event to avoid
12901285-- using a CAS based modification.
1291−{-# NOINLINE pushWorkerPar #-}
1292−pushWorkerPar :: MonadAsync m => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1286+{-# INLINE pushWorkerPar #-}
1287+pushWorkerPar
1288+ :: MonadAsync m
1289+ => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
12931290pushWorkerPar sv wloop =
1294− -- We do not use workerCount in case of ParallelVar but still there is no
1295− -- harm in maintaining it correctly.
1296−#ifdef DIAGNOSTICS
1297− do
1298− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1299− recordMaxWorkers sv
1300− -- This allocation matters when significant number of workers are being
1301− -- sent. We allocate it only when needed. The overhead increases by 4x.
1302− winfo <-
1303− case yieldRateInfo sv of
1304− Nothing -> return Nothing
1305− Just _ -> liftIO $ do
1306− cntRef <- newIORef 0
1307− t <- getTime Monotonic
1308− lat <- newIORef (0, t)
1309− return $ Just $ WorkerInfo
1310− { workerYieldMax = 0
1311− , workerYieldCount = cntRef
1312− , workerLatencyStart = lat
1313− }
1291+ if svarInspectMode sv
1292+ then forkWithDiag
1293+ else doFork (wloop Nothing) (handleChildException sv) >>= modifyThread sv
13141294
1315− doFork (wloop winfo) (handleChildException sv) >>= modifyThread sv
1316−#else
1317− doFork (wloop Nothing) (handleChildException sv) >>= modifyThread sv
1318−#endif
1295+ where
1296+
1297+ {-# NOINLINE forkWithDiag #-}
1298+ forkWithDiag = do
1299+ -- We do not use workerCount in case of ParallelVar but still there is
1300+ -- no harm in maintaining it correctly.
1301+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1302+ recordMaxWorkers sv
1303+ -- This allocation matters when significant number of workers are being
1304+ -- sent. We allocate it only when needed. The overhead increases by 4x.
1305+ winfo <-
1306+ case yieldRateInfo sv of
1307+ Nothing -> return Nothing
1308+ Just _ -> liftIO $ do
1309+ cntRef <- newIORef 0
1310+ t <- getTime Monotonic
1311+ lat <- newIORef (0, t)
1312+ return $ Just $ WorkerInfo
1313+ { workerYieldMax = 0
1314+ , workerYieldCount = cntRef
1315+ , workerLatencyStart = lat
1316+ }
1317+
1318+ doFork (wloop winfo) (handleChildException sv) >>= modifyThread sv
13191319
13201320-- Returns:
13211321-- True: can dispatch more
@@ -1547,8 +1547,8 @@ updateWorkerPollingInterval yinfo latency = do
15471547-- whereas the average is used for future estimates e.g. how many workers
15481548-- should be maintained to maintain the rate.
15491549-- CAUTION! keep it in sync with getWorkerLatency
1550−collectLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1551−collectLatency _ss yinfo = do
1550+collectLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1551+collectLatency sv yinfo = do
15521552 let cur = workerPendingLatency yinfo
15531553 col = workerCollectedLatency yinfo
15541554 longTerm = svarAllTimeLatency yinfo
@@ -1567,15 +1567,15 @@ collectLatency _ss yinfo = do
15671567
15681568 if pendingCount > 0
15691569 then do
1570− let new = pendingTime `div` fromIntegral pendingCount
1571−#ifdef DIAGNOSTICS
1572− minLat <- readIORef (minWorkerLatency _ss)
1573− when (new < minLat || minLat == 0) $
1574− writeIORef (minWorkerLatency _ss) new
1575−
1576− maxLat <- readIORef (maxWorkerLatency _ss)
1577− when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new
1578−#endif
1570+ let new = pendingTime `div` (fromIntegral pendingCount)
1571+ when (svarInspectMode sv) $ do
1572+ let ss = svarStats sv
1573+ minLat <- readIORef (minWorkerLatency ss)
1574+ when (new < minLat || minLat == 0) $
1575+ writeIORef (minWorkerLatency ss) new
1576+
1577+ maxLat <- readIORef (maxWorkerLatency ss)
1578+ when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
15791579 -- When we have collected a significant sized batch we compute the new
15801580 -- latency using that batch and return the new latency, otherwise we
15811581 -- return the previous latency derived from the previous batch.
@@ -1585,13 +1585,13 @@ collectLatency _ss yinfo = do
15851585 in prev > 0 && (r > 2 || r < 0.5))
15861586 || (prev == 0)
15871587 then do
1588+ when (svarInspectMode sv) $ do
1589+ let ss = svarStats sv
1590+ modifyIORef (avgWorkerLatency ss) $
1591+ \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
15881592 updateWorkerPollingInterval yinfo (max new prev)
15891593 writeIORef col (0, 0)
15901594 writeIORef measured ((prev + new) `div` 2)
1591−#ifdef DIAGNOSTICS
1592− modifyIORef (avgWorkerLatency _ss) $
1593− \(cnt, t) -> (cnt + pendingCount, t + pendingTime)
1594−#endif
15951595 modifyIORef longTerm $ \(_, t) -> (lcount', t)
15961596 return $ tripleWith new
15971597 else do
@@ -1613,7 +1613,7 @@ dispatchWorkerPaced sv = do
16131613 (svarYields, svarElapsed, wLatency) <- do
16141614 now <- liftIO $ getTime Monotonic
16151615 (yieldCount, baseTime, lat) <-
1616− liftIO $ collectLatency (svarStats sv) yinfo
1616+ liftIO $ collectLatency sv yinfo
16171617 let elapsed = fromInteger $ toNanoSecs $ now - baseTime
16181618 let latency =
16191619 if lat == 0
@@ -1811,7 +1811,7 @@ sendWorkerWait delay dispatch sv = do
18111811 if canDoMore
18121812 then sendWorkerWait delay dispatch sv
18131813 else do
1814− liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
1814+ liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
18151815 $ takeMVar (outputDoorBell sv)
18161816 (_, len) <- liftIO $ readIORef (outputQueue sv)
18171817 when (len <= 0) $ sendWorkerWait delay dispatch sv
@@ -1820,10 +1820,10 @@ sendWorkerWait delay dispatch sv = do
18201820readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
18211821readOutputQRaw sv = do
18221822 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
1823−#ifdef DIAGNOSTICS
1824− oqLen <- readIORef (maxOutQSize $ svarStats sv)
1825− when (len > oqLen) $ writeIORef (maxOutQSize $ svarStats sv) len
1826−#endif
1823+ when (svarInspectMode sv) $ do
1824+ let ref = maxOutQSize $ svarStats sv
1825+ oqLen <- readIORef ref
1826+ when (len > oqLen) $ writeIORef ref len
18271827 return (list, len)
18281828
18291829readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
@@ -1985,9 +1985,7 @@ getAheadSVar st f = do
19851985 maxLat <- newIORef (NanoSecs 0)
19861986 minLat <- newIORef (NanoSecs 0)
19871987 stpTime <- newIORef Nothing
1988−#ifdef DIAGNOSTICS
19891988 tid <- myThreadId
1990−#endif
19911989
19921990 let getSVar sv readOutput postProc = SVar
19931991 { outputQueue = outQ
@@ -2009,11 +2007,10 @@ getAheadSVar st f = do
20092007 , accountThread = delThread sv
20102008 , workerStopMVar = stopMVar
20112009 , svarRef = Nothing
2012−#ifdef DIAGNOSTICS
2010+ , svarInspectMode = getInspectMode st
20132011 , svarCreator = tid
20142012 , aheadWorkQueue = q
20152013 , outputHeap = outH
2016−#endif
20172014 , svarStats = SVarStats
20182015 { totalDispatches = disp
20192016 , maxWorkers = maxWrk
@@ -2081,9 +2078,7 @@ getParallelSVar st = do
20812078 maxLat <- newIORef (NanoSecs 0)
20822079 minLat <- newIORef (NanoSecs 0)
20832080 stpTime <- newIORef Nothing
2084−#ifdef DIAGNOSTICS
20852081 tid <- myThreadId
2086−#endif
20872082
20882083 let sv =
20892084 SVar { outputQueue = outQ
@@ -2106,11 +2101,10 @@ getParallelSVar st = do
21062101 , accountThread = modifyThread sv
21072102 , workerStopMVar = undefined
21082103 , svarRef = Nothing
2109−#ifdef DIAGNOSTICS
2104+ , svarInspectMode = getInspectMode st
21102105 , svarCreator = tid
21112106 , aheadWorkQueue = undefined
21122107 , outputHeap = undefined
2113−#endif
21142108 , svarStats = SVarStats
21152109 { totalDispatches = disp
21162110 , maxWorkers = maxWrk
@@ -2128,10 +2122,11 @@ getParallelSVar st = do
21282122 where
21292123
21302124 readOutputQPar sv = liftIO $ do
2131− withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)
2125+ withDiagMVar sv "readOutputQPar: doorbell"
2126+ $ takeMVar (outputDoorBell sv)
21322127 case yieldRateInfo sv of
21332128 Nothing -> return ()
2134− Just yinfo -> void $ collectLatency (svarStats sv) yinfo
2129+ Just yinfo -> void $ collectLatency sv yinfo
21352130 fst `fmap` readOutputQRaw sv
21362131
21372132sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)