3f5c79e · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix scheduling of WAsync style stream

3f5c79eHarendra Kumar7 years ago

Changelog.md+1

compare with master

unifiedsplit

@@ -11,6 +11,7 @@

11 stream, especially when built with `-threaded` and used with `-N` RTS option.

12 The issue occurs only in cases when a worker thread happens to be used

13 continuously for a long time.

14+* Fix scheduling of WAsyncT stream style to be in round-robin fashion.

15### Behavior change

16### Behavior change

src/Streamly/Internal/Data/Stream/Async.hs+8−7

@@ -225,13 +225,14 @@ workLoopFIFO q st sv winfo = run

225 res <- liftIO $ sendYield sv winfo (ChildYield a)

226 return $ if res then Continue else Suspend

227

228+ -- XXX in general we would like to yield "n" elements from a single stream

229+ -- before moving on to the next. Single element granularity could be too

230+ -- expensive in certain cases. Similarly, we can use time limit for

231+ -- yielding.

228 yieldk a r = do

232 yieldk a r = do

229 res <- liftIO $ sendYield sv winfo (ChildYield a)

233 res <- liftIO $ sendYield sv winfo (ChildYield a)

230− if res

234+ liftIO $ enqueueFIFO sv q r

231− then foldStreamShared st yieldk single (return Continue) r

235+ return $ if res then Continue else Suspend

232− else liftIO $ do

233− enqueueFIFO sv q r

234− return Suspend

235

236

236{-# INLINE workLoopFIFOLimited #-}

237{-# INLINE workLoopFIFOLimited #-}

237workLoopFIFOLimited

238workLoopFIFOLimited

@@ -273,12 +274,12 @@ workLoopFIFOLimited q st sv winfo = run

273

274

274 yieldk a r = do

275 yieldk a r = do

275 res <- liftIO $ sendYield sv winfo (ChildYield a)

276 res <- liftIO $ sendYield sv winfo (ChildYield a)

277+ liftIO $ enqueueFIFO sv q r

276 yieldLimitOk <- liftIO $ decrementYieldLimit sv

278 yieldLimitOk <- liftIO $ decrementYieldLimit sv

277 if res && yieldLimitOk

279 if res && yieldLimitOk

278− then foldStreamShared st yieldk single incrContinue r

280+ then return Continue

279 else liftIO $ do

281 else liftIO $ do

280 incrementYieldLimit sv

282 incrementYieldLimit sv

281− enqueueFIFO sv q r

282 return Suspend

283 return Suspend

283

284

284-------------------------------------------------------------------------------

285-------------------------------------------------------------------------------