e2b1e4c · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix worker dispatch

e2b1e4cHarendra Kumar9 years ago

src/Asyncly/AsyncT.hs+11−9

compare with master

unifiedsplit

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

298

299{-# INLINE sendWorkerWait #-}

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

301−sendWorkerWait ctx = do

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

302+

303+ where

304+

305+ dispatch = do

302− liftIO $ threadDelay 200

306+ liftIO $ threadDelay 200

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

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

304− when (null output) $ do

308+ 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

309+ done <- queueEmpty ctx

308− when (not done) (pushWorker ctx)

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

309− void $ liftIO $ takeMVar (doorBell ctx)

311+

310

312

311-- Note: This is performance sensitive code.

313-- Note: This is performance sensitive code.

312{-# NOINLINE pullWorker #-}

314{-# NOINLINE pullWorker #-}

test/Main.hs+17−4

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

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

4import Control.Monad (mzero)

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

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

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

7−-- import Data.List (nub)

6+import Control.Concurrent (threadDelay)

8import Data.List (sort)

7import Data.List (sort)

9import Data.Monoid ((<>))

8import Data.Monoid ((<>))

10import Prelude hiding (take, drop)

9import Prelude hiding (take, drop)

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

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

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

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

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

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

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

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

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

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

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

39 -- This is not predicatable

40 -- This is not predicatable

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

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

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

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

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

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

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

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

46 describe "Transformation" $ transformOps (<>)

47 describe "Transformation" $ transformOps (<>)

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

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

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

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

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

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+

66compose

79compose

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

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

68 -> ([Int] -> [Int])

81 -> ([Int] -> [Int])