e16a980 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use zipWith from StreamD in zipAsyncWith etc.

e16a980Harendra Kumar7 years ago

src/Streamly/Internal/Prelude.hs+31−5

compare with master

unifiedsplit

@@ -248,8 +248,8 @@ module Streamly.Internal.Prelude

248 -- ** Zipping

249 , zipWith

250 , zipWithM

251− , Z.zipAsyncWith

251+ , zipAsyncWith

252− , Z.zipAsyncWithM

252+ , zipAsyncWithM

253

254 -- ** Nested Streams

255 , concatMapM

@@ -435,7 +435,7 @@ import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))

435import Streamly.Internal.Data.Unfold.Types (Unfold)

436import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)

437-- import Streamly.Memory.Ring (Ring)

438−import Streamly.Internal.Data.SVar (MonadAsync, defState)

438+import Streamly.Internal.Data.SVar (MonadAsync, defState, adaptState)

439import Streamly.Streams.Async (mkAsync')

440import Streamly.Streams.Combinators (inspectMode, maxYields)

441import Streamly.Streams.Prelude

@@ -456,11 +456,9 @@ import qualified Streamly.Internal.Data.Fold.Types as FL

456import qualified Streamly.Streams.Prelude as P

457import qualified Streamly.Streams.StreamK as K

458import qualified Streamly.Streams.StreamD as D

459−import qualified Streamly.Streams.Zip as Z

460

459

461#ifdef USE_STREAMK_ONLY

460#ifdef USE_STREAMK_ONLY

462import qualified Streamly.Streams.StreamK as S

461import qualified Streamly.Streams.StreamK as S

463−import qualified Streamly.Streams.Zip as S

464#else

462#else

465import qualified Streamly.Streams.StreamD as S

463import qualified Streamly.Streams.StreamD as S

466#endif

464#endif

@@ -2034,6 +2032,34 @@ zipWithM f m1 m2 = fromStreamS $ S.zipWithM f (toStreamS m1) (toStreamS m2)

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

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

2035zipWith f m1 m2 = fromStreamS $ S.zipWith f (toStreamS m1) (toStreamS m2)

2033zipWith f m1 m2 = fromStreamS $ S.zipWith f (toStreamS m1) (toStreamS m2)

2036

2034

2035+------------------------------------------------------------------------------

2036+-- Parallel Zipping

2037+------------------------------------------------------------------------------

2038+

2039+-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped

2040+-- are generated concurrently.

2041+--

2042+-- @since 0.1.0

2043+{-# INLINE zipAsyncWith #-}

2044+zipAsyncWith :: (IsStream t, MonadAsync m)

2045+ => (a -> b -> c) -> t m a -> t m b -> t m c

2046+zipAsyncWith f m1 m2 = K.mkStream $ \st stp sng yld -> do

2047+ ma <- Par.mkParallel (adaptState st) m1

2048+ mb <- Par.mkParallel (adaptState st) m2

2049+ K.foldStream st stp sng yld $ zipWith f ma mb

2050+

2051+-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped

2052+-- are generated concurrently.

2053+--

2054+-- @since 0.4.0

2055+{-# INLINABLE zipAsyncWithM #-}

2056+zipAsyncWithM :: (IsStream t, MonadAsync m)

2057+ => (a -> b -> m c) -> t m a -> t m b -> t m c

2058+zipAsyncWithM f m1 m2 = K.mkStream $ \st stp sng yld -> do

2059+ ma <- Par.mkParallel (adaptState st) m1

2060+ mb <- Par.mkParallel (adaptState st) m2

2061+ K.foldStream st stp sng yld $ zipWithM f ma mb

2062+

2037------------------------------------------------------------------------------

2063------------------------------------------------------------------------------

2038-- Comparison

2064-- Comparison

2039------------------------------------------------------------------------------

2065------------------------------------------------------------------------------

src/Streamly/Streams/Zip.hs+7−38

@@ -20,12 +20,7 @@

20--

21module Streamly.Streams.Zip

22 (

23− K.zipWith

24− , K.zipWithM

25− , zipAsyncWith

26− , zipAsyncWithM

27−

28− , ZipSerialM

23+ ZipSerialM

29 , ZipSerial

24 , ZipSerial

30 , ZipStream -- deprecated

25 , ZipStream -- deprecated

31 , zipSerially

26 , zipSerially

@@ -53,10 +48,10 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,

53 readListPrecDefault)

48 readListPrecDefault)

54import Prelude hiding (map, repeat, zipWith)

49import Prelude hiding (map, repeat, zipWith)

56−import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream)

51+import Streamly.Streams.StreamK (IsStream(..), Stream)

57−import Streamly.Streams.Parallel (mkParallel)

58import Streamly.Streams.Serial (map)

52import Streamly.Streams.Serial (map)

59−import Streamly.Internal.Data.SVar (MonadAsync, adaptState)

53+import Streamly.Internal.Data.SVar (MonadAsync)

54+import Streamly.Internal.Prelude (zipWith, zipAsyncWith)

61import qualified Streamly.Streams.Prelude as P

56import qualified Streamly.Streams.Prelude as P

62import qualified Streamly.Streams.StreamK as K

57import qualified Streamly.Streams.StreamK as K

@@ -136,39 +131,12 @@ instance Monad m => Functor (ZipSerialM m) where

136

131

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

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

138 pure = ZipSerialM . K.repeat

133 pure = ZipSerialM . K.repeat

134+ {-# INLINE (<*>) #-}

139− (<*>) = K.zipWith id

135+ (<*>) = zipWith id

140

136

141FOLDABLE_INSTANCE(ZipSerialM)

137FOLDABLE_INSTANCE(ZipSerialM)

142TRAVERSABLE_INSTANCE(ZipSerialM)

138TRAVERSABLE_INSTANCE(ZipSerialM)

143

139

144−------------------------------------------------------------------------------

145−-- Parallel Zipping

146−------------------------------------------------------------------------------

147−

148−-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped

149−-- are generated concurrently.

150−--

151−-- @since 0.1.0

152−{-# INLINABLE zipAsyncWith #-}

153−zipAsyncWith :: (IsStream t, MonadAsync m)

154− => (a -> b -> c) -> t m a -> t m b -> t m c

155−zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do

156− ma <- mkParallel (adaptState st) m1

157− mb <- mkParallel (adaptState st) m2

158− foldStream st stp sng yld (K.zipWith f ma mb)

159−

160−-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped

161−-- are generated concurrently.

162−--

163−-- @since 0.4.0

164−{-# INLINABLE zipAsyncWithM #-}

165−zipAsyncWithM :: (IsStream t, MonadAsync m)

166− => (a -> b -> m c) -> t m a -> t m b -> t m c

167−zipAsyncWithM f m1 m2 = mkStream $ \st stp sng yld -> do

168− ma <- mkParallel (adaptState st) m1

169− mb <- mkParallel (adaptState st) m2

170− foldStream st stp sng yld (K.zipWithM f ma mb)

171−

172------------------------------------------------------------------------------

140------------------------------------------------------------------------------

173-- Parallely Zipping Streams

141-- Parallely Zipping Streams

174------------------------------------------------------------------------------

142------------------------------------------------------------------------------

@@ -233,4 +201,5 @@ instance Monad m => Functor (ZipAsyncM m) where

233

201

234instance MonadAsync m => Applicative (ZipAsyncM m) where

202instance MonadAsync m => Applicative (ZipAsyncM m) where

235 pure = ZipAsyncM . K.repeat

203 pure = ZipAsyncM . K.repeat

204+ {-# INLINE (<*>) #-}

236 m1 <*> m2 = zipAsyncWith id m1 m2

205 m1 <*> m2 = zipAsyncWith id m1 m2