55d49bd · streamly · git.iko.soy

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

compare with master

unifiedsplit

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

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

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

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

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

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

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

31import Data.IORef (newIORef, modifyIORef')

32import Data.IORef (newIORef, modifyIORef')

32import GHC.Generics (Generic)

33import GHC.Generics (Generic)

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

1610

1611

1611o_1_space_applicative :: Int -> [Benchmark]

1612o_1_space_applicative :: Int -> [Benchmark]

1612o_1_space_applicative value =

1613o_1_space_applicative value =

1613− [ bgroup "applicative"

1614+ [ bgroup "Applicative"

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

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

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

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

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

1615 ]

1619 ]

1616 ]

1620 ]

1617

1621

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

1636o_1_space_monad :: Int -> [Benchmark]

1640o_1_space_monad :: Int -> [Benchmark]

1637o_1_space_monad value =

1641o_1_space_monad value =

1638 [ bgroup "Monad"

1642 [ bgroup "Monad"

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

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

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

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

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

1641 filterAllOutM value serially

1646 filterAllOutM value serially

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

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

1643 filterAllInM value serially

1648 filterAllInM value serially

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

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

1645 filterSome value serially

1650 filterSome value serially

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

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

1647 breakAfterSome value serially

1652 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)" $

1649 toNullM3 value serially

1654 toNullM3 value serially

1650 ]

1655 ]

1651 ]

1656 ]

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

1657 iterateSingleton ((>>) . pure) value

1662 iterateSingleton ((>>) . pure) value

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

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

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

1664 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)" $

1661 toListM value serially

1666 toListM value serially

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

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

1663 toListSome value serially

1668 toListSome value serially

1664 ]

1669 ]

1665 ]

1670 ]

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

1714{-# INLINE iterateStateT #-}

1719{-# INLINE iterateStateT #-}

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

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

1716iterateStateT n = do

1721iterateStateT n = do

1717− x <- get

1722+ x <- lift get

1718 if x > n

1723 if x > n

1719 then do

1724 then do

1720− put (x - 1)

1725+ lift $ put (x - 1)

1721 iterateStateT n

1726 iterateStateT n

1722 else return x

1727 else return x

1723

1728

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

1734 iterateState n

1739 iterateState n

1735 else return x

1740 else return x

1736

1741

1737−o_n_space_transformer :: Int -> [Benchmark]

1742+o_n_heap_transformer :: Int -> [Benchmark]

1738−o_n_space_transformer value =

1743+o_n_heap_transformer value =

1739 [ bgroup "transformer"

1744 [ bgroup "transformer"

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

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

1741 State.evalStateT (iterateStateIO n) value

1746 State.evalStateT (iterateStateIO n) value

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

1805

1810

1806 -- transformation

1811 -- transformation

1807 , o_n_heap_reordering size

1812 , o_n_heap_reordering size

1813+ , o_n_heap_transformer size

1808 ]

1814 ]

1809 , bgroup (o_n_space_prefix moduleName) $ Prelude.concat

1815 , bgroup (o_n_space_prefix moduleName) $ Prelude.concat

1810 [ o_n_space_elimination_foldable size

1816 [ o_n_space_elimination_foldable size

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

1819 , o_n_space_functor size

1825 , o_n_space_functor size

1820 , o_n_space_applicative size

1826 , o_n_space_applicative size

1821 , o_n_space_monad size

1827 , o_n_space_monad size

1822− , o_n_space_transformer size

1823 ]

1828 ]

1824 ]

1829 ]

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

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

12

13module Streamly.Benchmark.Prelude where

14

15+import Control.Applicative (liftA2)

15import Control.DeepSeq (NFData(..))

16import Control.DeepSeq (NFData(..))

16import Control.Exception (try)

17import Control.Exception (try)

17import Data.Functor.Identity (Identity)

18import Data.Functor.Identity (Identity)

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

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

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

346runToList = S.toList

347runToList = S.toList

347

348

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+

348{-# INLINE toNullAp #-}

385{-# INLINE toNullAp #-}

349toNullAp

386toNullAp

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

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

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

357

394

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

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

359

396

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+

360{-# INLINE toNullM #-}

409{-# INLINE toNullM #-}

361toNullM

410toNullM

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

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

benchmark/streamly-benchmarks.cabal+1−3

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

108 , streamly >= 0.7.0

109 , random >= 1.0 && < 2.0

110 , gauge >= 0.2.4 && < 0.3

111+ , transformers >= 0.4 && < 0.6

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

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

112 build-depends:

113 build-depends:

113 fusion-plugin >= 0.2 && < 0.3

114 fusion-plugin >= 0.2 && < 0.3

114− if impl(ghc < 8.0)

115− build-depends:

116− transformers >= 0.4 && < 0.6

117 if flag(inspection)

115 if flag(inspection)

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

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

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

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

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

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

5# RTS_OPTIONS: additional RTS options

6# QUICK_MODE: whether we are in quick mode

7

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+

8#------------------------------------------------------------------------------

21#------------------------------------------------------------------------------

9# RTS Options

22# RTS Options

10#------------------------------------------------------------------------------

23#------------------------------------------------------------------------------

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

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

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

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

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

37

50

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

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

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

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

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

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

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

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

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

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

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

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

44

57

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

65# Speed options

78# Speed options

66#------------------------------------------------------------------------------

79#------------------------------------------------------------------------------

67

80

68−if test "$QUICK_MODE" -eq 0

81+if test -n "$QUICK_MODE"

69then

82then

83+ if test "$QUICK_MODE" -eq 0

84+ then

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

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

86+ fi

71fi

87fi

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

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

73

89

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

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

7

8#define MONADPARALLEL , MonadAsync m

9

10−#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

10+#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

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

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

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

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

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

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

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

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

14 \

16 \

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

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

16 liftBase = liftBaseDefault }; \

18 liftBase = liftBaseDefault }; \

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

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

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

34 \

36 \

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

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

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

36 get = lift get; \

39 get = lift get; \

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

37 put x = lift (put x); \

41 put x = lift (put x); \

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

38 state k = lift (state k) }

43 state k = lift (state k) }

39

44

40------------------------------------------------------------------------------

45------------------------------------------------------------------------------

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

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

146--

147-- @since 0.2.0

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

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

149 deriving (Semigroup, Monoid, MonadTrans)

150 deriving (Semigroup, Monoid, MonadTrans)

150

151

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

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

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

182

183

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

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

184 return = pure

185 return = pure

186+

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

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

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

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

189+

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

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

188 (>>) = (*>)

191 (>>) = (*>)

189

192

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

218

221

219{-# INLINE apSerial #-}

222{-# INLINE apSerial #-}

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

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

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

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

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

222

226

223{-# INLINE apSequence #-}

227{-# INLINE apSequence #-}

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

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

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

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

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

226

231

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.

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

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

228 {-# INLINE pure #-}

240 {-# INLINE pure #-}

229 pure = SerialT . K.yield

241 pure = SerialT . K.yield

242+

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

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

231 (<*>) = apSerial

244 (<*>) = 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+

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

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

233 (*>) = apSequence

253 (*>) = apSequence

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

255+

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

257+ (<*) = apDiscardSnd

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

234

259

235MONAD_COMMON_INSTANCES(SerialT,)

260MONAD_COMMON_INSTANCES(SerialT,)

236LIST_INSTANCES(SerialT)

261LIST_INSTANCES(SerialT)

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

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

182 {-# INLINE_LATE step' #-}

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

184

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

186+ (<$) = fmap . const

187+

185------------------------------------------------------------------------------

188------------------------------------------------------------------------------

186-- concatMap

189-- concatMap

187------------------------------------------------------------------------------

190------------------------------------------------------------------------------

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

260

263

261{-# INLINE_NORMAL apSequence #-}

264{-# INLINE_NORMAL apSequence #-}

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

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

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

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

267+ Stream step (Left statea)

268+

264− where

269+ where

270+

265 {-# INLINE_LATE step #-}

271 {-# INLINE_LATE step #-}

266 step gst (Left st) =

272 step gst (Left st) =

267 fmap

273 fmap

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

280 Stop -> Skip (Left ostate))

286 Stop -> Skip (Left ostate))

281 (stepb gst st)

287 (stepb gst st)

282

288

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+

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

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

284 {-# INLINE pure #-}

315 {-# INLINE pure #-}

285 pure = yield

316 pure = yield

317+

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

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

287 (<*>) = concatAp

319 (<*>) = concatAp

320+

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

322+ {-# INLINE liftA2 #-}

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

324+#endif

325+

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

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

289 (*>) = apSequence

327 (*>) = apSequence

290

328

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

330+ (<*) = apDiscardSnd

291

331

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

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

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

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

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

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

295 {-# INLINE return #-}

335 {-# INLINE return #-}

296 return = pure

336 return = pure

337+

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

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

298 (>>=) = flip concatMap

339 (>>=) = flip concatMap

340+

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

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

300 (>>) = (*>)

342 (>>) = (*>)

301

343

302instance MonadTrans Stream where

344instance MonadTrans Stream where

345+ {-# INLINE lift #-}

303 lift = yieldM

346 lift = yieldM

304

347

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

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

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

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

171 , concatMapBy

172 , concatMap

173 , bindWith

174+ , apWith

175+ , apSerial

176+ , apSerialDiscardFst

177+ , apSerialDiscardSnd

174

178

175 -- ** Transformation comprehensions

179 -- ** Transformation comprehensions

176 , the

180 , the

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

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

80 , concatMapBy

81 , concatMap

82 , bindWith

83+ , apWith

84+ , apSerial

85+ , apSerialDiscardFst

86+ , apSerialDiscardSnd

83

87

84 , Streaming -- deprecated

88 , Streaming -- deprecated

85 )

89 )

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

886-------------------------------------------------------------------------------

890-------------------------------------------------------------------------------

887

891

888instance MonadTrans Stream where

892instance MonadTrans Stream where

893+ {-# INLINE lift #-}

889 lift = yieldM

894 lift = yieldM

890

895

891-------------------------------------------------------------------------------

896-------------------------------------------------------------------------------

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

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

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

899 foldStream st yld sng stp x

904 foldStream st yld sng stp x

900

905

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+

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

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

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

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

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

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

115 , concat

116 , concatMapM

117 , outerProduct

118+ , ap

119+ , apDiscardFst

120+ , apDiscardSnd

118

121

119 -- * Exceptions

122 -- * Exceptions

120 , gbracket

123 , gbracket

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

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

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

690 Stop -> Skip (OuterProductOuter ost sy)

693 Stop -> Skip (OuterProductOuter ost sy)

691

694

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+

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

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

693

724

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

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

test/Prop.hs+24−1

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

4

5module Main (main) where

6

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

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

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

9import Control.Exception

10 (BlockedIndefinitelyOnMVar(..), catches,

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

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

881 listEquals eq stream list

882

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+

883-------------------------------------------------------------------------------

905-------------------------------------------------------------------------------

884-- Zip operations

906-- Zip operations

885-------------------------------------------------------------------------------

907-------------------------------------------------------------------------------

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

1139 -- XXX applicative with three arguments

1161 -- XXX applicative with three arguments

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

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

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

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

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

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

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

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

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

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

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