2e4b883 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Remove monad instances from StreamD/StreamK

2e4b883Harendra Kumar4 years ago

benchmark/Streamly/Benchmark/Data/Stream/StreamD.hs+30

compare with master

unifiedsplit

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

9{-# LANGUAGE FlexibleContexts #-}

10{-# LANGUAGE ScopedTypeVariables #-}

11{-# LANGUAGE TupleSections #-}

12+{-# OPTIONS_GHC -Wno-orphans #-}

13#ifdef __HADDOCK_VERSION__

14#ifdef __HADDOCK_VERSION__

14#undef INSPECTION

15#undef INSPECTION

@@ -21,6 +22,7 @@

22module Main (main) where

23module Main (main) where

25+import Control.Applicative (liftA2)

24import Control.Monad (when)

26import Control.Monad (when)

25import Data.Maybe (isJust)

27import Data.Maybe (isJust)

26import Gauge (bench, nfIO, bgroup, Benchmark, defaultMain)

28import Gauge (bench, nfIO, bgroup, Benchmark, defaultMain)

@@ -322,6 +324,34 @@ filterMap n = composeN n $ S.map (subtract 1) . S.filter (<= maxValue)

322-- Nested Composition

324-- Nested Composition

323-------------------------------------------------------------------------------

325-------------------------------------------------------------------------------

324

326

327+instance Applicative f => Applicative (S.Stream f) where

328+ {-# INLINE pure #-}

329+ pure = S.fromPure

330+

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

332+ (<*>) = S.crossApply

333+

334+ {-# INLINE liftA2 #-}

335+ liftA2 f x = (<*>) (fmap f x)

336+

337+ {-# INLINE (*>) #-}

338+ (*>) = S.crossApplySnd

339+

340+ {-# INLINE (<*) #-}

341+ (<*) = S.crossApplyFst

342+

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

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

345+instance Monad m => Monad (S.Stream m) where

346+ {-# INLINE return #-}

347+ return = pure

348+

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

350+ (>>=) = flip S.concatMap

351+

352+ {-# INLINE (>>) #-}

353+ (>>) = (*>)

354+

325{-# INLINE toNullApNested #-}

355{-# INLINE toNullApNested #-}

326toNullApNested :: Monad m => Stream m Int -> m ()

356toNullApNested :: Monad m => Stream m Int -> m ()

327toNullApNested s = runStream $ do

357toNullApNested s = runStream $ do

benchmark/Streamly/Benchmark/Data/Stream/StreamK.hs+28

@@ -8,6 +8,7 @@

8{-# LANGUAGE CPP #-}

9{-# LANGUAGE FlexibleContexts #-}

10{-# LANGUAGE ScopedTypeVariables #-}

11+{-# OPTIONS_GHC -Wno-orphans #-}

12#ifdef __HADDOCK_VERSION__

13#ifdef __HADDOCK_VERSION__

13#undef INSPECTION

14#undef INSPECTION

@@ -20,6 +21,7 @@

21module Main (main) where

22module Main (main) where

24+import Control.Applicative (liftA2)

23import Control.Monad (when)

25import Control.Monad (when)

24import Data.Maybe (isJust)

26import Data.Maybe (isJust)

25import System.Random (randomRIO)

27import System.Random (randomRIO)

@@ -500,6 +502,32 @@ concatMapBySerial outer inner n =

500-- Nested Composition

502-- Nested Composition

501-------------------------------------------------------------------------------

503-------------------------------------------------------------------------------

502

504

505+instance Monad m => Applicative (S.Stream m) where

506+ {-# INLINE pure #-}

507+ pure = S.fromPure

508+

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

510+ (<*>) = S.crossApply

511+

512+ {-# INLINE liftA2 #-}

513+ liftA2 f x = (<*>) (fmap f x)

514+

515+ {-# INLINE (*>) #-}

516+ (*>) = S.crossApplySnd

517+

518+ {-# INLINE (<*) #-}

519+ (<*) = S.crossApplyFst

520+

521+-- NOTE: even though concatMap for StreamD is 3x faster compared to StreamK,

522+-- the monad instance of StreamD is slower than StreamK after foldr/build

523+-- fusion.

524+instance Monad m => Monad (S.Stream m) where

525+ {-# INLINE return #-}

526+ return = pure

527+

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

529+ (>>=) = flip S.concatMap

530+

503{-# INLINE drainApplicative #-}

531{-# INLINE drainApplicative #-}

504drainApplicative :: Monad m => Stream m Int -> m ()

532drainApplicative :: Monad m => Stream m Int -> m ()

505drainApplicative s = drain $ do

533drainApplicative s = drain $ do

core/src/Streamly/Internal/Data/Stream/Cross.hs+10−38

@@ -17,7 +17,6 @@ module Streamly.Internal.Data.Stream.Cross

17where

19import Control.Applicative (liftA2)

20−import Control.Monad.Catch (MonadThrow, throwM)

21import Control.Monad.IO.Class (MonadIO(..))

20import Control.Monad.IO.Class (MonadIO(..))

22import Data.Functor.Identity (Identity(..))

21import Data.Functor.Identity (Identity(..))

23import GHC.Exts (IsList(..), IsString(..))

22import GHC.Exts (IsList(..), IsString(..))

@@ -62,7 +61,7 @@ import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

62newtype CrossStream m a = CrossStream {getCrossStream :: Stream m a}

61newtype CrossStream m a = CrossStream {getCrossStream :: Stream m a}

63 deriving (Functor, Semigroup, Monoid, Foldable)

62 deriving (Functor, Semigroup, Monoid, Foldable)

65−-- deriving instance (Foldable m, Monad m) => Foldable (CrossStream m)

64+-- Pure (Identity monad) stream instances

66deriving instance Traversable (CrossStream Identity)

65deriving instance Traversable (CrossStream Identity)

67deriving instance IsList (CrossStream Identity a)

66deriving instance IsList (CrossStream Identity a)

68deriving instance (a ~ Char) => IsString (CrossStream Identity a)

67deriving instance (a ~ Char) => IsString (CrossStream Identity a)

@@ -71,37 +70,10 @@ deriving instance Ord a => Ord (CrossStream Identity a)

71deriving instance Show a => Show (CrossStream Identity a)

70deriving instance Show a => Show (CrossStream Identity a)

72deriving instance Read a => Read (CrossStream Identity a)

71deriving instance Read a => Read (CrossStream Identity a)

74−{-

75−instance Show a => Show (CrossStream Identity a) where

76− showsPrec p dl = showParen (p > 10) $

77− showString "fromList " . shows (toList dl)

78−

79−instance Read a => Read (ZipStream Identity a) where

80− readPrec = parens $ prec 10 $ do

81− Ident "fromList" <- lexP

82− fromList <$> readPrec

83− readListPrec = readListPrecDefault

84−-}

85−

86------------------------------------------------------------------------------

73------------------------------------------------------------------------------

87-- Applicative

74-- Applicative

88------------------------------------------------------------------------------

75------------------------------------------------------------------------------

90−{-# INLINE apSerial #-}

91−apSerial :: Monad m => Stream m (a -> b) -> Stream m a -> Stream m b

92−apSerial m1 m2 =

93− Stream.fromStreamD $ Stream.toStreamD m1 <*> Stream.toStreamD m2

94−

95−{-# INLINE apSequence #-}

96−apSequence :: Monad m => Stream m a -> Stream m b -> Stream m b

97−apSequence m1 m2 =

98− Stream.fromStreamD $ Stream.toStreamD m1 *> Stream.toStreamD m2

99−

100−{-# INLINE apDiscardSnd #-}

101−apDiscardSnd :: Monad m => Stream m a -> Stream m b -> Stream m a

102−apDiscardSnd m1 m2 =

103− Stream.fromStreamD $ Stream.toStreamD m1 <* Stream.toStreamD m2

104−

105-- Note: we need to define all the typeclass operations because we want to

77-- Note: we need to define all the typeclass operations because we want to

106-- INLINE them.

78-- INLINE them.

107instance Monad m => Applicative (CrossStream m) where

79instance Monad m => Applicative (CrossStream m) where

@@ -109,19 +81,22 @@ instance Monad m => Applicative (CrossStream m) where

109 pure x = CrossStream (Stream.fromPure x)

81 pure x = CrossStream (Stream.fromPure x)

110

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

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

112− (CrossStream s1) <*> (CrossStream s2) = CrossStream (apSerial s1 s2)

84+ (CrossStream s1) <*> (CrossStream s2) =

113− -- (<*>) = K.apSerial

85+ CrossStream (Stream.crossApply s1 s2)

86+ -- (<*>) = K.crossApply

114

115 {-# INLINE liftA2 #-}

88 {-# INLINE liftA2 #-}

116 liftA2 f x = (<*>) (fmap f x)

89 liftA2 f x = (<*>) (fmap f x)

117

118 {-# INLINE (*>) #-}

91 {-# INLINE (*>) #-}

119− (CrossStream s1) *> (CrossStream s2) = CrossStream (apSequence s1 s2)

92+ (CrossStream s1) *> (CrossStream s2) =

120− -- (*>) = K.apSerialDiscardFst

93+ CrossStream (Stream.crossApplySnd s1 s2)

94+ -- (*>) = K.crossApplySnd

121

122 {-# INLINE (<*) #-}

96 {-# INLINE (<*) #-}

123− (CrossStream s1) <* (CrossStream s2) = CrossStream (apDiscardSnd s1 s2)

97+ (CrossStream s1) <* (CrossStream s2) =

124− -- (<*) = K.apSerialDiscardSnd

98+ CrossStream (Stream.crossApplyFst s1 s2)

99+ -- (<*) = K.crossApplyFst

125

100

126------------------------------------------------------------------------------

101------------------------------------------------------------------------------

127-- Monad

102-- Monad

@@ -158,6 +133,3 @@ instance Monad m => Monad (CrossStream m) where

158

133

159instance (MonadIO m) => MonadIO (CrossStream m) where

134instance (MonadIO m) => MonadIO (CrossStream m) where

160 liftIO x = CrossStream (Stream.fromEffect $ liftIO x)

135 liftIO x = CrossStream (Stream.fromEffect $ liftIO x)

161−

162−instance (MonadThrow m) => MonadThrow (CrossStream m) where

163− throwM x = CrossStream (Stream.fromEffect $ throwM x)

core/src/Streamly/Internal/Data/Stream/StreamD/Type.hs+37−29

@@ -78,22 +78,28 @@ module Streamly.Internal.Data.Stream.StreamD.Type

78 , takeEndByM

80 -- * Nesting

81+ , crossApply

82+ , crossApplyFst

83+ , crossApplySnd

84+

81 , ConcatMapUState (..)

85 , ConcatMapUState (..)

82 , unfoldMany

86 , unfoldMany

87+

83 , concatMap

88 , concatMap

84 , concatMapM

89 , concatMapM

90+ , concatEffect

91+

85 , FoldMany (..) -- for inspection testing

92 , FoldMany (..) -- for inspection testing

86 , FoldManyPost (..)

93 , FoldManyPost (..)

87 , foldMany

94 , foldMany

88 , foldManyPost

95 , foldManyPost

89− , refoldMany

90 , chunksOf

96 , chunksOf

97+

98+ , refoldMany

91 )

99 )

92where

100where

101

94import Control.Applicative (liftA2)

102import Control.Applicative (liftA2)

95−import Control.Monad.Catch (MonadThrow, throwM)

96−import Control.Monad.Trans.Class (lift, MonadTrans)

97import Data.Functor (($>))

103import Data.Functor (($>))

98import Data.Functor.Identity (Identity(..))

104import Data.Functor.Identity (Identity(..))

99import Fusion.Plugin.Types (Fuse(..))

105import Fusion.Plugin.Types (Fuse(..))

@@ -421,9 +427,9 @@ foldrS

421foldrS f final (Stream step state) = go SPEC state

427foldrS f final (Stream step state) = go SPEC state

422 where

428 where

423 {-# INLINE_LATE go #-}

429 {-# INLINE_LATE go #-}

424− go !_ st = do

425− -- defState??

426− r <- fromEffect $ step defState st

430+ go !_ st = concatEffect $ fmap g $ step defState st

431+

432+ g r =

427 case r of

433 case r of

428 Yield x s -> f x (go SPEC s)

434 Yield x s -> f x (go SPEC s)

429 Skip s -> go SPEC s

435 Skip s -> go SPEC s

@@ -681,9 +687,9 @@ takeEndBy f = takeEndByM (return . f)

681-- Combine N Streams - concatAp

687-- Combine N Streams - concatAp

682------------------------------------------------------------------------------

688------------------------------------------------------------------------------

683

689

684−{-# INLINE_NORMAL concatAp #-}

690+{-# INLINE_NORMAL crossApply #-}

685−concatAp :: Functor f => Stream f (a -> b) -> Stream f a -> Stream f b

691+crossApply :: Functor f => Stream f (a -> b) -> Stream f a -> Stream f b

686−concatAp (Stream stepa statea) (Stream stepb stateb) =

692+crossApply (Stream stepa statea) (Stream stepb stateb) =

687 Stream step' (Left statea)

693 Stream step' (Left statea)

688

694

689 where

695 where

@@ -702,9 +708,9 @@ concatAp (Stream stepa statea) (Stream stepb stateb) =

702 Stop -> Skip (Left os))

708 Stop -> Skip (Left os))

703 (stepb (adaptState gst) st)

709 (stepb (adaptState gst) st)

704

710

705−{-# INLINE_NORMAL apSequence #-}

711+{-# INLINE_NORMAL crossApplySnd #-}

706−apSequence :: Functor f => Stream f a -> Stream f b -> Stream f b

712+crossApplySnd :: Functor f => Stream f a -> Stream f b -> Stream f b

707−apSequence (Stream stepa statea) (Stream stepb stateb) =

713+crossApplySnd (Stream stepa statea) (Stream stepb stateb) =

708 Stream step (Left statea)

714 Stream step (Left statea)

709

715

710 where

716 where

@@ -725,9 +731,9 @@ apSequence (Stream stepa statea) (Stream stepb stateb) =

725 Stop -> Skip (Left ostate))

731 Stop -> Skip (Left ostate))

726 (stepb gst st)

732 (stepb gst st)

727

733

728−{-# INLINE_NORMAL apDiscardSnd #-}

734+{-# INLINE_NORMAL crossApplyFst #-}

729−apDiscardSnd :: Functor f => Stream f a -> Stream f b -> Stream f a

735+crossApplyFst :: Functor f => Stream f a -> Stream f b -> Stream f a

730−apDiscardSnd (Stream stepa statea) (Stream stepb stateb) =

736+crossApplyFst (Stream stepa statea) (Stream stepb stateb) =

731 Stream step (Left statea)

737 Stream step (Left statea)

732

738

733 where

739 where

@@ -748,21 +754,23 @@ apDiscardSnd (Stream stepa statea) (Stream stepb stateb) =

748 Stop -> Skip (Left ostate))

754 Stop -> Skip (Left ostate))

749 (stepb (adaptState gst) st)

755 (stepb (adaptState gst) st)

750

756

757+{-

751instance Applicative f => Applicative (Stream f) where

758instance Applicative f => Applicative (Stream f) where

752 {-# INLINE pure #-}

759 {-# INLINE pure #-}

753 pure = fromPure

760 pure = fromPure

754

761

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

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

756− (<*>) = concatAp

763+ (<*>) = crossApply

757

764

758 {-# INLINE liftA2 #-}

765 {-# INLINE liftA2 #-}

759 liftA2 f x = (<*>) (fmap f x)

766 liftA2 f x = (<*>) (fmap f x)

760

767

761 {-# INLINE (*>) #-}

768 {-# INLINE (*>) #-}

762− (*>) = apSequence

769+ (*>) = crossApplySnd

763

770

764 {-# INLINE (<*) #-}

771 {-# INLINE (<*) #-}

765− (<*) = apDiscardSnd

772+ (<*) = crossApplyFst

773+-}

766

774

767------------------------------------------------------------------------------

775------------------------------------------------------------------------------

768-- Combine N Streams - unfoldMany

776-- Combine N Streams - unfoldMany

@@ -854,6 +862,16 @@ concatMap f = concatMapM (return . f)

854-- {-# RULES "concatMap Array.toStreamD"

862-- {-# RULES "concatMap Array.toStreamD"

855-- concatMap Array.toStreamD = Array.flattenArray #-}

863-- concatMap Array.toStreamD = Array.flattenArray #-}

856

864

865+-- |

866+-- Definition:

867+--

868+-- > concatEffect generator = concatMapM (\() -> generator) (fromPure ())

869+--

870+{-# INLINE concatEffect #-}

871+concatEffect :: Monad m => m (Stream m a) -> Stream m a

872+concatEffect generator = concatMapM (\() -> generator) (fromPure ())

873+

874+{-

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

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

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

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

859instance Monad m => Monad (Stream m) where

877instance Monad m => Monad (Stream m) where

@@ -865,6 +883,7 @@ instance Monad m => Monad (Stream m) where

865

883

866 {-# INLINE (>>) #-}

884 {-# INLINE (>>) #-}

867 (>>) = (*>)

885 (>>) = (*>)

886+-}

868

887

869------------------------------------------------------------------------------

888------------------------------------------------------------------------------

870-- Grouping/Splitting

889-- Grouping/Splitting

@@ -1012,14 +1031,3 @@ refoldMany (Refold fstep inject extract) action (Stream step state) =

1012 return $ Skip (FoldManyYield b FoldManyDone)

1031 return $ Skip (FoldManyYield b FoldManyDone)

1013 step' _ (FoldManyYield b next) = return $ Yield b next

1032 step' _ (FoldManyYield b next) = return $ Yield b next

1014 step' _ FoldManyDone = return Stop

1033 step' _ FoldManyDone = return Stop

1015−

1016−------------------------------------------------------------------------------

1017−-- Other instances

1018−------------------------------------------------------------------------------

1019−

1020−instance MonadTrans Stream where

1021− {-# INLINE lift #-}

1022− lift = fromEffect

1023−

1024−instance (MonadThrow m) => MonadThrow (Stream m) where

1025− throwM = lift . throwM

core/src/Streamly/Internal/Data/Stream/StreamK.hs+5−4

@@ -152,14 +152,15 @@ module Streamly.Internal.Data.Stream.StreamK

152 , mergeByM

153

154 -- ** Nesting

155+ , crossApplyWith

156+ , crossApply

157+ , crossApplySnd

158+ , crossApplyFst

159+

155 , concatMapWith

160 , concatMapWith

156 , concatMap

161 , concatMap

157 , bindWith

162 , bindWith

158 , concatPairsWith

163 , concatPairsWith

159− , apWith

160− , apSerial

161− , apSerialDiscardFst

162− , apSerialDiscardSnd

163

164

164 -- ** Transformation comprehensions

165 -- ** Transformation comprehensions

165 , the

166 , the

core/src/Streamly/Internal/Data/Stream/StreamK/Transformer.hs+7−7

@@ -19,7 +19,7 @@ where

19import Control.Monad.Trans.Class (MonadTrans(lift))

20import Control.Monad.Trans.State.Strict (StateT)

21import Streamly.Internal.Data.Stream.StreamK

22− (Stream, nil, cons, uncons, fromEffect)

22+ (Stream, nil, cons, uncons, concatEffect)

24import qualified Control.Monad.Trans.State.Strict as State

@@ -57,9 +57,9 @@ evalStateT = go

58 where

60− -- XXX Do not use stream monad

61− go st m1 = do

62− (res, s1) <- lift $ st >>= State.runStateT (uncons m1)

60+ go st m1 = concatEffect $ fmap f (st >>= State.runStateT (uncons m1))

61+

62+ f (res, s1) =

63 case res of

64 Just (h, t) -> cons h (go (return s1) t)

65 Nothing -> nil

@@ -71,9 +71,9 @@ liftInner = go

72 where

74− -- XXX Do not use stream monad

75− go m1 = do

76− res <- fromEffect $ lift $ uncons m1

74+ go m1 = concatEffect $ fmap f $ lift $ uncons m1

75+

76+ f res =

77 case res of

78 Just (h, t) -> cons h (go t)

79 Nothing -> nil

core/src/Streamly/Internal/Data/Stream/StreamK/Type.hs+35−64

@@ -78,14 +78,17 @@ module Streamly.Internal.Data.Stream.StreamK.Type

78 , mapMWith

79 , mapMSerial

80 , unShare

81+

82+ , crossApplyWith

83+ , crossApply

84+ , crossApplySnd

85+ , crossApplyFst

86+

81 , concatMapWith

87 , concatMapWith

82 , concatMap

88 , concatMap

83 , bindWith

89 , bindWith

84 , concatPairsWith

90 , concatPairsWith

85− , apWith

91+

86− , apSerial

87− , apSerialDiscardFst

88− , apSerialDiscardSnd

89 , foldlS

92 , foldlS

90 , reverse

93 , reverse

@@ -97,20 +100,14 @@ module Streamly.Internal.Data.Stream.StreamK.Type

97 , interleave

100 , interleave

98 , interleaveFst

101 , interleaveFst

99 , interleaveMin

102 , interleaveMin

100−

101− -- * Reader

102− , evalStateT

103− , liftInner

104 )

103 )

105where

104where

106

105

107−import Control.Monad (ap, (>=>))

106+-- import Control.Applicative (liftA2)

108−import Control.Monad.Trans.Class (MonadTrans(lift))

107+import Control.Monad ((>=>))

109−import Control.Monad.Trans.State.Strict (StateT)

110import Data.Function (fix)

108import Data.Function (fix)

111import Streamly.Internal.Data.SVar.Type (State, adaptState, defState)

109import Streamly.Internal.Data.SVar.Type (State, adaptState, defState)

112

110

113−import qualified Control.Monad.Trans.State.Strict as State

114import qualified Prelude

111import qualified Prelude

115

112

116import Prelude hiding

113import Prelude hiding

@@ -866,14 +863,6 @@ mapMWith cns f = go

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

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

867 fmap = map

864 fmap = map

868

865

869−-------------------------------------------------------------------------------

870−-- Transformers

871−-------------------------------------------------------------------------------

872−

873−instance MonadTrans Stream where

874− {-# INLINE lift #-}

875− lift = fromEffect

876−

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

866-------------------------------------------------------------------------------

878-- Nesting

867-- Nesting

879-------------------------------------------------------------------------------

868-------------------------------------------------------------------------------

@@ -885,13 +874,13 @@ unShare x = mkStream $ \st yld sng stp ->

885 foldStream st yld sng stp x

874 foldStream st yld sng stp x

886

875

887-- XXX the function stream and value stream can run in parallel

876-- XXX the function stream and value stream can run in parallel

888−{-# INLINE apWith #-}

877+{-# INLINE crossApplyWith #-}

889−apWith ::

878+crossApplyWith ::

890 (Stream m b -> Stream m b -> Stream m b)

879 (Stream m b -> Stream m b -> Stream m b)

891 -> Stream m (a -> b)

880 -> Stream m (a -> b)

892 -> Stream m a

881 -> Stream m a

893 -> Stream m b

882 -> Stream m b

894−apWith par fstream stream = go1 fstream

883+crossApplyWith par fstream stream = go1 fstream

895

884

896 where

885 where

897

886

@@ -908,12 +897,12 @@ apWith par fstream stream = go1 fstream

908 yieldk a r = yld (f a) (go2 f r)

897 yieldk a r = yld (f a) (go2 f r)

909 in foldStream (adaptState st) yieldk single stp m

898 in foldStream (adaptState st) yieldk single stp m

910

899

911−{-# INLINE apSerial #-}

900+{-# INLINE crossApply #-}

912−apSerial ::

901+crossApply ::

913 Stream m (a -> b)

902 Stream m (a -> b)

914 -> Stream m a

903 -> Stream m a

915 -> Stream m b

904 -> Stream m b

916−apSerial fstream stream = go1 fstream

905+crossApply fstream stream = go1 fstream

917

906

918 where

907 where

919

908

@@ -938,12 +927,12 @@ apSerial fstream stream = go1 fstream

938 yieldk a r = yld (f a) (go3 f r)

927 yieldk a r = yld (f a) (go3 f r)

939 in foldStream (adaptState st) yieldk single stp m

928 in foldStream (adaptState st) yieldk single stp m

940

929

941−{-# INLINE apSerialDiscardFst #-}

930+{-# INLINE crossApplySnd #-}

942−apSerialDiscardFst ::

931+crossApplySnd ::

943 Stream m a

932 Stream m a

944 -> Stream m b

933 -> Stream m b

945 -> Stream m b

934 -> Stream m b

946−apSerialDiscardFst fstream stream = go1 fstream

935+crossApplySnd fstream stream = go1 fstream

947

936

948 where

937 where

949

938

@@ -962,12 +951,12 @@ apSerialDiscardFst fstream stream = go1 fstream

962 yieldk a r = yld a (go2 r1 r)

951 yieldk a r = yld a (go2 r1 r)

963 in foldStream st yieldk single stop m

952 in foldStream st yieldk single stop m

964

953

965−{-# INLINE apSerialDiscardSnd #-}

954+{-# INLINE crossApplyFst #-}

966−apSerialDiscardSnd ::

955+crossApplyFst ::

967 Stream m a

956 Stream m a

968 -> Stream m b

957 -> Stream m b

969 -> Stream m a

958 -> Stream m a

970−apSerialDiscardSnd fstream stream = go1 fstream

959+crossApplyFst fstream stream = go1 fstream

971

960

972 where

961 where

973

962

@@ -1116,11 +1105,22 @@ concatPairsWith combine f str = go (leafPairs str)

1116 yieldk a r = yld (a1 `combine` a) $ nonLeafPairs r

1105 yieldk a r = yld (a1 `combine` a) $ nonLeafPairs r

1117 in foldStream (adaptState st) yieldk single stop stream

1106 in foldStream (adaptState st) yieldk single stop stream

1118

1107

1108+{-

1119instance Monad m => Applicative (Stream m) where

1109instance Monad m => Applicative (Stream m) where

1120 {-# INLINE pure #-}

1110 {-# INLINE pure #-}

1121 pure = fromPure

1111 pure = fromPure

1112+

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

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

1123− (<*>) = ap

1114+ (<*>) = crossApply

1115+

1116+ {-# INLINE liftA2 #-}

1117+ liftA2 f x = (<*>) (fmap f x)

1118+

1119+ {-# INLINE (*>) #-}

1120+ (*>) = crossApplySnd

1121+

1122+ {-# INLINE (<*) #-}

1123+ (<*) = crossApplyFst

1124

1125-- NOTE: even though concatMap for StreamD is 3x faster compared to StreamK,

1126-- the monad instance of StreamD is slower than StreamK after foldr/build

@@ -1128,8 +1128,10 @@ instance Monad m => Applicative (Stream m) where

1128instance Monad m => Monad (Stream m) where

1129 {-# INLINE return #-}

1130 return = pure

1131+

1131 {-# INLINE (>>=) #-}

1132 {-# INLINE (>>=) #-}

1132 (>>=) = flip concatMap

1133 (>>=) = flip concatMap

1134+-}

1133

1135

1134{-

1136{-

1135-- Like concatMap but generates stream using an unfold function. Similar to

1137-- Like concatMap but generates stream using an unfold function. Similar to

@@ -1386,34 +1388,3 @@ concatMapEffect :: Monad m => (b -> Stream m a) -> m b -> Stream m a

1386concatMapEffect f action =

1388concatMapEffect f action =

1387 mkStream $ \st yld sng stp ->

1389 mkStream $ \st yld sng stp ->

1388 action >>= foldStreamShared st yld sng stp . f

1390 action >>= foldStreamShared st yld sng stp . f

1389−

1390−------------------------------------------------------------------------------

1391−-- Lifting inner monad

1392−------------------------------------------------------------------------------

1393−

1394−{-# INLINE evalStateT #-}

1395−evalStateT :: Monad m => m s -> Stream (StateT s m) a -> Stream m a

1396−evalStateT = go

1397−

1398− where

1399−

1400− -- XXX Do not use stream monad

1401− go st m1 = do

1402− (res, s1) <- lift $ st >>= State.runStateT (uncons m1)

1403− case res of

1404− Just (h, t) -> cons h (go (return s1) t)

1405− Nothing -> nil

1406−

1407−{-# INLINE liftInner #-}

1408−liftInner :: (Monad m, MonadTrans t, Monad (t m)) =>

1409− Stream m a -> Stream (t m) a

1410−liftInner = go

1411−

1412− where

1413−

1414− -- XXX Do not use stream monad

1415− go m1 = do

1416− res <- fromEffect $ lift $ uncons m1

1417− case res of

1418− Just (h, t) -> cons h (go t)

1419− Nothing -> nil

core/src/Streamly/Internal/Data/Stream/Type.hs+28

@@ -28,6 +28,11 @@ module Streamly.Internal.Data.Stream.Type

28 , fromPure

29 , fromEffect

31+ -- * Applicative

32+ , crossApply

33+ , crossApplySnd

34+ , crossApplyFst

35+

31 -- * Bind/Concat

36 -- * Bind/Concat

32 , bindWith

37 , bindWith

33 , concatMapWith

38 , concatMapWith

@@ -378,10 +383,32 @@ fromPure = fromStreamK . K.fromPure

378fromEffect :: Monad m => m a -> Stream m a

383fromEffect :: Monad m => m a -> Stream m a

379fromEffect = fromStreamK . K.fromEffect

384fromEffect = fromStreamK . K.fromEffect

380

385

386+-------------------------------------------------------------------------------

387+-- Applicative

388+-------------------------------------------------------------------------------

389+

390+{-# INLINE crossApply #-}

391+crossApply :: Monad m => Stream m (a -> b) -> Stream m a -> Stream m b

392+crossApply m1 m2 =

393+ fromStreamD $ D.crossApply (toStreamD m1) (toStreamD m2)

394+

395+{-# INLINE crossApplySnd #-}

396+crossApplySnd :: Monad m => Stream m a -> Stream m b -> Stream m b

397+crossApplySnd m1 m2 =

398+ fromStreamD $ D.crossApplySnd (toStreamD m1) (toStreamD m2)

399+

400+{-# INLINE crossApplyFst #-}

401+crossApplyFst :: Monad m => Stream m a -> Stream m b -> Stream m a

402+crossApplyFst m1 m2 =

403+ fromStreamD $ D.crossApplyFst (toStreamD m1) (toStreamD m2)

404+

381-------------------------------------------------------------------------------

405-------------------------------------------------------------------------------

382-- Bind/Concat

406-- Bind/Concat

383-------------------------------------------------------------------------------

407-------------------------------------------------------------------------------

384

408

409+-- |

410+--

411+-- /CPS/

385{-# INLINE bindWith #-}

412{-# INLINE bindWith #-}

386bindWith

413bindWith

387 :: (Stream m b -> Stream m b -> Stream m b)

414 :: (Stream m b -> Stream m b -> Stream m b)

@@ -400,6 +427,7 @@ bindWith par m1 f =

400-- elements in the input @stream@ and the @mixer@ function is used to merge

427-- elements in the input @stream@ and the @mixer@ function is used to merge

401-- those streams.

428-- those streams.

402--

429--

430+-- /CPS/

403{-# INLINE concatMapWith #-}

431{-# INLINE concatMapWith #-}

404concatMapWith

432concatMapWith

405 :: (Stream m b -> Stream m b -> Stream m b)

433 :: (Stream m b -> Stream m b -> Stream m b)

src/Streamly/Internal/Data/Stream/Ahead.hs+5−2

@@ -690,7 +690,10 @@ consM m (AheadT r) = AheadT $ aheadK (K.fromEffect m) r

690--

691-- @since 0.8.0

692newtype AheadT m a = AheadT {getAheadT :: Stream m a}

693+

693− deriving (MonadTrans)

694+instance MonadTrans AheadT where

695+ {-# INLINE lift #-}

696+ lift = AheadT . K.fromEffect

694

697

695-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See

698-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See

696-- 'AheadT' documentation for more details.

699-- 'AheadT' documentation for more details.

@@ -727,7 +730,7 @@ instance MonadAsync m => Monoid (AheadT m a) where

727{-# INLINE apAhead #-}

730{-# INLINE apAhead #-}

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

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

729apAhead (AheadT m1) (AheadT m2) =

732apAhead (AheadT m1) (AheadT m2) =

730− let f x1 = K.concatMapWith aheadK (pure . x1) m2

733+ let f x1 = K.concatMapWith aheadK (K.fromPure . x1) m2

731 in AheadT $ K.concatMapWith aheadK f m1

734 in AheadT $ K.concatMapWith aheadK f m1

732

735

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

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

src/Streamly/Internal/Data/Stream/Async.hs+10−4

@@ -742,7 +742,10 @@ consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r

742--

743-- @since 0.8.0

744newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}

745+

745− deriving (MonadTrans)

746+instance MonadTrans AsyncT where

747+ {-# INLINE lift #-}

748+ lift = AsyncT . K.fromEffect

746

749

747-- | A demand driven left biased parallely composing IO stream of elements of

750-- | A demand driven left biased parallely composing IO stream of elements of

748-- type @a@. See 'AsyncT' documentation for more details.

751-- type @a@. See 'AsyncT' documentation for more details.

@@ -782,7 +785,7 @@ instance MonadAsync m => Monoid (AsyncT m a) where

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

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

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

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

784apAsync (AsyncT m1) (AsyncT m2) =

787apAsync (AsyncT m1) (AsyncT m2) =

785− let f x1 = K.concatMapWith asyncK (pure . x1) m2

788+ let f x1 = K.concatMapWith asyncK (K.fromPure . x1) m2

786 in AsyncT $ K.concatMapWith asyncK f m1

789 in AsyncT $ K.concatMapWith asyncK f m1

787

790

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

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

@@ -965,7 +968,10 @@ consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r

965-- @

968-- @

966--

969--

967newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}

970newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}

971+

968− deriving (MonadTrans)

972+instance MonadTrans WAsyncT where

973+ {-# INLINE lift #-}

974+ lift = WAsyncT . K.fromEffect

969

975

970-- | A round robin parallely composing IO stream of elements of type @a@.

976-- | A round robin parallely composing IO stream of elements of type @a@.

971-- See 'WAsyncT' documentation for more details.

977-- See 'WAsyncT' documentation for more details.

@@ -1004,7 +1010,7 @@ instance MonadAsync m => Monoid (WAsyncT m a) where

1004 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}

1010 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}

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

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

1006apWAsync (WAsyncT m1) (WAsyncT m2) =

1012apWAsync (WAsyncT m1) (WAsyncT m2) =

1007− let f x1 = K.concatMapWith wAsyncK (pure . x1) m2

1013+ let f x1 = K.concatMapWith wAsyncK (K.fromPure . x1) m2

1008 in WAsyncT $ K.concatMapWith wAsyncK f m1

1014 in WAsyncT $ K.concatMapWith wAsyncK f m1

1009

1015

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

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

src/Streamly/Internal/Data/Stream/Parallel.hs+5−2

@@ -436,7 +436,10 @@ tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit

436--

437-- @since 0.8.0

438newtype ParallelT m a = ParallelT {getParallelT :: K.Stream m a}

439+

439− deriving (MonadTrans)

440+instance MonadTrans ParallelT where

441+ {-# INLINE lift #-}

442+ lift = ParallelT . K.fromEffect

440

443

441-- | A parallely composing IO stream of elements of type @a@.

444-- | A parallely composing IO stream of elements of type @a@.

442-- See 'ParallelT' documentation for more details.

445-- See 'ParallelT' documentation for more details.

@@ -476,7 +479,7 @@ instance MonadAsync m => Monoid (ParallelT m a) where

476apParallel :: MonadAsync m =>

479apParallel :: MonadAsync m =>

477 ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b

480 ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b

478apParallel (ParallelT m1) (ParallelT m2) =

481apParallel (ParallelT m1) (ParallelT m2) =

479− let f x1 = K.concatMapWith parallelK (pure . x1) m2

482+ let f x1 = K.concatMapWith parallelK (K.fromPure . x1) m2

480 in ParallelT $ K.concatMapWith parallelK f m1

483 in ParallelT $ K.concatMapWith parallelK f m1

481

484

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

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

src/Streamly/Internal/Data/Stream/Serial.hs+11−23

@@ -69,13 +69,13 @@ import Streamly.Data.Stream (Stream)

69import qualified Streamly.Data.Fold as Fold

70import qualified Streamly.Data.Stream as Stream

71import qualified Streamly.Internal.Data.Stream as Stream

72− (toStreamD, fromStreamD, toStreamK, fromStreamK)

72+ (toStreamK, fromStreamK, crossApply, crossApplySnd, crossApplyFst)

73import qualified Streamly.Internal.Data.Stream.Common as P

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

75 (fromStreamK, toStreamK, mapM)

76import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

77 (Stream, mkStream, foldStream, cons, consM, nil, concatMapWith, fromPure

78− , bindWith, interleave, interleaveFst, interleaveMin, serial)

78+ , bindWith, interleave, interleaveFst, interleaveMin, serial, fromEffect)

80import Prelude hiding (map, mapM, repeat, filter)

@@ -99,21 +99,6 @@ withLocal f m =

99-- Applicative - orphan instances

100------------------------------------------------------------------------------

101

102−{-# INLINE apSerial #-}

103−apSerial :: Monad m => Stream m (a -> b) -> Stream m a -> Stream m b

104−apSerial m1 m2 =

105− Stream.fromStreamD $ Stream.toStreamD m1 <*> Stream.toStreamD m2

106−

107−{-# INLINE apSequence #-}

108−apSequence :: Monad m => Stream m a -> Stream m b -> Stream m b

109−apSequence m1 m2 =

110− Stream.fromStreamD $ Stream.toStreamD m1 *> Stream.toStreamD m2

111−

112−{-# INLINE apDiscardSnd #-}

113−apDiscardSnd :: Monad m => Stream m a -> Stream m b -> Stream m a

114−apDiscardSnd m1 m2 =

115− Stream.fromStreamD $ Stream.toStreamD m1 <* Stream.toStreamD m2

116−

117-- Note: we need to define all the typeclass operations because we want to

102-- Note: we need to define all the typeclass operations because we want to

118-- INLINE them.

103-- INLINE them.

119instance Monad m => Applicative (Stream m) where

104instance Monad m => Applicative (Stream m) where

@@ -121,18 +106,18 @@ instance Monad m => Applicative (Stream m) where

121 pure = Stream.fromStreamK . K.fromPure

106 pure = Stream.fromStreamK . K.fromPure

122

107

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

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

124− (<*>) = apSerial

109+ (<*>) = Stream.crossApply

125 -- (<*>) = K.apSerial

110 -- (<*>) = K.apSerial

126

111

127 {-# INLINE liftA2 #-}

112 {-# INLINE liftA2 #-}

128 liftA2 f x = (<*>) (fmap f x)

113 liftA2 f x = (<*>) (fmap f x)

129

114

130 {-# INLINE (*>) #-}

115 {-# INLINE (*>) #-}

131− (*>) = apSequence

116+ (*>) = Stream.crossApplySnd

132 -- (*>) = K.apSerialDiscardFst

117 -- (*>) = K.apSerialDiscardFst

133

118

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

119 {-# INLINE (<*) #-}

135− (<*) = apDiscardSnd

120+ (<*) = Stream.crossApplyFst

136 -- (<*) = K.apSerialDiscardSnd

121 -- (<*) = K.apSerialDiscardSnd

137

122

138------------------------------------------------------------------------------

123------------------------------------------------------------------------------

@@ -288,7 +273,10 @@ map f = Stream.mapM (return . f)

288--

273--

289-- @since 0.8.0

274-- @since 0.8.0

290newtype WSerialT m a = WSerialT {getWSerialT :: K.Stream m a}

275newtype WSerialT m a = WSerialT {getWSerialT :: K.Stream m a}

276+

291− deriving (MonadTrans)

277+instance MonadTrans WSerialT where

278+ {-# INLINE lift #-}

279+ lift = WSerialT . K.fromEffect

292

280

293-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'

281-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'

294-- documentation for more details.

282-- documentation for more details.

@@ -353,9 +341,9 @@ instance Monoid (WSerialT m a) where

353 mappend = (<>)

341 mappend = (<>)

354

342

355{-# INLINE apWSerial #-}

343{-# INLINE apWSerial #-}

356−apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b

344+apWSerial :: WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b

357apWSerial (WSerialT m1) (WSerialT m2) =

345apWSerial (WSerialT m1) (WSerialT m2) =

358− let f x1 = K.concatMapWith K.interleave (pure . x1) m2

346+ let f x1 = K.concatMapWith K.interleave (K.fromPure . x1) m2

359 in WSerialT $ K.concatMapWith K.interleave f m1

347 in WSerialT $ K.concatMapWith K.interleave f m1

360

348

361instance Monad m => Applicative (WSerialT m) where

349instance Monad m => Applicative (WSerialT m) where