@@ -40,8 +40,8 @@ module Streamly.SVar
4040 , setMaxThreads
4141 , getMaxBuffer
4242 , setMaxBuffer
43− , getMaxStreamRate
44− , setMaxStreamRate
43+ , getStreamRate
44+ , setStreamRate
4545 , setStreamLatency
4646 , getYieldLimit
4747 , setYieldLimit
@@ -72,6 +72,7 @@ module Streamly.SVar
7272 , dequeueAhead
7373 , dequeueFromHeap
7474
75+ , Rate (..)
7576 , getYieldRateInfo
7677 , collectLatency
7778 , workerUpdateLatency
@@ -234,9 +235,52 @@ data WorkerInfo = WorkerInfo
234235 , workerLatencyStart :: IORef (Count, TimeSpec)
235236 }
236237
238+-- | Specify the stream yield rate.
239+--
240+-- @since 0.5.0
241+data Rate =
242+ AvgRate Double
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.
248+ | MinRate Double
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.
253+ | MaxRate Double
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.
260+ | ConstRate Double
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.
274+
275+data LatencyRange = LatencyRange
276+ { minLatency :: NanoSecs
277+ , maxLatency :: NanoSecs
278+ } deriving Show
279+
237280-- Rate control.
238281data YieldRateInfo = YieldRateInfo
239− { svarExpectedLatency :: NanoSecs
282+ { svarLatencyTarget :: NanoSecs
283+ , svarLatencyRange :: LatencyRange
240284
241285 -- Actual latency/througput as seen from the consumer side, we count the
242286 -- yields and the time it took to generates those yields. This is used to
@@ -256,7 +300,7 @@ data YieldRateInfo = YieldRateInfo
256300 -- long time, in such cases the consumer can change it.
257301 -- 0 means no latency computation
258302 -- XXX this is derivable from workerMeasuredLatency, can be removed.
259− , workerLatencyUpdateInterval :: IORef Count
303+ , workerPollingInterval :: IORef Count
260304
261305 -- This is in progress latency stats maintained by the workers which we
262306 -- empty into workerCollectedLatency stats at certain intervals - whenever
@@ -355,8 +399,8 @@ data State t m a = State
355399 , _threadsHigh :: Limit
356400 , _bufferHigh :: Limit
357401 -- XXX these two can be collapsed into a single type
358− , _streamLatency :: Maybe NanoSecs
359− , _maxStreamRate :: Maybe Double -- yields per second
402+ , _streamLatency :: Maybe NanoSecs -- bootstrap latency
403+ , _maxStreamRate :: Maybe Rate
360404 }
361405
362406-------------------------------------------------------------------------------
@@ -444,18 +488,11 @@ setMaxBuffer n st =
444488getMaxBuffer :: State t m a -> Limit
445489getMaxBuffer = _bufferHigh
446490
447−setMaxStreamRate :: Double -> State t m a -> State t m a
448−setMaxStreamRate n st =
449− st { _maxStreamRate =
450− if n < 0
451− then Nothing
452− else if n == 0
453− then Nothing
454− else Just n
455− }
491+setStreamRate :: Maybe Rate -> State t m a -> State t m a
492+setStreamRate r st = st { _maxStreamRate = r }
456493
457−getMaxStreamRate :: State t m a -> Maybe Double
458−getMaxStreamRate = _maxStreamRate
494+getStreamRate :: State t m a -> Maybe Rate
495+getStreamRate = _maxStreamRate
459496
460497setStreamLatency :: Int -> State t m a -> State t m a
461498setStreamLatency n st =
@@ -651,11 +688,13 @@ dumpSVar sv = do
651688 , "---------CURRENT STATE-----------"
652689 , "outputQueue length computed = " ++ show (length oqList)
653690 , "outputQueue length maintained = " ++ show oqLen
691+ -- XXX print the types of events in the outputQueue, first 5
654692 , "outputDoorBell = " ++ show db
655693 ]
656694 ++ aheadDump ++ unlines
657695 [ "needDoorBell = " ++ show waiting
658696 , "running threads = " ++ show rthread
697+ -- XXX print the status of first 5 threads
659698 , "running thread count = " ++ show workers
660699 ]
661700 ++ "---------STATS-----------\n"
@@ -853,7 +892,8 @@ checkRatePeriodic :: SVar t m a
853892 -> Count
854893 -> IO Bool
855894checkRatePeriodic 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
857897 if (i /= 0 && (ycnt `mod` i) == 0)
858898 then do
859899 workerUpdateLatency yinfo winfo
@@ -886,7 +926,7 @@ sendYield sv winfo msg = do
886926{-# INLINE workerStopUpdate #-}
887927workerStopUpdate :: WorkerInfo -> YieldRateInfo -> Int -> IO ()
888928workerStopUpdate winfo info n = do
889− i <- readIORef (workerLatencyUpdateInterval info)
929+ i <- readIORef (workerPollingInterval info)
890930 when (i /= 0) $ workerUpdateLatency info winfo
891931 when (n == 1) $ do
892932 t <- getTime Monotonic
@@ -1265,51 +1305,100 @@ minThreadDelay = 10^(6 :: Int)
12651305-- | Another magic number! When we have to start more workers to cover up a
12661306-- number of yields that we are lagging by then we cannot start one worker for
12671307-- 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
1269−-- as the duration in which we want to cover any lag. In other words we
1270−-- accumulate the lag (lagging number of yields) such that it covers this much
1271−-- duration for a single worker to produce those yields, before we start an
1272−-- extra worker.
1273−coverUpLatency :: NanoSecs
1274−coverUpLatency = 1000000
1308+-- workers is low these number of yields could be very high. We assume that we
1309+-- run each extra worker for at least this much time.
1310+rateRecoveryTime :: NanoSecs
1311+rateRecoveryTime = 1000000
12751312
12761313nanoToMicroSecs :: NanoSecs -> Int
12771314nanoToMicroSecs s = (fromIntegral s) `div` 1000
12781315
1316+-- We either block, or send one worker with limited yield count or one or more
1317+-- workers with unlimited yield count.
1318+data Work
1319+ = BlockWait NanoSecs
1320+ | PartialWorker Count
1321+ | ManyWorkers Int Count
1322+ deriving Show
1323+
12791324-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1280−countDispatchWorkers
1281− :: Limit -> Count -> NanoSecs -> NanoSecs -> NanoSecs -> Int
1282−countDispatchWorkers workerLimit svarYields svarElapsed wLatency expectedLat =
1283− -- Calculate how many workers do we need to maintain the rate.
1284−
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
1289− -- on that.
1290− --
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.
1293− --
1325+estimateWorkers
1326+ :: Limit
1327+ -> Count
1328+ -> NanoSecs
1329+ -> NanoSecs
1330+ -> NanoSecs
1331+ -> LatencyRange
1332+ -> Work
1333+estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =
12941334 -- XXX we can have a maxEfficiency combinator as well which runs the
12951335 -- producer at the maximal efficiency i.e. the number of workers are chosen
12961336 -- such that the latency is minimum or within a range. Or we can call it
12971337 -- maxWorkerLatency.
12981338 --
1299− let nWorkers = (fromIntegral wLatency) / (fromIntegral expectedLat)
1339+ let
1340+ -- How many workers do we need to acheive the required rate?
1341+ --
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.
1348+ --
1349+ -- When the workers are purely CPU bound, increasing the workers beyond
1350+ -- the number of CPUs won't help.
1351+ --
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.
1358+
13001359 -- Calculate how many yields are we ahead or behind to match the exact
13011360 -- required rate. Based on that we increase or decrease the effective
13021361 -- workers.
1303− d = fromIntegral svarElapsed
1304− e = fromIntegral expectedLat
1305− n = max 1 (fromIntegral coverUpLatency / fromIntegral wLatency)
1306− :: Double
1307− extraYields = (d / e) - (fromIntegral svarYields)
1308− extraWorkers = extraYields / n
1309− requiredWorkers = round $ nWorkers + extraWorkers
1310− in case workerLimit of
1311− Unlimited -> requiredWorkers
1312− Limited x -> min requiredWorkers (fromIntegral x)
1362+ --
1363+ -- When the worker latency is lower than required latency we begin with
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
1367+
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
1372+ then
1373+ let deltaYieldsFreq :: Double
1374+ deltaYieldsFreq =
1375+ fromIntegral deltaYields /
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))
1381+ (maxLatency range)
1382+ in assert (requiredLat > 0) $
1383+ if wLatency <= requiredLat
1384+ then PartialWorker deltaYields
1385+ else ManyWorkers ( fromIntegral
1386+ $ withLimit
1387+ $ wLatency `div` requiredLat) deltaYields
1388+ else
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)
1397+ where
1398+ withLimit n =
1399+ case workerLimit of
1400+ Unlimited -> n
1401+ Limited x -> min n (fromIntegral x)
13131402
13141403-- | Get the worker latency without resetting workerPendingLatency
13151404-- Returns (total yield count, base time, measured latency)
@@ -1341,19 +1430,33 @@ isBeyondMaxRate sv yinfo = do
13411430 (count, tstamp, wLatency) <- getWorkerLatency yinfo
13421431 now <- getTime Monotonic
13431432 let duration = fromInteger $ toNanoSecs $ now - tstamp
1344− let expectedLat = svarExpectedLatency yinfo
1345− let netWorkers = countDispatchWorkers (maxWorkerLimit sv) count duration
1346− wLatency expectedLat
1433+ let targetLat = svarLatencyTarget yinfo
1434+ let work = estimateWorkers (maxWorkerLimit sv) count duration wLatency
1435+ targetLat (svarLatencyRange yinfo)
13471436 cnt <- readIORef $ workerCount sv
1348− if netWorkers <= 0
1349− then do
1350− let sleepTime = ((fromIntegral count) * expectedLat) - duration
1351− if sleepTime >= minThreadDelay
1352− then return True
1353− else return False
1354− else if cnt > netWorkers
1355− then return True
1356− else return False
1437+ return $ case work of
1438+ -- XXX set the worker's maxYields or polling interval based on yields
1439+ PartialWorker _yields -> cnt > 1
1440+ ManyWorkers n _ -> cnt > n
1441+ BlockWait _ -> True
1442+
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.
1448+--
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
1456+ cnt = max 1 $ minThreadDelay `div` latency
1457+ period = min cnt (fromIntegral magicMaxBuffer)
1458+
1459+ writeIORef periodRef (fromIntegral period)
13571460
13581461useCollectedBatch :: YieldRateInfo
13591462 -> NanoSecs
@@ -1361,14 +1464,10 @@ useCollectedBatch :: YieldRateInfo
13611464 -> IORef NanoSecs
13621465 -> IO ()
13631466useCollectedBatch yinfo latency colRef measRef = do
1364− let periodRef = workerLatencyUpdateInterval yinfo
1365− cnt = max 1 $ minThreadDelay `div` latency
1366− period = min cnt (fromIntegral magicMaxBuffer)
1367−
1467+ updateWorkerPollingInterval yinfo latency
1468+ writeIORef colRef (0, 0)
13681469 -- XXX we should use some weight for the old latency as well in arriving at
13691470 -- the new latency, we have the counts, we can use that for some weighting.
1370− writeIORef periodRef (fromIntegral period)
1371− writeIORef colRef (0, 0)
13721471 writeIORef measRef latency
13731472
13741473-- Returns a triple, (1) yield count since last collection, (2) the base time
@@ -1467,51 +1566,76 @@ dispatchWorkerPaced sv = do
14671566 then return False
14681567 else do
14691568 let workerLimit = maxWorkerLimit sv
1470− let expectedLat = svarExpectedLatency yinfo
1471− let netWorkers = countDispatchWorkers
1472− workerLimit svarYields svarElapsed
1473− wLatency expectedLat
1569+ let targetLat = svarLatencyTarget yinfo
1570+ let range = svarLatencyRange yinfo
1571+ let work = estimateWorkers workerLimit svarYields svarElapsed
1572+ wLatency targetLat range
14741573
14751574 -- XXX we need to take yieldLimit into account here. If we are at the
14761575 -- end of the limit as well as the time, we should not be sleeping.
14771576 -- If we are not actually planning to dispatch any more workers we need
14781577 -- to take that in account.
1479− if netWorkers <= 0
1480− then do
1481− let expectedDuration = (fromIntegral svarYields) * expectedLat
1482− let sleepTime = expectedDuration - svarElapsed
1578+ case work of
1579+ BlockWait s -> do
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
1584+ -- result arrives.
1585+ --
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
1595+ return False
1596+ PartialWorker yields -> do
1597+ assert (yields > 0) (return ())
1598+ done <- allThreadsDone sv
1599+ when done $ void $ dispatchWorker yields sv
1600+ return False
1601+ ManyWorkers netWorkers yields -> do
1602+ assert (netWorkers >= 1) (return ())
1603+ assert (yields >= 0) (return ())
1604+
1605+ let periodRef = workerPollingInterval yinfo
1606+ ycnt = max 1 $ yields `div` fromIntegral netWorkers
1607+ period = min ycnt (fromIntegral magicMaxBuffer)
1608+
1609+ old <- liftIO $ readIORef periodRef
1610+ when (period < old) $
1611+ liftIO $ writeIORef periodRef period
1612+
1613+ cnt <- liftIO $ readIORef $ workerCount sv
1614+ if (cnt < netWorkers)
1615+ then do
1616+ let total = netWorkers - cnt
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
1625+ return r
1626+ else return False
14831627
1484− when (sleepTime >= minThreadDelay) $ do
1485− liftIO $ threadDelay $ nanoToMicroSecs sleepTime
1628+ where
14861629
1487− cnt <- liftIO $ readIORef $ workerCount sv
1488− when (cnt < 1) $ void $ do
1489− let n = countDispatchYields (expectedLat - wLatency)
1490− in dispatchWorker (fromIntegral (max n 1)) sv
1491− return False
1630+ dispatchN n = do
1631+ if n == 0
1632+ then return True
14921633 else do
1493− -- XXX stagger the workers over a period
1494− cnt <- liftIO $ readIORef $ workerCount sv
1495− if (cnt < netWorkers)
1496− then dispatchWorker 0 sv
1634+ r <- dispatchWorker 0 sv
1635+ if r
1636+ then dispatchN (n - 1)
14971637 else return False
14981638
1499− where
1500−
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
1507− then
1508− if idleTime == 0
1509− then maxYieldsPerWorker
1510− else
1511− let n = (minThreadDelay + idleTime) `div` idleTime
1512− in min (fromIntegral n) maxYieldsPerWorker
1513− else 1
1514−
15151639sendWorkerDelayPaced :: SVar t m a -> IO ()
15161640sendWorkerDelayPaced _ = return ()
15171641
@@ -1540,7 +1664,7 @@ sendWorkerWait
15401664sendWorkerWait delay dispatch sv = do
15411665 -- Note that we are guaranteed to have at least one outstanding worker when
15421666 -- 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.
15441668
15451669 liftIO $ delay sv
15461670 (_, n) <- liftIO $ readIORef (outputQueue sv)
@@ -1680,8 +1804,10 @@ postProcessBounded sv = do
16801804 if workersDone
16811805 then do
16821806 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.
16841809 when (not r) $ pushWorker 0 sv
1810+ -- XXX do we need to dispatch many here?
16851811 -- void $ dispatchWorker sv
16861812 return r
16871813 else return False
@@ -1699,34 +1825,63 @@ postProcessPaced sv = do
16991825 let yinfo = fromJust $ yieldRateInfo sv
17001826 liftIO $ adjustLatencies yinfo
17011827 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
17021833 return r
17031834 else return False
17041835
17051836getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
17061837getYieldRateInfo st = do
1707− case getMaxStreamRate st of
1708− Just rate -> do
1709− let latency = round $ 1.0e9 / rate
1710− measured <- newIORef 0
1711− wcur <- newIORef (0,0)
1712− wcol <- newIORef (0,0)
1713− now <- getTime Monotonic
1714− wlong <- newIORef (0,now)
1715− period <- newIORef 1
1716− stopTime <- newIORef $ fromNanoSecs 0
1717−
1718− return $ Just YieldRateInfo
1719− { svarExpectedLatency = latency
1720− , workerBootstrapLatency = getStreamLatency st
1721− , workerLatencyUpdateInterval = period
1722− , workerMeasuredLatency = measured
1723− , workerPendingLatency = wcur
1724− , workerStopTimeStamp = stopTime
1725− , workerCollectedLatency = wcol
1726− , svarAllTimeLatency = wlong
1727− }
1838+ let toLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1839+ toMaxLatency r = if r <= 0 then maxBound else r
1840+ case getStreamRate st of
1841+ Just (AvgRate rate) ->
1842+ let l = toLatency rate
1843+ mx = toMaxLatency (l * 2)
1844+ in mkYieldRateInfo l (LatencyRange (l `div` 2) mx)
1845+ Just (MinRate rate) ->
1846+ let l = toLatency rate
1847+ in mkYieldRateInfo l (LatencyRange (l `div` 2) l)
1848+ Just (MaxRate rate) ->
1849+ let l = toLatency rate
1850+ mx = toMaxLatency (l * 2)
1851+ in mkYieldRateInfo l (LatencyRange l mx)
1852+ Just (ConstRate rate) ->
1853+ let l = toLatency rate
1854+ in mkYieldRateInfo l (LatencyRange l l)
1855+ Just (Rate rate minr maxr) ->
1856+ let l = toLatency rate
1857+ minl = toLatency minr
1858+ maxl = toLatency maxr
1859+ in mkYieldRateInfo l (LatencyRange minl maxl)
17281860 Nothing -> return Nothing
17291861
1862+ where
1863+
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
1872+
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
1883+ }
1884+
17301885getAheadSVar :: MonadAsync m
17311886 => State t m a
17321887 -> ( IORef ([t m a], Int)
@@ -1801,7 +1956,7 @@ getAheadSVar st f = do
18011956 }
18021957
18031958 let sv =
1804− case getMaxStreamRate st of
1959+ case getStreamRate st of
18051960 Nothing -> getSVar sv readOutputQBounded postProcessBounded
18061961 Just _ -> getSVar sv readOutputQPaced postProcessPaced
18071962 in return sv
@@ -1909,7 +2064,10 @@ sendFirstWorker sv m = do
19092064 liftIO $ enqueue sv m
19102065 case yieldRateInfo sv of
19112066 Nothing -> pushWorker 0 sv
1912− Just _ -> pushWorker 1 sv
2067+ Just yinfo -> do
2068+ if svarLatencyTarget yinfo == maxBound
2069+ then liftIO $ threadDelay maxBound
2070+ else pushWorker 1 sv
19132071 return sv
19142072
19152073{-# INLINABLE newAheadVar #-}