e094283 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Revert "Fix zipWith(M) to work concurrently according to the stream type"

e094283Harendra Kumar5 years ago

Changelog.md−3

compare with master

unifiedsplit

@@ -15,9 +15,6 @@

15* `encodeLatin1` now silently truncates any character beyond 255 to incorrect

16 characters in the input stream. Use `encodeLatin1'` to recover previous

17 functionality.

18−* The zipping functions `Streamly.Prelude.zipWith` and

19− `Streamly.Prelude.zipWithM` are now applied concurrently for concurrent

20− streams.

22### Breaking type changes

19### Breaking type changes

benchmark/Streamly/Benchmark/Data/Stream/StreamK.hs+1−1

@@ -372,7 +372,7 @@ iterateDropWhileTrue streamLen iterStreamLen maxIters = iterateSource iterStream

372-------------------------------------------------------------------------------

373

374{-# INLINE zipWith #-}

375−zipWith :: S.MonadAsync m => Stream m Int -> m ()

375+zipWith :: Monad m => Stream m Int -> m ()

376zipWith src = drain $ S.zipWith (,) src src

377

378-------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/StreamK.hs+6−12

@@ -995,40 +995,34 @@ mapMaybe f m = go m

995-- Serial Zipping

996------------------------------------------------------------------------------

997

998−-- XXX We can probably implement zipWith in terms of zipWithM

999−-- | Zip two streams serially using a pure zipping function. The zipping

998+-- | Zip two streams serially using a pure zipping function.

1000−-- function is applied concurrently for concurrent streams.

1001--

999--

1002-- @since 0.1.0

1000-- @since 0.1.0

1003{-# INLINABLE zipWith #-}

1001{-# INLINABLE zipWith #-}

1004−zipWith ::

1005− (IsStream t, MonadAsync m) => (a -> b -> c) -> t m a -> t m b -> t m c

1002+zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

1006zipWith f = go

1003zipWith f = go

1007 where

1004 where

1008 go mx my = mkStream $ \st yld sng stp -> do

1005 go mx my = mkStream $ \st yld sng stp -> do

1009 let merge a ra =

1006 let merge a ra =

1010 let single2 b = sng (f a b)

1007 let single2 b = sng (f a b)

1011− runIt x = foldStream st yld sng stp x

1012− yield2 b rb = runIt (return (f a b) `consM` go ra rb)

1008+ yield2 b rb = yld (f a b) (go ra rb)

1013 in foldStream (adaptState st) yield2 single2 stp my

1009 in foldStream (adaptState st) yield2 single2 stp my

1014 let single1 a = merge a nil

1010 let single1 a = merge a nil

1015 yield1 = merge

1011 yield1 = merge

1016 foldStream (adaptState st) yield1 single1 stp mx

1012 foldStream (adaptState st) yield1 single1 stp mx

1017

1013

1018−-- | Zip two streams serially using a monadic zipping function. The zipping

1014+-- | Zip two streams serially using a monadic zipping function.

1019−-- function is applied concurrently for concurrent streams.

1020--

1015--

1021-- @since 0.1.0

1016-- @since 0.1.0

1022{-# INLINABLE zipWithM #-}

1017{-# INLINABLE zipWithM #-}

1023−zipWithM ::

1024− (IsStream t, MonadAsync m) => (a -> b -> m c) -> t m a -> t m b -> t m c

1018+zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

1025zipWithM f m1 m2 = go m1 m2

1019zipWithM f m1 m2 = go m1 m2

1026 where

1020 where

1027 go mx my = mkStream $ \st yld sng stp -> do

1021 go mx my = mkStream $ \st yld sng stp -> do

1028 let merge a ra =

1022 let merge a ra =

1029 let runIt x = foldStream st yld sng stp x

1023 let runIt x = foldStream st yld sng stp x

1030 single2 b = f a b >>= sng

1024 single2 b = f a b >>= sng

1031− yield2 b rb = runIt (f a b `consM` go ra rb)

1025+ yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)

1032 in foldStream (adaptState st) yield2 single2 stp my

1026 in foldStream (adaptState st) yield2 single2 stp my

1033 let single1 a = merge a nil

1027 let single1 a = merge a nil

1034 yield1 = merge

1028 yield1 = merge

src/Streamly/Internal/Data/Stream/Zip.hs+19−37

@@ -35,8 +35,6 @@ module Streamly.Internal.Data.Stream.Zip

35 )

36where

38−#include "inline.hs"

39−

40import Control.Applicative (liftA2)

38import Control.Applicative (liftA2)

41import Control.DeepSeq (NFData(..))

39import Control.DeepSeq (NFData(..))

42#if MIN_VERSION_deepseq(1,4,3)

40#if MIN_VERSION_deepseq(1,4,3)

@@ -58,13 +56,18 @@ import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream)

58import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

56import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

59import Streamly.Internal.Data.SVar (MonadAsync)

57import Streamly.Internal.Data.SVar (MonadAsync)

61−import Streamly.Internal.Data.Stream.Serial (SerialT)

62−

63import qualified Streamly.Internal.Data.Stream.Parallel as Par

59import qualified Streamly.Internal.Data.Stream.Parallel as Par

64import qualified Streamly.Internal.Data.Stream.Prelude as P

60import qualified Streamly.Internal.Data.Stream.Prelude as P

65− (cmpBy, eqBy, foldl', foldr, fromList, toList)

61+ (cmpBy, eqBy, foldl', foldr, fromList, toList, fromStreamS, toStreamS)

66−import qualified Streamly.Internal.Data.Stream.StreamK as K

62+import qualified Streamly.Internal.Data.Stream.StreamK as K (repeat)

63+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

67−import qualified Streamly.Internal.Data.Stream.StreamD as D

64+import qualified Streamly.Internal.Data.Stream.StreamD as D (zipWithM)

65+import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

66+#ifdef USE_STREAMK_ONLY

67+import qualified Streamly.Internal.Data.Stream.StreamK as S (zipWith, zipWithM)

68+#else

69+import qualified Streamly.Internal.Data.Stream.StreamD as S (zipWith, zipWithM)

70+#endif

69import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)

72import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)

@@ -83,21 +86,11 @@ import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)

83-- | Like 'zipWith' but using a monadic zipping function.

86-- | Like 'zipWith' but using a monadic zipping function.

84--

87--

85-- @since 0.4.0

88-- @since 0.4.0

86−{-# INLINE_EARLY zipWithM #-}

89+{-# INLINABLE zipWithM #-}

87−zipWithM ::

88− (IsStream t, MonadAsync m) => (a -> b -> m c) -> t m a -> t m b -> t m c

89−zipWithM f m1 m2 =

90− K.fromStream $ K.zipWithM f (K.toStream m1) (K.toStream m2)

91−

92−{-# RULES "zipWithM serial" zipWithM = zipWithMSerial #-}

93−{-# INLINE zipWithMSerial #-}

94−zipWithMSerial ::

95− Monad m => (a -> b -> m c) -> SerialT m a -> SerialT m b -> SerialT m c

90+zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

96−zipWithMSerial f m1 m2 =

91+zipWithM f m1 m2 = P.fromStreamS $ S.zipWithM f (P.toStreamS m1) (P.toStreamS m2)

97− D.fromStreamD $ D.zipWithM f (D.toStreamD m1) (D.toStreamD m2)

92+

98−

99−-- | Zip two streams serially using a pure zipping function. The zipping

93+-- | Zip two streams serially using a pure zipping function.

100−-- function is applied concurrently for concurrent streams.

101--

94--

102-- @

95-- @

103-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])

96-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])

@@ -105,17 +98,9 @@ zipWithMSerial f m1 m2 =

105-- @

98-- @

106--

99--

107-- @since 0.1.0

100-- @since 0.1.0

108−{-# INLINE_EARLY zipWith #-}

101+{-# INLINABLE zipWith #-}

109−zipWith ::

110− (IsStream t, MonadAsync m) => (a -> b -> c) -> t m a -> t m b -> t m c

111−zipWith f m1 m2 = K.fromStream $ K.zipWith f (K.toStream m1) (K.toStream m2)

112−

113−{-# RULES "zipWith serial" zipWith = zipWithSerial #-}

114−{-# INLINE zipWithSerial #-}

115−zipWithSerial ::

116− Monad m => (a -> b -> c) -> SerialT m a -> SerialT m b -> SerialT m c

102+zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c

117−zipWithSerial f m1 m2 =

103+zipWith f m1 m2 = P.fromStreamS $ S.zipWith f (P.toStreamS m1) (P.toStreamS m2)

118− D.fromStreamD $ D.zipWith f (D.toStreamD m1) (D.toStreamD m2)

119

104

120------------------------------------------------------------------------------

105------------------------------------------------------------------------------

121-- Parallel Zipping

106-- Parallel Zipping

@@ -220,10 +205,7 @@ instance Monad m => Functor (ZipSerialM m) where

220instance Monad m => Applicative (ZipSerialM m) where

205instance Monad m => Applicative (ZipSerialM m) where

221 pure = ZipSerialM . K.repeat

206 pure = ZipSerialM . K.repeat

222 {-# INLINE (<*>) #-}

207 {-# INLINE (<*>) #-}

223− m1 <*> m2 = fromStream $ toStream $ zipWithSerial id m1_ m2_

208+ (<*>) = zipWith id

224− where

225− m1_ = fromStream (toStream m1)

226− m2_ = fromStream (toStream m2)

227

209

228FOLDABLE_INSTANCE(ZipSerialM)

210FOLDABLE_INSTANCE(ZipSerialM)

229TRAVERSABLE_INSTANCE(ZipSerialM)

211TRAVERSABLE_INSTANCE(ZipSerialM)