git.iko.soy/streamly
on master help sign in
Add CPS annotation on CPS ops
commitb6185ba2c30e77564abb3794bd0ab4498f46779a authorHarendra Kumar <harendra@composewell.com> dateWed 14 Dec 2022, 17:45 (4 years ago) parent238bb2b
8 files changed+64−18
compare with master
unifiedsplit
core/src/Streamly/Data/Stream.hs+9 core/src/Streamly/Internal/Data/Fold.hs+1−1 core/src/Streamly/Internal/Data/Stream/Bottom.hs+17−3 core/src/Streamly/Internal/Data/Stream/Eliminate.hs+2−2 core/src/Streamly/Internal/Data/Stream/Exception.hs+2 core/src/Streamly/Internal/Data/Stream/Expand.hs+29−8 core/src/Streamly/Internal/Data/Stream/Generate.hs+2−2 core/src/Streamly/Internal/Data/Stream/Type.hs+2−2
core/src/Streamly/Data/Stream.hs+9
@@ -79,6 +79,15 @@
7979-- combinators. See the <https://github.com/composewell/streamly-examples>
8080-- repository for many more real world examples of stream programming.
8181--
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+--
8291module Streamly.Data.Stream
8392 (
8493 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
381381--
382382-- >>> driveBreak = flip Stream.foldBreak
383383--
384−-- /Not fused/
384+-- /CPS/
385385--
386386{-# INLINE driveBreak #-}
387387driveBreak :: 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
5353
5454 -- * Expand
5555 , concatEffect
56+ , concatEffect2
5657 , concatMapM
5758 , concatMap
5859
@@ -228,7 +229,7 @@ fold fl strm = D.fold fl $ D.fromStreamK $ toStreamK strm
228229-- | Like 'fold' but also returns the remaining stream. The resulting stream
229230-- would be 'Stream.nil' if the stream finished before the fold.
230231--
231−-- /Not fused/
232+-- /CPS/
232233--
233234{-# INLINE foldBreak #-}
234235foldBreak :: 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
540541
541542-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
542543--
544+-- /O(n) space/
545+--
543546-- /Pre-release/
544547{-# INLINE reverse' #-}
545548reverse' :: (MonadIO m, Unbox a) => Stream m a -> Stream m a
@@ -562,6 +565,8 @@ reverse' =
562565-- generation function is monadic, unlike 'concatMap', it can produce an
563566-- effect at the beginning of each iteration of the inner loop.
564567--
568+-- See 'unfoldMany' for a fusible alternative.
569+--
565570{-# INLINE concatMapM #-}
566571concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b
567572concatMapM 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)
574579-- >>> concatMap f = Stream.concat . fmap f
575580-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
576581--
582+-- See 'unfoldMany' for a fusible alternative.
583+--
577584{-# INLINE concatMap #-}
578585concatMap ::Monad m => (a -> Stream m b) -> Stream m a -> Stream m b
579586concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
@@ -588,14 +595,21 @@ concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
588595--
589596-- See also: 'concat', 'sequence'
590597--
591−-- /Inhibits stream fusion/
598+-- See 'concatEffect2' for a fusible alternative.
599+--
600+-- /CPS/
592601--
593602{-# INLINE concatEffect #-}
594603concatEffect :: Monad m => m (Stream m a) -> Stream m a
595−-- concatEffect generator = concatMapM (\() -> generator) (fromPure ())
596604concatEffect generator =
597605 fromStreamK $ K.concatEffect $ fmap toStreamK generator
598606
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+
599613-- XXX Need a more intuitive name, and need to reconcile the names
600614-- foldMany/fold/parse/parseMany/parseManyPost etc.
601615
core/src/Streamly/Internal/Data/Stream/Eliminate.hs+2−2
@@ -135,7 +135,7 @@ import Prelude hiding (foldr, init, reverse)
135135-- (Just h, t) -> Just (h, t)
136136-- :}
137137--
138−-- /Not fused/
138+-- /CPS/
139139--
140140{-# INLINE uncons #-}
141141uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
@@ -252,7 +252,7 @@ parseBreakD parser strm = do
252252
253253-- | Parse a stream using the supplied 'Parser'.
254254--
255−-- /Not fused/
255+-- /CPS/
256256--
257257{-# INLINE parseBreak #-}
258258parseBreak :: 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
176176-- stream is abandoned @onGC@ is executed, if the stream encounters an
177177-- exception @onException@ is executed.
178178--
179+-- /Inhibits stream fusion/
180+--
179181-- /Pre-release/
180182{-# INLINE bracketIO3 #-}
181183bracketIO3 :: (MonadIO m, MonadCatch m)
core/src/Streamly/Internal/Data/Stream/Expand.hs+29−8
@@ -39,10 +39,12 @@ module Streamly.Internal.Data.Stream.Expand
3939
4040 -- ** Interleave
4141 , interleave
42+ , interleave2
4243 , interleaveFst
43− , interleaveMin
44− , interleaveFstSuffix2
4544 , interleaveFst2
45+ , interleaveFstSuffix2
46+ , interleaveMin
47+ , interleaveMin2
4648
4749 -- ** Round Robin
4850 , roundrobin
@@ -187,8 +189,9 @@ infixr 6 `append`
187189-- be used to efficiently fold an infinite lazy container of streams
188190-- 'concatMapWith' et. al.
189191--
190−-- /Not fused/
192+-- See 'append2' for a fusible alternative.
191193--
194+-- /CPS/
192195{-# INLINE append #-}
193196append :: Stream m a -> Stream m a -> Stream m a
194197append = (<>)
@@ -206,16 +209,23 @@ append = (<>)
206209-- weighting it can be used with 'concatMapWith' even on a large number of
207210-- streams.
208211--
209−-- /Not fused/
212+-- See 'interleave2' for a fusible alternative.
210213--
214+-- /CPS/
211215{-# INLINE interleave #-}
212216interleave :: Stream m a -> Stream m a -> Stream m a
213217interleave s1 s2 = fromStreamK $ K.interleave (toStreamK s1) (toStreamK s2)
214218
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+
215223-- | Like `interleave` but stops interleaving as soon as the first stream
216224-- stops.
217225--
218−-- /Not fused/
226+-- See 'interleaveFst2' for a fusible alternative.
227+--
228+-- /CPS/
219229{-# INLINE interleaveFst #-}
220230interleaveFst :: Stream m a -> Stream m a -> Stream m a
221231interleaveFst s1 s2 =
@@ -224,12 +234,19 @@ interleaveFst s1 s2 =
224234-- | Like `interleave` but stops interleaving as soon as any of the two streams
225235-- stops.
226236--
227−-- /Not fused/
237+-- See 'interleaveMin2' for a fusible alternative.
238+--
239+-- /CPS/
228240{-# INLINE interleaveMin #-}
229241interleaveMin :: Stream m a -> Stream m a -> Stream m a
230242interleaveMin s1 s2 =
231243 fromStreamK $ K.interleaveMin (toStreamK s1) (toStreamK s2)
232244
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+
233250-- | Interleaves the outputs of two streams, yielding elements from each stream
234251-- alternately, starting from the first stream. As soon as the first stream
235252-- 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)
315332-- >>> Stream.fold Fold.toList $ Stream.mergeBy compare s1 s2
316333-- [1,2,3,4,5,6,8]
317334--
318−-- See also: 'mergeByM2'
335+-- See 'mergeByM2' for a fusible alternative.
319336--
337+-- /CPS/
320338{-# INLINE mergeBy #-}
321339mergeBy :: (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
322340mergeBy 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)
349367-- :}
350368-- [1,1,2,1,1,2,1,1,2]
351369--
352−-- See also: 'mergeByM2'
370+-- See 'mergeByM2' for a fusible alternative.
353371--
372+-- /CPS/
354373{-# INLINE mergeByM #-}
355374mergeByM
356375 :: Monad m
@@ -590,6 +609,8 @@ concatSmapMWith combine f initial =
590609--
591610-- /Caution: the stream of streams must be finite/
592611--
612+-- /CPS/
613+--
593614-- /Pre-release/
594615--
595616{-# INLINE mergeMapWith #-}
core/src/Streamly/Internal/Data/Stream/Generate.hs+2−2
@@ -387,7 +387,7 @@ iterateM step = fromStreamD . D.iterateM step
387387-- Note that the function @f@ must be lazy in its argument, that's why we use
388388-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.
389389--
390−-- /Not fused/
390+-- /CPS/
391391--
392392-- /Pre-release/
393393{-# INLINE mfix #-}
@@ -403,7 +403,7 @@ mfix f = fromStreamK $ K.mfix (toStreamK . f)
403403--
404404-- Construct a stream from a 'Foldable' containing pure values:
405405--
406−-- /Not fused/
406+-- /CPS/
407407--
408408{-# INLINE fromFoldable #-}
409409fromFoldable :: Foldable f => f a -> Stream m a
core/src/Streamly/Internal/Data/Stream/Type.hs+2−2
@@ -309,7 +309,7 @@ infixr 5 `cons`
309309--
310310-- >>> cons x xs = return x `Stream.consM` xs
311311--
312−-- /Not fused/
312+-- /CPS/
313313--
314314{-# INLINE_NORMAL cons #-}
315315cons :: a -> Stream m a -> Stream m a
@@ -333,7 +333,7 @@ infixr 5 `consM`
333333--
334334-- >>> consM x xs = Stream.fromEffect x `Stream.append` xs
335335--
336−-- /Not fused/
336+-- /CPS/
337337--
338338{-# INLINE consM #-}
339339{-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}