@@ -10,7 +10,7 @@
1010
1111module Streamly.Internal.Data.Parser.Chunked
1212 (
13− ParserChunked (..)
13+ ParserChunked (..) -- XXX rename to ChunkParser
1414 , fromParserD
1515 , parseBreak
1616 , K.fromPure
@@ -76,11 +76,10 @@ backTrack = go
7676
7777-- | A continuation to extract the result when a CPS parser is done.
7878{-# INLINE parserDone #-}
79−parserDone :: Monad m => (Int, Int) -> K.Parse b -> m (K.Step m a b)
80−parserDone (0,_) (K.Success n b) = return $ K.Done n b
81−parserDone st (K.Success _ _) =
82− error $ "Bug: fromParserK: inside alternative: " ++ show st
83−parserDone _ (K.Failure e) = return $ K.Error e
79+parserDone :: Monad m =>
80+ K.ParseResult b -> Int -> Maybe (Array a) -> m (K.Step a m b)
81+parserDone (K.Success n b) _ _ = return $ K.Done n b
82+parserDone (K.Failure n e) _ _ = return $ K.Error n e
8483
8584-- | Run a 'Parser' over a stream and return rest of the Stream.
8685{-# INLINE_NORMAL parseBreak #-}
@@ -90,47 +89,11 @@ parseBreak
9089 -> Stream m (Array a)
9190 -> m (b, Stream m (Array a))
9291parseBreak parser input = do
93− pRes <- K.runParser parser 0 (0,0) parserDone
94− case pRes of
95− K.Done n b -> assert (n == 0) (return (b, input))
96− K.Error e -> throwM (D.ParseError e)
97− K.Partial n parserk ->
98− assert (n == 0) (go [] parserk (Stream.toStreamK input))
99− K.Continue n parserk ->
100− assert (n == 0) (go [] parserk (Stream.toStreamK input))
92+ let parserk = \arr -> K.runParser parser 0 0 arr parserDone
93+ in go [] parserk (Stream.toStreamK input)
10194
10295 where
10396
104− -- This is a simplified yieldk
105− extractYieldK backBuf parserk arr stream = do
106− pRes <- parserk (Just arr)
107− case pRes of
108− K.Partial 0 cont1 ->
109− goExtract [] cont1 stream
110− K.Partial n cont1 -> do
111− assertM(n <= sum (map Array.length (arr:backBuf)))
112− goExtract [] cont1 (fst $ backTrack n (arr:backBuf) stream)
113− K.Continue 0 cont1 ->
114− go (arr:backBuf) cont1 stream
115− K.Continue n cont1 -> do
116− assertM(n <= sum (map Array.length (arr:backBuf)))
117− let (s1, backBuf1) = backTrack n (arr:backBuf) stream
118− in go backBuf1 cont1 s1
119− K.Done 0 b ->
120− return (b, Stream.fromStreamK stream)
121− K.Done n b -> do
122− assertM(n <= sum (map Array.length (arr:backBuf)))
123− let (s1, _) = backTrack n (arr:backBuf) stream
124− in return (b, Stream.fromStreamK s1)
125− K.Error err -> throwM $ D.ParseError err
126−
127− goExtract backBuf parserk stream = do
128− let stop = goStop backBuf parserk
129− single a = extractYieldK backBuf parserk a StreamK.nil
130− in StreamK.foldStream
131− defState (extractYieldK backBuf parserk) single stop stream
132−
133− -- This is a simplified goExtract
13497 {-# INLINE goStop #-}
13598 goStop backBuf parserk = do
13699 pRes <- parserk Nothing
@@ -138,49 +101,71 @@ parseBreak parser input = do
138101 -- If we stop in an alternative, it will try calling the next
139102 -- parser, the next parser may call initial returning Partial and
140103 -- then immediately we have to call extract on it.
141− K.Partial 0 cont1 -> do
142− goExtract [] cont1 StreamK.nil
104+ K.Partial 0 cont1 ->
105+ go [] cont1 StreamK.nil
143106 K.Partial n cont1 -> do
144− -- error $ "Bug: parseBreak: Partial in extract, n = " ++ show n
145− assertM(n <= sum (map Array.length backBuf))
146− let (s1, backBuf1) = backTrack n backBuf StreamK.nil
147− in goExtract backBuf1 cont1 s1
148− K.Continue 0 cont1 -> do
149− -- error "parseBreak: extract, Continue 0 creates infinite loop"
107+ let n1 = negate n
108+ assertM(n1 >= 0 && n1 <= sum (map Array.length backBuf))
109+ let (s1, backBuf1) = backTrack n1 backBuf StreamK.nil
110+ in go backBuf1 cont1 s1
111+ K.Continue 0 cont1 ->
150112 go backBuf cont1 StreamK.nil
151113 K.Continue n cont1 -> do
152− assertM(n <= sum (map Array.length backBuf))
153− let (s1, backBuf1) = backTrack n backBuf StreamK.nil
154− in goExtract backBuf1 cont1 s1
155− K.Done 0 b -> return (b, Stream.nil)
114+ let n1 = negate n
115+ assertM(n1 >= 0 && n1 <= sum (map Array.length backBuf))
116+ let (s1, backBuf1) = backTrack n1 backBuf StreamK.nil
117+ in go backBuf1 cont1 s1
118+ K.Done 0 b ->
119+ return (b, Stream.nil)
156120 K.Done n b -> do
157− assertM(n <= sum (map Array.length backBuf))
158− let (s1, _) = backTrack n backBuf StreamK.nil
121+ let n1 = negate n
122+ assertM(n1 >= 0 && n1 <= sum (map Array.length backBuf))
123+ let (s1, _) = backTrack n1 backBuf StreamK.nil
159124 in return (b, Stream.fromStreamK s1)
160− K.Error err -> throwM $ D.ParseError err
125+ K.Error _ err -> throwM $ D.ParseError err
161126
162− -- SPECIALIZE this on backBuf?
163− yieldk backBuf !parserk arr stream = do
127+ seekErr n len =
128+ error $ "parseBreak: Partial: forward seek not implemented n = "
129+ ++ show n ++ " len = " ++ show len
130+
131+ yieldk backBuf parserk arr stream = do
164132 pRes <- parserk (Just arr)
133+ let len = Array.length arr
165134 case pRes of
166− K.Partial 0 cont1 ->
167− go [] cont1 stream
168− K.Partial n cont1 -> do
169− assertM(n <= sum (map Array.length (arr:backBuf)))
170− go [] cont1 (fst $ backTrack n (arr:backBuf) stream)
171− K.Continue 0 cont1 ->
172− go (arr:backBuf) cont1 stream
173− K.Continue n cont1 -> do
174− assertM(n <= sum (map Array.length (arr:backBuf)))
175− let (s1, backBuf1) = backTrack n (arr:backBuf) stream
176− in go backBuf1 cont1 s1
177− K.Done 0 b ->
178− return (b, Stream.fromStreamK stream)
135+ K.Partial n cont1 ->
136+ case compare n len of
137+ EQ -> go [] cont1 stream
138+ LT -> do
139+ if n >= 0
140+ then yieldk [] cont1 arr stream
141+ else do
142+ let n1 = negate n
143+ bufLen = sum (map Array.length backBuf)
144+ s = StreamK.cons arr stream
145+ assertM(n1 >= 0 && n1 <= bufLen)
146+ let (s1, _) = backTrack n1 backBuf s
147+ go [] cont1 s1
148+ GT -> seekErr n len
149+ K.Continue n cont1 ->
150+ case compare n len of
151+ EQ -> go (arr:backBuf) cont1 stream
152+ LT -> do
153+ if n >= 0
154+ then yieldk backBuf cont1 arr stream
155+ else do
156+ let n1 = negate n
157+ bufLen = sum (map Array.length backBuf)
158+ s = StreamK.cons arr stream
159+ assertM(n1 >= 0 && n1 <= bufLen)
160+ let (s1, backBuf1) = backTrack n1 backBuf s
161+ go backBuf1 cont1 s1
162+ GT -> seekErr n len
179163 K.Done n b -> do
180− assertM(n <= sum (map Array.length (arr:backBuf)))
181− let (s1, _) = backTrack n (arr:backBuf) stream
164+ let n1 = len - n
165+ assertM(n1 <= sum (map Array.length (arr:backBuf)))
166+ let (s1, _) = backTrack n1 (arr:backBuf) stream
182167 in return (b, Stream.fromStreamK s1)
183− K.Error err -> throwM $ D.ParseError err
168+ K.Error _ err -> throwM $ D.ParseError err
184169
185170 go backBuf parserk stream = do
186171 let stop = goStop backBuf parserk
@@ -188,6 +173,12 @@ parseBreak parser input = do
188173 in StreamK.foldStream
189174 defState (yieldk backBuf parserk) single stop stream
190175
176+data ChunkResult s b =
177+ ChunkDone !Int !b
178+ | ChunkPartial !Int !s
179+ | ChunkContinue !Int !s
180+ | ChunkError !Int String
181+
191182-- This is very similar to fromParserD in the Array/Unboxed/Fold module.
192183{-# INLINE parseChunk #-}
193184parseChunk
@@ -195,122 +186,126 @@ parseChunk
195186 => (s -> a -> m (Step s b))
196187 -> s
197188 -> Array a
198− -> m (Step s b)
199−parseChunk pstep state (Array contents start end) = do
200− go SPEC start state
189+ -> Int
190+ -> m (ChunkResult s b)
191+parseChunk pstep state (Array contents start end) offset = do
192+ if offset >= 0
193+ then go SPEC (start + offset * SIZE_OF(a)) state
194+ else return $ ChunkContinue offset state
201195
202196 where
203197
204− {-# INLINE partial #-}
205− partial arrRem cur next elemSize st n fs1 = do
206− let next1 = next - (n * elemSize)
207− if next1 >= start && cur < end
208− then go SPEC next1 fs1
209− else return $ st (arrRem + n) fs1
198+ {-# INLINE onBack #-}
199+ onBack offset1 elemSize constr pst = do
200+ let pos = offset1 - start
201+ in if pos >= 0
202+ then go SPEC offset1 pst
203+ else return $ constr (pos `div` elemSize) pst
210204
211− go !_ !cur !pst | cur >= end = return $ Continue 0 pst
205+ -- Note: div may be expensive but the alternative is to maintain an element
206+ -- offset in addition to a byte offset or just the element offset and use
207+ -- multiplication to get the byte offset every time, both these options
208+ -- turned out to be more expensive than using div.
209+ go !_ !cur !pst | cur >= end =
210+ return $ ChunkContinue ((end - start) `div` SIZE_OF(a)) pst
212211 go !_ !cur !pst = do
213212 x <- liftIO $ peekWith contents cur
214213 pRes <- pstep pst x
215214 let elemSize = SIZE_OF(a)
216215 next = INDEX_NEXT(cur,a)
217− arrRem = (end - next) `div` elemSize
216+ back n = next - n * elemSize
217+ curOff = (cur - start) `div` elemSize
218+ nextOff = (next - start) `div` elemSize
218219 case pRes of
219− Done n b -> do
220− return $ Done (arrRem + n) b
220+ Done 0 b ->
221+ return $ ChunkDone nextOff b
222+ Done 1 b ->
223+ return $ ChunkDone curOff b
224+ Done n b ->
225+ return $ ChunkDone ((back n - start) `div` elemSize) b
226+ Partial 0 pst1 ->
227+ go SPEC next pst1
228+ Partial 1 pst1 ->
229+ go SPEC cur pst1
221230 Partial n pst1 ->
222− partial arrRem cur next elemSize Partial n pst1
223− Continue n pst1 -> do
224− partial arrRem cur next elemSize Continue n pst1
225− Error err -> return $ Error err
231+ onBack (back n) elemSize ChunkPartial pst1
232+ Continue 0 pst1 ->
233+ go SPEC next pst1
234+ Continue 1 pst1 ->
235+ go SPEC cur pst1
236+ Continue n pst1 ->
237+ onBack (back n) elemSize ChunkContinue pst1
238+ Error err ->
239+ return $ ChunkError curOff err
226240
227−{-# INLINE_NORMAL parseDToK #-}
241+{-# INLINE parseDToK #-}
228242parseDToK
229− :: (MonadIO m, Unbox a)
243+ :: forall m a s b r. (MonadIO m, Unbox a)
230244 => (s -> a -> m (Step s b))
231245 -> m (Initial s b)
232246 -> (s -> m (Step s b))
233247 -> Int
234− -> (Int, Int)
235− -> ((Int, Int) -> K.Parse b -> m (K.Step m a r))
236− -> m (K.Step m a r)
237−-- Non 'Alternative' case.
238−parseDToK pstep initial extract leftover (0, _) cont = do
248+ -> Int
249+ -> Maybe (Array a)
250+ -> (K.ParseResult b -> Int -> Maybe (Array a) -> m (K.Step a m r))
251+ -> m (K.Step a m r)
252+parseDToK pstep initial extract offset usedCount input cont = do
239253 res <- initial
240254 case res of
241− IPartial r -> return $ K.Partial leftover (parseCont (return r))
242− IDone b -> cont state (K.Success 0 b)
243− IError err -> cont state (K.Failure err)
255+ IPartial pst -> do
256+ case input of
257+ Just arr -> parseContJust usedCount offset pst arr
258+ Nothing -> parseContNothing usedCount pst
259+ IDone b -> cont (K.Success offset b) usedCount input
260+ IError err -> cont (K.Failure offset err) usedCount input
244261
245262 where
246263
247− -- The continuation is called with (0,0) state i.e. Alternative level
248− -- is 0 and the used count is 0. Alternative level is 0 because the level
249− -- passed in the argument above is 0, the "used" count is 0 because it is
250− -- not useful when Alternative level is 0. We should probably use a Maybe
251− -- type for the state but that might impact performance, need to measure.
252− state = (0,0)
253−
254− -- We could pass a stream or array here and drive the ParserD with fusion.
255− parseCont pst (Just arr) = do
256− r <- pst
257− pRes <- parseChunk pstep r arr
264+ -- XXX We can maintain an absolute position instead of relative that will
265+ -- help in reporting of error location in the stream.
266+ {-# NOINLINE parseContJust #-}
267+ parseContJust count off pst arr = do
268+ pRes <- parseChunk pstep pst arr off
269+ -- The "n" here is stream position index wrt the array start, and not
270+ -- the backtrack count as returned by byte stream parsers.
258271 case pRes of
259− Done n b -> cont state (K.Success n b)
260− Error err -> cont state (K.Failure err)
261− Partial n pst1 -> return $ K.Partial n (parseCont (return pst1))
262− Continue n pst1 -> return $ K.Continue n (parseCont (return pst1))
272+ ChunkDone n b ->
273+ assert (n <= Array.length arr)
274+ (cont (K.Success n b) (count + n - off) (Just arr))
275+ ChunkPartial n pst1 ->
276+ assert (n < 0 || n >= Array.length arr)
277+ (return $ K.Partial n (parseCont (count + n - off) pst1))
278+ ChunkContinue n pst1 ->
279+ assert (n < 0 || n >= Array.length arr)
280+ (return $ K.Continue n (parseCont (count + n - off) pst1))
281+ ChunkError n err ->
282+ cont (K.Failure n err) (count + n - off) (Just arr)
263283
264− parseCont acc Nothing = do
265− pst <- acc
284+ {-# NOINLINE parseContNothing #-}
285+ parseContNothing count pst = do
266286 r <- extract pst
267287 case r of
268− Done n b -> cont state (K.Success n b)
269− Error err -> cont state (K.Failure err)
270− Partial _ _ -> error "Bug: parseDToK Partial unreachable"
271− Continue n pst1 -> return $ K.Continue n (parseCont (return pst1))
272−
273−-- 'Alternative' case. Used count needs to be maintained when inside an
274−-- Alternative.
275−parseDToK pstep initial extract leftover (level, count) cont = do
276− res <- initial
277− case res of
278− IPartial r -> return $ K.Partial leftover (parseCont count (return r))
279− IDone b -> cont (level,count) (K.Success 0 b)
280− IError err -> cont (level,count) (K.Failure err)
281−
282− where
283−
284− parseCont !cnt pst (Just arr) = do
285− let !cnt1 = cnt + Array.length arr
286− r <- pst
287− pRes <- parseChunk pstep r arr
288− case pRes of
289− Done n b -> do
290− assertM(n <= cnt1)
291− cont (level, cnt1 - n) (K.Success n b)
288+ -- IMPORTANT: the n here is from the byte stream parser, that means
289+ -- it is the backtrack element count and not the stream position
290+ -- index into the current input array.
291+ Done n b ->
292+ assert (n >= 0)
293+ (cont (K.Success (- n) b) (count - n) Nothing)
294+ Continue n pst1 ->
295+ assert (n >= 0)
296+ (return $ K.Continue (- n) (parseCont (count - n) pst1))
292297 Error err ->
293− cont (level, cnt1) (K.Failure err)
294− Partial n pst1 -> do
295− assertM(n <= cnt1)
296− return $ K.Partial n (parseCont (cnt1 - n) (return pst1))
297− Continue n pst1 -> do
298− assertM(n <= cnt1)
299− return $ K.Continue n (parseCont (cnt1 - n) (return pst1))
300− parseCont cnt acc Nothing = do
301− pst <- acc
302− r <- extract pst
303− let s = (level, cnt)
304− case r of
305− Done n b -> do
306− assertM(n <= cnt)
307− cont s (K.Success n b)
298+ -- XXX It is called only when there is no input arr. So using 0
299+ -- as the position is correct?
300+ cont (K.Failure 0 err) count Nothing
308301 Partial _ _ -> error "Bug: parseDToK Partial unreachable"
309− Error e ->
310− cont s (K.Failure e)
311− Continue n pst1 -> do
312− assertM(n <= cnt)
313− return $ K.Continue n (parseCont (cnt - n) (return pst1))
302+
303+ -- XXX Maybe we can use two separate continuations instead of using
304+ -- Just/Nothing cases here. That may help in avoiding the parseContJust
305+ -- function call.
306+ {-# INLINE parseCont #-}
307+ parseCont cnt pst (Just arr) = parseContJust cnt 0 pst arr
308+ parseCont cnt pst Nothing = parseContNothing cnt pst
314309
315310-- | Convert a raw byte 'Parser' to a chunked parser.
316311--