b6185ba · streamly · git.iko.soy

core/src/Streamly/Data/Stream.hs+9

compare with master

unifiedsplit

@@ -79,6 +79,15 @@

79-- combinators. See the <https://github.com/composewell/streamly-examples>

80-- repository for many more real world examples of stream programming.

81--

82+-- == Performance Notes

83+--

84+-- Operations annotated as /CPS/ force the use of continuation passing style

85+-- streams. Therefore, such operations are not subjected to stream fusion.

86+-- However, for some of these operations you can find fusible alternatives in

87+-- the internal modules, which are perfectly fine to use but you need to

88+-- understand the implications especially the O(n^2) nature of those

89+-- operations..

90+--

82module Streamly.Data.Stream

91module Streamly.Data.Stream

83 (

92 (

84 Stream

93 Stream

core/src/Streamly/Internal/Data/Fold.hs+1−1

@@ -381,7 +381,7 @@ drive strm fl = StreamD.fold fl $ StreamD.fromStreamK $ Stream.toStreamK strm

381--

382-- >>> driveBreak = flip Stream.foldBreak

383--

384−-- /Not fused/

384+-- /CPS/

385--

386{-# INLINE driveBreak #-}

387driveBreak :: Monad m => Stream m a -> Fold m a b -> m (b, Stream m a)

core/src/Streamly/Internal/Data/Stream/Bottom.hs+17−3

@@ -53,6 +53,7 @@ module Streamly.Internal.Data.Stream.Bottom

53

54 -- * Expand

55 , concatEffect

56+ , concatEffect2

56 , concatMapM

57 , concatMapM

57 , concatMap

58 , concatMap

58

59

@@ -228,7 +229,7 @@ fold fl strm = D.fold fl $ D.fromStreamK $ toStreamK strm

228-- | Like 'fold' but also returns the remaining stream. The resulting stream

229-- | Like 'fold' but also returns the remaining stream. The resulting stream

229-- would be 'Stream.nil' if the stream finished before the fold.

230-- would be 'Stream.nil' if the stream finished before the fold.

230--

231--

231−-- /Not fused/

232+-- /CPS/

232--

233--

233{-# INLINE foldBreak #-}

234{-# INLINE foldBreak #-}

234foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)

235foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)

@@ -540,6 +541,8 @@ reverse s = fromStreamD $ D.reverse $ toStreamD s

540

541

541-- | Like 'reverse' but several times faster, requires a 'Storable' instance.

542-- | Like 'reverse' but several times faster, requires a 'Storable' instance.

542--

543--

544+-- /O(n) space/

545+--

543-- /Pre-release/

546-- /Pre-release/

544{-# INLINE reverse' #-}

547{-# INLINE reverse' #-}

545reverse' :: (MonadIO m, Unbox a) => Stream m a -> Stream m a

548reverse' :: (MonadIO m, Unbox a) => Stream m a -> Stream m a

@@ -562,6 +565,8 @@ reverse' =

562-- generation function is monadic, unlike 'concatMap', it can produce an

565-- generation function is monadic, unlike 'concatMap', it can produce an

563-- effect at the beginning of each iteration of the inner loop.

566-- effect at the beginning of each iteration of the inner loop.

564--

567--

568+-- See 'unfoldMany' for a fusible alternative.

569+--

565{-# INLINE concatMapM #-}

570{-# INLINE concatMapM #-}

566concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b

571concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b

567concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)

572concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)

@@ -574,6 +579,8 @@ concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)

574-- >>> concatMap f = Stream.concat . fmap f

579-- >>> concatMap f = Stream.concat . fmap f

575-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)

580-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)

576--

581--

582+-- See 'unfoldMany' for a fusible alternative.

583+--

577{-# INLINE concatMap #-}

584{-# INLINE concatMap #-}

578concatMap ::Monad m => (a -> Stream m b) -> Stream m a -> Stream m b

585concatMap ::Monad m => (a -> Stream m b) -> Stream m a -> Stream m b

579concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)

586concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)

@@ -588,14 +595,21 @@ concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)

588--

595--

589-- See also: 'concat', 'sequence'

596-- See also: 'concat', 'sequence'

590--

597--

591−-- /Inhibits stream fusion/

598+-- See 'concatEffect2' for a fusible alternative.

599+--

600+-- /CPS/

592--

601--

593{-# INLINE concatEffect #-}

602{-# INLINE concatEffect #-}

594concatEffect :: Monad m => m (Stream m a) -> Stream m a

603concatEffect :: Monad m => m (Stream m a) -> Stream m a

595−-- concatEffect generator = concatMapM (\() -> generator) (fromPure ())

596concatEffect generator =

604concatEffect generator =

597 fromStreamK $ K.concatEffect $ fmap toStreamK generator

605 fromStreamK $ K.concatEffect $ fmap toStreamK generator

598

606

607+{-# INLINE concatEffect2 #-}

608+concatEffect2 :: Monad m => m (Stream m a) -> Stream m a

609+-- concatEffect generator = concatMapM (\() -> generator) (fromPure ())

610+concatEffect2 generator =

611+ fromStreamD $ D.concatEffect $ fmap toStreamD generator

612+

599-- XXX Need a more intuitive name, and need to reconcile the names

613-- XXX Need a more intuitive name, and need to reconcile the names

600-- foldMany/fold/parse/parseMany/parseManyPost etc.

614-- foldMany/fold/parse/parseMany/parseManyPost etc.

601

615

core/src/Streamly/Internal/Data/Stream/Eliminate.hs+2−2

@@ -135,7 +135,7 @@ import Prelude hiding (foldr, init, reverse)

135-- (Just h, t) -> Just (h, t)

136-- :}

137--

138−-- /Not fused/

138+-- /CPS/

139--

140{-# INLINE uncons #-}

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

@@ -252,7 +252,7 @@ parseBreakD parser strm = do

252

253-- | Parse a stream using the supplied 'Parser'.

254--

255−-- /Not fused/

255+-- /CPS/

256--

257{-# INLINE parseBreak #-}

258parseBreak :: Monad m => Parser a m b -> Stream m a -> m (Either ParseError b, Stream m a)

core/src/Streamly/Internal/Data/Stream/Exception.hs+2

@@ -176,6 +176,8 @@ bracketIO bef aft = bracketIO3 bef aft aft aft

176-- stream is abandoned @onGC@ is executed, if the stream encounters an

177-- exception @onException@ is executed.

178--

179+-- /Inhibits stream fusion/

180+--

179-- /Pre-release/

181-- /Pre-release/

180{-# INLINE bracketIO3 #-}

182{-# INLINE bracketIO3 #-}

181bracketIO3 :: (MonadIO m, MonadCatch m)

183bracketIO3 :: (MonadIO m, MonadCatch m)

core/src/Streamly/Internal/Data/Stream/Expand.hs+29−8

@@ -39,10 +39,12 @@ module Streamly.Internal.Data.Stream.Expand

39

40 -- ** Interleave

41 , interleave

42+ , interleave2

42 , interleaveFst

43 , interleaveFst

43− , interleaveMin

44− , interleaveFstSuffix2

45 , interleaveFst2

44 , interleaveFst2

45+ , interleaveFstSuffix2

46+ , interleaveMin

47+ , interleaveMin2

46

48

47 -- ** Round Robin

49 -- ** Round Robin

48 , roundrobin

50 , roundrobin

@@ -187,8 +189,9 @@ infixr 6 `append`

187-- be used to efficiently fold an infinite lazy container of streams

189-- be used to efficiently fold an infinite lazy container of streams

188-- 'concatMapWith' et. al.

190-- 'concatMapWith' et. al.

189--

191--

190−-- /Not fused/

192+-- See 'append2' for a fusible alternative.

191--

193--

194+-- /CPS/

192{-# INLINE append #-}

195{-# INLINE append #-}

193append :: Stream m a -> Stream m a -> Stream m a

196append :: Stream m a -> Stream m a -> Stream m a

194append = (<>)

197append = (<>)

@@ -206,16 +209,23 @@ append = (<>)

206-- weighting it can be used with 'concatMapWith' even on a large number of

209-- weighting it can be used with 'concatMapWith' even on a large number of

207-- streams.

210-- streams.

208--

211--

209−-- /Not fused/

212+-- See 'interleave2' for a fusible alternative.

210--

213--

214+-- /CPS/

211{-# INLINE interleave #-}

215{-# INLINE interleave #-}

212interleave :: Stream m a -> Stream m a -> Stream m a

216interleave :: Stream m a -> Stream m a -> Stream m a

213interleave s1 s2 = fromStreamK $ K.interleave (toStreamK s1) (toStreamK s2)

217interleave s1 s2 = fromStreamK $ K.interleave (toStreamK s1) (toStreamK s2)

214

218

219+{-# INLINE interleave2 #-}

220+interleave2 :: Monad m => Stream m a -> Stream m a -> Stream m a

221+interleave2 s1 s2 = fromStreamD $ D.interleave (toStreamD s1) (toStreamD s2)

222+

215-- | Like `interleave` but stops interleaving as soon as the first stream

223-- | Like `interleave` but stops interleaving as soon as the first stream

216-- stops.

224-- stops.

217--

225--

218−-- /Not fused/

226+-- See 'interleaveFst2' for a fusible alternative.

227+--

228+-- /CPS/

219{-# INLINE interleaveFst #-}

229{-# INLINE interleaveFst #-}

220interleaveFst :: Stream m a -> Stream m a -> Stream m a

230interleaveFst :: Stream m a -> Stream m a -> Stream m a

221interleaveFst s1 s2 =

231interleaveFst s1 s2 =

@@ -224,12 +234,19 @@ interleaveFst s1 s2 =

224-- | Like `interleave` but stops interleaving as soon as any of the two streams

234-- | Like `interleave` but stops interleaving as soon as any of the two streams

225-- stops.

235-- stops.

226--

236--

227−-- /Not fused/

237+-- See 'interleaveMin2' for a fusible alternative.

238+--

239+-- /CPS/

228{-# INLINE interleaveMin #-}

240{-# INLINE interleaveMin #-}

229interleaveMin :: Stream m a -> Stream m a -> Stream m a

241interleaveMin :: Stream m a -> Stream m a -> Stream m a

230interleaveMin s1 s2 =

242interleaveMin s1 s2 =

231 fromStreamK $ K.interleaveMin (toStreamK s1) (toStreamK s2)

243 fromStreamK $ K.interleaveMin (toStreamK s1) (toStreamK s2)

232

244

245+{-# INLINE interleaveMin2 #-}

246+interleaveMin2 :: Monad m => Stream m a -> Stream m a -> Stream m a

247+interleaveMin2 s1 s2 =

248+ fromStreamD $ D.interleaveMin (toStreamD s1) (toStreamD s2)

249+

233-- | Interleaves the outputs of two streams, yielding elements from each stream

250-- | Interleaves the outputs of two streams, yielding elements from each stream

234-- alternately, starting from the first stream. As soon as the first stream

251-- alternately, starting from the first stream. As soon as the first stream

235-- finishes, the output stops, discarding the remaining part of the second

252-- finishes, the output stops, discarding the remaining part of the second

@@ -315,8 +332,9 @@ roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)

315-- >>> Stream.fold Fold.toList $ Stream.mergeBy compare s1 s2

332-- >>> Stream.fold Fold.toList $ Stream.mergeBy compare s1 s2

316-- [1,2,3,4,5,6,8]

333-- [1,2,3,4,5,6,8]

317--

334--

318−-- See also: 'mergeByM2'

335+-- See 'mergeByM2' for a fusible alternative.

319--

336--

337+-- /CPS/

320{-# INLINE mergeBy #-}

338{-# INLINE mergeBy #-}

321mergeBy :: (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

339mergeBy :: (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

322mergeBy f m1 m2 = fromStreamK $ K.mergeBy f (toStreamK m1) (toStreamK m2)

340mergeBy f m1 m2 = fromStreamK $ K.mergeBy f (toStreamK m1) (toStreamK m2)

@@ -349,8 +367,9 @@ mergeBy f m1 m2 = fromStreamK $ K.mergeBy f (toStreamK m1) (toStreamK m2)

349-- :}

367-- :}

350-- [1,1,2,1,1,2,1,1,2]

368-- [1,1,2,1,1,2,1,1,2]

351--

369--

352−-- See also: 'mergeByM2'

370+-- See 'mergeByM2' for a fusible alternative.

353--

371--

372+-- /CPS/

354{-# INLINE mergeByM #-}

373{-# INLINE mergeByM #-}

355mergeByM

374mergeByM

356 :: Monad m

375 :: Monad m

@@ -590,6 +609,8 @@ concatSmapMWith combine f initial =

590--

609--

591-- /Caution: the stream of streams must be finite/

610-- /Caution: the stream of streams must be finite/

592--

611--

612+-- /CPS/

613+--

593-- /Pre-release/

614-- /Pre-release/

594--

615--

595{-# INLINE mergeMapWith #-}

616{-# INLINE mergeMapWith #-}

core/src/Streamly/Internal/Data/Stream/Generate.hs+2−2

@@ -387,7 +387,7 @@ iterateM step = fromStreamD . D.iterateM step

387-- Note that the function @f@ must be lazy in its argument, that's why we use

388-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.

389--

390−-- /Not fused/

390+-- /CPS/

391--

392-- /Pre-release/

393{-# INLINE mfix #-}

@@ -403,7 +403,7 @@ mfix f = fromStreamK $ K.mfix (toStreamK . f)

403--

404-- Construct a stream from a 'Foldable' containing pure values:

405--

406−-- /Not fused/

406+-- /CPS/

407--

408{-# INLINE fromFoldable #-}

409fromFoldable :: Foldable f => f a -> Stream m a

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

@@ -309,7 +309,7 @@ infixr 5 `cons`

309--

310-- >>> cons x xs = return x `Stream.consM` xs

311--

312−-- /Not fused/

312+-- /CPS/

313--

314{-# INLINE_NORMAL cons #-}

315cons :: a -> Stream m a -> Stream m a

@@ -333,7 +333,7 @@ infixr 5 `consM`

333--

334-- >>> consM x xs = Stream.fromEffect x `Stream.append` xs

335--

336−-- /Not fused/

336+-- /CPS/

337--

338{-# INLINE consM #-}

339{-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}