ebdfa83 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

implement rateBuffer to forget distant gains/losses

ebdfa83Harendra Kumar8 years ago

src/Streamly/SVar.hs+65−23

compare with master

unifiedsplit

@@ -235,16 +235,32 @@ data WorkerInfo = WorkerInfo

235 , workerLatencyStart :: IORef (Count, TimeSpec)

236 }

237

238+

238−-- | Specify the stream yield rate in yields per second (@Hertz@). We try to

239+-- | Specifies the stream yield rate in yields per second (@Hertz@).

239−-- keep the average rate at 'rateGoal', if the actual average rate goes above

240+-- We keep accumulating yield credits at 'rateGoal'. At any point of time we

241+-- allow only as many yields as we have accumulated as per 'rateGoal' since the

242+-- start of time. If the consumer or the producer is slower or faster, the

243+-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap

240−-- or below the goal we try to recover it by increasing or decreasing the

244+-- between the two by increasing or decreasing the pull rate from the producer.

241−-- instantaneous rate but keeping it within 'rateLow' and 'rateHigh' limits.

245+-- However, if the gap becomes more than 'rateBuffer' we try to recover only as

246+-- much as 'rateBuffer'.

247+--

248+-- 'rateLow' puts a bound on how low the instantaneous rate can go when

249+-- recovering the rate gap. In other words, it determines the maximum yield

250+-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous

251+-- rate can go when recovering the rate gap. In other words, it determines the

252+-- minimum yield latency. We reduce the latency by increasing concurrency,

253+-- therefore we can say that it puts an upper bound on concurrency.

254+--

255+-- If the 'rateGoal' is 0 or negative the stream never yields a value.

256+-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.

242--

257--

243-- @since 0.5.0

258-- @since 0.5.0

244data Rate = Rate

259data Rate = Rate

245− { rateLow :: Double -- ^ The lower rate limit

260+ { rateLow :: Double -- ^ The lower rate limit

246− , rateGoal :: Double -- ^ The target rate we want to achieve

261+ , rateGoal :: Double -- ^ The target rate we want to achieve

247− , rateHigh :: Double -- ^ The upper rate limit

262+ , rateHigh :: Double -- ^ The upper rate limit

263+ , rateBuffer :: Int -- ^ Maximum slack from the goal

248 }

264 }

249

265

250data LatencyRange = LatencyRange

266data LatencyRange = LatencyRange

@@ -254,8 +270,10 @@ data LatencyRange = LatencyRange

254

270

255-- Rate control.

271-- Rate control.

256data YieldRateInfo = YieldRateInfo

272data YieldRateInfo = YieldRateInfo

257− { svarLatencyTarget :: NanoSecs

273+ { svarLatencyTarget :: NanoSecs

258− , svarLatencyRange :: LatencyRange

274+ , svarLatencyRange :: LatencyRange

275+ , svarRateBuffer :: Int

276+ , svarGainedLostYields :: IORef Count

259

277

260 -- Actual latency/througput as seen from the consumer side, we count the

278 -- Actual latency/througput as seen from the consumer side, we count the

261 -- yields and the time it took to generates those yields. This is used to

279 -- yields and the time it took to generates those yields. This is used to

@@ -585,22 +603,23 @@ dumpSVarStats sv ss style = do

585 minLat <- readIORef $ minWorkerLatency ss

603 minLat <- readIORef $ minWorkerLatency ss

586 maxLat <- readIORef $ maxWorkerLatency ss

604 maxLat <- readIORef $ maxWorkerLatency ss

587 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss

605 (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss

588− (svarCnt, svarLat) <- case yieldRateInfo sv of

606+ (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of

589− Nothing -> return (0, 0)

607+ Nothing -> return (0, 0, 0)

590 Just yinfo -> do

608 Just yinfo -> do

591 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo

609 (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo

592 if cnt > 0

610 if cnt > 0

593 then do

611 then do

594 t <- readIORef (svarStopTime ss)

612 t <- readIORef (svarStopTime ss)

613+ gl <- readIORef (svarGainedLostYields yinfo)

595 case t of

614 case t of

596 Nothing -> do

615 Nothing -> do

597 now <- getTime Monotonic

616 now <- getTime Monotonic

598 let interval = toNanoSecs (now - startTime)

617 let interval = toNanoSecs (now - startTime)

599− return $ (cnt, interval `div` fromIntegral cnt)

618+ return $ (cnt, gl, interval `div` fromIntegral cnt)

600 Just stopTime -> do

619 Just stopTime -> do

601 let interval = toNanoSecs (stopTime - startTime)

620 let interval = toNanoSecs (stopTime - startTime)

602− return $ (cnt, interval `div` fromIntegral cnt)

621+ return $ (cnt, gl, interval `div` fromIntegral cnt)

603− else return (0, 0)

622+ else return (0, 0, 0)

604

623

605 return $ unlines

624 return $ unlines

606 [ "total dispatches = " ++ show dispatches

625 [ "total dispatches = " ++ show dispatches

@@ -629,6 +648,9 @@ dumpSVarStats sv ss style = do

629 ++ (if svarCnt > 0

648 ++ (if svarCnt > 0

630 then "\nSVar yield count = " ++ show svarCnt

649 then "\nSVar yield count = " ++ show svarCnt

631 else "")

650 else "")

651+ ++ (if svarGainLossCnt > 0

652+ then "\nSVar gain/loss yield count = " ++ show svarGainLossCnt

653+ else "")

632 ]

654 ]

633

655

634{-# NOINLINE dumpSVar #-}

656{-# NOINLINE dumpSVar #-}

@@ -1301,12 +1323,14 @@ data Work

1301estimateWorkers

1323estimateWorkers

1302 :: Limit

1324 :: Limit

1303 -> Count

1325 -> Count

1326+ -> Count

1304 -> NanoSecs

1327 -> NanoSecs

1305 -> NanoSecs

1328 -> NanoSecs

1306 -> NanoSecs

1329 -> NanoSecs

1307 -> LatencyRange

1330 -> LatencyRange

1308 -> Work

1331 -> Work

1332+estimateWorkers workerLimit svarYields gainLossYields

1309−estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =

1333+ svarElapsed wLatency targetLat range =

1310 -- 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

1311 -- 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

1312 -- 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

@@ -1339,7 +1363,8 @@ estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =

1339 -- When the worker latency is lower than required latency we begin with

1363 -- When the worker latency is lower than required latency we begin with

1340 -- a yield and then wait rather than first waiting and then yielding.

1364 -- a yield and then wait rather than first waiting and then yielding.

1341 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat

1365 targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat

1366+ effectiveYields = svarYields + gainLossYields

1342− deltaYields = fromIntegral targetYields - svarYields

1367+ deltaYields = fromIntegral targetYields - effectiveYields

1343

1368

1344 -- We recover the deficit by running at a higher/lower rate for a

1369 -- We recover the deficit by running at a higher/lower rate for a

1345 -- certain amount of time. To keep the effective rate in reasonable

1370 -- certain amount of time. To keep the effective rate in reasonable

@@ -1362,7 +1387,7 @@ estimateWorkers workerLimit svarYields svarElapsed wLatency targetLat range =

1362 $ withLimit

1387 $ withLimit

1363 $ wLatency `div` adjustedLat) deltaYields

1388 $ wLatency `div` adjustedLat) deltaYields

1364 else

1389 else

1365− let expectedDuration = (fromIntegral svarYields) * targetLat

1390+ let expectedDuration = fromIntegral effectiveYields * targetLat

1366 sleepTime = expectedDuration - svarElapsed

1391 sleepTime = expectedDuration - svarElapsed

1367 maxSleepTime = maxLatency range - wLatency

1392 maxSleepTime = maxLatency range - wLatency

1368 s = min sleepTime maxSleepTime

1393 s = min sleepTime maxSleepTime

@@ -1407,8 +1432,9 @@ isBeyondMaxRate sv yinfo = do

1407 now <- getTime Monotonic

1432 now <- getTime Monotonic

1408 let duration = fromInteger $ toNanoSecs $ now - tstamp

1433 let duration = fromInteger $ toNanoSecs $ now - tstamp

1409 let targetLat = svarLatencyTarget yinfo

1434 let targetLat = svarLatencyTarget yinfo

1435+ gainLoss <- readIORef (svarGainedLostYields yinfo)

1410− let work = estimateWorkers (maxWorkerLimit sv) count duration wLatency

1436+ let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration

1411− targetLat (svarLatencyRange yinfo)

1437+ wLatency targetLat (svarLatencyRange yinfo)

1412 cnt <- readIORef $ workerCount sv

1438 cnt <- readIORef $ workerCount sv

1413 return $ case work of

1439 return $ case work of

1414 -- XXX set the worker's maxYields or polling interval based on yields

1440 -- XXX set the worker's maxYields or polling interval based on yields

@@ -1536,7 +1562,8 @@ dispatchWorkerPaced sv = do

1536 let workerLimit = maxWorkerLimit sv

1562 let workerLimit = maxWorkerLimit sv

1537 let targetLat = svarLatencyTarget yinfo

1563 let targetLat = svarLatencyTarget yinfo

1538 let range = svarLatencyRange yinfo

1564 let range = svarLatencyRange yinfo

1565+ gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)

1539− let work = estimateWorkers workerLimit svarYields svarElapsed

1566+ let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed

1540 wLatency targetLat range

1567 wLatency targetLat range

1541

1568

1542 -- XXX we need to take yieldLimit into account here. If we are at the

1569 -- XXX we need to take yieldLimit into account here. If we are at the

@@ -1563,12 +1590,15 @@ dispatchWorkerPaced sv = do

1563 return False

1590 return False

1564 PartialWorker yields -> do

1591 PartialWorker yields -> do

1565 assert (yields > 0) (return ())

1592 assert (yields > 0) (return ())

1593+ updateGainedLostYields yinfo yields

1594+

1566 done <- allThreadsDone sv

1595 done <- allThreadsDone sv

1567 when done $ void $ dispatchWorker yields sv

1596 when done $ void $ dispatchWorker yields sv

1568 return False

1597 return False

1569 ManyWorkers netWorkers yields -> do

1598 ManyWorkers netWorkers yields -> do

1570 assert (netWorkers >= 1) (return ())

1599 assert (netWorkers >= 1) (return ())

1571 assert (yields >= 0) (return ())

1600 assert (yields >= 0) (return ())

1601+ updateGainedLostYields yinfo yields

1572

1602

1573 let periodRef = workerPollingInterval yinfo

1603 let periodRef = workerPollingInterval yinfo

1574 ycnt = max 1 $ yields `div` fromIntegral netWorkers

1604 ycnt = max 1 $ yields `div` fromIntegral netWorkers

@@ -1595,6 +1625,15 @@ dispatchWorkerPaced sv = do

1595

1625

1596 where

1626 where

1597

1627

1628+ updateGainedLostYields yinfo yields = do

1629+ let buf = fromIntegral $ svarRateBuffer yinfo

1630+ when (yields /= 0 && abs yields > buf) $ do

1631+ let delta =

1632+ if yields > 0

1633+ then yields - buf

1634+ else yields + buf

1635+ liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)

1636+

1598 dispatchN n = do

1637 dispatchN n = do

1599 if n == 0

1638 if n == 0

1600 then return True

1639 then return True

@@ -1806,16 +1845,16 @@ getYieldRateInfo st = do

1806 -- convert rate in Hertz to latency in Nanoseconds

1845 -- convert rate in Hertz to latency in Nanoseconds

1807 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r

1846 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r

1808 case getStreamRate st of

1847 case getStreamRate st of

1809− Just (Rate low goal high) ->

1848+ Just (Rate low goal high buf) ->

1810 let l = rateToLatency goal

1849 let l = rateToLatency goal

1811 minl = rateToLatency high

1850 minl = rateToLatency high

1812 maxl = rateToLatency low

1851 maxl = rateToLatency low

1813− in mkYieldRateInfo l (LatencyRange minl maxl)

1852+ in mkYieldRateInfo l (LatencyRange minl maxl) buf

1814 Nothing -> return Nothing

1853 Nothing -> return Nothing

1815

1854

1816 where

1855 where

1817

1856

1818− mkYieldRateInfo latency latRange = do

1857+ mkYieldRateInfo latency latRange buf = do

1819 measured <- newIORef 0

1858 measured <- newIORef 0

1820 wcur <- newIORef (0,0)

1859 wcur <- newIORef (0,0)

1821 wcol <- newIORef (0,0)

1860 wcol <- newIORef (0,0)

@@ -1823,10 +1862,13 @@ getYieldRateInfo st = do

1823 wlong <- newIORef (0,now)

1862 wlong <- newIORef (0,now)

1824 period <- newIORef 1

1863 period <- newIORef 1

1825 stopTime <- newIORef $ fromNanoSecs 0

1864 stopTime <- newIORef $ fromNanoSecs 0

1865+ gainLoss <- newIORef (Count 0)

1826

1866

1827 return $ Just YieldRateInfo

1867 return $ Just YieldRateInfo

1828 { svarLatencyTarget = latency

1868 { svarLatencyTarget = latency

1829 , svarLatencyRange = latRange

1869 , svarLatencyRange = latRange

1870+ , svarRateBuffer = buf

1871+ , svarGainedLostYields = gainLoss

1830 , workerBootstrapLatency = getStreamLatency st

1872 , workerBootstrapLatency = getStreamLatency st

1831 , workerPollingInterval = period

1873 , workerPollingInterval = period

1832 , workerMeasuredLatency = measured

1874 , workerMeasuredLatency = measured

src/Streamly/Streams/SVar.hs+14−16

@@ -188,11 +188,9 @@ maxBufferSerial :: Int -> SerialT m a -> SerialT m a

188maxBufferSerial _ = id

189-}

190

191−-- | Specify the pull rate of a stream in number of yields per second

191+-- | Specify the pull rate of a stream.

192−-- (i.e. @Hertz@). A 'Nothing' value resets the rate to default which is

192+-- A 'Nothing' value resets the rate to default which is unlimited. When the

193−-- unlimited. If the specified rate is 0 or negative the stream never yields a

194−-- value.

195−-- When the rate is specified, concurrent production may be ramped up or down

193+-- rate is specified, concurrent production may be ramped up or down

196-- automatically to achieve the specified yield rate. The specific behavior for

194-- automatically to achieve the specified yield rate. The specific behavior for

197-- different styles of 'Rate' specifications is documented under 'Rate'. The

195-- different styles of 'Rate' specifications is documented under 'Rate'. The

198-- effective maximum production rate achieved by a stream is governed by:

196-- effective maximum production rate achieved by a stream is governed by:

@@ -207,11 +205,11 @@ maxBufferSerial _ = id

207rate :: IsStream t => Maybe Rate -> t m a -> t m a

205rate :: IsStream t => Maybe Rate -> t m a -> t m a

208rate r m = fromStream $ Stream $ \st stp sng yld -> do

206rate r m = fromStream $ Stream $ \st stp sng yld -> do

209 case r of

207 case r of

210− Just (Rate low goal _) | goal < low ->

208+ Just (Rate low goal _ _) | goal < low ->

211 error "rate: Target rate cannot be lower than minimum rate."

209 error "rate: Target rate cannot be lower than minimum rate."

212− Just (Rate _ goal high) | goal > high ->

210+ Just (Rate _ goal high _) | goal > high ->

213 error "rate: Target rate cannot be greater than maximum rate."

211 error "rate: Target rate cannot be greater than maximum rate."

214− Just (Rate low _ high) | low > high ->

212+ Just (Rate low _ high _) | low > high ->

215 error "rate: Minimum rate cannot be greater than maximum rate."

213 error "rate: Minimum rate cannot be greater than maximum rate."

216 _ -> unStream (toStream m) (setStreamRate r st) stp sng yld

214 _ -> unStream (toStream m) (setStreamRate r st) stp sng yld

217

215

@@ -221,7 +219,7 @@ yieldRateSerial :: Double -> SerialT m a -> SerialT m a

221yieldRateSerial _ = id

219yieldRateSerial _ = id

222-}

220-}

223

221

224−-- | Same as @rate (Just $ Rate (r/2) r (2*r))@

222+-- | Same as @rate (Just $ Rate (r/2) r (2*r) maxBound)@

225--

223--

226-- Specifies the average production rate of a stream in number of yields

224-- Specifies the average production rate of a stream in number of yields

227-- per second (i.e. @Hertz@). Concurrent production is ramped up or down

225-- per second (i.e. @Hertz@). Concurrent production is ramped up or down

@@ -231,9 +229,9 @@ yieldRateSerial _ = id

231--

229--

232-- @since 0.5.0

230-- @since 0.5.0

233avgRate :: IsStream t => Double -> t m a -> t m a

231avgRate :: IsStream t => Double -> t m a -> t m a

234−avgRate r = rate (Just $ Rate (r/2) r (2*r))

232+avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)

235

233

236−-- | Same as @rate (Just $ Rate r r (2*r))@

234+-- | Same as @rate (Just $ Rate r r (2*r) maxBound)@

237--

235--

238-- Specifies the minimum rate at which the stream should yield values. As

236-- Specifies the minimum rate at which the stream should yield values. As

239-- far as possible the yield rate would never be allowed to go below the

237-- far as possible the yield rate would never be allowed to go below the

@@ -242,9 +240,9 @@ avgRate r = rate (Just $ Rate (r/2) r (2*r))

242--

240--

243-- @since 0.5.0

241-- @since 0.5.0

244minRate :: IsStream t => Double -> t m a -> t m a

242minRate :: IsStream t => Double -> t m a -> t m a

245−minRate r = rate (Just $ Rate r r (2*r))

243+minRate r = rate (Just $ Rate r r (2*r) maxBound)

246

244

247−-- | Same as @rate (Just $ Rate (r/2) r r)@

245+-- | Same as @rate (Just $ Rate (r/2) r r maxBound)@

248--

246--

249-- Specifies the maximum rate at which the stream should yield values. As

247-- Specifies the maximum rate at which the stream should yield values. As

250-- far as possible the yield rate would never be allowed to go above the

248-- far as possible the yield rate would never be allowed to go above the

@@ -255,9 +253,9 @@ minRate r = rate (Just $ Rate r r (2*r))

255--

253--

256-- @since 0.5.0

254-- @since 0.5.0

257maxRate :: IsStream t => Double -> t m a -> t m a

255maxRate :: IsStream t => Double -> t m a -> t m a

258−maxRate r = rate (Just $ Rate (r/2) r r)

256+maxRate r = rate (Just $ Rate (r/2) r r maxBound)

259

257

260−-- | Same as @rate (Just $ Rate r r r)@

258+-- | Same as @rate (Just $ Rate r r r 0)@

261--

259--

262-- Specifies a constant yield rate. If for some reason the actual rate

260-- Specifies a constant yield rate. If for some reason the actual rate

263-- goes above or below the specified rate we do not try to recover it by

261-- goes above or below the specified rate we do not try to recover it by

@@ -267,7 +265,7 @@ maxRate r = rate (Just $ Rate (r/2) r r)

267--

265--

268-- @since 0.5.0

266-- @since 0.5.0

269constRate :: IsStream t => Double -> t m a -> t m a

267constRate :: IsStream t => Double -> t m a -> t m a

270−constRate r = rate (Just $ Rate r r r)

268+constRate r = rate (Just $ Rate r r r 0)

271

269

272-- | Specify the average latency, in nanoseconds, of a single threaded action

270-- | Specify the average latency, in nanoseconds, of a single threaded action

273-- in a concurrent composition. Streamly can measure the latencies, but that is

271-- in a concurrent composition. Streamly can measure the latencies, but that is

test/MaxRate.hs+3

@@ -115,8 +115,11 @@ main = hspec $ do

115 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 1 range)

116

117 describe "asyncly with 1 sec producer delay and some consumer delay" $ do

118+ -- ideally it should take 10 x 1 + 1 seconds

118 forM_ [1] (\r -> measureRate "asyncly" asyncly r 1 1 (11, 16))

119 forM_ [1] (\r -> measureRate "asyncly" asyncly r 1 1 (11, 16))

120+ -- ideally it should take 10 x 2 + 1 seconds

119 forM_ [1] (\r -> measureRate "asyncly" asyncly r 2 1 (21, 23))

121 forM_ [1] (\r -> measureRate "asyncly" asyncly r 2 1 (21, 23))

122+ -- ideally it should take 10 x 3 + 1 seconds

120 forM_ [1] (\r -> measureRate "asyncly" asyncly r 3 1 (31, 33))

123 forM_ [1] (\r -> measureRate "asyncly" asyncly r 3 1 (31, 33))

121

124

122 describe "aheadly with 1 sec producer delay and some consumer delay" $ do

125 describe "aheadly with 1 sec producer delay and some consumer delay" $ do

test/rate-timestamp.hs+1−1

@@ -14,7 +14,7 @@ withTimeStamp msg = do

14-- acidRain :: MonadAsync m => SerialT m Event

15producer =

16 asyncly

17− $ rate (Just $ AvgRate 1)

17+ $ avgRate 1

18 $ S.repeatM

19 $ liftIO $ do

20 withTimeStamp "produced"