1f72b61 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add inspectMode combinator

1f72b61Harendra Kumar8 years ago

.travis.yml+1−1

compare with master

unifiedsplit

@@ -86,7 +86,7 @@ matrix:

86 addons: {apt: {packages: [cabal-install-1.24,libsdl1.2-dev], sources: [hvr-ghc]}}

88 # Use cabal-install-1.24 for stack solver to work

89− - env: BUILD=stack GHCVER=8.2.2 RESOLVER=lts-11 GHC_OPTIONS="" DISABLE_SDIST_BUILD=y DISABLE_DIST_CHECKS=y STACK_BUILD_OPTIONS="--flag streamly:diag"

89+ - env: BUILD=stack GHCVER=8.2.2 RESOLVER=lts-11 GHC_OPTIONS="" DISABLE_SDIST_BUILD=y DISABLE_DIST_CHECKS=y

90 addons: {apt: {packages: [cabal-install-1.24,libsdl1.2-dev], sources: [hvr-ghc]}}

92 # Use cabal-install-1.24 for stack solver to work

src/Streamly/Internal.hsnew file+19

@@ -0,0 +1,19 @@

1+-- |

2+-- Module : Streamly.Internal

4+--

5+-- License : BSD3

6+-- Maintainer : harendra.kumar@gmail.com

7+-- Stability : experimental

8+-- Portability : GHC

9+--

10+-- This module is only for internal use. There is no warranty for the routines

11+-- in this module to work correctly, please use at your own risk. These

12+-- routines are subject to change or be removed without notice.

13+--

14+module Streamly.Internal

15+ ( inspectMode

16+ )

17+where

18+

19+import Streamly.Streams.SVar

src/Streamly/SVar.hs+111−116

@@ -19,11 +19,6 @@

19-- Maintainer : harendra.kumar@gmail.com

20-- Stability : experimental

21-- Portability : GHC

22−--

23−--

24−#ifdef DIAGNOSTICS_VERBOSE

25−#define DIAGNOSTICS

26−#endif

28module Streamly.SVar

23module Streamly.SVar

29 (

24 (

@@ -45,6 +40,8 @@ module Streamly.SVar

45 , setStreamLatency

40 , setStreamLatency

46 , getYieldLimit

41 , getYieldLimit

47 , setYieldLimit

42 , setYieldLimit

43+ , getInspectMode

44+ , setInspectMode

49 , cleanupSVar

46 , cleanupSVar

50 , cleanupSVarFromWorker

47 , cleanupSVarFromWorker

@@ -99,17 +96,18 @@ module Streamly.SVar

99 , toStreamVar

96 , toStreamVar

100 , SVarStats (..)

97 , SVarStats (..)

101 , NanoSecs (..)

98 , NanoSecs (..)

102−#ifdef DIAGNOSTICS

103 , dumpSVar

99 , dumpSVar

104−#endif

105 )

100 )

106where

101where

107

102

108import Control.Concurrent

103import Control.Concurrent

109 (ThreadId, myThreadId, threadDelay, throwTo)

104 (ThreadId, myThreadId, threadDelay, throwTo)

110import Control.Concurrent.MVar

105import Control.Concurrent.MVar

111− (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar)

106+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, tryReadMVar)

107+import Control.Exception

112−import Control.Exception (SomeException(..), catch, mask, assert, Exception)

108+ (SomeException(..), catch, mask, assert, Exception, catches,

109+ throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),

110+ BlockedIndefinitelyOnSTM(..))

113import Control.Monad (when)

111import Control.Monad (when)

114import Control.Monad.Catch (MonadThrow)

112import Control.Monad.Catch (MonadThrow)

115import Control.Monad.IO.Class (MonadIO(..))

113import Control.Monad.IO.Class (MonadIO(..))

@@ -130,24 +128,12 @@ import GHC.Conc (ThreadId(..))

130import GHC.Exts

128import GHC.Exts

131import GHC.IO (IO(..))

129import GHC.IO (IO(..))

132import System.Clock (TimeSpec, Clock(Monotonic), getTime, toNanoSecs)

130import System.Clock (TimeSpec, Clock(Monotonic), getTime, toNanoSecs)

131+import System.IO (hPutStrLn, stderr)

132+import Text.Printf (printf)

133

134import qualified Data.Heap as H

135import qualified Data.Set as S

136

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−

151-- Always use signed arithmetic to avoid inadvertant overflows of signed values

137-- Always use signed arithmetic to avoid inadvertant overflows of signed values

152-- on conversion when comparing unsigned quantities with signed.

138-- on conversion when comparing unsigned quantities with signed.

153newtype NanoSecs = NanoSecs Int64

139newtype NanoSecs = NanoSecs Int64

@@ -371,13 +357,14 @@ data SVar t m a = SVar

371 , svarStats :: SVarStats

357 , svarStats :: SVarStats

372 -- to track garbage collection of SVar

358 -- to track garbage collection of SVar

373 , svarRef :: Maybe (IORef ())

359 , svarRef :: Maybe (IORef ())

374−#ifdef DIAGNOSTICS

360+

361+ -- Only for diagnostics

362+ , svarInspectMode :: Bool

375 , svarCreator :: ThreadId

363 , svarCreator :: ThreadId

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

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

377 , Maybe Int)

365 , Maybe Int)

378 -- Shared work queue (stream, seqNo)

366 -- Shared work queue (stream, seqNo)

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

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

380−#endif

381 }

368 }

382

369

383-------------------------------------------------------------------------------

370-------------------------------------------------------------------------------

@@ -401,6 +388,7 @@ data State t m a = State

401 -- XXX these two can be collapsed into a single type

388 -- XXX these two can be collapsed into a single type

402 , _streamLatency :: Maybe NanoSecs -- bootstrap latency

389 , _streamLatency :: Maybe NanoSecs -- bootstrap latency

403 , _maxStreamRate :: Maybe Rate

390 , _maxStreamRate :: Maybe Rate

391+ , _inspectMode :: Bool

404 }

392 }

405

393

406-------------------------------------------------------------------------------

394-------------------------------------------------------------------------------

@@ -428,6 +416,7 @@ defState = State

428 , _bufferHigh = defaultMaxBuffer

416 , _bufferHigh = defaultMaxBuffer

429 , _maxStreamRate = Nothing

417 , _maxStreamRate = Nothing

430 , _streamLatency = Nothing

418 , _streamLatency = Nothing

419+ , _inspectMode = False

431 }

420 }

432

421

433-- XXX if perf gets affected we can have all the Nothing params in a single

422-- XXX if perf gets affected we can have all the Nothing params in a single

@@ -505,6 +494,12 @@ setStreamLatency n st =

505getStreamLatency :: State t m a -> Maybe NanoSecs

494getStreamLatency :: State t m a -> Maybe NanoSecs

506getStreamLatency = _streamLatency

495getStreamLatency = _streamLatency

507

496

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+

508-------------------------------------------------------------------------------

503-------------------------------------------------------------------------------

509-- Cleanup

504-- Cleanup

510-------------------------------------------------------------------------------

505-------------------------------------------------------------------------------

@@ -526,7 +521,6 @@ cleanupSVarFromWorker sv = do

526-- Dumping the SVar for debug/diag

521-- Dumping the SVar for debug/diag

527-------------------------------------------------------------------------------

522-------------------------------------------------------------------------------

528

523

529−#ifdef DIAGNOSTICS

530-- | Convert a number of seconds to a string. The string will consist

524-- | Convert a number of seconds to a string. The string will consist

531-- of four decimal places, followed by a short description of the time

525-- of four decimal places, followed by a short description of the time

532-- units.

526-- units.

@@ -553,8 +547,8 @@ secs k

553 | otherwise = printf "%.3f %s" t u

547 | otherwise = printf "%.3f %s" t u

554

548

555-- XXX Code duplicated from collectLatency

549-- XXX Code duplicated from collectLatency

556−drainLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)

550+drainLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)

557−drainLatency _ss yinfo = do

551+drainLatency sv yinfo = do

558 let cur = workerPendingLatency yinfo

552 let cur = workerPendingLatency yinfo

559 col = workerCollectedLatency yinfo

553 col = workerCollectedLatency yinfo

560 longTerm = svarAllTimeLatency yinfo

554 longTerm = svarAllTimeLatency yinfo

@@ -574,21 +568,19 @@ drainLatency _ss yinfo = do

574 if (pendingCount > 0)

568 if (pendingCount > 0)

575 then do

569 then do

576 let new = pendingTime `div` (fromIntegral pendingCount)

570 let new = pendingTime `div` (fromIntegral pendingCount)

577−#ifdef DIAGNOSTICS

571+ when (svarInspectMode sv) $ do

572+ let ss = svarStats sv

578− minLat <- readIORef (minWorkerLatency _ss)

573+ minLat <- readIORef (minWorkerLatency ss)

579− when (new < minLat || minLat == 0) $

574+ when (new < minLat || minLat == 0) $

580− writeIORef (minWorkerLatency _ss) new

575+ writeIORef (minWorkerLatency ss) new

581−

576+

582− maxLat <- readIORef (maxWorkerLatency _ss)

577+ maxLat <- readIORef (maxWorkerLatency ss)

583− when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new

578+ when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new

584−#endif

579+ modifyIORef (avgWorkerLatency ss) $

580+ \(cnt, t) -> (cnt + pendingCount, t + pendingTime)

585 -- To avoid minor fluctuations update in batches

581 -- To avoid minor fluctuations update in batches

586 writeIORef col (0, 0)

582 writeIORef col (0, 0)

587 writeIORef measured new

583 writeIORef measured new

588−#ifdef DIAGNOSTICS

589− modifyIORef (avgWorkerLatency _ss) $

590− \(cnt, t) -> (cnt + pendingCount, t + pendingTime)

591−#endif

592 modifyIORef longTerm $ \(_, t) -> (lcount', t)

584 modifyIORef longTerm $ \(_, t) -> (lcount', t)

593 return (lcount', ltime, new)

585 return (lcount', ltime, new)

594 else return notUpdated

586 else return notUpdated

@@ -598,7 +590,7 @@ dumpSVarStats sv ss style = do

598 case yieldRateInfo sv of

590 case yieldRateInfo sv of

599 Nothing -> return ()

591 Nothing -> return ()

600 Just yinfo -> do

592 Just yinfo -> do

601− _ <- liftIO $ drainLatency (svarStats sv) yinfo

593+ _ <- liftIO $ drainLatency sv yinfo

602 return ()

594 return ()

603

595

604 dispatches <- readIORef $ totalDispatches ss

596 dispatches <- readIORef $ totalDispatches ss

@@ -662,7 +654,7 @@ dumpSVarStats sv ss style = do

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

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

663dumpSVar sv = do

655dumpSVar sv = do

664 (oqList, oqLen) <- readIORef $ outputQueue sv

656 (oqList, oqLen) <- readIORef $ outputQueue sv

665− db <- tryTakeMVar $ outputDoorBell sv

657+ db <- tryReadMVar $ outputDoorBell sv

666 aheadDump <-

658 aheadDump <-

667 if svarStyle sv == AheadVar

659 if svarStyle sv == AheadVar

668 then do

660 then do

@@ -686,15 +678,17 @@ dumpSVar sv = do

686 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)

678 stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)

687

679

688 return $ unlines

680 return $ unlines

681+ [

689− [ "Creator tid = " ++ show (svarCreator sv)

682+ "Creator tid = " ++ show (svarCreator sv),

690− , "style = " ++ show (svarStyle sv)

683+ "style = " ++ show (svarStyle sv)

691 , "---------CURRENT STATE-----------"

684 , "---------CURRENT STATE-----------"

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

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

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

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

694 -- XXX print the types of events in the outputQueue, first 5

687 -- XXX print the types of events in the outputQueue, first 5

695 , "outputDoorBell = " ++ show db

688 , "outputDoorBell = " ++ show db

696 ]

689 ]

697− ++ aheadDump ++ unlines

690+ ++ aheadDump

691+ ++ unlines

698 [ "needDoorBell = " ++ show waiting

692 [ "needDoorBell = " ++ show waiting

699 , "running threads = " ++ show rthread

693 , "running threads = " ++ show rthread

700 -- XXX print the status of first 5 threads

694 -- XXX print the status of first 5 threads

@@ -703,6 +697,11 @@ dumpSVar sv = do

703 ++ "---------STATS-----------\n"

697 ++ "---------STATS-----------\n"

704 ++ stats

698 ++ stats

705

699

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+

706{-# NOINLINE mvarExcHandler #-}

705{-# NOINLINE mvarExcHandler #-}

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

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

708mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do

707mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do

@@ -717,15 +716,14 @@ stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do

717 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo

716 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo

718 throwIO e

717 throwIO e

719

718

720−withDBGMVar :: SVar t m a -> String -> IO () -> IO ()

719+withDiagMVar :: SVar t m a -> String -> IO () -> IO ()

721−withDBGMVar sv label action =

720+withDiagMVar sv label action =

721+ if svarInspectMode sv

722+ then

722− action `catches` [ Handler (mvarExcHandler sv label)

723+ action `catches` [ Handler (mvarExcHandler sv label)

723− , Handler (stmExcHandler sv label)

724+ , Handler (stmExcHandler sv label)

724− ]

725+ ]

725−#else

726+ else action

726−withDBGMVar :: SVar t m a -> String -> IO () -> IO ()

727−withDBGMVar _ _ action = action

728−#endif

729

727

730-------------------------------------------------------------------------------

728-------------------------------------------------------------------------------

731-- CAS

729-- CAS

@@ -1248,22 +1246,19 @@ handleChildException sv e = do

1248 tid <- myThreadId

1246 tid <- myThreadId

1249 void $ send sv (ChildStop tid (Just e))

1247 void $ send sv (ChildStop tid (Just e))

1250

1248

1251−#ifdef DIAGNOSTICS

1249+{-# NOINLINE recordMaxWorkers #-}

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

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

1253recordMaxWorkers sv = liftIO $ do

1251recordMaxWorkers sv = liftIO $ do

1254 active <- readIORef (workerCount sv)

1252 active <- readIORef (workerCount sv)

1255 maxWrk <- readIORef (maxWorkers $ svarStats sv)

1253 maxWrk <- readIORef (maxWorkers $ svarStats sv)

1256 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active

1254 when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active

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

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

1258−#endif

1259

1256

1260{-# NOINLINE pushWorker #-}

1257{-# NOINLINE pushWorker #-}

1261pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()

1258pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()

1262pushWorker yieldMax sv = do

1259pushWorker yieldMax sv = do

1263 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1

1260 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1

1264−#ifdef DIAGNOSTICS

1265− recordMaxWorkers sv

1261+ when (svarInspectMode sv) $ recordMaxWorkers sv

1266−#endif

1267 -- This allocation matters when significant number of workers are being

1262 -- This allocation matters when significant number of workers are being

1268 -- sent. We allocate it only when needed.

1263 -- sent. We allocate it only when needed.

1269 winfo <-

1264 winfo <-

@@ -1288,34 +1283,39 @@ pushWorker yieldMax sv = do

1288-- producer side. So we need to use a thread safe modification of

1283-- producer side. So we need to use a thread safe modification of

1289-- workerThreads. Alternatively, we can use a CreateThread event to avoid

1284-- workerThreads. Alternatively, we can use a CreateThread event to avoid

1290-- using a CAS based modification.

1285-- using a CAS based modification.

1291−{-# NOINLINE pushWorkerPar #-}

1286+{-# INLINE pushWorkerPar #-}

1287+pushWorkerPar

1288+ :: MonadAsync m

1292−pushWorkerPar :: MonadAsync m => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()

1289+ => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()

1293pushWorkerPar sv wloop =

1290pushWorkerPar sv wloop =

1294− -- We do not use workerCount in case of ParallelVar but still there is no

1291+ if svarInspectMode sv

1295− -- harm in maintaining it correctly.

1292+ then forkWithDiag

1296−#ifdef DIAGNOSTICS

1293+ else doFork (wloop Nothing) (handleChildException sv) >>= modifyThread sv

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− }

1314

1294

1315− doFork (wloop winfo) (handleChildException sv) >>= modifyThread sv

1295+ where

1316−#else

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+

1317− doFork (wloop Nothing) (handleChildException sv) >>= modifyThread sv

1318+ doFork (wloop winfo) (handleChildException sv) >>= modifyThread sv

1318−#endif

1319

1320-- Returns:

1321-- True: can dispatch more

@@ -1547,8 +1547,8 @@ updateWorkerPollingInterval yinfo latency = do

1547-- whereas the average is used for future estimates e.g. how many workers

1548-- should be maintained to maintain the rate.

1549-- CAUTION! keep it in sync with getWorkerLatency

1550−collectLatency :: SVarStats -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)

1550+collectLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)

1551−collectLatency _ss yinfo = do

1551+collectLatency sv yinfo = do

1552 let cur = workerPendingLatency yinfo

1553 col = workerCollectedLatency yinfo

1554 longTerm = svarAllTimeLatency yinfo

@@ -1567,15 +1567,15 @@ collectLatency _ss yinfo = do

1567

1568 if pendingCount > 0

1569 then do

1570− let new = pendingTime `div` fromIntegral pendingCount

1570+ let new = pendingTime `div` (fromIntegral pendingCount)

1571−#ifdef DIAGNOSTICS

1571+ when (svarInspectMode sv) $ do

1572+ let ss = svarStats sv

1572− minLat <- readIORef (minWorkerLatency _ss)

1573+ minLat <- readIORef (minWorkerLatency ss)

1573− when (new < minLat || minLat == 0) $

1574+ when (new < minLat || minLat == 0) $

1574− writeIORef (minWorkerLatency _ss) new

1575+ writeIORef (minWorkerLatency ss) new

1575−

1576+

1576− maxLat <- readIORef (maxWorkerLatency _ss)

1577+ maxLat <- readIORef (maxWorkerLatency ss)

1577− when (new > maxLat) $ writeIORef (maxWorkerLatency _ss) new

1578+ when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new

1578−#endif

1579 -- When we have collected a significant sized batch we compute the new

1580 -- latency using that batch and return the new latency, otherwise we

1581 -- return the previous latency derived from the previous batch.

@@ -1585,13 +1585,13 @@ collectLatency _ss yinfo = do

1585 in prev > 0 && (r > 2 || r < 0.5))

1586 || (prev == 0)

1587 then do

1588+ when (svarInspectMode sv) $ do

1589+ let ss = svarStats sv

1590+ modifyIORef (avgWorkerLatency ss) $

1591+ \(cnt, t) -> (cnt + pendingCount, t + pendingTime)

1588 updateWorkerPollingInterval yinfo (max new prev)

1592 updateWorkerPollingInterval yinfo (max new prev)

1589 writeIORef col (0, 0)

1593 writeIORef col (0, 0)

1590 writeIORef measured ((prev + new) `div` 2)

1594 writeIORef measured ((prev + new) `div` 2)

1591−#ifdef DIAGNOSTICS

1592− modifyIORef (avgWorkerLatency _ss) $

1593− \(cnt, t) -> (cnt + pendingCount, t + pendingTime)

1594−#endif

1595 modifyIORef longTerm $ \(_, t) -> (lcount', t)

1596 return $ tripleWith new

1597 else do

@@ -1613,7 +1613,7 @@ dispatchWorkerPaced sv = do

1613 (svarYields, svarElapsed, wLatency) <- do

1614 now <- liftIO $ getTime Monotonic

1615 (yieldCount, baseTime, lat) <-

1616− liftIO $ collectLatency (svarStats sv) yinfo

1616+ liftIO $ collectLatency sv yinfo

1617 let elapsed = fromInteger $ toNanoSecs $ now - baseTime

1618 let latency =

1619 if lat == 0

@@ -1811,7 +1811,7 @@ sendWorkerWait delay dispatch sv = do

1811 if canDoMore

1812 then sendWorkerWait delay dispatch sv

1813 else do

1814− liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"

1814+ liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"

1815 $ takeMVar (outputDoorBell sv)

1816 (_, len) <- liftIO $ readIORef (outputQueue sv)

1817 when (len <= 0) $ sendWorkerWait delay dispatch sv

@@ -1820,10 +1820,10 @@ sendWorkerWait delay dispatch sv = do

1820readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)

1821readOutputQRaw sv = do

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

1823−#ifdef DIAGNOSTICS

1823+ when (svarInspectMode sv) $ do

1824+ let ref = maxOutQSize $ svarStats sv

1824− oqLen <- readIORef (maxOutQSize $ svarStats sv)

1825+ oqLen <- readIORef ref

1825− when (len > oqLen) $ writeIORef (maxOutQSize $ svarStats sv) len

1826+ when (len > oqLen) $ writeIORef ref len

1826−#endif

1827 return (list, len)

1828

1829readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]

@@ -1985,9 +1985,7 @@ getAheadSVar st f = do

1985 maxLat <- newIORef (NanoSecs 0)

1986 minLat <- newIORef (NanoSecs 0)

1987 stpTime <- newIORef Nothing

1988−#ifdef DIAGNOSTICS

1989 tid <- myThreadId

1988 tid <- myThreadId

1990−#endif

1991

1989

1992 let getSVar sv readOutput postProc = SVar

1990 let getSVar sv readOutput postProc = SVar

1993 { outputQueue = outQ

1991 { outputQueue = outQ

@@ -2009,11 +2007,10 @@ getAheadSVar st f = do

2009 , accountThread = delThread sv

2007 , accountThread = delThread sv

2010 , workerStopMVar = stopMVar

2008 , workerStopMVar = stopMVar

2011 , svarRef = Nothing

2009 , svarRef = Nothing

2012−#ifdef DIAGNOSTICS

2010+ , svarInspectMode = getInspectMode st

2013 , svarCreator = tid

2011 , svarCreator = tid

2014 , aheadWorkQueue = q

2012 , aheadWorkQueue = q

2015 , outputHeap = outH

2013 , outputHeap = outH

2016−#endif

2017 , svarStats = SVarStats

2014 , svarStats = SVarStats

2018 { totalDispatches = disp

2015 { totalDispatches = disp

2019 , maxWorkers = maxWrk

2016 , maxWorkers = maxWrk

@@ -2081,9 +2078,7 @@ getParallelSVar st = do

2081 maxLat <- newIORef (NanoSecs 0)

2078 maxLat <- newIORef (NanoSecs 0)

2082 minLat <- newIORef (NanoSecs 0)

2079 minLat <- newIORef (NanoSecs 0)

2083 stpTime <- newIORef Nothing

2080 stpTime <- newIORef Nothing

2084−#ifdef DIAGNOSTICS

2085 tid <- myThreadId

2081 tid <- myThreadId

2086−#endif

2087

2082

2088 let sv =

2083 let sv =

2089 SVar { outputQueue = outQ

2084 SVar { outputQueue = outQ

@@ -2106,11 +2101,10 @@ getParallelSVar st = do

2106 , accountThread = modifyThread sv

2101 , accountThread = modifyThread sv

2107 , workerStopMVar = undefined

2102 , workerStopMVar = undefined

2108 , svarRef = Nothing

2103 , svarRef = Nothing

2109−#ifdef DIAGNOSTICS

2104+ , svarInspectMode = getInspectMode st

2110 , svarCreator = tid

2105 , svarCreator = tid

2111 , aheadWorkQueue = undefined

2106 , aheadWorkQueue = undefined

2112 , outputHeap = undefined

2107 , outputHeap = undefined

2113−#endif

2114 , svarStats = SVarStats

2108 , svarStats = SVarStats

2115 { totalDispatches = disp

2109 { totalDispatches = disp

2116 , maxWorkers = maxWrk

2110 , maxWorkers = maxWrk

@@ -2128,10 +2122,11 @@ getParallelSVar st = do

2128 where

2122 where

2129

2123

2130 readOutputQPar sv = liftIO $ do

2124 readOutputQPar sv = liftIO $ do

2125+ withDiagMVar sv "readOutputQPar: doorbell"

2131− withDBGMVar sv "readOutputQPar: doorbell" $ takeMVar (outputDoorBell sv)

2126+ $ takeMVar (outputDoorBell sv)

2132 case yieldRateInfo sv of

2127 case yieldRateInfo sv of

2133 Nothing -> return ()

2128 Nothing -> return ()

2134− Just yinfo -> void $ collectLatency (svarStats sv) yinfo

2129+ Just yinfo -> void $ collectLatency sv yinfo

2135 fst `fmap` readOutputQRaw sv

2130 fst `fmap` readOutputQRaw sv

2136

2131

2137sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)

2132sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)

src/Streamly/Streams/Ahead.hs+10−23

@@ -8,10 +8,6 @@

8{-# LANGUAGE StandaloneDeriving #-}

9{-# LANGUAGE UndecidableInstances #-} -- XXX

11−#ifdef DIAGNOSTICS_VERBOSE

12−#define DIAGNOSTICS

13−#endif

14−

15-- |

11-- |

16-- Module : Streamly.Streams.Ahead

12-- Module : Streamly.Streams.Ahead

@@ -32,7 +28,7 @@ module Streamly.Streams.Ahead

32where

28where

34import Control.Concurrent.MVar (putMVar, takeMVar)

30import Control.Concurrent.MVar (putMVar, takeMVar)

35−import Control.Monad (ap, void)

31+import Control.Monad (ap, void, when)

36import Control.Monad.Base (MonadBase(..), liftBaseDefault)

32import Control.Monad.Base (MonadBase(..), liftBaseDefault)

37import Control.Monad.Catch (MonadThrow, throwM)

33import Control.Monad.Catch (MonadThrow, throwM)

38-- import Control.Monad.Error.Class (MonadError(..))

34-- import Control.Monad.Error.Class (MonadError(..))

@@ -41,7 +37,7 @@ import Control.Monad.Reader.Class (MonadReader(..))

41import Control.Monad.State.Class (MonadState(..))

37import Control.Monad.State.Class (MonadState(..))

42import Control.Monad.Trans.Class (MonadTrans(lift))

38import Control.Monad.Trans.Class (MonadTrans(lift))

43import Data.Heap (Heap, Entry(..))

39import Data.Heap (Heap, Entry(..))

44−import Data.IORef (IORef, readIORef, atomicModifyIORef)

40+import Data.IORef (IORef, readIORef, atomicModifyIORef, writeIORef)

45import Data.Maybe (fromJust)

41import Data.Maybe (fromJust)

46import Data.Semigroup (Semigroup(..))

42import Data.Semigroup (Semigroup(..))

47import GHC.Exts (inline)

43import GHC.Exts (inline)

@@ -54,10 +50,6 @@ import Streamly.SVar

54import Streamly.Streams.StreamK (IsStream(..), Stream(..))

50import Streamly.Streams.StreamK (IsStream(..), Stream(..))

55import qualified Streamly.Streams.StreamK as K

51import qualified Streamly.Streams.StreamK as K

57−#ifdef DIAGNOSTICS

58−import Control.Monad (when)

59−import Data.IORef (writeIORef)

60−#endif

61import Prelude hiding (map)

53import Prelude hiding (map)

63#include "Instances.hs"

55#include "Instances.hs"

@@ -360,12 +352,12 @@ processWithoutToken q heap st sv winfo m sno = do

360 let hp' = H.insert (Entry seqNo ent) hp

352 let hp' = H.insert (Entry seqNo ent) hp

361 in ((hp', snum), hp')

353 in ((hp', snum), hp')

362

354

363−#ifdef DIAGNOSTICS

355+ when (svarInspectMode sv) $

364− liftIO $ do

356+ liftIO $ do

365− maxHp <- readIORef (maxHeapSize $ svarStats sv)

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

366− when (H.size newHp > maxHp) $

358+ when (H.size newHp > maxHp) $

367− writeIORef (maxHeapSize $ svarStats sv) (H.size newHp)

359+ writeIORef (maxHeapSize $ svarStats sv) (H.size newHp)

368−#endif

360+

369 heapOk <- liftIO $ underMaxHeap sv newHp

361 heapOk <- liftIO $ underMaxHeap sv newHp

370 let drainAndStop = drainHeap q heap st sv winfo

362 let drainAndStop = drainHeap q heap st sv winfo

371 mainLoop = workLoopAhead q heap st sv winfo

363 mainLoop = workLoopAhead q heap st sv winfo

@@ -486,6 +478,8 @@ processWithToken q heap st sv winfo action sno = do

486-- hooks can be used for a more general implementation to even check predicates

478-- hooks can be used for a more general implementation to even check predicates

487-- and not just yield limit.

479-- and not just yield limit.

488

480

481+-- XXX we can remove the sv parameter as it can be derived from st

482+

489workLoopAhead :: MonadIO m

483workLoopAhead :: MonadIO m

490 => IORef ([Stream m a], Int)

484 => IORef ([Stream m a], Int)

491 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

485 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

@@ -494,13 +488,6 @@ workLoopAhead :: MonadIO m

494 -> Maybe WorkerInfo

488 -> Maybe WorkerInfo

495 -> m ()

489 -> m ()

496workLoopAhead q heap st sv winfo = do

490workLoopAhead q heap st sv winfo = do

497−#ifdef DIAGNOSTICS

498− liftIO $ do

499− maxHp <- readIORef (maxHeapSize $ svarStats sv)

500− (hp, _) <- readIORef heap

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

502− (H.size hp)

503−#endif

504 r <- liftIO $ dequeueFromHeap heap

491 r <- liftIO $ dequeueFromHeap heap

505 case r of

492 case r of

506 Ready (Entry seqNo hent) ->

493 Ready (Entry seqNo hent) ->

src/Streamly/Streams/Async.hs+32−20

@@ -6,13 +6,10 @@

6{-# LANGUAGE InstanceSigs #-}

7{-# LANGUAGE LambdaCase #-}

8{-# LANGUAGE MultiParamTypeClasses #-}

9+{-# LANGUAGE ScopedTypeVariables #-}

9{-# LANGUAGE StandaloneDeriving #-}

10{-# LANGUAGE StandaloneDeriving #-}

10{-# LANGUAGE UndecidableInstances #-} -- XXX

11{-# LANGUAGE UndecidableInstances #-} -- XXX

12−#ifdef DIAGNOSTICS_VERBOSE

13−#define DIAGNOSTICS

14−#endif

15−

16-- |

13-- |

17-- Module : Streamly.Streams.Async

14-- Module : Streamly.Streams.Async

@@ -40,6 +37,7 @@ module Streamly.Streams.Async

40 )

37 )

41where

38where

40+import Control.Concurrent (myThreadId)

43import Control.Monad (ap)

41import Control.Monad (ap)

44import Control.Monad.Base (MonadBase(..), liftBaseDefault)

42import Control.Monad.Base (MonadBase(..), liftBaseDefault)

45import Control.Monad.Catch (MonadThrow, throwM)

43import Control.Monad.Catch (MonadThrow, throwM)

@@ -63,10 +61,6 @@ import Streamly.SVar

63import Streamly.Streams.StreamK (IsStream(..), Stream(..), adapt)

61import Streamly.Streams.StreamK (IsStream(..), Stream(..), adapt)

64import qualified Streamly.Streams.StreamK as K

62import qualified Streamly.Streams.StreamK as K

66−#ifdef DIAGNOSTICS

67−import Control.Concurrent (myThreadId)

68−#endif

69−

70#include "Instances.hs"

64#include "Instances.hs"

72-------------------------------------------------------------------------------

66-------------------------------------------------------------------------------

@@ -172,6 +166,8 @@ workLoopLIFOLimited q st sv winfo = run

172-- WAsync

166-- WAsync

173-------------------------------------------------------------------------------

167-------------------------------------------------------------------------------

174

168

169+-- XXX we can remove sv as it is derivable from st

170+

175{-# INLINE workLoopFIFO #-}

171{-# INLINE workLoopFIFO #-}

176workLoopFIFO

172workLoopFIFO

177 :: MonadIO m

173 :: MonadIO m

@@ -253,7 +249,8 @@ workLoopFIFOLimited q st sv winfo = run

253-- function argument to this function results in a perf degradation of more

249-- function argument to this function results in a perf degradation of more

254-- than 10%. Need to investigate what the root cause is.

250-- than 10%. Need to investigate what the root cause is.

255-- Interestingly, the same thing does not make any difference for Ahead.

251-- Interestingly, the same thing does not make any difference for Ahead.

252+getLifoSVar :: forall m a. MonadAsync m

256−getLifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)

253+ => State Stream m a -> IO (SVar Stream m a)

257getLifoSVar st = do

254getLifoSVar st = do

258 outQ <- newIORef ([], 0)

255 outQ <- newIORef ([], 0)

259 outQMv <- newEmptyMVar

256 outQMv <- newEmptyMVar

@@ -275,9 +272,7 @@ getLifoSVar st = do

275 maxLat <- newIORef (NanoSecs 0)

272 maxLat <- newIORef (NanoSecs 0)

276 minLat <- newIORef (NanoSecs 0)

273 minLat <- newIORef (NanoSecs 0)

277 stpTime <- newIORef Nothing

274 stpTime <- newIORef Nothing

278−#ifdef DIAGNOSTICS

279 tid <- myThreadId

275 tid <- myThreadId

280−#endif

281

276

282 let isWorkFinished _ = null <$> readIORef q

277 let isWorkFinished _ = null <$> readIORef q

283

278

@@ -291,7 +286,17 @@ getLifoSVar st = do

291 qEmpty <- null <$> readIORef q

286 qEmpty <- null <$> readIORef q

292 return $ qEmpty || yieldsDone

287 return $ qEmpty || yieldsDone

293

288

289+ let getSVar :: SVar Stream m a

290+ -> (SVar Stream m a -> m [ChildEvent a])

291+ -> (SVar Stream m a -> m Bool)

292+ -> (SVar Stream m a -> IO Bool)

293+ -> (IORef [Stream m a]

294+ -> State Stream m a

295+ -> SVar Stream m a

296+ -> Maybe WorkerInfo

297+ -> m())

298+ -> SVar Stream m a

294− let getSVar sv readOutput postProc workDone wloop = SVar

299+ getSVar sv readOutput postProc workDone wloop = SVar

295 { outputQueue = outQ

300 { outputQueue = outQ

296 , remainingWork = yl

301 , remainingWork = yl

297 , maxBufferLimit = getMaxBuffer st

302 , maxBufferLimit = getMaxBuffer st

@@ -311,11 +316,10 @@ getLifoSVar st = do

311 , accountThread = delThread sv

316 , accountThread = delThread sv

312 , workerStopMVar = undefined

317 , workerStopMVar = undefined

313 , svarRef = Nothing

318 , svarRef = Nothing

314−#ifdef DIAGNOSTICS

319+ , svarInspectMode = getInspectMode st

315 , svarCreator = tid

320 , svarCreator = tid

316 , aheadWorkQueue = undefined

321 , aheadWorkQueue = undefined

317 , outputHeap = undefined

322 , outputHeap = undefined

318−#endif

319 , svarStats = SVarStats

323 , svarStats = SVarStats

320 { totalDispatches = disp

324 { totalDispatches = disp

321 , maxWorkers = maxWrk

325 , maxWorkers = maxWrk

@@ -353,7 +357,8 @@ getLifoSVar st = do

353 workLoopLIFOLimited

357 workLoopLIFOLimited

354 in return sv

358 in return sv

355

359

360+getFifoSVar :: forall m a. MonadAsync m

356−getFifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)

361+ => State Stream m a -> IO (SVar Stream m a)

357getFifoSVar st = do

362getFifoSVar st = do

358 outQ <- newIORef ([], 0)

363 outQ <- newIORef ([], 0)

359 outQMv <- newEmptyMVar

364 outQMv <- newEmptyMVar

@@ -375,9 +380,7 @@ getFifoSVar st = do

375 maxLat <- newIORef (NanoSecs 0)

380 maxLat <- newIORef (NanoSecs 0)

376 minLat <- newIORef (NanoSecs 0)

381 minLat <- newIORef (NanoSecs 0)

377 stpTime <- newIORef Nothing

382 stpTime <- newIORef Nothing

378−#ifdef DIAGNOSTICS

379 tid <- myThreadId

383 tid <- myThreadId

380−#endif

381

384

382 let isWorkFinished _ = nullQ q

385 let isWorkFinished _ = nullQ q

383 let isWorkFinishedLimited sv = do

386 let isWorkFinishedLimited sv = do

@@ -390,7 +393,17 @@ getFifoSVar st = do

390 qEmpty <- nullQ q

393 qEmpty <- nullQ q

391 return $ qEmpty || yieldsDone

394 return $ qEmpty || yieldsDone

392

395

396+ let getSVar :: SVar Stream m a

397+ -> (SVar Stream m a -> m [ChildEvent a])

398+ -> (SVar Stream m a -> m Bool)

399+ -> (SVar Stream m a -> IO Bool)

400+ -> (LinkedQueue (Stream m a)

401+ -> State Stream m a

402+ -> SVar Stream m a

403+ -> Maybe WorkerInfo

404+ -> m())

405+ -> SVar Stream m a

393− let getSVar sv readOutput postProc workDone wloop = SVar

406+ getSVar sv readOutput postProc workDone wloop = SVar

394 { outputQueue = outQ

407 { outputQueue = outQ

395 , remainingWork = yl

408 , remainingWork = yl

396 , maxBufferLimit = getMaxBuffer st

409 , maxBufferLimit = getMaxBuffer st

@@ -410,11 +423,10 @@ getFifoSVar st = do

410 , accountThread = delThread sv

423 , accountThread = delThread sv

411 , workerStopMVar = undefined

424 , workerStopMVar = undefined

412 , svarRef = Nothing

425 , svarRef = Nothing

413−#ifdef DIAGNOSTICS

426+ , svarInspectMode = getInspectMode st

414 , svarCreator = tid

427 , svarCreator = tid

415 , aheadWorkQueue = undefined

428 , aheadWorkQueue = undefined

416 , outputHeap = undefined

429 , outputHeap = undefined

417−#endif

418 , svarStats = SVarStats

430 , svarStats = SVarStats

419 { totalDispatches = disp

431 { totalDispatches = disp

420 , maxWorkers = maxWrk

432 , maxWorkers = maxWrk

src/Streamly/Streams/SVar.hs+32−29

@@ -11,10 +11,6 @@

12#include "inline.h"

14−#ifdef DIAGNOSTICS_VERBOSE

15−#define DIAGNOSTICS

16−#endif

17−

18-- |

14-- |

19-- Module : Streamly.Streams.SVar

15-- Module : Streamly.Streams.SVar

@@ -37,37 +33,38 @@ module Streamly.Streams.SVar

37 , minRate

33 , minRate

38 , maxRate

34 , maxRate

39 , constRate

35 , constRate

36+ , inspectMode

37+ , printState

40 )

38 )

41where

39where

43import Control.Exception (fromException)

41import Control.Exception (fromException)

42+import Control.Monad (when)

44import Control.Monad.Catch (throwM)

43import Control.Monad.Catch (throwM)

44+import Control.Monad.IO.Class (MonadIO(liftIO))

45import Data.Int (Int64)

46−import Control.Monad.IO.Class (liftIO)

47−import Data.IORef (newIORef, mkWeakIORef)

46+import Data.IORef (newIORef, readIORef, mkWeakIORef)

48−#ifdef DIAGNOSTICS

49import Data.IORef (writeIORef)

47import Data.IORef (writeIORef)

48+import Data.Maybe (isNothing)

50import System.IO (hPutStrLn, stderr)

49import System.IO (hPutStrLn, stderr)

51import System.Clock (Clock(Monotonic), getTime)

50import System.Clock (Clock(Monotonic), getTime)

52−#endif

51+import System.Mem (performMajorGC)

54import Streamly.SVar

53import Streamly.SVar

55import Streamly.Streams.StreamK

54import Streamly.Streams.StreamK

56import Streamly.Streams.Serial (SerialT)

55import Streamly.Streams.Serial (SerialT)

58−-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we

59−-- can keep it on in production to debug problems quickly if and when they

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

61−-- until they are GCed because the consumer went away.

62−

63−#ifdef DIAGNOSTICS

64−#ifdef DIAGNOSTICS_VERBOSE

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

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

66printSVar sv how = do

58printSVar sv how = do

67 svInfo <- dumpSVar sv

59 svInfo <- dumpSVar sv

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

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

69−#endif

61+

70−#endif

62+printState :: MonadIO m => State Stream m a -> m ()

63+printState st = liftIO $ do

64+ let msv = streamVar st

65+ case msv of

66+ Just sv -> dumpSVar sv >>= putStrLn

67+ Nothing -> putStrLn "No SVar"

72-- | Pull a stream from an SVar.

69-- | Pull a stream from an SVar.

73{-# NOINLINE fromStreamVar #-}

70{-# NOINLINE fromStreamVar #-}

@@ -81,16 +78,12 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do

82 where

79 where

84− allDone stp =

81+ allDone stp = do

85−#ifdef DIAGNOSTICS

82+ when (svarInspectMode sv) $ do

86− do

87 t <- liftIO $ getTime Monotonic

83 t <- liftIO $ getTime Monotonic

88 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)

84 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)

89−#ifdef DIAGNOSTICS_VERBOSE

90 liftIO $ printSVar sv "SVar Done"

85 liftIO $ printSVar sv "SVar Done"

91−#endif

92−#endif

93− stp

86+ stp

95 {-# INLINE processEvents #-}

88 {-# INLINE processEvents #-}

96 processEvents [] = Stream $ \st stp sng yld -> do

89 processEvents [] = Stream $ \st stp sng yld -> do

@@ -125,12 +118,15 @@ fromSVar sv =

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

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

126 where

119 where

127

120

128− hook =

121+ hook = do

129−#ifdef DIAGNOSTICS_VERBOSE

122+ when (svarInspectMode sv) $ do

130− do

123+ r <- liftIO $ readIORef (svarStopTime (svarStats sv))

124+ when (isNothing r) $

131− printSVar sv "SVar Garbage Collected"

125+ printSVar sv "SVar Garbage Collected"

132−#endif

133 cleanupSVar sv

126 cleanupSVar sv

127+ -- If there are any SVars referenced by this SVar a GC will prompt

128+ -- them to be cleaned up quickly.

129+ when (svarInspectMode sv) performMajorGC

134

130

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

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

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

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

@@ -215,6 +211,8 @@ rate r m = fromStream $ Stream $ \st stp sng yld ->

215 error "rate: Minimum rate cannot be greater than maximum rate."

211 error "rate: Minimum rate cannot be greater than maximum rate."

216 _ -> unStream (toStream m) (setStreamRate r st) stp sng yld

212 _ -> unStream (toStream m) (setStreamRate r st) stp sng yld

217

213

214+-- XXX implement for serial streams as well, as a simple delay

215+

218{-

216{-

219{-# RULES "rate serial" rate = yieldRateSerial #-}

217{-# RULES "rate serial" rate = yieldRateSerial #-}

220yieldRateSerial :: Double -> SerialT m a -> SerialT m a

218yieldRateSerial :: Double -> SerialT m a -> SerialT m a

@@ -304,3 +302,8 @@ maxYields n m = fromStream $ Stream $ \st stp sng yld ->

304{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}

302{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}

305maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a

303maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a

306maxYieldsSerial _ = id

304maxYieldsSerial _ = id

305+

306+-- | Print debug information about an SVar when the stream ends

307+inspectMode :: IsStream t => t m a -> t m a

308+inspectMode m = fromStream $ Stream $ \st stp sng yld -> do

309+ unStream (toStream m) (setInspectMode st) stp sng yld

stack.yaml+1−3

@@ -8,9 +8,7 @@ extra-deps:

8 - Chart-1.9

9 - Chart-diagrams-1.9

10 - SVGFonts-1.6.0.3

11−

11+ - bench-show-0.2.0

12− - git: https://github.com/composewell/bench-show

13− commit: 38cb1b547d1a7ac62de09d59e28f4a1f71dba31b

15flags: {}

13flags: {}

16extra-package-dbs: []

14extra-package-dbs: []

streamly.cabal+1−9

@@ -87,11 +87,6 @@ source-repository head

87 type: git

88 location: https://github.com/composewell/streamly

90−flag diag

91− description: Diagnostics build

92− manual: True

93− default: False

94−

95flag dev

90flag dev

96 description: Development build

91 description: Development build

97 manual: True

92 manual: True

@@ -138,6 +133,7 @@ library

138 , Streamly.Time

133 , Streamly.Time

139 , Streamly

134 , Streamly

140 , Streamly.Tutorial

135 , Streamly.Tutorial

136+ , Streamly.Internal

141

137

142 default-language: Haskell2010

138 default-language: Haskell2010

143 ghc-options: -Wall

139 ghc-options: -Wall

@@ -148,11 +144,7 @@ library

148 if flag(no-fusion)

144 if flag(no-fusion)

149 cpp-options: -DDISABLE_FUSION

145 cpp-options: -DDISABLE_FUSION

150

146

151− if flag(diag)

152− cpp-options: -DDIAGNOSTICS

153−

154 if flag(dev)

147 if flag(dev)

155− cpp-options: -DDIAGNOSTICS

156 ghc-options: -Wmissed-specialisations

148 ghc-options: -Wmissed-specialisations

157 -Wall-missed-specialisations

149 -Wall-missed-specialisations

158 -fno-ignore-asserts

150 -fno-ignore-asserts