7a526f2 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

use direct style streams for linear composition

7a526f2Harendra Kumar8 years ago

Changelog.md+3

compare with master

unifiedsplit

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

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

10* 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

12## 0.3.0

15## 0.3.0

benchmark/Linear.hs+2

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

40 [ -- Most basic, barely stream continuations running

41 benchSrcIO serially "unfoldr" $ Ops.sourceUnfoldr

42 , benchSrcIO serially "unfoldrM" Ops.sourceUnfoldrM

43+ , benchSrcIO serially "fromList" Ops.sourceFromList

44+ , benchSrcIO serially "fromListM" Ops.sourceFromListM

43 -- These are essentially cons and consM

45 -- These are essentially cons and consM

44 , benchSrcIO serially "fromFoldable" Ops.sourceFromFoldable

46 , benchSrcIO serially "fromFoldable" Ops.sourceFromFoldable

45 , benchSrcIO serially "fromFoldableM" Ops.sourceFromFoldableM

47 , benchSrcIO serially "fromFoldableM" Ops.sourceFromFoldableM

benchmark/LinearOps.hs+9−1

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

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

74source n = S.serially $ sourceUnfoldrM n

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+

76{-# INLINE sourceFromFoldable #-}

84{-# INLINE sourceFromFoldable #-}

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

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

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

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

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

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

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

101

94{-# INLINE sourceUnfoldr #-}

102{-# INLINE sourceUnfoldr #-}

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

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

96sourceUnfoldr n = S.unfoldr step n

104sourceUnfoldr n = S.unfoldr step n

97 where

105 where

98 step cnt =

106 step cnt =

benchmark/NestedOps.hs+1−1

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

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

44{-# INLINE sourceUnfoldr #-}

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

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

46sourceUnfoldr start n = S.unfoldr step start

47 where

48 step cnt =

src/Streamly/Prelude.hs+45−28

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

1{-# LANGUAGE BangPatterns #-}

2+{-# LANGUAGE CPP #-}

2{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleInstances #-}

4{-# LANGUAGE FlexibleInstances #-}

4{-# LANGUAGE MultiParamTypeClasses #-}

5{-# LANGUAGE MultiParamTypeClasses #-}

5{-# LANGUAGE RankNTypes #-}

6{-# LANGUAGE RankNTypes #-}

6{-# LANGUAGE UndecidableInstances #-} -- XXX

7{-# LANGUAGE UndecidableInstances #-} -- XXX

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

10+

11+#include "Streams/inline.h"

12+

8-- |

13-- |

9-- Module : Streamly.Prelude

14-- Module : Streamly.Prelude

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

60 , repeatM

65 , repeatM

61 , iterate

66 , iterate

62 , iterateM

67 , iterateM

68+ , fromList

69+ , fromListM

63− , fromFoldable

70+ , K.fromFoldable

64 , fromFoldableM

71 , fromFoldableM

66 -- * Deconstruction

73 -- * Deconstruction

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

123 , toHandle

130 , toHandle

124

131

125 -- * Deprecated

132 -- * Deprecated

126− , once

133+ , K.once

127 , each

134 , each

128 , scan

135 , scan

129 , foldl

136 , foldl

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

144import qualified Prelude

151import qualified Prelude

145import qualified System.IO as IO

152import qualified System.IO as IO

146

153

147−import Streamly.Streams.StreamK

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

148import qualified Streamly.Streams.StreamK as K

155import qualified Streamly.Streams.StreamK as K

156+import qualified Streamly.Streams.StreamD as D

149import Streamly.Streams.Zip

157import Streamly.Streams.Zip

150import Streamly.Streams.Serial

158import Streamly.Streams.Serial

151import Streamly.SVar (MonadAsync)

159import Streamly.SVar (MonadAsync)

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

171-- @

179-- @

172--

180--

173-- @since 0.1.0

181-- @since 0.1.0

174−{-# INLINE unfoldr #-}

182+{-# INLINE_EARLY unfoldr #-}

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

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

176−unfoldr step = fromStream . go

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

177− where

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

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

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

179− case step s of

180− Nothing -> stp

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

182

187

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

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

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

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

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

211-- /Concurrent/

216-- /Concurrent/

212--

217--

213-- /Since: 0.1.0/

218-- /Since: 0.1.0/

214−{-# INLINE unfoldrM #-}

219+{-# INLINE_EARLY unfoldrM #-}

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

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

216−unfoldrM step = go

221+unfoldrM = K.unfoldrM

217− where

222+

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

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

219− mayb <- step s

224+{-# INLINE_EARLY unfoldrMSerial #-}

220− case mayb of

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

221− Nothing -> stp

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

222− Just (a, b) ->

227+

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

224−

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

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.

226−--

230+--

227−-- @since 0.2.0

231+-- @since 0.4.0

228−{-# INLINE fromFoldable #-}

232+{-# INLINE_EARLY fromList #-}

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

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

230−fromFoldable = Prelude.foldr cons nil

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 #-}

231

248

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

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

233--

250--

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

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

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

250{-# INLINE each #-}

267{-# INLINE each #-}

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

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

252−each = fromFoldable

269+each = K.fromFoldable

253

270

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

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

255--

272--

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

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

9{-# LANGUAGE StandaloneDeriving #-}

10{-# LANGUAGE UndecidableInstances #-} -- XXX

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

13+

14+#include "inline.h"

15+

12-- |

16-- |

13-- Module : Streamly.Streams.Prelude

17-- Module : Streamly.Streams.Prelude

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

35import Streamly.Streams.Serial (SerialT)

39import Streamly.Streams.Serial (SerialT)

36import Streamly.Streams.StreamK hiding (runStream)

40import Streamly.Streams.StreamK hiding (runStream)

41+import qualified Streamly.Streams.StreamD as D

37import qualified Streamly.Streams.StreamK as K

42import qualified Streamly.Streams.StreamK as K

39------------------------------------------------------------------------------

44------------------------------------------------------------------------------

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

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

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

46--

51--

47-- @since 0.2.0

52-- @since 0.2.0

53+{-# INLINE_EARLY runStream #-}

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

54runStream :: 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 #-}

51-- | Same as 'runStream'

59-- | Same as 'runStream'

52--

60--

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

90forEachWith :: (IsStream t, Foldable f)

98forEachWith :: (IsStream t, Foldable f)

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

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

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

100forEachWith 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

47 , repeat

48 , enumFromStepN

49 , fromList

50+ , fromListM

51 -- * Transformation

52 -- * Transformation

52 , mapM

53 , mapM

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

58 )

59 )

59where

60where

62+import Streamly.SVar (MonadAsync)

61import qualified Streamly.Streams.StreamK as K

63import qualified Streamly.Streams.StreamK as K

62import Prelude hiding (mapM, mapM_, repeat)

64import Prelude hiding (mapM, mapM_, repeat)

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

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

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

82-------------------------------------------------------------------------------

84-------------------------------------------------------------------------------

83−-- IsStream Stream

85+-- Conversion to and from StreamK

84-------------------------------------------------------------------------------

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

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

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

87−{-# INLINE_NORMAL toStreamK #-}

89+{-# INLINE_LATE toStreamK #-}

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

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

89toStreamK (Stream step state) = go state

91toStreamK (Stream step state) = go state

90 where

92 where

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

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

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

95 Stop -> stp

97 Stop -> stp

97−{-# INLINE_NORMAL fromStreamK #-}

99+{-# INLINE_LATE fromStreamK #-}

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

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

99fromStreamK m = Stream step m

101fromStreamK m = Stream step m

100 where

102 where

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

104 yieldk a r = return $ Yield a r

106 yieldk a r = return $ Yield a r

105 in K.unStream m1 Nothing stop single yieldk

107 in K.unStream m1 Nothing stop single yieldk

106

108

107−{-

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

108−instance K.IsStream Stream where

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

109− toStream = toStreamK

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

110− fromStream = fromStreamK

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

111−

112− consM = undefined

113− (|:) = undefined

114−-}

115

113

116------------------------------------------------------------------------------

114------------------------------------------------------------------------------

117-- Construction

115-- Construction

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

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

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

156 Nothing -> Stop

154 Nothing -> Stop

157

155

158−{-# INLINE_NORMAL unfoldr #-}

156+{-# INLINE_LATE unfoldr #-}

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

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

160unfoldr f = unfoldrM (return . f)

158unfoldr f = unfoldrM (return . f)

161

159

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

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

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

176 | otherwise = return $ Stop

174 | otherwise = return $ Stop

177

175

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+

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

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

179−{-# INLINE_NORMAL fromList #-}

187+{-# INLINE_LATE fromList #-}

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

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

181fromList zs = Stream step zs

189fromList zs = Stream step zs

182 where

190 where

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

215 Stop -> Nothing

223 Stop -> Nothing

216

224

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

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

218−{-# INLINE_NORMAL runStream #-}

226+{-# INLINE_LATE runStream #-}

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

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

220runStream (Stream step state) = go state

228runStream (Stream step state) = go state

221 where

229 where

src/Streamly/Streams/StreamK.hs+34

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

52 , consK

54 -- * Generation

55+ , unfoldr

56+ , unfoldrM

55 , repeat

57 , repeat

58+ , fromFoldable

57 -- * Semigroup Style Composition

60 -- * Semigroup Style Composition

58 , serial

61 , serial

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

334 yieldk _ r = go (toStream r)

337 yieldk _ r = go (toStream r)

335 in (unStream m1) Nothing stop single yieldk

338 in (unStream m1) Nothing stop single yieldk

336

339

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+

337-------------------------------------------------------------------------------

364-------------------------------------------------------------------------------

338-- Special generation

365-- Special generation

339-------------------------------------------------------------------------------

366-------------------------------------------------------------------------------

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

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

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

345repeat a = let x = cons a x in x

372repeat a = let x = cons a x in x

346

373

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+

347------------------------------------------------------------------------------

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

348-- Semigroup

382-- Semigroup

349------------------------------------------------------------------------------

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