55d49bd · streamly · git.iko.soy

git.iko.soy/streamly

Optimize Functor/Applicative/Monad etc for serial

Some of the benchmarks were order of magnitude off due to missing INLINE for type class operations. Now, all of them are in reasonable limits. Benchmarks affected for serial streams: * Functor, Applicative, Monad, transformers We need to do a similar exercise for other types of streams and for folds/parsers as well.

commit55d49bd50cc88f080295791d5ed47e691677ec09 authorHarendra Kumar <harendra@composewell.com> dateTue 23 Jun 2020, 12:31 (6 years ago) parentb4cef11

11 files changed+349−33

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Prelude/Serial.hs+20−15 benchmark/lib/Streamly/Benchmark/Prelude.hs+49 benchmark/streamly-benchmarks.cabal+1−3 bin/bench-exec-one.sh+22−6 src/Streamly/Internal/Data/Stream/Instances.hs+9−4 src/Streamly/Internal/Data/Stream/Serial.hs+27−2 src/Streamly/Internal/Data/Stream/StreamD/Type.hs+45−2 src/Streamly/Internal/Data/Stream/StreamK.hs+4 src/Streamly/Internal/Data/Stream/StreamK/Type.hs+117 src/Streamly/Internal/Data/Unfold.hs+31 test/Prop.hs+24−1

benchmark/Streamly/Benchmark/Prelude/Serial.hs+20−15

@@ -27,6 +27,7 @@ import Control.Monad (when)

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

2828import Control.Monad.State.Strict (StateT, get, put, MonadState)

2929import qualified Control.Monad.State.Strict as State

30+import Control.Monad.Trans.Class (lift)

3031import Data.Functor.Identity (Identity, runIdentity)

3132import Data.IORef (newIORef, modifyIORef')

3233import GHC.Generics (Generic)

@@ -1610,8 +1611,11 @@ o_1_space_concat value = sqrtVal `seq`

16101611

16111612o_1_space_applicative :: Int -> [Benchmark]

16121613o_1_space_applicative value =

1613− [ bgroup "applicative"

1614− [ benchIO "outer product (sqrt n x sqrt n)" $ toNullAp value serially

1614+ [ bgroup "Applicative"

1615+ [ benchIO "(*>) (sqrt n x sqrt n)" $ apDiscardFst value serially

1616+ , benchIO "(<*) (sqrt n x sqrt n)" $ apDiscardSnd value serially

1617+ , benchIO "(<*>) (sqrt n x sqrt n)" $ toNullAp value serially

1618+ , benchIO "liftA2 (sqrt n x sqrt n)" $ apLiftA2 value serially

16151619 ]

16161620 ]

16171621

@@ -1636,16 +1640,17 @@ o_n_space_applicative value =

16361640o_1_space_monad :: Int -> [Benchmark]

16371641o_1_space_monad value =

16381642 [ bgroup "Monad"

1639− [ benchIO "outer product (sqrt n x sqrt n)" $ toNullM value serially

1640− , benchIO "outer product (sqrt n x sqrt n) (filterAllOut)" $

1643+ [ benchIO "(>>) (sqrt n x sqrt n)" $ monadThen value serially

1644+ , benchIO "(>>=) (sqrt n x sqrt n)" $ toNullM value serially

1645+ , benchIO "(>>=) (sqrt n x sqrt n) (filterAllOut)" $

16411646 filterAllOutM value serially

1642− , benchIO "outer product (sqrt n x sqrt n) (filterAllIn)" $

1647+ , benchIO "(>>=) (sqrt n x sqrt n) (filterAllIn)" $

16431648 filterAllInM value serially

1644− , benchIO "outer product (sqrt n x sqrt n) (filterSome)" $

1649+ , benchIO "(>>=) (sqrt n x sqrt n) (filterSome)" $

16451650 filterSome value serially

1646− , benchIO "outer product (sqrt n x sqrt n) (breakAfterSome)" $

1651+ , benchIO "(>>=) (sqrt n x sqrt n) (breakAfterSome)" $

16471652 breakAfterSome value serially

1648− , benchIO "outer product (cubert n x cubert n x cubert n)" $

1653+ , benchIO "(>>=) (cubert n x cubert n x cubert n)" $

16491654 toNullM3 value serially

16501655 ]

16511656 ]

@@ -1657,9 +1662,9 @@ o_n_space_monad value =

16571662 iterateSingleton ((>>) . pure) value

16581663 , benchIOSrc serially "(>>=) (n times)" $

16591664 iterateSingleton (\x xs -> xs >>= \y -> return (x + y)) value

1660− , benchIO "outer product (sqrt n x sqrt n) (toList)" $

1665+ , benchIO "(>>=) (sqrt n x sqrt n) (toList)" $

16611666 toListM value serially

1662− , benchIO "outer product (sqrt n x sqrt n) (toListSome)" $

1667+ , benchIO "(>>=) (sqrt n x sqrt n) (toListSome)" $

16631668 toListSome value serially

16641669 ]

16651670 ]

@@ -1714,10 +1719,10 @@ iterateStateIO n = do

17141719{-# INLINE iterateStateT #-}

17151720iterateStateT :: Int -> SerialT (StateT Int IO) Int

17161721iterateStateT n = do

1717− x <- get

1722+ x <- lift get

17181723 if x > n

17191724 then do

1720− put (x - 1)

1725+ lift $ put (x - 1)

17211726 iterateStateT n

17221727 else return x

17231728

@@ -1734,8 +1739,8 @@ iterateState n = do

17341739 iterateState n

17351740 else return x

17361741

1737−o_n_space_transformer :: Int -> [Benchmark]

1738−o_n_space_transformer value =

1742+o_n_heap_transformer :: Int -> [Benchmark]

1743+o_n_heap_transformer value =

17391744 [ bgroup "transformer"

17401745 [ benchIO "StateT Int IO (n times) (baseline)" $ \n ->

17411746 State.evalStateT (iterateStateIO n) value

@@ -1805,6 +1810,7 @@ main = do

18051810

18061811 -- transformation

18071812 , o_n_heap_reordering size

1813+ , o_n_heap_transformer size

18081814 ]

18091815 , bgroup (o_n_space_prefix moduleName) $ Prelude.concat

18101816 [ o_n_space_elimination_foldable size

@@ -1819,6 +1825,5 @@ main = do

18191825 , o_n_space_functor size

18201826 , o_n_space_applicative size

18211827 , o_n_space_monad size

1822− , o_n_space_transformer size

18231828 ]

18241829 ]

benchmark/lib/Streamly/Benchmark/Prelude.hs+49

@@ -12,6 +12,7 @@

1212

1313module Streamly.Benchmark.Prelude where

1414

15+import Control.Applicative (liftA2)

1516import Control.DeepSeq (NFData(..))

1617import Control.Exception (try)

1718import Data.Functor.Identity (Identity)

@@ -345,6 +346,42 @@ concatStreamsWith op outer inner n =

345346runToList :: Monad m => S.SerialT m a -> m [a]

346347runToList = S.toList

347348

349+{-# INLINE apDiscardFst #-}

350+apDiscardFst

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

352+ => Int -> (t m Int -> S.SerialT m Int) -> Int -> m ()

353+apDiscardFst linearCount t start = S.drain . t $

354+ S.serially (sourceUnfoldrM nestedCount2 start)

355+ *> S.serially (sourceUnfoldrM nestedCount2 start)

356+

357+ where

358+

359+ nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

360+

361+{-# INLINE apDiscardSnd #-}

362+apDiscardSnd

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

364+ => Int -> (t m Int -> S.SerialT m Int) -> Int -> m ()

365+apDiscardSnd linearCount t start = S.drain . t $

366+ S.serially (sourceUnfoldrM nestedCount2 start)

367+ <* S.serially (sourceUnfoldrM nestedCount2 start)

368+

369+ where

370+

371+ nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

372+

373+{-# INLINE apLiftA2 #-}

374+apLiftA2

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

376+ => Int -> (t m Int -> S.SerialT m Int) -> Int -> m ()

377+apLiftA2 linearCount t start = S.drain . t $

378+ liftA2 (+) (S.serially (sourceUnfoldrM nestedCount2 start))

379+ (S.serially (sourceUnfoldrM nestedCount2 start))

380+

381+ where

382+

383+ nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

384+

348385{-# INLINE toNullAp #-}

349386toNullAp

350387 :: (S.IsStream t, S.MonadAsync m, Applicative (t m))

@@ -357,6 +394,18 @@ toNullAp linearCount t start = S.drain . t $

357394

358395 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

359396

397+{-# INLINE monadThen #-}

398+monadThen

399+ :: (S.IsStream t, S.MonadAsync m, Monad (t m))

400+ => Int -> (t m Int -> S.SerialT m Int) -> Int -> m ()

401+monadThen linearCount t start = S.drain . t $ do

402+ (S.serially $ sourceUnfoldrM nestedCount2 start) >>

403+ (S.serially $ sourceUnfoldrM nestedCount2 start)

404+

405+ where

406+

407+ nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

408+

360409{-# INLINE toNullM #-}

361410toNullM

362411 :: (S.IsStream t, S.MonadAsync m, Monad (t m))

benchmark/streamly-benchmarks.cabal+1−3

@@ -108,12 +108,10 @@ common bench-depends

108108 , streamly >= 0.7.0

109109 , random >= 1.0 && < 2.0

110110 , gauge >= 0.2.4 && < 0.3

111+ , transformers >= 0.4 && < 0.6

111112 if flag(fusion-plugin) && !impl(ghcjs) && !impl(ghc < 8.6)

112113 build-depends:

113114 fusion-plugin >= 0.2 && < 0.3

114− if impl(ghc < 8.0)

115− build-depends:

116− transformers >= 0.4 && < 0.6

117115 if flag(inspection)

118116 build-depends: template-haskell >= 2.14 && < 2.17

119117 , inspection-testing >= 0.4 && < 0.5

bin/bench-exec-one.sh+22−6

@@ -5,6 +5,19 @@

55# RTS_OPTIONS: additional RTS options

66# QUICK_MODE: whether we are in quick mode

8+# $1: message

9+die () {

10+ >&2 echo -e "Error: $1"

11+ exit 1

12+}

13+

14+warn () {

15+ >&2 echo -e "Warning: $1"

16+}

17+

18+test -n "$BENCH_EXEC_PATH" || die "BENCH_EXEC_PATH env var must be set"

19+test -n "$QUICK_MODE" || warn "QUICK_MODE env var not set (to 0 or 1)"

20+

821#------------------------------------------------------------------------------

922# RTS Options

1023#------------------------------------------------------------------------------

@@ -35,10 +48,10 @@ bench_rts_opts_specific () {

3548 Prelude.Parallel/o-n-heap/monad-outer-product/*) echo -n "-M256M" ;;

3649 Prelude.Parallel/o-n-space/monad-outer-product/*) echo -n "-K4M -M256M" ;;

3750

38− Prelude.Serial/o-n-space/Functor/*) echo -n "-K4M -M256M" ;;

39− Prelude.Serial/o-n-space/Applicative/*) echo -n "-K8M -M256M" ;;

40− Prelude.Serial/o-n-space/Monad/*) echo -n "-K8M -M256M" ;;

41− Prelude.Serial/o-n-space/transformer/*) echo -n "-K8M -M256M" ;;

51+ Prelude.Serial/o-n-space/Functor/*) echo -n "-K4M -M64M" ;;

52+ Prelude.Serial/o-n-space/Applicative/*) echo -n "-K8M -M128M" ;;

53+ Prelude.Serial/o-n-space/Monad/*) echo -n "-K8M -M64M" ;;

54+ Prelude.Serial/o-n-heap/transformer/*) echo -n "-M64M" ;;

4255 Prelude.Serial/o-n-space/grouping/*) echo -n "" ;;

4356 Prelude.Serial/o-n-space/*) echo -n "-K4M" ;;

4457

@@ -65,9 +78,12 @@ bench_rts_opts_specific () {

6578# Speed options

6679#------------------------------------------------------------------------------

6780

68−if test "$QUICK_MODE" -eq 0

81+if test -n "$QUICK_MODE"

6982then

70−QUICK_OPTIONS="--min-samples 3"

83+ if test "$QUICK_MODE" -eq 0

84+ then

85+ QUICK_OPTIONS="--min-samples 3"

86+ fi

7187fi

7288SUPER_QUICK_OPTIONS="--min-samples 1 --include-first-iter"

7389

src/Streamly/Internal/Data/Stream/Instances.hs+9−4

@@ -7,10 +7,12 @@

88#define MONADPARALLEL , MonadAsync m

10−#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

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

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

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

10+#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

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

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

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

14+ {-# INLINE (<$) #-}; \

15+ (<$) = fmap . const }; \

1416 \

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

1618 liftBase = liftBaseDefault }; \

@@ -33,8 +35,11 @@ instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \

3335 local f m = fromStream $ K.withLocal f (toStream m) }; \

3436 \

3537instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \

38+ {-# INLINE get #-}; \

3639 get = lift get; \

40+ {-# INLINE put #-}; \

3741 put x = lift (put x); \

42+ {-# INLINE state #-}; \

3843 state k = lift (state k) }

3944

4045------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/Serial.hs+27−2

@@ -146,6 +146,7 @@ import qualified Streamly.Internal.Data.Stream.StreamD as D

146146--

147147-- @since 0.2.0

148148newtype SerialT m a = SerialT {getSerialT :: Stream m a}

149+ -- XXX when deriving do we inherit an INLINE?

149150 deriving (Semigroup, Monoid, MonadTrans)

150151

151152-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation

@@ -182,8 +183,10 @@ instance IsStream SerialT where

182183

183184instance Monad m => Monad (SerialT m) where

184185 return = pure

186+

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

186188 (>>=) = K.bindWith K.serial

189+

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

188191 (>>) = (*>)

189192

@@ -218,19 +221,41 @@ map f = mapM (return . f)

218221

219222{-# INLINE apSerial #-}

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

221−apSerial (SerialT m1) (SerialT m2) = D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2

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

225+ D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2

222226

223227{-# INLINE apSequence #-}

224228apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b

225−apSequence (SerialT m1) (SerialT m2) = D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2

229+apSequence (SerialT m1) (SerialT m2) =

230+ D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2

226231

232+{-# INLINE apDiscardSnd #-}

233+apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a

234+apDiscardSnd (SerialT m1) (SerialT m2) =

235+ D.fromStreamD $ D.toStreamD m1 <* D.toStreamD m2

236+

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

238+-- INLINE them.

227239instance Monad m => Applicative (SerialT m) where

228240 {-# INLINE pure #-}

229241 pure = SerialT . K.yield

242+

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

231244 (<*>) = apSerial

245+ -- (<*>) = K.apSerial

246+

247+#if MIN_VERSION_base(4,10,0)

248+ {-# INLINE liftA2 #-}

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

250+#endif

251+

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

233253 (*>) = apSequence

254+ -- (*>) = K.apSerialDiscardFst

255+

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

257+ (<*) = apDiscardSnd

258+ -- (<*) = K.apSerialDiscardSnd

234259

235260MONAD_COMMON_INSTANCES(SerialT,)

236261LIST_INSTANCES(SerialT)

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

@@ -182,6 +182,9 @@ instance Functor m => Functor (Stream m) where

182182 {-# INLINE_LATE step' #-}

183183 step' gst st = fmap (fmap f) (step (adaptState gst) st)

184184

185+ {-# INLINE (<$) #-}

186+ (<$) = fmap . const

187+

185188------------------------------------------------------------------------------

186189-- concatMap

187190------------------------------------------------------------------------------

@@ -260,8 +263,11 @@ concatAp (Stream stepa statea) (Stream stepb stateb) = Stream step' (Left statea

260263

261264{-# INLINE_NORMAL apSequence #-}

262265apSequence :: Functor f => Stream f a -> Stream f b -> Stream f b

263−apSequence (Stream stepa statea) (Stream stepb stateb) = Stream step (Left statea)

264− where

266+apSequence (Stream stepa statea) (Stream stepb stateb) =

267+ Stream step (Left statea)

268+

269+ where

270+

265271 {-# INLINE_LATE step #-}

266272 step gst (Left st) =

267273 fmap

@@ -280,26 +286,63 @@ apSequence (Stream stepa statea) (Stream stepb stateb) = Stream step (Left state

280286 Stop -> Skip (Left ostate))

281287 (stepb gst st)

282288

289+{-# INLINE_NORMAL apDiscardSnd #-}

290+apDiscardSnd :: Functor f => Stream f a -> Stream f b -> Stream f a

291+apDiscardSnd (Stream stepa statea) (Stream stepb stateb) =

292+ Stream step (Left statea)

293+

294+ where

295+

296+ {-# INLINE_LATE step #-}

297+ step gst (Left st) =

298+ fmap

299+ (\r ->

300+ case r of

301+ Yield b s -> Skip (Right (s, stateb, b))

302+ Skip s -> Skip (Left s)

303+ Stop -> Stop)

304+ (stepa gst st)

305+ step gst (Right (ostate, st, b)) =

306+ fmap

307+ (\r ->

308+ case r of

309+ Yield _ s -> Yield b (Right (ostate, s, b))

310+ Skip s -> Skip (Right (ostate, s, b))

311+ Stop -> Skip (Left ostate))

312+ (stepb (adaptState gst) st)

313+

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

284315 {-# INLINE pure #-}

285316 pure = yield

317+

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

287319 (<*>) = concatAp

320+

321+#if MIN_VERSION_base(4,10,0)

322+ {-# INLINE liftA2 #-}

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

324+#endif

325+

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

289327 (*>) = apSequence

290328

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

330+ (<*) = apDiscardSnd

291331

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

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

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

295335 {-# INLINE return #-}

296336 return = pure

337+

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

298339 (>>=) = flip concatMap

340+

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

300342 (>>) = (*>)

301343

302344instance MonadTrans Stream where

345+ {-# INLINE lift #-}

303346 lift = yieldM

304347

305348instance (MonadThrow m) => MonadThrow (Stream m) where

src/Streamly/Internal/Data/Stream/StreamK.hs+4

@@ -171,6 +171,10 @@ module Streamly.Internal.Data.Stream.StreamK

171171 , concatMapBy

172172 , concatMap

173173 , bindWith

174+ , apWith

175+ , apSerial

176+ , apSerialDiscardFst

177+ , apSerialDiscardSnd

174178

175179 -- ** Transformation comprehensions

176180 , the

src/Streamly/Internal/Data/Stream/StreamK/Type.hs+117

@@ -80,6 +80,10 @@ module Streamly.Internal.Data.Stream.StreamK.Type

8080 , concatMapBy

8181 , concatMap

8282 , bindWith

83+ , apWith

84+ , apSerial

85+ , apSerialDiscardFst

86+ , apSerialDiscardSnd

8387

8488 , Streaming -- deprecated

8589 )

@@ -886,6 +890,7 @@ instance Monad m => Functor (Stream m) where

886890-------------------------------------------------------------------------------

887891

888892instance MonadTrans Stream where

893+ {-# INLINE lift #-}

889894 lift = yieldM

890895

891896-------------------------------------------------------------------------------

@@ -898,6 +903,118 @@ unShare :: IsStream t => t m a -> t m a

898903unShare x = mkStream $ \st yld sng stp ->

899904 foldStream st yld sng stp x

900905

906+-- XXX the function stream and value stream can run in parallel

907+{-# INLINE apWith #-}

908+apWith

909+ :: IsStream t

910+ => (t m b -> t m b -> t m b)

911+ -> t m (a -> b)

912+ -> t m a

913+ -> t m b

914+apWith par fstream stream = go1 fstream

915+

916+ where

917+

918+ go1 m =

919+ mkStream $ \st yld sng stp ->

920+ let foldShared = foldStreamShared st yld sng stp

921+ single f = foldShared $ unShare (go2 f stream)

922+ yieldk f r = foldShared $ unShare (go2 f stream) `par` go1 r

923+ in foldStream (adaptState st) yieldk single stp m

924+

925+ go2 f m =

926+ mkStream $ \st yld sng stp ->

927+ let single a = sng (f a)

928+ yieldk a r = yld (f a) (go2 f r)

929+ in foldStream (adaptState st) yieldk single stp m

930+

931+{-# INLINE apSerial #-}

932+apSerial

933+ :: IsStream t

934+ => t m (a -> b)

935+ -> t m a

936+ -> t m b

937+apSerial fstream stream = go1 fstream

938+

939+ where

940+

941+ go1 m =

942+ mkStream $ \st yld sng stp ->

943+ let foldShared = foldStreamShared st yld sng stp

944+ single f = foldShared $ go3 f stream

945+ yieldk f r = foldShared $ go2 f r stream

946+ in foldStream (adaptState st) yieldk single stp m

947+

948+ go2 f r1 m =

949+ mkStream $ \st yld sng stp ->

950+ let foldShared = foldStreamShared st yld sng stp

951+ stop = foldShared $ go1 r1

952+ single a = yld (f a) (go1 r1)

953+ yieldk a r = yld (f a) (go2 f r1 r)

954+ in foldStream (adaptState st) yieldk single stop m

955+

956+ go3 f m =

957+ mkStream $ \st yld sng stp ->

958+ let single a = sng (f a)

959+ yieldk a r = yld (f a) (go3 f r)

960+ in foldStream (adaptState st) yieldk single stp m

961+

962+{-# INLINE apSerialDiscardFst #-}

963+apSerialDiscardFst

964+ :: IsStream t

965+ => t m a

966+ -> t m b

967+ -> t m b

968+apSerialDiscardFst fstream stream = go1 fstream

969+

970+ where

971+

972+ go1 m =

973+ mkStream $ \st yld sng stp ->

974+ let foldShared = foldStreamShared st yld sng stp

975+ single _ = foldShared $ stream

976+ yieldk _ r = foldShared $ go2 r stream

977+ in foldStream (adaptState st) yieldk single stp m

978+

979+ go2 r1 m =

980+ mkStream $ \st yld sng stp ->

981+ let foldShared = foldStreamShared st yld sng stp

982+ stop = foldShared $ go1 r1

983+ single a = yld a (go1 r1)

984+ yieldk a r = yld a (go2 r1 r)

985+ in foldStream st yieldk single stop m

986+

987+{-# INLINE apSerialDiscardSnd #-}

988+apSerialDiscardSnd

989+ :: IsStream t

990+ => t m a

991+ -> t m b

992+ -> t m a

993+apSerialDiscardSnd fstream stream = go1 fstream

994+

995+ where

996+

997+ go1 m =

998+ mkStream $ \st yld sng stp ->

999+ let foldShared = foldStreamShared st yld sng stp

1000+ single f = foldShared $ go3 f stream

1001+ yieldk f r = foldShared $ go2 f r stream

1002+ in foldStream st yieldk single stp m

1003+

1004+ go2 f r1 m =

1005+ mkStream $ \st yld sng stp ->

1006+ let foldShared = foldStreamShared st yld sng stp

1007+ stop = foldShared $ go1 r1

1008+ single _ = yld f (go1 r1)

1009+ yieldk _ r = yld f (go2 f r1 r)

1010+ in foldStream (adaptState st) yieldk single stop m

1011+

1012+ go3 f m =

1013+ mkStream $ \st yld sng stp ->

1014+ let single _ = sng f

1015+ yieldk _ r = yld f (go3 f r)

1016+ in foldStream (adaptState st) yieldk single stp m

1017+

9011018-- XXX This is just concatMapBy with arguments flipped. We need to keep this

9021019-- instead of using a concatMap style definition because the bind

9031020-- implementation in Async and WAsync streams show significant perf degradation

src/Streamly/Internal/Data/Unfold.hs+31

@@ -115,6 +115,9 @@ module Streamly.Internal.Data.Unfold

115115 , concat

116116 , concatMapM

117117 , outerProduct

118+ , ap

119+ , apDiscardFst

120+ , apDiscardSnd

118121

119122 -- * Exceptions

120123 , gbracket

@@ -689,6 +692,34 @@ outerProduct (Unfold step1 inject1) (Unfold step2 inject2) = Unfold step inject

689692 Skip s -> Skip (OuterProductInner ost sy s x)

690693 Stop -> Skip (OuterProductOuter ost sy)

691694

695+-- Special cases of outer product

696+-- | Outer product with a function application.

697+--

698+-- /Unimplemented/

699+--

700+{-# INLINE_NORMAL ap #-}

701+ap :: -- Monad m =>

702+ Unfold m a (b -> c) -> Unfold m d b -> Unfold m (a, d) c

703+ap (Unfold _step1 _inject1) (Unfold _step2 _inject2) = undefined

704+

705+-- | Outer product discarding the first element.

706+--

707+-- /Unimplemented/

708+--

709+{-# INLINE_NORMAL apDiscardFst #-}

710+apDiscardFst :: -- Monad m =>

711+ Unfold m a b -> Unfold m c d -> Unfold m (a, c) d

712+apDiscardFst (Unfold _step1 _inject1) (Unfold _step2 _inject2) = undefined

713+

714+-- | Outer product discarding the second element.

715+--

716+-- /Unimplemented/

717+--

718+{-# INLINE_NORMAL apDiscardSnd #-}

719+apDiscardSnd :: -- Monad m =>

720+ Unfold m a b -> Unfold m c d -> Unfold m (a, c) b

721+apDiscardSnd (Unfold _step1 _inject1) (Unfold _step2 _inject2) = undefined

722+

692723-- XXX This can be used to implement a Monad instance for "Unfold m ()".

693724

694725data ConcatMapState s1 s2 = ConcatMapOuter s1 | ConcatMapInner s1 s2

test/Prop.hs+24−1

@@ -4,7 +4,7 @@

55module Main (main) where

7−import Control.Applicative (ZipList(..))

7+import Control.Applicative (ZipList(..), liftA2)

88import Control.Concurrent (MVar, takeMVar, putMVar, newEmptyMVar)

99import Control.Exception

1010 (BlockedIndefinitelyOnMVar(..), catches,

@@ -880,6 +880,28 @@ applicativeOps constr eq t (a, b) = withMaxSuccess maxTestCount $

880880 let list = (,) <$> a <*> b

881881 listEquals eq stream list

882882

883+ stream1 <- run ((S.toList . t) (liftA2 (,) (constr a) (constr b)))

884+ listEquals eq stream1 list

885+

886+-- XXX we can combine this with applicativeOps by making the type sufficiently

887+-- polymorphic.

888+applicativeOps1

889+ :: Applicative (t IO)

890+ => ([Int] -> t IO Int)

891+ -> ([Int] -> [Int] -> Bool)

892+ -> (t IO Int -> SerialT IO Int)

893+ -> ([Int], [Int])

894+ -> Property

895+applicativeOps1 constr eq t (a, b) = withMaxSuccess maxTestCount $

896+ monadicIO $ do

897+ stream <- run ((S.toList . t) (constr a *> constr b))

898+ let list = a *> b

899+ listEquals eq stream list

900+

901+ stream1 <- run ((S.toList . t) (constr a <* constr b))

902+ let list1 = a <* b

903+ listEquals eq stream1 list1

904+

883905-------------------------------------------------------------------------------

884906-- Zip operations

885907-------------------------------------------------------------------------------

@@ -1139,6 +1161,7 @@ main = hspec

11391161 -- XXX applicative with three arguments

11401162 serialOps $ prop "serially applicative" . applicativeOps S.fromFoldable (==)

11411163 serialOps $ prop "serially applicative folded" . applicativeOps folded (==)

1164+ serialOps $ prop "serially applicative discard" . applicativeOps1 S.fromFoldable (==)

11421165 wSerialOps $ prop "wSerially applicative" . applicativeOps S.fromFoldable sortEq

11431166 wSerialOps $ prop "wSerially applicative folded" . applicativeOps folded sortEq

11441167 aheadOps $ prop "aheadly applicative" . applicativeOps S.fromFoldable (==)