git.iko.soy/streamly
on master help sign in
Fix append and interleave nested inside parallel
5919be5Harendra Kumar
src/Asyncly/AsyncT.hs+10−7
compare with master
unifiedsplit
@@ -131,11 +131,11 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
131instance Semigroup (AsyncT m a) where
131instance Semigroup (AsyncT m a) where
132 m1 <> m2 = go m1
132 m1 <> m2 = go m1
133 where
133 where
134− go (AsyncT m) = AsyncT $ \ctx stp yld ->
134+ go (AsyncT m) = AsyncT $ \_ stp yld ->
135− let stop = (runAsyncT m2) ctx stp yld
135+ let stop = (runAsyncT m2) Nothing stp yld
136 yield a Nothing = yld a (Just m2)
136 yield a Nothing = yld a (Just m2)
137 yield a (Just r) = yld a (Just (go r))
137 yield a (Just r) = yld a (Just (go r))
138− in m ctx stop yield
138+ in m Nothing stop yield
139
139
140instance Monoid (AsyncT m a) where
140instance Monoid (AsyncT m a) where
141 mempty = AsyncT $ \_ stp _ -> stp
141 mempty = AsyncT $ \_ stp _ -> stp
@@ -143,11 +143,11 @@ instance Monoid (AsyncT m a) where
143
143
144-- | Same as '<=>'.
144-- | Same as '<=>'.
145interleave :: AsyncT m a -> AsyncT m a -> AsyncT m a
145interleave :: AsyncT m a -> AsyncT m a -> AsyncT m a
146−interleave m1 m2 = AsyncT $ \ctx stp yld -> do
146+interleave m1 m2 = AsyncT $ \_ stp yld -> do
147− let stop = (runAsyncT m2) ctx stp yld
147+ let stop = (runAsyncT m2) Nothing stp yld
148 yield a Nothing = yld a (Just m2)
148 yield a Nothing = yld a (Just m2)
149 yield a (Just r) = yld a (Just (interleave m2 r))
149 yield a (Just r) = yld a (Just (interleave m2 r))
150− (runAsyncT m1) ctx stop yield
150+ (runAsyncT m1) Nothing stop yield
151
151
152infixr 5 <=>
152infixr 5 <=>
153
153
@@ -374,7 +374,10 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
374 res <- liftIO $ tryTakeMVar (doorBell ctx)
374 res <- liftIO $ tryTakeMVar (doorBell ctx)
375 when (isNothing res) $ sendWorkerWait ctx
375 when (isNothing res) $ sendWorkerWait ctx
376 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
376 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
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?
377− (runAsyncT $ processEvents pctx list) Nothing stp yld
380+ (runAsyncT $ processEvents pctx (reverse list)) Nothing stp yld
378
381
379 where
382 where
380
383
test/Main.hs+14−9
@@ -48,8 +48,8 @@ main = hspec $ do
48 -- Monoidal Composition ordering checks
48 -- Monoidal Composition ordering checks
49 ---------------------------------------------------------------------------
49 ---------------------------------------------------------------------------
50
50
51− describe "Serial interleaved (<=>)" $ interleaved (<=>) id
51+ describe "Serial interleaved (<=>)" $ interleaved (<=>)
52− describe "Parallel interleaved (<|>)" $ interleaved (<|>) reverse
52+ describe "Parallel interleaved (<|>)" $ interleaved (<|>)
53 describe "Left biased parallel time order check" $ parallelCheck (<|)
53 describe "Left biased parallel time order check" $ parallelCheck (<|)
54 describe "Fair parallel time order check" $ parallelCheck (<|>)
54 describe "Fair parallel time order check" $ parallelCheck (<|>)
55
55
@@ -104,14 +104,20 @@ main = hspec $ do
104 let t = timed
104 let t = timed
105 in toList (
105 in toList (
106 ((t 8 <|> t 4) <=> (t 2 <|> t 0))
106 ((t 8 <|> t 4) <=> (t 2 <|> t 0))
107− <|> ((t 8 <|> t 4) <=> (t 2 <|> t 0)))
107+ <|> ((t 9 <|> t 4) <=> (t 2 <|> t 0)))
108− `shouldReturn` ([4,4,0,0,8,8,2,2])
108+ `shouldReturn` ([4,4,0,0,8,2,9,2])
109 it "Nest <|>, <=>, <|> (2)" $
109 it "Nest <|>, <=>, <|> (2)" $
110 let t = timed
110 let t = timed
111 in toList (
111 in toList (
112 ((t 4 <|> t 8) <=> (t 0 <|> t 2))
112 ((t 4 <|> t 8) <=> (t 0 <|> t 2))
113− <|> ((t 4 <|> t 8) <=> (t 0 <|> t 2)))
113+ <|> ((t 4 <|> t 9) <=> (t 0 <|> t 2)))
114− `shouldReturn` ([4,4,0,0,8,8,2,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])
115
121
116 ---------------------------------------------------------------------------
122 ---------------------------------------------------------------------------
117 -- Monoidal composition recursion loops
123 -- Monoidal composition recursion loops
@@ -165,12 +171,11 @@ pureBind f = do
165
171
166interleaved
172interleaved
167 :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)
173 :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)
168− -> ([Int] -> [Int])
169 -> Spec
174 -> Spec
170−interleaved f srt =
175+interleaved f =
171 it "Interleave four" $
176 it "Interleave four" $
172 toList ((return 0 <> return 1) `f` (return 100 <> return 101))
177 toList ((return 0 <> return 1) `f` (return 100 <> return 101))
173− `shouldReturn` (srt [0, 100, 1, 101])
178+ `shouldReturn` ([0, 100, 1, 101])
174
179
175parallelCheck :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec
180parallelCheck :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec
176parallelCheck f = do
181parallelCheck f = do