9fe6dc1 · streamly · git.iko.soy

git.iko.soy/streamly

use direct style zipWith and zipWithM

commit9fe6dc1726436f3d9d0f99e4bd3b22ac066672f1 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 27 Jun 2018, 08:33 (8 years ago) parent6530e76

7 files changed+89−14

compare with master

unifiedsplit

Changelog.md+1 benchmark/Linear.hs+2 benchmark/LinearOps.hs+21−4 src/Streamly/Prelude.hs+22−3 src/Streamly/Streams/StreamD.hs+37−1 src/Streamly/Streams/Zip.hs+5−5 test/Prop.hs+1−1

Changelog.md+1

@@ -2,6 +2,7 @@

33### Breaking changes

5+* Signatures of `zipWithM` and `zipAsyncWithM` have changed

56* Some functions in prelude now require an additional Monad constraint for the

67 underlying type of the stream.

benchmark/Linear.hs+2

@@ -97,6 +97,7 @@ main = do

9797 , benchIO "dropWhileM-true" Ops.dropWhileMTrue

9898 ]

9999 , benchIO "zip" $ Ops.zip

100+ , benchIO "zipM" $ Ops.zipM

100101 , bgroup "compose"

101102 [ benchIO "mapM" Ops.composeMapM

102103 , benchIO "map-with-all-in-filter" Ops.composeMapAllInFilter

@@ -151,5 +152,6 @@ main = do

151152 , benchIO "mapM" $ Ops.mapM parallely

152153 -- Zip has only one parallel flavor

153154 , benchIO "zip" $ Ops.zipAsync

155+ , benchIO "zipM" $ Ops.zipAsyncM

154156 ]

155157 ]

benchmark/LinearOps.hs+21−4

@@ -9,6 +9,7 @@

1010module LinearOps where

1111

12+import Data.Maybe (fromJust)

1213import Prelude

1314 (Monad, Int, (+), ($), (.), return, fmap, even, (>), (<=),

1415 subtract, undefined, Maybe(..), odd, Bool, not)

@@ -42,13 +43,14 @@ maxValue = value + 1000

4243{-# INLINE dropWhileTrue #-}

4344{-# INLINE dropWhileMTrue #-}

4445{-# INLINE zip #-}

46+{-# INLINE zipM #-}

4547{-# INLINE concat #-}

4648{-# INLINE composeAllInFilters #-}

4749{-# INLINE composeAllOutFilters #-}

4850{-# INLINE composeMapAllInFilter #-}

4951uncons, nullHeadTail, scan, mapM_, map, fmap, mapMaybe, filterEven, filterAllOut,

5052 filterAllIn, takeOne, takeAll, takeWhileTrue, takeWhileMTrue, dropAll,

51− dropWhileTrue, dropWhileMTrue, zip,

53+ dropWhileTrue, dropWhileMTrue, zip, zipM,

5254 concat, composeAllInFilters, composeAllOutFilters,

5355 composeMapAllInFilter

5456 :: Monad m

@@ -56,8 +58,9 @@ uncons, nullHeadTail, scan, mapM_, map, fmap, mapMaybe, filterEven, filterAllOut

5658

5759{-# INLINE composeMapM #-}

5860{-# INLINE zipAsync #-}

61+{-# INLINE zipAsyncM #-}

5962{-# INLINE mapMaybeM #-}

60−composeMapM, zipAsync, mapMaybeM :: S.MonadAsync m => Stream m Int -> m ()

63+composeMapM, zipAsync, zipAsyncM, mapMaybeM :: S.MonadAsync m => Stream m Int -> m ()

6164

6265{-# INLINE toList #-}

6366{-# INLINE foldr #-}

@@ -228,8 +231,22 @@ dropWhileMTrue = transform . S.dropWhileM (return . (<= maxValue))

228231-- Zipping and concat

229232-------------------------------------------------------------------------------

230233

231−zip src = transform $ (S.zipWith (,) src src)

232−zipAsync src = transform $ (S.zipAsyncWith (,) src src)

234+zip src = do

235+ r <- S.tail src

236+ let src1 = fromJust r

237+ transform $ (S.zipWith (,) src src1)

238+zipM src = do

239+ r <- S.tail src

240+ let src1 = fromJust r

241+ transform $ (S.zipWithM (\a b -> return (a,b)) src src1)

242+zipAsync src = do

243+ r <- S.tail src

244+ let src1 = fromJust r

245+ transform $ (S.zipAsyncWith (,) src src1)

246+zipAsyncM src = do

247+ r <- S.tail src

248+ let src1 = fromJust r

249+ transform $ (S.zipAsyncWithM (\a b -> return (a,b)) src src1)

233250concat _n = return ()

234251

235252-------------------------------------------------------------------------------

src/Streamly/Prelude.hs+22−3

@@ -141,8 +141,8 @@ module Streamly.Prelude

141141 -- * Zipping

142142 , zipWith

143143 , zipWithM

144− , zipAsyncWith

145− , zipAsyncWithM

144+ , Z.zipAsyncWith

145+ , Z.zipAsyncWithM

146146

147147 -- * Deprecated

148148 , K.once

@@ -166,13 +166,14 @@ import qualified System.IO as IO

166166import Streamly.SVar (MonadAsync)

167167import Streamly.Streams.StreamK (IsStream(..))

168168import Streamly.Streams.Serial (SerialT)

169−import Streamly.Streams.Zip (zipWith, zipWithM, zipAsyncWith, zipAsyncWithM)

170169

171170import qualified Streamly.Streams.StreamK as K

172171import qualified Streamly.Streams.StreamD as D

172+import qualified Streamly.Streams.Zip as Z

173173

174174#ifdef USE_STREAMK_ONLY

175175import qualified Streamly.Streams.StreamK as S

176+import qualified Streamly.Streams.Zip as S

176177#else

177178import qualified Streamly.Streams.StreamD as S

178179#endif

@@ -801,3 +802,21 @@ reverse m = fromStream $ go K.nil (toStream m)

801802 single a = runIt $ a `K.cons` rev

802803 yieldk a r = runIt $ go (a `K.cons` rev) r

803804 in K.unStream rest Nothing stop single yieldk

805+

806+------------------------------------------------------------------------------

807+-- Zipping

808+------------------------------------------------------------------------------

809+

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

811+--

812+-- @since 0.4.0

813+{-# INLINABLE zipWithM #-}

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

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

816+

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

818+--

819+-- @since 0.1.0

820+{-# INLINABLE zipWith #-}

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

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

src/Streamly/Streams/StreamD.hs+37−1

@@ -114,6 +114,10 @@ module Streamly.Streams.StreamD

114114 -- ** Map and Filter

115115 , mapMaybe

116116 , mapMaybeM

117+

118+ -- * Zipping

119+ , zipWith

120+ , zipWithM

117121 )

118122where

119123

@@ -122,7 +126,7 @@ import GHC.Types ( SPEC(..) )

122126import Prelude

123127 hiding (map, mapM, mapM_, repeat, foldr, last, take, filter,

124128 takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,

125− notElem, null, head, tail)

129+ notElem, null, head, tail, zipWith)

126130

127131import Streamly.SVar (MonadAsync)

128132import qualified Streamly.Streams.StreamK as K

@@ -624,6 +628,7 @@ map f = mapM (return . f)

624628-- Transformation by Map and Filter

625629------------------------------------------------------------------------------

626630

631+-- XXX Will this always fuse properly?

627632{-# INLINE_NORMAL mapMaybe #-}

628633mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b

629634mapMaybe f = fmap fromJust . filter isJust . map f

@@ -636,6 +641,37 @@ mapMaybeM f = fmap fromJust . filter isJust . mapM f

636641-- Instances

637642------------------------------------------------------------------------------

638643

644+{-# INLINE_NORMAL zipWithM #-}

645+zipWithM :: Monad m

646+ => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c

647+zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)

648+ where

649+ {-# INLINE_LATE step #-}

650+ step (sa, sb, Nothing) = do

651+ r <- stepa sa

652+ case r of

653+ Yield x sa' -> step (sa', sb, Just x)

654+ Stop -> return Stop

655+

656+ step (sa, sb, Just x) = do

657+ r <- stepb sb

658+ case r of

659+ Yield y sb' -> do

660+ z <- f x y

661+ return $ Yield z (sa, sb', Nothing)

662+ Stop -> return Stop

663+

664+{-# RULES "zipWithM xs xs"

665+ forall f xs. zipWithM f xs xs = mapM (\x -> f x x) xs #-}

666+

667+{-# INLINE zipWith #-}

668+zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

669+zipWith f = zipWithM (\a b -> return (f a b))

670+

671+------------------------------------------------------------------------------

672+-- Instances

673+------------------------------------------------------------------------------

674+

639675instance Monad m => Functor (Stream m) where

640676 {-# INLINE fmap #-}

641677 fmap = map

src/Streamly/Streams/Zip.hs+5−5

@@ -77,14 +77,14 @@ zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)

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

7878--

7979-- @since 0.1.0

80−zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c

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

8181zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

8282 where

8383 go mx my = Stream $ \_ stp sng yld -> do

8484 let merge a ra =

8585 let runIt x = unStream x Nothing stp sng yld

86− single2 b = runIt $ toStream (f a b)

87− yield2 b rb = runIt $ toStream (f a b) <> go ra rb

86+ single2 b = f a b >>= sng

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

8888 in unStream my Nothing stp single2 yield2

8989 let single1 a = merge a K.nil

9090 yield1 a ra = merge a ra

@@ -177,9 +177,9 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

177177-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

178178-- concurrently) using a monadic zipping function.

179179--

180−-- @since 0.1.0

180+-- @since 0.4.0

181181zipAsyncWithM :: (IsStream t, MonadAsync m)

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

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

183183zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

184184 ma <- mkAsync m1

185185 mb <- mkAsync m2

test/Prop.hs+1−1

@@ -525,7 +525,7 @@ zipApplicative constr t eq (a, b) = withMaxSuccess maxTestCount $

525525 equals eq stream3 list

526526

527527zipMonadic

528− :: (IsStream t, Monad (t IO))

528+ :: IsStream t

529529 => ([Int] -> t IO Int)

530530 -> (t IO (Int, Int) -> SerialT IO (Int, Int))

531531 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)