cccc7a4 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add a direct style version of StreamK

cccc7a4Harendra Kumar7 years ago

benchmark/BaseStreams.hs+16−1

compare with master

unifiedsplit

@@ -14,6 +14,8 @@ import System.Random (randomRIO)

14import Gauge

15import qualified StreamDOps as D

16import qualified StreamKOps as K

17+import qualified StreamDKOps as DK

18+import qualified Data.List as List

18-- We need a monadic bind here to make sure that the function f does not get

20-- We need a monadic bind here to make sure that the function f does not get

19-- completely optimized out by the compiler in some cases.

21-- completely optimized out by the compiler in some cases.

@@ -195,7 +197,10 @@ main =

195 ]

197 ]

196 ]

198 ]

197 , bgroup "list"

199 , bgroup "list"

200+ [ bgroup "elimination"

201+ [ benchPure "last" (\xs -> [List.last xs]) (K.sourceUnfoldrList K.value)

202+ ]

198− [ bgroup "nested"

203+ , bgroup "nested"

199 [ benchPure "toNullAp" K.toNullApNestedList (K.sourceUnfoldrList K.value2)

204 [ benchPure "toNullAp" K.toNullApNestedList (K.sourceUnfoldrList K.value2)

200 , benchPure "toNull" K.toNullNestedList (K.sourceUnfoldrList K.value2)

205 , benchPure "toNull" K.toNullNestedList (K.sourceUnfoldrList K.value2)

201 , benchPure "toNull3" K.toNullNestedList3 (K.sourceUnfoldrList K.value3)

206 , benchPure "toNull3" K.toNullNestedList3 (K.sourceUnfoldrList K.value3)

@@ -335,4 +340,14 @@ main =

335 , benchK "dropWhileTrue" K.iterateDropWhileTrue

340 , benchK "dropWhileTrue" K.iterateDropWhileTrue

336 ]

341 ]

337 ]

342 ]

343+ , bgroup "streamDK"

344+ [ bgroup "generation"

345+ [ benchIO "unfoldr" DK.toNull DK.sourceUnfoldr

346+ , benchIO "unfoldrM" DK.toNull DK.sourceUnfoldrM

347+ ]

348+ , bgroup "elimination"

349+ [ benchIO "toNull" DK.toNull DK.sourceUnfoldrM

350+ , benchIO "uncons" DK.uncons DK.sourceUnfoldrM

351+ ]

352+ ]

338 ]

353 ]

benchmark/StreamDKOps.hsnew file+424

@@ -0,0 +1,424 @@

1+-- |

2+-- Module : StreamDKOps

4+--

5+-- License : BSD3

6+-- Maintainer : harendra.kumar@gmail.com

8+{-# LANGUAGE FlexibleContexts #-}

9+{-# LANGUAGE ScopedTypeVariables #-}

10+

11+module StreamDKOps where

12+

13+-- import Control.Monad (when)

14+-- import Data.Maybe (isJust)

15+import Prelude

16+ (Monad, Int, (+), ($), (.), return, even, (>), (<=), div,

17+ subtract, undefined, Maybe(..), not, (>>=),

18+ maxBound, flip, (<$>), (<*>), round, (/), (**), (<))

19+import qualified Prelude as P

20+-- import qualified Data.List as List

21+

22+import qualified Streamly.Streams.StreamDK as S

23+-- import qualified Streamly.Streams.Prelude as SP

24+-- import qualified Streamly.SVar as S

25+

26+value, value2, value3, value16, maxValue :: Int

27+value = 100000

28+value2 = round (P.fromIntegral value**(1/2::P.Double)) -- double nested loop

29+value3 = round (P.fromIntegral value**(1/3::P.Double)) -- triple nested loop

30+value16 = round (P.fromIntegral value**(1/16::P.Double)) -- triple nested loop

31+maxValue = value

32+

33+-------------------------------------------------------------------------------

34+-- Benchmark ops

35+-------------------------------------------------------------------------------

36+

37+-------------------------------------------------------------------------------

38+-- Stream generation and elimination

39+-------------------------------------------------------------------------------

40+

41+type Stream m a = S.Stream m a

42+

43+{-# INLINE sourceUnfoldr #-}

44+sourceUnfoldr :: Monad m => Int -> Stream m Int

45+sourceUnfoldr n = S.unfoldr step n

46+ where

47+ step cnt =

48+ if cnt > n + value

49+ then Nothing

50+ else Just (cnt, cnt + 1)

51+

52+{-# INLINE sourceUnfoldrN #-}

53+sourceUnfoldrN :: Monad m => Int -> Int -> Stream m Int

54+sourceUnfoldrN m n = S.unfoldr step n

55+ where

56+ step cnt =

57+ if cnt > n + m

58+ then Nothing

59+ else Just (cnt, cnt + 1)

60+

61+{-# INLINE sourceUnfoldrM #-}

62+sourceUnfoldrM :: Monad m => Int -> Stream m Int

63+sourceUnfoldrM n = S.unfoldrM step n

64+ where

65+ step cnt =

66+ if cnt > n + value

67+ then return Nothing

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

69+

70+{-# INLINE sourceUnfoldrMN #-}

71+sourceUnfoldrMN :: Monad m => Int -> Int -> Stream m Int

72+sourceUnfoldrMN m n = S.unfoldrM step n

73+ where

74+ step cnt =

75+ if cnt > n + m

76+ then return Nothing

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

78+

79+{-

80+{-# INLINE sourceFromEnum #-}

81+sourceFromEnum :: Monad m => Int -> Stream m Int

82+sourceFromEnum n = S.enumFromStepN n 1 value

83+-}

84+

85+{-

86+{-# INLINE sourceFromFoldable #-}

87+sourceFromFoldable :: Int -> Stream m Int

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

89+-}

90+

91+{-

92+{-# INLINE sourceFromFoldableM #-}

93+sourceFromFoldableM :: S.MonadAsync m => Int -> Stream m Int

94+sourceFromFoldableM n = S.fromFoldableM (Prelude.fmap return [n..n+value])

95+-}

96+

97+{-

98+{-# INLINE sourceFoldMapWith #-}

99+sourceFoldMapWith :: Int -> Stream m Int

100+sourceFoldMapWith n = SP.foldMapWith S.serial S.yield [n..n+value]

101+

102+{-# INLINE sourceFoldMapWithM #-}

103+sourceFoldMapWithM :: Monad m => Int -> Stream m Int

104+sourceFoldMapWithM n = SP.foldMapWith S.serial (S.yieldM . return) [n..n+value]

105+-}

106+

107+{-# INLINE source #-}

108+source :: Monad m => Int -> Stream m Int

109+source = sourceUnfoldrM

110+

111+-------------------------------------------------------------------------------

112+-- Elimination

113+-------------------------------------------------------------------------------

114+

115+{-# INLINE runStream #-}

116+runStream :: Monad m => Stream m a -> m ()

117+runStream = S.drain

118+-- runStream = S.mapM_ (\_ -> return ())

119+

120+{-

121+{-# INLINE mapM_ #-}

122+mapM_ :: Monad m => Stream m a -> m ()

123+mapM_ = S.mapM_ (\_ -> return ())

124+-}

125+

126+{-# INLINE toNull #-}

127+toNull :: Monad m => Stream m Int -> m ()

128+toNull = runStream

129+

130+{-# INLINE uncons #-}

131+uncons :: Monad m => Stream m Int -> m ()

132+uncons s = do

133+ r <- S.uncons s

134+ case r of

135+ Nothing -> return ()

136+ Just (_, t) -> uncons t

137+

138+{-

139+{-# INLINE init #-}

140+init :: (Monad m, S.IsStream t) => t m a -> m ()

141+init s = do

142+ t <- S.init s

143+ P.mapM_ S.drain t

144+

145+{-# INLINE tail #-}

146+tail :: (Monad m, S.IsStream t) => t m a -> m ()

147+tail s = S.tail s >>= P.mapM_ tail

148+

149+{-# INLINE nullTail #-}

150+{-# INLINE headTail #-}

151+{-# INLINE zip #-}

152+nullTail, headTail, zip

153+ :: Monad m

154+ => Stream m Int -> m ()

155+

156+nullTail s = do

157+ r <- S.null s

158+ when (not r) $ S.tail s >>= P.mapM_ nullTail

159+

160+headTail s = do

161+ h <- S.head s

162+ when (isJust h) $ S.tail s >>= P.mapM_ headTail

163+

164+{-# INLINE toList #-}

165+toList :: Monad m => Stream m Int -> m [Int]

166+toList = S.toList

167+

168+{-# INLINE foldl #-}

169+foldl :: Monad m => Stream m Int -> m Int

170+foldl = S.foldl' (+) 0

171+

172+{-# INLINE last #-}

173+last :: Monad m => Stream m Int -> m (Maybe Int)

174+last = S.last

175+-}

176+

177+-------------------------------------------------------------------------------

178+-- Transformation

179+-------------------------------------------------------------------------------

180+

181+{-# INLINE transform #-}

182+transform :: Monad m => Stream m a -> m ()

183+transform = runStream

184+

185+{-# INLINE composeN #-}

186+composeN

187+ :: Monad m

188+ => Int -> (Stream m Int -> Stream m Int) -> Stream m Int -> m ()

189+composeN n f =

190+ case n of

191+ 1 -> transform . f

192+ 2 -> transform . f . f

193+ 3 -> transform . f . f . f

194+ 4 -> transform . f . f . f . f

195+ _ -> undefined

196+

197+{-

198+{-# INLINE scan #-}

199+{-# INLINE map #-}

200+{-# INLINE fmap #-}

201+{-# INLINE filterEven #-}

202+{-# INLINE filterAllOut #-}

203+{-# INLINE filterAllIn #-}

204+{-# INLINE takeOne #-}

205+{-# INLINE takeAll #-}

206+{-# INLINE takeWhileTrue #-}

207+{-# INLINE dropOne #-}

208+{-# INLINE dropAll #-}

209+{-# INLINE dropWhileTrue #-}

210+{-# INLINE dropWhileFalse #-}

211+{-# INLINE foldlS #-}

212+{-# INLINE concatMap #-}

213+scan, map, fmap, filterEven, filterAllOut,

214+ filterAllIn, takeOne, takeAll, takeWhileTrue, dropAll, dropOne,

215+ dropWhileTrue, dropWhileFalse, foldlS, concatMap

216+ :: Monad m

217+ => Int -> Stream m Int -> m ()

218+

219+{-# INLINE mapM #-}

220+{-# INLINE mapMSerial #-}

221+{-# INLINE intersperse #-}

222+mapM, mapMSerial, intersperse

223+ :: S.MonadAsync m => Int -> Stream m Int -> m ()

224+

225+scan n = composeN n $ S.scanl' (+) 0

226+map n = composeN n $ P.fmap (+1)

227+fmap n = composeN n $ P.fmap (+1)

228+mapM n = composeN n $ S.mapM return

229+mapMSerial n = composeN n $ S.mapMSerial return

230+filterEven n = composeN n $ S.filter even

231+filterAllOut n = composeN n $ S.filter (> maxValue)

232+filterAllIn n = composeN n $ S.filter (<= maxValue)

233+takeOne n = composeN n $ S.take 1

234+takeAll n = composeN n $ S.take maxValue

235+takeWhileTrue n = composeN n $ S.takeWhile (<= maxValue)

236+dropOne n = composeN n $ S.drop 1

237+dropAll n = composeN n $ S.drop maxValue

238+dropWhileTrue n = composeN n $ S.dropWhile (<= maxValue)

239+dropWhileFalse n = composeN n $ S.dropWhile (<= 1)

240+foldlS n = composeN n $ S.foldlS (flip S.cons) S.nil

241+-- We use a (sqrt n) element stream as source and then concat the same stream

242+-- for each element to produce an n element stream.

243+concatMap n = composeN n $ (\s -> S.concatMap (\_ -> s) s)

244+intersperse n = composeN n $ S.intersperse maxValue

245+

246+-------------------------------------------------------------------------------

247+-- Iteration

248+-------------------------------------------------------------------------------

249+

250+iterStreamLen, maxIters :: Int

251+iterStreamLen = 10

252+maxIters = 10000

253+

254+{-# INLINE iterateSource #-}

255+iterateSource

256+ :: S.MonadAsync m

257+ => (Stream m Int -> Stream m Int) -> Int -> Int -> Stream m Int

258+iterateSource g i n = f i (sourceUnfoldrMN iterStreamLen n)

259+ where

260+ f (0 :: Int) m = g m

261+ f x m = g (f (x P.- 1) m)

262+

263+{-# INLINE iterateMapM #-}

264+{-# INLINE iterateScan #-}

265+{-# INLINE iterateFilterEven #-}

266+{-# INLINE iterateTakeAll #-}

267+{-# INLINE iterateDropOne #-}

268+{-# INLINE iterateDropWhileFalse #-}

269+{-# INLINE iterateDropWhileTrue #-}

270+iterateMapM, iterateScan, iterateFilterEven, iterateTakeAll, iterateDropOne,

271+ iterateDropWhileFalse, iterateDropWhileTrue

272+ :: S.MonadAsync m

273+ => Int -> Stream m Int

274+

275+-- this is quadratic

276+iterateScan = iterateSource (S.scanl' (+) 0) (maxIters `div` 10)

277+iterateDropWhileFalse = iterateSource (S.dropWhile (> maxValue))

278+ (maxIters `div` 10)

279+

280+iterateMapM = iterateSource (S.mapM return) maxIters

281+iterateFilterEven = iterateSource (S.filter even) maxIters

282+iterateTakeAll = iterateSource (S.take maxValue) maxIters

283+iterateDropOne = iterateSource (S.drop 1) maxIters

284+iterateDropWhileTrue = iterateSource (S.dropWhile (<= maxValue)) maxIters

285+

286+-------------------------------------------------------------------------------

287+-- Zipping and concat

288+-------------------------------------------------------------------------------

289+

290+zip src = transform $ S.zipWith (,) src src

291+

292+{-# INLINE concatMapRepl4xN #-}

293+concatMapRepl4xN :: Monad m => Stream m Int -> m ()

294+concatMapRepl4xN src = transform $ (S.concatMap (S.replicate 4) src)

295+

296+-------------------------------------------------------------------------------

297+-- Mixed Composition

298+-------------------------------------------------------------------------------

299+

300+{-# INLINE scanMap #-}

301+{-# INLINE dropMap #-}

302+{-# INLINE dropScan #-}

303+{-# INLINE takeDrop #-}

304+{-# INLINE takeScan #-}

305+{-# INLINE takeMap #-}

306+{-# INLINE filterDrop #-}

307+{-# INLINE filterTake #-}

308+{-# INLINE filterScan #-}

309+{-# INLINE filterMap #-}

310+scanMap, dropMap, dropScan, takeDrop, takeScan, takeMap, filterDrop,

311+ filterTake, filterScan, filterMap

312+ :: Monad m => Int -> Stream m Int -> m ()

313+

314+scanMap n = composeN n $ S.map (subtract 1) . S.scanl' (+) 0

315+dropMap n = composeN n $ S.map (subtract 1) . S.drop 1

316+dropScan n = composeN n $ S.scanl' (+) 0 . S.drop 1

317+takeDrop n = composeN n $ S.drop 1 . S.take maxValue

318+takeScan n = composeN n $ S.scanl' (+) 0 . S.take maxValue

319+takeMap n = composeN n $ S.map (subtract 1) . S.take maxValue

320+filterDrop n = composeN n $ S.drop 1 . S.filter (<= maxValue)

321+filterTake n = composeN n $ S.take maxValue . S.filter (<= maxValue)

322+filterScan n = composeN n $ S.scanl' (+) 0 . S.filter (<= maxBound)

323+filterMap n = composeN n $ S.map (subtract 1) . S.filter (<= maxValue)

324+

325+-------------------------------------------------------------------------------

326+-- Nested Composition

327+-------------------------------------------------------------------------------

328+

329+{-# INLINE toNullApNested #-}

330+toNullApNested :: Monad m => Stream m Int -> m ()

331+toNullApNested s = runStream $ do

332+ (+) <$> s <*> s

333+

334+{-# INLINE toNullNested #-}

335+toNullNested :: Monad m => Stream m Int -> m ()

336+toNullNested s = runStream $ do

337+ x <- s

338+ y <- s

339+ return $ x + y

340+

341+{-# INLINE toNullNested3 #-}

342+toNullNested3 :: Monad m => Stream m Int -> m ()

343+toNullNested3 s = runStream $ do

344+ x <- s

345+ y <- s

346+ z <- s

347+ return $ x + y + z

348+

349+{-# INLINE filterAllOutNested #-}

350+filterAllOutNested

351+ :: Monad m

352+ => Stream m Int -> m ()

353+filterAllOutNested str = runStream $ do

354+ x <- str

355+ y <- str

356+ let s = x + y

357+ if s < 0

358+ then return s

359+ else S.nil

360+

361+{-# INLINE filterAllInNested #-}

362+filterAllInNested

363+ :: Monad m

364+ => Stream m Int -> m ()

365+filterAllInNested str = runStream $ do

366+ x <- str

367+ y <- str

368+ let s = x + y

369+ if s > 0

370+ then return s

371+ else S.nil

372+

373+-------------------------------------------------------------------------------

374+-- Nested Composition Pure lists

375+-------------------------------------------------------------------------------

376+

377+{-# INLINE sourceUnfoldrList #-}

378+sourceUnfoldrList :: Int -> Int -> [Int]

379+sourceUnfoldrList maxval n = List.unfoldr step n

380+ where

381+ step cnt =

382+ if cnt > n + maxval

383+ then Nothing

384+ else Just (cnt, cnt + 1)

385+

386+{-# INLINE toNullApNestedList #-}

387+toNullApNestedList :: [Int] -> [Int]

388+toNullApNestedList s = (+) <$> s <*> s

389+

390+{-# INLINE toNullNestedList #-}

391+toNullNestedList :: [Int] -> [Int]

392+toNullNestedList s = do

393+ x <- s

394+ y <- s

395+ return $ x + y

396+

397+{-# INLINE toNullNestedList3 #-}

398+toNullNestedList3 :: [Int] -> [Int]

399+toNullNestedList3 s = do

400+ x <- s

401+ y <- s

402+ z <- s

403+ return $ x + y + z

404+

405+{-# INLINE filterAllOutNestedList #-}

406+filterAllOutNestedList :: [Int] -> [Int]

407+filterAllOutNestedList str = do

408+ x <- str

409+ y <- str

410+ let s = x + y

411+ if s < 0

412+ then return s

413+ else []

414+

415+{-# INLINE filterAllInNestedList #-}

416+filterAllInNestedList :: [Int] -> [Int]

417+filterAllInNestedList str = do

418+ x <- str

419+ y <- str

420+ let s = x + y

421+ if s > 0

422+ then return s

423+ else []

424+-}

src/Streamly/Streams/StreamDK.hsnew file+165

@@ -0,0 +1,165 @@

1+{-# LANGUAGE BangPatterns #-}

2+{-# LANGUAGE CPP #-}

3+{-# LANGUAGE ExistentialQuantification #-}

4+{-# LANGUAGE FlexibleContexts #-}

5+{-# LANGUAGE PatternSynonyms #-}

6+{-# LANGUAGE RankNTypes #-}

7+{-# LANGUAGE RecordWildCards #-}

8+-- {-# LANGUAGE ScopedTypeVariables #-}

10+#include "inline.hs"

11+

12+-- |

13+-- Module : Streamly.Streams.StreamDK

15+-- License : BSD3

16+-- Maintainer : harendra.kumar@gmail.com

17+-- Stability : experimental

18+-- Portability : GHC

19+--

20+

21+module Streamly.Streams.StreamDK

22+ (

23+ -- * Stream Type

24+

25+ Stream

26+ , Step (..)

27+

28+ -- * Construction

29+ , nil

30+ , cons

31+ , consM

32+ , unfoldr

33+ , unfoldrM

34+ , replicateM

35+

36+ -- * Folding

37+ , uncons

38+ , foldrS

39+

40+ -- * Specific Folds

41+ , drain

42+ )

43+where

44+

45+import Streamly.Streams.StreamDK.Type (Stream(..), Step(..))

46+

47+-------------------------------------------------------------------------------

48+-- Construction

49+-------------------------------------------------------------------------------

50+

51+nil :: Monad m => Stream m a

52+nil = Stream $ return Stop

53+

54+{-# INLINE_NORMAL cons #-}

55+cons :: Monad m => a -> Stream m a -> Stream m a

56+cons x xs = Stream $ return $ Yield x xs

57+

58+consM :: Monad m => m a -> Stream m a -> Stream m a

59+consM eff xs = Stream $ eff >>= \x -> return $ Yield x xs

60+

61+unfoldrM :: Monad m => (s -> m (Maybe (a, s))) -> s -> Stream m a

62+unfoldrM next state = Stream (step' state)

63+ where

64+ step' st = do

65+ r <- next st

66+ return $ case r of

67+ Just (x, s) -> Yield x (Stream (step' s))

68+ Nothing -> Stop

69+{-

70+unfoldrM next s0 = buildM $ \yld stp ->

71+ let go s = do

72+ r <- next s

73+ case r of

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

75+ Nothing -> stp

76+ in go s0

77+-}

78+

79+{-# INLINE unfoldr #-}

80+unfoldr :: Monad m => (b -> Maybe (a, b)) -> b -> Stream m a

81+unfoldr next s0 = build $ \yld stp ->

82+ let go s =

83+ case next s of

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

85+ Nothing -> stp

86+ in go s0

87+

88+replicateM :: Monad m => Int -> a -> Stream m a

89+replicateM n x = Stream (step n)

90+ where

91+ step i = return $

92+ if i <= 0

93+ then Stop

94+ else Yield x (Stream (step (i - 1)))

95+

96+-------------------------------------------------------------------------------

97+-- Folding

98+-------------------------------------------------------------------------------

99+

100+uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))

101+uncons (Stream step) = do

102+ r <- step

103+ return $ case r of

104+ Yield x xs -> Just (x, xs)

105+ Stop -> Nothing

106+

107+-- | Lazy right associative fold to a stream.

108+{-# INLINE_NORMAL foldrS #-}

109+foldrS :: Monad m

110+ => (a -> Stream m b -> Stream m b)

111+ -> Stream m b

112+ -> Stream m a

113+ -> Stream m b

114+foldrS f streamb = go

115+ where

116+ go (Stream stepa) = Stream $ do

117+ r <- stepa

118+ case r of

119+ Yield x xs -> let Stream step = f x (go xs) in step

120+ Stop -> let Stream step = streamb in step

121+

122+{-# INLINE_LATE foldrM #-}

123+foldrM :: Monad m => (a -> m b -> m b) -> m b -> Stream m a -> m b

124+foldrM fstep acc ys = go ys

125+ where

126+ go (Stream step) = do

127+ r <- step

128+ case r of

129+ Yield x xs -> fstep x (go xs)

130+ Stop -> acc

131+

132+{-# INLINE_NORMAL build #-}

133+build :: Monad m

134+ => forall a. (forall b. (a -> b -> b) -> b -> b) -> Stream m a

135+build g = g cons nil

136+

137+{-# RULES

138+"foldrM/build" forall k z (g :: forall b. (a -> b -> b) -> b -> b).

139+ foldrM k z (build g) = g k z #-}

140+

141+{-

142+-- To fuse foldrM with unfoldrM we need the type m1 to be polymorphic such that

143+-- it is either Monad m or Stream m. So that we can use cons/nil as well as

144+-- monadic construction function as its arguments.

145+--

146+{-# INLINE_NORMAL buildM #-}

147+buildM :: Monad m

148+ => forall a. (forall b. (a -> m1 b -> m1 b) -> m1 b -> m1 b) -> Stream m a

149+buildM g = g cons nil

150+-}

151+

152+-------------------------------------------------------------------------------

153+-- Specific folds

154+-------------------------------------------------------------------------------

155+

156+{-# INLINE drain #-}

157+drain :: Monad m => Stream m a -> m ()

158+drain = foldrM (\_ xs -> xs) (return ())

159+{-

160+drain (Stream step) = do

161+ r <- step

162+ case r of

163+ Yield _ next -> drain next

164+ Stop -> return ()

165+ -}

src/Streamly/Streams/StreamDK/Type.hsnew file+77

@@ -0,0 +1,77 @@

1+{-# LANGUAGE CPP #-}

2+{-# LANGUAGE ExistentialQuantification #-}

3+{-# LANGUAGE FlexibleContexts #-}

5+-- |

6+-- Module : Streamly.StreamDK.Type

8+-- License : BSD3

9+-- Maintainer : harendra.kumar@gmail.com

10+-- Stability : experimental

11+-- Portability : GHC

12+--

13+-- A CPS style stream using a constructor based representation instead of a

14+-- function based representation.

15+--

16+-- StreamD is a non-recursive data type in which the state of the stream and

17+-- the step function are separate. When the step function is called, a stream

18+-- element and the new stream state is yielded. The generated element and the

19+-- state are passed to the next consumer in the loop. The state is threaded

20+-- around in the loop until control returns back to the original step function

21+-- to run the next step. This creates a closed loop representation with state

22+-- of each step being hidden/abstracted or existential within that step. This

23+-- creates a loop reprentation identical to the for an while loop constructs in

24+-- imperative languages, the states of the steps combined together constitutes

25+-- the state of the loop iteration.

26+--

27+-- Because of a closed representation consing too many elements in this type of

28+-- stream does not scale. Each cons creates a layer that needs to return the

29+-- control back to the caller. However, this representation can be optimized

30+-- very efficiently by the compiler because the state is explicitly separated

31+-- from the functions, represented using pure data constructors and visible to

32+-- the compiler, the stream steps can be fused using case-of-case

33+-- transformations and the state can be specialized using spec-constructor

34+-- optimization, yielding a C like tight loop/state machine with no

35+-- constructors and therefore no unnecessary allocation. Because of the closed

36+-- nature of the stream, exception handling is disciplined in this model

37+-- because control flow is structured in the loop and cannot be arbitrary.

38+-- Also, because of the closed nature of the loop, it is not possible to uncons

39+-- an item from the stream.

40+--

41+-- This type is amenable to stream fusion.

42+--

43+-- StreamDK i.e. the stream defined in this module, like StreamK, is a

44+-- recursive data type which has no explicit state, each step yields an element

45+-- and a computation representing the rest of the stream. Stream state is part

46+-- of the function representing the rest of the stream. This creates an open

47+-- computation representation, or essentially a continuation passing style

48+-- computation. After the stream step is executed, the caller is free to

49+-- consume the produced element and then send the control wherever it wants,

50+-- there is no restriction on the control to return back somewhere, the control

51+-- is free to go anywhere. The caller may decide not to consume the rest of the

52+-- stream.

53+--

54+-- This model allows generating arbitrarily large streams lazily using the cons

55+-- operation. Addition of more elements does not add any cost. Similarly uncons

56+-- operation is possible in this stream representation. The control flow can be

57+-- manipulated arbitrarily, there is no structure dictated by the stream. This

58+-- allows interoperation/interconversion with other monads, and other stream

59+-- types. Scalable cons operation and the possibility of uncons operation makes

60+-- this representation very similar to standard lists. Also, this type

61+-- facilitates a scalable monadic composition.

62+--

63+-- This model has a cost because the open representation requires a constructor

64+-- which may not always be eliminated.

65+--

66+-- This type is amenable to foldr/build fusion.

67+

68+module Streamly.Streams.StreamDK.Type

69+ ( Step(..)

70+ , Stream (..)

71+ )

72+where

73+

74+-- XXX Use Cons and Nil instead of Yield and Stop?

75+data Step m a = Yield a (Stream m a) | Stop

76+

77+data Stream m a = Stream (m (Step m a))

streamly.cabal+5

@@ -274,6 +274,8 @@ library

274 -- Base streams

275 , Streamly.Streams.StreamK.Type

276 , Streamly.Streams.StreamK

277+ , Streamly.Streams.StreamDK.Type

278+ , Streamly.Streams.StreamDK

277 , Streamly.Streams.StreamD.Type

279 , Streamly.Streams.StreamD.Type

278 , Streamly.Streams.StreamD

280 , Streamly.Streams.StreamD

279 , Streamly.Streams.Prelude

281 , Streamly.Streams.Prelude

@@ -653,6 +655,8 @@ benchmark base

653 , Streamly.Time.Units

655 , Streamly.Time.Units

654 , Streamly.Time.Clock

656 , Streamly.Time.Clock

655 , Streamly.SVar

657 , Streamly.SVar

658+ , Streamly.Streams.StreamDK.Type

659+ , Streamly.Streams.StreamDK

656 , Streamly.Streams.StreamK.Type

660 , Streamly.Streams.StreamK.Type

657 , Streamly.Streams.StreamK

661 , Streamly.Streams.StreamK

658 , Streamly.Streams.StreamD.Type

662 , Streamly.Streams.StreamD.Type

@@ -662,6 +666,7 @@ benchmark base

662

666

663 , StreamDOps

667 , StreamDOps

664 , StreamKOps

668 , StreamKOps

669+ , StreamDKOps

665

670

666 if flag(dev)

671 if flag(dev)

667 buildable: True

672 buildable: True