e95145c · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use fused annotation on "many", and update docs

e95145cHarendra Kumar4 years ago

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

compare with master

unifiedsplit

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

1601 = ManyFirst !s1 !s2

1602 | ManyLoop !s1 !s2

1603

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

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

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

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

1606−-- fold results using @collect@.

1606+-- using the @second@ fold.

1607--

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

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

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

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

1612--

1613−-- Stops when @collect@ stops.

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

1614--

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

1616--

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

1666 Partial s -> cextract s

1667 Done b -> return b

1668

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

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

1670−-- received.

1670+-- input is received.

1671--

1672-- /Internal/

1673--

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

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

208import Data.Bifunctor (Bifunctor(..))

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

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

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

212

211

213import qualified Control.Monad.Fail as Fail

212import qualified Control.Monad.Fail as Fail

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

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

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

922 assertM(n == cnt)

921 assertM(n == cnt)

923 return $ Continue n (AltParseL 0 s)

922 return $ Continue n (AltParseL 0 s)

924

923

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

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

926+

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

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

926--

928--

927-- /Pre-release/

929-- /Pre-release/

928--

930--

929{-# INLINE splitMany #-}

931{-# 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

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

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

932 Parser step initial extract

934 Parser step initial extract

933

935

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

941 FL.Partial fs -> do

943 FL.Partial fs -> do

942 pres <- initial1

944 pres <- initial1

943 case pres of

945 case pres of

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

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

945 IDone pb ->

947 IDone pb ->

946 runCollectorWith (handleCollect partial done) fs pb

948 runCollectorWith (handleCollect partial done) fs pb

947 IError _ -> done <$> fextract fs

949 IError _ -> done <$> fextract fs

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

949

951

950 runCollectorWith cont fs pb = fstep fs pb >>= cont

952 runCollectorWith cont fs pb = fstep fs pb >>= cont

951

953

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

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

952 initial = finitial >>= handleCollect IPartial IDone

956 initial = finitial >>= handleCollect IPartial IDone

953

957

954 {-# INLINE step #-}

958 {-# INLINE step #-}

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

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

956 r <- step1 st a

960 r <- step1 st a

957 let cnt1 = cnt + 1

961 let cnt1 = cnt + 1

958 case r of

962 case r of

959 Partial n s -> do

963 Partial n s -> do

960 assertM(cnt1 - n >= 0)

964 assertM(cnt1 - n >= 0)

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

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

962 Continue n s -> do

966 Continue n s -> do

963 assertM(cnt1 - n >= 0)

967 assertM(cnt1 - n >= 0)

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

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

965 Done n b -> do

969 Done n b -> do

966 assertM(cnt1 - n >= 0)

970 assertM(cnt1 - n >= 0)

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

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

968 Error _ -> do

972 Error _ -> do

969 xs <- fextract fs

973 xs <- fextract fs

970− return $ Done cnt1 xs

974+ return $ Done cnt xs

971

975

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

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

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

977+ extract (Fused3 s cnt fs) = do

974 r <- extract1 s

978 r <- extract1 s

975 case r of

979 case r of

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

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

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

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

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

984 Continue n s1 -> do

988 Continue n s1 -> do

985 assertM(n == cnt)

989 assertM(n == cnt)

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

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

987

991

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

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

989-- is received.

993-- is received.

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

1005 FL.Partial fs -> do

1009 FL.Partial fs -> do

1006 pres <- initial1

1010 pres <- initial1

1007 case pres of

1011 case pres of

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

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

1009 IDone pb ->

1013 IDone pb ->

1010 runCollectorWith (handleCollect partial done) fs pb

1014 runCollectorWith (handleCollect partial done) fs pb

1011 IError _ -> done <$> fextract fs

1015 IError _ -> done <$> fextract fs

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

1016 initial = finitial >>= handleCollect IPartial IDone

1020 initial = finitial >>= handleCollect IPartial IDone

1017

1021

1018 {-# INLINE step #-}

1022 {-# INLINE step #-}

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

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

1020 r <- step1 st a

1024 r <- step1 st a

1021 let cnt1 = cnt + 1

1025 let cnt1 = cnt + 1

1022 case r of

1026 case r of

1023 Partial n s -> do

1027 Partial n s -> do

1024 assertM(cnt1 - n >= 0)

1028 assertM(cnt1 - n >= 0)

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

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

1026 Continue n s -> do

1030 Continue n s -> do

1027 assertM(cnt1 - n >= 0)

1031 assertM(cnt1 - n >= 0)

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

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

1029 Done n b -> do

1033 Done n b -> do

1030 assertM(cnt1 - n >= 0)

1034 assertM(cnt1 - n >= 0)

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

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

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

1033 xs <- fextract fs

1037 xs <- fextract fs

1034 return $ Done cnt1 xs

1038 return $ Done cnt1 xs

1035

1039

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

1040+ extract (Fused3 s cnt fs) = do

1037 r <- extract1 s

1041 r <- extract1 s

1038 case r of

1042 case r of

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

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

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

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

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

1047 Continue n s1 -> do

1051 Continue n s1 -> do

1048 assertM(n == cnt)

1052 assertM(n == cnt)

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

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

1050

1054

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

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

1052--

1056--

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

1067 FL.Partial fs -> do

1071 FL.Partial fs -> do

1068 pres <- initial1

1072 pres <- initial1

1069 case pres of

1073 case pres of

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

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

1071 IDone pb ->

1075 IDone pb ->

1072 runCollectorWith (handleCollect partial done) fs pb

1076 runCollectorWith (handleCollect partial done) fs pb

1073 IError _ -> done <$> fextract fs

1077 IError _ -> done <$> fextract fs

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

1081 FL.Partial fs -> do

1085 FL.Partial fs -> do

1082 pres <- initial1

1086 pres <- initial1

1083 case pres of

1087 case pres of

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

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

1085 IDone pb ->

1089 IDone pb ->

1086 runCollectorWith (handleCollect IPartial IDone) fs pb

1090 runCollectorWith (handleCollect IPartial IDone) fs pb

1087 IError err -> return $ IError err

1091 IError err -> return $ IError err

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

1092 ++ " consuming any elements."

1096 ++ " consuming any elements."

1093

1097

1094 {-# INLINE step #-}

1098 {-# INLINE step #-}

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

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

1096 r <- step1 st a

1100 r <- step1 st a

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

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

1098 let cnt1 = cnt + 1

1102 let cnt1 = cnt + 1

1099 case r of

1103 case r of

1100 Partial n s -> do

1104 Partial n s -> do

1101 assertM(cnt1 - n >= 0)

1105 assertM(cnt1 - n >= 0)

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

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

1103 Continue n s -> do

1107 Continue n s -> do

1104 assertM(cnt1 - n >= 0)

1108 assertM(cnt1 - n >= 0)

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

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

1106 Done n b -> do

1110 Done n b -> do

1107 assertM(cnt1 - n >= 0)

1111 assertM(cnt1 - n >= 0)

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

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

1109 Error err -> return $ Error err

1113 Error err -> return $ Error err

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

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

1111 r <- step1 st a

1115 r <- step1 st a

1112 let cnt1 = cnt + 1

1116 let cnt1 = cnt + 1

1113 case r of

1117 case r of

1114 Partial n s -> do

1118 Partial n s -> do

1115 assertM(cnt1 - n >= 0)

1119 assertM(cnt1 - n >= 0)

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

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

1117 Continue n s -> do

1121 Continue n s -> do

1118 assertM(cnt1 - n >= 0)

1122 assertM(cnt1 - n >= 0)

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

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

1120 Done n b -> do

1124 Done n b -> do

1121 assertM(cnt1 - n >= 0)

1125 assertM(cnt1 - n >= 0)

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

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

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

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

1124

1128

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

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

1126 r <- extract1 s

1130 r <- extract1 s

1127 case r of

1131 case r of

1128 Error err -> return (Error err)

1132 Error err -> return (Error err)

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

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

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

1136 Continue n s1 -> do

1140 Continue n s1 -> do

1137 assertM(n == cnt)

1141 assertM(n == cnt)

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

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

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

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

1140 r <- extract1 s

1144 r <- extract1 s

1141 case r of

1145 case r of

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

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

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

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

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

1150 Continue n s1 -> do

1154 Continue n s1 -> do

1151 assertM(n == cnt)

1155 assertM(n == cnt)

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

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

1153

1157

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

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

1155-- any input.

1159-- any input.

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

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

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

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

1717

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+

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

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

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

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

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

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

1721--

1727--

1722-- Example, empty stream:

1728-- Example, empty stream:

1723--

1729--

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

1786

1792

1787-- XXX Nested foldMany does not fuse.

1793-- XXX Nested foldMany does not fuse.

1788

1794

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

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

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

1796+-- stream.

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

1792--

1797--

1793-- Definition:

1798-- Definition:

1794--

1799--

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

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

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

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

41-- 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+

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

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

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

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

45--

48--

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

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

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

50-- >>> 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]]

49--

56--

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

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

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

58-- 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'

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

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

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

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

107

114

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

116+

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

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

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

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

110--

119--