3095

pending completion

Build # 3095

Build Type

Pull #654

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/composewell/streamly/commit/<a class=hub.com/composewell/streamly/commit/ebac72fdf867efbaa3ce9059ab6ce96c4e7cf0de">ebac72fdf<a href="https://github.com/composewell/streamly/commit/ebac72fdf867efbaa3ce9059ab6ce96c4e7cf0de">">Merge </a><a class="double-link" href="https://github.com/composewell/streamly/commit/<a class="double-link" href="https://github.com/composewell/streamly/commit/1507d73ba7c64d032235196492d4354b268654d2">1507d73ba</a>">1507d73ba</a><a href="https://github.com/composewell/streamly/commit/ebac72fdf867efbaa3ce9059ab6ce96c4e7cf0de"> into bf3780f65">bf3780f65</a>

Pull Request Pull Request #654: Pass the failed Stream as an argument to ghandle, gbracket

Run Details

8 of 8 new or added lines in 1 file covered. (100.0%)

3350 of 9323 relevant lines covered (35.93%)

0.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

28.36

/src/Streamly/Internal/Data/Parser/ParserD/Tee.hs

{-# OPTIONS_GHC -Wno-incomplete-uni-patterns #-}

#include "inline.hs"

-- |
-- Module      : Streamly.Internal.Data.Parser.ParserD.Tee
-- Copyright   : (c) 2020 Composewell Technologies
-- License     : BSD-3-Clause
-- Maintainer  : streamly@composewell.com
-- Stability   : experimental
-- Portability : GHC
--
-- Parallel parsers. Distributing the input to multiple parsers at the same
-- time.
--
-- For simplicity, we are using code where a particular state is unreachable
-- but it is not prevented by types.  Somehow uni-pattern match using "let"
-- produces better optimized code compared to using @case@ match and using
-- explicit error messages in unreachable cases.
--
-- There seem to be no way to silence individual warnings so we use a global
-- incomplete uni-pattern match warning suppression option for the file.
-- Disabling the warning for other code as well  has the potential to mask off
-- some legit warnings, therefore, we have segregated only the code that uses
-- uni-pattern matches in this module.

module Streamly.Internal.Data.Parser.ParserD.Tee
    (
    -- Parallel zipped
      teeWith
    , teeWithFst
    , teeWithMin

    -- Parallel alternatives
    , shortest
    , longest
    )
where

import Control.Exception (assert)
import Control.Monad.Catch (MonadCatch, try)
import Prelude
       hiding (any, all, takeWhile)

import Fusion.Plugin.Types (Fuse(..))
import Streamly.Internal.Data.Parser.ParserD.Types
       (Parser(..), Step(..), ParseError)

-------------------------------------------------------------------------------
-- Distribute input to two parsers and collect both results
-------------------------------------------------------------------------------

-- When the input stream is distributed to two parsers, both the parsers can
-- backtrack independently. Therefore, we need separate buffer state for each
-- parser.
--
-- ParserK
--
-- We can keep the state of each parser in the zipper and pass around that
-- zipper to the parsers. Each parser can consume from the zipper and then pass
-- around the zipper to the other parser.
--
-- ParserD
--
-- In the approach we have taken here, the driver pushes one element at a time
-- to the tee and each of the parsers in the tee may buffer it independently
-- for backtracking. So they do not need to depend on the original stream
-- source for individual parser backtracking. Problem arises when both the
-- parsers backtrack and they do not need any input from the driver rather they
-- must consume from their buffers. For such situation we may need a
-- "Continue" style driver command from the tee so that the driver runs
-- the tee without providing it any input. Or we may need a local driver loop
-- until new input is to be demanded from the input stream.
--
-- When the tee errors out or stops, the tee driver may have to backtrack by
-- the specified amount (or the tee must return the leftover input). Therefore,
-- the tee driver also has to buffer, this leads to triple buffering.
--
-- When the tee stops we need to determine the backtracking amount from the
-- leftover of both the parsers. Since both the parsers may have consumed
-- different lengths of the stream we consider the maximum of the two as
-- consumed.
--
{-# ANN type StepState Fuse #-}
data StepState s a = StepState s | StepResult a

-- | State of the pair of parsers in a tee composition
-- Note: strictness annotation is important for fusing the constructors
{-# ANN type TeeState Fuse #-}
data TeeState sL sR x a b =
-- @TeePair (past buffer, parser state, future-buffer1, future-buffer2) ...@
    TeePair !([x], StepState sL a, [x], [x]) !([x], StepState sR b, [x], [x])

{-# ANN type Res Fuse #-}
data Res = Yld Int | Stp Int | Skp | Err String

-- | See 'Streamly.Internal.Data.Parser.teeWith'.
--
-- /Broken/
--
{-# INLINE teeWith #-}
teeWith :: Monad m
    => (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
teeWith zf (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
    Parser step initial extract

    where

    {-# INLINE_LATE initial #-}
    initial = do
        sL <- initialL
        sR <- initialR
        return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])

    {-# INLINE consume #-}
    consume buf inp1 inp2 stp st y = do
        let (x, inp11, inp21) =
                case inp1 of
                    [] -> (y, [], [])
                    z : [] -> (z, reverse (x:inp2), [])
                    z : zs -> (z, zs, x:inp2)
        r <- stp st x
        let buf1 = x:buf
        return (buf1, r, inp11, inp21)

    -- XXX This is currently broken, even though both the parsers need to
    -- consume from their buffers after backtracking the driver would still be
    -- pushing more input to the buffers.
    --
    -- consume one input item and return the next state of the fold
    {-# INLINE useStream #-}
    useStream buf inp1 inp2 stp st y = do
        (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
        case r of
            Partial 0 s ->
                let state = ([], StepState s, inp11, inp21)
                 in return (state, Yld 0)
            Partial n s ->
                let src0 = Prelude.take n buf1
                    src  = Prelude.reverse src0
                    state = ([], StepState s, src ++ inp11, inp21)
                 in assert (n <= length buf1) (return (state, Yld n))
            Done n b ->
                let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
                 in assert (n <= length buf1) (return (state, Stp n))
            -- Continue 0 s -> (buf1, Right s, inp11, inp21)
            Continue n s ->
                let (src0, buf2) = splitAt n buf1
                    src  = Prelude.reverse src0
                    state = (buf2, StepState s, src ++ inp11, inp21)
                 in assert (n <= length buf1) (return (state, Skp))
            Error err -> return (undefined, Err err)

    {-# INLINE_LATE step #-}
    step (TeePair (bufL, StepState sL, inpL1, inpL2)
                  (bufR, StepState sR, inpR1, inpR2)) x = do
        (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
        (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
        let next = TeePair l r
        return $ case (stL,stR) of
            (Yld n1, Yld n2) -> Partial (min n1 n2) next
            (Yld n1, Stp n2) -> Partial (min n1 n2) next
            (Stp n1, Yld n2) -> Partial (min n1 n2) next
            (Stp n1, Stp n2) ->
                -- Uni-pattern match results in better optimized code compared
                -- to a case match.
                let (_, StepResult rL, _, _) = l
                    (_, StepResult rR, _, _) = r
                 in Done (min n1 n2) (zf rL rR)
            (Err err, _) -> Error err
            (_, Err err) -> Error err
            _ -> Continue 0 next

    step (TeePair (bufL, StepState sL, inpL1, inpL2)
                r@(_, StepResult rR, _, _)) x = do
        (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
        let next = TeePair l r
        -- XXX If the unused count of this stream is lower than the unused
        -- count of the stopped stream, only then this will be correct. We need
        -- to fix the other case. We need to keep incrementing the unused count
        -- of the stopped stream and take the min of the two.
        return $ case stL of
            Yld n -> Partial n next
            Stp n ->
                let (_, StepResult rL, _, _) = l
                 in Done n (zf rL rR)
            Skp -> Continue 0 next
            Err err -> Error err

    step (TeePair l@(_, StepResult rL, _, _)
                    (bufR, StepState sR, inpR1, inpR2)) x = do
        (r, stR) <- useStream bufR inpR1 inpR2 stepR sR x
        let next = TeePair l r
        -- XXX If the unused count of this stream is lower than the unused
        -- count of the stopped stream, only then this will be correct. We need
        -- to fix the other case. We need to keep incrementing the unused count
        -- of the stopped stream and take the min of the two.
        return $ case stR of
            Yld n -> Partial n next
            Stp n ->
                let (_, StepResult rR, _, _) = r
                 in Done n (zf rL rR)
            Skp -> Continue 0 next
            Err err -> Error err

    step _ _ = undefined

    {-# INLINE_LATE extract #-}
    extract st =
        case st of
            TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
                rL <- extractL sL
                rR <- extractR sR
                return $ zf rL rR
            TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
                rL <- extractL sL
                return $ zf rL rR
            TeePair (_, StepResult  rL, _, _) (_, StepState sR, _, _) -> do
                rR <- extractR sR
                return $ zf rL rR
            TeePair (_, StepResult rL, _, _) (_, StepResult rR, _, _) ->
                return $ zf rL rR

-- | See 'Streamly.Internal.Data.Parser.teeWithFst'.
--
-- /Broken/
--
{-# INLINE teeWithFst #-}
teeWithFst :: Monad m
    => (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
teeWithFst zf (Parser stepL initialL extractL)
              (Parser stepR initialR extractR) =
    Parser step initial extract

    where

    {-# INLINE_LATE initial #-}
    initial = do
        sL <- initialL
        sR <- initialR
        return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])

    {-# INLINE consume #-}
    consume buf inp1 inp2 stp st y = do
        let (x, inp11, inp21) =
                case inp1 of
                    [] -> (y, [], [])
                    z : [] -> (z, reverse (x:inp2), [])
                    z : zs -> (z, zs, x:inp2)
        r <- stp st x
        let buf1 = x:buf
        return (buf1, r, inp11, inp21)

    -- consume one input item and return the next state of the fold
    {-# INLINE useStream #-}
    useStream buf inp1 inp2 stp st y = do
        (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
        case r of
            Partial 0 s ->
                let state = ([], StepState s, inp11, inp21)
                 in return (state, Yld 0)
            Partial n _ -> return (undefined, Yld n) -- Not implemented
            Done n b ->
                let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
                 in assert (n <= length buf1) (return (state, Stp n))
            -- Continue 0 s -> (buf1, Right s, inp11, inp21)
            Continue n s ->
                let (src0, buf2) = splitAt n buf1
                    src  = Prelude.reverse src0
                    state = (buf2, StepState s, src ++ inp11, inp21)
                 in assert (n <= length buf1) (return (state, Skp))
            Error err -> return (undefined, Err err)

    {-# INLINE_LATE step #-}
    step (TeePair (bufL, StepState sL, inpL1, inpL2)
                  (bufR, StepState sR, inpR1, inpR2)) x = do
        (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
        (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
        let next = TeePair l r
        case (stL,stR) of
            -- XXX what if the first parser returns an unused count which is
            -- more than the second parser's unused count? It does not make
            -- sense for the second parser to consume more than the first
            -- parser. We reset the input cursor based on the first parser.
            -- Error out if the second one has consumed more then the first?
            (Stp n1, Stp _) ->
                -- Uni-pattern match results in better optimized code compared
                -- to a case match.
                let (_, StepResult rL, _, _) = l
                    (_, StepResult rR, _, _) = r
                 in return $ Done n1 (zf rL rR)
            (Stp n1, Yld _) ->
                let (_, StepResult rL, _, _) = l
                    (_, StepState  ssR, _, _) = r
                 in do
                    rR <- extractR ssR
                    return $ Done n1 (zf rL rR)
            (Yld n1, Yld n2) -> return $ Partial (min n1 n2) next
            (Yld n1, Stp n2) -> return $ Partial (min n1 n2) next
            (Err err, _) -> return $ Error err
            (_, Err err) -> return $ Error err
            _ -> return $ Continue 0 next

    step (TeePair (bufL, StepState sL, inpL1, inpL2)
                r@(_, StepResult rR, _, _)) x = do
        (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
        let next = TeePair l r
        -- XXX If the unused count of this stream is lower than the unused
        -- count of the stopped stream, only then this will be correct. We need
        -- to fix the other case. We need to keep incrementing the unused count
        -- of the stopped stream and take the min of the two.
        return $ case stL of
            Yld n -> Partial n next
            Stp n ->
                let (_, StepResult rL, _, _) = l
                 in Done n (zf rL rR)
            Skp -> Continue 0 next
            Err err -> Error err

    step _ _ = undefined

    {-# INLINE_LATE extract #-}
    extract st =
        case st of
            TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
                rL <- extractL sL
                rR <- extractR sR
                return $ zf rL rR
            TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
                rL <- extractL sL
                return $ zf rL rR
            _ -> error "unreachable"

-- | See 'Streamly.Internal.Data.Parser.teeWithMin'.
--
-- /Unimplemented/
--
{-# INLINE teeWithMin #-}
teeWithMin ::
    -- Monad m =>
    (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
teeWithMin = undefined

-------------------------------------------------------------------------------
-- Distribute input to two parsers and choose one result
-------------------------------------------------------------------------------

-- | See 'Streamly.Internal.Data.Parser.shortest'.
--
-- /Broken/
--
{-# INLINE shortest #-}
shortest :: Monad m => Parser m x a -> Parser m x a -> Parser m x a
shortest (Parser stepL initialL extractL) (Parser stepR initialR _) =
    Parser step initial extract

    where

    {-# INLINE_LATE initial #-}
    initial = do
        sL <- initialL
        sR <- initialR
        return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])

    {-# INLINE consume #-}
    consume buf inp1 inp2 stp st y = do
        let (x, inp11, inp21) =
                case inp1 of
                    [] -> (y, [], [])
                    z : [] -> (z, reverse (x:inp2), [])
                    z : zs -> (z, zs, x:inp2)
        r <- stp st x
        let buf1 = x:buf
        return (buf1, r, inp11, inp21)

    -- consume one input item and return the next state of the fold
    {-# INLINE useStream #-}
    useStream buf inp1 inp2 stp st y = do
        (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
        case r of
            Partial 0 s ->
                let state = ([], StepState s, inp11, inp21)
                 in return (state, Yld 0)
            Partial n _ -> return (undefined, Yld n) -- Not implemented
            Done n b ->
                let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
                 in assert (n <= length buf1) (return (state, Stp n))
            -- Continue 0 s -> (buf1, Right s, inp11, inp21)
            Continue n s ->
                let (src0, buf2) = splitAt n buf1
                    src  = Prelude.reverse src0
                    state = (buf2, StepState s, src ++ inp11, inp21)
                 in assert (n <= length buf1) (return (state, Skp))
            Error err -> return (undefined, Err err)

    -- XXX Even if a parse finished earlier it may not be shortest if the other
    -- parser finishes later but returns a lot of unconsumed input. Our current
    -- criterion of shortest is whichever parse decided to stop earlier.
    {-# INLINE_LATE step #-}
    step (TeePair (bufL, StepState sL, inpL1, inpL2)
                  (bufR, StepState sR, inpR1, inpR2)) x = do
        (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
        (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
        let next = TeePair l r
        return $ case (stL,stR) of
            (Stp n1, _) ->
                let (_, StepResult rL, _, _) = l
                 in Done n1 rL
            (_, Stp n2) ->
                let (_, StepResult rR, _, _) = r
                 in Done n2 rR
            (Yld n1, Yld n2) -> Partial (min n1 n2) next
            (Err err, _) -> Error err
            (_, Err err) -> Error err
            _ -> Continue 0 next

    step _ _ = undefined

    {-# INLINE_LATE extract #-}
    extract st =
        case st of
            TeePair (_, StepState sL, _, _) _ -> extractL sL
            _ -> error "unreachable"

-- | See 'Streamly.Internal.Data.Parser.longest'.
--
-- /Broken/
--
{-# INLINE longest #-}
longest :: MonadCatch m => Parser m x a -> Parser m x a -> Parser m x a
longest (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
    Parser step initial extract

    where

    {-# INLINE_LATE initial #-}
    initial = do
        sL <- initialL
        sR <- initialR
        return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])

    {-# INLINE consume #-}
    consume buf inp1 inp2 stp st y = do
        let (x, inp11, inp21) =
                case inp1 of
                    [] -> (y, [], [])
                    z : [] -> (z, reverse (x:inp2), [])
                    z : zs -> (z, zs, x:inp2)
        r <- stp st x
        let buf1 = x:buf
        return (buf1, r, inp11, inp21)

    -- consume one input item and return the next state of the fold
    {-# INLINE useStream #-}
    useStream buf inp1 inp2 stp st y = do
        (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
        case r of
            Partial 0 s ->
                let state = ([], StepState s, inp11, inp21)
                 in return (state, Yld 0)
            Partial n _ -> return (undefined, Yld n) -- Not implemented
            Done n b ->
                let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
                 in assert (n <= length buf1) (return (state, Stp n))
            -- Continue 0 s -> (buf1, Right s, inp11, inp21)
            Continue n s ->
                let (src0, buf2) = splitAt n buf1
                    src  = Prelude.reverse src0
                    state = (buf2, StepState s, src ++ inp11, inp21)
                 in assert (n <= length buf1) (return (state, Skp))
            Error err -> return (undefined, Err err)

    {-# INLINE_LATE step #-}
    step (TeePair (bufL, StepState sL, inpL1, inpL2)
                  (bufR, StepState sR, inpR1, inpR2)) x = do
        (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
        (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
        let next = TeePair l r
        return $ case (stL,stR) of
            (Yld n1, Yld n2) -> Partial (min n1 n2) next
            (Yld n1, Stp n2) -> Partial (min n1 n2) next
            (Stp n1, Yld n2) -> Partial (min n1 n2) next
            (Stp n1, Stp n2) ->
                let (_, StepResult rL, _, _) = l
                    (_, StepResult rR, _, _) = r
                 in Done (max n1 n2) (if n1 >= n2 then rL else rR)
            (Err err, _) -> Error err
            (_, Err err) -> Error err
            _ -> Continue 0 next

    -- XXX the parser that finishes last may not be the longest because it may
    -- return a lot of unused input which makes it shorter. Our current
    -- criterion of deciding longest is based on whoever decides to finish
    -- last and not whoever consumed more input.
    --
    -- To actually know who made more progress we need to keep an account of
    -- how many items are unconsumed since the last yield.
    --
    step (TeePair (bufL, StepState sL, inpL1, inpL2)
                r@(_, StepResult _, _, _)) x = do
        (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
        let next = TeePair l r
        return $ case stL of
            Yld n -> Partial n next
            Stp n ->
                let (_, StepResult rL, _, _) = l
                 in Done n rL
            Skp -> Continue 0 next
            Err err -> Error err

    step (TeePair l@(_, StepResult _, _, _)
                    (bufR, StepState sR, inpR1, inpR2)) x = do
        (r, stR) <- useStream bufR inpR1 inpR2 stepR sR x
        let next = TeePair l r
        return $ case stR of
            Yld n -> Partial n next
            Stp n ->
                let (_, StepResult rR, _, _) = r
                 in Done n rR
            Skp -> Continue 0 next
            Err err -> Error err

    step _ _ = undefined

    {-# INLINE_LATE extract #-}
    extract st =
        -- XXX When results are partial we may not be able to precisely compare
        -- which parser has made more progress till now.  One way to do that is
        -- to figure out the actually consumed input up to the last yield.
        --
        case st of
            TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
                r <- try $ extractL sL
                case r of
                    Left (_ :: ParseError) -> extractR sR
                    Right b -> return b
            TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
                r <- try $ extractL sL
                case r of
                    Left (_ :: ParseError) -> return rR
                    Right b -> return b
            TeePair (_, StepResult rL, _, _) (_, StepState sR, _, _) -> do
                r <- try $ extractR sR
                case r of
                    Left (_ :: ParseError) -> return rL
                    Right b -> return b
            TeePair (_, StepResult _, _, _) (_, StepResult _, _, _) ->
                error "unreachable"

1	{-# OPTIONS_GHC -Wno-incomplete-uni-patterns #-}
2
3	#include "inline.hs"
4
5	-- \|
6	-- Module : Streamly.Internal.Data.Parser.ParserD.Tee
7	-- Copyright : (c) 2020 Composewell Technologies
8	-- License : BSD-3-Clause
9	-- Maintainer : streamly@composewell.com
10	-- Stability : experimental
11	-- Portability : GHC
12	--
13	-- Parallel parsers. Distributing the input to multiple parsers at the same
14	-- time.
15	--
16	-- For simplicity, we are using code where a particular state is unreachable
17	-- but it is not prevented by types. Somehow uni-pattern match using "let"
18	-- produces better optimized code compared to using @case@ match and using
19	-- explicit error messages in unreachable cases.
20	--
21	-- There seem to be no way to silence individual warnings so we use a global
22	-- incomplete uni-pattern match warning suppression option for the file.
23	-- Disabling the warning for other code as well has the potential to mask off
24	-- some legit warnings, therefore, we have segregated only the code that uses
25	-- uni-pattern matches in this module.
26
27	module Streamly.Internal.Data.Parser.ParserD.Tee
28	(
29	-- Parallel zipped
30	teeWith
31	, teeWithFst
32	, teeWithMin
33
34	-- Parallel alternatives
35	, shortest
36	, longest
37	)
38	where
39
40	import Control.Exception (assert)
41	import Control.Monad.Catch (MonadCatch, try)
42	import Prelude
43	hiding (any, all, takeWhile)
44
45	import Fusion.Plugin.Types (Fuse(..))
46	import Streamly.Internal.Data.Parser.ParserD.Types
47	(Parser(..), Step(..), ParseError)
48
49	-------------------------------------------------------------------------------
50	-- Distribute input to two parsers and collect both results
51	-------------------------------------------------------------------------------
52
53	-- When the input stream is distributed to two parsers, both the parsers can
54	-- backtrack independently. Therefore, we need separate buffer state for each
55	-- parser.
56	--
57	-- ParserK
58	--
59	-- We can keep the state of each parser in the zipper and pass around that
60	-- zipper to the parsers. Each parser can consume from the zipper and then pass
61	-- around the zipper to the other parser.
62	--
63	-- ParserD
64	--
65	-- In the approach we have taken here, the driver pushes one element at a time
66	-- to the tee and each of the parsers in the tee may buffer it independently
67	-- for backtracking. So they do not need to depend on the original stream
68	-- source for individual parser backtracking. Problem arises when both the
69	-- parsers backtrack and they do not need any input from the driver rather they
70	-- must consume from their buffers. For such situation we may need a
71	-- "Continue" style driver command from the tee so that the driver runs
72	-- the tee without providing it any input. Or we may need a local driver loop
73	-- until new input is to be demanded from the input stream.
74	--
75	-- When the tee errors out or stops, the tee driver may have to backtrack by
76	-- the specified amount (or the tee must return the leftover input). Therefore,
77	-- the tee driver also has to buffer, this leads to triple buffering.
78	--
79	-- When the tee stops we need to determine the backtracking amount from the
80	-- leftover of both the parsers. Since both the parsers may have consumed
81	-- different lengths of the stream we consider the maximum of the two as
82	-- consumed.
83	--
84	{-# ANN type StepState Fuse #-}
85	data StepState s a = StepState s \| StepResult a
86
87	-- \| State of the pair of parsers in a tee composition
88	-- Note: strictness annotation is important for fusing the constructors
89	{-# ANN type TeeState Fuse #-}
90	data TeeState sL sR x a b =
91	-- @TeePair (past buffer, parser state, future-buffer1, future-buffer2) ...@
92	TeePair !([x], StepState sL a, [x], [x]) !([x], StepState sR b, [x], [x])
93
94	{-# ANN type Res Fuse #-}
95	data Res = Yld Int \| Stp Int \| Skp \| Err String
96
97	-- \| See 'Streamly.Internal.Data.Parser.teeWith'.
98	--
99	-- /Broken/
100	--
101	{-# INLINE teeWith #-}
102	teeWith :: Monad m
103	=> (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
104	teeWith zf (Parser stepL initialL extractL) (Parser stepR initialR extractR) =	1✔
105	Parser step initial extract	1✔
106
107	where
108
109	{-# INLINE_LATE initial #-}
110	initial = do	1✔
111	sL <- initialL	1✔
112	sR <- initialR	1✔
113	return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])	×
114
115	{-# INLINE consume #-}
116	consume buf inp1 inp2 stp st y = do	1✔
117	let (x, inp11, inp21) =
118	case inp1 of	1✔
119	[] -> (y, [], [])	×
120	z : [] -> (z, reverse (x:inp2), [])	×
121	z : zs -> (z, zs, x:inp2)	×
122	r <- stp st x	1✔
123	let buf1 = x:buf	×
124	return (buf1, r, inp11, inp21)	×
125
126	-- XXX This is currently broken, even though both the parsers need to
127	-- consume from their buffers after backtracking the driver would still be
128	-- pushing more input to the buffers.
129	--
130	-- consume one input item and return the next state of the fold
131	{-# INLINE useStream #-}
132	useStream buf inp1 inp2 stp st y = do	1✔
133	(buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y	×
134	case r of	1✔
135	Partial 0 s ->
136	let state = ([], StepState s, inp11, inp21)	×
137	in return (state, Yld 0)	1✔
138	Partial n s ->
139	let src0 = Prelude.take n buf1	×
140	src = Prelude.reverse src0	×
141	state = ([], StepState s, src ++ inp11, inp21)	×
142	in assert (n <= length buf1) (return (state, Yld n))	×
143	Done n b ->
144	let state = (Prelude.take n buf1, StepResult b, inp11, inp21)	×
145	in assert (n <= length buf1) (return (state, Stp n))	×
146	-- Continue 0 s -> (buf1, Right s, inp11, inp21)
147	Continue n s ->
148	let (src0, buf2) = splitAt n buf1	×
149	src = Prelude.reverse src0	×
150	state = (buf2, StepState s, src ++ inp11, inp21)	×
151	in assert (n <= length buf1) (return (state, Skp))	×
152	Error err -> return (undefined, Err err)	×
153
154	{-# INLINE_LATE step #-}
155	step (TeePair (bufL, StepState sL, inpL1, inpL2)	1✔
156	(bufR, StepState sR, inpR1, inpR2)) x = do	1✔
157	(l,stL) <- useStream bufL inpL1 inpL2 stepL sL x	×
158	(r,stR) <- useStream bufR inpR1 inpR2 stepR sR x	×
159	let next = TeePair l r	1✔
160	return $ case (stL,stR) of	1✔
161	(Yld n1, Yld n2) -> Partial (min n1 n2) next	1✔
162	(Yld n1, Stp n2) -> Partial (min n1 n2) next	×
163	(Stp n1, Yld n2) -> Partial (min n1 n2) next	×
164	(Stp n1, Stp n2) ->
165	-- Uni-pattern match results in better optimized code compared
166	-- to a case match.
167	let (_, StepResult rL, _, _) = l	×
168	(_, StepResult rR, _, _) = r	×
169	in Done (min n1 n2) (zf rL rR)	×
170	(Err err, _) -> Error err	×
171	(_, Err err) -> Error err	×
172	_ -> Continue 0 next	×
173
174	step (TeePair (bufL, StepState sL, inpL1, inpL2)
175	r@(_, StepResult rR, _, _)) x = do	×
176	(l,stL) <- useStream bufL inpL1 inpL2 stepL sL x	×
177	let next = TeePair l r	×
178	-- XXX If the unused count of this stream is lower than the unused
179	-- count of the stopped stream, only then this will be correct. We need
180	-- to fix the other case. We need to keep incrementing the unused count
181	-- of the stopped stream and take the min of the two.
182	return $ case stL of	×
183	Yld n -> Partial n next	×
184	Stp n ->
185	let (_, StepResult rL, _, _) = l	×
186	in Done n (zf rL rR)	×
187	Skp -> Continue 0 next	×
188	Err err -> Error err	×
189
190	step (TeePair l@(_, StepResult rL, _, _)
191	(bufR, StepState sR, inpR1, inpR2)) x = do	×
192	(r, stR) <- useStream bufR inpR1 inpR2 stepR sR x	×
193	let next = TeePair l r	×
194	-- XXX If the unused count of this stream is lower than the unused
195	-- count of the stopped stream, only then this will be correct. We need
196	-- to fix the other case. We need to keep incrementing the unused count
197	-- of the stopped stream and take the min of the two.
198	return $ case stR of	×
199	Yld n -> Partial n next	×
200	Stp n ->
201	let (_, StepResult rR, _, _) = r	×
202	in Done n (zf rL rR)	×
203	Skp -> Continue 0 next	×
204	Err err -> Error err	×
205
206	step _ _ = undefined	×
207
208	{-# INLINE_LATE extract #-}
209	extract st =	1✔
210	case st of	1✔
211	TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do	1✔
212	rL <- extractL sL	1✔
213	rR <- extractR sR	1✔
214	return $ zf rL rR	1✔
215	TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do	×
216	rL <- extractL sL	×
217	return $ zf rL rR	×
218	TeePair (_, StepResult rL, _, _) (_, StepState sR, _, _) -> do	×
219	rR <- extractR sR	×
220	return $ zf rL rR	×
221	TeePair (_, StepResult rL, _, _) (_, StepResult rR, _, _) ->
222	return $ zf rL rR	×
223
224	-- \| See 'Streamly.Internal.Data.Parser.teeWithFst'.
225	--
226	-- /Broken/
227	--
228	{-# INLINE teeWithFst #-}
229	teeWithFst :: Monad m
230	=> (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
231	teeWithFst zf (Parser stepL initialL extractL)	×
232	(Parser stepR initialR extractR) =
233	Parser step initial extract	×
234
235	where
236
237	{-# INLINE_LATE initial #-}
238	initial = do	×
239	sL <- initialL	×
240	sR <- initialR	×
241	return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])	×
242
243	{-# INLINE consume #-}
244	consume buf inp1 inp2 stp st y = do	×
245	let (x, inp11, inp21) =
246	case inp1 of	×
247	[] -> (y, [], [])	×
248	z : [] -> (z, reverse (x:inp2), [])	×
249	z : zs -> (z, zs, x:inp2)	×
250	r <- stp st x	×
251	let buf1 = x:buf	×
252	return (buf1, r, inp11, inp21)	×
253
254	-- consume one input item and return the next state of the fold
255	{-# INLINE useStream #-}
256	useStream buf inp1 inp2 stp st y = do	×
257	(buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y	×
258	case r of	×
259	Partial 0 s ->
260	let state = ([], StepState s, inp11, inp21)	×
261	in return (state, Yld 0)	×
262	Partial n _ -> return (undefined, Yld n) -- Not implemented	×
263	Done n b ->
264	let state = (Prelude.take n buf1, StepResult b, inp11, inp21)	×
265	in assert (n <= length buf1) (return (state, Stp n))	×
266	-- Continue 0 s -> (buf1, Right s, inp11, inp21)
267	Continue n s ->
268	let (src0, buf2) = splitAt n buf1	×
269	src = Prelude.reverse src0	×
270	state = (buf2, StepState s, src ++ inp11, inp21)	×
271	in assert (n <= length buf1) (return (state, Skp))	×
272	Error err -> return (undefined, Err err)	×
273
274	{-# INLINE_LATE step #-}
275	step (TeePair (bufL, StepState sL, inpL1, inpL2)	×
276	(bufR, StepState sR, inpR1, inpR2)) x = do	×
277	(l,stL) <- useStream bufL inpL1 inpL2 stepL sL x	×
278	(r,stR) <- useStream bufR inpR1 inpR2 stepR sR x	×
279	let next = TeePair l r	×
280	case (stL,stR) of	×
281	-- XXX what if the first parser returns an unused count which is
282	-- more than the second parser's unused count? It does not make
283	-- sense for the second parser to consume more than the first
284	-- parser. We reset the input cursor based on the first parser.
285	-- Error out if the second one has consumed more then the first?
286	(Stp n1, Stp _) ->
287	-- Uni-pattern match results in better optimized code compared
288	-- to a case match.
289	let (_, StepResult rL, _, _) = l	×
290	(_, StepResult rR, _, _) = r	×
291	in return $ Done n1 (zf rL rR)	×
292	(Stp n1, Yld _) ->
293	let (_, StepResult rL, _, _) = l	×
294	(_, StepState ssR, _, _) = r	×
295	in do
296	rR <- extractR ssR	×
297	return $ Done n1 (zf rL rR)	×
298	(Yld n1, Yld n2) -> return $ Partial (min n1 n2) next	×
299	(Yld n1, Stp n2) -> return $ Partial (min n1 n2) next	×
300	(Err err, _) -> return $ Error err	×
301	(_, Err err) -> return $ Error err	×
302	_ -> return $ Continue 0 next	×
303
304	step (TeePair (bufL, StepState sL, inpL1, inpL2)
305	r@(_, StepResult rR, _, _)) x = do	×
306	(l,stL) <- useStream bufL inpL1 inpL2 stepL sL x	×
307	let next = TeePair l r	×
308	-- XXX If the unused count of this stream is lower than the unused
309	-- count of the stopped stream, only then this will be correct. We need
310	-- to fix the other case. We need to keep incrementing the unused count
311	-- of the stopped stream and take the min of the two.
312	return $ case stL of	×
313	Yld n -> Partial n next	×
314	Stp n ->
315	let (_, StepResult rL, _, _) = l	×
316	in Done n (zf rL rR)	×
317	Skp -> Continue 0 next	×
318	Err err -> Error err	×
319
320	step _ _ = undefined	×
321
322	{-# INLINE_LATE extract #-}
323	extract st =	×
324	case st of	×
325	TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do	×
326	rL <- extractL sL	×
327	rR <- extractR sR	×
328	return $ zf rL rR	×
329	TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do	×
330	rL <- extractL sL	×
331	return $ zf rL rR	×
332	_ -> error "unreachable"	×
333
334	-- \| See 'Streamly.Internal.Data.Parser.teeWithMin'.
335	--
336	-- /Unimplemented/
337	--
338	{-# INLINE teeWithMin #-}
339	teeWithMin ::
340	-- Monad m =>
341	(a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
342	teeWithMin = undefined	×
343
344	-------------------------------------------------------------------------------
345	-- Distribute input to two parsers and choose one result
346	-------------------------------------------------------------------------------
347
348	-- \| See 'Streamly.Internal.Data.Parser.shortest'.
349	--
350	-- /Broken/
351	--
352	{-# INLINE shortest #-}
353	shortest :: Monad m => Parser m x a -> Parser m x a -> Parser m x a
354	shortest (Parser stepL initialL extractL) (Parser stepR initialR _) =	1✔
355	Parser step initial extract	1✔
356
357	where
358
359	{-# INLINE_LATE initial #-}
360	initial = do	1✔
361	sL <- initialL	1✔
362	sR <- initialR	1✔
363	return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])	×
364
365	{-# INLINE consume #-}
366	consume buf inp1 inp2 stp st y = do	1✔
367	let (x, inp11, inp21) =
368	case inp1 of	1✔
369	[] -> (y, [], [])	×
370	z : [] -> (z, reverse (x:inp2), [])	×
371	z : zs -> (z, zs, x:inp2)	×
372	r <- stp st x	1✔
373	let buf1 = x:buf	×
374	return (buf1, r, inp11, inp21)	×
375
376	-- consume one input item and return the next state of the fold
377	{-# INLINE useStream #-}
378	useStream buf inp1 inp2 stp st y = do	1✔
379	(buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y	×
380	case r of	1✔
381	Partial 0 s ->
382	let state = ([], StepState s, inp11, inp21)	×
383	in return (state, Yld 0)	1✔
384	Partial n _ -> return (undefined, Yld n) -- Not implemented	×
385	Done n b ->
386	let state = (Prelude.take n buf1, StepResult b, inp11, inp21)	×
387	in assert (n <= length buf1) (return (state, Stp n))	×
388	-- Continue 0 s -> (buf1, Right s, inp11, inp21)
389	Continue n s ->
390	let (src0, buf2) = splitAt n buf1	×
391	src = Prelude.reverse src0	×
392	state = (buf2, StepState s, src ++ inp11, inp21)	×
393	in assert (n <= length buf1) (return (state, Skp))	×
394	Error err -> return (undefined, Err err)	×
395
396	-- XXX Even if a parse finished earlier it may not be shortest if the other
397	-- parser finishes later but returns a lot of unconsumed input. Our current
398	-- criterion of shortest is whichever parse decided to stop earlier.
399	{-# INLINE_LATE step #-}
400	step (TeePair (bufL, StepState sL, inpL1, inpL2)	1✔
401	(bufR, StepState sR, inpR1, inpR2)) x = do	1✔
402	(l,stL) <- useStream bufL inpL1 inpL2 stepL sL x	×
403	(r,stR) <- useStream bufR inpR1 inpR2 stepR sR x	×
404	let next = TeePair l r	1✔
405	return $ case (stL,stR) of	1✔
406	(Stp n1, _) ->
407	let (_, StepResult rL, _, _) = l	1✔
408	in Done n1 rL	×
409	(_, Stp n2) ->
410	let (_, StepResult rR, _, _) = r	1✔
411	in Done n2 rR	×
412	(Yld n1, Yld n2) -> Partial (min n1 n2) next	1✔
413	(Err err, _) -> Error err	×
414	(_, Err err) -> Error err	×
415	_ -> Continue 0 next	×
416
417	step _ _ = undefined	×
418
419	{-# INLINE_LATE extract #-}
420	extract st =	1✔
421	case st of	1✔
422	TeePair (_, StepState sL, _, _) _ -> extractL sL	1✔
423	_ -> error "unreachable"	×
424
425	-- \| See 'Streamly.Internal.Data.Parser.longest'.
426	--
427	-- /Broken/
428	--
429	{-# INLINE longest #-}
430	longest :: MonadCatch m => Parser m x a -> Parser m x a -> Parser m x a
431	longest (Parser stepL initialL extractL) (Parser stepR initialR extractR) =	1✔
432	Parser step initial extract	1✔
433
434	where
435
436	{-# INLINE_LATE initial #-}
437	initial = do	1✔
438	sL <- initialL	1✔
439	sR <- initialR	1✔
440	return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])	×
441
442	{-# INLINE consume #-}
443	consume buf inp1 inp2 stp st y = do	1✔
444	let (x, inp11, inp21) =
445	case inp1 of	1✔
446	[] -> (y, [], [])	×
447	z : [] -> (z, reverse (x:inp2), [])	×
448	z : zs -> (z, zs, x:inp2)	×
449	r <- stp st x	1✔
450	let buf1 = x:buf	×
451	return (buf1, r, inp11, inp21)	×
452
453	-- consume one input item and return the next state of the fold
454	{-# INLINE useStream #-}
455	useStream buf inp1 inp2 stp st y = do	1✔
456	(buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y	×
457	case r of	1✔
458	Partial 0 s ->
459	let state = ([], StepState s, inp11, inp21)	×
460	in return (state, Yld 0)	1✔
461	Partial n _ -> return (undefined, Yld n) -- Not implemented	×
462	Done n b ->
463	let state = (Prelude.take n buf1, StepResult b, inp11, inp21)	×
464	in assert (n <= length buf1) (return (state, Stp n))	×
465	-- Continue 0 s -> (buf1, Right s, inp11, inp21)
466	Continue n s ->
467	let (src0, buf2) = splitAt n buf1	×
468	src = Prelude.reverse src0	×
469	state = (buf2, StepState s, src ++ inp11, inp21)	×
470	in assert (n <= length buf1) (return (state, Skp))	×
471	Error err -> return (undefined, Err err)	×
472
473	{-# INLINE_LATE step #-}
474	step (TeePair (bufL, StepState sL, inpL1, inpL2)	1✔
475	(bufR, StepState sR, inpR1, inpR2)) x = do	1✔
476	(l,stL) <- useStream bufL inpL1 inpL2 stepL sL x	×
477	(r,stR) <- useStream bufR inpR1 inpR2 stepR sR x	×
478	let next = TeePair l r	1✔
479	return $ case (stL,stR) of	1✔
480	(Yld n1, Yld n2) -> Partial (min n1 n2) next	1✔
481	(Yld n1, Stp n2) -> Partial (min n1 n2) next	×
482	(Stp n1, Yld n2) -> Partial (min n1 n2) next	1✔
483	(Stp n1, Stp n2) ->
484	let (_, StepResult rL, _, _) = l	1✔
485	(_, StepResult rR, _, _) = r	×
486	in Done (max n1 n2) (if n1 >= n2 then rL else rR)	×
487	(Err err, _) -> Error err	×
488	(_, Err err) -> Error err	×
489	_ -> Continue 0 next	×
490
491	-- XXX the parser that finishes last may not be the longest because it may
492	-- return a lot of unused input which makes it shorter. Our current
493	-- criterion of deciding longest is based on whoever decides to finish
494	-- last and not whoever consumed more input.
495	--
496	-- To actually know who made more progress we need to keep an account of
497	-- how many items are unconsumed since the last yield.
498	--
499	step (TeePair (bufL, StepState sL, inpL1, inpL2)
500	r@(_, StepResult _, _, _)) x = do	×
501	(l,stL) <- useStream bufL inpL1 inpL2 stepL sL x	×
502	let next = TeePair l r	×
503	return $ case stL of	×
504	Yld n -> Partial n next	×
505	Stp n ->
506	let (_, StepResult rL, _, _) = l	×
507	in Done n rL	×
508	Skp -> Continue 0 next	×
509	Err err -> Error err	×
510
511	step (TeePair l@(_, StepResult _, _, _)
512	(bufR, StepState sR, inpR1, inpR2)) x = do	1✔
513	(r, stR) <- useStream bufR inpR1 inpR2 stepR sR x	×
514	let next = TeePair l r	1✔
515	return $ case stR of	1✔
516	Yld n -> Partial n next	1✔
517	Stp n ->
518	let (_, StepResult rR, _, _) = r	1✔
519	in Done n rR	×
520	Skp -> Continue 0 next	×
521	Err err -> Error err	×
522
523	step _ _ = undefined	×
524
525	{-# INLINE_LATE extract #-}
526	extract st =	1✔
527	-- XXX When results are partial we may not be able to precisely compare
528	-- which parser has made more progress till now. One way to do that is
529	-- to figure out the actually consumed input up to the last yield.
530	--
531	case st of	1✔
532	TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do	1✔
533	r <- try $ extractL sL	1✔
534	case r of	1✔
535	Left (_ :: ParseError) -> extractR sR	×
536	Right b -> return b	1✔
537	TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do	×
538	r <- try $ extractL sL	×
539	case r of	×
540	Left (_ :: ParseError) -> return rR	×
541	Right b -> return b	×
542	TeePair (_, StepResult rL, _, _) (_, StepState sR, _, _) -> do	1✔
543	r <- try $ extractR sR	1✔
544	case r of	1✔
545	Left (_ :: ParseError) -> return rL	×
546	Right b -> return b	1✔
547	TeePair (_, StepResult _, _, _) (_, StepResult _, _, _) ->
548	error "unreachable"	×

composewell / streamly / 3095

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous