1ddcfc4 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

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

1ddcfc4Adithya Kumar6 years ago

Changelog.md+3

compare with master

unifiedsplit

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

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 function in `Streamly.Prelude.zipWith` and

19+ `Streamly.Prelude.zipWithM` is now applied concurrently for concurrent

20+ streams.

19### Breaking changes

22### Breaking changes

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

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

352-------------------------------------------------------------------------------

353

354{-# INLINE zipWith #-}

355−zipWith :: Monad m => Stream m Int -> m ()

355+zipWith :: S.MonadAsync m => Stream m Int -> m ()

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

357

358-------------------------------------------------------------------------------

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

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

995-- Serial Zipping

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

997

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

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

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

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

999--

1001--

1000-- @since 0.1.0

1002-- @since 0.1.0

1001{-# INLINABLE zipWith #-}

1003{-# INLINABLE zipWith #-}

1004+zipWith ::

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

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

1003zipWith f = go

1006zipWith f = go

1004 where

1007 where

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

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

1006 let merge a ra =

1009 let merge a ra =

1007 let single2 b = sng (f a b)

1010 let single2 b = sng (f a b)

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

1011+ runIt x = foldStream st yld sng stp x

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

1009 in foldStream (adaptState st) yield2 single2 stp my

1013 in foldStream (adaptState st) yield2 single2 stp my

1010 let single1 a = merge a nil

1014 let single1 a = merge a nil

1011 yield1 = merge

1015 yield1 = merge

1012 foldStream (adaptState st) yield1 single1 stp mx

1016 foldStream (adaptState st) yield1 single1 stp mx

1013

1017

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

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

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

1015--

1020--

1016-- @since 0.1.0

1021-- @since 0.1.0

1017{-# INLINABLE zipWithM #-}

1022{-# INLINABLE zipWithM #-}

1023+zipWithM ::

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

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

1019zipWithM f m1 m2 = go m1 m2

1025zipWithM f m1 m2 = go m1 m2

1020 where

1026 where

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

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

1022 let merge a ra =

1028 let merge a ra =

1023 let runIt x = foldStream st yld sng stp x

1029 let runIt x = foldStream st yld sng stp x

1024 single2 b = f a b >>= sng

1030 single2 b = f a b >>= sng

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

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

1026 in foldStream (adaptState st) yield2 single2 stp my

1032 in foldStream (adaptState st) yield2 single2 stp my

1027 let single1 a = merge a nil

1033 let single1 a = merge a nil

1028 yield1 = merge

1034 yield1 = merge

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

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

35 )

36where

38+#include "inline.hs"

39+

38import Control.Applicative (liftA2)

40import Control.Applicative (liftA2)

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

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

40#if MIN_VERSION_deepseq(1,4,3)

42#if MIN_VERSION_deepseq(1,4,3)

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

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

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

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

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

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

62+

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -85,11 +82,21 @@ import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)

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

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

86--

83--

87-- @since 0.4.0

84-- @since 0.4.0

88−{-# INLINABLE zipWithM #-}

85+{-# INLINE_EARLY zipWithM #-}

86+zipWithM ::

87+ (IsStream t, MonadAsync m) => (a -> b -> m c) -> t m a -> t m b -> t m c

88+zipWithM f m1 m2 =

89+ K.fromStream $ K.zipWithM f (K.toStream m1) (K.toStream m2)

90+

91+{-# RULES "zipWithM serial" zipWithM = zipWithMSerial #-}

92+{-# INLINE zipWithMSerial #-}

93+zipWithMSerial ::

89−zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

94+ Monad m => (a -> b -> m c) -> SerialT m a -> SerialT m b -> SerialT m c

90−zipWithM f m1 m2 = P.fromStreamS $ S.zipWithM f (P.toStreamS m1) (P.toStreamS m2)

95+zipWithMSerial f m1 m2 =

91−

96+ D.fromStreamD $ D.zipWithM f (D.toStreamD m1) (D.toStreamD m2)

97+

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

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

99+-- function is applied concurrently for concurrent streams.

93--

100--

94-- @

101-- @

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

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

@@ -97,9 +104,17 @@ zipWithM f m1 m2 = P.fromStreamS $ S.zipWithM f (P.toStreamS m1) (P.toStreamS m2

97-- @

104-- @

98--

105--

99-- @since 0.1.0

106-- @since 0.1.0

100−{-# INLINABLE zipWith #-}

107+{-# INLINE_EARLY zipWith #-}

108+zipWith ::

109+ (IsStream t, MonadAsync m) => (a -> b -> c) -> t m a -> t m b -> t m c

110+zipWith f m1 m2 = K.fromStream $ K.zipWith f (K.toStream m1) (K.toStream m2)

111+

112+{-# RULES "zipWith serial" zipWith = zipWithSerial #-}

113+{-# INLINE zipWithSerial #-}

114+zipWithSerial ::

101−zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c

115+ Monad m => (a -> b -> c) -> SerialT m a -> SerialT m b -> SerialT m c

102−zipWith f m1 m2 = P.fromStreamS $ S.zipWith f (P.toStreamS m1) (P.toStreamS m2)

116+zipWithSerial f m1 m2 =

117+ D.fromStreamD $ D.zipWith f (D.toStreamD m1) (D.toStreamD m2)

103

118

104------------------------------------------------------------------------------

119------------------------------------------------------------------------------

105-- Parallel Zipping

120-- Parallel Zipping

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

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

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

205 pure = ZipSerialM . K.repeat

220 pure = ZipSerialM . K.repeat

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

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

207− (<*>) = zipWith id

222+ m1 <*> m2 = fromStream $ toStream $ zipWithSerial id m1_ m2_

223+ where

224+ m1_ = fromStream (toStream m1)

225+ m2_ = fromStream (toStream m2)

208

226

209FOLDABLE_INSTANCE(ZipSerialM)

227FOLDABLE_INSTANCE(ZipSerialM)

210TRAVERSABLE_INSTANCE(ZipSerialM)

228TRAVERSABLE_INSTANCE(ZipSerialM)