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