on master help sign in
Add concatParse to chain parsers on a stream

Add mconcatTo, update fold docs Add fold docs/snippets using Monoids. Add benchmark for sum using foldMap Add concatParse benchmark Add splitParseTill, update docs

commit684d2786ebe67205e4a2301c4892945db090f996 authorHarendra Kumar <harendra@composewell.com> dateWed 22 Apr 2020, 07:28 (6 years ago) parent688d68d
9 files changed+304−30
benchmark/Streamly/Benchmark/Data/Fold.hs+2−1 benchmark/Streamly/Benchmark/Data/Parser.hs+50−13 benchmark/Streamly/Benchmark/Data/ParserD.hs+9−4 benchmark/Streamly/Benchmark/FileIO/Stream.hs+1−1 benchmark/streamly-benchmarks.cabal+3−3 src/Streamly/Data/Fold.hs+6 src/Streamly/Internal/Data/Fold.hs+65−5 src/Streamly/Internal/Data/Stream/StreamD.hs+107 src/Streamly/Internal/Prelude.hs+61−3
benchmark/Streamly/Benchmark/Data/Fold.hs+2−1
@@ -10,7 +10,7 @@
1010module Main (main) where
1111
1212import Control.DeepSeq (NFData(..))
13−import Data.Monoid (Last(..))
13+import Data.Monoid (Last(..), Sum(..))
1414
1515import System.Random (randomRIO)
1616import Prelude (IO, Int, Double, String, (>), (<*>), (<$>), (+), ($),
@@ -82,6 +82,7 @@ o_1_space_serial_folds value =
8282 , benchIOSink value "lastN.10" (S.fold (IA.lastN 10))
8383 , benchIOSink value "length" (S.fold FL.length)
8484 , benchIOSink value "sum" (S.fold FL.sum)
85+ , benchIOSink value "sum (foldMap)" (S.fold (FL.foldMap Sum))
8586 , benchIOSink value "product" (S.fold FL.product)
8687 , benchIOSink value "maximumBy" (S.fold (FL.maximumBy compare))
8788 , benchIOSink value "maximum" (S.fold FL.maximum)
benchmark/Streamly/Benchmark/Data/Parser.hs+50−13
@@ -16,10 +16,12 @@ module Main
1616import Control.DeepSeq (NFData(..))
1717import Control.Monad.Catch (MonadCatch)
1818import Data.Foldable (asum)
19+import Data.Monoid (Sum(..))
1920import System.Random (randomRIO)
2021import Prelude hiding (any, all, take, sequence, sequenceA, takeWhile)
2122
2223import qualified Data.Traversable as TR
24+import qualified Data.Foldable as F
2325import qualified Control.Applicative as AP
2426import qualified Streamly as S hiding (runStream)
2527import qualified Streamly.Prelude as S
@@ -148,29 +150,49 @@ lookAhead :: MonadCatch m => Int -> SerialT m Int -> m ()
148150lookAhead value =
149151 IP.parse (PR.lookAhead (PR.takeWhile (<= value) FL.drain) *> pure ())
150152
151−-- quadratic complexity
152153{-# INLINE sequenceA #-}
153154sequenceA :: MonadCatch m => Int -> SerialT m Int -> m Int
154155sequenceA value xs = do
155156 x <- IP.parse (TR.sequenceA (replicate value (PR.satisfy (> 0)))) xs
156157 return $ length x
157158
158−-- quadratic complexity
159+{-# INLINE sequenceA_ #-}
160+sequenceA_ :: MonadCatch m => Int -> SerialT m Int -> m ()
161+sequenceA_ value =
162+ IP.parse (F.sequenceA_ $ replicate value (PR.satisfy (> 0)))
163+
159164{-# INLINE sequence #-}
160165sequence :: MonadCatch m => Int -> SerialT m Int -> m Int
161166sequence value xs = do
162167 x <- IP.parse (TR.sequence (replicate value (PR.satisfy (> 0)))) xs
163168 return $ length x
164169
165−-- choice using the "Alternative" instance with direct style parser type has
166−-- quadratic performance complexity.
167−--
168170{-# INLINE choice #-}
169171choice :: MonadCatch m => Int -> SerialT m Int -> m Int
170172choice value =
171173 IP.parse (asum (replicate value (PR.satisfy (< 0)))
172174 AP.<|> PR.satisfy (> 0))
173175
176+-------------------------------------------------------------------------------
177+-- Stream transformation
178+-------------------------------------------------------------------------------
179+
180+{-# INLINE splitParse #-}
181+splitParse :: MonadCatch m => SerialT m Int -> m ()
182+splitParse =
183+ S.drain
184+ . S.map getSum
185+ . IP.splitParse (PR.take 2 FL.mconcat)
186+ . S.map Sum
187+
188+{-# INLINE concatParse #-}
189+concatParse :: MonadCatch m => SerialT m Int -> m ()
190+concatParse =
191+ S.drain
192+ . S.map getSum
193+ . IP.concatParse (\b -> (PR.take 2 (FL.mconcatTo b))) (Sum 0)
194+ . S.map Sum
195+
174196-------------------------------------------------------------------------------
175197-- Benchmarks
176198-------------------------------------------------------------------------------
@@ -185,21 +207,36 @@ o_1_space_serial_parse value =
185207 , benchIOSink value "splitWith (all,any)" $ splitWithAllAny value
186208 , benchIOSink value "many" many
187209 , benchIOSink value "some" some
188− , benchIOSink value "choice" $ choice value
189210 , benchIOSink value "tee (all,any)" $ teeAllAny value
190211 , benchIOSink value "teeFst (all,any)" $ teeFstAllAny value
191212 , benchIOSink value "shortest (all,any)" $ shortestAllAny value
192213 , benchIOSink value "longest (all,any)" $ longestAllAny value
193− , benchIOSink value "sequenceA" $ sequenceA value
194− , benchIOSink value "sequence" $ sequence value
214+ , benchIOSink value "splitParse" splitParse
215+ , benchIOSink value "concatParse" concatParse
195216 ]
196217
197−o_1_heap_serial_parse :: Int -> [Benchmark]
198−o_1_heap_serial_parse value =
199− [ benchIOSink value "lookAhead" $ lookAhead value
218+o_n_heap_serial_parse :: Int -> [Benchmark]
219+o_n_heap_serial_parse value =
220+ [
221+ -- lookahead benchmark holds the entire input till end
222+ benchIOSink value "lookAhead" $ lookAhead value
223+
224+ -- accumulates the results in a list
225+ , benchIOSink value "sequence" $ sequence value
226+ , benchIOSink value "sequenceA" $ sequenceA value
227+
228+ -- XXX why should this take O(n) heap, it discards the results?
229+ , benchIOSink value "sequenceA_" $ sequenceA_ value
230+ -- XXX why O(n)?
231+ , benchIOSink value "choice" $ choice value
232+
233+ -- XXX this should get better with "yield with backtrack" fix
234+ , benchIOSink value "manyTill" $ manyTill value
235+
236+ -- These show non-linear time complexity.
237+ -- They accumulate the results in a list.
200238 , benchIOSink value "manyAlt" manyAlt
201239 , benchIOSink value "someAlt" someAlt
202− , benchIOSink value "manyTill" $ manyTill value
203240 ]
204241
205242-------------------------------------------------------------------------------
@@ -223,7 +260,7 @@ main = do
223260 , bgroup "o-n-heap"
224261 [ bgroup "parser" $ concat
225262 [
226− o_1_heap_serial_parse value
263+ o_n_heap_serial_parse value
227264 ]
228265 ]
229266 ]
benchmark/Streamly/Benchmark/Data/ParserD.hs+9−4
@@ -192,14 +192,19 @@ o_1_space_serial_parse value =
192192 , benchIOSink value "longest (all,any)" $ longestAllAny value
193193 ]
194194
195−-- These show non-linear time complexity
196−-- Increase in rss is 20-40% on doubling the stream size
197195o_n_heap_serial_parse :: Int -> [Benchmark]
198196o_n_heap_serial_parse value =
199− [ benchIOSink value "lookAhead" $ lookAhead value
197+ [
198+ -- lookahead benchmark holds the entire input till end
199+ benchIOSink value "lookAhead" $ lookAhead value
200+
201+ -- XXX this should get better with "yield with backtrack" fix
202+ , benchIOSink value "manyTill" $ manyTill value
203+
204+ -- These show non-linear time complexity
205+ -- They accumulate the results in a list.
200206 , benchIOSink value "manyAlt" manyAlt
201207 , benchIOSink value "someAlt" someAlt
202− , benchIOSink value "manyTill" $ manyTill value
203208 ]
204209
205210-- accumulate results in a list in IO
benchmark/Streamly/Benchmark/FileIO/Stream.hs+1−1
@@ -88,7 +88,7 @@ import qualified Streamly.Data.Unicode.Stream as SS
8888import qualified Streamly.Internal.Data.Unicode.Stream as IUS
8989import qualified Streamly.Internal.Memory.Unicode.Array as IUA
9090import qualified Streamly.Internal.Data.Unfold as IUF
91−import qualified Streamly.Internal.Data.Parser.ParserD as PR
91+import qualified Streamly.Internal.Data.Parser as PR
9292import qualified Streamly.Internal.Prelude as IP
9393import qualified Streamly.Internal.Data.Stream.StreamD as D
9494
benchmark/streamly-benchmarks.cabal+3−3
@@ -227,7 +227,7 @@ benchmark serial-o-n-space
227227benchmark fold
228228 import: bench-options
229229 type: exitcode-stdio-1.0
230− ghc-options: -rtsopts
230+ ghc-options: -with-rtsopts "-T -K128K -M16M"
231231 hs-source-dirs: Streamly/Benchmark/Data
232232 main-is: Fold.hs
233233 if impl(ghcjs)
@@ -250,7 +250,7 @@ benchmark unfold
250250benchmark parserD
251251 import: bench-options
252252 type: exitcode-stdio-1.0
253− ghc-options: -with-rtsopts "-T -K36K -M16M"
253+ ghc-options: -with-rtsopts "-T -K64K -M16M"
254254 hs-source-dirs: ., Streamly/Benchmark/Data
255255 main-is: ParserD.hs
256256 if impl(ghcjs)
@@ -274,7 +274,7 @@ benchmark parserK
274274benchmark parser
275275 import: bench-options
276276 type: exitcode-stdio-1.0
277− ghc-options: -with-rtsopts "-T -K36K -M32M"
277+ ghc-options: -with-rtsopts "-T -K128K -M32M"
278278 hs-source-dirs: Streamly/Benchmark/Data
279279 main-is: Parser.hs
280280 if impl(ghcjs)
src/Streamly/Data/Fold.hs+6
@@ -22,6 +22,12 @@
2222-- constituent folds and then combines the resulting fold outputs. Similarly,
2323-- a partitioning combinator divides the input among constituent folds.
2424--
25+-- = Monoids
26+--
27+-- Monoids allow modular folding in general. Most of the folds in this module
28+-- can be expressed using 'mconcat' and a suitable 'Monoid'. Instead of writing
29+-- folds we can write Monoids instead and turn them into folds.
30+--
2531-- = Performance Notes
2632--
2733-- 'Fold' representation is more efficient than using streams when splitting
src/Streamly/Internal/Data/Fold.hs+65−5
@@ -60,6 +60,7 @@ module Streamly.Internal.Data.Fold
6060
6161 -- ** Full Folds (Monoidal)
6262 , mconcat
63+ , mconcatTo
6364 , foldMap
6465 , foldMapM
6566
@@ -386,6 +387,8 @@ _Fold1 step = Fold step_ (return Nothing') (return . toMaybe)
386387-- | A fold that drains all its input, running the effects and discarding the
387388-- results.
388389--
390+-- > drain = drainBy (const (return ()))
391+--
389392-- @since 0.7.0
390393{-# INLINABLE drain #-}
391394drain :: Monad m => Fold m a ()
@@ -397,6 +400,7 @@ drain = Fold step begin done
397400
398401-- |
399402-- > drainBy f = lmapM f drain
403+-- > drainBy = FL.foldMapM (void . f)
400404--
401405-- Drain all input after passing it through a monadic function. This is the
402406-- dual of mapM_ on stream producers.
@@ -412,6 +416,8 @@ drainBy2 f = Fold2 (const (void . f)) (\_ -> return ()) return
412416
413417-- | Extract the last element of the input stream, if any.
414418--
419+-- > last = fmap getLast $ FL.foldMap (Last . Just)
420+--
415421-- @since 0.7.0
416422{-# INLINABLE last #-}
417423last :: Monad m => Fold m a (Maybe a)
@@ -423,6 +429,8 @@ last = _Fold1 (flip const)
423429
424430-- | Like 'length', except with a more general 'Num' return value
425431--
432+-- > genericLength = fmap getSum $ foldMap (Sum . const 1)
433+--
426434-- @since 0.7.0
427435{-# INLINABLE genericLength #-}
428436genericLength :: (Monad m, Num b) => Fold m a b
@@ -430,6 +438,8 @@ genericLength = Fold (\n _ -> return $ n + 1) (return 0) return
430438
431439-- | Determine the length of the input stream.
432440--
441+-- > length = fmap getSum $ foldMap (Sum . const 1)
442+--
433443-- @since 0.7.0
434444{-# INLINABLE length #-}
435445length :: Monad m => Fold m a Int
@@ -439,14 +449,18 @@ length = genericLength
439449-- identity (@0@) when the stream is empty. Note that this is not numerically
440450-- stable for floating point numbers.
441451--
452+-- > sum = fmap getSum $ FL.foldMap Sum
453+--
442454-- @since 0.7.0
443−{-# INLINABLE sum #-}
455+{-# INLINE sum #-}
444456sum :: (Monad m, Num a) => Fold m a a
445457sum = Fold (\x a -> return $ x + a) (return 0) return
446458
447459-- | Determine the product of all elements of a stream of numbers. Returns
448460-- multiplicative identity (@1@) when the stream is empty.
449461--
462+-- > product = fmap getProduct $ FL.foldMap Product
463+--
450464-- @since 0.7.0
451465{-# INLINABLE product #-}
452466product :: (Monad m, Num a) => Fold m a a
@@ -475,6 +489,8 @@ maximumBy cmp = _Fold1 max'
475489--
476490-- Determine the maximum element in a stream.
477491--
492+-- Compare with @FL.foldMap Max@.
493+--
478494-- @since 0.7.0
479495{-# INLINABLE maximum #-}
480496maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
@@ -494,6 +510,12 @@ minimumBy cmp = _Fold1 min'
494510-- | Determine the minimum element in a stream using the supplied comparison
495511-- function.
496512--
513+-- @
514+-- minimum = 'minimumBy' compare
515+-- @
516+--
517+-- Compare with @FL.foldMap Min@.
518+--
497519-- @since 0.7.0
498520{-# INLINABLE minimum #-}
499521minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
@@ -591,18 +613,31 @@ rollingHashFirstN n = ltake n rollingHash
591613-- Monoidal left folds
592614------------------------------------------------------------------------------
593615
616+-- | 'mappend' the elements of an input stream to a provided starting value.
617+--
618+-- > S.fold (FL.mconcatTo 10) (S.map Sum $ S.enumerateFromTo 1 10)
619+--
620+-- This could be faster than (fmap (m0 <>) FL.mconcat) especially when we have
621+-- to perform the operation many times.
622+--
623+-- /Internal/
624+--
625+{-# INLINE mconcatTo #-}
626+mconcatTo :: (Monad m, Monoid a) => a -> Fold m a a
627+mconcatTo i = Fold (\x a -> return $ mappend x a) (return i) return
628+
594629-- | Fold an input stream consisting of monoidal elements using 'mappend'
595630-- and 'mempty'.
596631--
597632-- > S.fold FL.mconcat (S.map Sum $ S.enumerateFromTo 1 10)
598633--
599634-- @since 0.7.0
600−{-# INLINABLE mconcat #-}
635+{-# INLINE mconcat #-}
601636mconcat :: (Monad m, Monoid a) => Fold m a a
602−mconcat = Fold (\x a -> return $ mappend x a) (return mempty) return
637+mconcat = mconcatTo mempty
603638
604639-- |
605−-- > foldMap f = map f mconcat
640+-- > foldMap f = lmap f mconcat
606641--
607642-- Make a fold from a pure function that folds the output of the function
608643-- using 'mappend' and 'mempty'.
@@ -615,7 +650,7 @@ foldMap :: (Monad m, Monoid b) => (a -> b) -> Fold m a b
615650foldMap f = lmap f mconcat
616651
617652-- |
618−-- > foldMapM f = mapM f mconcat
653+-- > foldMapM f = lmapM f mconcat
619654--
620655-- Make a fold from a monadic function that folds the output of the function
621656-- using 'mappend' and 'mempty'.
@@ -698,6 +733,8 @@ index = genericIndex
698733
699734-- | Extract the first element of the stream, if any.
700735--
736+-- > head = fmap getFirst $ FL.foldMap (First . Just)
737+--
701738-- @since 0.7.0
702739{-# INLINABLE head #-}
703740head :: Monad m => Fold m a (Maybe a)
@@ -705,6 +742,11 @@ head = _Fold1 const
705742
706743-- | Returns the first element that satisfies the given predicate.
707744--
745+-- @
746+-- find p = fmap getFirst $
747+-- foldMap (\x -> First (if p x then Just x else Nothing))
748+-- @
749+--
708750-- @since 0.7.0
709751{-# INLINABLE find #-}
710752find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
@@ -720,6 +762,8 @@ find predicate = Fold step (return Nothing') (return . toMaybe)
720762-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
721763-- first pair where the key equals the given value @a@.
722764--
765+-- > lookup = snd <$> find ((==) . fst)
766+--
723767-- @since 0.7.0
724768{-# INLINABLE lookup #-}
725769lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
@@ -755,6 +799,8 @@ findIndex predicate = Fold step (return $ Left' 0) (return . hush)
755799
756800-- | Returns the first index where a given value is found in the stream.
757801--
802+-- > elemIndex a = findIndex (== a)
803+--
758804-- @since 0.7.0
759805{-# INLINABLE elemIndex #-}
760806elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
@@ -766,6 +812,8 @@ elemIndex a = findIndex (a ==)
766812
767813-- | Return 'True' if the input stream is empty.
768814--
815+-- > null = fmap isJust head
816+--
769817-- @since 0.7.0
770818{-# INLINABLE null #-}
771819null :: Monad m => Fold m a Bool
@@ -773,6 +821,7 @@ null = Fold (\_ _ -> return False) (return True) return
773821
774822-- |
775823-- > any p = lmap p or
824+-- > any p = fmap getAny . FL.foldMap (Any . p)
776825--
777826-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
778827--
@@ -783,6 +832,8 @@ any predicate = Fold (\x a -> return $ x || predicate a) (return False) return
783832
784833-- | Return 'True' if the given element is present in the stream.
785834--
835+-- > elem a = any (== a)
836+--
786837-- @since 0.7.0
787838{-# INLINABLE elem #-}
788839elem :: (Eq a, Monad m) => a -> Fold m a Bool
@@ -790,6 +841,7 @@ elem a = any (a ==)
790841
791842-- |
792843-- > all p = lmap p and
844+-- > all p = fmap getAll . FL.foldMap (All . p)
793845--
794846-- | Returns 'True' if all elements of a stream satisfy a predicate.
795847--
@@ -800,6 +852,8 @@ all predicate = Fold (\x a -> return $ x && predicate a) (return True) return
800852
801853-- | Returns 'True' if the given element is not present in the stream.
802854--
855+-- > notElem a = all (/= a)
856+--
803857-- @since 0.7.0
804858{-# INLINABLE notElem #-}
805859notElem :: (Eq a, Monad m) => a -> Fold m a Bool
@@ -807,6 +861,9 @@ notElem a = all (a /=)
807861
808862-- | Returns 'True' if all elements are 'True', 'False' otherwise
809863--
864+-- > and = all (== True)
865+-- > and = fmap getAll . FL.foldMap All
866+--
810867-- @since 0.7.0
811868{-# INLINABLE and #-}
812869and :: Monad m => Fold m Bool Bool
@@ -814,6 +871,9 @@ and = Fold (\x a -> return $ x && a) (return True) return
814871
815872-- | Returns 'True' if any element is 'True', 'False' otherwise
816873--
874+-- > or = any (== True)
875+-- > or = fmap getAny . FL.foldMap Any
876+--
817877-- @since 0.7.0
818878{-# INLINABLE or #-}
819879or :: Monad m => Fold m Bool Bool
src/Streamly/Internal/Data/Stream/StreamD.hs+107
@@ -130,6 +130,7 @@ module Streamly.Internal.Data.Stream.StreamD
130130
131131 , parselMx'
132132 , splitParse
133+ , concatParse
133134
134135 -- ** Specialized Folds
135136 , tap
@@ -1121,6 +1122,112 @@ splitParse (PRD.Parser pstep initial extract) (Stream step state) =
11211122
11221123 stepOuter _ (ParseChunksYield a next) = return $ Yield a next
11231124
1125+{-# ANN type ConcatParseState Fuse #-}
1126+data ConcatParseState x inpBuf st p =
1127+ ConcatParseInit inpBuf st p
1128+ | ConcatParseInitLeftOver inpBuf
1129+ | ConcatParseStream st inpBuf p
1130+ | ConcatParseBuf inpBuf st inpBuf p
1131+ | ConcatParseYield x (ConcatParseState x inpBuf st p)
1132+
1133+{-# INLINE_NORMAL concatParse #-}
1134+concatParse
1135+ :: MonadThrow m
1136+ => (b -> PRD.Parser m a b)
1137+ -> b
1138+ -> Stream m a
1139+ -> Stream m b
1140+concatParse func seed (Stream step state) =
1141+ Stream stepOuter (ConcatParseInit [] state (func seed))
1142+
1143+ where
1144+
1145+ {-# INLINE_LATE stepOuter #-}
1146+ -- Buffer is empty, go to stream processing loop
1147+ stepOuter _ (ConcatParseInit [] st (PRD.Parser pstep initial extract)) = do
1148+ initial >>= \r -> return $ Skip $ ConcatParseStream st []
1149+ (PRD.Parser pstep (return r) extract)
1150+
1151+ -- Buffer is not empty, go to buffered processing loop
1152+ stepOuter _ (ConcatParseInit src st
1153+ (PRD.Parser pstep initial extract)) = do
1154+ initial >>= \r -> return $ Skip $ ConcatParseBuf src st []
1155+ (PRD.Parser pstep (return r) extract)
1156+
1157+ -- XXX we just discard any leftover input at the end
1158+ stepOuter _ (ConcatParseInitLeftOver _) = return Stop
1159+
1160+ -- Buffer is empty process elements from the stream
1161+ stepOuter gst (ConcatParseStream st buf
1162+ p@(PRD.Parser pstep initial extract)) = do
1163+ r <- step (adaptState gst) st
1164+ case r of
1165+ Yield x s -> do
1166+ pst <- initial
1167+ pRes <- pstep pst x
1168+ case pRes of
1169+ -- PR.Yield 0 pst1 -> go SPEC s [] pst1
1170+ PR.Yield n pst1 -> do
1171+ assert (n <= length (x:buf)) (return ())
1172+ let buf1 = Prelude.take n (x:buf)
1173+ return $ Skip $ ConcatParseStream s buf1
1174+ (PRD.Parser pstep (return pst1) extract)
1175+ -- PR.Skip 0 pst1 ->
1176+ -- return $ Skip $ ConcatParseStream s (x:buf) pst1
1177+ PR.Skip n pst1 -> do
1178+ assert (n <= length (x:buf)) (return ())
1179+ let (src0, buf1) = splitAt n (x:buf)
1180+ src = Prelude.reverse src0
1181+ return $ Skip $ ConcatParseBuf src s buf1
1182+ (PRD.Parser pstep (return pst1) extract)
1183+ -- XXX Specialize for Stop 0 common case?
1184+ PR.Stop n b -> do
1185+ assert (n <= length (x:buf)) (return ())
1186+ let src = Prelude.reverse (Prelude.take n (x:buf))
1187+ return $ Skip $
1188+ ConcatParseYield b (ConcatParseInit src s (func b))
1189+ PR.Error err -> throwM $ ParseError err
1190+ Skip s -> return $ Skip $ ConcatParseStream s buf p
1191+ Stop -> do
1192+ pst <- initial
1193+ b <- extract pst
1194+ let src = Prelude.reverse buf
1195+ return $ Skip $
1196+ ConcatParseYield b (ConcatParseInitLeftOver src)
1197+
1198+ -- go back to stream processing mode
1199+ stepOuter _ (ConcatParseBuf [] s buf p) =
1200+ return $ Skip $ ConcatParseStream s buf p
1201+
1202+ -- buffered processing loop
1203+ stepOuter _ (ConcatParseBuf (x:xs) s buf
1204+ (PRD.Parser pstep initial extract)) = do
1205+ pst <- initial
1206+ pRes <- pstep pst x
1207+ case pRes of
1208+ -- PR.Yield 0 pst1 ->
1209+ PR.Yield n pst1 -> do
1210+ assert (n <= length (x:buf)) (return ())
1211+ let buf1 = Prelude.take n (x:buf)
1212+ return $ Skip $ ConcatParseBuf xs s buf1
1213+ (PRD.Parser pstep (return pst1) extract)
1214+ -- PR.Skip 0 pst1 -> return $ Skip $ ConcatParseBuf xs s (x:buf) pst1
1215+ PR.Skip n pst1 -> do
1216+ assert (n <= length (x:buf)) (return ())
1217+ let (src0, buf1) = splitAt n (x:buf)
1218+ src = Prelude.reverse src0 ++ xs
1219+ return $ Skip $ ConcatParseBuf src s buf1
1220+ (PRD.Parser pstep (return pst1) extract)
1221+ -- XXX Specialize for Stop 0 common case?
1222+ PR.Stop n b -> do
1223+ assert (n <= length (x:buf)) (return ())
1224+ let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs
1225+ return $ Skip $ ConcatParseYield b
1226+ (ConcatParseInit src s (func b))
1227+ PR.Error err -> throwM $ ParseError err
1228+
1229+ stepOuter _ (ConcatParseYield a next) = return $ Yield a next
1230+
11241231------------------------------------------------------------------------------
11251232-- Specialized Folds
11261233------------------------------------------------------------------------------
src/Streamly/Internal/Prelude.hs+61−3
@@ -270,9 +270,13 @@ module Streamly.Internal.Prelude
270270
271271 -- ** Parsing
272272 , splitParse
273+ , splitParseTill
274+ , concatParse
273275
274276 -- ** Trimming
275277 , take
278+ -- , takeGE
279+ -- , takeBetween
276280 , takeByTime
277281 -- , takeEnd
278282 , takeWhile
@@ -3141,22 +3145,76 @@ concatMapTreeYieldLeavesWith combine f = concatMapLoopWith combine f yield
31413145-- expressed using splitParse. Operations like chunksOf, intervalsOf, split*,
31423146-- can be expressed using splitParse when used with an appropriate Parse.
31433147--
3144−-- | Apply a 'Parse' repeatedly on a stream and emit the parsed values in the
3148+-- The name splitParse is supposed to be opposite of concatMap. Alternative
3149+-- names could be "many", "parseMany" or "splitMany"? If we are going to have a
3150+-- separate combinators for Fold and Parser then having fold/parse in the name
3151+-- allows the distinction.
3152+--
3153+-- XXX We need takeGE/takeBetween to implement "some" using "many".
3154+
3155+-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
31453156-- output stream.
31463157--
3158+-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
3159+-- parse combinator.
3160+--
31473161-- >>> S.toList $ S.splitParse (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]
31483162-- > [3,7,11,15,19]
31493163--
31503164-- >>> S.toList $ S.splitParse (PR.line FL.toList) $ S.fromList "hello\nworld"
31513165-- > ["hello\n","world"]
31523166--
3167+-- /Internal
3168+--
31533169{-# INLINE splitParse #-}
31543170splitParse
31553171 :: (IsStream t, MonadThrow m)
3156− => PRD.Parser m a b
3172+ => Parser m a b
3173+ -> t m a
3174+ -> t m b
3175+splitParse p m =
3176+ D.fromStreamD $ D.splitParse (PRD.fromParserK p) (D.toStreamD m)
3177+
3178+-- | @splitParseTill collect test stream@ tries the parser @test@ on the input,
3179+-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
3180+-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
3181+-- output of @test@ is discarded and the output of @collect@ is emitted in the
3182+-- output stream. The parser fails if @collect@ fails.
3183+--
3184+-- /Unimplemented/
3185+--
3186+{-# INLINE splitParseTill #-}
3187+splitParseTill ::
3188+ -- (IsStream t, MonadThrow m) =>
3189+ Parser m a b
3190+ -> Parser m a x
3191+ -> t m a
3192+ -> t m b
3193+splitParseTill = undefined
3194+
3195+-- Rename to iterateParse?
3196+--
3197+-- | Iterate a parser generating function on a stream. The initial value @b@ is
3198+-- used to generate the first parser, the parser is applied on the stream and
3199+-- the result is used generate the next parser and so on.
3200+--
3201+-- >>> S.toList $ S.concatParse (\b -> PR.take 2 (FL.mconcatTo b)) 0 $ S.fromList [1..10]
3202+-- > [3,10,21,36,55,55]
3203+--
3204+-- This is the streaming equivalent of monad like sequenced application of
3205+-- parsers where next parser is dependent on the previous parser.
3206+--
3207+-- /Internal/
3208+--
3209+{-# INLINE concatParse #-}
3210+concatParse
3211+ :: (IsStream t, MonadThrow m)
3212+ => (b -> Parser m a b)
3213+ -> b
31573214 -> t m a
31583215 -> t m b
3159−splitParse f m = D.fromStreamD $ D.splitParse f (D.toStreamD m)
3216+concatParse f i m = D.fromStreamD $
3217+ D.concatParse (\b -> PRD.fromParserK $ f b) i (D.toStreamD m)
31603218
31613219------------------------------------------------------------------------------
31623220-- Grouping/Splitting