git.iko.soy/streamly
on ef76e4b help sign in
streamly/core/src/Streamly/Internal/Data/Parser/ParserD/Tee.hs
613 lines 25 kB history raw
1{-# OPTIONS_GHC -Wno-incomplete-uni-patterns #-}
3#include "inline.hs"
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.
27module Streamly.Internal.Data.Parser.ParserD.Tee
28 (
29 -- Parallel zipped
30 teeWith
31 , teeWithFst
32 , teeWithMin
34 -- Parallel alternatives
35 , shortest
36 , longest
37 )
38where
40import Control.Exception (assert)
41import Control.Monad.Catch (MonadCatch, try)
42import Prelude
43 hiding (any, all, takeWhile)
45import Fusion.Plugin.Types (Fuse(..))
46import Streamly.Internal.Data.Parser.ParserD.Type
47 (Initial(..), Parser(..), Step(..), ParseError)
49-------------------------------------------------------------------------------
50-- Distribute input to two parsers and collect both results
51-------------------------------------------------------------------------------
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 -- XXX We can use Initial instead of StepState
85{-# ANN type StepState Fuse #-}
86data StepState s a = StepState s | StepResult a
88-- | State of the pair of parsers in a tee composition
89-- Note: strictness annotation is important for fusing the constructors
90{-# ANN type TeeState Fuse #-}
91data TeeState sL sR x a b =
92-- @TeePair (past buffer, parser state, future-buffer1, future-buffer2) ...@
93 TeePair !([x], StepState sL a, [x], [x]) !([x], StepState sR b, [x], [x])
95{-# ANN type Res Fuse #-}
96data Res = Yld Int | Stp Int | Skp | Err String
98-- | See 'Streamly.Internal.Data.Parser.teeWith'.
99--
100-- /Broken/
101--
102{-# INLINE teeWith #-}
103teeWith :: Monad m
104 => (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
105teeWith zf (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
106 Parser step initial extract
108 where
110 {-# INLINE_LATE initial #-}
111 initial = do
112 resL <- initialL
113 resR <- initialR
114 return $ case resL of
115 IPartial sl ->
116 case resR of
117 IPartial sr -> IPartial $ TeePair ([], StepState sl, [], [])
118 ([], StepState sr, [], [])
119 IDone br -> IPartial $ TeePair ([], StepState sl, [], [])
120 ([], StepResult br, [], [])
121 IError err -> IError err
122 IDone bl ->
123 case resR of
124 IPartial sr ->
125 IPartial $ TeePair ([], StepResult bl, [], [])
126 ([], StepState sr, [], [])
127 IDone br -> IDone $ zf bl br
128 IError err -> IError err
129 IError err -> IError err
131 {-# INLINE consume #-}
132 consume buf inp1 inp2 stp st y = do
133 let (x, inp11, inp21) =
134 case inp1 of
135 [] -> (y, [], [])
136 z : [] -> (z, reverse (x:inp2), [])
137 z : zs -> (z, zs, x:inp2)
138 r <- stp st x
139 let buf1 = x:buf
140 return (buf1, r, inp11, inp21)
142 -- XXX This is currently broken, even though both the parsers need to
143 -- consume from their buffers after backtracking the driver would still be
144 -- pushing more input to the buffers.
145 --
146 -- consume one input item and return the next state of the fold
147 {-# INLINE useStream #-}
148 useStream buf inp1 inp2 stp st y = do
149 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
150 case r of
151 Partial 0 s ->
152 let state = ([], StepState s, inp11, inp21)
153 in return (state, Yld 0)
154 Partial n s ->
155 let src0 = Prelude.take n buf1
156 src = Prelude.reverse src0
157 state = ([], StepState s, src ++ inp11, inp21)
158 in assert (n <= length buf1) (return (state, Yld n))
159 Done n b ->
160 let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
161 in assert (n <= length buf1) (return (state, Stp n))
162 -- Continue 0 s -> (buf1, Right s, inp11, inp21)
163 Continue n s ->
164 let (src0, buf2) = splitAt n buf1
165 src = Prelude.reverse src0
166 state = (buf2, StepState s, src ++ inp11, inp21)
167 in assert (n <= length buf1) (return (state, Skp))
168 Error err -> return (undefined, Err err)
170 {-# INLINE_LATE step #-}
171 step (TeePair (bufL, StepState sL, inpL1, inpL2)
172 (bufR, StepState sR, inpR1, inpR2)) x = do
173 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
174 (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
175 let next = TeePair l r
176 return $ case (stL,stR) of
177 (Yld n1, Yld n2) -> Partial (min n1 n2) next
178 (Yld n1, Stp n2) -> Partial (min n1 n2) next
179 (Stp n1, Yld n2) -> Partial (min n1 n2) next
180 (Stp n1, Stp n2) ->
181 -- Uni-pattern match results in better optimized code compared
182 -- to a case match.
183 let (_, StepResult rL, _, _) = l
184 (_, StepResult rR, _, _) = r
185 in Done (min n1 n2) (zf rL rR)
186 (Err err, _) -> Error err
187 (_, Err err) -> Error err
188 _ -> Continue 0 next
190 step (TeePair (bufL, StepState sL, inpL1, inpL2)
191 r@(_, StepResult rR, _, _)) x = do
192 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL 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 stL of
199 Yld n -> Partial n next
200 Stp n ->
201 let (_, StepResult rL, _, _) = l
202 in Done n (zf rL rR)
203 Skp -> Continue 0 next
204 Err err -> Error err
206 step (TeePair l@(_, StepResult rL, _, _)
207 (bufR, StepState sR, inpR1, inpR2)) x = do
208 (r, stR) <- useStream bufR inpR1 inpR2 stepR sR x
209 let next = TeePair l r
210 -- XXX If the unused count of this stream is lower than the unused
211 -- count of the stopped stream, only then this will be correct. We need
212 -- to fix the other case. We need to keep incrementing the unused count
213 -- of the stopped stream and take the min of the two.
214 return $ case stR of
215 Yld n -> Partial n next
216 Stp n ->
217 let (_, StepResult rR, _, _) = r
218 in Done n (zf rL rR)
219 Skp -> Continue 0 next
220 Err err -> Error err
222 step _ _ = undefined
224 {-# INLINE_LATE extract #-}
225 extract st =
226 case st of
227 TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
228 rL <- extractL sL
229 rR <- extractR sR
230 return $ zf rL rR
231 TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
232 rL <- extractL sL
233 return $ zf rL rR
234 TeePair (_, StepResult rL, _, _) (_, StepState sR, _, _) -> do
235 rR <- extractR sR
236 return $ zf rL rR
237 TeePair (_, StepResult rL, _, _) (_, StepResult rR, _, _) ->
238 return $ zf rL rR
240-- | See 'Streamly.Internal.Data.Parser.teeWithFst'.
241--
242-- /Broken/
243--
244{-# INLINE teeWithFst #-}
245teeWithFst :: Monad m
246 => (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
247teeWithFst zf (Parser stepL initialL extractL)
248 (Parser stepR initialR extractR) =
249 Parser step initial extract
251 where
253 {-# INLINE_LATE initial #-}
254 initial = do
255 resL <- initialL
256 resR <- initialR
257 case resL of
258 IPartial sl ->
259 return $ case resR of
260 IPartial sr -> IPartial $ TeePair ([], StepState sl, [], [])
261 ([], StepState sr, [], [])
262 IDone br -> IPartial $ TeePair ([], StepState sl, [], [])
263 ([], StepResult br, [], [])
264 IError err -> IError err
265 IDone bl ->
266 case resR of
267 IPartial sr -> IDone . zf bl <$> extractR sr
268 IDone br -> return $ IDone $ zf bl br
269 IError err -> return $ IError err
270 IError err -> return $ IError err
272 {-# INLINE consume #-}
273 consume buf inp1 inp2 stp st y = do
274 let (x, inp11, inp21) =
275 case inp1 of
276 [] -> (y, [], [])
277 z : [] -> (z, reverse (x:inp2), [])
278 z : zs -> (z, zs, x:inp2)
279 r <- stp st x
280 let buf1 = x:buf
281 return (buf1, r, inp11, inp21)
283 -- consume one input item and return the next state of the fold
284 {-# INLINE useStream #-}
285 useStream buf inp1 inp2 stp st y = do
286 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
287 case r of
288 Partial 0 s ->
289 let state = ([], StepState s, inp11, inp21)
290 in return (state, Yld 0)
291 Partial n _ -> return (undefined, Yld n) -- Not implemented
292 Done n b ->
293 let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
294 in assert (n <= length buf1) (return (state, Stp n))
295 -- Continue 0 s -> (buf1, Right s, inp11, inp21)
296 Continue n s ->
297 let (src0, buf2) = splitAt n buf1
298 src = Prelude.reverse src0
299 state = (buf2, StepState s, src ++ inp11, inp21)
300 in assert (n <= length buf1) (return (state, Skp))
301 Error err -> return (undefined, Err err)
303 {-# INLINE_LATE step #-}
304 step (TeePair (bufL, StepState sL, inpL1, inpL2)
305 (bufR, StepState sR, inpR1, inpR2)) x = do
306 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
307 (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
308 let next = TeePair l r
309 case (stL,stR) of
310 -- XXX what if the first parser returns an unused count which is
311 -- more than the second parser's unused count? It does not make
312 -- sense for the second parser to consume more than the first
313 -- parser. We reset the input cursor based on the first parser.
314 -- Error out if the second one has consumed more then the first?
315 (Stp n1, Stp _) ->
316 -- Uni-pattern match results in better optimized code compared
317 -- to a case match.
318 let (_, StepResult rL, _, _) = l
319 (_, StepResult rR, _, _) = r
320 in return $ Done n1 (zf rL rR)
321 (Stp n1, Yld _) ->
322 let (_, StepResult rL, _, _) = l
323 (_, StepState ssR, _, _) = r
324 in do
325 rR <- extractR ssR
326 return $ Done n1 (zf rL rR)
327 (Yld n1, Yld n2) -> return $ Partial (min n1 n2) next
328 (Yld n1, Stp n2) -> return $ Partial (min n1 n2) next
329 (Err err, _) -> return $ Error err
330 (_, Err err) -> return $ Error err
331 _ -> return $ Continue 0 next
333 step (TeePair (bufL, StepState sL, inpL1, inpL2)
334 r@(_, StepResult rR, _, _)) x = do
335 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
336 let next = TeePair l r
337 -- XXX If the unused count of this stream is lower than the unused
338 -- count of the stopped stream, only then this will be correct. We need
339 -- to fix the other case. We need to keep incrementing the unused count
340 -- of the stopped stream and take the min of the two.
341 return $ case stL of
342 Yld n -> Partial n next
343 Stp n ->
344 let (_, StepResult rL, _, _) = l
345 in Done n (zf rL rR)
346 Skp -> Continue 0 next
347 Err err -> Error err
349 step _ _ = undefined
351 {-# INLINE_LATE extract #-}
352 extract st =
353 case st of
354 TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
355 rL <- extractL sL
356 rR <- extractR sR
357 return $ zf rL rR
358 TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
359 rL <- extractL sL
360 return $ zf rL rR
361 _ -> error "unreachable"
363-- | See 'Streamly.Internal.Data.Parser.teeWithMin'.
364--
365-- /Unimplemented/
366--
367{-# INLINE teeWithMin #-}
368teeWithMin ::
369 -- Monad m =>
370 (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
371teeWithMin = undefined
373-------------------------------------------------------------------------------
374-- Distribute input to two parsers and choose one result
375-------------------------------------------------------------------------------
377-- | See 'Streamly.Internal.Data.Parser.shortest'.
378--
379-- /Broken/
380--
381{-# INLINE shortest #-}
382shortest :: Monad m => Parser m x a -> Parser m x a -> Parser m x a
383shortest (Parser stepL initialL extractL) (Parser stepR initialR _) =
384 Parser step initial extract
386 where
388 {-# INLINE_LATE initial #-}
389 initial = do
390 resL <- initialL
391 resR <- initialR
392 return $ case resL of
393 IPartial sl ->
394 case resR of
395 IPartial sr -> IPartial $ TeePair ([], StepState sl, [], [])
396 ([], StepState sr, [], [])
397 IDone br -> IDone br
398 IError err -> IError err
399 IDone bl -> IDone bl
400 IError errL ->
401 case resR of
402 IPartial _ -> IError errL
403 IDone br -> IDone br
404 IError errR -> IError errR
406 {-# INLINE consume #-}
407 consume buf inp1 inp2 stp st y = do
408 let (x, inp11, inp21) =
409 case inp1 of
410 [] -> (y, [], [])
411 z : [] -> (z, reverse (x:inp2), [])
412 z : zs -> (z, zs, x:inp2)
413 r <- stp st x
414 let buf1 = x:buf
415 return (buf1, r, inp11, inp21)
417 -- consume one input item and return the next state of the fold
418 {-# INLINE useStream #-}
419 useStream buf inp1 inp2 stp st y = do
420 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
421 case r of
422 Partial 0 s ->
423 let state = ([], StepState s, inp11, inp21)
424 in return (state, Yld 0)
425 Partial n _ -> return (undefined, Yld n) -- Not implemented
426 Done n b ->
427 let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
428 in assert (n <= length buf1) (return (state, Stp n))
429 -- Continue 0 s -> (buf1, Right s, inp11, inp21)
430 Continue n s ->
431 let (src0, buf2) = splitAt n buf1
432 src = Prelude.reverse src0
433 state = (buf2, StepState s, src ++ inp11, inp21)
434 in assert (n <= length buf1) (return (state, Skp))
435 Error err -> return (undefined, Err err)
437 -- XXX Even if a parse finished earlier it may not be shortest if the other
438 -- parser finishes later but returns a lot of unconsumed input. Our current
439 -- criterion of shortest is whichever parse decided to stop earlier.
440 {-# INLINE_LATE step #-}
441 step (TeePair (bufL, StepState sL, inpL1, inpL2)
442 (bufR, StepState sR, inpR1, inpR2)) x = do
443 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
444 (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
445 let next = TeePair l r
446 return $ case (stL,stR) of
447 (Stp n1, _) ->
448 let (_, StepResult rL, _, _) = l
449 in Done n1 rL
450 (_, Stp n2) ->
451 let (_, StepResult rR, _, _) = r
452 in Done n2 rR
453 (Yld n1, Yld n2) -> Partial (min n1 n2) next
454 (Err err, _) -> Error err
455 (_, Err err) -> Error err
456 _ -> Continue 0 next
458 step _ _ = undefined
460 {-# INLINE_LATE extract #-}
461 extract st =
462 case st of
463 TeePair (_, StepState sL, _, _) _ -> extractL sL
464 _ -> error "unreachable"
466-- | See 'Streamly.Internal.Data.Parser.longest'.
467--
468-- /Broken/
469--
470{-# INLINE longest #-}
471longest :: MonadCatch m => Parser m x a -> Parser m x a -> Parser m x a
472longest (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
473 Parser step initial extract
475 where
478 {-# INLINE_LATE initial #-}
479 initial = do
480 resL <- initialL
481 resR <- initialR
482 return $ case resL of
483 IPartial sl ->
484 case resR of
485 IPartial sr -> IPartial $ TeePair ([], StepState sl, [], [])
486 ([], StepState sr, [], [])
487 IDone br -> IPartial $ TeePair ([], StepState sl, [], [])
488 ([], StepResult br, [], [])
489 IError _ ->
490 IPartial $ TeePair ([], StepState sl, [], [])
491 ([], StepResult undefined, [], [])
492 IDone bl ->
493 case resR of
494 IPartial sr ->
495 IPartial $ TeePair ([], StepResult bl, [], [])
496 ([], StepState sr, [], [])
497 IDone _ -> IDone bl
498 IError _ -> IDone bl
499 IError _ ->
500 case resR of
501 IPartial sr ->
502 IPartial $ TeePair ([], StepResult undefined, [], [])
503 ([], StepState sr, [], [])
504 IDone br -> IDone br
505 IError err -> IError err
507 {-# INLINE consume #-}
508 consume buf inp1 inp2 stp st y = do
509 let (x, inp11, inp21) =
510 case inp1 of
511 [] -> (y, [], [])
512 z : [] -> (z, reverse (x:inp2), [])
513 z : zs -> (z, zs, x:inp2)
514 r <- stp st x
515 let buf1 = x:buf
516 return (buf1, r, inp11, inp21)
518 -- consume one input item and return the next state of the fold
519 {-# INLINE useStream #-}
520 useStream buf inp1 inp2 stp st y = do
521 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
522 case r of
523 Partial 0 s ->
524 let state = ([], StepState s, inp11, inp21)
525 in return (state, Yld 0)
526 Partial n _ -> return (undefined, Yld n) -- Not implemented
527 Done n b ->
528 let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
529 in assert (n <= length buf1) (return (state, Stp n))
530 -- Continue 0 s -> (buf1, Right s, inp11, inp21)
531 Continue n s ->
532 let (src0, buf2) = splitAt n buf1
533 src = Prelude.reverse src0
534 state = (buf2, StepState s, src ++ inp11, inp21)
535 in assert (n <= length buf1) (return (state, Skp))
536 Error err -> return (undefined, Err err)
538 {-# INLINE_LATE step #-}
539 step (TeePair (bufL, StepState sL, inpL1, inpL2)
540 (bufR, StepState sR, inpR1, inpR2)) x = do
541 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
542 (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
543 let next = TeePair l r
544 return $ case (stL,stR) of
545 (Yld n1, Yld n2) -> Partial (min n1 n2) next
546 (Yld n1, Stp n2) -> Partial (min n1 n2) next
547 (Stp n1, Yld n2) -> Partial (min n1 n2) next
548 (Stp n1, Stp n2) ->
549 let (_, StepResult rL, _, _) = l
550 (_, StepResult rR, _, _) = r
551 in Done (max n1 n2) (if n1 >= n2 then rL else rR)
552 (Err err, _) -> Error err
553 (_, Err err) -> Error err
554 _ -> Continue 0 next
556 -- XXX the parser that finishes last may not be the longest because it may
557 -- return a lot of unused input which makes it shorter. Our current
558 -- criterion of deciding longest is based on whoever decides to finish
559 -- last and not whoever consumed more input.
560 --
561 -- To actually know who made more progress we need to keep an account of
562 -- how many items are unconsumed since the last yield.
563 --
564 step (TeePair (bufL, StepState sL, inpL1, inpL2)
565 r@(_, StepResult _, _, _)) x = do
566 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
567 let next = TeePair l r
568 return $ case stL of
569 Yld n -> Partial n next
570 Stp n ->
571 let (_, StepResult rL, _, _) = l
572 in Done n rL
573 Skp -> Continue 0 next
574 Err err -> Error err
576 step (TeePair l@(_, StepResult _, _, _)
577 (bufR, StepState sR, inpR1, inpR2)) x = do
578 (r, stR) <- useStream bufR inpR1 inpR2 stepR sR x
579 let next = TeePair l r
580 return $ case stR of
581 Yld n -> Partial n next
582 Stp n ->
583 let (_, StepResult rR, _, _) = r
584 in Done n rR
585 Skp -> Continue 0 next
586 Err err -> Error err
588 step _ _ = undefined
590 {-# INLINE_LATE extract #-}
591 extract st =
592 -- XXX When results are partial we may not be able to precisely compare
593 -- which parser has made more progress till now. One way to do that is
594 -- to figure out the actually consumed input up to the last yield.
595 --
596 case st of
597 TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
598 r <- try $ extractL sL
599 case r of
600 Left (_ :: ParseError) -> extractR sR
601 Right b -> return b
602 TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
603 r <- try $ extractL sL
604 case r of
605 Left (_ :: ParseError) -> return rR
606 Right b -> return b
607 TeePair (_, StepResult rL, _, _) (_, StepState sR, _, _) -> do
608 r <- try $ extractR sR
609 case r of
610 Left (_ :: ParseError) -> return rL
611 Right b -> return b
612 TeePair (_, StepResult _, _, _) (_, StepResult _, _, _) ->
613 error "unreachable"