2-- Module : Streamly.Internal.Data.Stream.Concurrent.Channel.Worker
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Collecting results from child workers in a streamed fashion
11module Streamly.Internal.Data.Stream.Channel.Worker
22 , handleChildException -- XXX rename to sendException
26import Control.Concurrent (myThreadId)
27import Control.Concurrent.MVar (MVar, tryPutMVar)
28import Control.Exception (SomeException(..), assert)
29import Control.Monad (when, void)
30import Data.IORef (IORef, readIORef, writeIORef)
31import Streamly.Internal.Data.Atomics
32 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)
33import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
34import Streamly.Internal.Data.Time.Units
35 (AbsTime, NanoSecond64(..), diffAbsTime64, fromRelTime64)
37import Streamly.Internal.Data.Stream.Channel.Types
39-------------------------------------------------------------------------------
41-------------------------------------------------------------------------------
43updateYieldCount :: WorkerInfo -> IO Count
44updateYieldCount winfo = do
45 cnt <- readIORef (workerYieldCount winfo)
47 writeIORef (workerYieldCount winfo) cnt1
50isBeyondMaxYield :: Count -> WorkerInfo -> Bool
51isBeyondMaxYield cnt winfo =
52 let ymax = workerYieldMax winfo
53 in ymax /= 0 && cnt >= ymax
55-------------------------------------------------------------------------------
56-- Sending results from worker
57-------------------------------------------------------------------------------
59{-# INLINE sendWithDoorBell #-}
61 IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int
62sendWithDoorBell q bell msg = do
63 -- XXX can the access to outputQueue be made faster somehow?
64 oldlen <- atomicModifyIORefCAS q $ \(es, n) ->
65 ((msg : es, n + 1), n)
66 when (oldlen <= 0) $ do
67 -- The wake up must happen only after the store has finished otherwise
68 -- we can have lost wakeup problems.
70 -- Since multiple workers can try this at the same time, it is possible
71 -- that we may put a spurious MVar after the consumer has already seen
72 -- the output. But that's harmless, at worst it may cause the consumer
73 -- to read the queue again and find it empty.
74 -- The important point is that the consumer is guaranteed to receive a
75 -- doorbell if something was added to the queue after it empties it.
76 void $ tryPutMVar bell ()
79-------------------------------------------------------------------------------
80-- Collect and update worker latency
81-------------------------------------------------------------------------------
83workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
84workerCollectLatency winfo = do
85 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
86 cnt1 <- readIORef (workerYieldCount winfo)
91 t1 <- getTime Monotonic
92 let period = fromRelTime64 $ diffAbsTime64 t1 t0
93 writeIORef (workerLatencyStart winfo) (cnt1, t1)
94 return $ Just (cnt, period)
97-- XXX There are a number of gotchas in measuring latencies.
98-- 1) We measure latencies only when a worker yields a value
99-- 2) It is possible that a stream calls the stop continuation, in which case
100-- the worker would not yield a value and we would not account that worker in
101-- latencies. Even though this case should ideally be accounted we do not
102-- account it because we cannot or do not distinguish it from the case
104-- 3) It is possible that a worker returns without yielding anything because it
105-- never got a chance to pick up work.
106-- 4) If the system timer resolution is lower than the latency, the latency
107-- computation turns out to be zero.
109-- We can fix this if we measure the latencies by counting the work items
110-- picked rather than based on the outputs yielded.
111workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
112workerUpdateLatency yinfo winfo = do
113 r <- workerCollectLatency winfo
115 Just (cnt, period) -> do
116 -- NOTE: On JS platform the timer resolution could be pretty low. When
117 -- the timer resolution is low, measurement of latencies could be
118 -- tricky. All the worker latencies will turn out to be zero if they
119 -- are lower than the resolution. We only take into account those
120 -- measurements which are more than the timer resolution.
122 let ref = workerPendingLatency yinfo
123 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
124 atomicModifyIORefCAS_ ref $
125 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
128-------------------------------------------------------------------------------
129-- Worker rate control
130-------------------------------------------------------------------------------
132-- We either block, or send one worker with limited yield count or one or more
133-- workers with unlimited yield count.
135 = BlockWait NanoSecond64
136 | PartialWorker Count
137 | ManyWorkers Int Count
140-- | Another magic number! When we have to start more workers to cover up a
141-- number of yields that we are lagging by then we cannot start one worker for
142-- each yield because that may be a very big number and if the latency of the
143-- workers is low these number of yields could be very high. We assume that we
144-- run each extra worker for at least this much time.
145rateRecoveryTime :: NanoSecond64
146rateRecoveryTime = 1000000
148-- | Get the worker latency without resetting workerPendingLatency
149-- Returns (total yield count, base time, measured latency)
150-- CAUTION! keep it in sync with collectLatency
151getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
152getWorkerLatency yinfo = do
153 let cur = workerPendingLatency yinfo
154 col = workerCollectedLatency yinfo
155 longTerm = svarAllTimeLatency yinfo
156 measured = workerMeasuredLatency yinfo
158 (curTotalCount, curCount, curTime) <- readIORef cur
159 (colTotalCount, colCount, colTime) <- readIORef col
160 (lcount, ltime) <- readIORef longTerm
161 prevLat <- readIORef measured
163 let latCount = colCount + curCount
164 latTime = colTime + curTime
165 totalCount = colTotalCount + curTotalCount
167 if latCount > 0 && latTime > 0
168 then let lat = latTime `div` fromIntegral latCount
169 -- XXX Give more weight to new?
170 in (lat + prevLat) `div` 2
172 return (lcount + totalCount, ltime, newLat)
174-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
184estimateWorkers workerLimit svarYields gainLossYields
185 svarElapsed wLatency targetLat range =
186 -- XXX we can have a maxEfficiency combinator as well which runs the
187 -- producer at the maximal efficiency i.e. the number of workers are chosen
188 -- such that the latency is minimum or within a range. Or we can call it
192 -- How many workers do we need to achieve the required rate?
194 -- When the workers are IO bound we can increase the throughput by
195 -- increasing the number of workers as long as the IO device has enough
196 -- capacity to process all the requests concurrently. If the IO
197 -- bandwidth is saturated increasing the workers won't help. Also, if
198 -- the CPU utilization in processing all these requests exceeds the CPU
199 -- bandwidth, then increasing the number of workers won't help.
201 -- When the workers are purely CPU bound, increasing the workers beyond
202 -- the number of CPUs won't help.
204 -- TODO - measure the CPU and IO requirements of the workers. Have a
205 -- way to specify the max bandwidth of the underlying IO mechanism and
206 -- use that to determine the max rate of workers, and also take the CPU
207 -- bandwidth into account. We can also discover the IO bandwidth if we
208 -- know that we are not CPU bound, then how much steady state rate are
209 -- we able to achieve. Design tests for CPU bound and IO bound cases.
211 -- Calculate how many yields are we ahead or behind to match the exact
212 -- required rate. Based on that we increase or decrease the effective
215 -- When the worker latency is lower than required latency we begin with
216 -- a yield and then wait rather than first waiting and then yielding.
217 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
218 effectiveYields = svarYields + gainLossYields
219 deltaYields = fromIntegral targetYields - effectiveYields
221 -- We recover the deficit by running at a higher/lower rate for a
222 -- certain amount of time. To keep the effective rate in reasonable
223 -- limits we use rateRecoveryTime, minLatency and maxLatency.
224 in if deltaYields > 0
226 let deltaYieldsFreq :: Double
228 fromIntegral deltaYields /
229 fromIntegral rateRecoveryTime
230 yieldsFreq = 1.0 / fromIntegral targetLat
231 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
232 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
233 adjustedLat = min (max requiredLat (minLatency range))
235 in assert (adjustedLat > 0) $
236 if wLatency <= adjustedLat
237 then PartialWorker deltaYields
238 else let workers = withLimit $ wLatency `div` adjustedLat
239 limited = min workers (fromIntegral deltaYields)
240 in ManyWorkers (fromIntegral limited) deltaYields
242 let expectedDuration = fromIntegral effectiveYields * targetLat
243 sleepTime = expectedDuration - svarElapsed
244 maxSleepTime = maxLatency range - wLatency
245 s = min sleepTime maxSleepTime
246 in assert (sleepTime >= 0) $
247 -- if s is less than 0 it means our maxSleepTime is less
248 -- than the worker latency.
249 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
254 Limited x -> min n (fromIntegral x)
256isBeyondMaxRate :: Limit -> IORef Int -> YieldRateInfo -> IO Bool
257isBeyondMaxRate workerLimit workerCount rateInfo = do
258 (count, tstamp, wLatency) <- getWorkerLatency rateInfo
259 now <- getTime Monotonic
260 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
261 let targetLat = svarLatencyTarget rateInfo
262 gainLoss <- readIORef (svarGainedLostYields rateInfo)
263 let work = estimateWorkers workerLimit count gainLoss duration
264 wLatency targetLat (svarLatencyRange rateInfo)
265 cnt <- readIORef workerCount
266 return $ case work of
267 -- XXX set the worker's maxYields or polling interval based on yields
268 PartialWorker _yields -> cnt > 1
269 ManyWorkers n _ -> cnt > n
272-- XXX we should do rate control periodically based on the total yields rather
273-- than based on the worker local yields as other workers may have yielded more
274-- and we should stop based on the aggregate yields. However, latency update
275-- period can be based on individual worker yields.
276{-# NOINLINE checkRatePeriodic #-}
284checkRatePeriodic workerLimit workerCount rateInfo workerInfo ycnt = do
285 i <- readIORef (workerPollingInterval rateInfo)
286 -- XXX use generation count to check if the interval has been updated
287 if i /= 0 && (ycnt `mod` i) == 0
289 workerUpdateLatency rateInfo workerInfo
290 -- XXX not required for parallel streams
291 isBeyondMaxRate workerLimit workerCount rateInfo
294-- | CAUTION! this also updates the yield count and therefore should be called
295-- only when we are actually yielding an element.
296{-# NOINLINE workerRateControl #-}
297workerRateControl :: Limit -> IORef Int -> YieldRateInfo -> WorkerInfo -> IO Bool
298workerRateControl workerLimit workerCount rateInfo workerInfo = do
299 cnt <- updateYieldCount workerInfo
300 beyondMaxRate <- checkRatePeriodic workerLimit workerCount rateInfo workerInfo cnt
301 return $ not (isBeyondMaxYield cnt workerInfo || beyondMaxRate)
303-------------------------------------------------------------------------------
305-------------------------------------------------------------------------------
307-- XXX we should do rate control here but not latency update in case of ahead
308-- streams. latency update must be done when we yield directly to outputQueue
309-- or when we yield to heap.
311-- | Returns whether the worker should continue (True) or stop (False).
312{-# INLINE sendYield #-}
318 -> Maybe YieldRateInfo
319 -> IORef ([ChildEvent a], Int)
323sendYield bufferLimit workerLimit workerCount workerInfo rateInfo q bell msg =
325 oldlen <- sendWithDoorBell q bell msg
328 Unlimited -> return True
330 active <- readIORef workerCount
331 return $ (oldlen + 1) < (fromIntegral lim - active)
336 Nothing -> return True
338 workerRateControl workerLimit workerCount yinfo winfo
339 Nothing -> return True
340 return $ bufferSpaceOk && rateLimitOk
342-------------------------------------------------------------------------------
344-------------------------------------------------------------------------------
346{-# INLINE workerStopUpdate #-}
347workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
348workerStopUpdate winfo info = do
349 i <- readIORef (workerPollingInterval info)
350 when (i /= 0) $ workerUpdateLatency info winfo
352{-# INLINABLE sendStop #-}
356 -> Maybe YieldRateInfo
357 -> IORef ([ChildEvent a], Int)
360sendStop workerCount workerInfo rateInfo q bell = do
361 atomicModifyIORefCAS_ workerCount $ \n -> n - 1
362 case (workerInfo, rateInfo) of
363 (Just winfo, Just rinfo) ->
364 workerStopUpdate winfo rinfo
367 myThreadId >>= \tid ->
368 void $ sendWithDoorBell q bell (ChildStop tid Nothing)
370{-# NOINLINE handleChildException #-}
371handleChildException ::
372 IORef ([ChildEvent a], Int) -> MVar () -> SomeException -> IO ()
373handleChildException q bell e = do
375 void $ sendWithDoorBell q bell (ChildStop tid (Just e))
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