git.iko.soy/streamly
on master help sign in
Fix a concurrency bug in filtering ops
c28f3f1Harendra Kumar
Changelog.md+7
compare with master
unifiedsplit
@@ -26,6 +26,13 @@
26 stream processing function application pipeline concurrently.
26 stream processing function application pipeline concurrently.
27* Added two functions `mapMaybe` and `mapMaybeM`.
27* Added two functions `mapMaybe` and `mapMaybeM`.
28
28
29+## 0.2.1
30+
31+### Bug Fixes
32+* Fixed a bug that caused some transformation ops to return incorrect results
33+ when used with concurrent streams. The affected ops are `take`, `filter`,
34+ `takeWhile`, `drop`, `dropWhile`, and `reverse`.
35+
29## 0.2.0
36## 0.2.0
30
37
31### Breaking changes
38### Breaking changes
src/Streamly/Prelude.hs+17−17
@@ -519,9 +519,9 @@ toList = foldrM (\a xs -> return (a : xs)) []
519take :: IsStream t => Int -> t m a -> t m a
519take :: IsStream t => Int -> t m a -> t m a
520take n m = fromStream $ go n (toStream m)
520take n m = fromStream $ go n (toStream m)
521 where
521 where
522− go n1 m1 = Stream $ \ctx stp sng yld ->
522+ go n1 m1 = Stream $ \_ stp sng yld ->
523 let yield a r = yld a (go (n1 - 1) r)
523 let yield a r = yld a (go (n1 - 1) r)
524− in if n1 <= 0 then stp else (S.runStream m1) ctx stp sng yield
524+ in if n1 <= 0 then stp else (S.runStream m1) Nothing stp sng yield
525
525
526-- | Include only those elements that pass a predicate.
526-- | Include only those elements that pass a predicate.
527--
527--
@@ -530,12 +530,12 @@ take n m = fromStream $ go n (toStream m)
530filter :: IsStream t => (a -> Bool) -> t m a -> t m a
530filter :: IsStream t => (a -> Bool) -> t m a -> t m a
531filter p m = fromStream $ go (toStream m)
531filter p m = fromStream $ go (toStream m)
532 where
532 where
533− go m1 = Stream $ \ctx stp sng yld ->
533+ go m1 = Stream $ \_ stp sng yld ->
534 let single a | p a = sng a
534 let single a | p a = sng a
535 | otherwise = stp
535 | otherwise = stp
536 yield a r | p a = yld a (go r)
536 yield a r | p a = yld a (go r)
537− | otherwise = (S.runStream r) ctx stp single yield
537+ | otherwise = (S.runStream r) Nothing stp single yield
538− in (S.runStream m1) ctx stp single yield
538+ in (S.runStream m1) Nothing stp single yield
539
539
540-- | End the stream as soon as the predicate fails on an element.
540-- | End the stream as soon as the predicate fails on an element.
541--
541--
@@ -544,12 +544,12 @@ filter p m = fromStream $ go (toStream m)
544takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
544takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
545takeWhile p m = fromStream $ go (toStream m)
545takeWhile p m = fromStream $ go (toStream m)
546 where
546 where
547− go m1 = Stream $ \ctx stp sng yld ->
547+ go m1 = Stream $ \_ stp sng yld ->
548 let single a | p a = sng a
548 let single a | p a = sng a
549 | otherwise = stp
549 | otherwise = stp
550 yield a r | p a = yld a (go r)
550 yield a r | p a = yld a (go r)
551 | otherwise = stp
551 | otherwise = stp
552− in (S.runStream m1) ctx stp single yield
552+ in (S.runStream m1) Nothing stp single yield
553
553
554-- | Discard first 'n' elements from the stream and take the rest.
554-- | Discard first 'n' elements from the stream and take the rest.
555--
555--
@@ -557,13 +557,13 @@ takeWhile p m = fromStream $ go (toStream m)
557drop :: IsStream t => Int -> t m a -> t m a
557drop :: IsStream t => Int -> t m a -> t m a
558drop n m = fromStream $ go n (toStream m)
558drop n m = fromStream $ go n (toStream m)
559 where
559 where
560− go n1 m1 = Stream $ \ctx stp sng yld ->
560+ go n1 m1 = Stream $ \_ stp sng yld ->
561 let single _ = stp
561 let single _ = stp
562− yield _ r = (S.runStream $ go (n1 - 1) r) ctx stp sng yld
562+ yield _ r = (S.runStream $ go (n1 - 1) r) Nothing stp sng yld
563 -- Somehow "<=" check performs better than a ">"
563 -- Somehow "<=" check performs better than a ">"
564 in if n1 <= 0
564 in if n1 <= 0
565− then (S.runStream m1) ctx stp sng yld
565+ then (S.runStream m1) Nothing stp sng yld
566− else (S.runStream m1) ctx stp single yield
566+ else (S.runStream m1) Nothing stp single yield
567
567
568-- | Drop elements in the stream as long as the predicate succeeds and then
568-- | Drop elements in the stream as long as the predicate succeeds and then
569-- take the rest of the stream.
569-- take the rest of the stream.
@@ -573,12 +573,12 @@ drop n m = fromStream $ go n (toStream m)
573dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
573dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
574dropWhile p m = fromStream $ go (toStream m)
574dropWhile p m = fromStream $ go (toStream m)
575 where
575 where
576− go m1 = Stream $ \ctx stp sng yld ->
576+ go m1 = Stream $ \_ stp sng yld ->
577 let single a | p a = stp
577 let single a | p a = stp
578 | otherwise = sng a
578 | otherwise = sng a
579− yield a r | p a = (S.runStream r) ctx stp single yield
579+ yield a r | p a = (S.runStream r) Nothing stp single yield
580 | otherwise = yld a r
580 | otherwise = yld a r
581− in (S.runStream m1) ctx stp single yield
581+ in (S.runStream m1) Nothing stp single yield
582
582
583-- | Map + filter: compute a stream of b from a stream of a and filter out
583-- | Map + filter: compute a stream of b from a stream of a and filter out
584-- elements that are mapped to Nothing. It is similar to 'mapMaybe' in
584-- elements that are mapped to Nothing. It is similar to 'mapMaybe' in
@@ -710,12 +710,12 @@ length = foldl (\n _ -> n + 1) 0 id
710reverse :: (IsStream t) => t m a -> t m a
710reverse :: (IsStream t) => t m a -> t m a
711reverse m = fromStream $ go S.nil (toStream m)
711reverse m = fromStream $ go S.nil (toStream m)
712 where
712 where
713− go rev rest = Stream $ \svr stp sng yld ->
713+ go rev rest = Stream $ \_ stp sng yld ->
714− let run x = S.runStream x svr stp sng yld
714+ let run x = S.runStream x Nothing stp sng yld
715 stop = run rev
715 stop = run rev
716 single a = run $ a `S.cons` rev
716 single a = run $ a `S.cons` rev
717 yield a r = run $ go (a `S.cons` rev) r
717 yield a r = run $ go (a `S.cons` rev) r
718− in S.runStream rest svr stop single yield
718+ in S.runStream rest Nothing stop single yield
719
719
720-- XXX replace the recursive "go" with continuation
720-- XXX replace the recursive "go" with continuation
721-- | Determine the minimum element in a stream.
721-- | Determine the minimum element in a stream.
src/Streamly/Streams.hs+4−4
@@ -415,10 +415,10 @@ x |&. f = f |$. x
415
415
416withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a
416withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a
417withLocal f m =
417withLocal f m =
418− S.Stream $ \svr stp sng yld ->
418+ S.Stream $ \_ stp sng yld ->
419 let single = local f . sng
419 let single = local f . sng
420 yield a r = local f $ yld a (withLocal f r)
420 yield a r = local f $ yld a (withLocal f r)
421− in (S.runStream m) svr (local f stp) single yield
421+ in (S.runStream m) Nothing (local f stp) single yield
422
422
423{-
423{-
424-- XXX handle and test cross thread state transfer
424-- XXX handle and test cross thread state transfer
@@ -426,8 +426,8 @@ withCatchError
426 :: MonadError e m
426 :: MonadError e m
427 => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a
427 => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a
428withCatchError m h =
428withCatchError m h =
429− S.Stream $ \svr stp sng yld ->
429+ S.Stream $ \_ stp sng yld ->
430− let run x = S.runStream x svr stp sng yield
430+ let run x = S.runStream x Nothing stp sng yield
431 handle r = r `catchError` \e -> run $ h e
431 handle r = r `catchError` \e -> run $ h e
432 yield a r = yld a (withCatchError r h)
432 yield a r = yld a (withCatchError r h)
433 in handle $ run m
433 in handle $ run m
streamly.cabal+1−1
@@ -1,5 +1,5 @@
1name: streamly
1name: streamly
2−version: 0.2.0
2+version: 0.2.1
3synopsis: Beautiful Streaming, Concurrent and Reactive Composition
3synopsis: Beautiful Streaming, Concurrent and Reactive Composition
4description:
4description:
5 Streamly, short for streaming concurrently, is a simple yet powerful
5 Streamly, short for streaming concurrently, is a simple yet powerful
test/Prop.hs+107
@@ -221,6 +221,24 @@ concurrentFoldrApplication n =
221 sourceUnfoldrM1 n |&. A.foldrM (\x xs -> return (x : xs)) []
221 sourceUnfoldrM1 n |&. A.foldrM (\x xs -> return (x : xs)) []
222 equals (==) stream list
222 equals (==) stream list
223
223
224+transformCombineFromList
225+ :: Semigroup (t IO Int)
226+ => ([Int] -> t IO Int)
227+ -> ([Int] -> [Int] -> Bool)
228+ -> ([Int] -> [Int])
229+ -> (t IO Int -> SerialT IO Int)
230+ -> (t IO Int -> t IO Int)
231+ -> [Int]
232+ -> [Int]
233+ -> [Int]
234+ -> Property
235+transformCombineFromList constr eq listOp t op a b c =
236+ monadicIO $ do
237+ stream <- run ((A.toList . t) $
238+ constr a <> op (constr b <> constr c))
239+ let list = a <> listOp (b <> c)
240+ equals eq stream list
241+
224foldFromList
242foldFromList
225 :: ([Int] -> t IO Int)
243 :: ([Int] -> t IO Int)
226 -> (t IO Int -> SerialT IO Int)
244 -> (t IO Int -> SerialT IO Int)
@@ -336,6 +354,69 @@ concurrentOps constr desc t eq = do
336 concurrentMapM constr eq $ \n mv stream ->
354 concurrentMapM constr eq $ \n mv stream ->
337 t $ A.mapM (mvarSequenceOp mv n) stream
355 t $ A.mapM (mvarSequenceOp mv n) stream
338
356
357+-- XXX add tests for MonadReader and MonadError etc. In case an SVar is
358+-- accidentally passed through them.
359+transformCombineOpsCommon
360+ :: (IsStream t, Semigroup (t IO Int))
361+ => ([Int] -> t IO Int)
362+ -> String
363+ -> (t IO Int -> SerialT IO Int)
364+ -> ([Int] -> [Int] -> Bool)
365+ -> Spec
366+transformCombineOpsCommon constr desc t eq = do
367+ let transform = transformCombineFromList constr eq
368+ -- Filtering
369+ prop (desc ++ " filter False") $
370+ transform (filter (const False)) t (A.filter (const False))
371+ prop (desc ++ " filter True") $
372+ transform (filter (const True)) t (A.filter (const True))
373+ prop (desc ++ " filter even") $
374+ transform (filter even) t (A.filter even)
375+
376+ prop (desc ++ " take maxBound") $
377+ transform (take maxBound) t (A.take maxBound)
378+ prop (desc ++ " take 0") $ transform (take 0) t (A.take 0)
379+
380+ prop (desc ++ " takeWhile True") $
381+ transform (takeWhile (const True)) t (A.takeWhile (const True))
382+ prop (desc ++ " takeWhile False") $
383+ transform (takeWhile (const False)) t (A.takeWhile (const False))
384+
385+ prop (desc ++ " drop maxBound") $
386+ transform (drop maxBound) t (A.drop maxBound)
387+ prop (desc ++ " drop 0") $ transform (drop 0) t (A.drop 0)
388+
389+ prop (desc ++ " dropWhile True") $
390+ transform (dropWhile (const True)) t (A.dropWhile (const True))
391+ prop (desc ++ " dropWhile False") $
392+ transform (dropWhile (const False)) t (A.dropWhile (const False))
393+ prop (desc ++ " scan") $ transform (scanl' (flip const) 0) t
394+ (A.scanl' (flip const) 0)
395+ prop (desc ++ " reverse") $ transform reverse t A.reverse
396+
397+transformCombineOpsOrdered
398+ :: (IsStream t, Semigroup (t IO Int))
399+ => ([Int] -> t IO Int)
400+ -> String
401+ -> (t IO Int -> SerialT IO Int)
402+ -> ([Int] -> [Int] -> Bool)
403+ -> Spec
404+transformCombineOpsOrdered constr desc t eq = do
405+ let transform = transformCombineFromList constr eq
406+ -- Filtering
407+ prop (desc ++ " take 1") $ transform (take 1) t (A.take 1)
408+ prop (desc ++ " take 10") $ transform (take 10) t (A.take 10)
409+
410+ prop (desc ++ " takeWhile > 0") $
411+ transform (takeWhile (> 0)) t (A.takeWhile (> 0))
412+
413+ prop (desc ++ " drop 1") $ transform (drop 1) t (A.drop 1)
414+ prop (desc ++ " drop 10") $ transform (drop 10) t (A.drop 10)
415+
416+ prop (desc ++ " dropWhile > 0") $
417+ transform (dropWhile (> 0)) t (A.dropWhile (> 0))
418+ prop (desc ++ " scan") $ transform (scanl' (+) 0) t (A.scanl' (+) 0)
419+
339wrapMaybe :: Eq a1 => ([a1] -> a2) -> [a1] -> Maybe a2
420wrapMaybe :: Eq a1 => ([a1] -> a2) -> [a1] -> Maybe a2
340wrapMaybe f =
421wrapMaybe f =
341 \x ->
422 \x ->
@@ -641,6 +722,32 @@ main = hspec $ do
641 prop "concurrent foldl application" $ withMaxSuccess maxTestCount $
722 prop "concurrent foldl application" $ withMaxSuccess maxTestCount $
642 concurrentFoldlApplication
723 concurrentFoldlApplication
643
724
725+ -- These tests are specifically targeted towards detecting illegal sharing
726+ -- of SVar across conurrent streams.
727+ describe "Stream transform and combine operations" $ do
728+ transformCombineOpsCommon A.fromFoldable "serially" serially (==)
729+ transformCombineOpsCommon A.fromFoldable "aheadly" aheadly (==)
730+ transformCombineOpsCommon A.fromFoldable "wSerially" wSerially sortEq
731+ transformCombineOpsCommon A.fromFoldable "zipSerially" zipSerially (==)
732+ transformCombineOpsCommon A.fromFoldable "zipAsyncly" zipAsyncly (==)
733+ transformCombineOpsCommon A.fromFoldable "asyncly" asyncly sortEq
734+ transformCombineOpsCommon A.fromFoldable "wAsyncly" wAsyncly sortEq
735+ transformCombineOpsCommon A.fromFoldable "parallely" parallely sortEq
736+
737+ transformCombineOpsCommon folded "serially" serially (==)
738+ transformCombineOpsCommon folded "aheadly" aheadly (==)
739+ transformCombineOpsCommon folded "wSerially" wSerially sortEq
740+ transformCombineOpsCommon folded "zipSerially" zipSerially (==)
741+ transformCombineOpsCommon folded "zipAsyncly" zipAsyncly (==)
742+ transformCombineOpsCommon folded "asyncly" asyncly sortEq
743+ transformCombineOpsCommon folded "wAsyncly" wAsyncly sortEq
744+ transformCombineOpsCommon folded "parallely" parallely sortEq
745+
746+ transformCombineOpsOrdered A.fromFoldable "serially" serially (==)
747+ transformCombineOpsOrdered A.fromFoldable "serially" aheadly (==)
748+ transformCombineOpsOrdered A.fromFoldable "zipSerially" zipSerially (==)
749+ transformCombineOpsOrdered A.fromFoldable "zipAsyncly" zipAsyncly (==)
750+
644 describe "Stream elimination operations" $ do
751 describe "Stream elimination operations" $ do
645 eliminationOps A.fromFoldable "serially" serially
752 eliminationOps A.fromFoldable "serially" serially
646 eliminationOps A.fromFoldable "aheadly" aheadly
753 eliminationOps A.fromFoldable "aheadly" aheadly