9346144 · streamly · git.iko.soy

git.iko.soy/streamly

print stats on garbage collection

commit93461444b8d53aa4d744d8d21a3bc561ada02a86 authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 13 Aug 2018, 22:31 (8 years ago) parent87aa6dd

4 files changed+116−50

compare with master

unifiedsplit

src/Streamly/SVar.hs+71−31 src/Streamly/Streams/Ahead.hs+6−4 src/Streamly/Streams/Async.hs+16−10 src/Streamly/Streams/SVar.hs+23−5

src/Streamly/SVar.hs+71−31

@@ -1,6 +1,7 @@

11{-# LANGUAGE CPP #-}

22{-# LANGUAGE KindSignatures #-}

33{-# LANGUAGE ConstraintKinds #-}

4+{-# LANGUAGE ExistentialQuantification #-}

45{-# LANGUAGE FlexibleContexts #-}

56{-# LANGUAGE FlexibleInstances #-}

67{-# LANGUAGE GeneralizedNewtypeDeriving #-}

@@ -79,6 +80,10 @@ module Streamly.SVar

7980 , delThread

8081

8182 , toStreamVar

83+#ifdef DIAGNOSTICS

84+ , SVarStats (..)

85+ , dumpSVar

86+#endif

8287 )

8388where

8489

@@ -122,6 +127,7 @@ import Control.Concurrent.MVar (tryTakeMVar)

122127import Control.Exception

123128 (catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),

124129 BlockedIndefinitelyOnSTM(..))

130+import Data.Maybe (isJust)

125131import System.IO (hPutStrLn, stderr)

126132#endif

127133

@@ -258,6 +264,19 @@ data YieldRateInfo = YieldRateInfo

258264 , workerStopTimeStamp :: IORef TimeSpec

259265 }

260266

267+#ifdef DIAGNOSTICS

268+data 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

277+}

278+#endif

279+

261280data Limit = Unlimited | Limited Word deriving Show

262281

263282data SVar t m a = SVar

@@ -290,14 +309,12 @@ data SVar t m a = SVar

290309 , workerStopMVar :: MVar ()

291310

292311#ifdef DIAGNOSTICS

312+ -- to track garbage collection of SVar

313+ , svarRef :: Maybe (IORef ())

293314 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)

294315 -- Shared work queue (stream, seqNo)

295− , aheadWorkQueue :: IORef ([t m a], Int)

296− , totalDispatches :: IORef Int

297− , maxWorkers :: IORef Int

298− , maxOutQSize :: IORef Int

299− , maxHeapSize :: IORef Int

300− , maxWorkQSize :: IORef Int

316+ , aheadWorkQueue :: IORef ([t m a], Int)

317+ , svarStats :: SVarStats

301318#endif

302319 }

303320

@@ -439,6 +456,22 @@ getStreamLatency = _streamLatency

439456-------------------------------------------------------------------------------

440457

441458#ifdef DIAGNOSTICS

459+dumpSVarStats :: SVarStats -> SVarStyle -> Bool -> IO String

460+dumpSVarStats ss style _paced = do

461+ dispatches <- readIORef $ totalDispatches ss

462+ maxWrk <- readIORef $ maxWorkers ss

463+ maxOq <- readIORef $ maxOutQSize ss

464+ maxHp <- readIORef $ maxHeapSize ss

465+

466+ return $ unlines

467+ [ "total dispatches = " ++ show dispatches

468+ , "max workers = " ++ show maxWrk

469+ , "max outQSize = " ++ show maxOq

470+ , if style == AheadVar

471+ then "heap max size = " ++ show maxHp

472+ else ""

473+ ]

474+

442475{-# NOINLINE dumpSVar #-}

443476dumpSVar :: SVar t m a -> IO String

444477dumpSVar sv = do

@@ -450,38 +483,39 @@ dumpSVar sv = do

450483 then do

451484 (oheap, oheapSeq) <- readIORef $ outputHeap sv

452485 (wq, wqSeq) <- readIORef $ aheadWorkQueue sv

453− maxHp <- readIORef $ maxHeapSize sv

454486 return $ unlines

455487 [ "heap length = " ++ show (H.size oheap)

456488 , "heap seqeunce = " ++ show oheapSeq

457489 , "work queue length = " ++ show (length wq)

458490 , "work queue sequence = " ++ show wqSeq

459− , "heap max size = " ++ show maxHp

460491 ]

461492 else return []

462493

463− waiting <- readIORef $ needDoorBell sv

494+ let style = svarStyle sv

495+ waiting <-

496+ if style /= ParallelVar

497+ then readIORef $ needDoorBell sv

498+ else return False

464499 rthread <- readIORef $ workerThreads sv

465500 workers <- readIORef $ workerCount sv

466− maxWrk <- readIORef $ maxWorkers sv

467− dispatches <- readIORef $ totalDispatches sv

468− maxOq <- readIORef $ maxOutQSize sv

501+ stats <- dumpSVarStats (svarStats sv) (svarStyle sv)

502+ (isJust (yieldRateInfo sv))

469503

470504 return $ unlines

471505 [ "tid = " ++ show tid

472506 , "style = " ++ show (svarStyle sv)

507+ , "---------CURRENT STATE-----------"

473508 , "outputQueue length computed = " ++ show (length oqList)

474509 , "outputQueue length maintained = " ++ show oqLen

475− , "output outputDoorBell = " ++ show db

476− , "total dispatches = " ++ show dispatches

477− , "max workers = " ++ show maxWrk

478− , "max outQSize = " ++ show maxOq

510+ , "outputDoorBell = " ++ show db

479511 ]

480512 ++ aheadDump ++ unlines

481513 [ "needDoorBell = " ++ show waiting

482514 , "running threads = " ++ show rthread

483515 , "running thread count = " ++ show workers

484516 ]

517+ ++ "---------STATS-----------\n"

518+ ++ stats

485519

486520{-# NOINLINE mvarExcHandler #-}

487521mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()

@@ -961,9 +995,9 @@ handleChildException sv e = do

961995recordMaxWorkers :: MonadIO m => SVar t m a -> m ()

962996recordMaxWorkers sv = liftIO $ do

963997 active <- readIORef (workerCount sv)

964− maxWrk <- readIORef (maxWorkers sv)

965− when (active > maxWrk) $ writeIORef (maxWorkers sv) active

966− modifyIORef (totalDispatches sv) (+1)

998+ maxWrk <- readIORef (maxWorkers $ svarStats sv)

999+ when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active

1000+ modifyIORef (totalDispatches $ svarStats sv) (+1)

9671001#endif

9681002

9691003{-# NOINLINE pushWorker #-}

@@ -1383,8 +1417,8 @@ readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)

13831417readOutputQRaw sv = do

13841418 (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)

13851419#ifdef DIAGNOSTICS

1386− oqLen <- readIORef (maxOutQSize sv)

1387− when (len > oqLen) $ writeIORef (maxOutQSize sv) len

1420+ oqLen <- readIORef (maxOutQSize $ svarStats sv)

1421+ when (len > oqLen) $ writeIORef (maxOutQSize $ svarStats sv) len

13881422#endif

13891423 return (list, len)

13901424

@@ -1540,13 +1574,16 @@ getAheadSVar st f = do

15401574 , accountThread = delThread sv

15411575 , workerStopMVar = stopMVar

15421576#ifdef DIAGNOSTICS

1577+ , svarRef = Nothing

15431578 , aheadWorkQueue = q

15441579 , outputHeap = outH

1545− , totalDispatches = disp

1546− , maxWorkers = maxWrk

1547− , maxOutQSize = maxOq

1548− , maxHeapSize = maxHs

1549− , maxWorkQSize = maxWq

1580+ , svarStats = SVarStats

1581+ { totalDispatches = disp

1582+ , maxWorkers = maxWrk

1583+ , maxOutQSize = maxOq

1584+ , maxHeapSize = maxHs

1585+ , maxWorkQSize = maxWq

1586+ }

15501587#endif

15511588 }

15521589

@@ -1611,13 +1648,16 @@ getParallelSVar = do

16111648 , accountThread = modifyThread sv

16121649 , workerStopMVar = undefined

16131650#ifdef DIAGNOSTICS

1651+ , svarRef = Nothing

16141652 , aheadWorkQueue = undefined

16151653 , outputHeap = undefined

1616− , totalDispatches = disp

1617− , maxWorkers = maxWrk

1618− , maxOutQSize = maxOq

1619− , maxHeapSize = maxHs

1620− , maxWorkQSize = maxWq

1654+ , svarStats = SVarStats

1655+ { totalDispatches = disp

1656+ , maxWorkers = maxWrk

1657+ , maxOutQSize = maxOq

1658+ , maxHeapSize = maxHs

1659+ , maxWorkQSize = maxWq

1660+ }

16211661#endif

16221662 }

16231663 in return sv

src/Streamly/Streams/Ahead.hs+6−4

@@ -214,9 +214,10 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

214214 loopHeap seqNo ent = do

215215#ifdef DIAGNOSTICS

216216 liftIO $ do

217− maxHp <- readIORef (maxHeapSize sv)

217+ maxHp <- readIORef (maxHeapSize $ svarStats sv)

218218 (hp, _) <- readIORef heap

219− when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)

219+ when (H.size hp > maxHp) $ writeIORef (maxHeapSize $ svarStats sv)

220+ (H.size hp)

220221#endif

221222 case ent of

222223 AheadEntryPure a -> do

@@ -449,9 +450,10 @@ workLoopAhead :: MonadIO m

449450workLoopAhead q heap st sv winfo = do

450451#ifdef DIAGNOSTICS

451452 liftIO $ do

452− maxHp <- readIORef (maxHeapSize sv)

453+ maxHp <- readIORef (maxHeapSize $ svarStats sv)

453454 (hp, _) <- readIORef heap

454− when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)

455+ when (H.size hp > maxHp) $ writeIORef (maxHeapSize $ svarStats sv)

456+ (H.size hp)

455457#endif

456458 ent <- liftIO $ dequeueFromHeap heap

457459 case ent of

src/Streamly/Streams/Async.hs+16−10

@@ -297,13 +297,16 @@ getLifoSVar st = do

297297 , accountThread = delThread sv

298298 , workerStopMVar = undefined

299299#ifdef DIAGNOSTICS

300+ , svarRef = Nothing

300301 , aheadWorkQueue = undefined

301302 , outputHeap = undefined

302− , maxWorkers = maxWrk

303− , totalDispatches = disp

304− , maxOutQSize = maxOq

305− , maxHeapSize = maxHs

306− , maxWorkQSize = maxWq

303+ , svarStats = SVarStats

304+ { totalDispatches = disp

305+ , maxWorkers = maxWrk

306+ , maxOutQSize = maxOq

307+ , maxHeapSize = maxHs

308+ , maxWorkQSize = maxWq

309+ }

307310#endif

308311 }

309312

@@ -382,13 +385,16 @@ getFifoSVar st = do

382385 , accountThread = delThread sv

383386 , workerStopMVar = undefined

384387#ifdef DIAGNOSTICS

388+ , svarRef = Nothing

385389 , aheadWorkQueue = undefined

386390 , outputHeap = undefined

387− , totalDispatches = disp

388− , maxWorkers = maxWrk

389− , maxOutQSize = maxOq

390− , maxHeapSize = maxHs

391− , maxWorkQSize = maxWq

391+ , svarStats = SVarStats

392+ { totalDispatches = disp

393+ , maxWorkers = maxWrk

394+ , maxOutQSize = maxOq

395+ , maxHeapSize = maxHs

396+ , maxWorkQSize = maxWq

397+ }

392398#endif

393399 }

394400

src/Streamly/Streams/SVar.hs+23−5

@@ -33,10 +33,12 @@ module Streamly.Streams.SVar

3333where

3434

3535import Control.Monad.Catch (throwM)

36−#ifdef DIAGNOSTICS_VERBOSE

36+import Data.Int (Int64)

37+#ifdef DIAGNOSTICS

3738import Control.Monad.IO.Class (liftIO)

39+import Data.IORef (newIORef, mkWeakIORef)

40+import System.IO (hPutStrLn, stderr)

3841#endif

39−import Data.Int (Int64)

4042

4143import Streamly.SVar

4244import Streamly.Streams.StreamK

@@ -47,6 +49,15 @@ import Streamly.Streams.Serial (SerialT)

4749-- happen, but it may result in unexpected output when threads are left hanging

4850-- until they are GCed because the consumer went away.

4951

52+#ifdef DIAGNOSTICS

53+#ifdef DIAGNOSTICS_VERBOSE

54+printSVar :: SVar t m a -> String -> IO ()

55+printSVar sv how = do

56+ svInfo <- dumpSVar sv

57+ hPutStrLn stderr $ "\n" ++ how ++ "\n" ++ svInfo

58+#endif

59+#endif

60+

5061-- | Pull a stream from an SVar.

5162{-# NOINLINE fromStreamVar #-}

5263fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a

@@ -62,8 +73,7 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do

6273 allDone stp = do

6374#ifdef DIAGNOSTICS

6475#ifdef DIAGNOSTICS_VERBOSE

65− svInfo <- liftIO $ dumpSVar sv

66− liftIO $ hPutStrLn stderr $ "fromStreamVar done\n" ++ svInfo

76+ liftIO $ printSVar sv "SVar Done"

6777#endif

6878#endif

6979 stp

@@ -87,7 +97,15 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do

8797

8898{-# INLINE fromSVar #-}

8999fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a

90−fromSVar sv = fromStream $ fromStreamVar sv

100+fromSVar sv = do

101+ fromStream $ Stream $ \st stp sng yld -> do

102+#ifdef DIAGNOSTICS_VERBOSE

103+ ref <- liftIO $ newIORef ()

104+ _ <- liftIO $ mkWeakIORef ref (printSVar sv "SVar Garbage Collected")

105+ unStream (fromStreamVar sv{svarRef = Just ref}) st stp sng yld

106+#else

107+ unStream (fromStreamVar sv) st stp sng yld

108+#endif

91109

92110-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

93111-- be read back from the SVar using 'fromSVar'.