Add a chunked parser implementation

core/src/Streamly/Internal/Data/Parser/Chunked.hs

@@ -0,0 +1,322 @@
+{-# OPTIONS_GHC -Wno-orphans  #-}
+
+-- |
+-- Module      : Streamly.Internal.Data.Parser.Chunked
+-- Copyright   : (c) 2020 Composewell Technologies
+-- License     : BSD-3-Clause
+-- Maintainer  : streamly@composewell.com
+-- Stability   : pre-release
+-- Portability : GHC
+
+module Streamly.Internal.Data.Parser.Chunked
+    (
+      ParserChunked (..)
+    , fromParserD
+    , parseBreak
+    , K.fromPure
+    , K.fromEffect
+    , K.die
+
+    -- We can expose all the parser combinators here, that way we can choose to
+    -- use custom or CPS equivalents where it makes sense.
+    )
+where
+
+#include "ArrayMacros.h"
+#include "assert.hs"
+#include "inline.hs"
+
+import Control.Monad.Catch (MonadThrow, throwM)
+import Control.Monad.IO.Class (MonadIO(..))
+import GHC.Types (SPEC(..))
+import Streamly.Internal.Data.Array.Type (Array(..))
+import Streamly.Internal.Data.Parser.Chunked.Type (ParserChunked (..))
+import Streamly.Internal.Data.Parser.ParserD.Type (Initial(..), Step(..))
+import Streamly.Internal.Data.Stream.Type (Stream)
+import Streamly.Internal.Data.SVar.Type (defState)
+import Streamly.Internal.Data.Unboxed (peekWith, sizeOf, Unbox)
+
+import qualified Streamly.Internal.Data.Array as Array
+import qualified Streamly.Internal.Data.Parser.Chunked.Type as K
+import qualified Streamly.Internal.Data.Parser.ParserD as D
+    hiding (fromParserK, toParserK)
+import qualified Streamly.Internal.Data.Stream.StreamK as StreamK
+import qualified Streamly.Internal.Data.Stream.Type as Stream
+
+-------------------------------------------------------------------------------
+-- Driver
+-------------------------------------------------------------------------------
+
+-- The backracking buffer consists of arrays in the most recent first order. We
+-- want to take a total of n array elements from this buffer. Note: when we
+-- have to take an array partially, we must take the last part of the array.
+{-# INLINE backTrack #-}
+backTrack :: forall m a. Unbox a =>
+       Int
+    -> [Array a]
+    -> StreamK.Stream m (Array a)
+    -> (StreamK.Stream m (Array a), [Array a])
+backTrack = go
+
+    where
+
+    go _ [] stream = (stream, [])
+    go n xs stream | n <= 0 = (stream, xs)
+    go n (x:xs) stream =
+        let len = Array.length x
+        in if n > len
+           then go (n - len) xs (StreamK.cons x stream)
+           else if n == len
+           then (StreamK.cons x stream, xs)
+           else let !(Array contents start end) = x
+                    !start1 = end - (n * SIZE_OF(a))
+                    arr1 = Array contents start1 end
+                    arr2 = Array contents start start1
+                 in (StreamK.cons arr1 stream, arr2:xs)
+
+-- | A continuation to extract the result when a CPS parser is done.
+{-# INLINE parserDone #-}
+parserDone :: Monad m => (Int, Int) -> K.Parse b -> m (K.Step m a b)
+parserDone (0,_) (K.Success n b) = return $ K.Done n b
+parserDone st (K.Success _ _) =
+    error $ "Bug: fromParserK: inside alternative: " ++ show st
+parserDone _ (K.Failure e) = return $ K.Error e
+
+-- | Run a 'Parser' over a stream and return rest of the Stream.
+{-# INLINE_NORMAL parseBreak #-}
+parseBreak
+    :: (MonadThrow m, Unbox a)
+    => ParserChunked a m b
+    -> Stream m (Array a)
+    -> m (b, Stream m (Array a))
+parseBreak parser input = do
+    pRes <- K.runParser parser 0 (0,0) parserDone
+    case pRes of
+        K.Done n b -> assert (n == 0) (return (b, input))
+        K.Error e -> throwM (D.ParseError e)
+        K.Partial n parserk ->
+            assert (n == 0) (go [] parserk (Stream.toStreamK input))
+        K.Continue n parserk ->
+            assert (n == 0) (go [] parserk (Stream.toStreamK input))
+
+    where
+
+    -- This is a simplified yieldk
+    extractYieldK backBuf parserk arr stream = do
+        pRes <- parserk (Just arr)
+        case pRes of
+            K.Partial 0 cont1 ->
+                goExtract [] cont1 stream
+            K.Partial n cont1 -> do
+                assertM(n <= sum (map Array.length (arr:backBuf)))
+                goExtract [] cont1 (fst $ backTrack n (arr:backBuf) stream)
+            K.Continue 0 cont1 ->
+                go (arr:backBuf) cont1 stream
+            K.Continue n cont1 -> do
+                assertM(n <= sum (map Array.length (arr:backBuf)))
+                let (s1, backBuf1) = backTrack n (arr:backBuf) stream
+                 in go backBuf1 cont1 s1
+            K.Done 0 b ->
+                return (b, Stream.fromStreamK stream)
+            K.Done n b -> do
+                assertM(n <= sum (map Array.length (arr:backBuf)))
+                let (s1, _) = backTrack n (arr:backBuf) stream
+                 in return (b, Stream.fromStreamK s1)
+            K.Error err -> throwM $ D.ParseError err
+
+    goExtract backBuf parserk stream = do
+        let stop = goStop backBuf parserk
+            single a = extractYieldK backBuf parserk a StreamK.nil
+         in StreamK.foldStream
+                defState (extractYieldK backBuf parserk) single stop stream
+
+    -- This is a simplified goExtract
+    {-# INLINE goStop #-}
+    goStop backBuf parserk = do
+        pRes <- parserk Nothing
+        case pRes of
+            -- If we stop in an alternative, it will try calling the next
+            -- parser, the next parser may call initial returning Partial and
+            -- then immediately we have to call extract on it.
+            K.Partial 0 cont1 -> do
+                 goExtract [] cont1 StreamK.nil
+            K.Partial n cont1 -> do
+                -- error $ "Bug: parseBreak: Partial in extract, n = " ++ show n
+                assertM(n <= sum (map Array.length backBuf))
+                let (s1, backBuf1) = backTrack n backBuf StreamK.nil
+                 in goExtract backBuf1 cont1 s1
+            K.Continue 0 cont1 -> do
+                -- error "parseBreak: extract, Continue 0 creates infinite loop"
+                go backBuf cont1 StreamK.nil
+            K.Continue n cont1 -> do
+                assertM(n <= sum (map Array.length backBuf))
+                let (s1, backBuf1) = backTrack n backBuf StreamK.nil
+                 in goExtract backBuf1 cont1 s1
+            K.Done 0 b -> return (b, Stream.nil)
+            K.Done n b -> do
+                assertM(n <= sum (map Array.length backBuf))
+                let (s1, _) = backTrack n backBuf StreamK.nil
+                 in return (b, Stream.fromStreamK s1)
+            K.Error err -> throwM $ D.ParseError err
+
+    -- SPECIALIZE this on backBuf?
+    yieldk backBuf !parserk arr stream = do
+        pRes <- parserk (Just arr)
+        case pRes of
+            K.Partial 0 cont1 ->
+                 go [] cont1 stream
+            K.Partial n cont1 -> do
+                assertM(n <= sum (map Array.length (arr:backBuf)))
+                go [] cont1 (fst $ backTrack n (arr:backBuf) stream)
+            K.Continue 0 cont1 ->
+                 go (arr:backBuf) cont1 stream
+            K.Continue n cont1 -> do
+                assertM(n <= sum (map Array.length (arr:backBuf)))
+                let (s1, backBuf1) = backTrack n (arr:backBuf) stream
+                 in go backBuf1 cont1 s1
+            K.Done 0 b ->
+                return (b, Stream.fromStreamK stream)
+            K.Done n b -> do
+                assertM(n <= sum (map Array.length (arr:backBuf)))
+                let (s1, _) = backTrack n (arr:backBuf) stream
+                 in return (b, Stream.fromStreamK s1)
+            K.Error err -> throwM $ D.ParseError err
+
+    go backBuf parserk stream = do
+        let stop = goStop backBuf parserk
+            single a = yieldk backBuf parserk a StreamK.nil
+         in StreamK.foldStream
+                defState (yieldk backBuf parserk) single stop stream
+
+-- This is very similar to fromParserD in the Array/Unboxed/Fold module.
+{-# INLINE parseChunk #-}
+parseChunk
+    :: forall m a s b. (MonadIO m, Unbox a)
+    => (s -> a -> m (Step s b))
+    -> s
+    -> Array a
+    -> m (Step s b)
+parseChunk pstep state (Array contents start end) = do
+    go SPEC start state
+
+    where
+
+    {-# INLINE partial #-}
+    partial arrRem cur next elemSize st n fs1 = do
+        let next1 = next - (n * elemSize)
+        if next1 >= start && cur < end
+        then go SPEC next1 fs1
+        else return $ st (arrRem + n) fs1
+
+    go !_ !cur !pst | cur >= end = return $ Continue 0 pst
+    go !_ !cur !pst = do
+        x <- liftIO $ peekWith contents cur
+        pRes <- pstep pst x
+        let elemSize = SIZE_OF(a)
+            next = INDEX_NEXT(cur,a)
+            arrRem = (end - next) `div` elemSize
+        case pRes of
+            Done n b -> do
+                return $ Done (arrRem + n) b
+            Partial n pst1 ->
+                partial arrRem cur next elemSize Partial n pst1
+            Continue n pst1 -> do
+                partial arrRem cur next elemSize Continue n pst1
+            Error err -> return $ Error err
+
+{-# INLINE_NORMAL parseDToK #-}
+parseDToK
+    :: (MonadIO m, Unbox a)
+    => (s -> a -> m (Step s b))
+    -> m (Initial s b)
+    -> (s -> m (Step s b))
+    -> Int
+    -> (Int, Int)
+    -> ((Int, Int) -> K.Parse b -> m (K.Step m a r))
+    -> m (K.Step m a r)
+-- Non 'Alternative' case.
+parseDToK pstep initial extract leftover (0, _) cont = do
+    res <- initial
+    case res of
+        IPartial r -> return $ K.Partial leftover (parseCont (return r))
+        IDone b -> cont state (K.Success 0 b)
+        IError err -> cont state (K.Failure err)
+
+    where
+
+    -- The continuation is called with (0,0) state i.e. Alternative level
+    -- is 0 and the used count is 0. Alternative level is 0 because the level
+    -- passed in the argument above is 0, the "used" count is 0 because it is
+    -- not useful when Alternative level is 0. We should probably use a Maybe
+    -- type for the state but that might impact performance, need to measure.
+    state = (0,0)
+
+    -- We could pass a stream or array here and drive the ParserD with fusion.
+    parseCont pst (Just arr) = do
+        r <- pst
+        pRes <- parseChunk pstep r arr
+        case pRes of
+            Done n b -> cont state (K.Success n b)
+            Error err -> cont state (K.Failure err)
+            Partial n pst1 -> return $ K.Partial n (parseCont (return pst1))
+            Continue n pst1 -> return $ K.Continue n (parseCont (return pst1))
+
+    parseCont acc Nothing = do
+        pst <- acc
+        r <- extract pst
+        case r of
+            Done n b -> cont state (K.Success n b)
+            Error err -> cont state (K.Failure err)
+            Partial _ _ -> error "Bug: parseDToK Partial unreachable"
+            Continue n pst1 -> return $ K.Continue n (parseCont (return pst1))
+
+-- 'Alternative' case. Used count needs to be maintained when inside an
+-- Alternative.
+parseDToK pstep initial extract leftover (level, count) cont = do
+    res <- initial
+    case res of
+        IPartial r -> return $ K.Partial leftover (parseCont count (return r))
+        IDone b -> cont (level,count) (K.Success 0 b)
+        IError err -> cont (level,count) (K.Failure err)
+
+    where
+
+    parseCont !cnt pst (Just arr) = do
+        let !cnt1 = cnt + Array.length arr
+        r <- pst
+        pRes <- parseChunk pstep r arr
+        case pRes of
+            Done n b -> do
+                assertM(n <= cnt1)
+                cont (level, cnt1 - n) (K.Success n b)
+            Error err ->
+                cont (level, cnt1) (K.Failure err)
+            Partial n pst1 -> do
+                assertM(n <= cnt1)
+                return $ K.Partial n (parseCont (cnt1 - n) (return pst1))
+            Continue n pst1 -> do
+                assertM(n <= cnt1)
+                return $ K.Continue n (parseCont (cnt1 - n) (return pst1))
+    parseCont cnt acc Nothing = do
+        pst <- acc
+        r <- extract pst
+        let s = (level, cnt)
+        case r of
+            Done n b -> do
+                assertM(n <= cnt)
+                cont s (K.Success n b)
+            Partial _ _ -> error "Bug: parseDToK Partial unreachable"
+            Error e ->
+                cont s (K.Failure e)
+            Continue n pst1 -> do
+                assertM(n <= cnt)
+                return $ K.Continue n (parseCont (cnt - n) (return pst1))
+
+-- | Convert a raw byte 'Parser' to a chunked parser.
+--
+-- /Pre-release/
+--
+{-# INLINE_LATE fromParserD #-}
+fromParserD :: (MonadIO m, Unbox a) => D.Parser a m b -> ParserChunked a m b
+fromParserD (D.Parser step initial extract) =
+    K.MkParser $ parseDToK step initial extract

core/src/Streamly/Internal/Data/Parser/Chunked/Type.hs

@@ -0,0 +1,332 @@
+{-# LANGUAGE UndecidableInstances #-}
+#include "inline.hs"
+
+-- |
+-- Module      : Streamly.Internal.Data.Parser.Chunked.Type
+-- Copyright   : (c) 2020 Composewell Technologies
+-- License     : BSD-3-Clause
+-- Maintainer  : streamly@composewell.com
+-- Stability   : experimental
+-- Portability : GHC
+--
+-- CPS style implementation of parsers.
+--
+-- The CPS representation allows linear performance for Applicative, sequenceA,
+-- Monad, sequence, and Alternative, choice operations compared to the
+-- quadratic complexity of the corresponding direct style operations. However,
+-- direct style operations allow fusion with ~10x better performance than CPS.
+--
+-- The direct style representation does not allow for recursive definitions of
+-- "some" and "many" whereas CPS allows that.
+
+-- XXX This code has only one line difference from the
+-- Data.Parser.ParserK/Type.
+
+module Streamly.Internal.Data.Parser.Chunked.Type
+    (
+      Step (..)
+    , Parse (..)
+    , ParserChunked (..)
+    , fromPure
+    , fromEffect
+    , die
+    )
+where
+
+import Control.Applicative (Alternative(..), liftA2)
+import Control.Monad (MonadPlus(..), ap)
+import Control.Monad.Catch (MonadCatch, MonadThrow(..))
+import Control.Monad.IO.Class (MonadIO, liftIO)
+import Control.Monad.Reader.Class (MonadReader, ask, local)
+import Control.Monad.State.Class (MonadState, get, put)
+import Streamly.Internal.Data.Array (Array)
+import qualified Control.Monad.Fail as Fail
+
+-- | The intermediate result of running a parser step. The parser driver may
+-- stop with a final result, pause with a continuation to resume, or fail with
+-- an error.
+--
+-- See ParserD docs. This is the same as the ParserD Step except that it uses a
+-- continuation in Partial and Continue constructors instead of a state in case
+-- of ParserD.
+--
+-- /Pre-release/
+--
+data Step m a r =
+      Done !Int r
+      -- XXX we can use a "resume" and a "stop" continuations instead of Maybe.
+      -- measure if that works any better.
+      -- XXX The Array is the only difference from element parser, we can pass
+      -- this as parameter?
+    | Partial !Int (Maybe (Array a) -> m (Step m a r))
+    | Continue !Int (Maybe (Array a) -> m (Step m a r))
+    | Error String
+
+instance Functor m => Functor (Step m a) where
+    fmap f (Done n r) = Done n (f r)
+    fmap f (Partial n k) = Partial n (fmap (fmap f) . k)
+    fmap f (Continue n k) = Continue n (fmap (fmap f) . k)
+    fmap _ (Error e) = Error e
+
+-- | The parser's result.
+--
+-- /Pre-release/
+--
+data Parse b =
+      Success !Int !b     -- Leftover count, result
+    | Failure !String     -- Error
+
+-- | Map a function over 'Success'.
+instance Functor Parse where
+    fmap f (Success n b) = Success n (f b)
+    fmap _ (Failure e) = Failure e
+
+-- | A continuation passing style parser representation. A continuation of
+-- 'Step's, each step passes a state and a parse result to the next 'Step'. The
+-- resulting 'Step' may carry a continuation that consumes input 'a' and
+-- results in another 'Step'. Essentially, the continuation may either consume
+-- input without a result or return a result with no further input to be
+-- consumed.
+--
+newtype ParserChunked a m b = MkParser
+    { runParser :: forall r.
+           -- leftover: the number of elements that were not used by the
+           -- previous consumer and should be carried forward.
+           Int
+           -- (alt nesting level, alt used elem count). Nesting level is
+           -- increased whenever we enter an Alternative composition and
+           -- decreased when it is done. The used element count is a count of
+           -- elements consumed by the Alternative. If the Alternative fails we
+           -- need to backtrack by this amount.
+           --
+           -- The nesting level is used in parseDToK to optimize the case when
+           -- we are not in an alternative, in that case we do not need to
+           -- maintain the element count for backtracking.
+        -> (Int, Int)
+           -- The first argument is the (nest level, used count) tuple as
+           -- described above. The leftover element count is carried as part of
+           -- 'Success' constructor of 'Parse'.
+        -> ((Int, Int) -> Parse b -> m (Step m a r))
+        -> m (Step m a r)
+    }
+
+-------------------------------------------------------------------------------
+-- Functor
+-------------------------------------------------------------------------------
+
+-- XXX rewrite this using ParserD, expose rmapM from ParserD.
+-- | Maps a function over the output of the parser.
+--
+instance Functor m => Functor (ParserChunked a m) where
+    {-# INLINE fmap #-}
+    fmap f parser = MkParser $ \n st k ->
+        let k1 s res = k s (fmap f res)
+         in runParser parser n st k1
+
+-------------------------------------------------------------------------------
+-- Sequential applicative
+-------------------------------------------------------------------------------
+
+-- This is the dual of stream "fromPure".
+--
+-- | A parser that always yields a pure value without consuming any input.
+--
+-- /Pre-release/
+--
+{-# INLINE fromPure #-}
+fromPure :: b -> ParserChunked a m b
+fromPure b = MkParser $ \n st k -> k st (Success n b)
+
+-- | See 'Streamly.Internal.Data.Parser.fromEffect'.
+--
+-- /Pre-release/
+--
+{-# INLINE fromEffect #-}
+fromEffect :: Monad m => m b -> ParserChunked a m b
+fromEffect eff = MkParser $ \n st k -> eff >>= \b -> k st (Success n b)
+
+-- | 'Applicative' form of 'Streamly.Internal.Data.Parser.serialWith'. Note that
+-- this operation does not fuse, use 'Streamly.Internal.Data.Parser.serialWith'
+-- when fusion is important.
+--
+instance Monad m => Applicative (ParserChunked a m) where
+    {-# INLINE pure #-}
+    pure = fromPure
+
+    {-# INLINE (<*>) #-}
+    (<*>) = ap
+
+    {-# INLINE (*>) #-}
+    p1 *> p2 = MkParser $ \n st k ->
+        let k1 s (Success n1 _) = runParser p2 n1 s k
+            k1 s (Failure e) = k s (Failure e)
+        in runParser p1 n st k1
+
+    {-# INLINE (<*) #-}
+    p1 <* p2 = MkParser $ \n st k ->
+        let k1 s1 (Success n1 b) =
+                let k2 s2 (Success n2 _) = k s2 (Success n2 b)
+                    k2 s2 (Failure e) = k s2 (Failure e)
+                in runParser p2 n1 s1 k2
+            k1 s1 (Failure e) = k s1 (Failure e)
+        in runParser p1 n st k1
+
+    {-# INLINE liftA2 #-}
+    liftA2 f p = (<*>) (fmap f p)
+
+-------------------------------------------------------------------------------
+-- Monad
+-------------------------------------------------------------------------------
+
+-- This is the dual of "nil".
+--
+-- | A parser that always fails with an error message without consuming
+-- any input.
+--
+-- /Pre-release/
+--
+{-# INLINE die #-}
+die :: String -> ParserChunked a m b
+die err = MkParser (\_ st k -> k st (Failure err))
+
+-- | Monad composition can be used for lookbehind parsers, we can make the
+-- future parses depend on the previously parsed values.
+--
+-- If we have to parse "a9" or "9a" but not "99" or "aa" we can use the
+-- following parser:
+--
+-- @
+-- backtracking :: MonadCatch m => PR.Parser Char m String
+-- backtracking =
+--     sequence [PR.satisfy isDigit, PR.satisfy isAlpha]
+--     '<|>'
+--     sequence [PR.satisfy isAlpha, PR.satisfy isDigit]
+-- @
+--
+-- We know that if the first parse resulted in a digit at the first place then
+-- the second parse is going to fail.  However, we waste that information and
+-- parse the first character again in the second parse only to know that it is
+-- not an alphabetic char.  By using lookbehind in a 'Monad' composition we can
+-- avoid redundant work:
+--
+-- @
+-- data DigitOrAlpha = Digit Char | Alpha Char
+--
+-- lookbehind :: MonadCatch m => PR.Parser Char m String
+-- lookbehind = do
+--     x1 \<-    Digit '<$>' PR.satisfy isDigit
+--          '<|>' Alpha '<$>' PR.satisfy isAlpha
+--
+--     -- Note: the parse depends on what we parsed already
+--     x2 <- case x1 of
+--         Digit _ -> PR.satisfy isAlpha
+--         Alpha _ -> PR.satisfy isDigit
+--
+--     return $ case x1 of
+--         Digit x -> [x,x2]
+--         Alpha x -> [x,x2]
+-- @
+--
+-- See also 'Streamly.Internal.Data.Parser.concatMap'. This monad instance
+-- does not fuse, use 'Streamly.Internal.Data.Parser.concatMap' when you need
+-- fusion.
+--
+instance Monad m => Monad (ParserChunked a m) where
+    {-# INLINE return #-}
+    return = pure
+
+    {-# INLINE (>>=) #-}
+    p >>= f = MkParser $ \n st k ->
+        let k1 s1 (Success n1 b) = runParser (f b) n1 s1 k
+            k1 s1 (Failure e) = k s1 (Failure e)
+         in runParser p n st k1
+
+    {-# INLINE (>>) #-}
+    (>>) = (*>)
+
+#if !(MIN_VERSION_base(4,13,0))
+    -- This is redefined instead of just being Fail.fail to be
+    -- compatible with base 4.8.
+    {-# INLINE fail #-}
+    fail = die
+#endif
+instance Monad m => Fail.MonadFail (ParserChunked a m) where
+    {-# INLINE fail #-}
+    fail = die
+
+instance (MonadThrow m, MonadReader r m, MonadCatch m) =>
+    MonadReader r (ParserChunked a m) where
+
+    {-# INLINE ask #-}
+    ask = fromEffect ask
+
+    {-# INLINE local #-}
+    local f p = MkParser $ \n st k -> local f $ runParser p n st k
+
+instance (MonadThrow m, MonadState s m) => MonadState s (ParserChunked a m) where
+    {-# INLINE get #-}
+    get = fromEffect get
+
+    {-# INLINE put #-}
+    put = fromEffect . put
+
+instance (MonadThrow m, MonadIO m) => MonadIO (ParserChunked a m) where
+    {-# INLINE liftIO #-}
+    liftIO = fromEffect . liftIO
+
+-------------------------------------------------------------------------------
+-- Alternative
+-------------------------------------------------------------------------------
+
+-- | 'Alternative' form of 'Streamly.Internal.Data.Parser.alt'. Backtrack and
+-- run the second parser if the first one fails.
+--
+-- The "some" and "many" operations of alternative accumulate results in a pure
+-- list which is not scalable and streaming. Instead use
+-- 'Streamly.Internal.Data.Parser.some' and
+-- 'Streamly.Internal.Data.Parser.many' for fusible operations with composable
+-- accumulation of results.
+--
+-- See also 'Streamly.Internal.Data.Parser.alt'. This 'Alternative' instance
+-- does not fuse, use 'Streamly.Internal.Data.Parser.alt' when you need
+-- fusion.
+--
+instance Monad m => Alternative (ParserChunked a m) where
+    {-# INLINE empty #-}
+    empty = die "empty"
+
+    {-# INLINE (<|>) #-}
+    p1 <|> p2 = MkParser $ \n (level, _) k ->
+        let k1 (0, _) _ = error "Bug: 0 nest level in Alternative"
+            k1 (l1, n1) (Failure _) = runParser p2 n1 (l1 - 1, 0) k
+            k1 (l1, _) success = k (l1 - 1, 0) success
+        in runParser p1 n (level + 1, 0) k1
+
+    -- some and many are implemented here instead of using default definitions
+    -- so that we can use INLINE on them. It gives 50% performance improvement.
+
+    {-# INLINE many #-}
+    many v = many_v
+
+        where
+
+        many_v = some_v <|> pure []
+        some_v = (:) <$> v <*> many_v
+
+    {-# INLINE some #-}
+    some v = some_v
+
+        where
+
+        many_v = some_v <|> pure []
+        some_v = (:) <$> v <*> many_v
+
+-- | 'mzero' is same as 'empty', it aborts the parser. 'mplus' is same as
+-- '<|>', it selects the first succeeding parser.
+--
+instance Monad m => MonadPlus (ParserChunked a m) where
+    {-# INLINE mzero #-}
+    mzero = die "mzero"
+
+    {-# INLINE mplus #-}
+    mplus = (<|>)

core/streamly-core.cabal

@@ -301,6 +301,8 @@ library
                      , Streamly.Internal.Data.Fold.Chunked
                      , Streamly.Internal.Data.Fold.Window
                      , Streamly.Internal.Data.Parser
+                     , Streamly.Internal.Data.Parser.Chunked.Type
+                     , Streamly.Internal.Data.Parser.Chunked
                      , Streamly.Internal.Data.Pipe
                      , Streamly.Internal.Data.Stream.Type
                      , Streamly.Internal.Data.Stream.Eliminate

hie.yaml

@@ -96,6 +96,8 @@ cradle:
               component: "test:Data.Stream.Concurrent"
             - path: "./test/Streamly/Test/Data/Stream/Common.hs"
               component: "test:Data.Stream.Concurrent"
+            - path: "./test/Streamly/Test/Data/Parser/Chunked.hs"
+              component: "test:Data.Parser.Chunked"
             - path: "./test/Streamly/Test/Data/Parser/ParserD.hs"
               component: "test:Data.Parser.ParserD"
             - path: "./test/Streamly/Test/Data/Unfold.hs"

streamly.cabal

@@ -98,6 +98,7 @@ extra-source-files:
     test/Streamly/Test/Data/Ring/Unboxed.hs
     test/Streamly/Test/Data/Array/Stream.hs
     test/Streamly/Test/Data/Parser/ParserD.hs
+    test/Streamly/Test/Data/Parser/Chunked.hs
     test/Streamly/Test/Data/Stream/Concurrent.hs
     test/Streamly/Test/FileSystem/Event.hs
     test/Streamly/Test/FileSystem/Event/Common.hs

test/streamly-tests.cabal

@@ -285,6 +285,13 @@ test-suite Data.Parser.ParserD
   if flag(limit-build-mem)
     ghc-options: +RTS -M1500M -RTS
 
+test-suite Data.Parser.Chunked
+  import: test-options
+  type: exitcode-stdio-1.0
+  main-is: Streamly/Test/Data/Parser/Chunked.hs
+  if flag(limit-build-mem)
+    ghc-options: +RTS -M1500M -RTS
+
 test-suite Data.SmallArray
   import: test-options
   type: exitcode-stdio-1.0