2e4b883 · streamly · git.iko.soy

git.iko.soy/streamly

Remove monad instances from StreamD/StreamK

commit2e4b883be17a2668e910da2ed8806ec1fd164ae9 authorHarendra Kumar <harendra@composewell.com> dateSun 20 Nov 2022, 10:00 (4 years ago) parent8c8b9c3

12 files changed+211−173

compare with master

unifiedsplit

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

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

99{-# LANGUAGE FlexibleContexts #-}

1010{-# LANGUAGE ScopedTypeVariables #-}

1111{-# LANGUAGE TupleSections #-}

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

1213

1314#ifdef __HADDOCK_VERSION__

1415#undef INSPECTION

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

2122

2223module Main (main) where

2324

25+import Control.Applicative (liftA2)

2426import Control.Monad (when)

2527import Data.Maybe (isJust)

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

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

322324-- Nested Composition

323325-------------------------------------------------------------------------------

324326

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+

325355{-# INLINE toNullApNested #-}

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

327357toNullApNested s = runStream $ do

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

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

88{-# LANGUAGE CPP #-}

99{-# LANGUAGE FlexibleContexts #-}

1010{-# LANGUAGE ScopedTypeVariables #-}

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

1112

1213#ifdef __HADDOCK_VERSION__

1314#undef INSPECTION

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

2021

2122module Main (main) where

2223

24+import Control.Applicative (liftA2)

2325import Control.Monad (when)

2426import Data.Maybe (isJust)

2527import System.Random (randomRIO)

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

500502-- Nested Composition

501503-------------------------------------------------------------------------------

502504

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+

503531{-# INLINE drainApplicative #-}

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

505533drainApplicative s = drain $ do

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

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

1717where

1818

1919import Control.Applicative (liftA2)

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

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

2221import Data.Functor.Identity (Identity(..))

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

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

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

6362 deriving (Functor, Semigroup, Monoid, Foldable)

6463

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

64+-- Pure (Identity monad) stream instances

6665deriving instance Traversable (CrossStream Identity)

6766deriving instance IsList (CrossStream Identity a)

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

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

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

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

7372

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−

8673------------------------------------------------------------------------------

8774-- Applicative

8875------------------------------------------------------------------------------

8976

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−

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

10678-- INLINE them.

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

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

10981 pure x = CrossStream (Stream.fromPure x)

11082

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

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

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

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

85+ CrossStream (Stream.crossApply s1 s2)

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

11487

11588 {-# INLINE liftA2 #-}

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

11790

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

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

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

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

93+ CrossStream (Stream.crossApplySnd s1 s2)

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

12195

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

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

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

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

98+ CrossStream (Stream.crossApplyFst s1 s2)

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

125100

126101------------------------------------------------------------------------------

127102-- Monad

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

158133

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

160135 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

7878 , takeEndByM

7979

8080 -- * Nesting

81+ , crossApply

82+ , crossApplyFst

83+ , crossApplySnd

84+

8185 , ConcatMapUState (..)

8286 , unfoldMany

87+

8388 , concatMap

8489 , concatMapM

90+ , concatEffect

91+

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

8693 , FoldManyPost (..)

8794 , foldMany

8895 , foldManyPost

89− , refoldMany

9096 , chunksOf

97+

98+ , refoldMany

9199 )

92100where

93101

94102import Control.Applicative (liftA2)

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

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

97103import Data.Functor (($>))

98104import Data.Functor.Identity (Identity(..))

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

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

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

422428 where

423429 {-# 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 =

427433 case r of

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

429435 Skip s -> go SPEC s

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

681687-- Combine N Streams - concatAp

682688------------------------------------------------------------------------------

683689

684−{-# INLINE_NORMAL concatAp #-}

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

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

690+{-# INLINE_NORMAL crossApply #-}

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

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

687693 Stream step' (Left statea)

688694

689695 where

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

702708 Stop -> Skip (Left os))

703709 (stepb (adaptState gst) st)

704710

705−{-# INLINE_NORMAL apSequence #-}

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

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

711+{-# INLINE_NORMAL crossApplySnd #-}

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

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

708714 Stream step (Left statea)

709715

710716 where

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

725731 Stop -> Skip (Left ostate))

726732 (stepb gst st)

727733

728−{-# INLINE_NORMAL apDiscardSnd #-}

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

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

734+{-# INLINE_NORMAL crossApplyFst #-}

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

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

731737 Stream step (Left statea)

732738

733739 where

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

748754 Stop -> Skip (Left ostate))

749755 (stepb (adaptState gst) st)

750756

757+{-

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

752759 {-# INLINE pure #-}

753760 pure = fromPure

754761

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

756− (<*>) = concatAp

763+ (<*>) = crossApply

757764

758765 {-# INLINE liftA2 #-}

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

760767

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

762− (*>) = apSequence

769+ (*>) = crossApplySnd

763770

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

765− (<*) = apDiscardSnd

772+ (<*) = crossApplyFst

773+-}

766774

767775------------------------------------------------------------------------------

768776-- Combine N Streams - unfoldMany

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

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

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

856864

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

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

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

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

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

865883

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

867885 (>>) = (*>)

886+-}

868887

869888------------------------------------------------------------------------------

870889-- Grouping/Splitting

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

10121031 return $ Skip (FoldManyYield b FoldManyDone)

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

10141033 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

152152 , mergeByM

153153

154154 -- ** Nesting

155+ , crossApplyWith

156+ , crossApply

157+ , crossApplySnd

158+ , crossApplyFst

159+

155160 , concatMapWith

156161 , concatMap

157162 , bindWith

158163 , concatPairsWith

159− , apWith

160− , apSerial

161− , apSerialDiscardFst

162− , apSerialDiscardSnd

163164

164165 -- ** Transformation comprehensions

165166 , the

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

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

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

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

2121import Streamly.Internal.Data.Stream.StreamK

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

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

2323

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

2525

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

5757

5858 where

5959

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) =

6363 case res of

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

6565 Nothing -> nil

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

7171

7272 where

7373

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 =

7777 case res of

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

7979 Nothing -> nil

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

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

7878 , mapMWith

7979 , mapMSerial

8080 , unShare

81+

82+ , crossApplyWith

83+ , crossApply

84+ , crossApplySnd

85+ , crossApplyFst

86+

8187 , concatMapWith

8288 , concatMap

8389 , bindWith

8490 , concatPairsWith

85− , apWith

86− , apSerial

87− , apSerialDiscardFst

88− , apSerialDiscardSnd

91+

8992 , foldlS

9093 , reverse

9194

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

97100 , interleave

98101 , interleaveFst

99102 , interleaveMin

100−

101− -- * Reader

102− , evalStateT

103− , liftInner

104103 )

105104where

106105

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

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

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

106+-- import Control.Applicative (liftA2)

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

110108import Data.Function (fix)

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

112110

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

114111import qualified Prelude

115112

116113import Prelude hiding

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

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

867864 fmap = map

868865

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

870−-- Transformers

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

872−

873−instance MonadTrans Stream where

874− {-# INLINE lift #-}

875− lift = fromEffect

876−

877866-------------------------------------------------------------------------------

878867-- Nesting

879868-------------------------------------------------------------------------------

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

885874 foldStream st yld sng stp x

886875

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

888−{-# INLINE apWith #-}

889−apWith ::

877+{-# INLINE crossApplyWith #-}

878+crossApplyWith ::

890879 (Stream m b -> Stream m b -> Stream m b)

891880 -> Stream m (a -> b)

892881 -> Stream m a

893882 -> Stream m b

894−apWith par fstream stream = go1 fstream

883+crossApplyWith par fstream stream = go1 fstream

895884

896885 where

897886

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

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

909898 in foldStream (adaptState st) yieldk single stp m

910899

911−{-# INLINE apSerial #-}

912−apSerial ::

900+{-# INLINE crossApply #-}

901+crossApply ::

913902 Stream m (a -> b)

914903 -> Stream m a

915904 -> Stream m b

916−apSerial fstream stream = go1 fstream

905+crossApply fstream stream = go1 fstream

917906

918907 where

919908

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

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

939928 in foldStream (adaptState st) yieldk single stp m

940929

941−{-# INLINE apSerialDiscardFst #-}

942−apSerialDiscardFst ::

930+{-# INLINE crossApplySnd #-}

931+crossApplySnd ::

943932 Stream m a

944933 -> Stream m b

945934 -> Stream m b

946−apSerialDiscardFst fstream stream = go1 fstream

935+crossApplySnd fstream stream = go1 fstream

947936

948937 where

949938

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

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

963952 in foldStream st yieldk single stop m

964953

965−{-# INLINE apSerialDiscardSnd #-}

966−apSerialDiscardSnd ::

954+{-# INLINE crossApplyFst #-}

955+crossApplyFst ::

967956 Stream m a

968957 -> Stream m b

969958 -> Stream m a

970−apSerialDiscardSnd fstream stream = go1 fstream

959+crossApplyFst fstream stream = go1 fstream

971960

972961 where

973962

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

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

11171106 in foldStream (adaptState st) yieldk single stop stream

11181107

1108+{-

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

11201110 {-# INLINE pure #-}

11211111 pure = fromPure

1112+

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

11241124

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

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

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

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

11291129 {-# INLINE return #-}

11301130 return = pure

1131+

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

11321133 (>>=) = flip concatMap

1134+-}

11331135

11341136{-

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

13861388concatMapEffect f action =

13871389 mkStream $ \st yld sng stp ->

13881390 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

2828 , fromPure

2929 , fromEffect

3030

31+ -- * Applicative

32+ , crossApply

33+ , crossApplySnd

34+ , crossApplyFst

35+

3136 -- * Bind/Concat

3237 , bindWith

3338 , concatMapWith

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

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

379384fromEffect = fromStreamK . K.fromEffect

380385

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+

381405-------------------------------------------------------------------------------

382406-- Bind/Concat

383407-------------------------------------------------------------------------------

384408

409+-- |

410+--

411+-- /CPS/

385412{-# INLINE bindWith #-}

386413bindWith

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

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

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

401428-- those streams.

402429--

430+-- /CPS/

403431{-# INLINE concatMapWith #-}

404432concatMapWith

405433 :: (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

690690--

691691-- @since 0.8.0

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

693− deriving (MonadTrans)

693+

694+instance MonadTrans AheadT where

695+ {-# INLINE lift #-}

696+ lift = AheadT . K.fromEffect

694697

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

696699-- 'AheadT' documentation for more details.

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

727730{-# INLINE apAhead #-}

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

729732apAhead (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

731734 in AheadT $ K.concatMapWith aheadK f m1

732735

733736instance (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

742742--

743743-- @since 0.8.0

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

745− deriving (MonadTrans)

745+

746+instance MonadTrans AsyncT where

747+ {-# INLINE lift #-}

748+ lift = AsyncT . K.fromEffect

746749

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

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

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

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

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

784787apAsync (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

786789 in AsyncT $ K.concatMapWith asyncK f m1

787790

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

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

965968-- @

966969--

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

968− deriving (MonadTrans)

971+

972+instance MonadTrans WAsyncT where

973+ {-# INLINE lift #-}

974+ lift = WAsyncT . K.fromEffect

969975

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

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

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

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

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

10061012apWAsync (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

10081014 in WAsyncT $ K.concatMapWith wAsyncK f m1

10091015

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

436436--

437437-- @since 0.8.0

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

439− deriving (MonadTrans)

439+

440+instance MonadTrans ParallelT where

441+ {-# INLINE lift #-}

442+ lift = ParallelT . K.fromEffect

440443

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

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

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

476479apParallel :: MonadAsync m =>

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

478481apParallel (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

480483 in ParallelT $ K.concatMapWith parallelK f m1

481484

482485instance (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)

6969import qualified Streamly.Data.Fold as Fold

7070import qualified Streamly.Data.Stream as Stream

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

72− (toStreamD, fromStreamD, toStreamK, fromStreamK)

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

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

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

7575 (fromStreamK, toStreamK, mapM)

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

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

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

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

7979

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

8181

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

9999-- Applicative - orphan instances

100100------------------------------------------------------------------------------

101101

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−

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

118103-- INLINE them.

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

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

121106 pure = Stream.fromStreamK . K.fromPure

122107

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

124− (<*>) = apSerial

109+ (<*>) = Stream.crossApply

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

126111

127112 {-# INLINE liftA2 #-}

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

129114

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

131− (*>) = apSequence

116+ (*>) = Stream.crossApplySnd

132117 -- (*>) = K.apSerialDiscardFst

133118

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

135− (<*) = apDiscardSnd

120+ (<*) = Stream.crossApplyFst

136121 -- (<*) = K.apSerialDiscardSnd

137122

138123------------------------------------------------------------------------------

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

288273--

289274-- @since 0.8.0

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

291− deriving (MonadTrans)

276+

277+instance MonadTrans WSerialT where

278+ {-# INLINE lift #-}

279+ lift = WSerialT . K.fromEffect

292280

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

294282-- documentation for more details.

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

353341 mappend = (<>)

354342

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

357345apWSerial (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

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

360348

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