e95145c · streamly · git.iko.soy

git.iko.soy/streamly

Use fused annotation on "many", and update docs

Add a comment about removing foldManyPost And manyPost, splitManyPost

commite95145ca40f15d8308d7ac0be4fb1a6208623afa authorHarendra Kumar <harendra@composewell.com> dateTue 21 Feb 2023, 10:03 (4 years ago) parentf0db109

4 files changed+61−43

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/Fold/Type.hs+6−6 core/src/Streamly/Internal/Data/Parser/ParserD/Type.hs+32−28 core/src/Streamly/Internal/Data/Stream/StreamD/Type.hs+11−6 src/Streamly/Internal/Data/Fold/Async.hs+12−3

core/src/Streamly/Internal/Data/Fold/Type.hs+6−6

@@ -1601,16 +1601,16 @@ data ManyState s1 s2

16011601 = ManyFirst !s1 !s2

16021602 | ManyLoop !s1 !s2

16031603

1604−-- | Collect zero or more applications of a fold. @many split collect@ applies

1605−-- the @split@ fold repeatedly on the input stream and accumulates zero or more

1606−-- fold results using @collect@.

1604+-- | Collect zero or more applications of a fold. @many first second@ applies

1605+-- the @first@ fold repeatedly on the input stream and accumulates it's results

1606+-- using the @second@ fold.

16071607--

16081608-- >>> two = Fold.take 2 Fold.toList

16091609-- >>> twos = Fold.many two Fold.toList

16101610-- >>> Stream.fold twos $ Stream.fromList [1..10]

16111611-- [[1,2],[3,4],[5,6],[7,8],[9,10]]

16121612--

1613−-- Stops when @collect@ stops.

1613+-- Stops when @second@ fold stops.

16141614--

16151615-- See also: 'Data.Stream.concatMap', 'Data.Stream.foldMany'

16161616--

@@ -1666,8 +1666,8 @@ many (Fold sstep sinitial sextract) (Fold cstep cinitial cextract) =

16661666 Partial s -> cextract s

16671667 Done b -> return b

16681668

1669−-- | Like many, but inner fold emits an output at the end even if no input is

1670−-- received.

1669+-- | Like many, but the "first" fold emits an output at the end even if no

1670+-- input is received.

16711671--

16721672-- /Internal/

16731673--

core/src/Streamly/Internal/Data/Parser/ParserD/Type.hs+32−28

@@ -208,7 +208,6 @@ import Control.Monad.IO.Class (MonadIO, liftIO)

208208import Data.Bifunctor (Bifunctor(..))

209209import Fusion.Plugin.Types (Fuse(..))

210210import Streamly.Internal.Data.Fold.Type (Fold(..), toList)

211−import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))

212211

213212import qualified Control.Monad.Fail as Fail

214213import qualified Streamly.Internal.Data.Fold.Type as FL

@@ -922,12 +921,15 @@ alt (Parser stepL initialL extractL) (Parser stepR initialR extractR) =

922921 assertM(n == cnt)

923922 return $ Continue n (AltParseL 0 s)

924923

924+{-# ANN type Fused3 Fuse #-}

925+data Fused3 a b c = Fused3 !a !b !c

926+

925927-- | See documentation of 'Streamly.Internal.Data.Parser.many'.

926928--

927929-- /Pre-release/

928930--

929931{-# INLINE splitMany #-}

930−splitMany :: Monad m => Parser a m b -> Fold m b c -> Parser a m c

932+splitMany :: Monad m => Parser a m b -> Fold m b c -> Parser a m c

931933splitMany (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

932934 Parser step initial extract

933935

@@ -941,7 +943,7 @@ splitMany (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

941943 FL.Partial fs -> do

942944 pres <- initial1

943945 case pres of

944− IPartial ps -> return $ partial $ Tuple3' ps 0 fs

946+ IPartial ps -> return $ partial $ Fused3 ps 0 fs

945947 IDone pb ->

946948 runCollectorWith (handleCollect partial done) fs pb

947949 IError _ -> done <$> fextract fs

@@ -949,28 +951,30 @@ splitMany (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

949951

950952 runCollectorWith cont fs pb = fstep fs pb >>= cont

951953

954+ -- See notes in Fold.many for the reason why the parser must be initialized

955+ -- right away instead of on first input.

952956 initial = finitial >>= handleCollect IPartial IDone

953957

954958 {-# INLINE step #-}

955− step (Tuple3' st cnt fs) a = do

959+ step (Fused3 st cnt fs) a = do

956960 r <- step1 st a

957961 let cnt1 = cnt + 1

958962 case r of

959963 Partial n s -> do

960964 assertM(cnt1 - n >= 0)

961− return $ Continue n (Tuple3' s (cnt1 - n) fs)

965+ return $ Continue n (Fused3 s (cnt1 - n) fs)

962966 Continue n s -> do

963967 assertM(cnt1 - n >= 0)

964− return $ Continue n (Tuple3' s (cnt1 - n) fs)

968+ return $ Continue n (Fused3 s (cnt1 - n) fs)

965969 Done n b -> do

966970 assertM(cnt1 - n >= 0)

967971 fstep fs b >>= handleCollect (Partial n) (Done n)

968972 Error _ -> do

969973 xs <- fextract fs

970− return $ Done cnt1 xs

974+ return $ Done cnt xs

971975

972− extract (Tuple3' _ 0 fs) = fmap (Done 0) (fextract fs)

973− extract (Tuple3' s cnt fs) = do

976+ extract (Fused3 _ 0 fs) = fmap (Done 0) (fextract fs)

977+ extract (Fused3 s cnt fs) = do

974978 r <- extract1 s

975979 case r of

976980 Error _ -> fmap (Done cnt) (fextract fs)

@@ -983,7 +987,7 @@ splitMany (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

983987 Partial _ _ -> error "splitMany: Partial in extract"

984988 Continue n s1 -> do

985989 assertM(n == cnt)

986− return (Continue n (Tuple3' s1 0 fs))

990+ return (Continue n (Fused3 s1 0 fs))

987991

988992-- | Like splitMany, but inner fold emits an output at the end even if no input

989993-- is received.

@@ -1005,7 +1009,7 @@ splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

10051009 FL.Partial fs -> do

10061010 pres <- initial1

10071011 case pres of

1008− IPartial ps -> return $ partial $ Tuple3' ps 0 fs

1012+ IPartial ps -> return $ partial $ Fused3 ps 0 fs

10091013 IDone pb ->

10101014 runCollectorWith (handleCollect partial done) fs pb

10111015 IError _ -> done <$> fextract fs

@@ -1016,16 +1020,16 @@ splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

10161020 initial = finitial >>= handleCollect IPartial IDone

10171021

10181022 {-# INLINE step #-}

1019− step (Tuple3' st cnt fs) a = do

1023+ step (Fused3 st cnt fs) a = do

10201024 r <- step1 st a

10211025 let cnt1 = cnt + 1

10221026 case r of

10231027 Partial n s -> do

10241028 assertM(cnt1 - n >= 0)

1025− return $ Continue n (Tuple3' s (cnt1 - n) fs)

1029+ return $ Continue n (Fused3 s (cnt1 - n) fs)

10261030 Continue n s -> do

10271031 assertM(cnt1 - n >= 0)

1028− return $ Continue n (Tuple3' s (cnt1 - n) fs)

1032+ return $ Continue n (Fused3 s (cnt1 - n) fs)

10291033 Done n b -> do

10301034 assertM(cnt1 - n >= 0)

10311035 fstep fs b >>= handleCollect (Partial n) (Done n)

@@ -1033,7 +1037,7 @@ splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

10331037 xs <- fextract fs

10341038 return $ Done cnt1 xs

10351039

1036− extract (Tuple3' s cnt fs) = do

1040+ extract (Fused3 s cnt fs) = do

10371041 r <- extract1 s

10381042 case r of

10391043 Error _ -> fmap (Done cnt) (fextract fs)

@@ -1046,7 +1050,7 @@ splitManyPost (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

10461050 Partial _ _ -> error "splitMany: Partial in extract"

10471051 Continue n s1 -> do

10481052 assertM(n == cnt)

1049− return (Continue n (Tuple3' s1 0 fs))

1053+ return (Continue n (Fused3 s1 0 fs))

10501054

10511055-- | See documentation of 'Streamly.Internal.Data.Parser.some'.

10521056--

@@ -1067,7 +1071,7 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

10671071 FL.Partial fs -> do

10681072 pres <- initial1

10691073 case pres of

1070− IPartial ps -> return $ partial $ Tuple3' ps 0 $ Right fs

1074+ IPartial ps -> return $ partial $ Fused3 ps 0 $ Right fs

10711075 IDone pb ->

10721076 runCollectorWith (handleCollect partial done) fs pb

10731077 IError _ -> done <$> fextract fs

@@ -1081,7 +1085,7 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

10811085 FL.Partial fs -> do

10821086 pres <- initial1

10831087 case pres of

1084− IPartial ps -> return $ IPartial $ Tuple3' ps 0 $ Left fs

1088+ IPartial ps -> return $ IPartial $ Fused3 ps 0 $ Left fs

10851089 IDone pb ->

10861090 runCollectorWith (handleCollect IPartial IDone) fs pb

10871091 IError err -> return $ IError err

@@ -1092,37 +1096,37 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

10921096 ++ " consuming any elements."

10931097

10941098 {-# INLINE step #-}

1095− step (Tuple3' st cnt (Left fs)) a = do

1099+ step (Fused3 st cnt (Left fs)) a = do

10961100 r <- step1 st a

10971101 -- In the Left state, count is used only for the assert

10981102 let cnt1 = cnt + 1

10991103 case r of

11001104 Partial n s -> do

11011105 assertM(cnt1 - n >= 0)

1102− return $ Continue n (Tuple3' s (cnt1 - n) (Left fs))

1106+ return $ Continue n (Fused3 s (cnt1 - n) (Left fs))

11031107 Continue n s -> do

11041108 assertM(cnt1 - n >= 0)

1105− return $ Continue n (Tuple3' s (cnt1 - n) (Left fs))

1109+ return $ Continue n (Fused3 s (cnt1 - n) (Left fs))

11061110 Done n b -> do

11071111 assertM(cnt1 - n >= 0)

11081112 fstep fs b >>= handleCollect (Partial n) (Done n)

11091113 Error err -> return $ Error err

1110− step (Tuple3' st cnt (Right fs)) a = do

1114+ step (Fused3 st cnt (Right fs)) a = do

11111115 r <- step1 st a

11121116 let cnt1 = cnt + 1

11131117 case r of

11141118 Partial n s -> do

11151119 assertM(cnt1 - n >= 0)

1116− return $ Partial n (Tuple3' s (cnt1 - n) (Right fs))

1120+ return $ Partial n (Fused3 s (cnt1 - n) (Right fs))

11171121 Continue n s -> do

11181122 assertM(cnt1 - n >= 0)

1119− return $ Continue n (Tuple3' s (cnt1 - n) (Right fs))

1123+ return $ Continue n (Fused3 s (cnt1 - n) (Right fs))

11201124 Done n b -> do

11211125 assertM(cnt1 - n >= 0)

11221126 fstep fs b >>= handleCollect (Partial n) (Done n)

11231127 Error _ -> Done cnt1 <$> fextract fs

11241128

1125− extract (Tuple3' s cnt (Left fs)) = do

1129+ extract (Fused3 s cnt (Left fs)) = do

11261130 r <- extract1 s

11271131 case r of

11281132 Error err -> return (Error err)

@@ -1135,8 +1139,8 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

11351139 Partial _ _ -> error "splitSome: Partial in extract"

11361140 Continue n s1 -> do

11371141 assertM(n == cnt)

1138− return (Continue n (Tuple3' s1 0 (Left fs)))

1139− extract (Tuple3' s cnt (Right fs)) = do

1142+ return (Continue n (Fused3 s1 0 (Left fs)))

1143+ extract (Fused3 s cnt (Right fs)) = do

11401144 r <- extract1 s

11411145 case r of

11421146 Error _ -> fmap (Done cnt) (fextract fs)

@@ -1149,7 +1153,7 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =

11491153 Partial _ _ -> error "splitSome: Partial in extract"

11501154 Continue n s1 -> do

11511155 assertM(n == cnt)

1152− return (Continue n (Tuple3' s1 0 (Right fs)))

1156+ return (Continue n (Fused3 s1 0 (Right fs)))

11531157

11541158-- | A parser that always fails with an error message without consuming

11551159-- any input.

core/src/Streamly/Internal/Data/Stream/StreamD/Type.hs+11−6

@@ -1715,9 +1715,15 @@ data FoldManyPost s fs b a

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

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

17171717

1718−-- | Like 'foldMany' but evaluates the fold before the stream, and yields its

1719−-- output even if the stream is empty, therefore, always results in a non-empty

1720−-- output even on an empty stream (default result of the fold).

1718+-- XXX foldManyPost keeps the last fold always partial. if the last fold is

1719+-- complete then another fold is applied on empty input. This is used for

1720+-- applying folds like takeEndBy such that the last element is not the

1721+-- separator (infix style). But that looks like a hack. We should remove this

1722+-- and use a custom combinator for infix parsing.

1723+

1724+-- | Like 'foldMany' but evaluates the fold even if the fold did not receive

1725+-- any input, therefore, always results in a non-empty output even on an empty

1726+-- stream (default result of the fold).

17211727--

17221728-- Example, empty stream:

17231729--

@@ -1786,9 +1792,8 @@ data FoldMany s fs b a

17861792

17871793-- XXX Nested foldMany does not fuse.

17881794

1789−-- | Apply a 'Fold' repeatedly on a stream and emit the results in the

1790−-- output stream. Unlike 'foldManyPost' it evaluates the fold after the stream,

1791−-- therefore, an empty input stream results in an empty output stream.

1795+-- | Apply a 'Fold' repeatedly on a stream and emit the results in the output

1796+-- stream.

17921797--

17931798-- Definition:

17941799--

src/Streamly/Internal/Data/Fold/Async.hs+12−3

@@ -39,13 +39,20 @@ import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))

3939-- XXX If n is 0 return immediately in initial.

4040-- XXX we should probably discard the input received after the timeout like

4141-- takeEndBy_.

42−--

42+

43+-- XXX The foldMany doctest is important for consistency with the "many" fold,

44+-- and intervalsOf. We should put it in tests.

45+

4346-- | @takeInterval n fold@ uses @fold@ to fold the input items arriving within

4447-- a window of first @n@ seconds.

4548--

46−-- >>> input = Stream.delay 0.1 $ Stream.fromList [1..]

49+-- >>> input = Stream.delay 0.2 $ Stream.fromList [1..10]

4750-- >>> Stream.fold (Fold.takeInterval 1.0 Fold.toList) input

48−-- [1,2,3,4,5,6,7,8,9,10,11]

51+-- [1,2,3,4,5,6]

52+--

53+-- >>> f = Fold.takeInterval 0.5 Fold.toList

54+-- >>> Stream.fold Fold.toList $ Stream.foldMany f input

55+-- [[1,2,3,4],[5,6,7],[8,9,10]]

4956--

5057-- Stops when @fold@ stops or when the timeout occurs. Note that the fold needs

5158-- an input after the timeout to stop. For example, if no input is pushed to

@@ -105,6 +112,8 @@ takeInterval n (Fold step initial done) = Fold step' initial' done'

105112-- each fold can group the input differently e.g. by one second, one minute and

106113-- one hour windows respectively and fold each resulting stream of folds.

107114

115+-- XXX This needs to be fixed like intervalsOf in Data.Stream.Time.

116+

108117-- | Group the input stream into windows of n second each using the first fold

109118-- and then fold the resulting groups using the second fold.

110119--