c28f3f1 · streamly · git.iko.soy

git.iko.soy/streamly

Fix a concurrency bug in filtering ops

The bug causes the ops take, takeWhile, drop, dropWhile, filter and reverse to behave incorrectly due to illegal sharing of SVar across ops.

commitc28f3f1c4705165671b9da261dd275efd29fba85 authorHarendra Kumar <harendra.kumar@gmail.com> dateTue 5 Jun 2018, 18:03 (8 years ago) parent9acbc6c

5 files changed+136−22

compare with master

unifiedsplit

Changelog.md+7 src/Streamly/Prelude.hs+17−17 src/Streamly/Streams.hs+4−4 streamly.cabal+1−1 test/Prop.hs+107

Changelog.md+7

@@ -26,6 +26,13 @@

2626 stream processing function application pipeline concurrently.

2727* Added two functions `mapMaybe` and `mapMaybeM`.

2828

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+

2936## 0.2.0

3037

3138### Breaking changes

src/Streamly/Prelude.hs+17−17

@@ -519,9 +519,9 @@ toList = foldrM (\a xs -> return (a : xs)) []

519519take :: IsStream t => Int -> t m a -> t m a

520520take n m = fromStream $ go n (toStream m)

521521 where

522− go n1 m1 = Stream $ \ctx stp sng yld ->

522+ go n1 m1 = Stream $ \_ stp sng yld ->

523523 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

525525

526526-- | Include only those elements that pass a predicate.

527527--

@@ -530,12 +530,12 @@ take n m = fromStream $ go n (toStream m)

530530filter :: IsStream t => (a -> Bool) -> t m a -> t m a

531531filter p m = fromStream $ go (toStream m)

532532 where

533− go m1 = Stream $ \ctx stp sng yld ->

533+ go m1 = Stream $ \_ stp sng yld ->

534534 let single a | p a = sng a

535535 | otherwise = stp

536536 yield a r | p a = yld a (go r)

537− | otherwise = (S.runStream r) ctx stp single yield

538− in (S.runStream m1) ctx stp single yield

537+ | otherwise = (S.runStream r) Nothing stp single yield

538+ in (S.runStream m1) Nothing stp single yield

539539

540540-- | End the stream as soon as the predicate fails on an element.

541541--

@@ -544,12 +544,12 @@ filter p m = fromStream $ go (toStream m)

544544takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

545545takeWhile p m = fromStream $ go (toStream m)

546546 where

547− go m1 = Stream $ \ctx stp sng yld ->

547+ go m1 = Stream $ \_ stp sng yld ->

548548 let single a | p a = sng a

549549 | otherwise = stp

550550 yield a r | p a = yld a (go r)

551551 | otherwise = stp

552− in (S.runStream m1) ctx stp single yield

552+ in (S.runStream m1) Nothing stp single yield

553553

554554-- | Discard first 'n' elements from the stream and take the rest.

555555--

@@ -557,13 +557,13 @@ takeWhile p m = fromStream $ go (toStream m)

557557drop :: IsStream t => Int -> t m a -> t m a

558558drop n m = fromStream $ go n (toStream m)

559559 where

560− go n1 m1 = Stream $ \ctx stp sng yld ->

560+ go n1 m1 = Stream $ \_ stp sng yld ->

561561 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

563563 -- Somehow "<=" check performs better than a ">"

564564 in if n1 <= 0

565− then (S.runStream m1) ctx stp sng yld

566− else (S.runStream m1) ctx stp single yield

565+ then (S.runStream m1) Nothing stp sng yld

566+ else (S.runStream m1) Nothing stp single yield

567567

568568-- | Drop elements in the stream as long as the predicate succeeds and then

569569-- take the rest of the stream.

@@ -573,12 +573,12 @@ drop n m = fromStream $ go n (toStream m)

573573dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

574574dropWhile p m = fromStream $ go (toStream m)

575575 where

576− go m1 = Stream $ \ctx stp sng yld ->

576+ go m1 = Stream $ \_ stp sng yld ->

577577 let single a | p a = stp

578578 | 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

580580 | otherwise = yld a r

581− in (S.runStream m1) ctx stp single yield

581+ in (S.runStream m1) Nothing stp single yield

582582

583583-- | Map + filter: compute a stream of b from a stream of a and filter out

584584-- elements that are mapped to Nothing. It is similar to 'mapMaybe' in

@@ -710,12 +710,12 @@ length = foldl (\n _ -> n + 1) 0 id

710710reverse :: (IsStream t) => t m a -> t m a

711711reverse m = fromStream $ go S.nil (toStream m)

712712 where

713− go rev rest = Stream $ \svr stp sng yld ->

714− let run x = S.runStream x svr stp sng yld

713+ go rev rest = Stream $ \_ stp sng yld ->

714+ let run x = S.runStream x Nothing stp sng yld

715715 stop = run rev

716716 single a = run $ a `S.cons` rev

717717 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

719719

720720-- XXX replace the recursive "go" with continuation

721721-- | Determine the minimum element in a stream.

src/Streamly/Streams.hs+4−4

@@ -415,10 +415,10 @@ x |&. f = f |$. x

415415

416416withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a

417417withLocal f m =

418− S.Stream $ \svr stp sng yld ->

418+ S.Stream $ \_ stp sng yld ->

419419 let single = local f . sng

420420 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

422422

423423{-

424424-- XXX handle and test cross thread state transfer

@@ -426,8 +426,8 @@ withCatchError

426426 :: MonadError e m

427427 => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a

428428withCatchError m h =

429− S.Stream $ \svr stp sng yld ->

430− let run x = S.runStream x svr stp sng yield

429+ S.Stream $ \_ stp sng yld ->

430+ let run x = S.runStream x Nothing stp sng yield

431431 handle r = r `catchError` \e -> run $ h e

432432 yield a r = yld a (withCatchError r h)

433433 in handle $ run m

streamly.cabal+1−1

@@ -1,5 +1,5 @@

11name: streamly

2−version: 0.2.0

2+version: 0.2.1

33synopsis: Beautiful Streaming, Concurrent and Reactive Composition

44description:

55 Streamly, short for streaming concurrently, is a simple yet powerful

test/Prop.hs+107

@@ -221,6 +221,24 @@ concurrentFoldrApplication n =

221221 sourceUnfoldrM1 n |&. A.foldrM (\x xs -> return (x : xs)) []

222222 equals (==) stream list

223223

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+

224242foldFromList

225243 :: ([Int] -> t IO Int)

226244 -> (t IO Int -> SerialT IO Int)

@@ -336,6 +354,69 @@ concurrentOps constr desc t eq = do

336354 concurrentMapM constr eq $ \n mv stream ->

337355 t $ A.mapM (mvarSequenceOp mv n) stream

338356

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+

339420wrapMaybe :: Eq a1 => ([a1] -> a2) -> [a1] -> Maybe a2

340421wrapMaybe f =

341422 \x ->

@@ -641,6 +722,32 @@ main = hspec $ do

641722 prop "concurrent foldl application" $ withMaxSuccess maxTestCount $

642723 concurrentFoldlApplication

643724

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+

644751 describe "Stream elimination operations" $ do

645752 eliminationOps A.fromFoldable "serially" serially

646753 eliminationOps A.fromFoldable "aheadly" aheadly