streamly-0.11.0: src/Streamly/Internal/Data/SVar/Worker.hs
{-# OPTIONS_GHC -Wno-deprecations #-}
-- |
-- Module : Streamly.Internal.Data.SVar.Worker
-- Copyright : (c) 2017 Composewell Technologies
-- License : BSD-3-Clause
-- Maintainer : streamly@composewell.com
-- Stability : experimental
-- Portability : GHC
--
--
module Streamly.Internal.Data.SVar.Worker
{-# DEPRECATED "The functionality is moved to Channel.*" #-}
(
-- * Adjusting Limits
decrementYieldLimit
, incrementYieldLimit
, decrementBufferLimit
, incrementBufferLimit
, resetBufferLimit
-- * Rate Control
, Work (..)
, isBeyondMaxRate
, estimateWorkers
, updateYieldCount
, minThreadDelay
, workerRateControl
, workerUpdateLatency
-- * Send Events
, send
, ringDoorBell
-- ** Yield
, sendYield
, sendToProducer
-- ** Stop
, sendStop
, sendStopToProducer
-- ** Exception
, handleChildException
, handleFoldException
)
where
#include "inline.hs"
import Control.Concurrent (myThreadId, takeMVar)
import Control.Concurrent.MVar (MVar, tryPutMVar)
import Control.Exception (SomeException(..), assert)
import Control.Monad (when, void)
import Control.Monad.IO.Class (MonadIO(liftIO))
import Data.IORef (IORef, readIORef, writeIORef)
import Streamly.Internal.Data.Atomics
(atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
storeLoadBarrier)
import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
import Streamly.Internal.Data.Time.Units
(AbsTime, NanoSecond64(..), diffAbsTime64, fromRelTime64)
import Streamly.Internal.Data.SVar.Type
------------------------------------------------------------------------------
-- Collecting results from child workers in a streamed fashion
------------------------------------------------------------------------------
-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
-- faster, along with the fields in sv required by send?
-- XXX make it noinline
--
-- XXX we may want to employ an increment and decrement in batches when the
-- througput is high or when the cost of synchronization is high. For example
-- if the application is distributed then inc/dec of a shared variable may be
-- very costly.
--
-- A worker decrements the yield limit before it executes an action. However,
-- the action may not result in an element being yielded, in that case we have
-- to increment the yield limit.
--
-- Note that we need it to be an Int type so that we have the ability to undo a
-- decrement that takes it below zero.
{-# INLINE decrementYieldLimit #-}
decrementYieldLimit :: SVar t m a -> IO Bool
decrementYieldLimit sv =
case remainingWork sv of
Nothing -> return True
Just ref -> do
r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
return $ r >= 1
{-# INLINE incrementYieldLimit #-}
incrementYieldLimit :: SVar t m a -> IO ()
incrementYieldLimit sv =
case remainingWork sv of
Nothing -> return ()
Just ref -> atomicModifyIORefCAS_ ref (+ 1)
-- XXX exception safety of all atomic/MVar operations
-- TBD Each worker can have their own queue and the consumer can empty one
-- queue at a time, that way contention can be reduced.
-- XXX Only yields should be counted in the buffer limit and not the Stop
-- events.
{-# INLINE decrementBufferLimit #-}
decrementBufferLimit :: SVar t m a -> IO ()
decrementBufferLimit sv =
case maxBufferLimit sv of
Unlimited -> return ()
Limited _ -> do
let ref = pushBufferSpace sv
old <- atomicModifyIORefCAS ref $ \x ->
(if x >= 1 then x - 1 else x, x)
when (old <= 0) $
case pushBufferPolicy sv of
PushBufferBlock -> blockAndRetry
PushBufferDropNew -> do
-- We just drop one item and proceed. It is possible
-- that by the time we drop the item the consumer
-- thread might have run and created space in the
-- buffer, but we do not care about that condition.
-- This is not pedantically correct but it should be
-- fine in practice.
-- XXX we may want to drop only if n == maxBuf
-- otherwise we must have space in the buffer and a
-- decrement should be possible.
block <- atomicModifyIORefCAS (outputQueue sv) $
\(es, n) ->
case es of
[] -> (([],n), True)
_ : xs -> ((xs, n - 1), False)
when block blockAndRetry
-- XXX need a dequeue or ring buffer for this
PushBufferDropOld -> undefined
where
blockAndRetry = do
let ref = pushBufferSpace sv
liftIO $ takeMVar (pushBufferMVar sv)
old <- atomicModifyIORefCAS ref $ \x ->
(if x >= 1 then x - 1 else x, x)
-- When multiple threads sleep on takeMVar, the first thread would
-- wakeup due to a putMVar by the consumer, but the rest of the threads
-- would have to put back the MVar after taking it and decrementing the
-- buffer count, otherwise all other threads will remain asleep.
if old >= 1
then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
-- We do not put the MVar back in this case, instead we
-- wait for the consumer to put it.
else blockAndRetry
{-# INLINE incrementBufferLimit #-}
incrementBufferLimit :: SVar t m a -> IO ()
incrementBufferLimit sv =
case maxBufferLimit sv of
Unlimited -> return ()
Limited _ -> do
atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
writeBarrier
void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
{-# INLINE resetBufferLimit #-}
resetBufferLimit :: SVar t m a -> IO ()
resetBufferLimit sv =
case maxBufferLimit sv of
Unlimited -> return ()
Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
(const (fromIntegral n))
-------------------------------------------------------------------------------
-- Yield control
-------------------------------------------------------------------------------
updateYieldCount :: WorkerInfo -> IO Count
updateYieldCount winfo = do
cnt <- readIORef (workerYieldCount winfo)
let cnt1 = cnt + 1
writeIORef (workerYieldCount winfo) cnt1
return cnt1
isBeyondMaxYield :: Count -> WorkerInfo -> Bool
isBeyondMaxYield cnt winfo =
let ymax = workerYieldMax winfo
in ymax /= 0 && cnt >= ymax
-------------------------------------------------------------------------------
-- Sending results from worker
-------------------------------------------------------------------------------
{-# INLINE ringDoorBell #-}
ringDoorBell :: SVar t m a -> IO ()
ringDoorBell sv = do
storeLoadBarrier
w <- readIORef $ needDoorBell sv
when w $ do
-- Note: the sequence of operations is important for correctness here.
-- We need to set the flag to false strictly before sending the
-- outputDoorBell, otherwise the outputDoorBell may get processed too
-- early and then we may set the flag to False to later making the
-- consumer lose the flag, even without receiving a outputDoorBell.
atomicModifyIORefCAS_ (needDoorBell sv) (const False)
void $ tryPutMVar (outputDoorBell sv) ()
{-# INLINE sendWithDoorBell #-}
sendWithDoorBell ::
IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int
sendWithDoorBell q bell msg = do
-- XXX can the access to outputQueue be made faster somehow?
oldlen <- atomicModifyIORefCAS q $ \(es, n) ->
((msg : es, n + 1), n)
when (oldlen <= 0) $ do
-- The wake up must happen only after the store has finished otherwise
-- we can have lost wakeup problems.
writeBarrier
-- Since multiple workers can try this at the same time, it is possible
-- that we may put a spurious MVar after the consumer has already seen
-- the output. But that's harmless, at worst it may cause the consumer
-- to read the queue again and find it empty.
-- The important point is that the consumer is guaranteed to receive a
-- doorbell if something was added to the queue after it empties it.
void $ tryPutMVar bell ()
return oldlen
-- | This function is used by the producer threads to queue output for the
-- consumer thread to consume. Returns whether the queue has more space.
send :: SVar t m a -> ChildEvent a -> IO Int
send sv = sendWithDoorBell (outputQueue sv) (outputDoorBell sv)
-- There is no bound implemented on the buffer, this is assumed to be low
-- traffic.
sendToProducer :: SVar t m a -> ChildEvent a -> IO Int
sendToProducer sv msg = do
-- In case the producer stream is blocked on pushing to the fold buffer
-- then wake it up so that it can check for the stop event or exception
-- being sent to it otherwise we will be deadlocked.
void $ tryPutMVar (pushBufferMVar sv) ()
sendWithDoorBell (outputQueueFromConsumer sv)
(outputDoorBellFromConsumer sv) msg
-------------------------------------------------------------------------------
-- Collect and update worker latency
-------------------------------------------------------------------------------
workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
workerCollectLatency winfo = do
(cnt0, t0) <- readIORef (workerLatencyStart winfo)
cnt1 <- readIORef (workerYieldCount winfo)
let cnt = cnt1 - cnt0
if cnt > 0
then do
t1 <- getTime Monotonic
let period = fromRelTime64 $ diffAbsTime64 t1 t0
writeIORef (workerLatencyStart winfo) (cnt1, t1)
return $ Just (cnt, period)
else return Nothing
-- XXX There are a number of gotchas in measuring latencies.
-- 1) We measure latencies only when a worker yields a value
-- 2) It is possible that a stream calls the stop continuation, in which case
-- the worker would not yield a value and we would not account that worker in
-- latencies. Even though this case should ideally be accounted we do not
-- account it because we cannot or do not distinguish it from the case
-- described next.
-- 3) It is possible that a worker returns without yielding anything because it
-- never got a chance to pick up work.
-- 4) If the system timer resolution is lower than the latency, the latency
-- computation turns out to be zero.
--
-- We can fix this if we measure the latencies by counting the work items
-- picked rather than based on the outputs yielded.
workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
workerUpdateLatency yinfo winfo = do
r <- workerCollectLatency winfo
case r of
Just (cnt, period) -> do
-- NOTE: On JS platform the timer resolution could be pretty low. When
-- the timer resolution is low, measurement of latencies could be
-- tricky. All the worker latencies will turn out to be zero if they
-- are lower than the resolution. We only take into account those
-- measurements which are more than the timer resolution.
let ref = workerPendingLatency yinfo
(cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
atomicModifyIORefCAS_ ref $
\(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
Nothing -> return ()
-------------------------------------------------------------------------------
-- Worker rate control
-------------------------------------------------------------------------------
-- We either block, or send one worker with limited yield count or one or more
-- workers with unlimited yield count.
data Work
= BlockWait NanoSecond64
| PartialWorker Count
| ManyWorkers Int Count
deriving Show
-- | This is a magic number and it is overloaded, and used at several places to
-- achieve batching:
--
-- 1. If we have to sleep to slowdown this is the minimum period that we
-- accumulate before we sleep. Also, workers do not stop until this much
-- sleep time is accumulated.
-- 3. Collected latencies are computed and transferred to measured latency
-- after a minimum of this period.
minThreadDelay :: NanoSecond64
minThreadDelay = 1000000
-- | Another magic number! When we have to start more workers to cover up a
-- number of yields that we are lagging by then we cannot start one worker for
-- each yield because that may be a very big number and if the latency of the
-- workers is low these number of yields could be very high. We assume that we
-- run each extra worker for at least this much time.
rateRecoveryTime :: NanoSecond64
rateRecoveryTime = 1000000
-- | Get the worker latency without resetting workerPendingLatency
-- Returns (total yield count, base time, measured latency)
-- CAUTION! keep it in sync with collectLatency
getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
getWorkerLatency yinfo = do
let cur = workerPendingLatency yinfo
col = workerCollectedLatency yinfo
longTerm = svarAllTimeLatency yinfo
measured = workerMeasuredLatency yinfo
(curTotalCount, curCount, curTime) <- readIORef cur
(colTotalCount, colCount, colTime) <- readIORef col
(lcount, ltime) <- readIORef longTerm
prevLat <- readIORef measured
let latCount = colCount + curCount
latTime = colTime + curTime
totalCount = colTotalCount + curTotalCount
newLat =
if latCount > 0 && latTime > 0
then let lat = latTime `div` fromIntegral latCount
-- XXX Give more weight to new?
in (lat + prevLat) `div` 2
else prevLat
return (lcount + totalCount, ltime, newLat)
-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
estimateWorkers
:: Limit
-> Count
-> Count
-> NanoSecond64
-> NanoSecond64
-> NanoSecond64
-> LatencyRange
-> Work
estimateWorkers workerLimit svarYields gainLossYields
svarElapsed wLatency targetLat range =
-- XXX we can have a maxEfficiency combinator as well which runs the
-- producer at the maximal efficiency i.e. the number of workers are chosen
-- such that the latency is minimum or within a range. Or we can call it
-- maxWorkerLatency.
--
let
-- How many workers do we need to achieve the required rate?
--
-- When the workers are IO bound we can increase the throughput by
-- increasing the number of workers as long as the IO device has enough
-- capacity to process all the requests concurrently. If the IO
-- bandwidth is saturated increasing the workers won't help. Also, if
-- the CPU utilization in processing all these requests exceeds the CPU
-- bandwidth, then increasing the number of workers won't help.
--
-- When the workers are purely CPU bound, increasing the workers beyond
-- the number of CPUs won't help.
--
-- TODO - measure the CPU and IO requirements of the workers. Have a
-- way to specify the max bandwidth of the underlying IO mechanism and
-- use that to determine the max rate of workers, and also take the CPU
-- bandwidth into account. We can also discover the IO bandwidth if we
-- know that we are not CPU bound, then how much steady state rate are
-- we able to achieve. Design tests for CPU bound and IO bound cases.
-- Calculate how many yields are we ahead or behind to match the exact
-- required rate. Based on that we increase or decrease the effective
-- workers.
--
-- When the worker latency is lower than required latency we begin with
-- a yield and then wait rather than first waiting and then yielding.
targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
effectiveYields = svarYields + gainLossYields
deltaYields = fromIntegral targetYields - effectiveYields
-- We recover the deficit by running at a higher/lower rate for a
-- certain amount of time. To keep the effective rate in reasonable
-- limits we use rateRecoveryTime, minLatency and maxLatency.
in if deltaYields > 0
then
let deltaYieldsFreq :: Double
deltaYieldsFreq =
fromIntegral deltaYields /
fromIntegral rateRecoveryTime
yieldsFreq = 1.0 / fromIntegral targetLat
totalYieldsFreq = yieldsFreq + deltaYieldsFreq
requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
adjustedLat = min (max requiredLat (minLatency range))
(maxLatency range)
in assert (adjustedLat > 0) $
if wLatency <= adjustedLat
then PartialWorker deltaYields
else let workers = withLimit $ wLatency `div` adjustedLat
limited = min workers (fromIntegral deltaYields)
in ManyWorkers (fromIntegral limited) deltaYields
else
let expectedDuration = fromIntegral effectiveYields * targetLat
sleepTime = expectedDuration - svarElapsed
maxSleepTime = maxLatency range - wLatency
s = min sleepTime maxSleepTime
in assert (sleepTime >= 0) $
-- if s is less than 0 it means our maxSleepTime is less
-- than the worker latency.
if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
where
withLimit n =
case workerLimit of
Unlimited -> n
Limited x -> min n (fromIntegral x)
isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
isBeyondMaxRate sv yinfo = do
(count, tstamp, wLatency) <- getWorkerLatency yinfo
now <- getTime Monotonic
let duration = fromRelTime64 $ diffAbsTime64 now tstamp
let targetLat = svarLatencyTarget yinfo
gainLoss <- readIORef (svarGainedLostYields yinfo)
let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
wLatency targetLat (svarLatencyRange yinfo)
cnt <- readIORef $ workerCount sv
return $ case work of
-- XXX set the worker's maxYields or polling interval based on yields
PartialWorker _yields -> cnt > 1
ManyWorkers n _ -> cnt > n
BlockWait _ -> True
-- XXX we should do rate control periodically based on the total yields rather
-- than based on the worker local yields as other workers may have yielded more
-- and we should stop based on the aggregate yields. However, latency update
-- period can be based on individual worker yields.
{-# NOINLINE checkRatePeriodic #-}
checkRatePeriodic :: SVar t m a
-> YieldRateInfo
-> WorkerInfo
-> Count
-> IO Bool
checkRatePeriodic sv yinfo winfo ycnt = do
i <- readIORef (workerPollingInterval yinfo)
-- XXX use generation count to check if the interval has been updated
if i /= 0 && (ycnt `mod` i) == 0
then do
workerUpdateLatency yinfo winfo
-- XXX not required for parallel streams
isBeyondMaxRate sv yinfo
else return False
-- CAUTION! this also updates the yield count and therefore should be called
-- only when we are actually yielding an element.
{-# NOINLINE workerRateControl #-}
workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
workerRateControl sv yinfo winfo = do
cnt <- updateYieldCount winfo
beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
-------------------------------------------------------------------------------
-- Send a yield event
-------------------------------------------------------------------------------
-- XXX we should do rate control here but not latency update in case of ahead
-- streams. latency update must be done when we yield directly to outputQueue
-- or when we yield to heap.
--
-- returns whether the worker should continue (True) or stop (False).
{-# INLINE sendYield #-}
sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
sendYield sv mwinfo msg = do
oldlen <- send sv msg
let limit = maxBufferLimit sv
bufferSpaceOk <- case limit of
Unlimited -> return True
Limited lim -> do
active <- readIORef (workerCount sv)
return $ (oldlen + 1) < (fromIntegral lim - active)
rateLimitOk <-
case mwinfo of
Just winfo ->
case yieldRateInfo sv of
Nothing -> return True
Just yinfo -> workerRateControl sv yinfo winfo
Nothing -> return True
return $ bufferSpaceOk && rateLimitOk
-------------------------------------------------------------------------------
-- Send a Stop event
-------------------------------------------------------------------------------
{-# INLINE workerStopUpdate #-}
workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
workerStopUpdate winfo info = do
i <- readIORef (workerPollingInterval info)
when (i /= 0) $ workerUpdateLatency info winfo
{-# INLINABLE sendStop #-}
sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
sendStop sv mwinfo = do
atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
case (mwinfo, yieldRateInfo sv) of
(Just winfo, Just info) ->
workerStopUpdate winfo info
_ ->
return ()
myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
-- {-# NOINLINE sendStopToProducer #-}
sendStopToProducer :: MonadIO m => SVar t m a -> m ()
sendStopToProducer sv = liftIO $ do
tid <- myThreadId
void $ sendToProducer sv (ChildStop tid Nothing)
-------------------------------------------------------------------------------
-- Send exceptions
-------------------------------------------------------------------------------
{-# NOINLINE handleFoldException #-}
handleFoldException :: SVar t m a -> SomeException -> IO ()
handleFoldException sv e = do
tid <- myThreadId
void $ sendToProducer sv (ChildStop tid (Just e))
{-# NOINLINE handleChildException #-}
handleChildException :: SVar t m a -> SomeException -> IO ()
handleChildException sv e = do
tid <- myThreadId
void $ send sv (ChildStop tid (Just e))