git.iko.soy/streamly
on da36a01 help sign in
streamly/core/src/Streamly/Internal/Data/Parser/ParserD/Tee.hs
617 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 {-
30 -- Parallel zipped
31 teeWith
32 , teeWithFst
33 , teeWithMin
35 -- Parallel alternatives
36 , shortest
37 , longest
38 -}
39 )
40where
42{-
43import Control.Exception (assert)
44import Control.Monad.Catch (MonadCatch, try)
45import Prelude
46 hiding (any, all, takeWhile)
48import Fusion.Plugin.Types (Fuse(..))
49import Streamly.Internal.Data.Parser.ParserD.Type
50 (Initial(..), Parser(..), Step(..), ParseError)
52-------------------------------------------------------------------------------
53-- Distribute input to two parsers and collect both results
54-------------------------------------------------------------------------------
56-- When the input stream is distributed to two parsers, both the parsers can
57-- backtrack independently. Therefore, we need separate buffer state for each
58-- parser.
59--
60-- ParserK
61--
62-- We can keep the state of each parser in the zipper and pass around that
63-- zipper to the parsers. Each parser can consume from the zipper and then pass
64-- around the zipper to the other parser.
65--
66-- ParserD
67--
68-- In the approach we have taken here, the driver pushes one element at a time
69-- to the tee and each of the parsers in the tee may buffer it independently
70-- for backtracking. So they do not need to depend on the original stream
71-- source for individual parser backtracking. Problem arises when both the
72-- parsers backtrack and they do not need any input from the driver rather they
73-- must consume from their buffers. For such situation we may need a
74-- "Continue" style driver command from the tee so that the driver runs
75-- the tee without providing it any input. Or we may need a local driver loop
76-- until new input is to be demanded from the input stream.
77--
78-- When the tee errors out or stops, the tee driver may have to backtrack by
79-- the specified amount (or the tee must return the leftover input). Therefore,
80-- the tee driver also has to buffer, this leads to triple buffering.
81--
82-- When the tee stops we need to determine the backtracking amount from the
83-- leftover of both the parsers. Since both the parsers may have consumed
84-- different lengths of the stream we consider the maximum of the two as
85-- consumed.
86--
87 -- XXX We can use Initial instead of StepState
88{-# ANN type StepState Fuse #-}
89data StepState s a = StepState s | StepResult a
91-- | State of the pair of parsers in a tee composition
92-- Note: strictness annotation is important for fusing the constructors
93{-# ANN type TeeState Fuse #-}
94data TeeState sL sR x a b =
95-- @TeePair (past buffer, parser state, future-buffer1, future-buffer2) ...@
96 TeePair !([x], StepState sL a, [x], [x]) !([x], StepState sR b, [x], [x])
98{-# ANN type Res Fuse #-}
99data Res = Yld Int | Stp Int | Skp | Err String
101-- | See 'Streamly.Internal.Data.Parser.teeWith'.
102--
103-- /Broken/
104--
105{-# INLINE teeWith #-}
106teeWith :: Monad m
107 => (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
108teeWith zf (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
109 Parser step initial extract
111 where
113 {-# INLINE_LATE initial #-}
114 initial = do
115 resL <- initialL
116 resR <- initialR
117 return $ case resL of
118 IPartial sl ->
119 case resR of
120 IPartial sr -> IPartial $ TeePair ([], StepState sl, [], [])
121 ([], StepState sr, [], [])
122 IDone br -> IPartial $ TeePair ([], StepState sl, [], [])
123 ([], StepResult br, [], [])
124 IError err -> IError err
125 IDone bl ->
126 case resR of
127 IPartial sr ->
128 IPartial $ TeePair ([], StepResult bl, [], [])
129 ([], StepState sr, [], [])
130 IDone br -> IDone $ zf bl br
131 IError err -> IError err
132 IError err -> IError err
134 {-# INLINE consume #-}
135 consume buf inp1 inp2 stp st y = do
136 let (x, inp11, inp21) =
137 case inp1 of
138 [] -> (y, [], [])
139 z : [] -> (z, reverse (x:inp2), [])
140 z : zs -> (z, zs, x:inp2)
141 r <- stp st x
142 let buf1 = x:buf
143 return (buf1, r, inp11, inp21)
145 -- XXX This is currently broken, even though both the parsers need to
146 -- consume from their buffers after backtracking the driver would still be
147 -- pushing more input to the buffers.
148 --
149 -- consume one input item and return the next state of the fold
150 {-# INLINE useStream #-}
151 useStream buf inp1 inp2 stp st y = do
152 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
153 case r of
154 Partial 0 s ->
155 let state = ([], StepState s, inp11, inp21)
156 in return (state, Yld 0)
157 Partial n s ->
158 let src0 = Prelude.take n buf1
159 src = Prelude.reverse src0
160 state = ([], StepState s, src ++ inp11, inp21)
161 in assert (n <= length buf1) (return (state, Yld n))
162 Done n b ->
163 let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
164 in assert (n <= length buf1) (return (state, Stp n))
165 -- Continue 0 s -> (buf1, Right s, inp11, inp21)
166 Continue n s ->
167 let (src0, buf2) = splitAt n buf1
168 src = Prelude.reverse src0
169 state = (buf2, StepState s, src ++ inp11, inp21)
170 in assert (n <= length buf1) (return (state, Skp))
171 Error err -> return (undefined, Err err)
173 {-# INLINE_LATE step #-}
174 step (TeePair (bufL, StepState sL, inpL1, inpL2)
175 (bufR, StepState sR, inpR1, inpR2)) x = do
176 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
177 (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
178 let next = TeePair l r
179 return $ case (stL,stR) of
180 (Yld n1, Yld n2) -> Partial (min n1 n2) next
181 (Yld n1, Stp n2) -> Partial (min n1 n2) next
182 (Stp n1, Yld n2) -> Partial (min n1 n2) next
183 (Stp n1, Stp n2) ->
184 -- Uni-pattern match results in better optimized code compared
185 -- to a case match.
186 let (_, StepResult rL, _, _) = l
187 (_, StepResult rR, _, _) = r
188 in Done (min n1 n2) (zf rL rR)
189 (Err err, _) -> Error err
190 (_, Err err) -> Error err
191 _ -> Continue 0 next
193 step (TeePair (bufL, StepState sL, inpL1, inpL2)
194 r@(_, StepResult rR, _, _)) x = do
195 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
196 let next = TeePair l r
197 -- XXX If the unused count of this stream is lower than the unused
198 -- count of the stopped stream, only then this will be correct. We need
199 -- to fix the other case. We need to keep incrementing the unused count
200 -- of the stopped stream and take the min of the two.
201 return $ case stL of
202 Yld n -> Partial n next
203 Stp n ->
204 let (_, StepResult rL, _, _) = l
205 in Done n (zf rL rR)
206 Skp -> Continue 0 next
207 Err err -> Error err
209 step (TeePair l@(_, StepResult rL, _, _)
210 (bufR, StepState sR, inpR1, inpR2)) x = do
211 (r, stR) <- useStream bufR inpR1 inpR2 stepR sR x
212 let next = TeePair l r
213 -- XXX If the unused count of this stream is lower than the unused
214 -- count of the stopped stream, only then this will be correct. We need
215 -- to fix the other case. We need to keep incrementing the unused count
216 -- of the stopped stream and take the min of the two.
217 return $ case stR of
218 Yld n -> Partial n next
219 Stp n ->
220 let (_, StepResult rR, _, _) = r
221 in Done n (zf rL rR)
222 Skp -> Continue 0 next
223 Err err -> Error err
225 step _ _ = undefined
227 {-# INLINE_LATE extract #-}
228 extract st =
229 case st of
230 TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
231 rL <- extractL sL
232 rR <- extractR sR
233 return $ zf rL rR
234 TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
235 rL <- extractL sL
236 return $ zf rL rR
237 TeePair (_, StepResult rL, _, _) (_, StepState sR, _, _) -> do
238 rR <- extractR sR
239 return $ zf rL rR
240 TeePair (_, StepResult rL, _, _) (_, StepResult rR, _, _) ->
241 return $ zf rL rR
243-- | See 'Streamly.Internal.Data.Parser.teeWithFst'.
244--
245-- /Broken/
246--
247{-# INLINE teeWithFst #-}
248teeWithFst :: Monad m
249 => (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
250teeWithFst zf (Parser stepL initialL extractL)
251 (Parser stepR initialR extractR) =
252 Parser step initial extract
254 where
256 {-# INLINE_LATE initial #-}
257 initial = do
258 resL <- initialL
259 resR <- initialR
260 case resL of
261 IPartial sl ->
262 return $ case resR of
263 IPartial sr -> IPartial $ TeePair ([], StepState sl, [], [])
264 ([], StepState sr, [], [])
265 IDone br -> IPartial $ TeePair ([], StepState sl, [], [])
266 ([], StepResult br, [], [])
267 IError err -> IError err
268 IDone bl ->
269 case resR of
270 IPartial sr -> IDone . zf bl <$> extractR sr
271 IDone br -> return $ IDone $ zf bl br
272 IError err -> return $ IError err
273 IError err -> return $ IError err
275 {-# INLINE consume #-}
276 consume buf inp1 inp2 stp st y = do
277 let (x, inp11, inp21) =
278 case inp1 of
279 [] -> (y, [], [])
280 z : [] -> (z, reverse (x:inp2), [])
281 z : zs -> (z, zs, x:inp2)
282 r <- stp st x
283 let buf1 = x:buf
284 return (buf1, r, inp11, inp21)
286 -- consume one input item and return the next state of the fold
287 {-# INLINE useStream #-}
288 useStream buf inp1 inp2 stp st y = do
289 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
290 case r of
291 Partial 0 s ->
292 let state = ([], StepState s, inp11, inp21)
293 in return (state, Yld 0)
294 Partial n _ -> return (undefined, Yld n) -- Not implemented
295 Done n b ->
296 let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
297 in assert (n <= length buf1) (return (state, Stp n))
298 -- Continue 0 s -> (buf1, Right s, inp11, inp21)
299 Continue n s ->
300 let (src0, buf2) = splitAt n buf1
301 src = Prelude.reverse src0
302 state = (buf2, StepState s, src ++ inp11, inp21)
303 in assert (n <= length buf1) (return (state, Skp))
304 Error err -> return (undefined, Err err)
306 {-# INLINE_LATE step #-}
307 step (TeePair (bufL, StepState sL, inpL1, inpL2)
308 (bufR, StepState sR, inpR1, inpR2)) x = do
309 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
310 (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
311 let next = TeePair l r
312 case (stL,stR) of
313 -- XXX what if the first parser returns an unused count which is
314 -- more than the second parser's unused count? It does not make
315 -- sense for the second parser to consume more than the first
316 -- parser. We reset the input cursor based on the first parser.
317 -- Error out if the second one has consumed more then the first?
318 (Stp n1, Stp _) ->
319 -- Uni-pattern match results in better optimized code compared
320 -- to a case match.
321 let (_, StepResult rL, _, _) = l
322 (_, StepResult rR, _, _) = r
323 in return $ Done n1 (zf rL rR)
324 (Stp n1, Yld _) ->
325 let (_, StepResult rL, _, _) = l
326 (_, StepState ssR, _, _) = r
327 in do
328 rR <- extractR ssR
329 return $ Done n1 (zf rL rR)
330 (Yld n1, Yld n2) -> return $ Partial (min n1 n2) next
331 (Yld n1, Stp n2) -> return $ Partial (min n1 n2) next
332 (Err err, _) -> return $ Error err
333 (_, Err err) -> return $ Error err
334 _ -> return $ Continue 0 next
336 step (TeePair (bufL, StepState sL, inpL1, inpL2)
337 r@(_, StepResult rR, _, _)) x = do
338 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
339 let next = TeePair l r
340 -- XXX If the unused count of this stream is lower than the unused
341 -- count of the stopped stream, only then this will be correct. We need
342 -- to fix the other case. We need to keep incrementing the unused count
343 -- of the stopped stream and take the min of the two.
344 return $ case stL of
345 Yld n -> Partial n next
346 Stp n ->
347 let (_, StepResult rL, _, _) = l
348 in Done n (zf rL rR)
349 Skp -> Continue 0 next
350 Err err -> Error err
352 step _ _ = undefined
354 {-# INLINE_LATE extract #-}
355 extract st =
356 case st of
357 TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
358 rL <- extractL sL
359 rR <- extractR sR
360 return $ zf rL rR
361 TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
362 rL <- extractL sL
363 return $ zf rL rR
364 _ -> error "unreachable"
366-- | See 'Streamly.Internal.Data.Parser.teeWithMin'.
367--
368-- /Unimplemented/
369--
370{-# INLINE teeWithMin #-}
371teeWithMin ::
372 -- Monad m =>
373 (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
374teeWithMin = undefined
376-------------------------------------------------------------------------------
377-- Distribute input to two parsers and choose one result
378-------------------------------------------------------------------------------
380-- | See 'Streamly.Internal.Data.Parser.shortest'.
381--
382-- /Broken/
383--
384{-# INLINE shortest #-}
385shortest :: Monad m => Parser m x a -> Parser m x a -> Parser m x a
386shortest (Parser stepL initialL extractL) (Parser stepR initialR _) =
387 Parser step initial extract
389 where
391 {-# INLINE_LATE initial #-}
392 initial = do
393 resL <- initialL
394 resR <- initialR
395 return $ case resL of
396 IPartial sl ->
397 case resR of
398 IPartial sr -> IPartial $ TeePair ([], StepState sl, [], [])
399 ([], StepState sr, [], [])
400 IDone br -> IDone br
401 IError err -> IError err
402 IDone bl -> IDone bl
403 IError errL ->
404 case resR of
405 IPartial _ -> IError errL
406 IDone br -> IDone br
407 IError errR -> IError errR
409 {-# INLINE consume #-}
410 consume buf inp1 inp2 stp st y = do
411 let (x, inp11, inp21) =
412 case inp1 of
413 [] -> (y, [], [])
414 z : [] -> (z, reverse (x:inp2), [])
415 z : zs -> (z, zs, x:inp2)
416 r <- stp st x
417 let buf1 = x:buf
418 return (buf1, r, inp11, inp21)
420 -- consume one input item and return the next state of the fold
421 {-# INLINE useStream #-}
422 useStream buf inp1 inp2 stp st y = do
423 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
424 case r of
425 Partial 0 s ->
426 let state = ([], StepState s, inp11, inp21)
427 in return (state, Yld 0)
428 Partial n _ -> return (undefined, Yld n) -- Not implemented
429 Done n b ->
430 let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
431 in assert (n <= length buf1) (return (state, Stp n))
432 -- Continue 0 s -> (buf1, Right s, inp11, inp21)
433 Continue n s ->
434 let (src0, buf2) = splitAt n buf1
435 src = Prelude.reverse src0
436 state = (buf2, StepState s, src ++ inp11, inp21)
437 in assert (n <= length buf1) (return (state, Skp))
438 Error err -> return (undefined, Err err)
440 -- XXX Even if a parse finished earlier it may not be shortest if the other
441 -- parser finishes later but returns a lot of unconsumed input. Our current
442 -- criterion of shortest is whichever parse decided to stop earlier.
443 {-# INLINE_LATE step #-}
444 step (TeePair (bufL, StepState sL, inpL1, inpL2)
445 (bufR, StepState sR, inpR1, inpR2)) x = do
446 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
447 (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
448 let next = TeePair l r
449 return $ case (stL,stR) of
450 (Stp n1, _) ->
451 let (_, StepResult rL, _, _) = l
452 in Done n1 rL
453 (_, Stp n2) ->
454 let (_, StepResult rR, _, _) = r
455 in Done n2 rR
456 (Yld n1, Yld n2) -> Partial (min n1 n2) next
457 (Err err, _) -> Error err
458 (_, Err err) -> Error err
459 _ -> Continue 0 next
461 step _ _ = undefined
463 {-# INLINE_LATE extract #-}
464 extract st =
465 case st of
466 TeePair (_, StepState sL, _, _) _ -> extractL sL
467 _ -> error "unreachable"
469-- | See 'Streamly.Internal.Data.Parser.longest'.
470--
471-- /Broken/
472--
473{-# INLINE longest #-}
474longest :: MonadCatch m => Parser m x a -> Parser m x a -> Parser m x a
475longest (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
476 Parser step initial extract
478 where
481 {-# INLINE_LATE initial #-}
482 initial = do
483 resL <- initialL
484 resR <- initialR
485 return $ case resL of
486 IPartial sl ->
487 case resR of
488 IPartial sr -> IPartial $ TeePair ([], StepState sl, [], [])
489 ([], StepState sr, [], [])
490 IDone br -> IPartial $ TeePair ([], StepState sl, [], [])
491 ([], StepResult br, [], [])
492 IError _ ->
493 IPartial $ TeePair ([], StepState sl, [], [])
494 ([], StepResult undefined, [], [])
495 IDone bl ->
496 case resR of
497 IPartial sr ->
498 IPartial $ TeePair ([], StepResult bl, [], [])
499 ([], StepState sr, [], [])
500 IDone _ -> IDone bl
501 IError _ -> IDone bl
502 IError _ ->
503 case resR of
504 IPartial sr ->
505 IPartial $ TeePair ([], StepResult undefined, [], [])
506 ([], StepState sr, [], [])
507 IDone br -> IDone br
508 IError err -> IError err
510 {-# INLINE consume #-}
511 consume buf inp1 inp2 stp st y = do
512 let (x, inp11, inp21) =
513 case inp1 of
514 [] -> (y, [], [])
515 z : [] -> (z, reverse (x:inp2), [])
516 z : zs -> (z, zs, x:inp2)
517 r <- stp st x
518 let buf1 = x:buf
519 return (buf1, r, inp11, inp21)
521 -- consume one input item and return the next state of the fold
522 {-# INLINE useStream #-}
523 useStream buf inp1 inp2 stp st y = do
524 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
525 case r of
526 Partial 0 s ->
527 let state = ([], StepState s, inp11, inp21)
528 in return (state, Yld 0)
529 Partial n _ -> return (undefined, Yld n) -- Not implemented
530 Done n b ->
531 let state = (Prelude.take n buf1, StepResult b, inp11, inp21)
532 in assert (n <= length buf1) (return (state, Stp n))
533 -- Continue 0 s -> (buf1, Right s, inp11, inp21)
534 Continue n s ->
535 let (src0, buf2) = splitAt n buf1
536 src = Prelude.reverse src0
537 state = (buf2, StepState s, src ++ inp11, inp21)
538 in assert (n <= length buf1) (return (state, Skp))
539 Error err -> return (undefined, Err err)
541 {-# INLINE_LATE step #-}
542 step (TeePair (bufL, StepState sL, inpL1, inpL2)
543 (bufR, StepState sR, inpR1, inpR2)) x = do
544 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
545 (r,stR) <- useStream bufR inpR1 inpR2 stepR sR x
546 let next = TeePair l r
547 return $ case (stL,stR) of
548 (Yld n1, Yld n2) -> Partial (min n1 n2) next
549 (Yld n1, Stp n2) -> Partial (min n1 n2) next
550 (Stp n1, Yld n2) -> Partial (min n1 n2) next
551 (Stp n1, Stp n2) ->
552 let (_, StepResult rL, _, _) = l
553 (_, StepResult rR, _, _) = r
554 in Done (max n1 n2) (if n1 >= n2 then rL else rR)
555 (Err err, _) -> Error err
556 (_, Err err) -> Error err
557 _ -> Continue 0 next
559 -- XXX the parser that finishes last may not be the longest because it may
560 -- return a lot of unused input which makes it shorter. Our current
561 -- criterion of deciding longest is based on whoever decides to finish
562 -- last and not whoever consumed more input.
563 --
564 -- To actually know who made more progress we need to keep an account of
565 -- how many items are unconsumed since the last yield.
566 --
567 step (TeePair (bufL, StepState sL, inpL1, inpL2)
568 r@(_, StepResult _, _, _)) x = do
569 (l,stL) <- useStream bufL inpL1 inpL2 stepL sL x
570 let next = TeePair l r
571 return $ case stL of
572 Yld n -> Partial n next
573 Stp n ->
574 let (_, StepResult rL, _, _) = l
575 in Done n rL
576 Skp -> Continue 0 next
577 Err err -> Error err
579 step (TeePair l@(_, StepResult _, _, _)
580 (bufR, StepState sR, inpR1, inpR2)) x = do
581 (r, stR) <- useStream bufR inpR1 inpR2 stepR sR x
582 let next = TeePair l r
583 return $ case stR of
584 Yld n -> Partial n next
585 Stp n ->
586 let (_, StepResult rR, _, _) = r
587 in Done n rR
588 Skp -> Continue 0 next
589 Err err -> Error err
591 step _ _ = undefined
593 {-# INLINE_LATE extract #-}
594 extract st =
595 -- XXX When results are partial we may not be able to precisely compare
596 -- which parser has made more progress till now. One way to do that is
597 -- to figure out the actually consumed input up to the last yield.
598 --
599 case st of
600 TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
601 r <- try $ extractL sL
602 case r of
603 Left (_ :: ParseError) -> extractR sR
604 Right b -> return b
605 TeePair (_, StepState sL, _, _) (_, StepResult rR, _, _) -> do
606 r <- try $ extractL sL
607 case r of
608 Left (_ :: ParseError) -> return rR
609 Right b -> return b
610 TeePair (_, StepResult rL, _, _) (_, StepState sR, _, _) -> do
611 r <- try $ extractR sR
612 case r of
613 Left (_ :: ParseError) -> return rL
614 Right b -> return b
615 TeePair (_, StepResult _, _, _) (_, StepResult _, _, _) ->
616 error "unreachable"
617-}