2-- Module : Streamly.Internal.Data.Stream.Channel.Dispatcher
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
10module Streamly.Internal.Data.Stream.Channel.Dispatcher
12 -- * Latency collection
16 -- * Thread accounting
29import Control.Concurrent (MVar, ThreadId)
30import Control.Concurrent.MVar (tryPutMVar)
31import Control.Exception (assert)
32import Control.Monad (when, void)
33import Control.Monad.IO.Class (MonadIO(liftIO))
34import Data.IORef (IORef, modifyIORef, readIORef, writeIORef)
35import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS, writeBarrier)
36import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
37import Streamly.Internal.Data.Time.Units
38 ( AbsTime, NanoSecond64(..), diffAbsTime64, showNanoSecond64
41import qualified Data.Set as S
43import Streamly.Internal.Data.Stream.Channel.Types
45-------------------------------------------------------------------------------
46-- Worker latency data processing
47-------------------------------------------------------------------------------
49-- | This is a magic number and it is overloaded, and used at several places to
52-- 1. If we have to sleep to slowdown this is the minimum period that we
53-- accumulate before we sleep. Also, workers do not stop until this much
54-- sleep time is accumulated.
55-- 3. Collected latencies are computed and transferred to measured latency
56-- after a minimum of this period.
57minThreadDelay :: NanoSecond64
58minThreadDelay = 1000000
60-- Every once in a while workers update the latencies and check the yield rate.
61-- They return if we are above the expected yield rate. If we check too often
62-- it may impact performance, if we check less often we may have a stale
63-- picture. We update every minThreadDelay but we translate that into a yield
64-- count based on latency so that the checking overhead is little.
66-- XXX use a generation count to indicate that the value is updated. If the
67-- value is updated an existing worker must check it again on the next yield.
68-- Otherwise it is possible that we may keep updating it and because of the mod
69-- worker keeps skipping it.
70updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
71updateWorkerPollingInterval yinfo latency = do
72 let periodRef = workerPollingInterval yinfo
73 cnt = max 1 $ minThreadDelay `div` latency
74 period = min cnt (fromIntegral magicMaxBuffer)
76 writeIORef periodRef (fromIntegral period)
78{-# INLINE recordMinMaxLatency #-}
79recordMinMaxLatency :: SVarStats -> NanoSecond64 -> IO ()
80recordMinMaxLatency ss new = do
81 minLat <- readIORef (minWorkerLatency ss)
82 when (new < minLat || minLat == 0) $
83 writeIORef (minWorkerLatency ss) new
85 maxLat <- readIORef (maxWorkerLatency ss)
86 when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
88recordAvgLatency :: SVarStats -> (Count, NanoSecond64) -> IO ()
89recordAvgLatency ss (count, time) = do
90 modifyIORef (avgWorkerLatency ss) $
91 \(cnt, t) -> (cnt + count, t + time)
93-- Pour the pending latency stats into a collection bucket
94{-# INLINE collectWorkerPendingLatency #-}
95collectWorkerPendingLatency
96 :: IORef (Count, Count, NanoSecond64)
97 -> IORef (Count, Count, NanoSecond64)
98 -> IO (Count, Maybe (Count, NanoSecond64))
99collectWorkerPendingLatency cur col = do
100 (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
102 (fcnt, cnt, t) <- readIORef col
103 let totalCount = fcnt + fcount
104 latCount = cnt + count
106 writeIORef col (totalCount, latCount, latTime)
108 assert (latCount == 0 || latTime /= 0) (return ())
110 if latCount > 0 && latTime > 0
111 then Just (latCount, latTime)
113 return (totalCount, latPair)
115{-# INLINE shouldUseCollectedBatch #-}
116shouldUseCollectedBatch
122shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
123 let r = fromIntegral newLat / fromIntegral prevLat :: Double
124 in (collectedYields > fromIntegral magicMaxBuffer)
125 || (collectedTime > minThreadDelay)
126 || (prevLat > 0 && (r > 2 || r < 0.5))
129-- Returns a triple, (1) yield count since last collection, (2) the base time
130-- when we started counting, (3) average latency in the last measurement
131-- period. The former two are used for accurate measurement of the going rate
132-- whereas the average is used for future estimates e.g. how many workers
133-- should be maintained to maintain the rate.
134-- CAUTION! keep it in sync with getWorkerLatency
140 -> IO (Count, AbsTime, NanoSecond64)
141collectLatency inspecting ss yinfo drain = do
142 let cur = workerPendingLatency yinfo
143 col = workerCollectedLatency yinfo
144 longTerm = svarAllTimeLatency yinfo
145 measured = workerMeasuredLatency yinfo
147 (newCount, newLatPair) <- collectWorkerPendingLatency cur col
148 (lcount, ltime) <- readIORef longTerm
149 prevLat <- readIORef measured
151 let newLcount = lcount + newCount
152 retWith lat = return (newLcount, ltime, lat)
155 Nothing -> retWith prevLat
156 Just (count, time) -> do
157 let newLat = time `div` fromIntegral count
158 when inspecting $ recordMinMaxLatency ss newLat
159 -- When we have collected a significant sized batch we compute the
160 -- new latency using that batch and return the new latency,
161 -- otherwise we return the previous latency derived from the
163 if shouldUseCollectedBatch newCount time newLat prevLat || drain
165 -- XXX make this NOINLINE?
166 updateWorkerPollingInterval yinfo (max newLat prevLat)
167 when inspecting $ recordAvgLatency ss (count, time)
168 writeIORef col (0, 0, 0)
169 writeIORef measured ((prevLat + newLat) `div` 2)
170 modifyIORef longTerm $ \(_, t) -> (newLcount, t)
174-------------------------------------------------------------------------------
175-- Dumping the SVar for debug/diag
176-------------------------------------------------------------------------------
178dumpSVarStats :: Bool -> Maybe YieldRateInfo -> SVarStats -> IO String
179dumpSVarStats inspecting rateInfo ss = do
183 _ <- liftIO $ collectLatency inspecting ss yinfo True
186 dispatches <- readIORef $ totalDispatches ss
187 maxWrk <- readIORef $ maxWorkers ss
188 maxOq <- readIORef $ maxOutQSize ss
189 -- maxHp <- readIORef $ maxHeapSize ss
190 minLat <- readIORef $ minWorkerLatency ss
191 maxLat <- readIORef $ maxWorkerLatency ss
192 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
193 (svarCnt, svarGainLossCnt, svarLat) <- case rateInfo of
194 Nothing -> return (0, 0, 0)
196 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
199 t <- readIORef (svarStopTime ss)
200 gl <- readIORef (svarGainedLostYields yinfo)
203 now <- getTime Monotonic
204 let interval = diffAbsTime64 now startTime
205 return (cnt, gl, interval `div` fromIntegral cnt)
207 let interval = diffAbsTime64 stopTime startTime
208 return (cnt, gl, interval `div` fromIntegral cnt)
209 else return (0, 0, 0)
212 [ "total dispatches = " <> show dispatches
213 , "max workers = " <> show maxWrk
214 , "max outQSize = " <> show maxOq
216 then "\nmin worker latency = " <> showNanoSecond64 minLat
219 then "\nmax worker latency = " <> showNanoSecond64 maxLat
222 then let lat = avgTime `div` fromIntegral avgCnt
223 in "\navg worker latency = " <> showNanoSecond64 lat
226 then "\nSVar latency = " <> showRelTime64 svarLat
229 then "\nSVar yield count = " <> show svarCnt
231 <> (if svarGainLossCnt > 0
232 then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
236-------------------------------------------------------------------------------
238-------------------------------------------------------------------------------
240-- Thread tracking is needed for two reasons:
242-- 1) Killing threads on exceptions. Threads may not be left to go away by
243-- themselves because they may run for significant times before going away or
244-- worse they may be stuck in IO and never go away.
246-- 2) To know when all threads are done and the stream has ended.
248{-# NOINLINE addThread #-}
249addThread :: MonadIO m => IORef (Set ThreadId) -> ThreadId -> m ()
250addThread workerSet tid =
251 liftIO $ modifyIORef workerSet (S.insert tid)
253-- This is cheaper than modifyThread because we do not have to send a
254-- outputDoorBell This can make a difference when more workers are being
256{-# INLINE delThread #-}
257delThread :: MonadIO m => IORef (Set ThreadId) -> ThreadId -> m ()
258delThread workerSet tid =
259 liftIO $ modifyIORef workerSet (S.delete tid)
261-- If present then delete else add. This takes care of out of order add and
262-- delete i.e. a delete arriving before we even added a thread.
263-- This occurs when the forked thread is done even before the 'addThread' right
264-- after the fork gets a chance to run.
265{-# INLINE modifyThread #-}
266modifyThread :: MonadIO m => IORef (Set ThreadId) -> MVar () -> ThreadId -> m ()
267modifyThread workerSet bell tid = do
268 changed <- liftIO $ atomicModifyIORefCAS workerSet $ \old ->
270 then let new = S.delete tid old in (new, new)
271 else let new = S.insert tid old in (new, old)
272 when (null changed) $
275 void $ tryPutMVar bell ()
277-- | This is safe even if we are adding more threads concurrently because if
278-- a child thread is adding another thread then anyway 'workerThreads' will
280{-# INLINE allThreadsDone #-}
281allThreadsDone :: MonadIO m => IORef (Set ThreadId) -> m Bool
282allThreadsDone ref = liftIO $ S.null <$> readIORef ref
284-------------------------------------------------------------------------------
285-- Dispatching workers
286-------------------------------------------------------------------------------
288{-# NOINLINE recordMaxWorkers #-}
289recordMaxWorkers :: MonadIO m => IORef Int -> SVarStats -> m ()
290recordMaxWorkers countRef ss = liftIO $ do
291 active <- readIORef countRef
292 maxWrk <- readIORef (maxWorkers ss)
293 when (active > maxWrk) $ writeIORef (maxWorkers ss) active
294 modifyIORef (totalDispatches ss) (+1)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list