e094283 · streamly · git.iko.soy

git.iko.soy/streamly

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

This reverts commit 1ddcfc46345cd4e4c3115a955668a6d5158a64cd. This requires MonadAsync constraint which breaks the existing zipWith for pure streams e.g. 'SerialT Identity' (for example in streaming-benchmarks package). We can possibly have different zipWith APIs for concurrent zipping.

commite09428332cb7a12effc52c5ad9c89dd6bcb36504 authorHarendra Kumar <harendra@composewell.com> dateTue 22 Jun 2021, 12:36 (5 years ago) parent6587585

4 files changed+26−53

compare with master

unifiedsplit

Changelog.md−3 benchmark/Streamly/Benchmark/Data/Stream/StreamK.hs+1−1 src/Streamly/Internal/Data/Stream/StreamK.hs+6−12 src/Streamly/Internal/Data/Stream/Zip.hs+19−37

Changelog.md−3

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

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

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

1717 functionality.

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

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

20− streams.

2118

2219### Breaking type changes

2320

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

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

372372-------------------------------------------------------------------------------

373373

374374{-# INLINE zipWith #-}

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

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

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

377377

378378-------------------------------------------------------------------------------

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

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

995995-- Serial Zipping

996996------------------------------------------------------------------------------

997997

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

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

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

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

1001999--

10021000-- @since 0.1.0

10031001{-# 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

10061003zipWith f = go

10071004 where

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

10091006 let merge a ra =

10101007 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)

10131009 in foldStream (adaptState st) yield2 single2 stp my

10141010 let single1 a = merge a nil

10151011 yield1 = merge

10161012 foldStream (adaptState st) yield1 single1 stp mx

10171013

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

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

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

10201015--

10211016-- @since 0.1.0

10221017{-# 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

10251019zipWithM f m1 m2 = go m1 m2

10261020 where

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

10281022 let merge a ra =

10291023 let runIt x = foldStream st yld sng stp x

10301024 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)

10321026 in foldStream (adaptState st) yield2 single2 stp my

10331027 let single1 a = merge a nil

10341028 yield1 = merge

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

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

3535 )

3636where

3737

38−#include "inline.hs"

39−

4038import Control.Applicative (liftA2)

4139import Control.DeepSeq (NFData(..))

4240#if MIN_VERSION_deepseq(1,4,3)

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

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

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

6058

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

62−

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

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

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

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

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

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

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

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

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

6871

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

7073

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

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

8487--

8588-- @since 0.4.0

86−{-# INLINE_EARLY 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

96−zipWithMSerial f m1 m2 =

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

98−

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

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

89+{-# INLINABLE zipWithM #-}

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

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

92+

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

10194--

10295-- @

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

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

10598-- @

10699--

107100-- @since 0.1.0

108−{-# INLINE_EARLY 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

117−zipWithSerial f m1 m2 =

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

101+{-# INLINABLE zipWith #-}

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

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

119104

120105------------------------------------------------------------------------------

121106-- Parallel Zipping

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

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

221206 pure = ZipSerialM . K.repeat

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

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

224− where

225− m1_ = fromStream (toStream m1)

226− m2_ = fromStream (toStream m2)

208+ (<*>) = zipWith id

227209

228210FOLDABLE_INSTANCE(ZipSerialM)

229211TRAVERSABLE_INSTANCE(ZipSerialM)