e2b1e4c · streamly · git.iko.soy

git.iko.soy/streamly

Fix worker dispatch

commite2b1e4c45d4cbdf22b08b42e4166f6b1d9adaf8a authorHarendra Kumar <harendra.kumar@gmail.com> dateTue 5 Sep 2017, 10:50 (9 years ago) parented5f00d

2 files changed+28−13

compare with master

unifiedsplit

src/Asyncly/AsyncT.hs+11−9 test/Main.hs+17−4

src/Asyncly/AsyncT.hs+11−9

@@ -298,15 +298,17 @@ pushWorker ctx =

298298

299299{-# INLINE sendWorkerWait #-}

300300sendWorkerWait :: MonadAsync m => Context m a -> m ()

301−sendWorkerWait ctx = do

302− liftIO $ threadDelay 200

303− output <- liftIO $ readIORef (outputQueue ctx)

304− when (null output) $ do

305− -- Send a worker only if workqueue is not empty otherwise we may keep

306− -- sending a worker in a loop running into a livelock

307− done <- queueEmpty ctx

308− when (not done) (pushWorker ctx)

309− void $ liftIO $ takeMVar (doorBell ctx)

301+sendWorkerWait ctx = dispatch >> void (liftIO $ takeMVar (doorBell ctx))

302+

303+ where

304+

305+ dispatch = do

306+ liftIO $ threadDelay 200

307+ output <- liftIO $ readIORef (outputQueue ctx)

308+ when (null output) $ do

309+ done <- queueEmpty ctx

310+ when (not done) $ (pushWorker ctx) >> dispatch

311+

310312

311313-- Note: This is performance sensitive code.

312314{-# NOINLINE pullWorker #-}

test/Main.hs+17−4

@@ -2,9 +2,8 @@ module Main (main) where

33import Control.Applicative ((<|>), empty)

44import Control.Monad (mzero)

5−-- import Control.Concurrent (myThreadId, threadDelay)

65import Control.Monad.IO.Class (liftIO)

7−-- import Data.List (nub)

6+import Control.Concurrent (threadDelay)

87import Data.List (sort)

98import Data.Monoid ((<>))

109import Prelude hiding (take, drop)

@@ -34,13 +33,15 @@ main = hspec $ do

3433

3534 describe "Serial Composition (<>)" $ compose (<>) id

3635 describe "Serial interleaved (<=>)" $ interleaved (<=>)

37− describe "Left Biased parallel Composition (<|)" $ compose (<|) sort

36+ describe "Left biased parallel Composition (<|)" $ compose (<|) sort

3837 describe "Fair parallel Composition (<|>)" $ compose (<|>) sort

38+ describe "Left biased parallel time order check" $ parallelCheck (<|)

39+ describe "Fair parallel time order check" $ parallelCheck (<|>)

3940 -- This is not predicatable

4041 -- describe "Parallel interleaved (<|>)" $ interleaved (<|>)

4142

4243 describe "Serial loops (<>)" $ loops (<>) id reverse

43− describe "Left Biased parallel loops (<|)" $ loops (<|) sort sort

44+ describe "Left biased parallel loops (<|)" $ loops (<|) sort sort

4445 describe "Fair parallel loops (<|>)" $ loops (<|>) sort sort

4546

4647 describe "Transformation" $ transformOps (<>)

@@ -63,6 +64,18 @@ interleaved f =

6364 toList ((return 0 <> return 1) `f` (return 100 <> return 101))

6465 `shouldReturn` ([0, 100, 1, 101])

6566

67+parallelCheck :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec

68+parallelCheck f = do

69+ it "Parallel ordering left associated" $

70+ toList (((event 4 `f` event 3) `f` event 2) `f` event 1)

71+ `shouldReturn` ([1..4])

72+

73+ it "Parallel ordering right associated" $

74+ toList (event 4 `f` (event 3 `f` (event 2 `f` event 1)))

75+ `shouldReturn` ([1..4])

76+

77+ where event n = (liftIO $ threadDelay (n * 100000)) >> (return n)

78+

6679compose

6780 :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

6881 -> ([Int] -> [Int])