1dcad0f · streamly · git.iko.soy

benchmark/Linear.hs+15

compare with master

unifiedsplit

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

440 , benchIOSrc serially "dropWhileTrue" Ops.iterateDropWhileTrue

441 ]

442 ]

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+ ]

443 ]

458 ]

benchmark/Nested.hs+8

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

93 , benchIO "filterSome" $ Ops.filterSome parallely

94 , benchIO "breakAfterSome" $ Ops.breakAfterSome parallely

95 ]

96+

97+ , bgroup "zipSerially"

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

99+ ]

100+

101+ , bgroup "zipAsyncly"

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

103+ ]

96 ]

104 ]

benchmark/NestedOps.hs+1−1

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

70

71{-# INLINE toNullAp #-}

72toNullAp

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

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

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

75toNullAp t start = runStream . t $

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

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

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

28

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

30import Control.Exception (assert)

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

31+import Control.Monad (void, when)

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

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

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

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

47import qualified Data.Heap as H

48

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

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

51import Streamly.Internal.Data.SVar

50import Streamly.Internal.Data.SVar

52import Streamly.Internal.Data.Stream.StreamK

51import Streamly.Internal.Data.Stream.StreamK

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

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

54 foldStreamSVar)

53 foldStreamSVar)

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

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

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

56

57import Prelude hiding (map)

58

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

683concatMapAhead f m = fromStream $

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

685

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

687− return = pure

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

686+{-# INLINE apAhead #-}

689− (>>=) = flip concatMapAhead

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

690

691

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

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

693+ {-# INLINE pure #-}

692 pure = AheadT . K.yield

694 pure = AheadT . K.yield

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

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

694− (<*>) = ap

696+ (<*>) = apAhead

697+

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

699+ return = pure

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

701+ (>>=) = flip concatMapAhead

695

702

696------------------------------------------------------------------------------

703------------------------------------------------------------------------------

697-- Other instances

704-- Other instances

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

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

37where

38

39import Control.Concurrent (myThreadId)

40−import Control.Monad (ap)

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

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

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

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

43import Control.Concurrent.MVar (newEmptyMVar)

42import Control.Concurrent.MVar (newEmptyMVar)

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

58

57

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

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

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

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

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

62import Streamly.Internal.Data.SVar

60import Streamly.Internal.Data.SVar

63import Streamly.Internal.Data.Stream.StreamK

61import Streamly.Internal.Data.Stream.StreamK

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

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

65 foldStreamSVar)

63 foldStreamSVar)

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

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

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

67

66

68#include "Instances.hs"

67#include "Instances.hs"

69

68

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

690 mempty = K.nil

689 mempty = K.nil

691 mappend = (<>)

690 mappend = (<>)

692

691

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+

693------------------------------------------------------------------------------

709------------------------------------------------------------------------------

694-- Monad

710-- Monad

695------------------------------------------------------------------------------

711------------------------------------------------------------------------------

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

707 return = pure

723 return = pure

708 (>>=) = bindAsync

724 (>>=) = bindAsync

709

725

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−

719------------------------------------------------------------------------------

726------------------------------------------------------------------------------

720-- Other instances

727-- Other instances

721------------------------------------------------------------------------------

728------------------------------------------------------------------------------

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

831 mempty = K.nil

838 mempty = K.nil

832 mappend = (<>)

839 mappend = (<>)

833

840

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+

834------------------------------------------------------------------------------

858------------------------------------------------------------------------------

835-- Monad

859-- Monad

836------------------------------------------------------------------------------

860------------------------------------------------------------------------------

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

848 return = pure

872 return = pure

849 (>>=) = bindWAsync

873 (>>=) = bindWAsync

850

874

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−

860------------------------------------------------------------------------------

875------------------------------------------------------------------------------

861-- Other instances

876-- Other instances

862------------------------------------------------------------------------------

877------------------------------------------------------------------------------

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

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

7

8#define MONADPARALLEL , MonadAsync m

9

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−

17#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

10#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

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

11instance 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 }; \

20 \

14 \

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

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

22 liftBase = liftBaseDefault }; \

16 liftBase = liftBaseDefault }; \

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

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

36

37import Control.Concurrent (myThreadId, takeMVar)

38import Control.Exception (SomeException(..))

39−import Control.Monad (ap, when)

39+import Control.Monad (when)

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

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

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

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

56

57import Streamly.Internal.Data.Stream.SVar

58 (fromSVar, fromProducer, fromConsumer)

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

60import Streamly.Internal.Data.Stream.StreamK

59import Streamly.Internal.Data.Stream.StreamK

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

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

62

61

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

551 mempty = K.nil

550 mempty = K.nil

552 mappend = (<>)

551 mappend = (<>)

553

552

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+

554------------------------------------------------------------------------------

570------------------------------------------------------------------------------

555-- Monad

571-- Monad

556------------------------------------------------------------------------------

572------------------------------------------------------------------------------

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

568-- Other instances

584-- Other instances

569------------------------------------------------------------------------------

585------------------------------------------------------------------------------

570

586

571−MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)

572MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

587MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

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

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

54#if MIN_VERSION_deepseq(1,4,3)

55import Control.DeepSeq (NFData1(..))

56#endif

57−import Control.Monad (ap)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

80

81#include "Instances.hs"

82#include "inline.hs"

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

190-- Other instances

191------------------------------------------------------------------------------

192

193−{-# INLINE_EARLY mapM #-}

193+{-# INLINE mapM #-}

194mapM :: (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

196

197-- |

198-- @

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

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

212map f = mapM (return . f)

213

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+

215MONAD_COMMON_INSTANCES(SerialT,)

226MONAD_COMMON_INSTANCES(SerialT,)

216LIST_INSTANCES(SerialT)

227LIST_INSTANCES(SerialT)

217NFDATA1_INSTANCE(SerialT)

228NFDATA1_INSTANCE(SerialT)

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

383 mempty = K.nil

394 mempty = K.nil

384 mappend = (<>)

395 mappend = (<>)

385

396

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+

386------------------------------------------------------------------------------

409------------------------------------------------------------------------------

387-- Monad

410-- Monad

388------------------------------------------------------------------------------

411------------------------------------------------------------------------------

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

396-- Other instances

419-- Other instances

397------------------------------------------------------------------------------

420------------------------------------------------------------------------------

398

421

399−MONAD_APPLICATIVE_INSTANCE(WSerialT,)

400MONAD_COMMON_INSTANCES(WSerialT,)

422MONAD_COMMON_INSTANCES(WSerialT,)

401LIST_INSTANCES(WSerialT)

423LIST_INSTANCES(WSerialT)

402NFDATA1_INSTANCE(WSerialT)

424NFDATA1_INSTANCE(WSerialT)

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

429-- /Internal/

451-- /Internal/

430--

452--

431{-# INLINE unfoldrM #-}

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

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

455unfoldrM 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

68where

69

70import Control.Applicative (liftA2)

71−import Control.Monad (ap, when)

71+import Control.Monad (when)

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

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

74import Data.Functor.Identity (Identity(..))

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

242 {-# INLINE pure #-}

243 pure = yield

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

246

251

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

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

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

253-- 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

789-- @since 0.2.0

790{-# INLINE serial #-}

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

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

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

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

793−{-

794serial m1 m2 = go m1

795 where

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

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

798 single a = yld a m2

799 yieldk a r = yld a (go r)

800 in foldStream st yieldk single stop m

801−-}

802

801

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

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

804{-# INLINE conjoin #-}

803{-# INLINE conjoin #-}

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

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

22 (

23 ZipSerialM

24 , ZipSerial

25− , ZipStream -- deprecated

26 , zipSerially

25 , zipSerially

27− , zipping -- deprecated

28

26

29 , ZipAsyncM

27 , ZipAsyncM

30 , ZipAsync

28 , ZipAsync

31 , zipAsyncly

29 , zipAsyncly

30+

31+ , zipWith

32+ , zipWithM

33+ , zipAsyncWith

34+ , zipAsyncWithM

35+

36+ -- * Deprecated

37+ , ZipStream

38+ , zipping

32− , zippingAsync -- deprecated

39+ , zippingAsync

33 )

40 )

34where

41where

35

42

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

49import Prelude hiding (map, repeat, zipWith)

56import Prelude hiding (map, repeat, zipWith)

50

57

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

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

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

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

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

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

55

60

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

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

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

62import 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

58

70

59#include "Instances.hs"

71#include "Instances.hs"

60

72

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+

61------------------------------------------------------------------------------

127------------------------------------------------------------------------------

62-- Serially Zipping Streams

128-- Serially Zipping Streams

63------------------------------------------------------------------------------

129------------------------------------------------------------------------------

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

127NFDATA1_INSTANCE(ZipSerialM)

193NFDATA1_INSTANCE(ZipSerialM)

128

194

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

195instance 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

131

198

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

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

133 pure = ZipSerialM . K.repeat

200 pure = ZipSerialM . K.repeat

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

197 (|:) = consMZipAsync

264 (|:) = consMZipAsync

198

265

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

266instance 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

201

269

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

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

203 pure = ZipAsyncM . K.repeat

271 pure = ZipAsyncM . K.repeat

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

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

253 , mergeAsyncByM

254

255 -- ** Zipping

256− , zipWith

256+ , Z.zipWith

257− , zipWithM

257+ , Z.zipWithM

258− , zipAsyncWith

258+ , Z.zipAsyncWith

259− , zipAsyncWithM

259+ , Z.zipAsyncWithM

260

261 -- ** Nested Streams

262 , concatMapM

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

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

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

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

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

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

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

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

452import Streamly.Internal.Data.Stream.Prelude

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

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

455import 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)

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

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

458import Streamly.Internal.Data.Time.Units

459import Streamly.Internal.Data.Time.Units

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

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

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

476

477

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

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

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

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

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

479

481

480------------------------------------------------------------------------------

482------------------------------------------------------------------------------

481-- Deconstruction

483-- Deconstruction

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

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

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

572unfoldrMSerial = Serial.unfoldrM

574unfoldrMSerial = Serial.unfoldrM

573

575

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+

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

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

575--

587--

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

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

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

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

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

2129indexedR n = fromStreamD . D.indexedR n . toStreamD

2141indexedR n = fromStreamD . D.indexedR n . toStreamD

2130

2142

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−

2184------------------------------------------------------------------------------

2143------------------------------------------------------------------------------

2185-- Comparison

2144-- Comparison

2186------------------------------------------------------------------------------

2145------------------------------------------------------------------------------

streamly.cabal+2−2

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

742benchmark base

743 import: bench-options

744 type: exitcode-stdio-1.0

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

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

746− src

746+ , src

747 if os(windows)

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

749 if os(darwin)