2-- Module : Streamly.Internal.Data.SVar.Worker
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
10module Streamly.Internal.Data.SVar.Worker
15 , decrementBufferLimit
16 , incrementBufferLimit
41 , handleChildException
48import Control.Concurrent (myThreadId, takeMVar)
49import Control.Concurrent.MVar (MVar, tryPutMVar)
50import Control.Exception (SomeException(..), assert)
51import Control.Monad (when, void)
52import Control.Monad.IO.Class (MonadIO(liftIO))
53import Data.IORef (IORef, readIORef, writeIORef)
54import Streamly.Internal.Data.Atomics
55 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
57import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
58import Streamly.Internal.Data.Time.Units
59 (AbsTime, NanoSecond64(..), diffAbsTime64, fromRelTime64)
61import Streamly.Internal.Data.SVar.Type
63------------------------------------------------------------------------------
64-- Collecting results from child workers in a streamed fashion
65------------------------------------------------------------------------------
67-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
68-- faster, along with the fields in sv required by send?
69-- XXX make it noinline
71-- XXX we may want to employ an increment and decrement in batches when the
72-- througput is high or when the cost of synchronization is high. For example
73-- if the application is distributed then inc/dec of a shared variable may be
76-- A worker decrements the yield limit before it executes an action. However,
77-- the action may not result in an element being yielded, in that case we have
78-- to increment the yield limit.
80-- Note that we need it to be an Int type so that we have the ability to undo a
81-- decrement that takes it below zero.
82{-# INLINE decrementYieldLimit #-}
83decrementYieldLimit :: SVar t m a -> IO Bool
84decrementYieldLimit sv =
85 case remainingWork sv of
86 Nothing -> return True
88 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
91{-# INLINE incrementYieldLimit #-}
92incrementYieldLimit :: SVar t m a -> IO ()
93incrementYieldLimit sv =
94 case remainingWork sv of
96 Just ref -> atomicModifyIORefCAS_ ref (+ 1)
98-- XXX exception safety of all atomic/MVar operations
100-- TBD Each worker can have their own queue and the consumer can empty one
101-- queue at a time, that way contention can be reduced.
103-- XXX Only yields should be counted in the buffer limit and not the Stop
106{-# INLINE decrementBufferLimit #-}
107decrementBufferLimit :: SVar t m a -> IO ()
108decrementBufferLimit sv =
109 case maxBufferLimit sv of
110 Unlimited -> return ()
112 let ref = pushBufferSpace sv
113 old <- atomicModifyIORefCAS ref $ \x ->
114 (if x >= 1 then x - 1 else x, x)
116 case pushBufferPolicy sv of
117 PushBufferBlock -> blockAndRetry
118 PushBufferDropNew -> do
119 -- We just drop one item and proceed. It is possible
120 -- that by the time we drop the item the consumer
121 -- thread might have run and created space in the
122 -- buffer, but we do not care about that condition.
123 -- This is not pedantically correct but it should be
125 -- XXX we may want to drop only if n == maxBuf
126 -- otherwise we must have space in the buffer and a
127 -- decrement should be possible.
128 block <- atomicModifyIORefCAS (outputQueue sv) $
132 _ : xs -> ((xs, n - 1), False)
133 when block blockAndRetry
134 -- XXX need a dequeue or ring buffer for this
135 PushBufferDropOld -> undefined
140 let ref = pushBufferSpace sv
141 liftIO $ takeMVar (pushBufferMVar sv)
142 old <- atomicModifyIORefCAS ref $ \x ->
143 (if x >= 1 then x - 1 else x, x)
144 -- When multiple threads sleep on takeMVar, the first thread would
145 -- wakeup due to a putMVar by the consumer, but the rest of the threads
146 -- would have to put back the MVar after taking it and decrementing the
147 -- buffer count, otherwise all other threads will remain asleep.
149 then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
150 -- We do not put the MVar back in this case, instead we
151 -- wait for the consumer to put it.
154{-# INLINE incrementBufferLimit #-}
155incrementBufferLimit :: SVar t m a -> IO ()
156incrementBufferLimit sv =
157 case maxBufferLimit sv of
158 Unlimited -> return ()
160 atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
162 void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
164{-# INLINE resetBufferLimit #-}
165resetBufferLimit :: SVar t m a -> IO ()
167 case maxBufferLimit sv of
168 Unlimited -> return ()
169 Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
170 (const (fromIntegral n))
172-------------------------------------------------------------------------------
174-------------------------------------------------------------------------------
176updateYieldCount :: WorkerInfo -> IO Count
177updateYieldCount winfo = do
178 cnt <- readIORef (workerYieldCount winfo)
180 writeIORef (workerYieldCount winfo) cnt1
183isBeyondMaxYield :: Count -> WorkerInfo -> Bool
184isBeyondMaxYield cnt winfo =
185 let ymax = workerYieldMax winfo
186 in ymax /= 0 && cnt >= ymax
188-------------------------------------------------------------------------------
189-- Sending results from worker
190-------------------------------------------------------------------------------
192{-# INLINE ringDoorBell #-}
193ringDoorBell :: SVar t m a -> IO ()
196 w <- readIORef $ needDoorBell sv
198 -- Note: the sequence of operations is important for correctness here.
199 -- We need to set the flag to false strictly before sending the
200 -- outputDoorBell, otherwise the outputDoorBell may get processed too
201 -- early and then we may set the flag to False to later making the
202 -- consumer lose the flag, even without receiving a outputDoorBell.
203 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
204 void $ tryPutMVar (outputDoorBell sv) ()
206{-# INLINE sendWithDoorBell #-}
208 IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int
209sendWithDoorBell q bell msg = do
210 -- XXX can the access to outputQueue be made faster somehow?
211 oldlen <- atomicModifyIORefCAS q $ \(es, n) ->
212 ((msg : es, n + 1), n)
213 when (oldlen <= 0) $ do
214 -- The wake up must happen only after the store has finished otherwise
215 -- we can have lost wakeup problems.
217 -- Since multiple workers can try this at the same time, it is possible
218 -- that we may put a spurious MVar after the consumer has already seen
219 -- the output. But that's harmless, at worst it may cause the consumer
220 -- to read the queue again and find it empty.
221 -- The important point is that the consumer is guaranteed to receive a
222 -- doorbell if something was added to the queue after it empties it.
223 void $ tryPutMVar bell ()
226-- | This function is used by the producer threads to queue output for the
227-- consumer thread to consume. Returns whether the queue has more space.
228send :: SVar t m a -> ChildEvent a -> IO Int
229send sv = sendWithDoorBell (outputQueue sv) (outputDoorBell sv)
231-- There is no bound implemented on the buffer, this is assumed to be low
233sendToProducer :: SVar t m a -> ChildEvent a -> IO Int
234sendToProducer sv msg = do
235 -- In case the producer stream is blocked on pushing to the fold buffer
236 -- then wake it up so that it can check for the stop event or exception
237 -- being sent to it otherwise we will be deadlocked.
238 void $ tryPutMVar (pushBufferMVar sv) ()
239 sendWithDoorBell (outputQueueFromConsumer sv)
240 (outputDoorBellFromConsumer sv) msg
242-------------------------------------------------------------------------------
243-- Collect and update worker latency
244-------------------------------------------------------------------------------
246workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
247workerCollectLatency winfo = do
248 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
249 cnt1 <- readIORef (workerYieldCount winfo)
250 let cnt = cnt1 - cnt0
254 t1 <- getTime Monotonic
255 let period = fromRelTime64 $ diffAbsTime64 t1 t0
256 writeIORef (workerLatencyStart winfo) (cnt1, t1)
257 return $ Just (cnt, period)
260-- XXX There are a number of gotchas in measuring latencies.
261-- 1) We measure latencies only when a worker yields a value
262-- 2) It is possible that a stream calls the stop continuation, in which case
263-- the worker would not yield a value and we would not account that worker in
264-- latencies. Even though this case should ideally be accounted we do not
265-- account it because we cannot or do not distinguish it from the case
267-- 3) It is possible that a worker returns without yielding anything because it
268-- never got a chance to pick up work.
269-- 4) If the system timer resolution is lower than the latency, the latency
270-- computation turns out to be zero.
272-- We can fix this if we measure the latencies by counting the work items
273-- picked rather than based on the outputs yielded.
274workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
275workerUpdateLatency yinfo winfo = do
276 r <- workerCollectLatency winfo
278 Just (cnt, period) -> do
279 -- NOTE: On JS platform the timer resolution could be pretty low. When
280 -- the timer resolution is low, measurement of latencies could be
281 -- tricky. All the worker latencies will turn out to be zero if they
282 -- are lower than the resolution. We only take into account those
283 -- measurements which are more than the timer resolution.
285 let ref = workerPendingLatency yinfo
286 (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
287 atomicModifyIORefCAS_ ref $
288 \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
291-------------------------------------------------------------------------------
292-- Worker rate control
293-------------------------------------------------------------------------------
295-- We either block, or send one worker with limited yield count or one or more
296-- workers with unlimited yield count.
298 = BlockWait NanoSecond64
299 | PartialWorker Count
300 | ManyWorkers Int Count
303-- | This is a magic number and it is overloaded, and used at several places to
306-- 1. If we have to sleep to slowdown this is the minimum period that we
307-- accumulate before we sleep. Also, workers do not stop until this much
308-- sleep time is accumulated.
309-- 3. Collected latencies are computed and transferred to measured latency
310-- after a minimum of this period.
311minThreadDelay :: NanoSecond64
312minThreadDelay = 1000000
314-- | Another magic number! When we have to start more workers to cover up a
315-- number of yields that we are lagging by then we cannot start one worker for
316-- each yield because that may be a very big number and if the latency of the
317-- workers is low these number of yields could be very high. We assume that we
318-- run each extra worker for at least this much time.
319rateRecoveryTime :: NanoSecond64
320rateRecoveryTime = 1000000
322-- | Get the worker latency without resetting workerPendingLatency
323-- Returns (total yield count, base time, measured latency)
324-- CAUTION! keep it in sync with collectLatency
325getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
326getWorkerLatency yinfo = do
327 let cur = workerPendingLatency yinfo
328 col = workerCollectedLatency yinfo
329 longTerm = svarAllTimeLatency yinfo
330 measured = workerMeasuredLatency yinfo
332 (curTotalCount, curCount, curTime) <- readIORef cur
333 (colTotalCount, colCount, colTime) <- readIORef col
334 (lcount, ltime) <- readIORef longTerm
335 prevLat <- readIORef measured
337 let latCount = colCount + curCount
338 latTime = colTime + curTime
339 totalCount = colTotalCount + curTotalCount
341 if latCount > 0 && latTime > 0
342 then let lat = latTime `div` fromIntegral latCount
343 -- XXX Give more weight to new?
344 in (lat + prevLat) `div` 2
346 return (lcount + totalCount, ltime, newLat)
348-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
358estimateWorkers workerLimit svarYields gainLossYields
359 svarElapsed wLatency targetLat range =
360 -- XXX we can have a maxEfficiency combinator as well which runs the
361 -- producer at the maximal efficiency i.e. the number of workers are chosen
362 -- such that the latency is minimum or within a range. Or we can call it
366 -- How many workers do we need to achieve the required rate?
368 -- When the workers are IO bound we can increase the throughput by
369 -- increasing the number of workers as long as the IO device has enough
370 -- capacity to process all the requests concurrently. If the IO
371 -- bandwidth is saturated increasing the workers won't help. Also, if
372 -- the CPU utilization in processing all these requests exceeds the CPU
373 -- bandwidth, then increasing the number of workers won't help.
375 -- When the workers are purely CPU bound, increasing the workers beyond
376 -- the number of CPUs won't help.
378 -- TODO - measure the CPU and IO requirements of the workers. Have a
379 -- way to specify the max bandwidth of the underlying IO mechanism and
380 -- use that to determine the max rate of workers, and also take the CPU
381 -- bandwidth into account. We can also discover the IO bandwidth if we
382 -- know that we are not CPU bound, then how much steady state rate are
383 -- we able to achieve. Design tests for CPU bound and IO bound cases.
385 -- Calculate how many yields are we ahead or behind to match the exact
386 -- required rate. Based on that we increase or decrease the effective
389 -- When the worker latency is lower than required latency we begin with
390 -- a yield and then wait rather than first waiting and then yielding.
391 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
392 effectiveYields = svarYields + gainLossYields
393 deltaYields = fromIntegral targetYields - effectiveYields
395 -- We recover the deficit by running at a higher/lower rate for a
396 -- certain amount of time. To keep the effective rate in reasonable
397 -- limits we use rateRecoveryTime, minLatency and maxLatency.
398 in if deltaYields > 0
400 let deltaYieldsFreq :: Double
402 fromIntegral deltaYields /
403 fromIntegral rateRecoveryTime
404 yieldsFreq = 1.0 / fromIntegral targetLat
405 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
406 requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
407 adjustedLat = min (max requiredLat (minLatency range))
409 in assert (adjustedLat > 0) $
410 if wLatency <= adjustedLat
411 then PartialWorker deltaYields
412 else let workers = withLimit $ wLatency `div` adjustedLat
413 limited = min workers (fromIntegral deltaYields)
414 in ManyWorkers (fromIntegral limited) deltaYields
416 let expectedDuration = fromIntegral effectiveYields * targetLat
417 sleepTime = expectedDuration - svarElapsed
418 maxSleepTime = maxLatency range - wLatency
419 s = min sleepTime maxSleepTime
420 in assert (sleepTime >= 0) $
421 -- if s is less than 0 it means our maxSleepTime is less
422 -- than the worker latency.
423 if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
428 Limited x -> min n (fromIntegral x)
430isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
431isBeyondMaxRate sv yinfo = do
432 (count, tstamp, wLatency) <- getWorkerLatency yinfo
433 now <- getTime Monotonic
434 let duration = fromRelTime64 $ diffAbsTime64 now tstamp
435 let targetLat = svarLatencyTarget yinfo
436 gainLoss <- readIORef (svarGainedLostYields yinfo)
437 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
438 wLatency targetLat (svarLatencyRange yinfo)
439 cnt <- readIORef $ workerCount sv
440 return $ case work of
441 -- XXX set the worker's maxYields or polling interval based on yields
442 PartialWorker _yields -> cnt > 1
443 ManyWorkers n _ -> cnt > n
446-- XXX we should do rate control periodically based on the total yields rather
447-- than based on the worker local yields as other workers may have yielded more
448-- and we should stop based on the aggregate yields. However, latency update
449-- period can be based on individual worker yields.
450{-# NOINLINE checkRatePeriodic #-}
451checkRatePeriodic :: SVar t m a
456checkRatePeriodic sv yinfo winfo ycnt = do
457 i <- readIORef (workerPollingInterval yinfo)
458 -- XXX use generation count to check if the interval has been updated
459 if i /= 0 && (ycnt `mod` i) == 0
461 workerUpdateLatency yinfo winfo
462 -- XXX not required for parallel streams
463 isBeyondMaxRate sv yinfo
466-- CAUTION! this also updates the yield count and therefore should be called
467-- only when we are actually yielding an element.
468{-# NOINLINE workerRateControl #-}
469workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
470workerRateControl sv yinfo winfo = do
471 cnt <- updateYieldCount winfo
472 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
473 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
475-------------------------------------------------------------------------------
477-------------------------------------------------------------------------------
479-- XXX we should do rate control here but not latency update in case of ahead
480-- streams. latency update must be done when we yield directly to outputQueue
481-- or when we yield to heap.
483-- returns whether the worker should continue (True) or stop (False).
484{-# INLINE sendYield #-}
485sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
486sendYield sv mwinfo msg = do
487 oldlen <- send sv msg
488 let limit = maxBufferLimit sv
489 bufferSpaceOk <- case limit of
490 Unlimited -> return True
492 active <- readIORef (workerCount sv)
493 return $ (oldlen + 1) < (fromIntegral lim - active)
497 case yieldRateInfo sv of
498 Nothing -> return True
499 Just yinfo -> workerRateControl sv yinfo winfo
500 Nothing -> return True
501 return $ bufferSpaceOk && rateLimitOk
503-------------------------------------------------------------------------------
505-------------------------------------------------------------------------------
507{-# INLINE workerStopUpdate #-}
508workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
509workerStopUpdate winfo info = do
510 i <- readIORef (workerPollingInterval info)
511 when (i /= 0) $ workerUpdateLatency info winfo
513{-# INLINABLE sendStop #-}
514sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
515sendStop sv mwinfo = do
516 atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
517 case (mwinfo, yieldRateInfo sv) of
518 (Just winfo, Just info) ->
519 workerStopUpdate winfo info
522 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
524-- {-# NOINLINE sendStopToProducer #-}
525sendStopToProducer :: MonadIO m => SVar t m a -> m ()
526sendStopToProducer sv = liftIO $ do
528 void $ sendToProducer sv (ChildStop tid Nothing)
530-------------------------------------------------------------------------------
532-------------------------------------------------------------------------------
534{-# NOINLINE handleFoldException #-}
535handleFoldException :: SVar t m a -> SomeException -> IO ()
536handleFoldException sv e = do
538 void $ sendToProducer sv (ChildStop tid (Just e))
540{-# NOINLINE handleChildException #-}
541handleChildException :: SVar t m a -> SomeException -> IO ()
542handleChildException sv e = do
544 void $ send sv (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