Remove the rewrite rule based Parser wrapper layer

Now we have explicit "Parser" and "ParserK" types for direct style and
CPS parsers, respectively. The programmer can use either type depending
on the use case.

benchmark/Streamly/Benchmark/Data/Parser.hs

@@ -18,7 +18,6 @@ module Main
   ) where
 
 import Control.DeepSeq (NFData(..))
-import Data.Foldable (asum)
 import Data.Functor (($>))
 import Data.Monoid (Sum(..))
 import GHC.Magic (inline)
@@ -30,8 +29,6 @@ import Streamly.Internal.Data.Stream.StreamD (Stream)
 import Prelude hiding
     (any, all, take, sequence, sequence_, sequenceA, takeWhile, dropWhile)
 
-import qualified Data.Traversable as TR
-import qualified Data.Foldable as F
 import qualified Control.Applicative as AP
 import qualified Streamly.FileSystem.Handle as Handle
 import qualified Streamly.Internal.Data.Array as Array
@@ -113,18 +110,6 @@ benchIOSink value name f =
 -- Parsers
 -------------------------------------------------------------------------------
 
-{-# INLINE one #-}
-one :: Monad m => Int -> Stream m Int -> m (Either ParseError (Maybe Int))
-one value = Stream.parse p
-
-    where
-
-    p = do
-        m <- PR.fromFold Fold.one
-        case m of
-          Just i -> if i >= value then pure m else p
-          Nothing -> pure Nothing
-
 {-# INLINE takeBetween #-}
 takeBetween :: Monad m => Int -> Stream m a -> m (Either ParseError ())
 takeBetween value = Stream.parse (PR.takeBetween 0 value Fold.drain)
@@ -187,14 +172,20 @@ wordBy :: Monad m => Int -> Stream m Int -> m (Either ParseError ())
 wordBy value = Stream.parse (PR.wordBy (>= value) Fold.drain)
 
 {-# INLINE sepByWords #-}
-sepByWords :: Monad m => Int -> Stream m Int -> m (Either ParseError ())
-sepByWords _ = Stream.parse (wrds even Fold.drain)
+sepByWords :: Monad m => Stream m Int -> m (Either ParseError ())
+sepByWords = Stream.parse (wrds even Fold.drain)
     where
     wrds p = PR.sepBy (PR.takeWhile (not . p) Fold.drain) (PR.dropWhile p)
 
+-- Returning a list to compare with the sepBy1 in ParserK
+{-# INLINE sepBy1 #-}
+sepBy1 :: Monad m => Stream m Int -> m (Either ParseError [Int])
+sepBy1 xs = do
+    Stream.parse (PR.sepBy1 (PR.satisfy odd) (PR.satisfy even) Fold.toList) xs
+
 {-# INLINE sepByWords1 #-}
-sepByWords1 :: Monad m => Int -> Stream m Int -> m (Either ParseError ())
-sepByWords1 _ = Stream.parse (wrds even Fold.drain)
+sepByWords1 :: Monad m => Stream m Int -> m (Either ParseError ())
+sepByWords1 = Stream.parse (wrds even Fold.drain)
     where
     wrds p = PR.sepBy1 (PR.takeWhile (not . p) Fold.drain) (PR.dropWhile p)
 
@@ -368,34 +359,6 @@ lookAhead :: Monad m => Int -> Stream m Int -> m (Either ParseError ())
 lookAhead value =
     Stream.parse (PR.lookAhead (PR.takeWhile (<= value) Fold.drain) $> ())
 
-{-# INLINE sequenceA #-}
-sequenceA :: Monad m => Int -> Stream m Int -> m Int
-sequenceA value xs = do
-    x <- Stream.parse (TR.sequenceA (replicate value (PR.satisfy (> 0)))) xs
-    return $ length x
-
-{-# INLINE sequenceA_ #-}
-sequenceA_ :: Monad m => Int -> Stream m Int -> m (Either ParseError ())
-sequenceA_ value =
-    Stream.parse (F.sequenceA_ $ replicate value (PR.satisfy (> 0)))
-
-{-# INLINE sequence #-}
-sequence :: Monad m => Int -> Stream m Int -> m Int
-sequence value xs = do
-    x <- Stream.parse (TR.sequence (replicate value (PR.satisfy (> 0)))) xs
-    return $ length x
-
-{-# INLINE sequence_ #-}
-sequence_ :: Monad m => Int -> Stream m Int -> m (Either ParseError ())
-sequence_ value =
-    Stream.parse (F.sequence_ $ replicate value (PR.satisfy (> 0)))
-
-{-# INLINE choiceAsum #-}
-choiceAsum :: Monad m => Int -> Stream m Int -> m (Either ParseError Int)
-choiceAsum value =
-    Stream.parse (asum (replicate value (PR.satisfy (< 0)))
-        AP.<|> PR.satisfy (> 0))
-
 {-
 {-# INLINE choice #-}
 choice :: Monad m => Int -> Stream m Int -> m (Either ParseError Int)
@@ -429,7 +392,8 @@ parseIterate n =
 
 {-# INLINE concatSequence #-}
 concatSequence :: Monad m => Stream m Int -> m (Either ParseError ())
-concatSequence = Stream.parse $ PR.concatSequence Fold.drain $ Stream.repeat PR.one
+concatSequence =
+    Stream.parse $ PR.sequence (Stream.repeat PR.one) Fold.drain
 
 {-# INLINE parseManyGroupBy #-}
 parseManyGroupBy :: Monad m => (Int -> Int -> Bool) -> Stream m Int -> m ()
@@ -449,8 +413,7 @@ instance NFData ParseError where
 
 o_1_space_serial :: Int -> [Benchmark]
 o_1_space_serial value =
-    [ benchIOSink value "one (fold)" $ one value
-    , benchIOSink value "takeBetween" $ takeBetween value
+    [ benchIOSink value "takeBetween" $ takeBetween value
     , benchIOSink value "takeEQ" $ takeEQ value
     , benchIOSink value "takeWhile" $ takeWhile value
     , benchIOSink value "takeWhileP" $ takeWhileP value
@@ -462,8 +425,9 @@ o_1_space_serial value =
     , benchIOSink value "groupBy" $ groupBy
     , benchIOSink value "groupByRolling" $ groupByRolling
     , benchIOSink value "wordBy" $ wordBy value
-    , benchIOSink value "sepBy (words)" $ sepByWords value
-    , benchIOSink value "sepBy1 (words)" $ sepByWords1 value
+    , benchIOSink value "sepBy (words)" sepByWords
+    , benchIOSink value "sepBy1" sepBy1
+    , benchIOSink value "sepBy1 (words)" sepByWords1
     , benchIOSink value "deintercalate" $ deintercalate value
     , benchIOSink value "splitAp" $ splitAp value
     , benchIOSink value "splitApBefore" $ splitApBefore value
@@ -513,17 +477,9 @@ o_n_heap_serial value =
     -- lookahead benchmark holds the entire input till end
       benchIOSink value "lookAhead" $ lookAhead value
 
-    -- accumulates the results in a list
-    , benchIOSink value "sequence" $ sequence value
-    , benchIOSink value "sequenceA" $ sequenceA value
-
-    -- XXX why should this take O(n) heap, it discards the results?
-    , benchIOSink value "sequence_" $ sequence_ value
-    , benchIOSink value "sequenceA_" $ sequenceA_ value
     -- non-linear time complexity (parserD)
     , benchIOSink value "split_" $ split_ value
     -- XXX why O(n) heap?
-    , benchIOSink value "choice (asum)" $ choiceAsum value
     -- , benchIOSink value "choice" $ choice value
 
     -- These show non-linear time complexity.

benchmark/Streamly/Benchmark/Data/Parser/ParserK.hs

@@ -19,11 +19,13 @@ import Data.Foldable (asum)
 import Streamly.Internal.Data.Parser (ParseError(..))
 import Streamly.Internal.Data.Stream.StreamD (Stream)
 import System.Random (randomRIO)
-import Prelude hiding (any, all, take, sequence, sequenceA, takeWhile)
+import Prelude hiding
+    (any, all, take, sequence, sequence_, sequenceA, takeWhile)
 
 import qualified Control.Applicative as AP
 import qualified Data.Foldable as F
 import qualified Data.Traversable as TR
+import qualified Streamly.Data.Parser.ParserK as ParserK
 import qualified Streamly.Internal.Data.Fold as FL
 import qualified Streamly.Internal.Data.Parser.ParserK.Type as PR
 import qualified Streamly.Internal.Data.Parser.ParserD as PRD
@@ -59,11 +61,19 @@ benchIOSink value name f =
 -- Parsers
 -------------------------------------------------------------------------------
 
-#ifdef FROM_PARSERK
-#define PARSE_OP (Stream.parseD . PRD.fromParserK)
-#else
-#define PARSE_OP Stream.parse
-#endif
+#define PARSE_OP (Stream.parse . PRD.fromParserK)
+
+{-# INLINE one #-}
+one :: Monad m => Int -> Stream m Int -> m (Either ParseError (Maybe Int))
+one value = Stream.parse (ParserK.toParser p)
+
+    where
+
+    p = do
+        m <- ParserK.fromFold FL.one
+        case m of
+          Just i -> if i >= value then pure m else p
+          Nothing -> pure Nothing
 
 {-# INLINE satisfy #-}
 satisfy :: Monad m => (a -> Bool) -> PR.Parser a m a
@@ -106,6 +116,13 @@ sequence value xs = do
     x <- PARSE_OP (TR.sequence list) xs
     return $ Prelude.length x
 
+{-# INLINE sequence_ #-}
+sequence_ :: Monad m => Int -> Stream m Int -> m (Either ParseError ())
+sequence_ value =
+    let parser = satisfy (> 0)
+        list = Prelude.replicate value parser
+     in PARSE_OP (F.sequence_ list)
+
 {-# INLINE manyAlt #-}
 manyAlt :: Monad m => Stream m Int -> m Int
 manyAlt xs = do
@@ -141,17 +158,40 @@ o_1_space_serial value =
     , benchIOSink value "splitApp" $ splitApp value
     ]
 
+{-# INLINE sepBy1 #-}
+sepBy1 :: Monad m => Stream m Int -> m Int
+sepBy1 xs = do
+    x <- PARSE_OP (parser (satisfy odd) (satisfy even)) xs
+    return $ Prelude.length x
+
+    where
+
+    parser p sep = do
+        x <- p
+        fmap (x :) $ AP.many (sep >> p)
+
 -- O(n) heap beacuse of accumulation of the list in strict IO monad?
 o_n_heap_serial :: Int -> [Benchmark]
 o_n_heap_serial value =
-    [ benchIOSink value "sequenceA" $ sequenceA value
+    [
+    -- accumulates the results in a list
+    -- XXX why should this take O(n) heap, it discards the results?
+      benchIOSink value "sequence_" $ sequence_ value
     , benchIOSink value "sequenceA_" $ sequenceA_ value
     , benchIOSink value "sequence" $ sequence value
+    , benchIOSink value "sequenceA" $ sequenceA value
     , benchIOSink value "manyAlt" manyAlt
+    , benchIOSink value "sepBy1" sepBy1
     , benchIOSink value "someAlt" someAlt
     , benchIOSink value "choice" $ choice value
     ]
 
+-- O(n) heap beacuse of accumulation of the list in strict IO monad?
+o_1_space_recursive :: Int -> [Benchmark]
+o_1_space_recursive value =
+    [ benchIOSink value "one (recursive)" $ one value
+    ]
+
 -------------------------------------------------------------------------------
 -- Driver
 -------------------------------------------------------------------------------
@@ -163,5 +203,6 @@ main = runWithCLIOpts defaultStreamSize allBenchmarks
 
     allBenchmarks value =
         [ bgroup (o_1_space_prefix moduleName) (o_1_space_serial value)
+        , bgroup (o_1_space_prefix moduleName) (o_1_space_recursive value)
         , bgroup (o_n_heap_prefix moduleName) (o_n_heap_serial value)
         ]

core/src/Streamly/Data/Parser.hs

@@ -17,6 +17,37 @@
 -- (which should be the case almost always) you can just use 'fromEffect' to
 -- execute the lower layer monad effects.
 --
+-- == Performance Notes
+--
+-- This module is designed for fusion, inline the operations in this module for
+-- fusion to occur, avoid using these operations in recursive calls, avoid
+-- operations like 'sequence', 'asum' on these parsers. If you need these then
+-- use the 'ParserK' module instead.
+--
+-- The 'Parser' type represents a stream consumer by composing state as data
+-- which enables stream fusion. Stream fusion generates a tight loop without
+-- any constructor allocations between the stages, providing C like performance
+-- for the loop. Stream fusion works when multiple functions are combined in a
+-- pipeline statically. Therefore, the operations in this module must be
+-- inlined and must not be used recursively to allow for stream fusion.
+--
+-- Using the 'Parser' type parsing operations like 'one', 'splitWith' etc.
+-- degrade quadratically (O(n^2)) when combined many times. If you need to
+-- combine these operations, say more than 50 times in a single loop, then you
+-- should use the continuation style parser type 'ParserK' instead. Also, if
+-- you need to use these operations in a recursive loop you should use
+-- 'ParserK' instead.
+--
+-- The 'ParserK' type represents a stream consumer by composing function calls,
+-- therefore, a function call overhead is incurred at each composition. It is
+-- quite fast in general but may be a few times slower than a fused parser.
+-- However, it allows for scalable dynamic composition especially parsers can
+-- be used in recursive calls. Using the 'ParserK' type operations like
+-- 'splitWith' provide linear (O(n)) performance with respect to the number of
+-- compositions..
+--
+-- 'Parser' and 'ParserK' types can be interconverted.
+--
 module Streamly.Data.Parser
     (
     -- * Parser Type

core/src/Streamly/Data/Parser/ParserK.hs

@@ -0,0 +1,54 @@
+-- |
+-- Module      : Streamly.Data.Parser.ParserK
+-- Copyright   : (c) 2023 Composewell Technologies
+-- License     : BSD-3-Clause
+-- Maintainer  : streamly@composewell.com
+-- Stability   : pre-release
+-- Portability : GHC
+--
+-- Parsers using Continuation Passing Style (CPS). See notes in
+-- "Streamly.Data.Parser" module to know when to use this module.
+--
+-- To run a 'ParserK' convert it to a 'Parser' and then run it.
+--
+module Streamly.Data.Parser.ParserK
+    (
+    -- * Parser Type
+      ParserK
+    , Parser
+
+    -- * Parsers
+    -- ** Conversions
+    , fromFold
+    , fromParser
+    , toParser
+
+    -- ** Without Input
+    , fromPure
+    , fromEffect
+    , die
+    )
+
+where
+
+import Streamly.Internal.Data.Fold (Fold)
+import Streamly.Internal.Data.Parser.ParserK.Type
+import qualified Streamly.Internal.Data.Parser.ParserD as ParserD
+
+-- | Convert a 'Fold' to a 'ParserK'.
+--
+{-# INLINE fromFold #-}
+fromFold :: Monad m => Fold m a b -> ParserK a m b
+fromFold = ParserD.toParserK . ParserD.fromFold
+
+-- | Convert a 'Parser' to a 'ParserK'.
+--
+{-# INLINE fromParser #-}
+fromParser :: Monad m => ParserD.Parser a m b -> ParserK a m b
+fromParser = ParserD.toParserK
+
+-- | Convert a 'ParserK' to a 'Parser'.
+--
+{-# INLINE toParser #-}
+toParser :: Monad m => ParserK a m b -> ParserD.Parser a m b
+toParser = ParserD.fromParserK

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

@@ -182,7 +182,7 @@ fromParserD (ParserD.Parser step1 initial1 extract1) =
 {-# INLINE fromParser #-}
 fromParser :: forall m a b. (MonadIO m, Unbox a) =>
     Parser.Parser a m b -> ChunkFold m a b
-fromParser = fromParserD . ParserD.fromParserK
+fromParser = fromParserD
 
 -- | Adapt an array stream fold.
 --

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

@@ -57,6 +57,8 @@ data Step a m r =
       -- Array a -> m (Step a m r), m (Step a m r)
       -- XXX The Array is the only difference from element parser, we can pass
       -- this as parameter?
+      -- XXX Unify element and chunked parser, by using the argument as
+      -- None | Single a | Chunk (Array a).
     | Partial !Int (Maybe (Array a) -> m (Step a m r))
     | Continue !Int (Maybe (Array a) -> m (Step a m r))
     | Error !Int String

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

@@ -214,6 +214,7 @@ import Fusion.Plugin.Types (Fuse(..))
 import Streamly.Internal.Data.Fold.Type (Fold(..), toList)
 import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))
 
+import qualified Control.Monad.Fail as Fail
 import qualified Streamly.Internal.Data.Fold.Type as FL
 import qualified Streamly.Internal.Data.Parser.ParserK.Type as K
 
@@ -224,6 +225,7 @@ import Prelude hiding (concatMap, filter)
 -- >>> import Control.Applicative ((<|>))
 -- >>> import Data.Bifunctor (second)
 -- >>> import Prelude hiding (concatMap)
+-- >>> import qualified Streamly.Data.Fold as Fold
 -- >>> import qualified Streamly.Data.Stream as Stream
 -- >>> import qualified Streamly.Internal.Data.Stream as Stream (parse)
 -- >>> import qualified Streamly.Internal.Data.Parser as Parser
@@ -625,9 +627,9 @@ fromParserK parser = Parser step initial extract
 -- Mapping on the output
 ------------------------------------------------------------------------------
 
--- | Map a monadic function on the output of a parser.
+-- | @rmapM f parser@ maps the monadic function @f@ on the output of the parser.
 --
--- /Pre-release/
+-- >>> rmap = fmap
 {-# INLINE rmapM #-}
 rmapM :: Monad m => (b -> m c) -> Parser a m b -> Parser a m c
 rmapM f (Parser step initial extract) =
@@ -644,17 +646,14 @@ rmapM f (Parser step initial extract) =
             IError err -> return $ IError err
     step1 s a = step s a >>= mapMStep f
 
--- | See 'Streamly.Internal.Data.Parser.fromPure'.
---
--- /Pre-release/
+-- | A parser that always yields a pure value without consuming any input.
 --
 {-# INLINE_NORMAL fromPure #-}
 fromPure :: Monad m => b -> Parser a m b
 fromPure b = Parser undefined (pure $ IDone b) undefined
 
--- | See 'Streamly.Internal.Data.Parser.fromEffect'.
---
--- /Pre-release/
+-- | A parser that always yields the result of an effectful action without
+-- consuming any input.
 --
 {-# INLINE fromEffect #-}
 fromEffect :: Monad m => m b -> Parser a m b
@@ -667,14 +666,52 @@ fromEffect b = Parser undefined (IDone <$> b) undefined
 {-# ANN type SeqParseState Fuse #-}
 data SeqParseState sl f sr = SeqParseL sl | SeqParseR f sr
 
--- | See 'Streamly.Internal.Data.Parser.splitWith'.
---
 -- Note: this implementation of splitWith is fast because of stream fusion but
 -- has quadratic time complexity, because each composition adds a new branch
 -- that each subsequent parse's input element has to go through, therefore, it
 -- cannot scale to a large number of compositions. After around 100
 -- compositions the performance starts dipping rapidly beyond a CPS style
 -- unfused implementation.
+
+-- | Sequential parser application. Apply two parsers sequentially to an input
+-- stream.  The input is provided to the first parser, when it is done the
+-- remaining input is provided to the second parser. If both the parsers
+-- succeed their outputs are combined using the supplied function. The
+-- operation fails if any of the parsers fail.
+--
+-- Note: This is a parsing dual of appending streams using
+-- 'Streamly.Data.Stream.append', it splits the streams using two parsers and zips
+-- the results.
+--
+-- This implementation is strict in the second argument, therefore, the
+-- following will fail:
+--
+-- >>> Stream.parse (Parser.splitWith const (Parser.satisfy (> 0)) undefined) $ Stream.fromList [1]
+-- *** Exception: Prelude.undefined
+-- ...
+--
+-- Compare with 'Applicative' instance method '<*>'. This implementation allows
+-- stream fusion but has quadratic complexity. This can fuse with other
+-- operations and can be faster than 'Applicative' instance for small number
+-- (less than 8) of compositions.
+--
+-- Many combinators can be expressed using @splitWith@ and other parser
+-- primitives. Some common idioms are described below,
+--
+-- @
+-- span :: (a -> Bool) -> Fold m a b -> Fold m a b -> Parser a m b
+-- span pred f1 f2 = splitWith (,) ('takeWhile' pred f1) ('fromFold' f2)
+-- @
+--
+-- @
+-- spanBy :: (a -> a -> Bool) -> Fold m a b -> Fold m a b -> Parser a m b
+-- spanBy eq f1 f2 = splitWith (,) ('groupBy' eq f1) ('fromFold' f2)
+-- @
+--
+-- @
+-- spanByRolling :: (a -> a -> Bool) -> Fold m a b -> Fold m a b -> Parser a m b
+-- spanByRolling eq f1 f2 = splitWith (,) ('groupByRolling' eq f1) ('fromFold' f2)
+-- @
 --
 -- /Pre-release/
 --
@@ -824,7 +861,24 @@ noErrorUnsafeSplitWith func (Parser stepL initialL extractL)
 data SeqAState sl sr = SeqAL sl | SeqAR sr
 
 -- This turns out to be slightly faster than splitWith
--- | See 'Streamly.Internal.Data.Parser.split_'.
+
+-- | Sequential parser application ignoring the output of the first parser.
+-- Apply two parsers sequentially to an input stream.  The input is provided to
+-- the first parser, when it is done the remaining input is provided to the
+-- second parser. The output of the parser is the output of the second parser.
+-- The operation fails if any of the parsers fail.
+--
+-- This implementation is strict in the second argument, therefore, the
+-- following will fail:
+--
+-- >>> Stream.parse (Parser.split_ (Parser.satisfy (> 0)) undefined) $ Stream.fromList [1]
+-- *** Exception: Prelude.undefined
+-- ...
+--
+-- Compare with 'Applicative' instance method '*>'. This implementation allows
+-- stream fusion but has quadratic complexity. This can fuse with other
+-- operations, and can be faster than 'Applicative' instance for small
+-- number (less than 8) of compositions.
 --
 -- /Pre-release/
 --
@@ -962,7 +1016,20 @@ data AltParseState sl sr = AltParseL Int sl | AltParseR sr
 -- each subsequent alternative's input element has to go through, therefore, it
 -- cannot scale to a large number of compositions
 
--- | See 'Streamly.Internal.Data.Parser.alt'.
+-- | Sequential alternative. Apply the input to the first parser and return the
+-- result if the parser succeeds. If the first parser fails then backtrack and
+-- apply the same input to the second parser and return the result.
+--
+-- Note: This implementation is not lazy in the second argument. The following
+-- will fail:
+--
+-- >> Stream.parse (Parser.satisfy (> 0) `Parser.alt` undefined) $ Stream.fromList [1..10]
+-- *** Exception: Prelude.undefined
+--
+-- Compare with ParserK 'Alternative' instance method '<|>'. This
+-- implementation allows stream fusion but has quadratic complexity. This can
+-- fuse with other operations and can be much faster than 'Alternative'
+-- instance for small number (less than 8) of alternatives.
 --
 -- /Time Complexity:/ O(n^2) where n is the number of compositions.
 --
@@ -1263,15 +1330,15 @@ splitSome (Parser step1 initial1 extract1) (Fold fstep finitial fextract) =
                 assertM(n == cnt)
                 return (Continue n (Tuple3' s1 0 (Right fs)))
 
--- | See 'Streamly.Internal.Data.Parser.die'.
---
--- /Pre-release/
+-- | A parser that always fails with an error message without consuming
+-- any input.
 --
 {-# INLINE_NORMAL die #-}
 die :: Monad m => String -> Parser a m b
 die err = Parser undefined (pure (IError err)) undefined
 
--- | See 'Streamly.Internal.Data.Parser.dieM'.
+-- | A parser that always fails with an effectful error message and without
+-- consuming any input.
 --
 -- /Pre-release/
 --
@@ -1292,7 +1359,7 @@ dieM err = Parser undefined (IError <$> err) undefined
 -- Note: The implementation of '<|>' is not lazy in the second
 -- argument. The following code will fail:
 --
--- >>> Stream.parse (ParserD.toParserK $ ParserD.satisfy (> 0) <|> undefined) $ Stream.fromList [1..10]
+-- >>> Stream.parse (ParserD.satisfy (> 0) <|> undefined) $ Stream.fromList [1..10]
 -- *** Exception: Prelude.undefined
 -- ...
 --
@@ -1314,7 +1381,7 @@ data ConcatParseState sl m a b =
       ConcatParseL sl
     | forall s. ConcatParseR (s -> a -> m (Step s b)) s (s -> m (Step s b))
 
--- | See 'Streamly.Internal.Data.Parser.concatMap'.
+-- | Map a 'Parser' returning function on the result of a 'Parser'.
 --
 -- /Pre-release/
 --
@@ -1469,6 +1536,10 @@ instance Monad m => Monad (Parser a m) where
     {-# INLINE (>>) #-}
     (>>) = (*>)
 
+instance Monad m => Fail.MonadFail (Parser a m) where
+    {-# INLINE fail #-}
+    fail = die
+
 -- | See documentation of 'Streamly.Internal.Data.Parser.ParserK.Type.Parser'.
 --
 instance Monad m => MonadPlus (Parser a m) where
@@ -1486,6 +1557,13 @@ instance (Monad m, MonadIO m) => MonadIO (Parser a m) where
 -- Mapping on input
 ------------------------------------------------------------------------------
 
+-- | @lmap f parser@ maps the function @f@ on the input of the parser.
+--
+-- >>> Stream.parse (Parser.lmap (\x -> x * x) (Parser.fromFold Fold.sum)) (Stream.enumerateFromTo 1 100)
+-- Right 338350
+--
+-- > lmap = Parser.lmapM return
+--
 {-# INLINE lmap #-}
 lmap :: (a -> b) -> Parser b m r -> Parser a m r
 lmap f (Parser step begin done) = Parser step1 begin done
@@ -1494,6 +1572,8 @@ lmap f (Parser step begin done) = Parser step1 begin done
 
     step1 x a = step x (f a)
 
+-- | @lmapM f parser@ maps the monadic function @f@ on the input of the parser.
+--
 {-# INLINE lmapM #-}
 lmapM :: Monad m => (a -> m b) -> Parser b m r -> Parser a m r
 lmapM f (Parser step begin done) = Parser step1 begin done
@@ -1502,6 +1582,11 @@ lmapM f (Parser step begin done) = Parser step1 begin done
 
     step1 x a = f a >>= step x
 
+-- | Include only those elements that pass a predicate.
+--
+-- >>> Stream.parse (Parser.filter (> 5) (Parser.fromFold Fold.sum)) $ Stream.fromList [1..10]
+-- Right 40
+--
 {-# INLINE filter #-}
 filter :: Monad m => (a -> Bool) -> Parser a m b -> Parser a m b
 filter f (Parser step initial extract) = Parser step1 initial extract

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

@@ -20,7 +20,8 @@ module Streamly.Internal.Data.Parser.ParserK.Type
     (
       Step (..)
     , Parse (..)
-    , Parser (..)
+    , Parser (..)  -- XXX Stop exporting this
+    , ParserK
     , fromPure
     , fromEffect
     , die
@@ -101,6 +102,8 @@ newtype Parser a m b = MkParser
         -> m (Step m a r)
     }
 
+type ParserK = Parser
+
 -------------------------------------------------------------------------------
 -- Functor
 -------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/Producer/Source.hs

@@ -282,7 +282,7 @@ parseManyD parser reader = Producer step return return
 -- /Pre-release/
 {-# INLINE parseMany #-}
 parseMany :: Monad m =>
-       ParserK.Parser a m b
+       ParserD.Parser a m b
     -> Producer m (Source x a) a
     -> Producer m (Source x a) (Either ParseError b)
-parseMany parser = parseManyD (ParserD.fromParserK parser)
+parseMany = parseManyD

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

@@ -96,7 +96,7 @@ import qualified Streamly.Internal.Data.Array.Mut.Stream as AS
 import qualified Streamly.Internal.Data.Fold.Type as FL (Fold(..), Step(..))
 import qualified Streamly.Internal.Data.Parser as PR
 import qualified Streamly.Internal.Data.Parser.ParserD as PRD
-    (Parser(..), Initial(..), fromParserK)
+    (Parser(..), Initial(..))
 import qualified Streamly.Internal.Data.Stream.StreamD as D
 import qualified Streamly.Internal.Data.Stream.StreamK as K
 
@@ -789,7 +789,7 @@ parseBreak ::
 parseBreak p s =
     fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)
 -}
-parseBreak p = parseBreakK (PRD.fromParserK p)
+parseBreak = parseBreakK
 
 -------------------------------------------------------------------------------
 -- Elimination - Running Array Folds and parsers

core/src/Streamly/Internal/Data/Stream/StreamD/Eliminate.hs

@@ -181,7 +181,7 @@ parseD parser strm = do
 --
 {-# INLINE [3] parse #-}
 parse :: Monad m => PR.Parser a m b -> Stream m a -> m (Either ParseError b)
-parse = parseD . PRD.fromParserK
+parse = parseD
 
 -- | Run a 'Parse' over a stream and return rest of the Stream.
 {-# INLINE_NORMAL parseBreakD #-}
@@ -347,7 +347,7 @@ parseBreakD (PRD.Parser pstep initial extract) stream@(Stream step state) = do
 --
 {-# INLINE parseBreak #-}
 parseBreak :: Monad m => PR.Parser a m b -> Stream m a -> m (Either ParseError b, Stream m a)
-parseBreak p = parseBreakD (PRD.fromParserK p)
+parseBreak = parseBreakD
 
 ------------------------------------------------------------------------------
 -- Specialized Folds

core/src/Streamly/Internal/Data/Stream/StreamD/Nesting.hs

@@ -1620,7 +1620,7 @@ parseMany
     => PR.Parser a m b
     -> Stream m a
     -> Stream m (Either ParseError b)
-parseMany p = parseManyD (PRD.fromParserK p)
+parseMany = parseManyD
 
 -- | Apply a stream of parsers to an input stream and emit the results in the
 -- output stream.
@@ -1890,7 +1890,7 @@ parseIterate
     -> b
     -> Stream m a
     -> Stream m (Either ParseError b)
-parseIterate f = parseIterateD (PRD.fromParserK . f)
+parseIterate = parseIterateD
 
 ------------------------------------------------------------------------------
 -- Grouping

core/src/Streamly/Internal/Data/Stream/StreamK.hs

@@ -1231,7 +1231,7 @@ parseBreakD (PR.Parser pstep initial extract) stream = do
 {-# INLINE parseBreak #-}
 parseBreak :: Monad m =>
     Parser.Parser a m b -> Stream m a -> m (Either ParseError b, Stream m a)
-parseBreak p = parseBreakD (PR.fromParserK p)
+parseBreak = parseBreakD
 
 {-# INLINE parse #-}
 parse :: Monad m =>

core/src/Streamly/Internal/Serialize/FromBytes.hs

@@ -54,7 +54,7 @@ import qualified Streamly.Internal.Data.Array as A
 import qualified Streamly.Internal.Data.Parser as PR
     (fromPure, either, satisfy, takeEQ)
 import qualified Streamly.Internal.Data.Parser.ParserD as PRD
-    (Parser(..), Initial(..), Step(..), toParserK)
+    (Parser(..), Initial(..), Step(..))
 
 -- Note: The () type does not need to have an on-disk representation in theory.
 -- But we use a concrete representation for it so that we count how many ()
@@ -155,7 +155,7 @@ word16beD = PRD.Parser step initial extract
 --
 {-# INLINE word16be #-}
 word16be :: Monad m => Parser Word8 m Word16
-word16be = PRD.toParserK word16beD
+word16be = word16beD
 
 -- | Little endian (LSB first) Word16
 {-# INLINE word16leD #-}
@@ -180,7 +180,7 @@ word16leD = PRD.Parser step initial extract
 --
 {-# INLINE word16le #-}
 word16le :: Monad m => Parser Word8 m Word16
-word16le = PRD.toParserK word16leD
+word16le = word16leD
 
 -- | Big endian (MSB first) Word32
 {-# INLINE word32beD #-}
@@ -207,7 +207,7 @@ word32beD = PRD.Parser step initial extract
 --
 {-# INLINE word32be #-}
 word32be :: Monad m => Parser Word8 m Word32
-word32be = PRD.toParserK word32beD
+word32be = word32beD
 
 -- | Little endian (LSB first) Word32
 {-# INLINE word32leD #-}
@@ -233,7 +233,7 @@ word32leD = PRD.Parser step initial extract
 --
 {-# INLINE word32le #-}
 word32le :: Monad m => Parser Word8 m Word32
-word32le = PRD.toParserK word32leD
+word32le = word32leD
 
 -- | Big endian (MSB first) Word64
 {-# INLINE word64beD #-}
@@ -260,7 +260,7 @@ word64beD = PRD.Parser step initial extract
 --
 {-# INLINE word64be #-}
 word64be :: Monad m => Parser Word8 m Word64
-word64be = PRD.toParserK word64beD
+word64be = word64beD
 
 -- | Little endian (LSB first) Word64
 {-# INLINE word64leD #-}
@@ -286,7 +286,7 @@ word64leD = PRD.Parser step initial extract
 --
 {-# INLINE word64le #-}
 word64le :: Monad m => Parser Word8 m Word64
-word64le = PRD.toParserK word64leD
+word64le = word64leD
 
 {-# INLINE int8 #-}
 int8 :: Monad m => Parser Word8 m Int8

core/src/Streamly/Internal/Unicode/Stream.hs

@@ -540,7 +540,7 @@ writeCharUtf8' =  ParserD.toFold (parseCharUtf8WithD ErrorOnCodingFailure)
 {-# INLINE parseCharUtf8With #-}
 parseCharUtf8With ::
        Monad m => CodingFailureMode -> Parser.Parser Word8 m Char
-parseCharUtf8With = ParserD.toParserK . parseCharUtf8WithD
+parseCharUtf8With = parseCharUtf8WithD
 
 -- XXX write it as a parser and use parseMany to decode a stream, need to check
 -- if that preserves the same performance. Or we can use a resumable parser

core/streamly-core.cabal

@@ -390,6 +390,7 @@ library
                      , Streamly.Data.Array.Mut
                      , Streamly.Data.Fold
                      , Streamly.Data.Parser
+                     , Streamly.Data.Parser.ParserK
                      , Streamly.Data.Stream
                      , Streamly.Data.Stream.StreamK
                      , Streamly.Data.Unfold
@@ -419,6 +420,7 @@ library
                       , Streamly.Internal.Data.Stream.Cross
                       , Streamly.Internal.Data.List
                       , Streamly.Data.Stream.Zip
+                      , Streamly.Internal.Data.Parser.ParserDK
 
     build-depends:
                     -- streamly-base

src/Streamly/Internal/Data/Stream/IsStream/Eliminate.hs

@@ -382,7 +382,7 @@ parseD p = D.parseD p . toStreamD
 -- /Internal/
 {-# INLINE parseK #-}
 parseK :: Monad m => PRK.Parser a m b -> SerialT m a -> m (Either PRD.ParseError b)
-parseK = parse
+parseK p = parse (PRD.fromParserK p)
 
 -- | Parse a stream using the supplied 'Parser'.
 --
@@ -404,7 +404,7 @@ parseK = parse
 --
 {-# INLINE [3] parse #-}
 parse :: Monad m => Parser a m b -> SerialT m a -> m (Either PRD.ParseError b)
-parse = parseD . PRD.fromParserK
+parse = parseD
 
 ------------------------------------------------------------------------------
 -- Specific Fold Functions

src/Streamly/Internal/Data/Stream/IsStream/Reduce.hs

@@ -189,8 +189,7 @@ import qualified Streamly.Internal.Data.Array.Type as A
 import qualified Streamly.Internal.Data.Fold as FL
     (Fold, Step(..), takeEndBy_, takeEndBy, catMaybes, take)
 import qualified Streamly.Internal.Data.IsMap as IsMap
-import qualified Streamly.Internal.Data.Parser.ParserD as PRD
-    (Parser(..), fromParserK)
+import qualified Streamly.Internal.Data.Parser.ParserD as PRD (Parser(..))
 import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
 import qualified Streamly.Internal.Data.Stream.StreamD as D
     ( foldMany
@@ -388,7 +387,7 @@ parseMany
     -> t m a
     -> t m (Either ParseError b)
 parseMany p m =
-    fromStreamD $ D.parseManyD (PRD.fromParserK p) (toStreamD m)
+    fromStreamD $ D.parseManyD p (toStreamD m)
 
 -- | Same as parseMany but for StreamD streams.
 --
@@ -456,7 +455,7 @@ parseIterate
     -> t m a
     -> t m (Either ParseError b)
 parseIterate f i m = fromStreamD $
-    D.parseIterateD (PRD.fromParserK . f) i (toStreamD m)
+    D.parseIterateD f i (toStreamD m)
 
 ------------------------------------------------------------------------------
 -- Grouping

test/Streamly/Test/Data/Parser.hs

@@ -13,6 +13,7 @@ import Test.QuickCheck.Monadic (monadicIO, assert, run)
 
 import Prelude hiding (sequence)
 
+import qualified Control.Monad.Fail as Fail
 import qualified Data.List as List
 import qualified Prelude
 import qualified Streamly.Internal.Data.Array as A
@@ -97,7 +98,7 @@ dieM =
 parserFail :: Property
 parserFail =
     property $
-        case runIdentity $ S.parse (fail err) (S.fromList [0 :: Int]) of
+        case runIdentity $ S.parse (Fail.fail err) (S.fromList [0 :: Int]) of
             Right _ -> False
             Left (ParseError e) -> err == e
     where
@@ -255,7 +256,7 @@ takeGE =
             let
                 list_length = Prelude.length ls
             in
-                case runIdentity $S.parse (P.takeGE n FL.toList) (S.fromList ls) of
+                case runIdentity $ S.parse (P.takeGE n FL.toList) (S.fromList ls) of
                     Right parsed_list ->
                         if n <= list_length
                         then checkListEqual parsed_list ls