9346144 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

print stats on garbage collection

9346144Harendra Kumar8 years ago

src/Streamly/SVar.hs+71−31

compare with master

unifiedsplit

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

1{-# LANGUAGE CPP #-}

2{-# LANGUAGE KindSignatures #-}

3{-# LANGUAGE ConstraintKinds #-}

4+{-# LANGUAGE ExistentialQuantification #-}

4{-# LANGUAGE FlexibleContexts #-}

5{-# LANGUAGE FlexibleContexts #-}

5{-# LANGUAGE FlexibleInstances #-}

6{-# LANGUAGE FlexibleInstances #-}

6{-# LANGUAGE GeneralizedNewtypeDeriving #-}

7{-# LANGUAGE GeneralizedNewtypeDeriving #-}

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

79 , delThread

80 , delThread

81 , toStreamVar

82 , toStreamVar

83+#ifdef DIAGNOSTICS

84+ , SVarStats (..)

85+ , dumpSVar

86+#endif

82 )

87 )

83where

88where

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

122import Control.Exception

127import Control.Exception

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

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

124 BlockedIndefinitelyOnSTM(..))

129 BlockedIndefinitelyOnSTM(..))

130+import Data.Maybe (isJust)

125import System.IO (hPutStrLn, stderr)

131import System.IO (hPutStrLn, stderr)

126#endif

132#endif

127

133

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

258 , workerStopTimeStamp :: IORef TimeSpec

264 , workerStopTimeStamp :: IORef TimeSpec

259 }

265 }

260

266

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+

261data Limit = Unlimited | Limited Word deriving Show

280data Limit = Unlimited | Limited Word deriving Show

262

281

263data SVar t m a = SVar

282data SVar t m a = SVar

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

290 , workerStopMVar :: MVar ()

309 , workerStopMVar :: MVar ()

291

310

292#ifdef DIAGNOSTICS

311#ifdef DIAGNOSTICS

312+ -- to track garbage collection of SVar

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

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

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

294 -- Shared work queue (stream, seqNo)

315 -- Shared work queue (stream, seqNo)

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

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

296− , totalDispatches :: IORef Int

317+ , svarStats :: SVarStats

297− , maxWorkers :: IORef Int

298− , maxOutQSize :: IORef Int

299− , maxHeapSize :: IORef Int

300− , maxWorkQSize :: IORef Int

301#endif

318#endif

302 }

319 }

303

320

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

439-------------------------------------------------------------------------------

456-------------------------------------------------------------------------------

440

457

441#ifdef DIAGNOSTICS

458#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+

442{-# NOINLINE dumpSVar #-}

475{-# NOINLINE dumpSVar #-}

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

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

444dumpSVar sv = do

477dumpSVar sv = do

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

450 then do

483 then do

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

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

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

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

453− maxHp <- readIORef $ maxHeapSize sv

454 return $ unlines

486 return $ unlines

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

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

456 , "heap seqeunce = " ++ show oheapSeq

488 , "heap seqeunce = " ++ show oheapSeq

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

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

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

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

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

460 ]

491 ]

461 else return []

492 else return []

462

493

494+ let style = svarStyle sv

495+ waiting <-

496+ if style /= ParallelVar

463− waiting <- readIORef $ needDoorBell sv

497+ then readIORef $ needDoorBell sv

498+ else return False

464 rthread <- readIORef $ workerThreads sv

499 rthread <- readIORef $ workerThreads sv

465 workers <- readIORef $ workerCount sv

500 workers <- readIORef $ workerCount sv

466− maxWrk <- readIORef $ maxWorkers sv

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

467− dispatches <- readIORef $ totalDispatches sv

502+ (isJust (yieldRateInfo sv))

468− maxOq <- readIORef $ maxOutQSize sv

469

503

470 return $ unlines

504 return $ unlines

471 [ "tid = " ++ show tid

505 [ "tid = " ++ show tid

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

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

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

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

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

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

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

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

510+ , "outputDoorBell = " ++ show db

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

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

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

479 ]

511 ]

480 ++ aheadDump ++ unlines

512 ++ aheadDump ++ unlines

481 [ "needDoorBell = " ++ show waiting

513 [ "needDoorBell = " ++ show waiting

482 , "running threads = " ++ show rthread

514 , "running threads = " ++ show rthread

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

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

484 ]

516 ]

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

518+ ++ stats

485

519

486{-# NOINLINE mvarExcHandler #-}

520{-# NOINLINE mvarExcHandler #-}

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

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

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

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

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

962recordMaxWorkers sv = liftIO $ do

996recordMaxWorkers sv = liftIO $ do

963 active <- readIORef (workerCount sv)

997 active <- readIORef (workerCount sv)

964− maxWrk <- readIORef (maxWorkers sv)

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

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

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

966− modifyIORef (totalDispatches sv) (+1)

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

967#endif

1001#endif

968

1002

969{-# NOINLINE pushWorker #-}

1003{-# NOINLINE pushWorker #-}

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

1383readOutputQRaw sv = do

1417readOutputQRaw sv = do

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

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

1385#ifdef DIAGNOSTICS

1419#ifdef DIAGNOSTICS

1386− oqLen <- readIORef (maxOutQSize sv)

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

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

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

1388#endif

1422#endif

1389 return (list, len)

1423 return (list, len)

1390

1424

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

1540 , accountThread = delThread sv

1574 , accountThread = delThread sv

1541 , workerStopMVar = stopMVar

1575 , workerStopMVar = stopMVar

1542#ifdef DIAGNOSTICS

1576#ifdef DIAGNOSTICS

1577+ , svarRef = Nothing

1543 , aheadWorkQueue = q

1578 , aheadWorkQueue = q

1544 , outputHeap = outH

1579 , outputHeap = outH

1580+ , svarStats = SVarStats

1545− , totalDispatches = disp

1581+ { totalDispatches = disp

1546− , maxWorkers = maxWrk

1582+ , maxWorkers = maxWrk

1547− , maxOutQSize = maxOq

1583+ , maxOutQSize = maxOq

1548− , maxHeapSize = maxHs

1584+ , maxHeapSize = maxHs

1549− , maxWorkQSize = maxWq

1585+ , maxWorkQSize = maxWq

1586+ }

1550#endif

1587#endif

1551 }

1588 }

1552

1589

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

1611 , accountThread = modifyThread sv

1648 , accountThread = modifyThread sv

1612 , workerStopMVar = undefined

1649 , workerStopMVar = undefined

1613#ifdef DIAGNOSTICS

1650#ifdef DIAGNOSTICS

1651+ , svarRef = Nothing

1614 , aheadWorkQueue = undefined

1652 , aheadWorkQueue = undefined

1615 , outputHeap = undefined

1653 , outputHeap = undefined

1654+ , svarStats = SVarStats

1616− , totalDispatches = disp

1655+ { totalDispatches = disp

1617− , maxWorkers = maxWrk

1656+ , maxWorkers = maxWrk

1618− , maxOutQSize = maxOq

1657+ , maxOutQSize = maxOq

1619− , maxHeapSize = maxHs

1658+ , maxHeapSize = maxHs

1620− , maxWorkQSize = maxWq

1659+ , maxWorkQSize = maxWq

1660+ }

1621#endif

1661#endif

1622 }

1662 }

1623 in return sv

1663 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

214 loopHeap seqNo ent = do

215#ifdef DIAGNOSTICS

216 liftIO $ do

217− maxHp <- readIORef (maxHeapSize sv)

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

218 (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)

220#endif

221#endif

221 case ent of

222 case ent of

222 AheadEntryPure a -> do

223 AheadEntryPure a -> do

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

449workLoopAhead q heap st sv winfo = do

450workLoopAhead q heap st sv winfo = do

450#ifdef DIAGNOSTICS

451#ifdef DIAGNOSTICS

451 liftIO $ do

452 liftIO $ do

452− maxHp <- readIORef (maxHeapSize sv)

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

453 (hp, _) <- readIORef heap

454 (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)

455#endif

457#endif

456 ent <- liftIO $ dequeueFromHeap heap

458 ent <- liftIO $ dequeueFromHeap heap

457 case ent of

459 case ent of

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

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

297 , accountThread = delThread sv

298 , workerStopMVar = undefined

299#ifdef DIAGNOSTICS

300+ , svarRef = Nothing

300 , aheadWorkQueue = undefined

301 , aheadWorkQueue = undefined

301 , outputHeap = undefined

302 , outputHeap = undefined

303+ , svarStats = SVarStats

304+ { totalDispatches = disp

302− , maxWorkers = maxWrk

305+ , maxWorkers = maxWrk

303− , totalDispatches = disp

304− , maxOutQSize = maxOq

306+ , maxOutQSize = maxOq

305− , maxHeapSize = maxHs

307+ , maxHeapSize = maxHs

306− , maxWorkQSize = maxWq

308+ , maxWorkQSize = maxWq

309+ }

307#endif

310#endif

308 }

311 }

309

312

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

382 , accountThread = delThread sv

385 , accountThread = delThread sv

383 , workerStopMVar = undefined

386 , workerStopMVar = undefined

384#ifdef DIAGNOSTICS

387#ifdef DIAGNOSTICS

388+ , svarRef = Nothing

385 , aheadWorkQueue = undefined

389 , aheadWorkQueue = undefined

386 , outputHeap = undefined

390 , outputHeap = undefined

391+ , svarStats = SVarStats

387− , totalDispatches = disp

392+ { totalDispatches = disp

388− , maxWorkers = maxWrk

393+ , maxWorkers = maxWrk

389− , maxOutQSize = maxOq

394+ , maxOutQSize = maxOq

390− , maxHeapSize = maxHs

395+ , maxHeapSize = maxHs

391− , maxWorkQSize = maxWq

396+ , maxWorkQSize = maxWq

397+ }

392#endif

398#endif

393 }

399 }

394

400

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

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

33where

35import Control.Monad.Catch (throwM)

36−#ifdef DIAGNOSTICS_VERBOSE

36+import Data.Int (Int64)

37+#ifdef DIAGNOSTICS

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

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

39+import Data.IORef (newIORef, mkWeakIORef)

40+import System.IO (hPutStrLn, stderr)

38#endif

41#endif

39−import Data.Int (Int64)

41import Streamly.SVar

43import Streamly.SVar

42import Streamly.Streams.StreamK

44import Streamly.Streams.StreamK

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

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

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

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

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

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+

50-- | Pull a stream from an SVar.

61-- | Pull a stream from an SVar.

51{-# NOINLINE fromStreamVar #-}

62{-# NOINLINE fromStreamVar #-}

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

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

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

62 allDone stp = do

73 allDone stp = do

63#ifdef DIAGNOSTICS

74#ifdef DIAGNOSTICS

64#ifdef DIAGNOSTICS_VERBOSE

75#ifdef DIAGNOSTICS_VERBOSE

65− svInfo <- liftIO $ dumpSVar sv

76+ liftIO $ printSVar sv "SVar Done"

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

67#endif

77#endif

68#endif

78#endif

69 stp

79 stp

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

88{-# INLINE fromSVar #-}

98{-# INLINE fromSVar #-}

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

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

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

90−fromSVar sv = fromStream $ fromStreamVar sv

107+ unStream (fromStreamVar sv) st stp sng yld

108+#endif

109

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

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

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

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