git.iko.soy/streamly
on c3dcabc help sign in
streamly/src/Streamly/Internal/Data/Parser/ParserD/Tee.hs
548 lines 22 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.Types
47 (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{-# ANN type StepState Fuse #-}
85data StepState s a = StepState s | StepResult a
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 #-}
90data 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])
94{-# ANN type Res Fuse #-}
95data Res = Yld Int | Stp Int | Skp | Err String
97-- | See 'Streamly.Internal.Data.Parser.teeWith'.
98--
99-- /Broken/
100--
101{-# INLINE teeWith #-}
102teeWith :: Monad m
103 => (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
104teeWith zf (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
105 Parser step initial extract
107 where
109 {-# INLINE_LATE initial #-}
110 initial = do
111 sL <- initialL
112 sR <- initialR
113 return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])
115 {-# INLINE consume #-}
116 consume buf inp1 inp2 stp st y = do
117 let (x, inp11, inp21) =
118 case inp1 of
119 [] -> (y, [], [])
120 z : [] -> (z, reverse (x:inp2), [])
121 z : zs -> (z, zs, x:inp2)
122 r <- stp st x
123 let buf1 = x:buf
124 return (buf1, r, inp11, inp21)
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
133 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
134 case r of
135 Partial 0 s ->
136 let state = ([], StepState s, inp11, inp21)
137 in return (state, Yld 0)
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)
154 {-# INLINE_LATE step #-}
155 step (TeePair (bufL, StepState sL, inpL1, inpL2)
156 (bufR, StepState sR, inpR1, inpR2)) x = do
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
160 return $ case (stL,stR) of
161 (Yld n1, Yld n2) -> Partial (min n1 n2) next
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
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
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
206 step _ _ = undefined
208 {-# INLINE_LATE extract #-}
209 extract st =
210 case st of
211 TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
212 rL <- extractL sL
213 rR <- extractR sR
214 return $ zf rL rR
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
224-- | See 'Streamly.Internal.Data.Parser.teeWithFst'.
225--
226-- /Broken/
227--
228{-# INLINE teeWithFst #-}
229teeWithFst :: Monad m
230 => (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
231teeWithFst zf (Parser stepL initialL extractL)
232 (Parser stepR initialR extractR) =
233 Parser step initial extract
235 where
237 {-# INLINE_LATE initial #-}
238 initial = do
239 sL <- initialL
240 sR <- initialR
241 return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])
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)
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)
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
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
320 step _ _ = undefined
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"
334-- | See 'Streamly.Internal.Data.Parser.teeWithMin'.
335--
336-- /Unimplemented/
337--
338{-# INLINE teeWithMin #-}
339teeWithMin ::
340 -- Monad m =>
341 (a -> b -> c) -> Parser m x a -> Parser m x b -> Parser m x c
342teeWithMin = undefined
344-------------------------------------------------------------------------------
345-- Distribute input to two parsers and choose one result
346-------------------------------------------------------------------------------
348-- | See 'Streamly.Internal.Data.Parser.shortest'.
349--
350-- /Broken/
351--
352{-# INLINE shortest #-}
353shortest :: Monad m => Parser m x a -> Parser m x a -> Parser m x a
354shortest (Parser stepL initialL extractL) (Parser stepR initialR _) =
355 Parser step initial extract
357 where
359 {-# INLINE_LATE initial #-}
360 initial = do
361 sL <- initialL
362 sR <- initialR
363 return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])
365 {-# INLINE consume #-}
366 consume buf inp1 inp2 stp st y = do
367 let (x, inp11, inp21) =
368 case inp1 of
369 [] -> (y, [], [])
370 z : [] -> (z, reverse (x:inp2), [])
371 z : zs -> (z, zs, x:inp2)
372 r <- stp st x
373 let buf1 = x:buf
374 return (buf1, r, inp11, inp21)
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
379 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
380 case r of
381 Partial 0 s ->
382 let state = ([], StepState s, inp11, inp21)
383 in return (state, Yld 0)
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)
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)
401 (bufR, StepState sR, inpR1, inpR2)) x = do
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
405 return $ case (stL,stR) of
406 (Stp n1, _) ->
407 let (_, StepResult rL, _, _) = l
408 in Done n1 rL
409 (_, Stp n2) ->
410 let (_, StepResult rR, _, _) = r
411 in Done n2 rR
412 (Yld n1, Yld n2) -> Partial (min n1 n2) next
413 (Err err, _) -> Error err
414 (_, Err err) -> Error err
415 _ -> Continue 0 next
417 step _ _ = undefined
419 {-# INLINE_LATE extract #-}
420 extract st =
421 case st of
422 TeePair (_, StepState sL, _, _) _ -> extractL sL
423 _ -> error "unreachable"
425-- | See 'Streamly.Internal.Data.Parser.longest'.
426--
427-- /Broken/
428--
429{-# INLINE longest #-}
430longest :: MonadCatch m => Parser m x a -> Parser m x a -> Parser m x a
431longest (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
432 Parser step initial extract
434 where
436 {-# INLINE_LATE initial #-}
437 initial = do
438 sL <- initialL
439 sR <- initialR
440 return $ TeePair ([], StepState sL, [], []) ([], StepState sR, [], [])
442 {-# INLINE consume #-}
443 consume buf inp1 inp2 stp st y = do
444 let (x, inp11, inp21) =
445 case inp1 of
446 [] -> (y, [], [])
447 z : [] -> (z, reverse (x:inp2), [])
448 z : zs -> (z, zs, x:inp2)
449 r <- stp st x
450 let buf1 = x:buf
451 return (buf1, r, inp11, inp21)
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
456 (buf1, r, inp11, inp21) <- consume buf inp1 inp2 stp st y
457 case r of
458 Partial 0 s ->
459 let state = ([], StepState s, inp11, inp21)
460 in return (state, Yld 0)
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)
473 {-# INLINE_LATE step #-}
474 step (TeePair (bufL, StepState sL, inpL1, inpL2)
475 (bufR, StepState sR, inpR1, inpR2)) x = do
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
479 return $ case (stL,stR) of
480 (Yld n1, Yld n2) -> Partial (min n1 n2) next
481 (Yld n1, Stp n2) -> Partial (min n1 n2) next
482 (Stp n1, Yld n2) -> Partial (min n1 n2) next
483 (Stp n1, Stp n2) ->
484 let (_, StepResult rL, _, _) = l
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
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
511 step (TeePair l@(_, StepResult _, _, _)
512 (bufR, StepState sR, inpR1, inpR2)) x = do
513 (r, stR) <- useStream bufR inpR1 inpR2 stepR sR x
514 let next = TeePair l r
515 return $ case stR of
516 Yld n -> Partial n next
517 Stp n ->
518 let (_, StepResult rR, _, _) = r
519 in Done n rR
520 Skp -> Continue 0 next
521 Err err -> Error err
523 step _ _ = undefined
525 {-# INLINE_LATE extract #-}
526 extract st =
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
532 TeePair (_, StepState sL, _, _) (_, StepState sR, _, _) -> do
533 r <- try $ extractL sL
534 case r of
535 Left (_ :: ParseError) -> extractR sR
536 Right b -> return b
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
543 r <- try $ extractR sR
544 case r of
545 Left (_ :: ParseError) -> return rL
546 Right b -> return b
547 TeePair (_, StepResult _, _, _) (_, StepResult _, _, _) ->
548 error "unreachable"