@@ -40,8 +40,8 @@ module Streamly.SVar
40 , setMaxThreads
40 , setMaxThreads
41 , getMaxBuffer
41 , getMaxBuffer
42 , setMaxBuffer
42 , setMaxBuffer
43− , getMaxStreamRate
43+ , getStreamRate
44− , setMaxStreamRate
44+ , setStreamRate
45 , setStreamLatency
45 , setStreamLatency
46 , getYieldLimit
46 , getYieldLimit
47 , setYieldLimit
47 , setYieldLimit
@@ -72,6 +72,7 @@ module Streamly.SVar
72 , dequeueAhead
72 , dequeueAhead
73 , dequeueFromHeap
73 , dequeueFromHeap
75 , getYieldRateInfo
76 , getYieldRateInfo
76 , collectLatency
77 , collectLatency
77 , workerUpdateLatency
78 , workerUpdateLatency
@@ -234,9 +235,52 @@ data WorkerInfo = WorkerInfo
234 , workerLatencyStart :: IORef (Count, TimeSpec)
235 , workerLatencyStart :: IORef (Count, TimeSpec)
238+-- | Specify the stream yield rate.
243+ -- ^ Specifies the average production rate of a stream in number of yields
244+ -- per second (i.e. @Hertz@). Concurrent production is ramped up or down
245+ -- automatically to achieve the specified average yield rate. The rate can
246+ -- go down to half of the specified rate on the lower side and double of
247+ -- the specified rate on the higher side.
249+ -- ^ Specifies the minimum rate at which the stream should yield values. As
250+ -- far as possible the yield rate would never be allowed to go below the
251+ -- specified rate, even though it may possibly go above it at times, the
252+ -- upper limit is double of the specified rate.
254+ -- ^ Specifies the maximum rate at which the stream should yield values. As
255+ -- far as possible the yield rate would never be allowed to go above the
256+ -- specified rate, even though it may possibly go below it at times, the
257+ -- lower limit is half of the specified rate. This can be useful in
258+ -- applications where certain resource usage must not be allowed to go
259+ -- beyond certain limits.
261+ -- ^ Specifies a constant yield rate. If for some reason the actual rate
262+ -- goes above or below the specified rate we do not try to recover it by
263+ -- increasing or decreasing the rate in future. This can be useful in
264+ -- applications like graphics frame refresh where we need to maintain a
265+ -- constant refresh rate.
266+ | Rate Double Double Double
267+ -- ^ This is the most general specification of the stream yield rate and
268+ -- all the other specifications can be expressed in terms of this. The
269+ -- first argument is the target rate, the second argument is the lower
270+ -- limit and the third argument is the upper limit. We try to keep the
271+ -- average rate at the target rate specified, if the actual rate goes above
272+ -- or below we try to recover the rate by increasing or decreasing it in
273+ -- future while keeping the maximum or minimum within the specified range.
275+data LatencyRange = LatencyRange
276+ { minLatency :: NanoSecs
277+ , maxLatency :: NanoSecs
237-- Rate control.
280-- Rate control.
238data YieldRateInfo = YieldRateInfo
281data YieldRateInfo = YieldRateInfo
239− { svarExpectedLatency :: NanoSecs
282+ { svarLatencyTarget :: NanoSecs
283+ , svarLatencyRange :: LatencyRange
241 -- Actual latency/througput as seen from the consumer side, we count the
285 -- Actual latency/througput as seen from the consumer side, we count the
242 -- yields and the time it took to generates those yields. This is used to
286 -- yields and the time it took to generates those yields. This is used to
@@ -256,7 +300,7 @@ data YieldRateInfo = YieldRateInfo
256 -- long time, in such cases the consumer can change it.
300 -- long time, in such cases the consumer can change it.
257 -- 0 means no latency computation
301 -- 0 means no latency computation
258 -- XXX this is derivable from workerMeasuredLatency, can be removed.
302 -- XXX this is derivable from workerMeasuredLatency, can be removed.
259− , workerLatencyUpdateInterval :: IORef Count
303+ , workerPollingInterval :: IORef Count
261 -- This is in progress latency stats maintained by the workers which we
305 -- This is in progress latency stats maintained by the workers which we
262 -- empty into workerCollectedLatency stats at certain intervals - whenever
306 -- empty into workerCollectedLatency stats at certain intervals - whenever
@@ -355,8 +399,8 @@ data State t m a = State
355 , _threadsHigh :: Limit
399 , _threadsHigh :: Limit
356 , _bufferHigh :: Limit
400 , _bufferHigh :: Limit
357 -- XXX these two can be collapsed into a single type
401 -- XXX these two can be collapsed into a single type
358− , _streamLatency :: Maybe NanoSecs
402+ , _streamLatency :: Maybe NanoSecs -- bootstrap latency
359− , _maxStreamRate :: Maybe Double -- yields per second
403+ , _maxStreamRate :: Maybe Rate
362-------------------------------------------------------------------------------
406-------------------------------------------------------------------------------
@@ -444,18 +488,11 @@ setMaxBuffer n st =
444getMaxBuffer :: State t m a -> Limit
488getMaxBuffer :: State t m a -> Limit
445getMaxBuffer = _bufferHigh
489getMaxBuffer = _bufferHigh
447−setMaxStreamRate :: Double -> State t m a -> State t m a
491+setStreamRate :: Maybe Rate -> State t m a -> State t m a
448−setMaxStreamRate n st =
449− st { _maxStreamRate =
492+setStreamRate r st = st { _maxStreamRate = r }
457−getMaxStreamRate :: State t m a -> Maybe Double
494+getStreamRate :: State t m a -> Maybe Rate
458−getMaxStreamRate = _maxStreamRate
495+getStreamRate = _maxStreamRate
460setStreamLatency :: Int -> State t m a -> State t m a
497setStreamLatency :: Int -> State t m a -> State t m a
461setStreamLatency n st =
498setStreamLatency n st =
@@ -651,11 +688,13 @@ dumpSVar sv = do
651 , "---------CURRENT STATE-----------"
688 , "---------CURRENT STATE-----------"
652 , "outputQueue length computed = " ++ show (length oqList)
689 , "outputQueue length computed = " ++ show (length oqList)
653 , "outputQueue length maintained = " ++ show oqLen
690 , "outputQueue length maintained = " ++ show oqLen
691+ -- XXX print the types of events in the outputQueue, first 5
654 , "outputDoorBell = " ++ show db
692 , "outputDoorBell = " ++ show db
656 ++ aheadDump ++ unlines
694 ++ aheadDump ++ unlines
657 [ "needDoorBell = " ++ show waiting
695 [ "needDoorBell = " ++ show waiting
658 , "running threads = " ++ show rthread
696 , "running threads = " ++ show rthread
697+ -- XXX print the status of first 5 threads
659 , "running thread count = " ++ show workers
698 , "running thread count = " ++ show workers
661 ++ "---------STATS-----------\n"
700 ++ "---------STATS-----------\n"
@@ -853,7 +892,8 @@ checkRatePeriodic :: SVar t m a
854 -> IO Bool
893 -> IO Bool
855checkRatePeriodic sv yinfo winfo ycnt = do
894checkRatePeriodic sv yinfo winfo ycnt = do
856− i <- readIORef (workerLatencyUpdateInterval yinfo)
895+ i <- readIORef (workerPollingInterval yinfo)
896+ -- XXX use generation count to check if the interval has been updated
857 if (i /= 0 && (ycnt `mod` i) == 0)
897 if (i /= 0 && (ycnt `mod` i) == 0)
859 workerUpdateLatency yinfo winfo
899 workerUpdateLatency yinfo winfo
@@ -886,7 +926,7 @@ sendYield sv winfo msg = do
886{-# INLINE workerStopUpdate #-}
926{-# INLINE workerStopUpdate #-}
887workerStopUpdate :: WorkerInfo -> YieldRateInfo -> Int -> IO ()
927workerStopUpdate :: WorkerInfo -> YieldRateInfo -> Int -> IO ()
888workerStopUpdate winfo info n = do
928workerStopUpdate winfo info n = do
889− i <- readIORef (workerLatencyUpdateInterval info)
929+ i <- readIORef (workerPollingInterval info)
890 when (i /= 0) $ workerUpdateLatency info winfo
930 when (i /= 0) $ workerUpdateLatency info winfo
891 when (n == 1) $ do
931 when (n == 1) $ do
892 t <- getTime Monotonic
932 t <- getTime Monotonic
@@ -1265,51 +1305,100 @@ minThreadDelay = 10^(6 :: Int)
1265-- | Another magic number! When we have to start more workers to cover up a
1305-- | Another magic number! When we have to start more workers to cover up a
1266-- number of yields that we are lagging by then we cannot start one worker for
1306-- number of yields that we are lagging by then we cannot start one worker for
1267-- each yield because that may be a very big number and if the latency of the
1307-- each yield because that may be a very big number and if the latency of the
1268−-- workers is low these number of yields could be very high. We use this number
1308+-- workers is low these number of yields could be very high. We assume that we
1269−-- as the duration in which we want to cover any lag. In other words we
1309+-- run each extra worker for at least this much time.
1270−-- accumulate the lag (lagging number of yields) such that it covers this much
1310+rateRecoveryTime :: NanoSecs
1271−-- duration for a single worker to produce those yields, before we start an
1311+rateRecoveryTime = 1000000
1273−coverUpLatency :: NanoSecs
1274−coverUpLatency = 1000000
1276nanoToMicroSecs :: NanoSecs -> Int
1313nanoToMicroSecs :: NanoSecs -> Int
1277nanoToMicroSecs s = (fromIntegral s) `div` 1000
1314nanoToMicroSecs s = (fromIntegral s) `div` 1000
1316+-- We either block, or send one worker with limited yield count or one or more
1317+-- workers with unlimited yield count.
1319+ = BlockWait NanoSecs
1320+ | PartialWorker Count
1321+ | ManyWorkers Int Count
1279-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1324-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1280−countDispatchWorkers
1325+estimateWorkers
1281− :: Limit -> Count -> NanoSecs -> NanoSecs -> NanoSecs -> Int
1326+ :: Limit
1282−countDispatchWorkers workerLimit svarYields svarElapsed wLatency expectedLat =
1333+estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
1283− -- Calculate how many workers do we need to maintain the rate.
1285− -- XXX Increasing the workers may reduce the overall latency per
1286− -- yield of the workers, sometimes quite a bit. We can keep a
1287− -- latency histogram for 1,2,4..2^20 workers so that we can decide
1288− -- which bucket is optimal and decide the number of workers based
1291− -- XXX put a cap on extraworkers? Should not be more than nWorkers.
1292− -- Bridge the gap slowly rather than in a burst in one go.
1294 -- XXX we can have a maxEfficiency combinator as well which runs the
1334 -- XXX we can have a maxEfficiency combinator as well which runs the
1295 -- producer at the maximal efficiency i.e. the number of workers are chosen
1335 -- producer at the maximal efficiency i.e. the number of workers are chosen
1296 -- such that the latency is minimum or within a range. Or we can call it
1336 -- such that the latency is minimum or within a range. Or we can call it
1297 -- maxWorkerLatency.
1337 -- maxWorkerLatency.
1299− let nWorkers = (fromIntegral wLatency) / (fromIntegral expectedLat)
1339+ let
1340+ -- How many workers do we need to acheive the required rate?
1342+ -- When the workers are IO bound we can increase the throughput by
1343+ -- increasing the number of workers as long as the IO device has enough
1344+ -- capacity to process all the requests concurrently. If the IO
1345+ -- bandwidth is saturated increasing the workers won't help. Also, if
1346+ -- the CPU utilization in processing all these requests exceeds the CPU
1347+ -- bandwidth, then increasing the number of workers won't help.
1349+ -- When the workers are purely CPU bound, increasing the workers beyond
1350+ -- the number of CPUs won't help.
1352+ -- TODO - measure the CPU and IO requirements of the workers. Have a
1353+ -- way to specify the max bandwidth of the underlying IO mechanism and
1354+ -- use that to determine the max rate of workers, and also take the CPU
1355+ -- bandwidth into account. We can also discover the IO bandwidth if we
1356+ -- know that we are not CPU bound, then how much steady state rate are
1357+ -- we able to acheive. Design tests for CPU bound and IO bound cases.
1300 -- Calculate how many yields are we ahead or behind to match the exact
1359 -- Calculate how many yields are we ahead or behind to match the exact
1301 -- required rate. Based on that we increase or decrease the effective
1360 -- required rate. Based on that we increase or decrease the effective
1302 -- workers.
1361 -- workers.
1303− d = fromIntegral svarElapsed
1362+ --
1304− e = fromIntegral expectedLat
1363+ -- When the worker latency is lower than required latency we begin with
1305− n = max 1 (fromIntegral coverUpLatency / fromIntegral wLatency)
1364+ -- a yield and then wait rather than first waiting and then yielding.
1365+ targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1366+ deltaYields = fromIntegral targetYields - svarYields
1368+ -- We recover the deficit by running at a higher/lower rate for a
1369+ -- certain amount of time. To keep the effective rate in reasonable
1370+ -- limits we use rateRecoveryTime, minLatency and maxLatency.
1371+ in if deltaYields > 0
1306− :: Double
1373+ let deltaYieldsFreq :: Double
1307− extraYields = (d / e) - (fromIntegral svarYields)
1374+ deltaYieldsFreq =
1308− extraWorkers = extraYields / n
1375+ fromIntegral deltaYields /
1309− requiredWorkers = round $ nWorkers + extraWorkers
1376+ fromIntegral rateRecoveryTime
1377+ yieldsFreq = 1.0 / fromIntegral targetLat
1378+ totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1379+ adjustedLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1380+ requiredLat = min (max adjustedLat (minLatency range))
1382+ in assert (requiredLat > 0) $
1383+ if wLatency <= requiredLat
1384+ then PartialWorker deltaYields
1385+ else ManyWorkers ( fromIntegral
1387+ $ wLatency `div` requiredLat) deltaYields
1389+ let expectedDuration = (fromIntegral svarYields) * targetLat
1390+ sleepTime = expectedDuration - svarElapsed
1391+ maxSleepTime = maxLatency range - wLatency
1392+ s = min sleepTime maxSleepTime
1393+ in assert (sleepTime >= 0) $
1394+ -- if s is less than 0 it means our maxSleepTime is less
1395+ -- than the worker latency.
1396+ if (s > 0) then BlockWait s else ManyWorkers 1 (Count 0)
1310− in case workerLimit of
1399+ case workerLimit of
1311− Unlimited -> requiredWorkers
1400+ Unlimited -> n
1312− Limited x -> min requiredWorkers (fromIntegral x)
1401+ Limited x -> min n (fromIntegral x)
1314-- | Get the worker latency without resetting workerPendingLatency
1403-- | Get the worker latency without resetting workerPendingLatency
1315-- Returns (total yield count, base time, measured latency)
1404-- Returns (total yield count, base time, measured latency)
@@ -1341,19 +1430,33 @@ isBeyondMaxRate sv yinfo = do
1341 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1430 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1342 now <- getTime Monotonic
1431 now <- getTime Monotonic
1343 let duration = fromInteger $ toNanoSecs $ now - tstamp
1432 let duration = fromInteger $ toNanoSecs $ now - tstamp
1344− let expectedLat = svarExpectedLatency yinfo
1433+ let targetLat = svarLatencyTarget yinfo
1345− let netWorkers = countDispatchWorkers (maxWorkerLimit sv) count duration
1434+ let work = estimateWorkers (maxWorkerLimit sv) count duration wLatency
1346− wLatency expectedLat
1435+ targetLat (svarLatencyRange yinfo)
1347 cnt <- readIORef $ workerCount sv
1436 cnt <- readIORef $ workerCount sv
1348− if netWorkers <= 0
1437+ return $ case work of
1349− then do
1438+ -- XXX set the worker's maxYields or polling interval based on yields
1350− let sleepTime = ((fromIntegral count) * expectedLat) - duration
1439+ PartialWorker _yields -> cnt > 1
1440+ ManyWorkers n _ -> cnt > n
1441+ BlockWait _ -> True
1443+-- Every once in a while workers update the latencies and check the yield rate.
1444+-- They return if we are above the expected yield rate. If we check too often
1445+-- it may impact performance, if we check less often we may have a stale
1446+-- picture. We update every minThreadDelay but we translate that into a yield
1447+-- count based on latency so that the checking overhead is little.
1449+-- XXX use a generation count to indicate that the value is updated. If the
1450+-- value is updated an existing worker must check it again on the next yield.
1451+-- Otherwise it is possible that we may keep updating it and because of the mod
1452+-- worker keeps skipping it.
1453+updateWorkerPollingInterval :: YieldRateInfo -> NanoSecs -> IO ()
1454+updateWorkerPollingInterval yinfo latency = do
1455+ let periodRef = workerPollingInterval yinfo
1351− if sleepTime >= minThreadDelay
1456+ cnt = max 1 $ minThreadDelay `div` latency
1352− then return True
1457+ period = min cnt (fromIntegral magicMaxBuffer)
1353− else return False
1458+
1354− else if cnt > netWorkers
1459+ writeIORef periodRef (fromIntegral period)
1358useCollectedBatch :: YieldRateInfo
1461useCollectedBatch :: YieldRateInfo
1359 -> NanoSecs
1462 -> NanoSecs
@@ -1361,14 +1464,10 @@ useCollectedBatch :: YieldRateInfo
1361 -> IORef NanoSecs
1464 -> IORef NanoSecs
1362 -> IO ()
1465 -> IO ()
1363useCollectedBatch yinfo latency colRef measRef = do
1466useCollectedBatch yinfo latency colRef measRef = do
1364− let periodRef = workerLatencyUpdateInterval yinfo
1467+ updateWorkerPollingInterval yinfo latency
1365− cnt = max 1 $ minThreadDelay `div` latency
1468+ writeIORef colRef (0, 0)
1366− period = min cnt (fromIntegral magicMaxBuffer)
1368 -- XXX we should use some weight for the old latency as well in arriving at
1469 -- XXX we should use some weight for the old latency as well in arriving at
1369 -- the new latency, we have the counts, we can use that for some weighting.
1470 -- the new latency, we have the counts, we can use that for some weighting.
1370− writeIORef periodRef (fromIntegral period)
1371− writeIORef colRef (0, 0)
1372 writeIORef measRef latency
1471 writeIORef measRef latency
1374-- Returns a triple, (1) yield count since last collection, (2) the base time
1473-- Returns a triple, (1) yield count since last collection, (2) the base time
@@ -1467,51 +1566,76 @@ dispatchWorkerPaced sv = do
1467 then return False
1566 then return False
1469 let workerLimit = maxWorkerLimit sv
1568 let workerLimit = maxWorkerLimit sv
1470− let expectedLat = svarExpectedLatency yinfo
1569+ let targetLat = svarLatencyTarget yinfo
1471− let netWorkers = countDispatchWorkers
1570+ let range = svarLatencyRange yinfo
1472− workerLimit svarYields svarElapsed
1571+ let work = estimateWorkers workerLimit svarYields svarElapsed
1473− wLatency expectedLat
1572+ wLatency targetLat range
1475 -- XXX we need to take yieldLimit into account here. If we are at the
1574 -- XXX we need to take yieldLimit into account here. If we are at the
1476 -- end of the limit as well as the time, we should not be sleeping.
1575 -- end of the limit as well as the time, we should not be sleeping.
1477 -- If we are not actually planning to dispatch any more workers we need
1576 -- If we are not actually planning to dispatch any more workers we need
1478 -- to take that in account.
1577 -- to take that in account.
1580+ assert (s >= 0) (return ())
1581+ -- XXX note that when we return from here we will block waiting
1582+ -- for the result from the existing worker. If that takes too
1583+ -- long we won't be able to send another worker until the
1586+ -- Sleep only if there are no active workers, otherwise we will
1587+ -- defer the output of those. Note we cannot use workerCount
1588+ -- here as it is not a reliable way to ensure there are
1589+ -- definitely no active workers. When workerCount is 0 we may
1590+ -- still have a Stop event waiting in the outputQueue.
1591+ done <- allThreadsDone sv
1592+ when done $ void $ do
1593+ liftIO $ threadDelay $ nanoToMicroSecs s
1594+ dispatchWorker 1 sv
1596+ PartialWorker yields -> do
1597+ assert (yields > 0) (return ())
1598+ done <- allThreadsDone sv
1599+ when done $ void $ dispatchWorker yields sv
1479− if netWorkers <= 0
1601+ ManyWorkers netWorkers yields -> do
1602+ assert (netWorkers >= 1) (return ())
1603+ assert (yields >= 0) (return ())
1605+ let periodRef = workerPollingInterval yinfo
1606+ ycnt = max 1 $ yields `div` fromIntegral netWorkers
1607+ period = min ycnt (fromIntegral magicMaxBuffer)
1609+ old <- liftIO $ readIORef periodRef
1610+ when (period < old) $
1611+ liftIO $ writeIORef periodRef period
1613+ cnt <- liftIO $ readIORef $ workerCount sv
1614+ if (cnt < netWorkers)
1480− then do
1615+ then do
1481− let expectedDuration = (fromIntegral svarYields) * expectedLat
1616+ let total = netWorkers - cnt
1482− let sleepTime = expectedDuration - svarElapsed
1617+ batch = max 1 $ fromIntegral $
1618+ minThreadDelay `div` targetLat
1619+ r <- dispatchN (min total batch)
1620+ -- XXX stagger the workers over a period?
1621+ -- XXX cannot sleep, as that would mean we cannot process the
1622+ -- outputs. need to try a different mechanism to stagger.
1623+ -- when (total > batch) $
1624+ -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1484− when (sleepTime >= minThreadDelay) $ do
1628+ where
1485− liftIO $ threadDelay $ nanoToMicroSecs sleepTime
1487− cnt <- liftIO $ readIORef $ workerCount sv
1630+ dispatchN n = do
1488− when (cnt < 1) $ void $ do
1631+ if n == 0
1489− let n = countDispatchYields (expectedLat - wLatency)
1490− in dispatchWorker (fromIntegral (max n 1)) sv
1491− return False
1632+ then return True
1493− -- XXX stagger the workers over a period
1634+ r <- dispatchWorker 0 sv
1494− cnt <- liftIO $ readIORef $ workerCount sv
1635+ if r
1495− if (cnt < netWorkers)
1636+ then dispatchN (n - 1)
1496− then dispatchWorker 0 sv
1497 else return False
1637 else return False
1501− countDispatchYields :: NanoSecs -> Int
1502− countDispatchYields idleTime =
1503− assert (idleTime >= 0) $
1504− -- we do not want to sleep for less than minThreadDelay
1505− let maxYieldsPerWorker = fromIntegral magicMaxBuffer
1506− in if idleTime < minThreadDelay
1509− then maxYieldsPerWorker
1511− let n = (minThreadDelay + idleTime) `div` idleTime
1512− in min (fromIntegral n) maxYieldsPerWorker
1515sendWorkerDelayPaced :: SVar t m a -> IO ()
1639sendWorkerDelayPaced :: SVar t m a -> IO ()
1516sendWorkerDelayPaced _ = return ()
1640sendWorkerDelayPaced _ = return ()
@@ -1540,7 +1664,7 @@ sendWorkerWait
1540sendWorkerWait delay dispatch sv = do
1664sendWorkerWait delay dispatch sv = do
1541 -- Note that we are guaranteed to have at least one outstanding worker when
1665 -- Note that we are guaranteed to have at least one outstanding worker when
1542 -- we enter this function. So if we sleep we are guaranteed to be woken up
1666 -- we enter this function. So if we sleep we are guaranteed to be woken up
1543− -- by a outputDoorBell, when the worker exits.
1667+ -- by an outputDoorBell, when the worker exits.
1545 liftIO $ delay sv
1669 liftIO $ delay sv
1546 (_, n) <- liftIO $ readIORef (outputQueue sv)
1670 (_, n) <- liftIO $ readIORef (outputQueue sv)
@@ -1680,8 +1804,10 @@ postProcessBounded sv = do
1680 if workersDone
1804 if workersDone
1682 r <- liftIO $ isWorkDone sv
1806 r <- liftIO $ isWorkDone sv
1683− -- XXX do we need to dispatch many here?
1807+ -- Note that we need to guarantee a worker, therefore we cannot just
1808+ -- use dispatchWorker which may or may not send a worker.
1684 when (not r) $ pushWorker 0 sv
1809 when (not r) $ pushWorker 0 sv
1810+ -- XXX do we need to dispatch many here?
1685 -- void $ dispatchWorker sv
1811 -- void $ dispatchWorker sv
1686 return r
1812 return r
1687 else return False
1813 else return False
@@ -1699,34 +1825,63 @@ postProcessPaced sv = do
1699 let yinfo = fromJust $ yieldRateInfo sv
1825 let yinfo = fromJust $ yieldRateInfo sv
1700 liftIO $ adjustLatencies yinfo
1826 liftIO $ adjustLatencies yinfo
1701 void $ dispatchWorkerPaced sv
1827 void $ dispatchWorkerPaced sv
1828+ -- Note that we need to guarantee a worker since the work is not
1829+ -- finished, therefore we cannot just rely on dispatchWorkerPaced
1830+ -- which may or may not send a worker.
1831+ noWorker <- allThreadsDone sv
1832+ when noWorker $ pushWorker 0 sv
1702 return r
1833 return r
1703 else return False
1834 else return False
1705getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1836getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
1706getYieldRateInfo st = do
1837getYieldRateInfo st = do
1707− case getMaxStreamRate st of
1838+ let toLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1708− Just rate -> do
1839+ toMaxLatency r = if r <= 0 then maxBound else r
1709− let latency = round $ 1.0e9 / rate
1840+ case getStreamRate st of
1710− measured <- newIORef 0
1841+ Just (AvgRate rate) ->
1711− wcur <- newIORef (0,0)
1842+ let l = toLatency rate
1712− wcol <- newIORef (0,0)
1843+ mx = toMaxLatency (l * 2)
1713− now <- getTime Monotonic
1844+ in mkYieldRateInfo l (LatencyRange (l `div` 2) mx)
1714− wlong <- newIORef (0,now)
1845+ Just (MinRate rate) ->
1715− period <- newIORef 1
1846+ let l = toLatency rate
1716− stopTime <- newIORef $ fromNanoSecs 0
1847+ in mkYieldRateInfo l (LatencyRange (l `div` 2) l)
1717−
1848+ Just (MaxRate rate) ->
1718− return $ Just YieldRateInfo
1849+ let l = toLatency rate
1719− { svarExpectedLatency = latency
1850+ mx = toMaxLatency (l * 2)
1720− , workerBootstrapLatency = getStreamLatency st
1851+ in mkYieldRateInfo l (LatencyRange l mx)
1721− , workerLatencyUpdateInterval = period
1852+ Just (ConstRate rate) ->
1722− , workerMeasuredLatency = measured
1853+ let l = toLatency rate
1723− , workerPendingLatency = wcur
1854+ in mkYieldRateInfo l (LatencyRange l l)
1724− , workerStopTimeStamp = stopTime
1855+ Just (Rate rate minr maxr) ->
1725− , workerCollectedLatency = wcol
1856+ let l = toLatency rate
1726− , svarAllTimeLatency = wlong
1857+ minl = toLatency minr
1727− }
1858+ maxl = toLatency maxr
1859+ in mkYieldRateInfo l (LatencyRange minl maxl)
1728 Nothing -> return Nothing
1860 Nothing -> return Nothing
1864+ mkYieldRateInfo latency latRange = do
1865+ measured <- newIORef 0
1866+ wcur <- newIORef (0,0)
1867+ wcol <- newIORef (0,0)
1868+ now <- getTime Monotonic
1869+ wlong <- newIORef (0,now)
1870+ period <- newIORef 1
1871+ stopTime <- newIORef $ fromNanoSecs 0
1873+ return $ Just YieldRateInfo
1874+ { svarLatencyTarget = latency
1875+ , svarLatencyRange = latRange
1876+ , workerBootstrapLatency = getStreamLatency st
1877+ , workerPollingInterval = period
1878+ , workerMeasuredLatency = measured
1879+ , workerPendingLatency = wcur
1880+ , workerStopTimeStamp = stopTime
1881+ , workerCollectedLatency = wcol
1882+ , svarAllTimeLatency = wlong
1730getAheadSVar :: MonadAsync m
1885getAheadSVar :: MonadAsync m
1731 => State t m a
1886 => State t m a
1732 -> ( IORef ([t m a], Int)
1887 -> ( IORef ([t m a], Int)
@@ -1801,7 +1956,7 @@ getAheadSVar st f = do
1803 let sv =
1958 let sv =
1804− case getMaxStreamRate st of
1959+ case getStreamRate st of
1805 Nothing -> getSVar sv readOutputQBounded postProcessBounded
1960 Nothing -> getSVar sv readOutputQBounded postProcessBounded
1806 Just _ -> getSVar sv readOutputQPaced postProcessPaced
1961 Just _ -> getSVar sv readOutputQPaced postProcessPaced
1807 in return sv
1962 in return sv
@@ -1909,7 +2064,10 @@ sendFirstWorker sv m = do
1909 liftIO $ enqueue sv m
2064 liftIO $ enqueue sv m
1910 case yieldRateInfo sv of
2065 case yieldRateInfo sv of
1911 Nothing -> pushWorker 0 sv
2066 Nothing -> pushWorker 0 sv
2068+ if svarLatencyTarget yinfo == maxBound
2069+ then liftIO $ threadDelay maxBound
1912− Just _ -> pushWorker 1 sv
2070+ else pushWorker 1 sv
1913 return sv
2071 return sv
1915{-# INLINABLE newAheadVar #-}
2073{-# INLINABLE newAheadVar #-}