git.iko.soy/streamly
on master help sign in
Fix append and interleave nested inside parallel
commit5919be5b368ff0ac408c185e9025c3b39c639fe3 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 20 Sep 2017, 21:55 (9 years ago) parent2e10f14
2 files changed+24−16
compare with master
unifiedsplit
src/Asyncly/AsyncT.hs+10−7 test/Main.hs+14−9
src/Asyncly/AsyncT.hs+10−7
@@ -131,11 +131,11 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
131131instance Semigroup (AsyncT m a) where
132132 m1 <> m2 = go m1
133133 where
134− go (AsyncT m) = AsyncT $ \ctx stp yld ->
135− let stop = (runAsyncT m2) ctx stp yld
134+ go (AsyncT m) = AsyncT $ \_ stp yld ->
135+ let stop = (runAsyncT m2) Nothing stp yld
136136 yield a Nothing = yld a (Just m2)
137137 yield a (Just r) = yld a (Just (go r))
138− in m ctx stop yield
138+ in m Nothing stop yield
139139
140140instance Monoid (AsyncT m a) where
141141 mempty = AsyncT $ \_ stp _ -> stp
@@ -143,11 +143,11 @@ instance Monoid (AsyncT m a) where
143143
144144-- | Same as '<=>'.
145145interleave :: AsyncT m a -> AsyncT m a -> AsyncT m a
146−interleave m1 m2 = AsyncT $ \ctx stp yld -> do
147− let stop = (runAsyncT m2) ctx stp yld
146+interleave m1 m2 = AsyncT $ \_ stp yld -> do
147+ let stop = (runAsyncT m2) Nothing stp yld
148148 yield a Nothing = yld a (Just m2)
149149 yield a (Just r) = yld a (Just (interleave m2 r))
150− (runAsyncT m1) ctx stop yield
150+ (runAsyncT m1) Nothing stop yield
151151
152152infixr 5 <=>
153153
@@ -374,7 +374,10 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
374374 res <- liftIO $ tryTakeMVar (doorBell ctx)
375375 when (isNothing res) $ sendWorkerWait ctx
376376 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
377− (runAsyncT $ processEvents pctx list) Nothing stp yld
377+ -- To avoid lock overhead we read all events at once instead of reading one
378+ -- at a time. We just reverse the list to process the events in the order
379+ -- they arrived. Maybe we can use a queue instead?
380+ (runAsyncT $ processEvents pctx (reverse list)) Nothing stp yld
378381
379382 where
380383
test/Main.hs+14−9
@@ -48,8 +48,8 @@ main = hspec $ do
4848 -- Monoidal Composition ordering checks
4949 ---------------------------------------------------------------------------
5050
51− describe "Serial interleaved (<=>)" $ interleaved (<=>) id
52− describe "Parallel interleaved (<|>)" $ interleaved (<|>) reverse
51+ describe "Serial interleaved (<=>)" $ interleaved (<=>)
52+ describe "Parallel interleaved (<|>)" $ interleaved (<|>)
5353 describe "Left biased parallel time order check" $ parallelCheck (<|)
5454 describe "Fair parallel time order check" $ parallelCheck (<|>)
5555
@@ -104,14 +104,20 @@ main = hspec $ do
104104 let t = timed
105105 in toList (
106106 ((t 8 <|> t 4) <=> (t 2 <|> t 0))
107− <|> ((t 8 <|> t 4) <=> (t 2 <|> t 0)))
108− `shouldReturn` ([4,4,0,0,8,8,2,2])
107+ <|> ((t 9 <|> t 4) <=> (t 2 <|> t 0)))
108+ `shouldReturn` ([4,4,0,0,8,2,9,2])
109109 it "Nest <|>, <=>, <|> (2)" $
110110 let t = timed
111111 in toList (
112112 ((t 4 <|> t 8) <=> (t 0 <|> t 2))
113− <|> ((t 4 <|> t 8) <=> (t 0 <|> t 2)))
114− `shouldReturn` ([4,4,0,0,8,8,2,2])
113+ <|> ((t 4 <|> t 9) <=> (t 0 <|> t 2)))
114+ `shouldReturn` ([4,4,0,0,8,2,9,2])
115+ it "Nest <|>, <|>, <|>" $
116+ let t = timed
117+ in toList (
118+ ((t 4 <|> t 8) <|> (t 0 <|> t 2))
119+ <|> ((t 4 <|> t 8) <|> (t 0 <|> t 2)))
120+ `shouldReturn` ([0,0,2,2,4,4,8,8])
115121
116122 ---------------------------------------------------------------------------
117123 -- Monoidal composition recursion loops
@@ -165,12 +171,11 @@ pureBind f = do
165171
166172interleaved
167173 :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)
168− -> ([Int] -> [Int])
169174 -> Spec
170−interleaved f srt =
175+interleaved f =
171176 it "Interleave four" $
172177 toList ((return 0 <> return 1) `f` (return 100 <> return 101))
173− `shouldReturn` (srt [0, 100, 1, 101])
178+ `shouldReturn` ([0, 100, 1, 101])
174179
175180parallelCheck :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec
176181parallelCheck f = do