e81d72f · streamly · git.iko.soy

core/src/Streamly/Internal/Data/Array/Unboxed/Stream.hs+6−5

compare with master

unifiedsplit

@@ -744,8 +744,8 @@ parseBreakK (PRD.Parser pstep initial extract) stream = do

744 pRes <- extract pst

745 case pRes of

746 PR.Partial _ _ -> error "Bug: parseBreak: Partial in extract"

747− PR.Continue 0 _ ->

747+ PR.Continue 0 s ->

748− error "parseBreak: extract, Continue 0 creates infinite loop"

748+ goStop s backBuf

749 PR.Continue n s -> do

750 assert (n <= Prelude.length backBuf) (return ())

751 let (src0, buf1) = splitAt n backBuf

@@ -897,8 +897,8 @@ runArrayParserDBreak

897 pRes <- extract pst

898 case pRes of

899 PR.Partial _ _ -> error "Bug: runArrayParserDBreak: Partial in extract"

900− PR.Continue 0 _ ->

900+ PR.Continue 0 pst1 ->

901− error "runArrayParserDBreak: extract, Continue 0 creates infinite loop"

901+ goStop backBuf pst1

902 PR.Continue n pst1 -> do

903 assert

904 (n <= sum (map Array.length (getList backBuf)))

@@ -1123,7 +1123,8 @@ runArrayFoldManyD

1123 pRes <- extract pst

1124 case pRes of

1125 PR.Partial _ _ -> error "runArrayFoldManyD: Partial in extract"

1126− PR.Continue 0 _ -> error "runArrayFoldManyD: Continue 0 in extract"

1126+ PR.Continue 0 pst1 ->

1127+ return $ D.Skip $ ParseChunksStop backBuf pst1

1127 PR.Continue n pst1 -> do

1128 PR.Continue n pst1 -> do

1128 assert (n <= sum (map Array.length backBuf)) (return ())

1129 assert (n <= sum (map Array.length backBuf)) (return ())

1129 let (src0, buf1) = splitAtArrayListRev n backBuf

1130 let (src0, buf1) = splitAtArrayListRev n backBuf

core/src/Streamly/Internal/Data/Parser/ParserD/Type.hs+12−4

@@ -510,6 +510,14 @@ parseDToK pstep initial extract leftover (0, _) cont = do

510 Partial _ _ -> error "Bug: parseDToK Partial unreachable"

511 Continue n pst1 -> return $ K.Continue n (parseCont (return pst1))

512

513+-- XXX Pass on the stream to parserK and inside parserK drive the parserD

514+-- directly from the stream. That way we can feed full chunks efficiently to

515+-- parserD with fusion. The remaining stream can be passed on to the

516+-- continuation and at the end the leftover stream can be returned.

517+--

518+-- contification of parserD completely breaks fusion. We can avoid that by

519+-- fusing the parserD with the source stream.

520+

513-- 'Alternative' case. Used count needs to be maintained when inside an

521-- 'Alternative' case. Used count needs to be maintained when inside an

514-- Alternative.

522-- Alternative.

515parseDToK pstep initial extract leftover (level, count) cont = do

523parseDToK pstep initial extract leftover (level, count) cont = do

@@ -569,9 +577,9 @@ toParserK (Parser step initial extract) =

569{-# INLINE parserDone #-}

577{-# INLINE parserDone #-}

570parserDone :: Monad m => (Int, Int) -> K.Parse b -> m (K.Step m a b)

578parserDone :: Monad m => (Int, Int) -> K.Parse b -> m (K.Step m a b)

571parserDone (0,_) (K.Success n b) = return $ K.Done n b

579parserDone (0,_) (K.Success n b) = return $ K.Done n b

572−parserDone st (K.Success _ _) =

580+parserDone (0,_) (K.Failure e) = return $ K.Error e

581+parserDone st _ =

573− error $ "Bug: fromParserK: inside alternative: " ++ show st

582+ error $ "Bug: fromParserK: parser done inside alternative: " ++ show st

574−parserDone _ (K.Failure e) = return $ K.Error e

575

583

576-- | Convert a CPS style 'K.Parser' to a direct style 'Parser'.

584-- | Convert a CPS style 'K.Parser' to a direct style 'Parser'.

577--

585--

@@ -1025,7 +1033,7 @@ alt (Parser stepL initialL extractL) (Parser stepR initialR extractR) =

1025 IPartial rR -> Continue cnt (AltParseR rR)

1033 IPartial rR -> Continue cnt (AltParseR rR)

1026 IDone b -> Done cnt b

1034 IDone b -> Done cnt b

1027 IError err -> Error err

1035 IError err -> Error err

1028− Partial _ _ -> error "Bug: serialWith extract 'Partial'"

1036+ Partial _ _ -> error "Bug: alt: extractL 'Partial'"

1029 Continue n s -> do

1037 Continue n s -> do

1030 assertM(n == cnt)

1038 assertM(n == cnt)

1031 return $ Continue n (AltParseL 0 s)

1039 return $ Continue n (AltParseL 0 s)

core/src/Streamly/Internal/Data/Parser/ParserK/Type.hs+7−7

@@ -84,14 +84,14 @@ instance Functor Parse where

84--

85newtype Parser m a b = MkParser

86 { runParser :: forall r.

87− -- The number of elements that were not used by the previous

87+ -- leftover: the number of elements that were not used by the

88− -- consumer and should be carried forward.

88+ -- previous consumer and should be carried forward.

89 Int

90− -- (nesting level, used elem count). Nesting level is increased

90+ -- (alt nesting level, alt used elem count). Nesting level is

91− -- whenever we enter an Alternative composition and decreased when

91+ -- increased whenever we enter an Alternative composition and

92− -- it is done. The used element count is a count of elements

92+ -- decreased when it is done. The used element count is a count of

93− -- consumed by the Alternative. If the Alternative fails we need to

93+ -- elements consumed by the Alternative. If the Alternative fails we

94− -- backtrack by this amount.

94+ -- need to backtrack by this amount.

95 --

96 -- The nesting level is used in parseDToK to optimize the case when

97 -- we are not in an alternative, in that case we do not need to

core/src/Streamly/Internal/Data/Producer/Source.hs+2−2

@@ -221,8 +221,8 @@ parseD

221 pRes <- extract pst

222 case pRes of

223 Partial _ _ -> error "Bug: parseD: Partial in extract"

224− Continue 0 _ ->

224+ Continue 0 pst1 ->

225− error "parseD: extract, Continue 0 creates infinite loop"

225+ goStop buf pst1

226 Continue n pst1 -> do

227 assert (n <= length (getList buf)) (return ())

228 let (src0, buf1) = splitAt n (getList buf)

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

@@ -221,6 +221,8 @@ foldlT f z s = D.foldlT f z (toStreamD s)

221parseD :: MonadThrow m => PRD.Parser m a b -> Stream m a -> m b

222parseD p = D.parse p . toStreamD

223

224+-- XXX Drive directly as parserK rather than converting to parserD first.

225+

224-- | Parse a stream using the supplied ParserK 'PRK.Parser'.

226-- | Parse a stream using the supplied ParserK 'PRK.Parser'.

225--

227--

226-- /Internal/

228-- /Internal/

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

@@ -253,13 +253,13 @@ parseBreak (PRD.Parser pstep initial extract) stream@(Stream step state) = do

253 PR.Error err -> throwM $ ParseError err

254

255 -- This is simplified goExtract

256+ -- XXX Use SPEC?

256 {-# INLINE goStop #-}

257 {-# INLINE goStop #-}

257 goStop buf pst = do

258 goStop buf pst = do

258 pRes <- extract pst

259 pRes <- extract pst

259 case pRes of

260 case pRes of

260 PR.Partial _ _ -> error "Bug: parseBreak: Partial in extract"

261 PR.Partial _ _ -> error "Bug: parseBreak: Partial in extract"

261− PR.Continue 0 _ ->

262+ PR.Continue 0 pst1 -> goStop buf pst1

262− error "parseBreak: extract, Continue 0 creates infinite loop"

263 PR.Continue n pst1 -> do

264 assert (n <= length (getList buf)) (return ())

265 let (src0, buf1) = splitAt n (getList buf)

core/src/Streamly/Internal/Data/Stream/StreamD/Nesting.hs+4−4

@@ -1300,8 +1300,8 @@ parseMany (PRD.Parser pstep initial extract) (Stream step state) =

1300 pRes <- extract pst

1301 case pRes of

1302 PR.Partial _ _ -> error "Bug: parseMany: Partial in extract"

1303− PR.Continue 0 _ ->

1303+ PR.Continue 0 pst1 ->

1304− error "parseMany: extract, Continue 0 creates infinite loop"

1304+ return $ Skip $ ParseChunksStop buf pst1

1305 PR.Continue n pst1 -> do

1306 assert (n <= length buf) (return ())

1307 let (src0, buf1) = splitAt n buf

@@ -1480,8 +1480,8 @@ parseIterate func seed (Stream step state) =

1480 pRes <- extract pst

1481 case pRes of

1482 PR.Partial _ _ -> error "Bug: parseIterate: Partial in extract"

1483− PR.Continue 0 _ ->

1483+ PR.Continue 0 pst1 ->

1484− error "parseIterate: extract, Continue 0 creates infinite loop"

1484+ return $ Skip $ ConcatParseStop buf pstep pst1 extract

1485 PR.Continue n pst1 -> do

1486 assert (n <= length buf) (return ())

1487 let (src0, buf1) = splitAt n buf

core/src/Streamly/Internal/Data/Stream/StreamK.hs+1−1

@@ -1054,7 +1054,7 @@ parseBreak (PR.Parser pstep initial extract) stream = do

1054 src = Prelude.reverse src0

1055 return (b, fromList src)

1056 PR.Partial _ _ -> error "Bug: parseBreak: Partial in extract"

1057− PR.Continue 0 _ -> error "parseBreak: extract, Continue 0 creates infinite loop"

1057+ PR.Continue 0 s -> goStream nil buf s

1058 PR.Continue n s -> do

1059 assertM(n <= length buf)

1060 let (src0, buf1) = splitAt n buf