7a526f2 · streamly · git.iko.soy

git.iko.soy/streamly

use direct style streams for linear composition

For some stream generation APIs. Also added, fromList/fromListM APIs.

commit7a526f22b993440bf3754e65561033255f0c62b4 authorHarendra Kumar <harendra.kumar@gmail.com> dateSun 24 Jun 2018, 06:19 (8 years ago) parent434ac89

8 files changed+126−72

compare with master

unifiedsplit

Changelog.md+3 benchmark/Linear.hs+2 benchmark/LinearOps.hs+9−1 benchmark/NestedOps.hs+1−1 src/Streamly/Prelude.hs+45−28 src/Streamly/Streams/Prelude.hs+9−27 src/Streamly/Streams/StreamD.hs+23−15 src/Streamly/Streams/StreamK.hs+34

Changelog.md+3

@@ -8,6 +8,9 @@

99* Add `yield` to construct a singleton stream from a pure value

1010* Add `repeat` to generate an infinite stream by repeating a pure value

11+* Add `fromList` and `fromListM` to generate streams from lists, faster than

12+ `fromFoldable` and `fromFoldableM`

13+* Significant performance improvements

1114

1215## 0.3.0

1316

benchmark/Linear.hs+2

@@ -40,6 +40,8 @@ main = do

4040 [ -- Most basic, barely stream continuations running

4141 benchSrcIO serially "unfoldr" $ Ops.sourceUnfoldr

4242 , benchSrcIO serially "unfoldrM" Ops.sourceUnfoldrM

43+ , benchSrcIO serially "fromList" Ops.sourceFromList

44+ , benchSrcIO serially "fromListM" Ops.sourceFromListM

4345 -- These are essentially cons and consM

4446 , benchSrcIO serially "fromFoldable" Ops.sourceFromFoldable

4547 , benchSrcIO serially "fromFoldableM" Ops.sourceFromFoldableM

benchmark/LinearOps.hs+9−1

@@ -73,6 +73,14 @@ type Stream m a = S.SerialT m a

7373source :: (S.MonadAsync m, S.IsStream t) => Int -> t m Int

7474source n = S.serially $ sourceUnfoldrM n

7575

76+{-# INLINE sourceFromList #-}

77+sourceFromList :: (Monad m, S.IsStream t) => Int -> t m Int

78+sourceFromList n = S.fromList [n..n+value]

79+

80+{-# INLINE sourceFromListM #-}

81+sourceFromListM :: (S.MonadAsync m, S.IsStream t) => Int -> t m Int

82+sourceFromListM n = S.fromListM (Prelude.fmap return [n..n+value])

83+

7684{-# INLINE sourceFromFoldable #-}

7785sourceFromFoldable :: S.IsStream t => Int -> t m Int

7886sourceFromFoldable n = S.fromFoldable [n..n+value]

@@ -92,7 +100,7 @@ sourceFoldMapWithM :: (S.IsStream t, Monad m, S.Semigroup (t m Int))

92100sourceFoldMapWithM n = S.foldMapWith (S.<>) (S.yieldM . return) [n..n+value]

93101

94102{-# INLINE sourceUnfoldr #-}

95−sourceUnfoldr :: S.IsStream t => Int -> t m Int

103+sourceUnfoldr :: (Monad m, S.IsStream t) => Int -> t m Int

96104sourceUnfoldr n = S.unfoldr step n

97105 where

98106 step cnt =

benchmark/NestedOps.hs+1−1

@@ -42,7 +42,7 @@ sourceUnfoldrM n value = S.serially $ S.unfoldrM step n

4242 else return (Just (cnt, cnt + 1))

4343

4444{-# INLINE sourceUnfoldr #-}

45−sourceUnfoldr :: S.IsStream t => Int -> Int -> t m Int

45+sourceUnfoldr :: (Monad m, S.IsStream t) => Int -> Int -> t m Int

4646sourceUnfoldr start n = S.unfoldr step start

4747 where

4848 step cnt =

src/Streamly/Prelude.hs+45−28

@@ -1,10 +1,15 @@

11{-# LANGUAGE BangPatterns #-}

2+{-# LANGUAGE CPP #-}

23{-# LANGUAGE FlexibleContexts #-}

34{-# LANGUAGE FlexibleInstances #-}

45{-# LANGUAGE MultiParamTypeClasses #-}

56{-# LANGUAGE RankNTypes #-}

67{-# LANGUAGE UndecidableInstances #-} -- XXX

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

10+

11+#include "Streams/inline.h"

12+

813-- |

914-- Module : Streamly.Prelude

@@ -60,7 +65,9 @@ module Streamly.Prelude

6065 , repeatM

6166 , iterate

6267 , iterateM

63− , fromFoldable

68+ , fromList

69+ , fromListM

70+ , K.fromFoldable

6471 , fromFoldableM

6572

6673 -- * Deconstruction

@@ -123,7 +130,7 @@ module Streamly.Prelude

123130 , toHandle

124131

125132 -- * Deprecated

126− , once

133+ , K.once

127134 , each

128135 , scan

129136 , foldl

@@ -144,8 +151,9 @@ import Prelude hiding (filter, drop, dropWhile, take,

144151import qualified Prelude

145152import qualified System.IO as IO

146153

147−import Streamly.Streams.StreamK

154+import Streamly.Streams.StreamK (IsStream(..), cons, nil, (.:))

148155import qualified Streamly.Streams.StreamK as K

156+import qualified Streamly.Streams.StreamD as D

149157import Streamly.Streams.Zip

150158import Streamly.Streams.Serial

151159import Streamly.SVar (MonadAsync)

@@ -171,14 +179,11 @@ import Streamly.SVar (MonadAsync)

171179-- @

172180--

173181-- @since 0.1.0

174−{-# INLINE unfoldr #-}

175−unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a

176−unfoldr step = fromStream . go

177− where

178− go s = K.Stream $ \_ stp _ yld ->

179− case step s of

180− Nothing -> stp

181− Just (a, b) -> yld a (go b)

182+{-# INLINE_EARLY unfoldr #-}

183+unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a

184+unfoldr step seed = fromStream $ D.toStreamK (D.unfoldr step seed)

185+{-# RULES "unfoldr fallback to StreamK" [1]

186+ forall a b. D.toStreamK (D.unfoldr a b) = K.unfoldr a b #-}

182187

183188-- | Build a stream by unfolding a /monadic/ step function starting from a

184189-- seed. The step function returns the next element in the stream and the next

@@ -211,23 +216,35 @@ unfoldr step = fromStream . go

211216-- /Concurrent/

212217--

213218-- /Since: 0.1.0/

214−{-# INLINE unfoldrM #-}

219+{-# INLINE_EARLY unfoldrM #-}

215220unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a

216−unfoldrM step = go

217− where

218− go s = fromStream $ K.Stream $ \svr stp sng yld -> do

219− mayb <- step s

220− case mayb of

221− Nothing -> stp

222− Just (a, b) ->

223− K.unStream (toStream (return a |: go b)) svr stp sng yld

224−

225−-- | Construct a stream from a 'Foldable' containing pure values.

226−--

227−-- @since 0.2.0

228−{-# INLINE fromFoldable #-}

229−fromFoldable :: (IsStream t, Foldable f) => f a -> t m a

230−fromFoldable = Prelude.foldr cons nil

221+unfoldrM = K.unfoldrM

222+

223+{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}

224+{-# INLINE_EARLY unfoldrMSerial #-}

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

226+unfoldrMSerial step seed = fromStream $ D.toStreamK (D.unfoldrM step seed)

227+

228+-- | Construct a stream from a list containing pure values. This can be more

229+-- efficient than 'K.fromFoldable' for lists as it can fuse the list.

230+--

231+-- @since 0.4.0

232+{-# INLINE_EARLY fromList #-}

233+fromList :: (Monad m, IsStream t) => [a] -> t m a

234+fromList = fromStream . D.toStreamK . D.fromList

235+{-# RULES "fromList fallback to StreamK" [1]

236+ forall a. D.toStreamK (D.fromList a) = K.fromFoldable a #-}

237+

238+-- | Construct a stream from a list containing monadic actions. This can be

239+-- more efficient than 'fromFoldableM' especially for serial streams as it can

240+-- fuse the list.

241+--

242+-- @since 0.4.0

243+{-# INLINE_EARLY fromListM #-}

244+fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a

245+fromListM = fromStream . D.toStreamK . D.fromListM

246+{-# RULES "fromListM fallback to StreamK" [1]

247+ forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}

231248

232249-- | Construct a stream from a 'Foldable' containing monadic actions.

233250--

@@ -249,7 +266,7 @@ fromFoldableM = Prelude.foldr consM nil

249266{-# DEPRECATED each "Please use fromFoldable instead." #-}

250267{-# INLINE each #-}

251268each :: (IsStream t, Foldable f) => f a -> t m a

252−each = fromFoldable

269+each = K.fromFoldable

253270

254271-- | Generate a stream by performing a monadic action @n@ times.

255272--

src/Streamly/Streams/Prelude.hs+9−27

@@ -9,6 +9,10 @@

99{-# LANGUAGE StandaloneDeriving #-}

1010{-# LANGUAGE UndecidableInstances #-} -- XXX

1111

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

13+

14+#include "inline.h"

15+

1216-- |

1317-- Module : Streamly.Streams.Prelude

@@ -34,6 +38,7 @@ where

3438

3539import Streamly.Streams.Serial (SerialT)

3640import Streamly.Streams.StreamK hiding (runStream)

41+import qualified Streamly.Streams.StreamD as D

3742import qualified Streamly.Streams.StreamK as K

3843

3944------------------------------------------------------------------------------

@@ -45,8 +50,11 @@ import qualified Streamly.Streams.StreamK as K

4550-- combinators for example @runStream . 'asyncly'@.

4651--

4752-- @since 0.2.0

53+{-# INLINE_EARLY runStream #-}

4854runStream :: Monad m => SerialT m a -> m ()

49−runStream = K.runStream

55+runStream m = D.runStream $ D.fromStreamK (toStream m)

56+{-# RULES "runStream fallback to CPS" [1]

57+ forall a. D.runStream (D.fromStreamK a) = K.runStream a #-}

5058

5159-- | Same as 'runStream'

5260--

@@ -90,29 +98,3 @@ foldMapWith f g = foldr (f . g) nil

9098forEachWith :: (IsStream t, Foldable f)

9199 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b

92100forEachWith f xs g = foldr (f . g) nil xs

93−

94−-------------------------------------------------------------------------------

95−-- Generation by unfold

96−-------------------------------------------------------------------------------

97−

98−{-

99−{-# INLINE unfoldr #-}

100−unfoldr :: (b -> Maybe (a, b)) -> b -> Stream m a

101−unfoldr step = go

102− where

103− go s = Stream $ \_ stp _ yld ->

104− case step s of

105− Nothing -> stp

106− Just (a, b) -> yld a (go b)

107−

108−{-# INLINE unfoldrM #-}

109−unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> Stream m a

110−unfoldrM step = go

111− where

112− go s = Stream $ \svr stp sng yld -> do

113− mayb <- step s

114− case mayb of

115− Nothing -> stp

116− Just (a, b) ->

117− K.runStream (a `cons` go b) svr stp sng yld

118−-}

src/Streamly/Streams/StreamD.hs+23−15

@@ -47,6 +47,7 @@ module Streamly.Streams.StreamD

4747 , repeat

4848 , enumFromStepN

4949 , fromList

50+ , fromListM

5051

5152 -- * Transformation

5253 , mapM

@@ -58,6 +59,7 @@ module Streamly.Streams.StreamD

5859 )

5960where

6061

62+import Streamly.SVar (MonadAsync)

6163import qualified Streamly.Streams.StreamK as K

6264import Prelude hiding (mapM, mapM_, repeat)

6365

@@ -80,11 +82,11 @@ instance Functor (Step s) where

8082data Stream m a = forall s. Stream (s -> m (Step s a)) s

8183

8284-------------------------------------------------------------------------------

83−-- IsStream Stream

85+-- Conversion to and from StreamK

8486-------------------------------------------------------------------------------

8587

8688-- Convert a direct stream to and from CPS encoded stream

87−{-# INLINE_NORMAL toStreamK #-}

89+{-# INLINE_LATE toStreamK #-}

8890toStreamK :: Monad m => Stream m a -> K.Stream m a

8991toStreamK (Stream step state) = go state

9092 where

@@ -94,7 +96,7 @@ toStreamK (Stream step state) = go state

9496 Yield x s -> yld x (go s)

9597 Stop -> stp

9698

97−{-# INLINE_NORMAL fromStreamK #-}

99+{-# INLINE_LATE fromStreamK #-}

98100fromStreamK :: Monad m => K.Stream m a -> Stream m a

99101fromStreamK m = Stream step m

100102 where

@@ -104,14 +106,10 @@ fromStreamK m = Stream step m

104106 yieldk a r = return $ Yield a r

105107 in K.unStream m1 Nothing stop single yieldk

106108

107−{-

108−instance K.IsStream Stream where

109− toStream = toStreamK

110− fromStream = fromStreamK

111−

112− consM = undefined

113− (|:) = undefined

114−-}

109+{-# RULES "fromStreamK/toStreamK fusion"

110+ forall s. toStreamK (fromStreamK s) = s #-}

111+{-# RULES "toStreamK/fromStreamK fusion"

112+ forall s. fromStreamK (toStreamK s) = s #-}

115113

116114------------------------------------------------------------------------------

117115-- Construction

@@ -155,7 +153,7 @@ unfoldrM next state = Stream step state

155153 Just (x, s) -> Yield x s

156154 Nothing -> Stop

157155

158−{-# INLINE_NORMAL unfoldr #-}

156+{-# INLINE_LATE unfoldr #-}

159157unfoldr :: Monad m => (s -> Maybe (a, s)) -> s -> Stream m a

160158unfoldr f = unfoldrM (return . f)

161159

@@ -175,8 +173,18 @@ enumFromStepN from stride n =

175173 step (x, i) | i > 0 = return $ Yield x (x + stride, i - 1)

176174 | otherwise = return $ Stop

177175

178−-- | Convert a list to a 'Stream'

179−{-# INLINE_NORMAL fromList #-}

176+-- XXX we need the MonadAsync constraint because of a rewrite rule.

177+-- | Convert a list of monadic actions to a 'Stream'

178+{-# INLINE_LATE fromListM #-}

179+fromListM :: MonadAsync m => [m a] -> Stream m a

180+fromListM zs = Stream step zs

181+ where

182+ {-# INLINE_LATE step #-}

183+ step (m:ms) = m >>= \x -> return $ Yield x ms

184+ step [] = return Stop

185+

186+-- | Convert a list of pure values to a 'Stream'

187+{-# INLINE_LATE fromList #-}

180188fromList :: Monad m => [a] -> Stream m a

181189fromList zs = Stream step zs

182190 where

@@ -215,7 +223,7 @@ uncons (Stream step state) = go state

215223 Stop -> Nothing

216224

217225-- | Run a streaming composition, discard the results.

218−{-# INLINE_NORMAL runStream #-}

226+{-# INLINE_LATE runStream #-}

219227runStream :: Monad m => Stream m a -> m ()

220228runStream (Stream step state) = go state

221229 where

src/Streamly/Streams/StreamK.hs+34

@@ -52,7 +52,10 @@ module Streamly.Streams.StreamK

5252 , consK

5353

5454 -- * Generation

55+ , unfoldr

56+ , unfoldrM

5557 , repeat

58+ , fromFoldable

5659

5760 -- * Semigroup Style Composition

5861 , serial

@@ -334,6 +337,30 @@ runStream m = go (toStream m)

334337 yieldk _ r = go (toStream r)

335338 in (unStream m1) Nothing stop single yieldk

336339

340+-------------------------------------------------------------------------------

341+-- Generation

342+-------------------------------------------------------------------------------

343+

344+{-# INLINE unfoldr #-}

345+unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a

346+unfoldr step = fromStream . go

347+ where

348+ go s = Stream $ \_ stp _ yld ->

349+ case step s of

350+ Nothing -> stp

351+ Just (a, b) -> yld a (go b)

352+

353+{-# INLINE unfoldrM #-}

354+unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a

355+unfoldrM step = go

356+ where

357+ go s = fromStream $ Stream $ \svr stp sng yld -> do

358+ mayb <- step s

359+ case mayb of

360+ Nothing -> stp

361+ Just (a, b) ->

362+ unStream (toStream (return a |: go b)) svr stp sng yld

363+

337364-------------------------------------------------------------------------------

338365-- Special generation

339366-------------------------------------------------------------------------------

@@ -344,6 +371,13 @@ runStream m = go (toStream m)

344371repeat :: IsStream t => a -> t m a

345372repeat a = let x = cons a x in x

346373

374+-- | Construct a stream from a 'Foldable' containing pure values.

375+--

376+-- @since 0.2.0

377+{-# INLINE fromFoldable #-}

378+fromFoldable :: (IsStream t, Foldable f) => f a -> t m a

379+fromFoldable = Prelude.foldr cons nil

380+

347381------------------------------------------------------------------------------

348382-- Semigroup

349383------------------------------------------------------------------------------