cd50805 · streamly · git.iko.soy

git.iko.soy/streamly

Do not share concurrency state in concurrent function application

commitcd50805c9252466b7ea1044cf42a276cf549aba9 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 8 Aug 2018, 17:17 (8 years ago) parent9b87957

3 files changed+13−1

unifiedsplit

Changelog.md+6

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

11## Unreleased

3+### Bug Fixes

5+* Fix a bug in concurrent function application that in certain cases would

6+ unnecessarily share the concurrency state resulting in incorrect output

7+ stream.

39### Enhancements

410

511* Add `foldl1'`, `foldr1`, `intersperseM`, `find`, `lookup`, `and`, `or`,

src/Streamly/Streams/Parallel.hs+1−1

@@ -130,7 +130,7 @@ applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b

130130applyWith f m = fromStream $ Stream $ \st stp sng yld -> do

131131 sv <- newParallelVar

132132 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))

133− unStream (toStream $ f $ fromSVar sv) st stp sng yld

133+ unStream (toStream $ f $ fromSVar sv) (rstState st) stp sng yld

134134

135135------------------------------------------------------------------------------

136136-- Stream runner concurrent function application

test/Prop.hs+6

@@ -8,6 +8,7 @@ import Control.Monad (when)

88import Control.Applicative (ZipList(..))

99import Control.Concurrent (MVar, takeMVar, putMVar, newEmptyMVar)

1010import Control.Monad (replicateM, replicateM_)

11+import Data.Function ((&))

1112import Data.IORef (readIORef, modifyIORef, newIORef)

1213import Data.List (sort, foldl', scanl', findIndices, findIndex, elemIndices,

1314 elemIndex, find, intersperse, foldl1')

@@ -466,6 +467,11 @@ transformCombineOpsOrdered constr desc t eq = do

466467 transform (dropWhile (> 0)) t (S.dropWhile (> 0))

467468 prop (desc ++ " scan") $ transform (scanl' (+) 0) t (S.scanl' (+) 0)

468469

470+ -- XXX this does not fail when the SVar is shared, need to fix.

471+ prop (desc ++ " concurrent application") $

472+ transform (& (map (+1))) t (|& (S.map (+1)))

473+

474+

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

470476wrapMaybe f =

471477 \x ->