1dcad0f · streamly · git.iko.soy

git.iko.soy/streamly

improve perf of functor and applicative instances

Functor instances of various stream types are directly expressed in terms of StreamD map impl. Applicative instances now use concatMap instead of using "ap" from Monad. Rewrite rules to specialize unfoldrM to WSerial and ZipSerial as well so that it performs well for those. This was causing benchmarks for these to be much slower beacause of unfoldrM used in the benchmarks being slow. For this, I had to move the zipWith* impls into the Zip module so that we can import the ZipSerial type in Prelude.

commit1dcad0f560a3b5fb012171f46f5796be46246cb2 authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 16 Dec 2019, 16:54 (7 years ago) parent634df53

13 files changed+226−119

compare with master

unifiedsplit

benchmark/Linear.hs+15

@@ -440,4 +440,19 @@ main =

440440 , benchIOSrc serially "dropWhileTrue" Ops.iterateDropWhileTrue

441441 ]

442442 ]

443+ , bgroup "wSerially"

444+ [ bgroup "transformation"

445+ [ benchIOSink "fmap" $ Ops.fmap' wSerially 1

446+ ]

447+ ]

448+ , bgroup "zipSerially"

449+ [ bgroup "transformation"

450+ [ benchIOSink "fmap" $ Ops.fmap' zipSerially 1

451+ ]

452+ ]

453+ , bgroup "zipAsyncly"

454+ [ bgroup "transformation"

455+ [ benchIOSink "fmap" $ Ops.fmap' zipAsyncly 1

456+ ]

457+ ]

443458 ]

benchmark/Nested.hs+8

@@ -93,4 +93,12 @@ main =

9393 , benchIO "filterSome" $ Ops.filterSome parallely

9494 , benchIO "breakAfterSome" $ Ops.breakAfterSome parallely

9595 ]

96+

97+ , bgroup "zipSerially"

98+ [ benchIO "toNullAp" $ Ops.toNullAp zipSerially

99+ ]

100+

101+ , bgroup "zipAsyncly"

102+ [ benchIO "toNullAp" $ Ops.toNullAp zipAsyncly

103+ ]

96104 ]

benchmark/NestedOps.hs+1−1

@@ -70,7 +70,7 @@ runToList = S.toList

7070

7171{-# INLINE toNullAp #-}

7272toNullAp

73− :: (S.IsStream t, S.MonadAsync m, Monad (t m))

73+ :: (S.IsStream t, S.MonadAsync m, Applicative (t m))

7474 => (t m Int -> S.SerialT m Int) -> Int -> m ()

7575toNullAp t start = runStream . t $

7676 (+) <$> source start nestedCount2 <*> source start nestedCount2

src/Streamly/Internal/Data/Stream/Ahead.hs+14−7

@@ -28,7 +28,7 @@ where

2828

2929import Control.Concurrent.MVar (putMVar, takeMVar)

3030import Control.Exception (assert)

31−import Control.Monad (ap, void, when)

31+import Control.Monad (void, when)

3232import Control.Monad.Base (MonadBase(..), liftBaseDefault)

3333import Control.Monad.Catch (MonadThrow, throwM)

3434-- import Control.Monad.Error.Class (MonadError(..))

@@ -47,12 +47,12 @@ import GHC.Exts (inline)

4747import qualified Data.Heap as H

4848

4949import Streamly.Internal.Data.Stream.SVar (fromSVar)

50−import Streamly.Internal.Data.Stream.Serial (map)

5150import Streamly.Internal.Data.SVar

5251import Streamly.Internal.Data.Stream.StreamK

5352 (IsStream(..), Stream, mkStream, foldStream, foldStreamShared,

5453 foldStreamSVar)

5554import qualified Streamly.Internal.Data.Stream.StreamK as K

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

5656

5757import Prelude hiding (map)

5858

@@ -683,15 +683,22 @@ concatMapAhead :: MonadAsync m => (a -> AheadT m b) -> AheadT m a -> AheadT m b

683683concatMapAhead f m = fromStream $

684684 K.concatMapBy ahead (\a -> K.adapt $ f a) (K.adapt m)

685685

686−instance MonadAsync m => Monad (AheadT m) where

687− return = pure

688− {-# INLINE (>>=) #-}

689− (>>=) = flip concatMapAhead

686+{-# INLINE apAhead #-}

687+apAhead :: MonadAsync m => AheadT m (a -> b) -> AheadT m a -> AheadT m b

688+apAhead (AheadT m1) (AheadT m2) =

689+ let f x1 = K.concatMapBy ahead (pure . x1) m2

690+ in AheadT $ K.concatMapBy ahead f m1

690691

691692instance (Monad m, MonadAsync m) => Applicative (AheadT m) where

693+ {-# INLINE pure #-}

692694 pure = AheadT . K.yield

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

694− (<*>) = ap

696+ (<*>) = apAhead

697+

698+instance MonadAsync m => Monad (AheadT m) where

699+ return = pure

700+ {-# INLINE (>>=) #-}

701+ (>>=) = flip concatMapAhead

695702

696703------------------------------------------------------------------------------

697704-- Other instances

src/Streamly/Internal/Data/Stream/Async.hs+35−20

@@ -37,7 +37,6 @@ module Streamly.Internal.Data.Stream.Async

3737where

3838

3939import Control.Concurrent (myThreadId)

40−import Control.Monad (ap)

4140import Control.Monad.Base (MonadBase(..), liftBaseDefault)

4241import Control.Monad.Catch (MonadThrow, throwM)

4342import Control.Concurrent.MVar (newEmptyMVar)

@@ -58,12 +57,12 @@ import qualified Data.Set as S

5857

5958import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)

6059import Streamly.Internal.Data.Stream.SVar (fromSVar)

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

6260import Streamly.Internal.Data.SVar

6361import Streamly.Internal.Data.Stream.StreamK

6462 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared,

6563 foldStreamSVar)

6664import qualified Streamly.Internal.Data.Stream.StreamK as K

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

6766

6867#include "Instances.hs"

6968

@@ -690,6 +689,23 @@ instance MonadAsync m => Monoid (AsyncT m a) where

690689 mempty = K.nil

691690 mappend = (<>)

692691

692+------------------------------------------------------------------------------

693+-- Applicative

694+------------------------------------------------------------------------------

695+

696+{-# INLINE apAsync #-}

697+{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}

698+apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b

699+apAsync (AsyncT m1) (AsyncT m2) =

700+ let f x1 = K.concatMapBy async (pure . x1) m2

701+ in AsyncT $ K.concatMapBy async f m1

702+

703+instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where

704+ {-# INLINE pure #-}

705+ pure = AsyncT . K.yield

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

707+ (<*>) = apAsync

708+

693709------------------------------------------------------------------------------

694710-- Monad

695711------------------------------------------------------------------------------

@@ -707,15 +723,6 @@ instance MonadAsync m => Monad (AsyncT m) where

707723 return = pure

708724 (>>=) = bindAsync

709725

710−{-# INLINE apAsync #-}

711−{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}

712−apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b

713−apAsync mf m = ap (adapt mf) (adapt m)

714−

715−instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where

716− pure = AsyncT . K.yield

717− (<*>) = apAsync

718−

719726------------------------------------------------------------------------------

720727-- Other instances

721728------------------------------------------------------------------------------

@@ -831,6 +838,23 @@ instance MonadAsync m => Monoid (WAsyncT m a) where

831838 mempty = K.nil

832839 mappend = (<>)

833840

841+------------------------------------------------------------------------------

842+-- Applicative

843+------------------------------------------------------------------------------

844+

845+{-# INLINE apWAsync #-}

846+{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}

847+apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b

848+apWAsync (WAsyncT m1) (WAsyncT m2) =

849+ let f x1 = K.concatMapBy wAsync (pure . x1) m2

850+ in WAsyncT $ K.concatMapBy wAsync f m1

851+

852+-- GHC: if we specify arguments in the definition of (<*>) we see a significant

853+-- performance degradation (~2x).

854+instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where

855+ pure = WAsyncT . K.yield

856+ (<*>) = apWAsync

857+

834858------------------------------------------------------------------------------

835859-- Monad

836860------------------------------------------------------------------------------

@@ -848,15 +872,6 @@ instance MonadAsync m => Monad (WAsyncT m) where

848872 return = pure

849873 (>>=) = bindWAsync

850874

851−{-# INLINE apWAsync #-}

852−{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}

853−apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b

854−apWAsync mf m = ap (adapt mf) (adapt m)

855−

856−instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where

857− pure = WAsyncT . K.yield

858− (<*>) = apWAsync

859−

860875------------------------------------------------------------------------------

861876-- Other instances

862877------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/Instances.hs+2−8

@@ -7,16 +7,10 @@

88#define MONADPARALLEL , MonadAsync m

10−#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \

11−instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \

12− {-# INLINE pure #-}; \

13− pure = STREAM . K.yield; \

14− {-# INLINE (<*>) #-}; \

15− (<*>) = ap }

16−

1710#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

1811instance Monad m => Functor (STREAM m) where { \

19− fmap = map }; \

12+ {-# INLINE fmap #-}; \

13+ fmap f (STREAM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m }; \

2014 \

2115instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\

2216 liftBase = liftBaseDefault }; \

src/Streamly/Internal/Data/Stream/Parallel.hs+18−3

@@ -36,7 +36,7 @@ where

3636

3737import Control.Concurrent (myThreadId, takeMVar)

3838import Control.Exception (SomeException(..))

39−import Control.Monad (ap, when)

39+import Control.Monad (when)

4040import Control.Monad.Base (MonadBase(..), liftBaseDefault)

4141import Control.Monad.Catch (MonadThrow, throwM)

4242-- import Control.Monad.Error.Class (MonadError(..))

@@ -56,7 +56,6 @@ import qualified Data.Set as Set

5656

5757import Streamly.Internal.Data.Stream.SVar

5858 (fromSVar, fromProducer, fromConsumer)

59−import Streamly.Internal.Data.Stream.Serial (map)

6059import Streamly.Internal.Data.Stream.StreamK

6160 (IsStream(..), Stream, mkStream, foldStream, foldStreamShared, adapt)

6261

@@ -551,6 +550,23 @@ instance MonadAsync m => Monoid (ParallelT m a) where

551550 mempty = K.nil

552551 mappend = (<>)

553552

553+------------------------------------------------------------------------------

554+-- Applicative

555+------------------------------------------------------------------------------

556+

557+{-# INLINE apParallel #-}

558+{-# SPECIALIZE apParallel :: ParallelT IO (a -> b) -> ParallelT IO a -> ParallelT IO b #-}

559+apParallel :: MonadAsync m => ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b

560+apParallel (ParallelT m1) (ParallelT m2) =

561+ let f x1 = K.concatMapBy parallel (pure . x1) m2

562+ in ParallelT $ K.concatMapBy parallel f m1

563+

564+instance (Monad m, MonadAsync m) => Applicative (ParallelT m) where

565+ {-# INLINE pure #-}

566+ pure = ParallelT . K.yield

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

568+ (<*>) = apParallel

569+

554570------------------------------------------------------------------------------

555571-- Monad

556572------------------------------------------------------------------------------

@@ -568,5 +584,4 @@ instance MonadAsync m => Monad (ParallelT m) where

568584-- Other instances

569585------------------------------------------------------------------------------

570586

571−MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)

572587MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

src/Streamly/Internal/Data/Stream/Serial.hs+28−6

@@ -54,7 +54,6 @@ import Control.DeepSeq (NFData(..))

5454#if MIN_VERSION_deepseq(1,4,3)

5555import Control.DeepSeq (NFData1(..))

5656#endif

57−import Control.Monad (ap)

5857import Control.Monad.Base (MonadBase(..), liftBaseDefault)

5958import Control.Monad.Catch (MonadThrow, throwM)

6059-- import Control.Monad.Error.Class (MonadError(..))

@@ -77,6 +76,7 @@ import Streamly.Internal.Data.Stream.StreamK (IsStream(..), adapt, Stream, mkStr

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

7877import qualified Streamly.Internal.Data.Stream.StreamK as K

7978import qualified Streamly.Internal.Data.Stream.StreamD as D

79+import qualified Streamly.Internal.Data.Unfold as UF

8080

8181#include "Instances.hs"

8282#include "inline.hs"

@@ -190,9 +190,9 @@ instance Monad m => Monad (SerialT m) where

190190-- Other instances

191191------------------------------------------------------------------------------

192192

193−{-# INLINE_EARLY mapM #-}

193+{-# INLINE mapM #-}

194194mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b

195−mapM f m = fromStream $ D.toStreamK $ D.mapM f $ D.fromStreamK (toStream m)

195+mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m

196196

197197-- |

198198-- @

@@ -211,7 +211,18 @@ mapM f m = fromStream $ D.toStreamK $ D.mapM f $ D.fromStreamK (toStream m)

211211map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b

212212map f = mapM (return . f)

213213

214−MONAD_APPLICATIVE_INSTANCE(SerialT,)

214+{-# INLINE apSerial #-}

215+apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b

216+apSerial (SerialT m1) (SerialT m2) =

217+ let f x1 = D.concatMapU (UF.singleton (pure . x1)) (D.toStreamD m2)

218+ in D.fromStreamD $ D.concatMap f (D.toStreamD m1)

219+

220+instance Monad m => Applicative (SerialT m) where

221+ {-# INLINE pure #-}

222+ pure = SerialT . K.yield

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

224+ (<*>) = apSerial

225+

215226MONAD_COMMON_INSTANCES(SerialT,)

216227LIST_INSTANCES(SerialT)

217228NFDATA1_INSTANCE(SerialT)

@@ -383,6 +394,18 @@ instance Monoid (WSerialT m a) where

383394 mempty = K.nil

384395 mappend = (<>)

385396

397+{-# INLINE apWSerial #-}

398+apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b

399+apWSerial (WSerialT m1) (WSerialT m2) =

400+ let f x1 = K.concatMapBy wSerial (pure . x1) m2

401+ in WSerialT $ K.concatMapBy wSerial f m1

402+

403+instance Monad m => Applicative (WSerialT m) where

404+ {-# INLINE pure #-}

405+ pure = WSerialT . K.yield

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

407+ (<*>) = apWSerial

408+

386409------------------------------------------------------------------------------

387410-- Monad

388411------------------------------------------------------------------------------

@@ -396,7 +419,6 @@ instance Monad m => Monad (WSerialT m) where

396419-- Other instances

397420------------------------------------------------------------------------------

398421

399−MONAD_APPLICATIVE_INSTANCE(WSerialT,)

400422MONAD_COMMON_INSTANCES(WSerialT,)

401423LIST_INSTANCES(WSerialT)

402424NFDATA1_INSTANCE(WSerialT)

@@ -429,5 +451,5 @@ TRAVERSABLE_INSTANCE(WSerialT)

429451-- /Internal/

430452--

431453{-# INLINE unfoldrM #-}

432−unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> SerialT m a

454+unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a

433455unfoldrM step seed = D.fromStreamD (D.unfoldrM step seed)

src/Streamly/Internal/Data/Stream/StreamD/Type.hs+7−2

@@ -68,7 +68,7 @@ module Streamly.Internal.Data.Stream.StreamD.Type

6868where

6969

7070import Control.Applicative (liftA2)

71−import Control.Monad (ap, when)

71+import Control.Monad (when)

7272import Control.Monad.Catch (MonadThrow, throwM)

7373import Control.Monad.Trans (lift, MonadTrans)

7474import Data.Functor.Identity (Identity(..))

@@ -242,7 +242,12 @@ instance Monad m => Applicative (Stream m) where

242242 {-# INLINE pure #-}

243243 pure = yield

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

245− (<*>) = ap

245+ m1 <*> m2 =

246+ -- XXX concatMapU is faster but it would require us to merge

247+ -- Unfold/Types.hs into this file so that we can use UF.singleton here.

248+ -- let f x1 = concatMapU (UF.singleton (pure . x1)) m2

249+ let f x1 = concatMap (\x2 -> pure (x1 x2)) m2

250+ in concatMap f m1

246251

247252-- NOTE: even though concatMap for StreamD is 4x faster compared to StreamK,

248253-- the monad instance does not seem to be significantly faster.

src/Streamly/Internal/Data/Stream/StreamK/Type.hs+2−3

@@ -789,8 +789,8 @@ foldrMKWith f step acc g = go g

789789-- @since 0.2.0

790790{-# INLINE serial #-}

791791serial :: IsStream t => t m a -> t m a -> t m a

792−serial xs ys = augmentS (\c n -> foldrS c n xs) ys

793−{-

792+-- XXX This doubles the time of toNullAp benchmark, may not be fusing properly

793+-- serial xs ys = augmentS (\c n -> foldrS c n xs) ys

794794serial m1 m2 = go m1

795795 where

796796 go m = mkStream $ \st yld sng stp ->

@@ -798,7 +798,6 @@ serial m1 m2 = go m1

798798 single a = yld a m2

799799 yieldk a r = yld a (go r)

800800 in foldStream st yieldk single stop m

801−-}

802801

803802-- join/merge/append streams depending on consM

804803{-# INLINE conjoin #-}

src/Streamly/Internal/Data/Stream/Zip.hs+76−8

@@ -22,14 +22,21 @@ module Streamly.Internal.Data.Stream.Zip

2222 (

2323 ZipSerialM

2424 , ZipSerial

25− , ZipStream -- deprecated

2625 , zipSerially

27− , zipping -- deprecated

2826

2927 , ZipAsyncM

3028 , ZipAsync

3129 , zipAsyncly

32− , zippingAsync -- deprecated

30+

31+ , zipWith

32+ , zipWithM

33+ , zipAsyncWith

34+ , zipAsyncWithM

35+

36+ -- * Deprecated

37+ , ZipStream

38+ , zipping

39+ , zippingAsync

3340 )

3441where

3542

@@ -49,15 +56,74 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,

4956import Prelude hiding (map, repeat, zipWith)

5057

5158import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream)

52−import Streamly.Internal.Data.Stream.Serial (map)

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

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

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

5560

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

5762import qualified Streamly.Internal.Data.Stream.StreamK as K

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

64+

65+#ifdef USE_STREAMK_ONLY

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

67+#else

68+import qualified Streamly.Internal.Data.Stream.StreamD as S

69+#endif

5870

5971#include "Instances.hs"

6072

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

74+--

75+-- @since 0.4.0

76+{-# INLINABLE zipWithM #-}

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

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

79+

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

81+--

82+-- @

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

84+-- [5,7,9]

85+-- @

86+--

87+-- @since 0.1.0

88+{-# INLINABLE zipWith #-}

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

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

91+

92+------------------------------------------------------------------------------

93+-- Parallel Zipping

94+------------------------------------------------------------------------------

95+

96+-- The CPS version and the direct version of zipAsyncWithM below seem to have

97+-- identical performance. However, we need to use the StreamD version of

98+-- mkParallel which uses a direct implementation of fromSVar. In comparison to

99+-- CPS version of fromSVar the direct version gives a 2x improvement.

100+

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

102+-- are generated concurrently.

103+--

104+-- @since 0.4.0

105+{-# INLINE zipAsyncWithM #-}

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

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

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

109+ ma <- D.mkParallel (adaptState st) (D.toStreamD m1)

110+ mb <- D.mkParallel (adaptState st) (D.toStreamD m2)

111+ K.foldStream st stp sng yld $ zipWithM f (D.fromStreamD ma)

112+ (D.fromStreamD mb)

113+{-

114+zipAsyncWithM f m1 m2 =

115+ fromStreamD $ D.zipAsyncWithM f (toStreamD m1) (toStreamD m2)

116+-}

117+

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

119+-- are generated concurrently.

120+--

121+-- @since 0.1.0

122+{-# INLINE zipAsyncWith #-}

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

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

125+zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))

126+

61127------------------------------------------------------------------------------

62128-- Serially Zipping Streams

63129------------------------------------------------------------------------------

@@ -127,7 +193,8 @@ LIST_INSTANCES(ZipSerialM)

127193NFDATA1_INSTANCE(ZipSerialM)

128194

129195instance Monad m => Functor (ZipSerialM m) where

130− fmap = map

196+ {-# INLINE fmap #-}

197+ fmap f (ZipSerialM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m

131198

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

133200 pure = ZipSerialM . K.repeat

@@ -197,7 +264,8 @@ instance IsStream ZipAsyncM where

197264 (|:) = consMZipAsync

198265

199266instance Monad m => Functor (ZipAsyncM m) where

200− fmap = map

267+ {-# INLINE fmap #-}

268+ fmap f (ZipAsyncM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m

201269

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

203271 pure = ZipAsyncM . K.repeat

src/Streamly/Internal/Prelude.hs+18−59

@@ -253,10 +253,10 @@ module Streamly.Internal.Prelude

253253 , mergeAsyncByM

254254

255255 -- ** Zipping

256− , zipWith

257− , zipWithM

258− , zipAsyncWith

259− , zipAsyncWithM

256+ , Z.zipWith

257+ , Z.zipWithM

258+ , Z.zipAsyncWith

259+ , Z.zipAsyncWithM

260260

261261 -- ** Nested Streams

262262 , concatMapM

@@ -446,14 +446,15 @@ import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))

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

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

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

449−import Streamly.Internal.Data.SVar (MonadAsync, State, defState, adaptState)

449+import Streamly.Internal.Data.SVar (MonadAsync, State, defState)

450450import Streamly.Internal.Data.Stream.Async (mkAsync')

451451import Streamly.Internal.Data.Stream.Combinators (inspectMode, maxYields)

452452import Streamly.Internal.Data.Stream.Prelude

453453 (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)

454454import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)

455455import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM))

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

456+import Streamly.Internal.Data.Stream.Serial (SerialT, WSerialT)

457+import Streamly.Internal.Data.Stream.Zip (ZipSerialM)

457458import Streamly.Internal.Data.Pipe.Types (Pipe (..))

458459import Streamly.Internal.Data.Time.Units

459460 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,

@@ -476,6 +477,7 @@ import qualified Streamly.Internal.Data.Stream.StreamD as S

476477

477478import qualified Streamly.Internal.Data.Stream.Serial as Serial

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

480+import qualified Streamly.Internal.Data.Stream.Zip as Z

479481

480482------------------------------------------------------------------------------

481483-- Deconstruction

@@ -571,6 +573,16 @@ unfoldrM = K.unfoldrM

571573unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a

572574unfoldrMSerial = Serial.unfoldrM

573575

576+{-# RULES "unfoldrM wSerial" unfoldrM = unfoldrMWSerial #-}

577+{-# INLINE_EARLY unfoldrMWSerial #-}

578+unfoldrMWSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> WSerialT m a

579+unfoldrMWSerial = Serial.unfoldrM

580+

581+{-# RULES "unfoldrM zipSerial" unfoldrM = unfoldrMZipSerial #-}

582+{-# INLINE_EARLY unfoldrMZipSerial #-}

583+unfoldrMZipSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> ZipSerialM m a

584+unfoldrMZipSerial = Serial.unfoldrM

585+

574586-- | Convert an 'Unfold' into a stream by supplying it an input seed.

575587--

576588-- >>> unfold UF.replicateM 10 (putStrLn "hello")

@@ -2128,59 +2140,6 @@ indexed = fromStreamD . D.indexed . toStreamD

21282140indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)

21292141indexedR n = fromStreamD . D.indexedR n . toStreamD

21302142

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

2132−--

2133−-- @since 0.4.0

2134−{-# INLINABLE zipWithM #-}

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

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

2137−

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

2139−--

2140−-- @

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

2142−-- [5,7,9]

2143−-- @

2144−--

2145−-- @since 0.1.0

2146−{-# INLINABLE zipWith #-}

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

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

2149−

2150−------------------------------------------------------------------------------

2151−-- Parallel Zipping

2152−------------------------------------------------------------------------------

2153−

2154−-- The CPS version and the direct version of zipAsyncWithM below seem to have

2155−-- identical performane. However, we need to use the StreamD version of

2156−-- mkParallel which uses a direct implementation of fromSVar. In comparison to

2157−-- CPS version of fromSVar the direct version gives a 2x improvement.

2158−

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

2160−-- are generated concurrently.

2161−--

2162−-- @since 0.4.0

2163−{-# INLINE zipAsyncWithM #-}

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

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

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

2167− ma <- mkParallel (adaptState st) m1

2168− mb <- mkParallel (adaptState st) m2

2169− K.foldStream st stp sng yld $ zipWithM f ma mb

2170−{-

2171−zipAsyncWithM f m1 m2 =

2172− fromStreamD $ D.zipAsyncWithM f (toStreamD m1) (toStreamD m2)

2173−-}

2174−

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

2176−-- are generated concurrently.

2177−--

2178−-- @since 0.1.0

2179−{-# INLINE zipAsyncWith #-}

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

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

2182−zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))

2183−

21842143------------------------------------------------------------------------------

21852144-- Comparison

21862145------------------------------------------------------------------------------

streamly.cabal+2−2

@@ -742,8 +742,8 @@ benchmark concurrent

742742benchmark base

743743 import: bench-options

744744 type: exitcode-stdio-1.0

745− include-dirs: src/Streamly/Internal/Data/Time

746− src

745+ include-dirs: src/Streamly/Internal/Data/Time

746+ , src

747747 if os(windows)

748748 c-sources: src/Streamly/Internal/Data/Time/Windows.c

749749 if os(darwin)