cca869f · streamly · git.iko.soy

git.iko.soy/streamly

Fix a rate control issue

We were not sending enough workers in the end when the streaming is finishing. Also, limit the estimated workers to the number of yields required.

commitcca869f7ed98fbe513cff1e67635917f9e16e113 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 12 Sep 2018, 11:13 (8 years ago) parente583f56

4 files changed+37−32

compare with master

unifiedsplit

src/Streamly/SVar.hs+17−14 src/Streamly/Streams/Async.hs+4−4 test/Main.hs+12−11 test/MaxRate.hs+4−3

src/Streamly/SVar.hs+17−14

@@ -346,7 +346,7 @@ data SVar t m a = SVar

346346 -- Combined/aggregate parameters

347347 , maxWorkerLimit :: Limit

348348 , maxBufferLimit :: Limit

349− , remainingYields :: Maybe (IORef Count)

349+ , remainingWork :: Maybe (IORef Count)

350350 , yieldRateInfo :: Maybe YieldRateInfo

351351

352352 -- Used only by bounded SVar types

@@ -779,7 +779,7 @@ doFork action exHandler =

779779 exHandler

780780 runInIO (return tid)

781781

782−-- XXX Can we make access to remainingYields and yieldRateInfo fields in sv

782+-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv

783783-- faster, along with the fields in sv required by send?

784784-- XXX make it noinline

785785--

@@ -793,7 +793,7 @@ doFork action exHandler =

793793{-# INLINE decrementYieldLimit #-}

794794decrementYieldLimit :: SVar t m a -> IO Bool

795795decrementYieldLimit sv =

796− case remainingYields sv of

796+ case remainingWork sv of

797797 Nothing -> return True

798798 Just ref -> do

799799 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

@@ -804,7 +804,7 @@ decrementYieldLimit sv =

804804{-# INLINE decrementYieldLimitPost #-}

805805decrementYieldLimitPost :: SVar t m a -> IO Bool

806806decrementYieldLimitPost sv =

807− case remainingYields sv of

807+ case remainingWork sv of

808808 Nothing -> return True

809809 Just ref -> do

810810 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

@@ -813,7 +813,7 @@ decrementYieldLimitPost sv =

813813{-# INLINE incrementYieldLimit #-}

814814incrementYieldLimit :: SVar t m a -> IO ()

815815incrementYieldLimit sv =

816− case remainingYields sv of

816+ case remainingWork sv of

817817 Nothing -> return ()

818818 Just ref -> atomicModifyIORefCAS_ ref (+ 1)

819819

@@ -1339,6 +1339,9 @@ dispatchWorker yieldCount sv = do

13391339 -- XXX in case of Ahead streams we should not send more than one worker

13401340 -- when the work queue is done but heap is not done.

13411341 done <- liftIO $ isWorkDone sv

1342+ -- Note, "done" may not mean that the work is actually finished if there

1343+ -- are workers active, because there may be a worker which has not yet

1344+ -- queued the leftover work.

13421345 if (not done)

13431346 then do

13441347 qDone <- liftIO $ isQueueDone sv

@@ -1353,7 +1356,7 @@ dispatchWorker yieldCount sv = do

13531356 -- executing. In that case we should either configure the maxWorker

13541357 -- count to higher or use parallel style instead of ahead or async

13551358 -- style.

1356− limit <- case remainingYields sv of

1359+ limit <- case remainingWork sv of

13571360 Nothing -> return workerLimit

13581361 Just ref -> do

13591362 n <- liftIO $ readIORef ref

@@ -1368,12 +1371,12 @@ dispatchWorker yieldCount sv = do

13681371 let dispatch = pushWorker yieldCount sv >> return True

13691372 in case limit of

13701373 Unlimited -> dispatch

1371− -- Note that the use of remainingYields and workerCount is not

1374+ -- Note that the use of remainingWork and workerCount is not

13721375 -- atomic and the counts may even have changed between reading and

13731376 -- using them here, so this is just approximate logic and we cannot

13741377 -- rely on it for correctness. We may actually dispatch more

13751378 -- workers than required.

1376− Limited lim | active < (fromIntegral lim) -> dispatch

1379+ Limited lim | lim > 0 -> dispatch

13771380 _ -> return False

13781381 else do

13791382 when (active <= 0) $ pushWorker 0 sv

@@ -1474,9 +1477,9 @@ estimateWorkers workerLimit svarYields gainLossYields

14741477 in assert (adjustedLat > 0) $

14751478 if wLatency <= adjustedLat

14761479 then PartialWorker deltaYields

1477− else ManyWorkers ( fromIntegral

1478− $ withLimit

1479− $ wLatency `div` adjustedLat) deltaYields

1480+ else let workers = withLimit $ wLatency `div` adjustedLat

1481+ limited = min workers (fromIntegral deltaYields)

1482+ in ManyWorkers (fromIntegral limited) deltaYields

14801483 else

14811484 let expectedDuration = fromIntegral effectiveYields * targetLat

14821485 sleepTime = expectedDuration - svarElapsed

@@ -2001,7 +2004,7 @@ getAheadSVar st f = do

20012004

20022005 let getSVar sv readOutput postProc = SVar

20032006 { outputQueue = outQ

2004− , remainingYields = yl

2007+ , remainingWork = yl

20052008 , maxBufferLimit = getMaxBuffer st

20062009 , maxWorkerLimit = getMaxThreads st

20072010 , yieldRateInfo = rateInfo

@@ -2049,7 +2052,7 @@ getAheadSVar st f = do

20492052 isQueueDoneAhead sv q = do

20502053 queueDone <- checkEmpty q

20512054 yieldsDone <-

2052− case remainingYields sv of

2055+ case remainingWork sv of

20532056 Just yref -> do

20542057 n <- readIORef yref

20552058 return (n <= 0)

@@ -2097,7 +2100,7 @@ getParallelSVar st = do

20972100

20982101 let sv =

20992102 SVar { outputQueue = outQ

2100− , remainingYields = yl

2103+ , remainingWork = yl

21012104 , maxBufferLimit = Unlimited

21022105 , maxWorkerLimit = Unlimited

21032106 -- Used only for diagnostics

src/Streamly/Streams/Async.hs+4−4

@@ -283,7 +283,7 @@ getLifoSVar st = do

283283

284284 let isWorkFinishedLimited sv = do

285285 yieldsDone <-

286− case remainingYields sv of

286+ case remainingWork sv of

287287 Just ref -> do

288288 n <- readIORef ref

289289 return (n <= 0)

@@ -293,7 +293,7 @@ getLifoSVar st = do

293293

294294 let getSVar sv readOutput postProc workDone wloop = SVar

295295 { outputQueue = outQ

296− , remainingYields = yl

296+ , remainingWork = yl

297297 , maxBufferLimit = getMaxBuffer st

298298 , maxWorkerLimit = getMaxThreads st

299299 , yieldRateInfo = rateInfo

@@ -382,7 +382,7 @@ getFifoSVar st = do

382382 let isWorkFinished _ = nullQ q

383383 let isWorkFinishedLimited sv = do

384384 yieldsDone <-

385− case remainingYields sv of

385+ case remainingWork sv of

386386 Just ref -> do

387387 n <- readIORef ref

388388 return (n <= 0)

@@ -392,7 +392,7 @@ getFifoSVar st = do

392392

393393 let getSVar sv readOutput postProc workDone wloop = SVar

394394 { outputQueue = outQ

395− , remainingYields = yl

395+ , remainingWork = yl

396396 , maxBufferLimit = getMaxBuffer st

397397 , maxWorkerLimit = getMaxThreads st

398398 , yieldRateInfo = rateInfo

test/Main.hs+12−11

@@ -84,19 +84,19 @@ main = hspec $ do

8484 `shouldReturn` ([0,0,2,2,4,4,8,8])

8585

8686 describe "restricts concurrency and cleans up extra tasks" $ do

87− it "take 1 asyncly" $ checkCleanup asyncly (S.take 1)

88− it "take 1 wAsyncly" $ checkCleanup wAsyncly (S.take 1)

89− it "take 1 aheadly" $ checkCleanup aheadly (S.take 1)

87+ it "take 1 asyncly" $ checkCleanup 2 asyncly (S.take 1)

88+ it "take 1 wAsyncly" $ checkCleanup 2 wAsyncly (S.take 1)

89+ it "take 1 aheadly" $ checkCleanup 2 aheadly (S.take 1)

9090

91− it "takeWhile (< 0) asyncly" $ checkCleanup asyncly (S.takeWhile (< 0))

92− it "takeWhile (< 0) wAsyncly" $ checkCleanup wAsyncly (S.takeWhile (< 0))

93− it "takeWhile (< 0) aheadly" $ checkCleanup aheadly (S.takeWhile (< 0))

91+ it "takeWhile (< 0) asyncly" $ checkCleanup 2 asyncly (S.takeWhile (< 0))

92+ it "takeWhile (< 0) wAsyncly" $ checkCleanup 2 wAsyncly (S.takeWhile (< 0))

93+ it "takeWhile (< 0) aheadly" $ checkCleanup 2 aheadly (S.takeWhile (< 0))

9494

9595#ifdef DEVBUILD

9696 -- parallely fails on CI machines, may need more difference in times of

9797 -- the events, but that would make tests even slower.

98− it "take 1 parallely" $ checkCleanup parallely (S.take 1)

99− it "takeWhile (< 0) parallely" $ checkCleanup parallely (S.takeWhile (< 0))

98+ it "take 1 parallely" $ checkCleanup 3 parallely (S.take 1)

99+ it "takeWhile (< 0) parallely" $ checkCleanup 3 parallely (S.takeWhile (< 0))

100100

101101 testFoldOpsCleanup "head" S.head

102102 testFoldOpsCleanup "null" S.null

@@ -138,10 +138,11 @@ main = hspec $ do

138138 describe "Parallel mappend time order check" $ parallelCheck parallely mappend

139139

140140checkCleanup :: IsStream t

141− => (t IO Int -> SerialT IO Int)

141+ => Int

142+ -> (t IO Int -> SerialT IO Int)

142143 -> (t IO Int -> t IO Int)

143144 -> IO ()

144−checkCleanup t op = do

145+checkCleanup d t op = do

145146 r <- newIORef (-1 :: Int)

146147 runStream . serially $ do

147148 _ <- t $ op $ delay r 0 S.|: delay r 1 S.|: delay r 2 S.|: S.nil

@@ -151,7 +152,7 @@ checkCleanup t op = do

151152 res <- readIORef r

152153 res `shouldBe` 0

153154 where

154− delay ref i = threadDelay (i*200000) >> writeIORef ref i >> return i

155+ delay ref i = threadDelay (i*d*100000) >> writeIORef ref i >> return i

155156

156157#ifdef DEVBUILD

157158checkCleanupFold :: IsStream t

test/MaxRate.hs+4−3

@@ -105,12 +105,13 @@ main = hspec $ do

105105 in describe "wAsyncly no consumer delay and 1 sec producer delay" $ do

106106 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 1 range)

107107

108− -- XXX does not work well at a million ops per second, need to fix.

109− let rates = [1, 10, 100, 1000, 10000, 100000]

108+ let rates = [1, 10, 100, 1000, 10000, 100000, 1000000]

110109 in describe "aheadly no consumer delay no producer delay" $ do

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

112111

113− let rates = [1, 10, 100, 1000, 10000, 25000]

112+ -- XXX after the change to stop workers when the heap is clearing

113+ -- thi does not work well at a 25000 ops per second, need to fix.

114+ let rates = [1, 10, 100, 1000, 10000, 12500]

114115 in describe "aheadly no consumer delay and 1 sec producer delay" $ do

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

116117