51fb1b8 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move common parts of Channel out of Async module

51fb1b8Harendra Kumar4 years ago

src/Streamly/Internal/Data/Stream/Async/Channel.hs+3−2

compare with master

unifiedsplit

@@ -43,7 +43,7 @@ where

43import Control.Monad.IO.Class (MonadIO(liftIO))

44import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

45import Streamly.Internal.Data.Stream.Async.Channel.Operations

46− (fromChannel, toChannel, toChannelK)

46+ (fromChannel, fromChannelK, toChannel, toChannelK)

47import Streamly.Internal.Data.SVar.Type (adaptState)

49import qualified Streamly.Internal.Data.Stream as Stream

@@ -53,6 +53,7 @@ import qualified Streamly.Internal.Data.Stream.Async.Channel.Interleave

53import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

55import Streamly.Internal.Data.Stream.Async.Channel.Type

56+import Streamly.Internal.Data.Stream.Channel.Types

57-- | Allocate a channel and evaluate the stream using the channel and the

58-- | Allocate a channel and evaluate the stream using the channel and the

58-- supplied evaluator function. The evaluator is run in a worker thread.

59-- supplied evaluator function. The evaluator is run in a worker thread.

@@ -69,7 +70,7 @@ withChannelK evaluator modifier input = K.concatEffect action

69 action = do

70 action = do

70 chan <- Append.newChannel modifier

71 chan <- Append.newChannel modifier

71 toChannelK chan (evaluator chan input)

72 toChannelK chan (evaluator chan input)

72− return $ Stream.toStreamK $ fromChannel chan

73+ return $ fromChannelK chan

74{-# INLINE withInterleaveChannelK #-}

75{-# INLINE withInterleaveChannelK #-}

75withInterleaveChannelK :: MonadAsync m =>

76withInterleaveChannelK :: MonadAsync m =>

src/Streamly/Internal/Data/Stream/Async/Channel/Append.hs+21−21

@@ -28,14 +28,14 @@ import Streamly.Internal.Control.Concurrent

28 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)

29import Streamly.Internal.Data.Atomics

30 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

31+import Streamly.Internal.Data.Stream.Channel.Dispatcher (delThread)

32import qualified Data.Set as Set

33import qualified Data.Set as Set

33import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

34import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

35import Streamly.Internal.Data.Stream.Async.Channel.Consumer

36import Streamly.Internal.Data.Stream.Async.Channel.Consumer

36−import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

37import Streamly.Internal.Data.Stream.Async.Channel.Type

38−import Streamly.Internal.Data.Stream.Async.Channel.Worker

38+import Streamly.Internal.Data.Stream.Channel.Types

40------------------------------------------------------------------------------

41-- Creating a channel

@@ -53,7 +53,7 @@ enqueueLIFO ::

53 -> IO ()

54enqueueLIFO sv q m = do

55 atomicModifyIORefCAS_ q $ \ms -> m : ms

56− ringDoorBell sv

56+ ringDoorBell (needDoorBell sv) (outputDoorBell sv)

58data WorkerStatus = Continue | Suspend

@@ -69,11 +69,10 @@ workLoopLIFO q sv winfo = run

70 where

72− stop = liftIO $ sendStop sv winfo

73 run = do

72 run = do

74 work <- dequeue

73 work <- dequeue

75 case work of

74 case work of

76− Nothing -> stop

75+ Nothing -> liftIO $ stop sv winfo

77 Just (RunInIO runin, m) -> do

76 Just (RunInIO runin, m) -> do

78 -- XXX when we finish we need to send the monadic state back to

77 -- XXX when we finish we need to send the monadic state back to

79 -- the parent so that the state can be merged back. We capture

78 -- the parent so that the state can be merged back. We capture

@@ -92,14 +91,14 @@ workLoopLIFO q sv winfo = run

92 res <- restoreM r

91 res <- restoreM r

93 case res of

92 case res of

94 Continue -> run

93 Continue -> run

95− Suspend -> stop

94+ Suspend -> liftIO $ stop sv winfo

97 single a = do

96 single a = do

98− res <- liftIO $ sendYield sv winfo (ChildYield a)

97+ res <- liftIO $ yield sv winfo a

99 return $ if res then Continue else Suspend

98 return $ if res then Continue else Suspend

100

101 yieldk a r = do

100 yieldk a r = do

102− res <- liftIO $ sendYield sv winfo (ChildYield a)

101+ res <- liftIO $ yield sv winfo a

103 if res

102 if res

104 then K.foldStreamShared undefined yieldk single (return Continue) r

103 then K.foldStreamShared undefined yieldk single (return Continue) r

105 else do

104 else do

@@ -128,12 +127,13 @@ workLoopLIFOLimited q sv winfo = run

128

127

129 where

128 where

130

129

130+ incrContinue =

131− incrContinue = liftIO (incrementYieldLimit sv) >> return Continue

131+ liftIO (incrementYieldLimit (remainingWork sv)) >> return Continue

132− stop = liftIO $ sendStop sv winfo

132+

133 run = do

134 work <- dequeue

135 case work of

136− Nothing -> stop

136+ Nothing -> liftIO $ stop sv winfo

137 Just (RunInIO runin, m) -> do

138 -- XXX This is just a best effort minimization of concurrency

139 -- to the yield limit. If the stream is made of concurrent

@@ -141,7 +141,7 @@ workLoopLIFOLimited q sv winfo = run

141 -- streams before executing the action. This can be done

142 -- though, by sharing the yield limit ref with downstream

143 -- actions via state passing. Just a todo.

144− yieldLimitOk <- liftIO $ decrementYieldLimit sv

144+ yieldLimitOk <- liftIO $ decrementYieldLimit (remainingWork sv)

145 if yieldLimitOk

146 then do

147 r <- liftIO $ runin $

@@ -154,28 +154,28 @@ workLoopLIFOLimited q sv winfo = run

154 res <- restoreM r

155 case res of

156 Continue -> run

157− Suspend -> stop

157+ Suspend -> liftIO $ stop sv winfo

158 -- Avoid any side effects, undo the yield limit decrement if we

159 -- never yielded anything.

160 else liftIO $ do

161 enqueueLIFO sv q (RunInIO runin, m)

162− incrementYieldLimit sv

162+ incrementYieldLimit (remainingWork sv)

163− sendStop sv winfo

163+ stop sv winfo

164

165 single a = do

166− res <- liftIO $ sendYield sv winfo (ChildYield a)

166+ res <- liftIO $ yield sv winfo a

167 return $ if res then Continue else Suspend

168

169 -- XXX can we pass on the yield limit downstream to limit the concurrency

170 -- of constituent streams.

171 yieldk a r = do

172− res <- liftIO $ sendYield sv winfo (ChildYield a)

172+ res <- liftIO $ yield sv winfo a

173− yieldLimitOk <- liftIO $ decrementYieldLimit sv

173+ yieldLimitOk <- liftIO $ decrementYieldLimit (remainingWork sv)

174 if res && yieldLimitOk

175 then K.foldStreamShared undefined yieldk single incrContinue r

176 else do

177 runInIO <- askRunInIO

178− liftIO $ incrementYieldLimit sv

178+ liftIO $ incrementYieldLimit (remainingWork sv)

179 liftIO $ enqueueLIFO sv q (runInIO, r)

180 return Suspend

181

@@ -205,7 +205,7 @@ getLifoSVar mrun cfg = do

205 yl <- case getYieldLimit cfg of

206 Nothing -> return Nothing

207 Just x -> Just <$> newIORef x

208− rateInfo <- getYieldRateInfo cfg

208+ rateInfo <- newRateInfo cfg

209

210 stats <- newSVarStats

211 tid <- myThreadId

@@ -248,7 +248,7 @@ getLifoSVar mrun cfg = do

248 , needDoorBell = wfw

249 , svarMrun = mrun

250 , workerCount = active

251− , accountThread = delThread sv

251+ , accountThread = delThread running

252 , workerStopMVar = undefined

253 , svarRef = Nothing

254 , svarInspectMode = getInspectMode cfg

src/Streamly/Internal/Data/Stream/Async/Channel/Consumer.hs+17−40

@@ -9,52 +9,39 @@

9module Streamly.Internal.Data.Stream.Async.Channel.Consumer

10 (

11 -- * Read Output

12− readOutputQBasic

13− , readOutputQRaw

14− , readOutputQPaced

12+ readOutputQPaced

15 , readOutputQBounded

13 , readOutputQBounded

17 -- * Postprocess Hook After Reading

15 -- * Postprocess Hook After Reading

18 , postProcessPaced

16 , postProcessPaced

19 , postProcessBounded

17 , postProcessBounded

20−

21− -- * Release Resources

22− , cleanupSVar

23 )

18 )

24where

19where

26import Control.Monad.Trans.Control (MonadBaseControl)

21import Control.Monad.Trans.Control (MonadBaseControl)

27−import Control.Concurrent (throwTo)

28import Control.Monad (when, void)

22import Control.Monad (when, void)

29import Control.Monad.IO.Class (MonadIO(liftIO))

23import Control.Monad.IO.Class (MonadIO(liftIO))

30−import Data.IORef (IORef, readIORef, writeIORef)

24+import Data.IORef (readIORef)

31−import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)

33−import Streamly.Internal.Data.Stream.Async.Channel.Type

34import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

26import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

27+import Streamly.Internal.Data.Stream.Async.Channel.Type

28+import Streamly.Internal.Data.Stream.Channel.Dispatcher

29+import Streamly.Internal.Data.Stream.Channel.Types

36-------------------------------------------------------------------------------

31-------------------------------------------------------------------------------

37-- Reading from the workers' output queue/buffer

32-- Reading from the workers' output queue/buffer

38-------------------------------------------------------------------------------

33-------------------------------------------------------------------------------

40−{-# INLINE readOutputQBasic #-}

35+{-# INLINE readOutputQChan #-}

41−readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)

42−readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)

43−

44−{-# INLINE readOutputQRaw #-}

45−readOutputQRaw :: Channel m a -> IO ([ChildEvent a], Int)

36+readOutputQChan :: Channel m a -> IO ([ChildEvent a], Int)

46−readOutputQRaw sv = do

37+readOutputQChan sv = do

47− (list, len) <- readOutputQBasic (outputQueue sv)

38+ let ss = if svarInspectMode sv then Just (svarStats sv) else Nothing

48− when (svarInspectMode sv) $ do

39+ in readOutputQRaw (outputQueue sv) ss

49− let ref = maxOutQSize $ svarStats sv

50− oqLen <- readIORef ref

51− when (len > oqLen) $ writeIORef ref len

52− return (list, len)

54readOutputQBounded :: (MonadIO m, MonadBaseControl IO m) =>

41readOutputQBounded :: (MonadIO m, MonadBaseControl IO m) =>

55 Channel m a -> m [ChildEvent a]

42 Channel m a -> m [ChildEvent a]

56readOutputQBounded sv = do

43readOutputQBounded sv = do

57− (list, len) <- liftIO $ readOutputQRaw sv

44+ (list, len) <- liftIO $ readOutputQChan sv

58 -- When there is no output seen we dispatch more workers to help

45 -- When there is no output seen we dispatch more workers to help

59 -- out if there is work pending in the work queue.

46 -- out if there is work pending in the work queue.

60 if len <= 0

47 if len <= 0

@@ -78,12 +65,12 @@ readOutputQBounded sv = do

78 {-# INLINE blockingRead #-}

65 {-# INLINE blockingRead #-}

79 blockingRead = do

66 blockingRead = do

80 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv

67 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv

81− liftIO (fst `fmap` readOutputQRaw sv)

68+ liftIO (fst `fmap` readOutputQChan sv)

83readOutputQPaced :: (MonadIO m, MonadBaseControl IO m) =>

70readOutputQPaced :: (MonadIO m, MonadBaseControl IO m) =>

84 Channel m a -> m [ChildEvent a]

71 Channel m a -> m [ChildEvent a]

85readOutputQPaced sv = do

72readOutputQPaced sv = do

86− (list, len) <- liftIO $ readOutputQRaw sv

73+ (list, len) <- liftIO $ readOutputQChan sv

87 if len <= 0

74 if len <= 0

88 then blockingRead

75 then blockingRead

89 else do

76 else do

@@ -97,11 +84,11 @@ readOutputQPaced sv = do

97 {-# INLINE blockingRead #-}

84 {-# INLINE blockingRead #-}

98 blockingRead = do

85 blockingRead = do

99 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv

86 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv

100− liftIO (fst `fmap` readOutputQRaw sv)

87+ liftIO (fst `fmap` readOutputQChan sv)

101

102postProcessPaced :: (MonadIO m, MonadBaseControl IO m) => Channel m a -> m Bool

89postProcessPaced :: (MonadIO m, MonadBaseControl IO m) => Channel m a -> m Bool

103postProcessPaced sv = do

90postProcessPaced sv = do

104− workersDone <- allThreadsDone sv

91+ workersDone <- allThreadsDone (workerThreads sv)

105 -- XXX If during consumption we figure out we are getting delayed then we

92 -- XXX If during consumption we figure out we are getting delayed then we

106 -- should trigger dispatch there as well. We should try to check on the

93 -- should trigger dispatch there as well. We should try to check on the

107 -- workers after consuming every n item from the buffer?

94 -- workers after consuming every n item from the buffer?

@@ -113,7 +100,7 @@ postProcessPaced sv = do

113 -- Note that we need to guarantee a worker since the work is not

100 -- Note that we need to guarantee a worker since the work is not

114 -- finished, therefore we cannot just rely on dispatchWorkerPaced

101 -- finished, therefore we cannot just rely on dispatchWorkerPaced

115 -- which may or may not send a worker.

102 -- which may or may not send a worker.

116− noWorker <- allThreadsDone sv

103+ noWorker <- allThreadsDone (workerThreads sv)

117 when noWorker $ pushWorker 0 sv

104 when noWorker $ pushWorker 0 sv

118 return r

105 return r

119 else return False

106 else return False

@@ -121,7 +108,7 @@ postProcessPaced sv = do

121postProcessBounded :: (MonadIO m, MonadBaseControl IO m) =>

108postProcessBounded :: (MonadIO m, MonadBaseControl IO m) =>

122 Channel m a -> m Bool

109 Channel m a -> m Bool

123postProcessBounded sv = do

110postProcessBounded sv = do

124− workersDone <- allThreadsDone sv

111+ workersDone <- allThreadsDone (workerThreads sv)

125 -- There may still be work pending even if there are no workers pending

112 -- There may still be work pending even if there are no workers pending

126 -- because all the workers may return if the outputQueue becomes full. In

113 -- because all the workers may return if the outputQueue becomes full. In

127 -- that case send off a worker to kickstart the work again.

114 -- that case send off a worker to kickstart the work again.

@@ -141,13 +128,3 @@ postProcessBounded sv = do

141 -- void $ dispatchWorker sv

128 -- void $ dispatchWorker sv

142 return r

129 return r

143 else return False

130 else return False

144−

145−-------------------------------------------------------------------------------

146−-- Cleanup

147−-------------------------------------------------------------------------------

148−

149−cleanupSVar :: Channel m a -> IO ()

150−cleanupSVar sv = do

151− workers <- readIORef (workerThreads sv)

152− Prelude.mapM_ (`throwTo` ThreadAbort)

153− workers

src/Streamly/Internal/Data/Stream/Async/Channel/Dispatcher.hs+28−337

@@ -9,22 +9,8 @@

9--

10module Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

11 (

12− -- * Latency collection

13− collectLatency

14−

15− -- * Diagnostics

16− , withDiagMVar

17− , dumpSVar

18− , printSVar

19−

20− -- * Thread accounting

21− , delThread

22− , modifyThread

23− , allThreadsDone

24−

25 -- * Dispatching

12 -- * Dispatching

26− , recordMaxWorkers

27− , pushWorker

13+ pushWorker

28 , dispatchWorker

14 , dispatchWorker

29 , dispatchWorkerPaced

15 , dispatchWorkerPaced

30 , sendWorkerWait

16 , sendWorkerWait

@@ -35,338 +21,33 @@ module Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

35where

21where

37import Control.Monad.Trans.Control (MonadBaseControl)

23import Control.Monad.Trans.Control (MonadBaseControl)

38−import Control.Concurrent (takeMVar, tryReadMVar, ThreadId, threadDelay)

24+import Control.Concurrent (takeMVar, threadDelay)

39−import Control.Concurrent.MVar (tryPutMVar)

25+import Control.Exception (assert)

40−import Control.Exception

41− (assert, catches, throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),

42− BlockedIndefinitelyOnSTM(..))

43import Control.Monad (when, void)

26import Control.Monad (when, void)

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

27import Control.Monad.IO.Class (MonadIO(liftIO))

45import Data.Maybe (fromJust, fromMaybe)

28import Data.Maybe (fromJust, fromMaybe)

46−import Data.IORef (IORef, modifyIORef, newIORef, readIORef, writeIORef)

29+import Data.IORef (modifyIORef, newIORef, readIORef, writeIORef)

47import Streamly.Internal.Control.ForkLifted (doFork)

30import Streamly.Internal.Control.ForkLifted (doFork)

48−import Streamly.Internal.Data.Atomics

31+import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_, storeLoadBarrier)

49− (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

50− storeLoadBarrier)

51import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

32import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

52import Streamly.Internal.Data.Time.Units

33import Streamly.Internal.Data.Time.Units

53− (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,

34+ (MicroSecond64(..), diffAbsTime64, fromRelTime64, toRelTime64)

54− fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)

55−import System.IO (hPutStrLn, stderr)

56−

57−-- import qualified Data.Heap as H

58−import qualified Data.Set as S

60import Streamly.Internal.Data.Stream.Async.Channel.Type

36import Streamly.Internal.Data.Stream.Async.Channel.Type

61−import Streamly.Internal.Data.Stream.Async.Channel.Worker

37+import Streamly.Internal.Data.Stream.Channel.Dispatcher

62−

38+import Streamly.Internal.Data.Stream.Channel.Types

63−-------------------------------------------------------------------------------

39+import Streamly.Internal.Data.Stream.Channel.Worker

64−-- Worker latency data processing

65−-------------------------------------------------------------------------------

66−

67−-- Every once in a while workers update the latencies and check the yield rate.

68−-- They return if we are above the expected yield rate. If we check too often

69−-- it may impact performance, if we check less often we may have a stale

70−-- picture. We update every minThreadDelay but we translate that into a yield

71−-- count based on latency so that the checking overhead is little.

72−--

73−-- XXX use a generation count to indicate that the value is updated. If the

74−-- value is updated an existing worker must check it again on the next yield.

75−-- Otherwise it is possible that we may keep updating it and because of the mod

76−-- worker keeps skipping it.

77−updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()

78−updateWorkerPollingInterval yinfo latency = do

79− let periodRef = workerPollingInterval yinfo

80− cnt = max 1 $ minThreadDelay `div` latency

81− period = min cnt (fromIntegral magicMaxBuffer)

82−

83− writeIORef periodRef (fromIntegral period)

84−

85−{-# INLINE recordMinMaxLatency #-}

86−recordMinMaxLatency :: Channel m a -> NanoSecond64 -> IO ()

87−recordMinMaxLatency sv new = do

88− let ss = svarStats sv

89− minLat <- readIORef (minWorkerLatency ss)

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

91− writeIORef (minWorkerLatency ss) new

92−

93− maxLat <- readIORef (maxWorkerLatency ss)

94− when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new

95−

96−recordAvgLatency :: Channel m a -> (Count, NanoSecond64) -> IO ()

97−recordAvgLatency sv (count, time) = do

98− let ss = svarStats sv

99− modifyIORef (avgWorkerLatency ss) $

100− \(cnt, t) -> (cnt + count, t + time)

101−

102−-- Pour the pending latency stats into a collection bucket

103−{-# INLINE collectWorkerPendingLatency #-}

104−collectWorkerPendingLatency

105− :: IORef (Count, Count, NanoSecond64)

106− -> IORef (Count, Count, NanoSecond64)

107− -> IO (Count, Maybe (Count, NanoSecond64))

108−collectWorkerPendingLatency cur col = do

109− (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)

110−

111− (fcnt, cnt, t) <- readIORef col

112− let totalCount = fcnt + fcount

113− latCount = cnt + count

114− latTime = t + time

115− writeIORef col (totalCount, latCount, latTime)

116−

117− assert (latCount == 0 || latTime /= 0) (return ())

118− let latPair =

119− if latCount > 0 && latTime > 0

120− then Just (latCount, latTime)

121− else Nothing

122− return (totalCount, latPair)

123−

124−{-# INLINE shouldUseCollectedBatch #-}

125−shouldUseCollectedBatch

126− :: Count

127− -> NanoSecond64

128− -> NanoSecond64

129− -> NanoSecond64

130− -> Bool

131−shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =

132− let r = fromIntegral newLat / fromIntegral prevLat :: Double

133− in (collectedYields > fromIntegral magicMaxBuffer)

134− || (collectedTime > minThreadDelay)

135− || (prevLat > 0 && (r > 2 || r < 0.5))

136− || (prevLat == 0)

137−

138−-- Returns a triple, (1) yield count since last collection, (2) the base time

139−-- when we started counting, (3) average latency in the last measurement

140−-- period. The former two are used for accurate measurement of the going rate

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

142−-- should be maintained to maintain the rate.

143−-- CAUTION! keep it in sync with getWorkerLatency

144−collectLatency :: Channel m a

145− -> YieldRateInfo

146− -> Bool

147− -> IO (Count, AbsTime, NanoSecond64)

148−collectLatency sv yinfo drain = do

149− let cur = workerPendingLatency yinfo

150− col = workerCollectedLatency yinfo

151− longTerm = svarAllTimeLatency yinfo

152− measured = workerMeasuredLatency yinfo

153−

154− (newCount, newLatPair) <- collectWorkerPendingLatency cur col

155− (lcount, ltime) <- readIORef longTerm

156− prevLat <- readIORef measured

157−

158− let newLcount = lcount + newCount

159− retWith lat = return (newLcount, ltime, lat)

160−

161− case newLatPair of

162− Nothing -> retWith prevLat

163− Just (count, time) -> do

164− let newLat = time `div` fromIntegral count

165− when (svarInspectMode sv) $ recordMinMaxLatency sv newLat

166− -- When we have collected a significant sized batch we compute the

167− -- new latency using that batch and return the new latency,

168− -- otherwise we return the previous latency derived from the

169− -- previous batch.

170− if shouldUseCollectedBatch newCount time newLat prevLat || drain

171− then do

172− -- XXX make this NOINLINE?

173− updateWorkerPollingInterval yinfo (max newLat prevLat)

174− when (svarInspectMode sv) $ recordAvgLatency sv (count, time)

175− writeIORef col (0, 0, 0)

176− writeIORef measured ((prevLat + newLat) `div` 2)

177− modifyIORef longTerm $ \(_, t) -> (newLcount, t)

178− retWith newLat

179− else retWith prevLat

180−

181−-------------------------------------------------------------------------------

182−-- Dumping the SVar for debug/diag

183−-------------------------------------------------------------------------------

184−

185−dumpSVarStats :: Channel m a -> SVarStats -> IO String

186−dumpSVarStats sv ss = do

187− case yieldRateInfo sv of

188− Nothing -> return ()

189− Just yinfo -> do

190− _ <- liftIO $ collectLatency sv yinfo True

191− return ()

192−

193− dispatches <- readIORef $ totalDispatches ss

194− maxWrk <- readIORef $ maxWorkers ss

195− maxOq <- readIORef $ maxOutQSize ss

196− -- maxHp <- readIORef $ maxHeapSize ss

197− minLat <- readIORef $ minWorkerLatency ss

198− maxLat <- readIORef $ maxWorkerLatency ss

199− (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss

200− (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of

201− Nothing -> return (0, 0, 0)

202− Just yinfo -> do

203− (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo

204− if cnt > 0

205− then do

206− t <- readIORef (svarStopTime ss)

207− gl <- readIORef (svarGainedLostYields yinfo)

208− case t of

209− Nothing -> do

210− now <- getTime Monotonic

211− let interval = diffAbsTime64 now startTime

212− return (cnt, gl, interval `div` fromIntegral cnt)

213− Just stopTime -> do

214− let interval = diffAbsTime64 stopTime startTime

215− return (cnt, gl, interval `div` fromIntegral cnt)

216− else return (0, 0, 0)

217−

218− return $ unlines

219− [ "total dispatches = " <> show dispatches

220− , "max workers = " <> show maxWrk

221− , "max outQSize = " <> show maxOq

222− <> (if minLat > 0

223− then "\nmin worker latency = " <> showNanoSecond64 minLat

224− else "")

225− <> (if maxLat > 0

226− then "\nmax worker latency = " <> showNanoSecond64 maxLat

227− else "")

228− <> (if avgCnt > 0

229− then let lat = avgTime `div` fromIntegral avgCnt

230− in "\navg worker latency = " <> showNanoSecond64 lat

231− else "")

232− <> (if svarLat > 0

233− then "\nSVar latency = " <> showRelTime64 svarLat

234− else "")

235− <> (if svarCnt > 0

236− then "\nSVar yield count = " <> show svarCnt

237− else "")

238− <> (if svarGainLossCnt > 0

239− then "\nSVar gain/loss yield count = " <> show svarGainLossCnt

240− else "")

241− ]

242−

243−{-# NOINLINE dumpSVar #-}

244−dumpSVar :: Channel m a -> IO String

245−dumpSVar sv = do

246− (oqList, oqLen) <- readIORef $ outputQueue sv

247− db <- tryReadMVar $ outputDoorBell sv

248−

249− waiting <- readIORef $ needDoorBell sv

250− rthread <- readIORef $ workerThreads sv

251− workers <- readIORef $ workerCount sv

252− stats <- dumpSVarStats sv (svarStats sv)

253−

254− return $ unlines

255− [

256− "Creator tid = " <> show (svarCreator sv)

257− , "---------CURRENT STATE-----------"

258− , "outputQueue length computed = " <> show (length oqList)

259− , "outputQueue length maintained = " <> show oqLen

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

261− , "outputDoorBell = " <> show db

262− ]

263− <> unlines

264− [ "needDoorBell = " <> show waiting

265− , "running threads = " <> show rthread

266− -- XXX print the status of first 5 threads

267− , "running thread count = " <> show workers

268− ]

269− <> "---------STATS-----------\n"

270− <> stats

271−

272−-- MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we

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

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

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

276−

277−{-# NOINLINE mvarExcHandler #-}

278−mvarExcHandler :: Channel m a -> String -> BlockedIndefinitelyOnMVar -> IO ()

279−mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do

280− svInfo <- dumpSVar sv

281− hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo

282− throwIO e

283−

284−{-# NOINLINE stmExcHandler #-}

285−stmExcHandler :: Channel m a -> String -> BlockedIndefinitelyOnSTM -> IO ()

286−stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do

287− svInfo <- dumpSVar sv

288− hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo

289− throwIO e

290−

291−withDiagMVar :: Channel m a -> String -> IO () -> IO ()

292−withDiagMVar sv label action =

293− if svarInspectMode sv

294− then

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

296− , Handler (stmExcHandler sv label)

297− ]

298− else action

299−

300−printSVar :: Channel m a -> String -> IO ()

301−printSVar sv how = do

302− svInfo <- dumpSVar sv

303− hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo

304−

305−-------------------------------------------------------------------------------

306−-- Thread accounting

307−-------------------------------------------------------------------------------

308−

309−-- Thread tracking is needed for two reasons:

310−--

311−-- 1) Killing threads on exceptions. Threads may not be left to go away by

312−-- themselves because they may run for significant times before going away or

313−-- worse they may be stuck in IO and never go away.

314−--

315−-- 2) To know when all threads are done and the stream has ended.

316−

317−{-# NOINLINE addThread #-}

318−addThread :: MonadIO m => Channel m a -> ThreadId -> m ()

319−addThread sv tid =

320− liftIO $ modifyIORef (workerThreads sv) (S.insert tid)

321−

322−-- This is cheaper than modifyThread because we do not have to send a

323−-- outputDoorBell This can make a difference when more workers are being

324−-- dispatched.

325−{-# INLINE delThread #-}

326−delThread :: MonadIO m => Channel m a -> ThreadId -> m ()

327−delThread sv tid =

328− liftIO $ modifyIORef (workerThreads sv) (S.delete tid)

329−

330−-- If present then delete else add. This takes care of out of order add and

331−-- delete i.e. a delete arriving before we even added a thread.

332−-- This occurs when the forked thread is done even before the 'addThread' right

333−-- after the fork gets a chance to run.

334−{-# INLINE modifyThread #-}

335−modifyThread :: MonadIO m => Channel m a -> ThreadId -> m ()

336−modifyThread sv tid = do

337− changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->

338− if S.member tid old

339− then let new = S.delete tid old in (new, new)

340− else let new = S.insert tid old in (new, old)

341− when (null changed) $

342− liftIO $ do

343− writeBarrier

344− void $ tryPutMVar (outputDoorBell sv) ()

345−

346−-- | This is safe even if we are adding more threads concurrently because if

347−-- a child thread is adding another thread then anyway 'workerThreads' will

348−-- not be empty.

349−{-# INLINE allThreadsDone #-}

350−allThreadsDone :: MonadIO m => Channel m a -> m Bool

351−allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)

352

353-------------------------------------------------------------------------------

41-------------------------------------------------------------------------------

354-- Dispatching workers

42-- Dispatching workers

355-------------------------------------------------------------------------------

43-------------------------------------------------------------------------------

356

357−{-# NOINLINE recordMaxWorkers #-}

358−recordMaxWorkers :: MonadIO m => Channel m a -> m ()

359−recordMaxWorkers sv = liftIO $ do

360− active <- readIORef (workerCount sv)

361− maxWrk <- readIORef (maxWorkers $ svarStats sv)

362− when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active

363− modifyIORef (totalDispatches $ svarStats sv) (+1)

364−

365{-# NOINLINE pushWorker #-}

45{-# NOINLINE pushWorker #-}

366pushWorker :: (MonadIO m, MonadBaseControl IO m) => Count -> Channel m a -> m ()

46pushWorker :: (MonadIO m, MonadBaseControl IO m) => Count -> Channel m a -> m ()

367pushWorker yieldMax sv = do

47pushWorker yieldMax sv = do

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

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

369− when (svarInspectMode sv) $ recordMaxWorkers sv

49+ when (svarInspectMode sv)

50+ $ recordMaxWorkers (workerCount sv) (svarStats sv)

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

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

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

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

372 winfo <-

53 winfo <-

@@ -381,8 +62,12 @@ pushWorker yieldMax sv = do

381 , workerYieldCount = cntRef

62 , workerYieldCount = cntRef

382 , workerLatencyStart = lat

63 , workerLatencyStart = lat

383 }

64 }

384− doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)

65+ doFork (workLoop sv winfo) (svarMrun sv) exception

385− >>= addThread sv

66+ >>= addThread (workerThreads sv)

67+

68+ where

69+

70+ exception = handleChildException (outputQueue sv) (outputDoorBell sv)

386

387-- Returns:

72-- Returns:

388-- True: can dispatch more

73-- True: can dispatch more

@@ -454,7 +139,9 @@ dispatchWorkerPaced sv = do

454 (svarYields, svarElapsed, wLatency) <- do

139 (svarYields, svarElapsed, wLatency) <- do

455 now <- liftIO $ getTime Monotonic

140 now <- liftIO $ getTime Monotonic

456 (yieldCount, baseTime, lat) <-

141 (yieldCount, baseTime, lat) <-

142+ liftIO

457− liftIO $ collectLatency sv yinfo False

143+ $ collectLatency

144+ (svarInspectMode sv) (svarStats sv) yinfo False

458 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime

145 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime

459 let latency =

146 let latency =

460 if lat == 0

147 if lat == 0

@@ -492,7 +179,7 @@ dispatchWorkerPaced sv = do

492 -- here as it is not a reliable way to ensure there are

179 -- here as it is not a reliable way to ensure there are

493 -- definitely no active workers. When workerCount is 0 we may

180 -- definitely no active workers. When workerCount is 0 we may

494 -- still have a Stop event waiting in the outputQueue.

181 -- still have a Stop event waiting in the outputQueue.

495− done <- allThreadsDone sv

182+ done <- allThreadsDone (workerThreads sv)

496 when done $ void $ do

183 when done $ void $ do

497 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64

184 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64

498 liftIO $ threadDelay (fromIntegral us)

185 liftIO $ threadDelay (fromIntegral us)

@@ -502,7 +189,7 @@ dispatchWorkerPaced sv = do

502 assert (yields > 0) (return ())

189 assert (yields > 0) (return ())

503 updateGainedLostYields yinfo yields

190 updateGainedLostYields yinfo yields

504

191

505− done <- allThreadsDone sv

192+ done <- allThreadsDone (workerThreads sv)

506 when done $ void $ dispatchWorker yields sv

193 when done $ void $ dispatchWorker yields sv

507 return False

194 return False

508 ManyWorkers netWorkers yields -> do

195 ManyWorkers netWorkers yields -> do

@@ -627,8 +314,12 @@ sendWorkerWait delay dispatch sv = do

627 if canDoMore

314 if canDoMore

628 then sendWorkerWait delay dispatch sv

315 then sendWorkerWait delay dispatch sv

629 else do

316 else do

317+ liftIO

318+ $ withDiagMVar

319+ (svarInspectMode sv)

320+ (dumpSVar sv)

630− liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"

321+ "sendWorkerWait: nothing to do"

631− $ takeMVar (outputDoorBell sv)

322+ $ takeMVar (outputDoorBell sv)

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

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

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

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

634

325

src/Streamly/Internal/Data/Stream/Async/Channel/Interleave.hs+21−21

@@ -22,14 +22,14 @@ import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pu

22import Data.IORef (newIORef, readIORef)

23import Streamly.Internal.Control.Concurrent

24 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)

25+import Streamly.Internal.Data.Stream.Channel.Dispatcher (delThread)

26import qualified Data.Set as Set

27import qualified Data.Set as Set

27import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

28import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

29import Streamly.Internal.Data.Stream.Async.Channel.Consumer

30import Streamly.Internal.Data.Stream.Async.Channel.Consumer

30−import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

31import Streamly.Internal.Data.Stream.Async.Channel.Type

32−import Streamly.Internal.Data.Stream.Async.Channel.Worker

32+import Streamly.Internal.Data.Stream.Channel.Types

34------------------------------------------------------------------------------

35-- Creating a channel

@@ -49,7 +49,7 @@ enqueueFIFO ::

49 -> IO ()

50enqueueFIFO sv q m = do

51 pushL q m

52− ringDoorBell sv

52+ ringDoorBell (needDoorBell sv) (outputDoorBell sv)

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

@@ -64,11 +64,10 @@ workLoopFIFO q sv winfo = run

65 where

67− stop = liftIO $ sendStop sv winfo

68 run = do

67 run = do

69 work <- liftIO $ tryPopR q

68 work <- liftIO $ tryPopR q

70 case work of

69 case work of

71− Nothing -> stop

70+ Nothing -> liftIO $ stop sv winfo

72 Just (RunInIO runin, m) -> do

71 Just (RunInIO runin, m) -> do

73 r <- liftIO

72 r <- liftIO

74 $ runin

73 $ runin

@@ -77,10 +76,10 @@ workLoopFIFO q sv winfo = run

77 res <- restoreM r

76 res <- restoreM r

78 case res of

77 case res of

79 Continue -> run

78 Continue -> run

80− Suspend -> stop

79+ Suspend -> liftIO $ stop sv winfo

82 single a = do

81 single a = do

83− res <- liftIO $ sendYield sv winfo (ChildYield a)

82+ res <- liftIO $ yield sv winfo a

84 return $ if res then Continue else Suspend

83 return $ if res then Continue else Suspend

86 -- XXX in general we would like to yield "n" elements from a single stream

85 -- XXX in general we would like to yield "n" elements from a single stream

@@ -88,7 +87,7 @@ workLoopFIFO q sv winfo = run

88 -- expensive in certain cases. Similarly, we can use time limit for

87 -- expensive in certain cases. Similarly, we can use time limit for

89 -- yielding.

88 -- yielding.

90 yieldk a r = do

89 yieldk a r = do

91− res <- liftIO $ sendYield sv winfo (ChildYield a)

90+ res <- liftIO $ yield sv winfo a

92 runInIO <- askRunInIO

91 runInIO <- askRunInIO

93 liftIO $ enqueueFIFO sv q (runInIO, r)

92 liftIO $ enqueueFIFO sv q (runInIO, r)

94 return $ if res then Continue else Suspend

93 return $ if res then Continue else Suspend

@@ -104,14 +103,15 @@ workLoopFIFOLimited q sv winfo = run

104

103

105 where

104 where

106

105

107− stop = liftIO $ sendStop sv winfo

106+ incrContinue =

108− incrContinue = liftIO (incrementYieldLimit sv) >> return Continue

107+ liftIO (incrementYieldLimit (remainingWork sv)) >> return Continue

108+

109 run = do

110 work <- liftIO $ tryPopR q

111 case work of

112− Nothing -> stop

112+ Nothing -> liftIO $ stop sv winfo

113 Just (RunInIO runin, m) -> do

114− yieldLimitOk <- liftIO $ decrementYieldLimit sv

114+ yieldLimitOk <- liftIO $ decrementYieldLimit (remainingWork sv)

115 if yieldLimitOk

116 then do

117 r <- liftIO

@@ -121,25 +121,25 @@ workLoopFIFOLimited q sv winfo = run

121 res <- restoreM r

122 case res of

123 Continue -> run

124− Suspend -> stop

124+ Suspend -> liftIO $ stop sv winfo

125 else liftIO $ do

126 enqueueFIFO sv q (RunInIO runin, m)

127− incrementYieldLimit sv

127+ incrementYieldLimit (remainingWork sv)

128− sendStop sv winfo

128+ stop sv winfo

129

130 single a = do

131− res <- liftIO $ sendYield sv winfo (ChildYield a)

131+ res <- liftIO $ yield sv winfo a

132 return $ if res then Continue else Suspend

133

134 yieldk a r = do

135− res <- liftIO $ sendYield sv winfo (ChildYield a)

135+ res <- liftIO $ yield sv winfo a

136 runInIO <- askRunInIO

137 liftIO $ enqueueFIFO sv q (runInIO, r)

138− yieldLimitOk <- liftIO $ decrementYieldLimit sv

138+ yieldLimitOk <- liftIO $ decrementYieldLimit (remainingWork sv)

139 if res && yieldLimitOk

140 then return Continue

141 else liftIO $ do

142− incrementYieldLimit sv

142+ incrementYieldLimit (remainingWork sv)

143 return Suspend

144

145-------------------------------------------------------------------------------

@@ -162,7 +162,7 @@ getFifoSVar mrun cfg = do

162 yl <- case getYieldLimit cfg of

163 Nothing -> return Nothing

164 Just x -> Just <$> newIORef x

165− rateInfo <- getYieldRateInfo cfg

165+ rateInfo <- newRateInfo cfg

166

167 stats <- newSVarStats

168 tid <- myThreadId

@@ -204,7 +204,7 @@ getFifoSVar mrun cfg = do

204 , needDoorBell = wfw

205 , svarMrun = mrun

206 , workerCount = active

207− , accountThread = delThread sv

207+ , accountThread = delThread running

208 , workerStopMVar = undefined

209 , svarRef = Nothing

210 , svarInspectMode = getInspectMode cfg

src/Streamly/Internal/Data/Stream/Async/Channel/Operations.hs+14−9

@@ -42,9 +42,9 @@ import qualified Streamly.Internal.Data.Stream as Stream

42import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

43import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

45−import Streamly.Internal.Data.Stream.Async.Channel.Consumer

46import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

45import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

47import Streamly.Internal.Data.Stream.Async.Channel.Type

46import Streamly.Internal.Data.Stream.Async.Channel.Type

47+import Streamly.Internal.Data.Stream.Channel.Types

49import Prelude hiding (map, concat, concatMap)

@@ -91,6 +91,7 @@ import Test.Inspection (inspect, hasNoTypeClassesExcept)

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

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

94+{-# INLINE toChannelK #-}

94toChannelK :: (MonadIO m, MonadBaseControl IO m) =>

95toChannelK :: (MonadIO m, MonadBaseControl IO m) =>

95 Channel m a -> K.Stream m a -> m ()

96 Channel m a -> K.Stream m a -> m ()

96toChannelK sv m = do

97toChannelK sv m = do

@@ -132,7 +133,7 @@ fromChannelRaw sv = K.MkStream $ \st yld sng stp -> do

132 when (svarInspectMode sv) $ do

133 when (svarInspectMode sv) $ do

133 t <- liftIO $ getTime Monotonic

134 t <- liftIO $ getTime Monotonic

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

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

135− liftIO $ printSVar sv "SVar Done"

136+ liftIO $ printSVar (dumpSVar sv) "SVar Done"

136 stp

137 stp

137

138

138 {-# INLINE processEvents #-}

139 {-# INLINE processEvents #-}

@@ -154,7 +155,9 @@ fromChannelRaw sv = K.MkStream $ \st yld sng stp -> do

154 case fromException ex of

155 case fromException ex of

155 Just ThreadAbort ->

156 Just ThreadAbort ->

156 K.foldStream st yld sng stp rest

157 K.foldStream st yld sng stp rest

158+ Nothing -> do

157− Nothing -> liftIO (cleanupSVar sv) >> throwM ex

159+ liftIO (cleanupSVar (workerThreads sv))

160+ throwM ex

158

161

159#if __GLASGOW_HASKELL__ < 810

162#if __GLASGOW_HASKELL__ < 810

160#ifdef INSPECTION

163#ifdef INSPECTION

@@ -199,8 +202,8 @@ fromChannelK sv =

199 when (svarInspectMode sv) $ do

202 when (svarInspectMode sv) $ do

200 r <- liftIO $ readIORef (svarStopTime (svarStats sv))

203 r <- liftIO $ readIORef (svarStopTime (svarStats sv))

201 when (isNothing r) $

204 when (isNothing r) $

202− printSVar sv "SVar Garbage Collected"

205+ printSVar (dumpSVar sv) "SVar Garbage Collected"

203− cleanupSVar sv

206+ cleanupSVar (workerThreads sv)

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

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

205 -- them to be cleaned up quickly.

208 -- them to be cleaned up quickly.

206 when (svarInspectMode sv) performMajorGC

209 when (svarInspectMode sv) performMajorGC

@@ -245,8 +248,8 @@ _fromChannelD svar = D.Stream step FromSVarInit

245 when (svarInspectMode svar) $ do

248 when (svarInspectMode svar) $ do

246 r <- liftIO $ readIORef (svarStopTime (svarStats svar))

249 r <- liftIO $ readIORef (svarStopTime (svarStats svar))

247 when (isNothing r) $

250 when (isNothing r) $

248− printSVar svar "SVar Garbage Collected"

251+ printSVar (dumpSVar svar) "SVar Garbage Collected"

249− cleanupSVar svar

252+ cleanupSVar (workerThreads svar)

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

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

251 -- them to be cleaned up quickly.

254 -- them to be cleaned up quickly.

252 when (svarInspectMode svar) performMajorGC

255 when (svarInspectMode svar) performMajorGC

@@ -275,11 +278,13 @@ _fromChannelD svar = D.Stream step FromSVarInit

275 case fromException ex of

278 case fromException ex of

276 Just ThreadAbort ->

279 Just ThreadAbort ->

277 return $ D.Skip (FromSVarLoop sv es)

280 return $ D.Skip (FromSVarLoop sv es)

281+ Nothing -> do

278− Nothing -> liftIO (cleanupSVar sv) >> throwM ex

282+ liftIO (cleanupSVar (workerThreads sv))

283+ throwM ex

279

284

280 step _ (FromSVarDone sv) = do

285 step _ (FromSVarDone sv) = do

281 when (svarInspectMode sv) $ do

286 when (svarInspectMode sv) $ do

282 t <- liftIO $ getTime Monotonic

287 t <- liftIO $ getTime Monotonic

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

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

284− liftIO $ printSVar sv "SVar Done"

289+ liftIO $ printSVar (dumpSVar sv) "SVar Done"

285 return D.Stop

290 return D.Stop

src/Streamly/Internal/Data/Stream/Async/Channel/Type.hs+48−506

@@ -8,222 +8,24 @@

9module Streamly.Internal.Data.Stream.Async.Channel.Type

10 (

11− -- * Parent child communication

12− ThreadAbort (..)

13− , ChildEvent (..)

14−

15− -- * SVar

16− , Count (..)

17− , Limit (..)

18− , SVarStats (..)

19− , WorkerInfo (..)

20− , LatencyRange (..)

21− , YieldRateInfo (..)

22− , Channel(..)

11+ Channel(..)

23−

12+ , yield

24− -- * Channel configuration

13+ , stop

25− , Rate (..)

14+ , dumpSVar

26− , Config

27−

28− -- ** Default config

29− , magicMaxBuffer

30− , defaultConfig

31−

32− -- ** Type cast

33−

34− -- ** State accessors

35− , maxThreads

36− , maxBuffer

37− , maxYields

38− , inspectMode

39−

40− , rate

41− , avgRate

42− , minRate

43− , maxRate

44− , constRate

45−

46− , getMaxThreads

47− , getMaxBuffer

48− , getStreamRate

49− , getStreamLatency

50− , setStreamLatency

51− , getYieldLimit

52− , getInspectMode

53−

54− , getYieldRateInfo

55− , newSVarStats

56 )

15 )

57where

16where

59import Control.Concurrent (ThreadId)

18import Control.Concurrent (ThreadId)

60import Control.Concurrent.MVar (MVar)

19import Control.Concurrent.MVar (MVar)

61−import Control.Exception (SomeException(..), Exception)

62−import Data.Int (Int64)

63−import Data.IORef (IORef, newIORef)

20+import Data.IORef (IORef)

64import Data.Set (Set)

21import Data.Set (Set)

66import Streamly.Internal.Control.Concurrent (RunInIO)

23import Streamly.Internal.Control.Concurrent (RunInIO)

24+import Streamly.Internal.Data.Stream.Channel.Dispatcher (dumpSVarStats)

25+import Streamly.Internal.Data.Stream.Channel.Worker (sendYield, sendStop)

67import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

26import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

68−import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

69−import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))

70−

71−newtype Count = Count Int64

72− deriving ( Eq

73− , Read

74− , Show

75− , Enum

76− , Bounded

77− , Num

78− , Real

79− , Integral

80− , Ord

81− )

82−

83−-- XXX We can use maxBound for unlimited?

84−

85−-- This is essentially a 'Maybe Word' type

86−data Limit = Unlimited | Limited Word deriving Show

87−

88−instance Eq Limit where

89− Unlimited == Unlimited = True

90− Unlimited == Limited _ = False

91− Limited _ == Unlimited = False

92− Limited x == Limited y = x == y

93−

94−instance Ord Limit where

95− Unlimited <= Unlimited = True

96− Unlimited <= Limited _ = False

97− Limited _ <= Unlimited = True

98− Limited x <= Limited y = x <= y

99−

100−------------------------------------------------------------------------------

101−-- Parent child thread communication type

102−------------------------------------------------------------------------------

103−

104−data ThreadAbort = ThreadAbort deriving Show

105−

106−instance Exception ThreadAbort

107−

108−-- | Events that a child thread may send to a parent thread.

109−data ChildEvent a =

110− ChildYield a

111− | ChildStop ThreadId (Maybe SomeException)

112

113−-- | An SVar or a Stream Var is a conduit to the output from multiple streams

28+import Streamly.Internal.Data.Stream.Channel.Types

114−-- running concurrently and asynchronously. An SVar can be thought of as an

115−-- asynchronous IO handle. We can write any number of streams to an SVar in a

116−-- non-blocking manner and then read them back at any time at any pace. The

117−-- SVar would run the streams asynchronously and accumulate results. An SVar

118−-- may not really execute the stream completely and accumulate all the results.

119−-- However, it ensures that the reader can read the results at whatever paces

120−-- it wants to read. The SVar monitors and adapts to the consumer's pace.

121−--

122−-- An SVar is a mini scheduler, it has an associated workLoop that holds the

123−-- stream tasks to be picked and run by a pool of worker threads. It has an

124−-- associated output queue where the output stream elements are placed by the

125−-- worker threads. A outputDoorBell is used by the worker threads to intimate the

126−-- consumer thread about availability of new results in the output queue. More

127−-- workers are added to the SVar by 'fromStreamVar' on demand if the output

128−-- produced is not keeping pace with the consumer. On bounded SVars, workers

129−-- block on the output queue to provide throttling of the producer when the

130−-- consumer is not pulling fast enough. The number of workers may even get

131−-- reduced depending on the consuming pace.

132−--

133−-- New work is enqueued either at the time of creation of the SVar or as a

134−-- result of executing the parallel combinators i.e. '<|' and '<|>' when the

135−-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.

136−

137−-- We measure the individual worker latencies to estimate the number of workers

138−-- needed or the amount of time we have to sleep between dispatches to achieve

139−-- a particular rate when controlled pace mode it used.

140−data WorkerInfo = WorkerInfo

141− { workerYieldMax :: Count -- 0 means unlimited

142− -- total number of yields by the worker till now

143− , workerYieldCount :: IORef Count

144− -- yieldCount at start, timestamp

145− , workerLatencyStart :: IORef (Count, AbsTime)

146− }

147−

148−data LatencyRange = LatencyRange

149− { minLatency :: NanoSecond64

150− , maxLatency :: NanoSecond64

151− } deriving Show

152−

153−-- Rate control.

154−data YieldRateInfo = YieldRateInfo

155− { svarLatencyTarget :: NanoSecond64

156− , svarLatencyRange :: LatencyRange

157− , svarRateBuffer :: Int

158−

159− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

160− -- read by the worker threads

161− , svarGainedLostYields :: IORef Count

162−

163− -- Actual latency/througput as seen from the consumer side, we count the

164− -- yields and the time it took to generates those yields. This is used to

165− -- increase or decrease the number of workers needed to achieve the desired

166− -- rate. The idle time of workers is adjusted in this, so that we only

167− -- account for the rate when the consumer actually demands data.

168− -- XXX interval latency is enough, we can move this under diagnostics build

169− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

170− -- read by the worker threads

171− , svarAllTimeLatency :: IORef (Count, AbsTime)

172−

173− -- XXX Worker latency specified by the user to be used before the first

174− -- actual measurement arrives. Not yet implemented

175− , workerBootstrapLatency :: Maybe NanoSecond64

176−

177− -- After how many yields the worker should update the latency information.

178− -- If the latency is high, this count is kept lower and vice-versa. XXX If

179− -- the latency suddenly becomes too high this count may remain too high for

180− -- long time, in such cases the consumer can change it.

181− -- 0 means no latency computation

182− -- XXX this is derivable from workerMeasuredLatency, can be removed.

183− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

184− -- read by the worker threads

185− , workerPollingInterval :: IORef Count

186−

187− -- This is in progress latency stats maintained by the workers which we

188− -- empty into workerCollectedLatency stats at certain intervals - whenever

189− -- we process the stream elements yielded in this period. The first count

190− -- is all yields, the second count is only those yields for which the

191− -- latency was measured to be non-zero (note that if the timer resolution

192− -- is low the measured latency may be zero e.g. on JS platform).

193− -- [LOCKING] Locked access. Modified by the consumer thread as well as

194− -- worker threads. Workers modify it periodically based on

195− -- workerPollingInterval and not on every yield to reduce the locking

196− -- overhead.

197− -- (allYieldCount, yieldCount, timeTaken)

198− , workerPendingLatency :: IORef (Count, Count, NanoSecond64)

199−

200− -- This is the second level stat which is an accmulation from

201− -- workerPendingLatency stats. We keep accumulating latencies in this

202− -- bucket until we have stats for a sufficient period and then we reset it

203− -- to start collecting for the next period and retain the computed average

204− -- latency for the last period in workerMeasuredLatency.

205− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

206− -- read by the worker threads

207− -- (allYieldCount, yieldCount, timeTaken)

208− , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)

209−

210− -- Latency as measured by workers, aggregated for the last period.

211− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

212− -- read by the worker threads

213− , workerMeasuredLatency :: IORef NanoSecond64

214− }

215−

216−data SVarStats = SVarStats {

217− totalDispatches :: IORef Int

218− , maxWorkers :: IORef Int

219− , maxOutQSize :: IORef Int

220− , maxHeapSize :: IORef Int

221− , maxWorkQSize :: IORef Int

222− , avgWorkerLatency :: IORef (Count, NanoSecond64)

223− , minWorkerLatency :: IORef NanoSecond64

224− , maxWorkerLatency :: IORef NanoSecond64

225− , svarStopTime :: IORef (Maybe AbsTime)

226−}

227

228-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have

30-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have

229-- references to the original SVar stored in several functions which will keep

31-- references to the original SVar stored in several functions which will keep

@@ -310,303 +112,43 @@ data Channel m a = Channel

310 , svarCreator :: ThreadId

112 , svarCreator :: ThreadId

311 }

113 }

312

114

313−-------------------------------------------------------------------------------

115+{-# INLINE yield #-}

314−-- Overall state threaded around a stream

116+yield :: Channel m a -> Maybe WorkerInfo -> a -> IO Bool

315−-------------------------------------------------------------------------------

117+yield sv winfo x =

316−

118+ sendYield

317−-- | Specifies the stream yield rate in yields per second (@Hertz@).

119+ (maxBufferLimit sv)

318−-- We keep accumulating yield credits at 'rateGoal'. At any point of time we

120+ (maxWorkerLimit sv)

319−-- allow only as many yields as we have accumulated as per 'rateGoal' since the

121+ (workerCount sv)

320−-- start of time. If the consumer or the producer is slower or faster, the

122+ winfo

321−-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap

123+ (yieldRateInfo sv)

322−-- between the two by increasing or decreasing the pull rate from the producer.

124+ (outputQueue sv)

323−-- However, if the gap becomes more than 'rateBuffer' we try to recover only as

125+ (outputDoorBell sv)

324−-- much as 'rateBuffer'.

126+ (ChildYield x)

325−--

127+

326−-- 'rateLow' puts a bound on how low the instantaneous rate can go when

128+{-# INLINE stop #-}

327−-- recovering the rate gap. In other words, it determines the maximum yield

129+stop :: Channel m a -> Maybe WorkerInfo -> IO ()

328−-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous

130+stop sv winfo =

329−-- rate can go when recovering the rate gap. In other words, it determines the

131+ sendStop

330−-- minimum yield latency. We reduce the latency by increasing concurrency,

132+ (workerCount sv)

331−-- therefore we can say that it puts an upper bound on concurrency.

133+ winfo

332−--

134+ (yieldRateInfo sv)

333−-- If the 'rateGoal' is 0 or negative the stream never yields a value.

135+ (outputQueue sv)

334−-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.

136+ (outputDoorBell sv)

335−--

137+

336−-- /Since: 0.5.0 ("Streamly")/

138+{-# NOINLINE dumpSVar #-}

337−--

139+dumpSVar :: Channel m a -> IO String

338−-- @since 0.8.0

140+dumpSVar sv = do

339−data Rate = Rate

141+ xs <- sequence

340− { rateLow :: Double -- ^ The lower rate limit

142+ [ return (dumpCreator (svarCreator sv))

341− , rateGoal :: Double -- ^ The target rate we want to achieve

143+ , return "---------CURRENT STATE-----------"

342− , rateHigh :: Double -- ^ The upper rate limit

144+ , dumpOutputQ (outputQueue sv)

343− , rateBuffer :: Int -- ^ Maximum slack from the goal

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

344− }

146+ , dumpDoorBell (outputDoorBell sv)

345−

147+ , dumpNeedDoorBell (needDoorBell sv)

346−-- XXX we can put the resettable fields in a oneShotConfig field and others in

148+ , dumpRunningThreads (workerThreads sv)

347−-- a persistentConfig field. That way reset would be fast and scalable

149+ -- XXX print the status of first 5 threads

348−-- irrespective of the number of fields.

150+ , dumpWorkerCount (workerCount sv)

349−--

151+ , return "---------STATS-----------\n"

350−-- XXX make all these Limited types and use phantom types to distinguish them

152+ , dumpSVarStats (svarInspectMode sv) (yieldRateInfo sv) (svarStats sv)

351−

153+ ]

352−-- | An abstract type for specifying the configuration parameters of a

154+ return $ concat xs

353−-- 'Channel'. Use @Config -> Config@ modifier functions to modify the

354−-- default configuration. See the modifiers for default values.

355−--

356−data Config = Config

357− { -- one shot configuration, automatically reset for each API call

358− -- streamVar :: Maybe (SVar t m a)

359− _yieldLimit :: Maybe Count

360−

361− -- persistent configuration, state that remains valid until changed by

362− -- an explicit setting via a combinator.

363− , _threadsHigh :: Limit

364− , _bufferHigh :: Limit

365−

366− -- XXX these two can be collapsed into a single type

367− , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency

368− , _maxStreamRate :: Maybe Rate

369− , _inspectMode :: Bool

370− }

371−

372−-------------------------------------------------------------------------------

373−-- State defaults and reset

374−-------------------------------------------------------------------------------

375−

376−-- A magical value for the buffer size arrived at by running the smallest

377−-- possible task and measuring the optimal value of the buffer for that. This

378−-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel

379−-- core-i7 processor.

380−magicMaxBuffer :: Word

381−magicMaxBuffer = 1500

382−

383−defaultMaxThreads, defaultMaxBuffer :: Limit

384−defaultMaxThreads = Limited magicMaxBuffer

385−defaultMaxBuffer = Limited magicMaxBuffer

386−

387−-- The fields prefixed by an _ are not to be accessed or updated directly but

388−-- via smart accessor APIs.

389−defaultConfig :: Config

390−defaultConfig = Config

391− { -- streamVar = Nothing

392− _yieldLimit = Nothing

393− , _threadsHigh = defaultMaxThreads

394− , _bufferHigh = defaultMaxBuffer

395− , _maxStreamRate = Nothing

396− , _streamLatency = Nothing

397− , _inspectMode = False

398− }

399−

400−-------------------------------------------------------------------------------

401−-- Smart get/set routines for State

402−-------------------------------------------------------------------------------

403−

404−-- Use get/set routines instead of directly accessing the State fields

405−maxYields :: Maybe Int64 -> Config -> Config

406−maxYields lim st =

407− st { _yieldLimit =

408− case lim of

409− Nothing -> Nothing

410− Just n ->

411− if n <= 0

412− then Just 0

413− else Just (fromIntegral n)

414− }

415−

416−getYieldLimit :: Config -> Maybe Count

417−getYieldLimit = _yieldLimit

418−

419−-- | Specify the maximum number of threads that can be spawned by the channel.

420−-- A value of 0 resets the thread limit to default, a negative value means

421−-- there is no limit. The default value is 1500.

422−--

423−-- When the actions in a stream are IO bound, having blocking IO calls, this

424−-- option can be used to control the maximum number of in-flight IO requests.

425−-- When the actions are CPU bound this option can be used to control the amount

426−-- of CPU used by the stream.

427−--

428−maxThreads :: Int -> Config -> Config

429−maxThreads n st =

430− st { _threadsHigh =

431− if n < 0

432− then Unlimited

433− else if n == 0

434− then defaultMaxThreads

435− else Limited (fromIntegral n)

436− }

437−

438−getMaxThreads :: Config -> Limit

439−getMaxThreads = _threadsHigh

440−

441−-- | Specify the maximum size of the buffer for storing the results from

442−-- concurrent computations. If the buffer becomes full we stop spawning more

443−-- concurrent tasks until there is space in the buffer.

444−-- A value of 0 resets the buffer size to default, a negative value means

445−-- there is no limit. The default value is 1500.

446−--

447−-- CAUTION! using an unbounded 'maxBuffer' value (i.e. a negative value)

448−-- coupled with an unbounded 'maxThreads' value is a recipe for disaster in

449−-- presence of infinite streams, or very large streams. Especially, it must

450−-- not be used when 'pure' is used in 'ZipAsyncM' streams as 'pure' in

451−-- applicative zip streams generates an infinite stream causing unbounded

452−-- concurrent generation with no limit on the buffer or threads.

453−--

454−maxBuffer :: Int -> Config -> Config

455−maxBuffer n st =

456− st { _bufferHigh =

457− if n < 0

458− then Unlimited

459− else if n == 0

460− then defaultMaxBuffer

461− else Limited (fromIntegral n)

462− }

463−

464−getMaxBuffer :: Config -> Limit

465−getMaxBuffer = _bufferHigh

466−

467−-- | Specify the pull rate of a channel.

468−-- A 'Nothing' value resets the rate to default which is unlimited. When the

469−-- rate is specified, concurrent production may be ramped up or down

470−-- automatically to achieve the specified yield rate. The specific behavior for

471−-- different styles of 'Rate' specifications is documented under 'Rate'. The

472−-- effective maximum production rate achieved by a channel is governed by:

473−--

474−-- * The 'maxThreads' limit

475−-- * The 'maxBuffer' limit

476−-- * The maximum rate that the stream producer can achieve

477−-- * The maximum rate that the stream consumer can achieve

478−--

479−rate :: Maybe Rate -> Config -> Config

480−rate r st = st { _maxStreamRate = r }

481−

482−getStreamRate :: Config -> Maybe Rate

483−getStreamRate = _maxStreamRate

484−

485−setStreamLatency :: Int -> Config -> Config

486−setStreamLatency n st =

487− st { _streamLatency =

488− if n <= 0

489− then Nothing

490− else Just (fromIntegral n)

491− }

492−

493−getStreamLatency :: Config -> Maybe NanoSecond64

494−getStreamLatency = _streamLatency

495−

496−inspectMode :: Config -> Config

497−inspectMode st = st { _inspectMode = True }

498−

499−getInspectMode :: Config -> Bool

500−getInspectMode = _inspectMode

501−

502−-------------------------------------------------------------------------------

503−-- Creating an SVar

504−-------------------------------------------------------------------------------

505−

506−getYieldRateInfo :: Config -> IO (Maybe YieldRateInfo)

507−getYieldRateInfo st = do

508− -- convert rate in Hertz to latency in Nanoseconds

509− let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r

510− case getStreamRate st of

511− Just (Rate low goal high buf) ->

512− let l = rateToLatency goal

513− minl = rateToLatency high

514− maxl = rateToLatency low

515− in mkYieldRateInfo l (LatencyRange minl maxl) buf

516− Nothing -> return Nothing

517−

518− where

519−

520− mkYieldRateInfo latency latRange buf = do

521− measured <- newIORef 0

522− wcur <- newIORef (0,0,0)

523− wcol <- newIORef (0,0,0)

524− now <- getTime Monotonic

525− wlong <- newIORef (0,now)

526− period <- newIORef 1

527− gainLoss <- newIORef (Count 0)

528−

529− return $ Just YieldRateInfo

530− { svarLatencyTarget = latency

531− , svarLatencyRange = latRange

532− , svarRateBuffer = buf

533− , svarGainedLostYields = gainLoss

534− , workerBootstrapLatency = getStreamLatency st

535− , workerPollingInterval = period

536− , workerMeasuredLatency = measured

537− , workerPendingLatency = wcur

538− , workerCollectedLatency = wcol

539− , svarAllTimeLatency = wlong

540− }

541−

542−newSVarStats :: IO SVarStats

543−newSVarStats = do

544− disp <- newIORef 0

545− maxWrk <- newIORef 0

546− maxOq <- newIORef 0

547− maxHs <- newIORef 0

548− maxWq <- newIORef 0

549− avgLat <- newIORef (0, NanoSecond64 0)

550− maxLat <- newIORef (NanoSecond64 0)

551− minLat <- newIORef (NanoSecond64 0)

552− stpTime <- newIORef Nothing

553−

554− return SVarStats

555− { totalDispatches = disp

556− , maxWorkers = maxWrk

557− , maxOutQSize = maxOq

558− , maxHeapSize = maxHs

559− , maxWorkQSize = maxWq

560− , avgWorkerLatency = avgLat

561− , minWorkerLatency = minLat

562− , maxWorkerLatency = maxLat

563− , svarStopTime = stpTime

564− }

565−

566−-------------------------------------------------------------------------------

567−-- Rate

568−-------------------------------------------------------------------------------

569−

570−-- | Same as @rate (Just $ Rate (r/2) r (2*r) maxBound)@

571−--

572−-- Specifies the average production rate of a stream in number of yields

573−-- per second (i.e. @Hertz@). Concurrent production is ramped up or down

574−-- automatically to achieve the specified average yield rate. The rate can

575−-- go down to half of the specified rate on the lower side and double of

576−-- the specified rate on the higher side.

577−--

578−avgRate :: Double -> Config -> Config

579−avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)

580−

581−-- | Same as @rate (Just $ Rate r r (2*r) maxBound)@

582−--

583−-- Specifies the minimum rate at which the stream should yield values. As

584−-- far as possible the yield rate would never be allowed to go below the

585−-- specified rate, even though it may possibly go above it at times, the

586−-- upper limit is double of the specified rate.

587−--

588−minRate :: Double -> Config -> Config

589−minRate r = rate (Just $ Rate r r (2*r) maxBound)

590−

591−-- | Same as @rate (Just $ Rate (r/2) r r maxBound)@

592−--

593−-- Specifies the maximum rate at which the stream should yield values. As

594−-- far as possible the yield rate would never be allowed to go above the

595−-- specified rate, even though it may possibly go below it at times, the

596−-- lower limit is half of the specified rate. This can be useful in

597−-- applications where certain resource usage must not be allowed to go

598−-- beyond certain limits.

599−--

600−maxRate :: Double -> Config -> Config

601−maxRate r = rate (Just $ Rate (r/2) r r maxBound)

602−

603−-- | Same as @rate (Just $ Rate r r r 0)@

604−--

605−-- Specifies a constant yield rate. If for some reason the actual rate

606−-- goes above or below the specified rate we do not try to recover it by

607−-- increasing or decreasing the rate in future. This can be useful in

608−-- applications like graphics frame refresh where we need to maintain a

609−-- constant refresh rate.

610−--

611−constRate :: Double -> Config -> Config

612−constRate r = rate (Just $ Rate r r r 0)

src/Streamly/Internal/Data/Stream/Channel/Dispatcher.hsnew file+297

@@ -0,0 +1,297 @@

1+-- |

2+-- Module : Streamly.Internal.Data.Stream.Channel.Dispatcher

4+-- License : BSD-3-Clause

5+-- Maintainer : streamly@composewell.com

6+-- Stability : experimental

7+-- Portability : GHC

8+--

9+--

10+module Streamly.Internal.Data.Stream.Channel.Dispatcher

11+ (

12+ -- * Latency collection

13+ minThreadDelay

14+ , collectLatency

15+

16+ -- * Thread accounting

17+ , addThread

18+ , delThread

19+ , modifyThread

20+ , allThreadsDone

21+ , recordMaxWorkers

22+

23+ -- * Diagnostics

24+ , dumpSVarStats

25+ )

26+where

27+

28+import Data.Set (Set)

29+import Control.Concurrent (MVar, ThreadId)

30+import Control.Concurrent.MVar (tryPutMVar)

31+import Control.Exception (assert)

32+import Control.Monad (when, void)

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

34+import Data.IORef (IORef, modifyIORef, readIORef, writeIORef)

35+#if __GLASGOW_HASKELL__ < 804

36+import Data.Semigroup ((<>))

37+#endif

38+import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS, writeBarrier)

39+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

40+import Streamly.Internal.Data.Time.Units

41+ ( AbsTime, NanoSecond64(..), diffAbsTime64, showNanoSecond64

42+ , showRelTime64)

43+

44+import qualified Data.Set as S

45+

46+import Streamly.Internal.Data.Stream.Channel.Types

47+

48+-------------------------------------------------------------------------------

49+-- Worker latency data processing

50+-------------------------------------------------------------------------------

51+

52+-- | This is a magic number and it is overloaded, and used at several places to

53+-- achieve batching:

54+--

55+-- 1. If we have to sleep to slowdown this is the minimum period that we

56+-- accumulate before we sleep. Also, workers do not stop until this much

57+-- sleep time is accumulated.

58+-- 3. Collected latencies are computed and transferred to measured latency

59+-- after a minimum of this period.

60+minThreadDelay :: NanoSecond64

61+minThreadDelay = 1000000

62+

63+-- Every once in a while workers update the latencies and check the yield rate.

64+-- They return if we are above the expected yield rate. If we check too often

65+-- it may impact performance, if we check less often we may have a stale

66+-- picture. We update every minThreadDelay but we translate that into a yield

67+-- count based on latency so that the checking overhead is little.

68+--

69+-- XXX use a generation count to indicate that the value is updated. If the

70+-- value is updated an existing worker must check it again on the next yield.

71+-- Otherwise it is possible that we may keep updating it and because of the mod

72+-- worker keeps skipping it.

73+updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()

74+updateWorkerPollingInterval yinfo latency = do

75+ let periodRef = workerPollingInterval yinfo

76+ cnt = max 1 $ minThreadDelay `div` latency

77+ period = min cnt (fromIntegral magicMaxBuffer)

78+

79+ writeIORef periodRef (fromIntegral period)

80+

81+{-# INLINE recordMinMaxLatency #-}

82+recordMinMaxLatency :: SVarStats -> NanoSecond64 -> IO ()

83+recordMinMaxLatency ss new = do

84+ minLat <- readIORef (minWorkerLatency ss)

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

86+ writeIORef (minWorkerLatency ss) new

87+

88+ maxLat <- readIORef (maxWorkerLatency ss)

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

90+

91+recordAvgLatency :: SVarStats -> (Count, NanoSecond64) -> IO ()

92+recordAvgLatency ss (count, time) = do

93+ modifyIORef (avgWorkerLatency ss) $

94+ \(cnt, t) -> (cnt + count, t + time)

95+

96+-- Pour the pending latency stats into a collection bucket

97+{-# INLINE collectWorkerPendingLatency #-}

98+collectWorkerPendingLatency

99+ :: IORef (Count, Count, NanoSecond64)

100+ -> IORef (Count, Count, NanoSecond64)

101+ -> IO (Count, Maybe (Count, NanoSecond64))

102+collectWorkerPendingLatency cur col = do

103+ (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)

104+

105+ (fcnt, cnt, t) <- readIORef col

106+ let totalCount = fcnt + fcount

107+ latCount = cnt + count

108+ latTime = t + time

109+ writeIORef col (totalCount, latCount, latTime)

110+

111+ assert (latCount == 0 || latTime /= 0) (return ())

112+ let latPair =

113+ if latCount > 0 && latTime > 0

114+ then Just (latCount, latTime)

115+ else Nothing

116+ return (totalCount, latPair)

117+

118+{-# INLINE shouldUseCollectedBatch #-}

119+shouldUseCollectedBatch

120+ :: Count

121+ -> NanoSecond64

122+ -> NanoSecond64

123+ -> NanoSecond64

124+ -> Bool

125+shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =

126+ let r = fromIntegral newLat / fromIntegral prevLat :: Double

127+ in (collectedYields > fromIntegral magicMaxBuffer)

128+ || (collectedTime > minThreadDelay)

129+ || (prevLat > 0 && (r > 2 || r < 0.5))

130+ || (prevLat == 0)

131+

132+-- Returns a triple, (1) yield count since last collection, (2) the base time

133+-- when we started counting, (3) average latency in the last measurement

134+-- period. The former two are used for accurate measurement of the going rate

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

136+-- should be maintained to maintain the rate.

137+-- CAUTION! keep it in sync with getWorkerLatency

138+collectLatency ::

139+ Bool

140+ -> SVarStats

141+ -> YieldRateInfo

142+ -> Bool

143+ -> IO (Count, AbsTime, NanoSecond64)

144+collectLatency inspect ss yinfo drain = do

145+ let cur = workerPendingLatency yinfo

146+ col = workerCollectedLatency yinfo

147+ longTerm = svarAllTimeLatency yinfo

148+ measured = workerMeasuredLatency yinfo

149+

150+ (newCount, newLatPair) <- collectWorkerPendingLatency cur col

151+ (lcount, ltime) <- readIORef longTerm

152+ prevLat <- readIORef measured

153+

154+ let newLcount = lcount + newCount

155+ retWith lat = return (newLcount, ltime, lat)

156+

157+ case newLatPair of

158+ Nothing -> retWith prevLat

159+ Just (count, time) -> do

160+ let newLat = time `div` fromIntegral count

161+ when inspect $ recordMinMaxLatency ss newLat

162+ -- When we have collected a significant sized batch we compute the

163+ -- new latency using that batch and return the new latency,

164+ -- otherwise we return the previous latency derived from the

165+ -- previous batch.

166+ if shouldUseCollectedBatch newCount time newLat prevLat || drain

167+ then do

168+ -- XXX make this NOINLINE?

169+ updateWorkerPollingInterval yinfo (max newLat prevLat)

170+ when inspect $ recordAvgLatency ss (count, time)

171+ writeIORef col (0, 0, 0)

172+ writeIORef measured ((prevLat + newLat) `div` 2)

173+ modifyIORef longTerm $ \(_, t) -> (newLcount, t)

174+ retWith newLat

175+ else retWith prevLat

176+

177+-------------------------------------------------------------------------------

178+-- Dumping the SVar for debug/diag

179+-------------------------------------------------------------------------------

180+

181+dumpSVarStats :: Bool -> Maybe YieldRateInfo -> SVarStats -> IO String

182+dumpSVarStats inspect rateInfo ss = do

183+ case rateInfo of

184+ Nothing -> return ()

185+ Just yinfo -> do

186+ _ <- liftIO $ collectLatency inspect ss yinfo True

187+ return ()

188+

189+ dispatches <- readIORef $ totalDispatches ss

190+ maxWrk <- readIORef $ maxWorkers ss

191+ maxOq <- readIORef $ maxOutQSize ss

192+ -- maxHp <- readIORef $ maxHeapSize ss

193+ minLat <- readIORef $ minWorkerLatency ss

194+ maxLat <- readIORef $ maxWorkerLatency ss

195+ (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss

196+ (svarCnt, svarGainLossCnt, svarLat) <- case rateInfo of

197+ Nothing -> return (0, 0, 0)

198+ Just yinfo -> do

199+ (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo

200+ if cnt > 0

201+ then do

202+ t <- readIORef (svarStopTime ss)

203+ gl <- readIORef (svarGainedLostYields yinfo)

204+ case t of

205+ Nothing -> do

206+ now <- getTime Monotonic

207+ let interval = diffAbsTime64 now startTime

208+ return (cnt, gl, interval `div` fromIntegral cnt)

209+ Just stopTime -> do

210+ let interval = diffAbsTime64 stopTime startTime

211+ return (cnt, gl, interval `div` fromIntegral cnt)

212+ else return (0, 0, 0)

213+

214+ return $ unlines

215+ [ "total dispatches = " <> show dispatches

216+ , "max workers = " <> show maxWrk

217+ , "max outQSize = " <> show maxOq

218+ <> (if minLat > 0

219+ then "\nmin worker latency = " <> showNanoSecond64 minLat

220+ else "")

221+ <> (if maxLat > 0

222+ then "\nmax worker latency = " <> showNanoSecond64 maxLat

223+ else "")

224+ <> (if avgCnt > 0

225+ then let lat = avgTime `div` fromIntegral avgCnt

226+ in "\navg worker latency = " <> showNanoSecond64 lat

227+ else "")

228+ <> (if svarLat > 0

229+ then "\nSVar latency = " <> showRelTime64 svarLat

230+ else "")

231+ <> (if svarCnt > 0

232+ then "\nSVar yield count = " <> show svarCnt

233+ else "")

234+ <> (if svarGainLossCnt > 0

235+ then "\nSVar gain/loss yield count = " <> show svarGainLossCnt

236+ else "")

237+ ]

238+

239+-------------------------------------------------------------------------------

240+-- Thread accounting

241+-------------------------------------------------------------------------------

242+

243+-- Thread tracking is needed for two reasons:

244+--

245+-- 1) Killing threads on exceptions. Threads may not be left to go away by

246+-- themselves because they may run for significant times before going away or

247+-- worse they may be stuck in IO and never go away.

248+--

249+-- 2) To know when all threads are done and the stream has ended.

250+

251+{-# NOINLINE addThread #-}

252+addThread :: MonadIO m => IORef (Set ThreadId) -> ThreadId -> m ()

253+addThread workerSet tid =

254+ liftIO $ modifyIORef workerSet (S.insert tid)

255+

256+-- This is cheaper than modifyThread because we do not have to send a

257+-- outputDoorBell This can make a difference when more workers are being

258+-- dispatched.

259+{-# INLINE delThread #-}

260+delThread :: MonadIO m => IORef (Set ThreadId) -> ThreadId -> m ()

261+delThread workerSet tid =

262+ liftIO $ modifyIORef workerSet (S.delete tid)

263+

264+-- If present then delete else add. This takes care of out of order add and

265+-- delete i.e. a delete arriving before we even added a thread.

266+-- This occurs when the forked thRead is done even before the 'addThread' right

267+-- after the fork gets a chance to run.

268+{-# INLINE modifyThread #-}

269+modifyThread :: MonadIO m => IORef (Set ThreadId) -> MVar () -> ThreadId -> m ()

270+modifyThread workerSet bell tid = do

271+ changed <- liftIO $ atomicModifyIORefCAS workerSet $ \old ->

272+ if S.member tid old

273+ then let new = S.delete tid old in (new, new)

274+ else let new = S.insert tid old in (new, old)

275+ when (null changed) $

276+ liftIO $ do

277+ writeBarrier

278+ void $ tryPutMVar bell ()

279+

280+-- | This is safe even if we are adding more threads concurrently because if

281+-- a child thread is adding another thread then anyway 'workerThreads' will

282+-- not be empty.

283+{-# INLINE allThreadsDone #-}

284+allThreadsDone :: MonadIO m => IORef (Set ThreadId) -> m Bool

285+allThreadsDone ref = liftIO $ S.null <$> readIORef ref

286+

287+-------------------------------------------------------------------------------

288+-- Dispatching workers

289+-------------------------------------------------------------------------------

290+

291+{-# NOINLINE recordMaxWorkers #-}

292+recordMaxWorkers :: MonadIO m => IORef Int -> SVarStats -> m ()

293+recordMaxWorkers countRef ss = liftIO $ do

294+ active <- readIORef countRef

295+ maxWrk <- readIORef (maxWorkers ss)

296+ when (active > maxWrk) $ writeIORef (maxWorkers ss) active

297+ modifyIORef (totalDispatches ss) (+1)

src/Streamly/Internal/Data/Stream/Channel/Types.hsnew file+710

@@ -0,0 +1,710 @@

1+-- |

2+-- Module : Streamly.Internal.Data.Stream.Channel.Types

4+-- License : BSD-3-Clause

5+-- Maintainer : streamly@composewell.com

6+-- Stability : experimental

7+-- Portability : GHC

8+--

9+-- A Channel is a place where streams join and new streams start. This module

10+-- defines low level data structures and functions to build channels. For

11+-- concrete Channels see the Channel modules of specific stream types.

12+--

13+-- A Channel is a conduit to the output from multiple streams running

14+-- concurrently and asynchronously. A channel can be thought of as an

15+-- asynchronous IO handle. We can write any number of streams to a channel in a

16+-- non-blocking manner and then read them back at any time at any pace. The

17+-- channel would run the streams asynchronously and accumulate results. A

18+-- channel may not really execute the stream completely and accumulate all the

19+-- results. However, it ensures that the reader can read the results at

20+-- whatever pace it wants to read. The channel monitors and adapts to the

21+-- consumer's pace.

22+--

23+-- A channel is a mini scheduler, it has an associated workLoop that holds the

24+-- stream tasks to be picked and run by a pool of worker threads. It has an

25+-- associated output queue where the output stream elements are placed by the

26+-- worker threads. An outputDoorBell is used by the worker threads to intimate the

27+-- consumer thread about availability of new results in the output queue. More

28+-- workers are added to the channel by 'fromChannel' on demand if the output

29+-- produced is not keeping pace with the consumer. On bounded channels, workers

30+-- block on the output queue to provide throttling of the producer when the

31+-- consumer is not pulling fast enough. The number of workers may even get

32+-- reduced depending on the consuming pace.

33+--

34+module Streamly.Internal.Data.Stream.Channel.Types

35+ (

36+ -- * Types

37+ Count (..)

38+ , Limit (..)

39+ , ThreadAbort (..)

40+ , ChildEvent (..)

41+

42+ -- * Stats

43+ , SVarStats (..)

44+ , newSVarStats

45+

46+ -- * Rate Control

47+ , WorkerInfo (..)

48+ , LatencyRange (..)

49+ , YieldRateInfo (..)

50+ , newRateInfo

51+

52+ -- * Output queue

53+ , readOutputQRaw

54+ , readOutputQBasic

55+ , ringDoorBell

56+

57+ -- * Yield Limit

58+ , decrementYieldLimit

59+ , incrementYieldLimit

60+

61+ -- * Configuration

62+ , Rate (..)

63+ , Config

64+

65+ -- ** Default config

66+ , magicMaxBuffer

67+ , defaultConfig

68+

69+ -- ** Set config

70+ , maxThreads

71+ , maxBuffer

72+ , maxYields

73+ , inspectMode

74+

75+ , rate

76+ , avgRate

77+ , minRate

78+ , maxRate

79+ , constRate

80+

81+ -- ** Get config

82+ , getMaxThreads

83+ , getMaxBuffer

84+ , getStreamRate

85+ , getStreamLatency

86+ , setStreamLatency

87+ , getYieldLimit

88+ , getInspectMode

89+

90+ -- * Cleanup

91+ , cleanupSVar

92+

93+ -- * Diagnostics

94+ , dumpCreator

95+ , dumpOutputQ

96+ , dumpDoorBell

97+ , dumpNeedDoorBell

98+ , dumpRunningThreads

99+ , dumpWorkerCount

100+

101+ , withDiagMVar

102+ , printSVar

103+ )

104+where

105+

106+import Control.Concurrent (ThreadId, myThreadId, throwTo, MVar, tryReadMVar)

107+import Control.Concurrent.MVar (tryPutMVar)

108+import Control.Exception

109+ ( SomeException(..), Exception, catches, throwIO, Handler(..)

110+ , BlockedIndefinitelyOnMVar(..), BlockedIndefinitelyOnSTM(..))

111+import Control.Monad (void, when)

112+import Data.Int (Int64)

113+import Data.IORef (IORef, newIORef, readIORef, writeIORef)

114+import Data.Set (Set)

115+import Streamly.Internal.Data.Atomics

116+ (atomicModifyIORefCAS, atomicModifyIORefCAS_, storeLoadBarrier)

117+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

118+import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))

119+import System.IO (hPutStrLn, stderr)

120+

121+import qualified Data.Set as Set

122+

123+------------------------------------------------------------------------------

124+-- Common types

125+------------------------------------------------------------------------------

126+

127+newtype Count = Count Int64

128+ deriving ( Eq

129+ , Read

130+ , Show

131+ , Enum

132+ , Bounded

133+ , Num

134+ , Real

135+ , Integral

136+ , Ord

137+ )

138+

139+-- XXX We can use maxBound for unlimited?

140+

141+-- This is essentially a 'Maybe Word' type

142+data Limit = Unlimited | Limited Word deriving Show

143+

144+instance Eq Limit where

145+ Unlimited == Unlimited = True

146+ Unlimited == Limited _ = False

147+ Limited _ == Unlimited = False

148+ Limited x == Limited y = x == y

149+

150+instance Ord Limit where

151+ Unlimited <= Unlimited = True

152+ Unlimited <= Limited _ = False

153+ Limited _ <= Unlimited = True

154+ Limited x <= Limited y = x <= y

155+

156+------------------------------------------------------------------------------

157+-- Parent child thread communication type

158+------------------------------------------------------------------------------

159+

160+data ThreadAbort = ThreadAbort deriving Show

161+

162+instance Exception ThreadAbort

163+

164+-- | Events that a child thread may send to a parent thread.

165+data ChildEvent a =

166+ ChildYield a

167+ | ChildStop ThreadId (Maybe SomeException)

168+

169+-- | We measure the individual worker latencies to estimate the number of workers

170+-- needed or the amount of time we have to sleep between dispatches to achieve

171+-- a particular rate when controlled pace mode it used.

172+data WorkerInfo = WorkerInfo

173+ {

174+ -- | 0 means unlimited

175+ workerYieldMax :: Count

176+ -- | total number of yields by the worker till now

177+ , workerYieldCount :: IORef Count

178+ -- | yieldCount at start, timestamp

179+ , workerLatencyStart :: IORef (Count, AbsTime)

180+ }

181+

182+data LatencyRange = LatencyRange

183+ { minLatency :: NanoSecond64

184+ , maxLatency :: NanoSecond64

185+ } deriving Show

186+

187+-- | Rate control.

188+data YieldRateInfo = YieldRateInfo

189+ { svarLatencyTarget :: NanoSecond64

190+ , svarLatencyRange :: LatencyRange

191+ , svarRateBuffer :: Int

192+

193+ -- | [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

194+ -- read by the worker threads

195+ , svarGainedLostYields :: IORef Count

196+

197+ -- | Actual latency/througput as seen from the consumer side, we count the

198+ -- yields and the time it took to generates those yields. This is used to

199+ -- increase or decrease the number of workers needed to achieve the desired

200+ -- rate. The idle time of workers is adjusted in this, so that we only

201+ -- account for the rate when the consumer actually demands data.

202+ -- XXX interval latency is enough, we can move this under diagnostics build

203+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

204+ -- read by the worker threads

205+ , svarAllTimeLatency :: IORef (Count, AbsTime)

206+

207+ -- XXX Worker latency specified by the user to be used before the first

208+ -- actual measurement arrives. Not yet implemented

209+ , workerBootstrapLatency :: Maybe NanoSecond64

210+

211+ -- | After how many yields the worker should update the latency information.

212+ -- If the latency is high, this count is kept lower and vice-versa. XXX If

213+ -- the latency suddenly becomes too high this count may remain too high for

214+ -- long time, in such cases the consumer can change it.

215+ -- 0 means no latency computation

216+ -- XXX this is derivable from workerMeasuredLatency, can be removed.

217+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

218+ -- read by the worker threads

219+ , workerPollingInterval :: IORef Count

220+

221+ -- | This is in progress latency stats maintained by the workers which we

222+ -- empty into workerCollectedLatency stats at certain intervals - whenever

223+ -- we process the stream elements yielded in this period. The first count

224+ -- is all yields, the second count is only those yields for which the

225+ -- latency was measured to be non-zero (note that if the timer resolution

226+ -- is low the measured latency may be zero e.g. on JS platform).

227+ -- [LOCKING] Locked access. Modified by the consumer thread as well as

228+ -- worker threads. Workers modify it periodically based on

229+ -- workerPollingInterval and not on every yield to reduce the locking

230+ -- overhead.

231+ -- (allYieldCount, yieldCount, timeTaken)

232+ , workerPendingLatency :: IORef (Count, Count, NanoSecond64)

233+

234+ -- | This is the second level stat which is an accmulation from

235+ -- workerPendingLatency stats. We keep accumulating latencies in this

236+ -- bucket until we have stats for a sufficient period and then we reset it

237+ -- to start collecting for the next period and retain the computed average

238+ -- latency for the last period in workerMeasuredLatency.

239+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

240+ -- read by the worker threads

241+ -- (allYieldCount, yieldCount, timeTaken)

242+ , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)

243+

244+ -- | Latency as measured by workers, aggregated for the last period.

245+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

246+ -- read by the worker threads

247+ , workerMeasuredLatency :: IORef NanoSecond64

248+ }

249+

250+data SVarStats = SVarStats {

251+ totalDispatches :: IORef Int

252+ , maxWorkers :: IORef Int

253+ , maxOutQSize :: IORef Int

254+ , maxHeapSize :: IORef Int

255+ , maxWorkQSize :: IORef Int

256+ , avgWorkerLatency :: IORef (Count, NanoSecond64)

257+ , minWorkerLatency :: IORef NanoSecond64

258+ , maxWorkerLatency :: IORef NanoSecond64

259+ , svarStopTime :: IORef (Maybe AbsTime)

260+}

261+

262+-------------------------------------------------------------------------------

263+-- Config

264+-------------------------------------------------------------------------------

265+

266+-- | Specifies the stream yield rate in yields per second (@Hertz@).

267+-- We keep accumulating yield credits at 'rateGoal'. At any point of time we

268+-- allow only as many yields as we have accumulated as per 'rateGoal' since the

269+-- start of time. If the consumer or the producer is slower or faster, the

270+-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap

271+-- between the two by increasing or decreasing the pull rate from the producer.

272+-- However, if the gap becomes more than 'rateBuffer' we try to recover only as

273+-- much as 'rateBuffer'.

274+--

275+-- 'rateLow' puts a bound on how low the instantaneous rate can go when

276+-- recovering the rate gap. In other words, it determines the maximum yield

277+-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous

278+-- rate can go when recovering the rate gap. In other words, it determines the

279+-- minimum yield latency. We reduce the latency by increasing concurrency,

280+-- therefore we can say that it puts an upper bound on concurrency.

281+--

282+-- If the 'rateGoal' is 0 or negative the stream never yields a value.

283+-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.

284+--

285+-- /Since: 0.5.0 ("Streamly")/

286+--

287+-- @since 0.8.0

288+data Rate = Rate

289+ { rateLow :: Double -- ^ The lower rate limit

290+ , rateGoal :: Double -- ^ The target rate we want to achieve

291+ , rateHigh :: Double -- ^ The upper rate limit

292+ , rateBuffer :: Int -- ^ Maximum slack from the goal

293+ }

294+

295+-- XXX we can put the resettable fields in a oneShotConfig field and others in

296+-- a persistentConfig field. That way reset would be fast and scalable

297+-- irrespective of the number of fields.

298+--

299+-- XXX make all these Limited types and use phantom types to distinguish them

300+

301+-- | An abstract type for specifying the configuration parameters of a

302+-- 'Channel'. Use @Config -> Config@ modifier functions to modify the

303+-- default configuration. See the modifiers for default values.

304+--

305+data Config = Config

306+ { -- one shot configuration, automatically reset for each API call

307+ -- streamVar :: Maybe (SVar t m a)

308+ _yieldLimit :: Maybe Count

309+

310+ -- persistent configuration, state that remains valid until changed by

311+ -- an explicit setting via a combinator.

312+ , _threadsHigh :: Limit

313+ , _bufferHigh :: Limit

314+

315+ -- XXX these two can be collapsed into a single type

316+ , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency

317+ , _maxStreamRate :: Maybe Rate

318+ , _inspectMode :: Bool

319+ }

320+

321+-------------------------------------------------------------------------------

322+-- State defaults and reset

323+-------------------------------------------------------------------------------

324+

325+-- | A magical value for the buffer size arrived at by running the smallest

326+-- possible task and measuring the optimal value of the buffer for that. This

327+-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel

328+-- core-i7 processor.

329+magicMaxBuffer :: Word

330+magicMaxBuffer = 1500

331+

332+defaultMaxThreads, defaultMaxBuffer :: Limit

333+defaultMaxThreads = Limited magicMaxBuffer

334+defaultMaxBuffer = Limited magicMaxBuffer

335+

336+-- | The fields prefixed by an _ are not to be accessed or updated directly but

337+-- via smart accessor APIs. Use get/set routines instead of directly accessing

338+-- the Config fields

339+defaultConfig :: Config

340+defaultConfig = Config

341+ { -- streamVar = Nothing

342+ _yieldLimit = Nothing

343+ , _threadsHigh = defaultMaxThreads

344+ , _bufferHigh = defaultMaxBuffer

345+ , _maxStreamRate = Nothing

346+ , _streamLatency = Nothing

347+ , _inspectMode = False

348+ }

349+

350+-------------------------------------------------------------------------------

351+-- Smart get/set routines for State

352+-------------------------------------------------------------------------------

353+

354+maxYields :: Maybe Int64 -> Config -> Config

355+maxYields lim st =

356+ st { _yieldLimit =

357+ case lim of

358+ Nothing -> Nothing

359+ Just n ->

360+ if n <= 0

361+ then Just 0

362+ else Just (fromIntegral n)

363+ }

364+

365+getYieldLimit :: Config -> Maybe Count

366+getYieldLimit = _yieldLimit

367+

368+-- | Specify the maximum number of threads that can be spawned by the channel.

369+-- A value of 0 resets the thread limit to default, a negative value means

370+-- there is no limit. The default value is 1500.

371+--

372+-- When the actions in a stream are IO bound, having blocking IO calls, this

373+-- option can be used to control the maximum number of in-flight IO requests.

374+-- When the actions are CPU bound this option can be used to control the amount

375+-- of CPU used by the stream.

376+--

377+maxThreads :: Int -> Config -> Config

378+maxThreads n st =

379+ st { _threadsHigh =

380+ if n < 0

381+ then Unlimited

382+ else if n == 0

383+ then defaultMaxThreads

384+ else Limited (fromIntegral n)

385+ }

386+

387+getMaxThreads :: Config -> Limit

388+getMaxThreads = _threadsHigh

389+

390+-- | Specify the maximum size of the buffer for storing the results from

391+-- concurrent computations. If the buffer becomes full we stop spawning more

392+-- concurrent tasks until there is space in the buffer.

393+-- A value of 0 resets the buffer size to default, a negative value means

394+-- there is no limit. The default value is 1500.

395+--

396+-- CAUTION! using an unbounded 'maxBuffer' value (i.e. a negative value)

397+-- coupled with an unbounded 'maxThreads' value is a recipe for disaster in

398+-- presence of infinite streams, or very large streams. Especially, it must

399+-- not be used when 'pure' is used in 'ZipAsyncM' streams as 'pure' in

400+-- applicative zip streams generates an infinite stream causing unbounded

401+-- concurrent generation with no limit on the buffer or threads.

402+--

403+maxBuffer :: Int -> Config -> Config

404+maxBuffer n st =

405+ st { _bufferHigh =

406+ if n < 0

407+ then Unlimited

408+ else if n == 0

409+ then defaultMaxBuffer

410+ else Limited (fromIntegral n)

411+ }

412+

413+getMaxBuffer :: Config -> Limit

414+getMaxBuffer = _bufferHigh

415+

416+-- | Specify the pull rate of a channel.

417+-- A 'Nothing' value resets the rate to default which is unlimited. When the

418+-- rate is specified, concurrent production may be ramped up or down

419+-- automatically to achieve the specified yield rate. The specific behavior for

420+-- different styles of 'Rate' specifications is documented under 'Rate'. The

421+-- effective maximum production rate achieved by a channel is governed by:

422+--

423+-- * The 'maxThreads' limit

424+-- * The 'maxBuffer' limit

425+-- * The maximum rate that the stream producer can achieve

426+-- * The maximum rate that the stream consumer can achieve

427+--

428+rate :: Maybe Rate -> Config -> Config

429+rate r st = st { _maxStreamRate = r }

430+

431+getStreamRate :: Config -> Maybe Rate

432+getStreamRate = _maxStreamRate

433+

434+setStreamLatency :: Int -> Config -> Config

435+setStreamLatency n st =

436+ st { _streamLatency =

437+ if n <= 0

438+ then Nothing

439+ else Just (fromIntegral n)

440+ }

441+

442+getStreamLatency :: Config -> Maybe NanoSecond64

443+getStreamLatency = _streamLatency

444+

445+inspectMode :: Config -> Config

446+inspectMode st = st { _inspectMode = True }

447+

448+getInspectMode :: Config -> Bool

449+getInspectMode = _inspectMode

450+

451+-------------------------------------------------------------------------------

452+-- Initialization

453+-------------------------------------------------------------------------------

454+

455+newRateInfo :: Config -> IO (Maybe YieldRateInfo)

456+newRateInfo st = do

457+ -- convert rate in Hertz to latency in Nanoseconds

458+ let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r

459+ case getStreamRate st of

460+ Just (Rate low goal high buf) ->

461+ let l = rateToLatency goal

462+ minl = rateToLatency high

463+ maxl = rateToLatency low

464+ in mkYieldRateInfo l (LatencyRange minl maxl) buf

465+ Nothing -> return Nothing

466+

467+ where

468+

469+ mkYieldRateInfo latency latRange buf = do

470+ measured <- newIORef 0

471+ wcur <- newIORef (0,0,0)

472+ wcol <- newIORef (0,0,0)

473+ now <- getTime Monotonic

474+ wlong <- newIORef (0,now)

475+ period <- newIORef 1

476+ gainLoss <- newIORef (Count 0)

477+

478+ return $ Just YieldRateInfo

479+ { svarLatencyTarget = latency

480+ , svarLatencyRange = latRange

481+ , svarRateBuffer = buf

482+ , svarGainedLostYields = gainLoss

483+ , workerBootstrapLatency = getStreamLatency st

484+ , workerPollingInterval = period

485+ , workerMeasuredLatency = measured

486+ , workerPendingLatency = wcur

487+ , workerCollectedLatency = wcol

488+ , svarAllTimeLatency = wlong

489+ }

490+

491+newSVarStats :: IO SVarStats

492+newSVarStats = do

493+ disp <- newIORef 0

494+ maxWrk <- newIORef 0

495+ maxOq <- newIORef 0

496+ maxHs <- newIORef 0

497+ maxWq <- newIORef 0

498+ avgLat <- newIORef (0, NanoSecond64 0)

499+ maxLat <- newIORef (NanoSecond64 0)

500+ minLat <- newIORef (NanoSecond64 0)

501+ stpTime <- newIORef Nothing

502+

503+ return SVarStats

504+ { totalDispatches = disp

505+ , maxWorkers = maxWrk

506+ , maxOutQSize = maxOq

507+ , maxHeapSize = maxHs

508+ , maxWorkQSize = maxWq

509+ , avgWorkerLatency = avgLat

510+ , minWorkerLatency = minLat

511+ , maxWorkerLatency = maxLat

512+ , svarStopTime = stpTime

513+ }

514+

515+-------------------------------------------------------------------------------

516+-- Rate

517+-------------------------------------------------------------------------------

518+

519+-- | Same as @rate (Just $ Rate (r/2) r (2*r) maxBound)@

520+--

521+-- Specifies the average production rate of a stream in number of yields

522+-- per second (i.e. @Hertz@). Concurrent production is ramped up or down

523+-- automatically to achieve the specified average yield rate. The rate can

524+-- go down to half of the specified rate on the lower side and double of

525+-- the specified rate on the higher side.

526+--

527+avgRate :: Double -> Config -> Config

528+avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)

529+

530+-- | Same as @rate (Just $ Rate r r (2*r) maxBound)@

531+--

532+-- Specifies the minimum rate at which the stream should yield values. As

533+-- far as possible the yield rate would never be allowed to go below the

534+-- specified rate, even though it may possibly go above it at times, the

535+-- upper limit is double of the specified rate.

536+--

537+minRate :: Double -> Config -> Config

538+minRate r = rate (Just $ Rate r r (2*r) maxBound)

539+

540+-- | Same as @rate (Just $ Rate (r/2) r r maxBound)@

541+--

542+-- Specifies the maximum rate at which the stream should yield values. As

543+-- far as possible the yield rate would never be allowed to go above the

544+-- specified rate, even though it may possibly go below it at times, the

545+-- lower limit is half of the specified rate. This can be useful in

546+-- applications where certain resource usage must not be allowed to go

547+-- beyond certain limits.

548+--

549+maxRate :: Double -> Config -> Config

550+maxRate r = rate (Just $ Rate (r/2) r r maxBound)

551+

552+-- | Same as @rate (Just $ Rate r r r 0)@

553+--

554+-- Specifies a constant yield rate. If for some reason the actual rate

555+-- goes above or below the specified rate we do not try to recover it by

556+-- increasing or decreasing the rate in future. This can be useful in

557+-- applications like graphics frame refresh where we need to maintain a

558+-- constant refresh rate.

559+--

560+constRate :: Double -> Config -> Config

561+constRate r = rate (Just $ Rate r r r 0)

562+

563+-------------------------------------------------------------------------------

564+-- Channel yield count

565+-------------------------------------------------------------------------------

566+

567+-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv

568+-- faster, along with the fields in sv required by send?

569+-- XXX make it noinline

570+--

571+-- XXX we may want to employ an increment and decrement in batches when the

572+-- througput is high or when the cost of synchronization is high. For example

573+-- if the application is distributed then inc/dec of a shared variable may be

574+-- very costly.

575+

576+-- | A worker decrements the yield limit before it executes an action. However,

577+-- the action may not result in an element being yielded, in that case we have

578+-- to increment the yield limit.

579+--

580+-- Note that we need it to be an Int type so that we have the ability to undo a

581+-- decrement that takes it below zero.

582+{-# INLINE decrementYieldLimit #-}

583+decrementYieldLimit :: Maybe (IORef Count) -> IO Bool

584+decrementYieldLimit remaining =

585+ case remaining of

586+ Nothing -> return True

587+ Just ref -> do

588+ r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

589+ return $ r >= 1

590+

591+{-# INLINE incrementYieldLimit #-}

592+incrementYieldLimit :: Maybe (IORef Count) -> IO ()

593+incrementYieldLimit remaining =

594+ case remaining of

595+ Nothing -> return ()

596+ Just ref -> atomicModifyIORefCAS_ ref (+ 1)

597+

598+-------------------------------------------------------------------------------

599+-- Output queue

600+-------------------------------------------------------------------------------

601+

602+{-# INLINE readOutputQBasic #-}

603+readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)

604+readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)

605+

606+{-# INLINE readOutputQRaw #-}

607+readOutputQRaw ::

608+ IORef ([ChildEvent a], Int) -> Maybe SVarStats -> IO ([ChildEvent a], Int)

609+readOutputQRaw q stats = do

610+ (list, len) <- readOutputQBasic q

611+ case stats of

612+ Just ss -> do

613+ let ref = maxOutQSize ss

614+ oqLen <- readIORef ref

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

616+ Nothing -> return ()

617+ return (list, len)

618+

619+{-# INLINE ringDoorBell #-}

620+ringDoorBell :: IORef Bool -> MVar () -> IO ()

621+ringDoorBell needBell bell = do

622+ storeLoadBarrier

623+ w <- readIORef needBell

624+ when w $ do

625+ -- Note: the sequence of operations is important for correctness here.

626+ -- We need to set the flag to false strictly before sending the

627+ -- outputDoorBell, otherwise the outputDoorBell may get processed too

628+ -- early and then we may set the flag to False to later making the

629+ -- consumer lose the flag, even without receiving a outputDoorBell.

630+ atomicModifyIORefCAS_ needBell (const False)

631+ void $ tryPutMVar bell ()

632+

633+-------------------------------------------------------------------------------

634+-- Diagnostics

635+-------------------------------------------------------------------------------

636+

637+dumpCreator :: Show a => a -> String

638+dumpCreator tid = "Creator tid = " <> show tid

639+

640+dumpOutputQ :: (Foldable t, Show a1) => IORef (t a2, a1) -> IO String

641+dumpOutputQ q = do

642+ (oqList, oqLen) <- readIORef q

643+ return $ unlines

644+ [ "outputQueue length computed = " <> show (length oqList)

645+ , "outputQueue length maintained = " <> show oqLen

646+ ]

647+

648+dumpDoorBell :: Show a => MVar a -> IO String

649+dumpDoorBell mvar = do

650+ db <- tryReadMVar mvar

651+ return $ "outputDoorBell = " <> show db

652+

653+dumpNeedDoorBell :: Show a => IORef a -> IO String

654+dumpNeedDoorBell ref = do

655+ waiting <- readIORef ref

656+ return $ "needDoorBell = " <> show waiting

657+

658+dumpRunningThreads :: Show a => IORef a -> IO String

659+dumpRunningThreads ref = do

660+ rthread <- readIORef ref

661+ return $ "running threads = " <> show rthread

662+

663+dumpWorkerCount :: Show a => IORef a -> IO String

664+dumpWorkerCount ref = do

665+ workers <- readIORef ref

666+ return $ "running thread count = " <> show workers

667+

668+{-# NOINLINE mvarExcHandler #-}

669+mvarExcHandler :: IO String -> String -> BlockedIndefinitelyOnMVar -> IO ()

670+mvarExcHandler dump label e@BlockedIndefinitelyOnMVar = do

671+ svInfo <- dump

672+ hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo

673+ throwIO e

674+

675+{-# NOINLINE stmExcHandler #-}

676+stmExcHandler :: IO String -> String -> BlockedIndefinitelyOnSTM -> IO ()

677+stmExcHandler dump label e@BlockedIndefinitelyOnSTM = do

678+ svInfo <- dump

679+ hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo

680+ throwIO e

681+

682+-- | MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we

683+-- can keep it on in production to debug problems quickly if and when they

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

685+-- until they are GCed because the consumer went away.

686+withDiagMVar :: Bool -> IO String -> String -> IO () -> IO ()

687+withDiagMVar inspect dump label action =

688+ if inspect

689+ then

690+ action `catches` [ Handler (mvarExcHandler dump label)

691+ , Handler (stmExcHandler dump label)

692+ ]

693+ else action

694+

695+printSVar :: IO String -> String -> IO ()

696+printSVar dump how = do

697+ svInfo <- dump

698+ hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo

699+

700+-------------------------------------------------------------------------------

701+-- Cleanup

702+-------------------------------------------------------------------------------

703+

704+-- Can be called from a worker as well, avoids throwing to itself

705+cleanupSVar :: IORef (Set ThreadId) -> IO ()

706+cleanupSVar workerSet = do

707+ workers <- readIORef workerSet

708+ self <- myThreadId

709+ Prelude.mapM_ (`throwTo` ThreadAbort)

710+ (Prelude.filter (/= self) $ Set.toList workers)

src/Streamly/Internal/Data/Stream/Async/Channel/Worker.hs → src/Streamly/Internal/Data/Stream/Channel/Worker.hsrenamed+73−134

@@ -6,34 +6,18 @@

6-- Stability : experimental

7-- Portability : GHC

8--

9−--

9+-- Collecting results from child workers in a streamed fashion

10−module Streamly.Internal.Data.Stream.Async.Channel.Worker

11− (

12− -- * Adjusting Limits

13− decrementYieldLimit

14− , incrementYieldLimit

16− -- * Rate Control

11+module Streamly.Internal.Data.Stream.Channel.Worker

12+ (

17− , Work (..)

13+ Work (..)

18− , isBeyondMaxRate

19 , estimateWorkers

14 , estimateWorkers

20− , updateYieldCount

21− , minThreadDelay

22− , workerRateControl

23− , workerUpdateLatency

25 -- * Send Events

16 -- * Send Events

26− , send

17+ , sendWithDoorBell

27− , ringDoorBell

28−

29− -- ** Yield

30 , sendYield

18 , sendYield

31−

32− -- ** Stop

33 , sendStop

19 , sendStop

34−

35− -- ** Exception

36− , handleChildException

20+ , handleChildException -- XXX rename to sendException

37 )

21 )

38where

22where

@@ -43,48 +27,12 @@ import Control.Exception (SomeException(..), assert)

43import Control.Monad (when, void)

27import Control.Monad (when, void)

44import Data.IORef (IORef, readIORef, writeIORef)

28import Data.IORef (IORef, readIORef, writeIORef)

45import Streamly.Internal.Data.Atomics

29import Streamly.Internal.Data.Atomics

46− (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

30+ (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)

47− storeLoadBarrier)

48import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

31import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

49import Streamly.Internal.Data.Time.Units

32import Streamly.Internal.Data.Time.Units

50 (AbsTime, NanoSecond64(..), diffAbsTime64, fromRelTime64)

33 (AbsTime, NanoSecond64(..), diffAbsTime64, fromRelTime64)

52−import Streamly.Internal.Data.Stream.Async.Channel.Type

35+import Streamly.Internal.Data.Stream.Channel.Types

53−

54−------------------------------------------------------------------------------

55−-- Collecting results from child workers in a streamed fashion

56−------------------------------------------------------------------------------

57−

58−-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv

59−-- faster, along with the fields in sv required by send?

60−-- XXX make it noinline

61−--

62−-- XXX we may want to employ an increment and decrement in batches when the

63−-- througput is high or when the cost of synchronization is high. For example

64−-- if the application is distributed then inc/dec of a shared variable may be

65−-- very costly.

66−--

67−-- A worker decrements the yield limit before it executes an action. However,

68−-- the action may not result in an element being yielded, in that case we have

69−-- to increment the yield limit.

70−--

71−-- Note that we need it to be an Int type so that we have the ability to undo a

72−-- decrement that takes it below zero.

73−{-# INLINE decrementYieldLimit #-}

74−decrementYieldLimit :: Channel m a -> IO Bool

75−decrementYieldLimit sv =

76− case remainingWork sv of

77− Nothing -> return True

78− Just ref -> do

79− r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

80− return $ r >= 1

81−

82−{-# INLINE incrementYieldLimit #-}

83−incrementYieldLimit :: Channel m a -> IO ()

84−incrementYieldLimit sv =

85− case remainingWork sv of

86− Nothing -> return ()

87− Just ref -> atomicModifyIORefCAS_ ref (+ 1)

89-------------------------------------------------------------------------------

37-------------------------------------------------------------------------------

90-- Yield control

38-- Yield control

@@ -106,20 +54,6 @@ isBeyondMaxYield cnt winfo =

106-- Sending results from worker

54-- Sending results from worker

107-------------------------------------------------------------------------------

55-------------------------------------------------------------------------------

108

109−{-# INLINE ringDoorBell #-}

110−ringDoorBell :: Channel m a -> IO ()

111−ringDoorBell sv = do

112− storeLoadBarrier

113− w <- readIORef $ needDoorBell sv

114− when w $ do

115− -- Note: the sequence of operations is important for correctness here.

116− -- We need to set the flag to false strictly before sending the

117− -- outputDoorBell, otherwise the outputDoorBell may get processed too

118− -- early and then we may set the flag to False to later making the

119− -- consumer lose the flag, even without receiving a outputDoorBell.

120− atomicModifyIORefCAS_ (needDoorBell sv) (const False)

121− void $ tryPutMVar (outputDoorBell sv) ()

122−

123{-# INLINE sendWithDoorBell #-}

57{-# INLINE sendWithDoorBell #-}

124sendWithDoorBell ::

58sendWithDoorBell ::

125 IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int

59 IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int

@@ -140,11 +74,6 @@ sendWithDoorBell q bell msg = do

140 void $ tryPutMVar bell ()

74 void $ tryPutMVar bell ()

141 return oldlen

75 return oldlen

142

143−-- | This function is used by the producer threads to queue output for the

144−-- consumer thread to consume. Returns whether the queue has more space.

145−send :: Channel m a -> ChildEvent a -> IO Int

146−send sv = sendWithDoorBell (outputQueue sv) (outputDoorBell sv)

147−

148-------------------------------------------------------------------------------

77-------------------------------------------------------------------------------

149-- Collect and update worker latency

78-- Collect and update worker latency

150-------------------------------------------------------------------------------

79-------------------------------------------------------------------------------

@@ -206,17 +135,6 @@ data Work

206 | ManyWorkers Int Count

135 | ManyWorkers Int Count

207 deriving Show

136 deriving Show

208

137

209−-- | This is a magic number and it is overloaded, and used at several places to

210−-- achieve batching:

211−--

212−-- 1. If we have to sleep to slowdown this is the minimum period that we

213−-- accumulate before we sleep. Also, workers do not stop until this much

214−-- sleep time is accumulated.

215−-- 3. Collected latencies are computed and transferred to measured latency

216−-- after a minimum of this period.

217−minThreadDelay :: NanoSecond64

218−minThreadDelay = 1000000

219−

220-- | Another magic number! When we have to start more workers to cover up a

138-- | Another magic number! When we have to start more workers to cover up a

221-- number of yields that we are lagging by then we cannot start one worker for

139-- number of yields that we are lagging by then we cannot start one worker for

222-- each yield because that may be a very big number and if the latency of the

140-- each yield because that may be a very big number and if the latency of the

@@ -333,16 +251,16 @@ estimateWorkers workerLimit svarYields gainLossYields

333 Unlimited -> n

251 Unlimited -> n

334 Limited x -> min n (fromIntegral x)

252 Limited x -> min n (fromIntegral x)

335

253

336−isBeyondMaxRate :: Channel m a -> YieldRateInfo -> IO Bool

254+isBeyondMaxRate :: Limit -> IORef Int -> YieldRateInfo -> IO Bool

337−isBeyondMaxRate sv yinfo = do

255+isBeyondMaxRate workerLimit workerCount rateInfo = do

338− (count, tstamp, wLatency) <- getWorkerLatency yinfo

256+ (count, tstamp, wLatency) <- getWorkerLatency rateInfo

339 now <- getTime Monotonic

257 now <- getTime Monotonic

340 let duration = fromRelTime64 $ diffAbsTime64 now tstamp

258 let duration = fromRelTime64 $ diffAbsTime64 now tstamp

341− let targetLat = svarLatencyTarget yinfo

259+ let targetLat = svarLatencyTarget rateInfo

342− gainLoss <- readIORef (svarGainedLostYields yinfo)

260+ gainLoss <- readIORef (svarGainedLostYields rateInfo)

343− let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration

261+ let work = estimateWorkers workerLimit count gainLoss duration

344− wLatency targetLat (svarLatencyRange yinfo)

262+ wLatency targetLat (svarLatencyRange rateInfo)

345− cnt <- readIORef $ workerCount sv

263+ cnt <- readIORef workerCount

346 return $ case work of

264 return $ case work of

347 -- XXX set the worker's maxYields or polling interval based on yields

265 -- XXX set the worker's maxYields or polling interval based on yields

348 PartialWorker _yields -> cnt > 1

266 PartialWorker _yields -> cnt > 1

@@ -354,29 +272,31 @@ isBeyondMaxRate sv yinfo = do

354-- and we should stop based on the aggregate yields. However, latency update

272-- and we should stop based on the aggregate yields. However, latency update

355-- period can be based on individual worker yields.

273-- period can be based on individual worker yields.

356{-# NOINLINE checkRatePeriodic #-}

274{-# NOINLINE checkRatePeriodic #-}

357−checkRatePeriodic :: Channel m a

275+checkRatePeriodic ::

276+ Limit

277+ -> IORef Int

358− -> YieldRateInfo

278+ -> YieldRateInfo

359− -> WorkerInfo

279+ -> WorkerInfo

360− -> Count

280+ -> Count

361− -> IO Bool

281+ -> IO Bool

362−checkRatePeriodic sv yinfo winfo ycnt = do

282+checkRatePeriodic workerLimit workerCount rateInfo workerInfo ycnt = do

363− i <- readIORef (workerPollingInterval yinfo)

283+ i <- readIORef (workerPollingInterval rateInfo)

364 -- XXX use generation count to check if the interval has been updated

284 -- XXX use generation count to check if the interval has been updated

365 if i /= 0 && (ycnt `mod` i) == 0

285 if i /= 0 && (ycnt `mod` i) == 0

366 then do

286 then do

367− workerUpdateLatency yinfo winfo

287+ workerUpdateLatency rateInfo workerInfo

368 -- XXX not required for parallel streams

288 -- XXX not required for parallel streams

369− isBeyondMaxRate sv yinfo

289+ isBeyondMaxRate workerLimit workerCount rateInfo

370 else return False

290 else return False

371

291

372−-- CAUTION! this also updates the yield count and therefore should be called

292+-- | CAUTION! this also updates the yield count and therefore should be called

373-- only when we are actually yielding an element.

293-- only when we are actually yielding an element.

374{-# NOINLINE workerRateControl #-}

294{-# NOINLINE workerRateControl #-}

375−workerRateControl :: Channel m a -> YieldRateInfo -> WorkerInfo -> IO Bool

295+workerRateControl :: Limit -> IORef Int -> YieldRateInfo -> WorkerInfo -> IO Bool

376−workerRateControl sv yinfo winfo = do

296+workerRateControl workerLimit workerCount rateInfo workerInfo = do

377− cnt <- updateYieldCount winfo

297+ cnt <- updateYieldCount workerInfo

378− beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt

298+ beyondMaxRate <- checkRatePeriodic workerLimit workerCount rateInfo workerInfo cnt

379− return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)

299+ return $ not (isBeyondMaxYield cnt workerInfo || beyondMaxRate)

380

300

381-------------------------------------------------------------------------------

301-------------------------------------------------------------------------------

382-- Send a yield event

302-- Send a yield event

@@ -385,24 +305,35 @@ workerRateControl sv yinfo winfo = do

385-- XXX we should do rate control here but not latency update in case of ahead

305-- XXX we should do rate control here but not latency update in case of ahead

386-- streams. latency update must be done when we yield directly to outputQueue

306-- streams. latency update must be done when we yield directly to outputQueue

387-- or when we yield to heap.

307-- or when we yield to heap.

388−--

308+

389−-- returns whether the worker should continue (True) or stop (False).

309+-- | Returns whether the worker should continue (True) or stop (False).

390{-# INLINE sendYield #-}

310{-# INLINE sendYield #-}

311+sendYield ::

312+ Limit

313+ -> Limit

314+ -> IORef Int

391−sendYield :: Channel m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool

315+ -> Maybe WorkerInfo

392−sendYield sv mwinfo msg = do

316+ -> Maybe YieldRateInfo

393− oldlen <- send sv msg

317+ -> IORef ([ChildEvent a], Int)

394− let limit = maxBufferLimit sv

318+ -> MVar ()

319+ -> ChildEvent a

320+ -> IO Bool

321+sendYield bufferLimit workerLimit workerCount workerInfo rateInfo q bell msg =

322+ do

323+ oldlen <- sendWithDoorBell q bell msg

395− bufferSpaceOk <- case limit of

324+ bufferSpaceOk <-

325+ case bufferLimit of

396 Unlimited -> return True

326 Unlimited -> return True

397 Limited lim -> do

327 Limited lim -> do

398− active <- readIORef (workerCount sv)

328+ active <- readIORef workerCount

399 return $ (oldlen + 1) < (fromIntegral lim - active)

329 return $ (oldlen + 1) < (fromIntegral lim - active)

400 rateLimitOk <-

330 rateLimitOk <-

401− case mwinfo of

331+ case workerInfo of

402 Just winfo ->

332 Just winfo ->

403− case yieldRateInfo sv of

333+ case rateInfo of

404 Nothing -> return True

334 Nothing -> return True

335+ Just yinfo ->

405− Just yinfo -> workerRateControl sv yinfo winfo

336+ workerRateControl workerLimit workerCount yinfo winfo

406 Nothing -> return True

337 Nothing -> return True

407 return $ bufferSpaceOk && rateLimitOk

338 return $ bufferSpaceOk && rateLimitOk

408

339

@@ -417,18 +348,26 @@ workerStopUpdate winfo info = do

417 when (i /= 0) $ workerUpdateLatency info winfo

348 when (i /= 0) $ workerUpdateLatency info winfo

418

349

419{-# INLINABLE sendStop #-}

350{-# INLINABLE sendStop #-}

351+sendStop ::

352+ IORef Int

420−sendStop :: Channel m a -> Maybe WorkerInfo -> IO ()

353+ -> Maybe WorkerInfo

421−sendStop sv mwinfo = do

354+ -> Maybe YieldRateInfo

355+ -> IORef ([ChildEvent a], Int)

356+ -> MVar ()

357+ -> IO ()

358+sendStop workerCount workerInfo rateInfo q bell = do

422− atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1

359+ atomicModifyIORefCAS_ workerCount $ \n -> n - 1

423− case (mwinfo, yieldRateInfo sv) of

360+ case (workerInfo, rateInfo) of

424− (Just winfo, Just info) ->

361+ (Just winfo, Just rinfo) ->

425− workerStopUpdate winfo info

362+ workerStopUpdate winfo rinfo

426 _ ->

363 _ ->

427 return ()

364 return ()

365+ myThreadId >>= \tid ->

428− myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

366+ void $ sendWithDoorBell q bell (ChildStop tid Nothing)

429

367

430{-# NOINLINE handleChildException #-}

368{-# NOINLINE handleChildException #-}

431−handleChildException :: Channel m a -> SomeException -> IO ()

369+handleChildException ::

370+ IORef ([ChildEvent a], Int) -> MVar () -> SomeException -> IO ()

432−handleChildException sv e = do

371+handleChildException q bell e = do

433 tid <- myThreadId

372 tid <- myThreadId

434− void $ send sv (ChildStop tid (Just e))

373+ void $ sendWithDoorBell q bell (ChildStop tid (Just e))

streamly.cabal+4−1

@@ -447,8 +447,11 @@ library

447

448 , Streamly.Internal.Data.Stream.Serial

449 , Streamly.Internal.Data.Stream.WSerial

450+ , Streamly.Internal.Data.Stream.Channel.Types

451+ , Streamly.Internal.Data.Stream.Channel.Dispatcher

452+ , Streamly.Internal.Data.Stream.Channel.Worker

453+

450 , Streamly.Internal.Data.Stream.Async.Channel.Type

454 , Streamly.Internal.Data.Stream.Async.Channel.Type

451− , Streamly.Internal.Data.Stream.Async.Channel.Worker

452 , Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

455 , Streamly.Internal.Data.Stream.Async.Channel.Dispatcher

453 , Streamly.Internal.Data.Stream.Async.Channel.Consumer

456 , Streamly.Internal.Data.Stream.Async.Channel.Consumer

454 , Streamly.Internal.Data.Stream.Async.Channel.Append

457 , Streamly.Internal.Data.Stream.Async.Channel.Append