git.iko.soy/streamly
on f67df51 help sign in
streamly/src/Streamly/Internal/Data/Zipper.hs
358 lines 13 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE ExistentialQuantification #-}
3{-# LANGUAGE BangPatterns #-}
4{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE ScopedTypeVariables #-}
7#include "inline.hs"
9-- |
10-- Module : Streamly.Internal.Data.Zipper
11-- Copyright : (c) 2020 Composewell Technologies
12-- License : BSD-3-Clause
13-- Maintainer : streamly@composewell.com
14-- Stability : experimental
15-- Portability : GHC
16--
17-- A special zipper derived from a stream to facilitate stream parsing.
18--
19-- = Zipper
20--
21-- The zipper is designed to enable moving backward and forward in the stream
22-- efficiently to be able to implement backtracking parsers. The zipper can
23-- buffer data until the parser indicates that it is not needed, we can move
24-- back in the buffered data if the parser needs to backtrack.
25--
26-- = Checkpointing
27--
28-- We have a checkpointing mechanism built into the zipper to handle
29-- backtracking. Whenever we enter a computation that can fail and has an
30-- alternative to run then we add a checkpoint to the zipper. The checkpoint
31-- ensures that we do not release the buffer upto the checkpoint until this
32-- computation succeeds, if the computation fails we can backtrack to focus the
33-- cursor back to the checkpoint position.
34--
35-- = Future Work
36--
37-- We can possibly have a typeclass providing a zipper interface so that we can
38-- derive a zipper from different data sources efficiently. For example,
39--
40-- * stream zipper
41-- * zippers from different types of arrays
42-- * zipper from a tree of arrays for dynamic in-memory storage
43-- * file handle zipper can utilize the seek to go back and forth in the file
44--
45-- A tree of zippers where a lower level zipper is dependent on (references
46-- data in) a higher level zipper could be useful in implementing distributive
47-- parsers.
48--
49-- See https://hackage.haskell.org/package/zipper for an existing type class.
50-- We need a monadic one.
52module Streamly.Internal.Data.Zipper
53 (
54 Zipper (..)
56 -- * Construction
57 , nil
58 , fromStream
59 , fromList
61 -- * Checkpointing
62 , checkpoint
63 , release
64 , restore
66 -- * Parsing
67 , parse
68 )
69where
71import Control.Exception (assert, Exception(..))
72import Control.Monad.Catch (MonadCatch, try)
73-- import GHC.Types (SPEC(..))
74import Prelude hiding (splitAt)
76import Streamly.Internal.Data.Stream.StreamK.Type (Stream(..))
77import Streamly.Internal.Data.SVar (defState)
79import qualified Streamly.Internal.Data.Parser.ParserD.Types as PR
80import qualified Streamly.Internal.Data.Stream.StreamK as K
82-------------------------------------------------------------------------------
83-- Zipper type
84-------------------------------------------------------------------------------
86-- | @Zipper checkpoints lefts rights tail@. The focus is on the first element
87-- of @rights@. @lefts@ is buffered data on the right of the cursor. Note that
88-- @lefts@ is stored as a reversed list, this helps is adding more items to the
89-- list quickly. @tail@ is a stream that is used to generate more data if the
90-- cursor moves past @rights@.
91--
92-- @checkpoints@ is a stack of checkpoints. A new checkpoint is created by a
93-- @checkpoint@ operation. A checkpoint consists of a count that tracks how
94-- many elements we have yielded after the checkpoint was created. We need
95-- this information to backtrack to the checkpoint. If we enter a nested
96-- alternative we add another checkpoint in the stack. When we exit an
97-- alternative we call a @release@ on the checkpoint. The @release@ removes the
98-- checkpoint from the stack and adds its element count to the previous
99-- checkpoint in the stack. When the last checkpoint is removed the buffer is
100-- released.
101--
102-- /Internal/
103--
104data Zipper m a = Zipper [Int] [a] [a] (Stream m a)
106-------------------------------------------------------------------------------
107-- Construction
108-------------------------------------------------------------------------------
110-- | Empty zipper.
111--
112-- /Internal/
113--
114{-# INLINE nil #-}
115nil :: Zipper m a
116nil = Zipper [] [] [] K.nil
118-- | Create a zipper from a stream.
119--
120-- /Internal/
121--
122{-# INLINE fromStream #-}
123fromStream :: Stream m a -> Zipper m a
124fromStream = Zipper [] [] []
126-- | Create a zipper from a list.
127--
128-- /Internal/
129--
130{-# INLINE fromList #-}
131fromList :: [a] -> Zipper m a
132fromList xs = Zipper [] [] xs K.nil
134-- | Add a checkpoint to the Zipper so that we do not release the buffer beyond
135-- the checkpoint.
136--
137-- /Internal/
138--
139{-# INLINE checkpoint #-}
140checkpoint :: Zipper m a -> Zipper m a
141checkpoint (Zipper cps xs ys stream) = Zipper (0:cps) xs ys stream
143-- | Release the latest checkpoint, releases any values held by the checkpoint.
144-- Note that the values may still be held by other checkpoints in the stack of
145-- checkpoints.
146--
147-- /Internal/
148--
149{-# INLINE release #-}
150release :: Zipper m a -> Zipper m a
151release (Zipper [] _ _ _) = error "Bug: release, no checkpoint exists!"
152release (Zipper (n:cps) xs ys stream) =
153 assert (n <= length xs) $
154 case cps of
155 [] -> assert (n == length xs) $ Zipper [] [] ys stream
156 cp:rest -> Zipper ((cp + n) : rest) xs ys stream
158-- | Rewind to restore the cursor to the latest checkpoint.
159--
160-- /Internal/
161--
162{-# INLINE restore #-}
163restore :: Zipper m a -> Zipper m a
164restore (Zipper [] _ _ _) = error "Bug: restore, no checkpoint exists!"
165restore (Zipper (n:cps) xs ys stream) =
166 assert (n <= length xs) $
167 let (src0, buf1) = splitAt n xs
168 src = Prelude.reverse src0
169 in Zipper cps buf1 (src ++ ys) stream
171-------------------------------------------------------------------------------
172-- Parse driver for a Zipper
173-------------------------------------------------------------------------------
175-- Inlined definition. Without the inline "serially/parser/take" benchmark
176-- degrades and splitParse does not fuse. Even using "inline" at the callsite
177-- does not help.
178{-# INLINE splitAt #-}
179splitAt :: Int -> [a] -> ([a],[a])
180splitAt n ls
181 | n <= 0 = ([], ls)
182 | otherwise = splitAt' n ls
183 where
184 splitAt' :: Int -> [a] -> ([a], [a])
185 splitAt' _ [] = ([], [])
186 splitAt' 1 (x:xs) = ([x], xs)
187 splitAt' m (x:xs) = (x:xs', xs'')
188 where
189 (xs', xs'') = splitAt' (m - 1) xs
191-- | Parse a stream zipper using a parser's @step@, @initial@ and @extract@
192-- functions.
193--
194{-# INLINE_NORMAL parse #-}
195parse
196 :: MonadCatch m
197 => (s -> a -> m (PR.Step s b))
198 -> m s
199 -> (s -> m b)
200 -> Zipper m a
201 -> m (Zipper m a, Either String b)
203-- The case when no checkpoints exist
204parse pstep initial extract (Zipper [] ls rs stream) =
205 case rs of
206 [] -> go stream ls initial
207 _ -> gobuf stream ls rs initial
209 where
211 {-# INLINE go #-}
212 go st buf !acc =
213 let stop = do
214 pst <- acc
215 r <- try $ extract pst
216 return $ case r of
217 Left (e :: PR.ParseError) ->
218 (nil, Left (displayException e))
219 Right b -> (nil, Right b)
220 single x = yieldk x K.nil
221 yieldk x r = do
222 acc1 <- acc >>= \b -> pstep b x
223 case acc1 of
224 -- PR.Yield 0 pst1 -> go SPEC s [] pst1
225 PR.Yield n pst1 -> do
226 assert (n <= length (x:buf)) (return ())
227 go r (Prelude.take n (x:buf)) (return pst1)
228 PR.YieldB n pst1 -> do
229 assert (n <= length (x:buf)) (return ())
230 let src0 = Prelude.take n (x:buf)
231 src = Prelude.reverse src0
232 gobuf r [] src (return pst1)
233 PR.Skip 0 pst1 -> go r (x:buf) (return pst1)
234 PR.Skip n pst1 -> do
235 assert (n <= length (x:buf)) (return ())
236 let (src0, buf1) = splitAt n (x:buf)
237 src = Prelude.reverse src0
238 gobuf r buf1 src (return pst1)
239 PR.Stop n b -> do
240 assert (n <= length (x:buf)) (return ())
241 let (src0, buf1) = splitAt n (x:buf)
242 src = Prelude.reverse src0
243 return (Zipper [] buf1 src r, Right b)
244 PR.Error err ->
245 return (Zipper [] (x:buf) [] r, Left err)
246 in K.foldStream defState yieldk single stop st
248 gobuf s buf [] !pst = go s buf pst
249 gobuf s buf (x:xs) !pst = do
250 r <- pst
251 pRes <- pstep r x
252 case pRes of
253 -- PR.Yield 0 pst1 -> go SPEC s [] pst1
254 PR.Yield n pst1 -> do
255 assert (n <= length (x:buf)) (return ())
256 gobuf s (Prelude.take n (x:buf)) xs (return pst1)
257 PR.YieldB n pst1 -> do
258 assert (n <= length (x:buf)) (return ())
259 let src0 = Prelude.take n (x:buf)
260 src = Prelude.reverse src0
261 gobuf s [] (src ++ xs) (return pst1)
262 PR.Skip 0 pst1 -> gobuf s (x:buf) xs (return pst1)
263 PR.Skip n pst1 -> do
264 assert (n <= length (x:buf)) (return ())
265 let (src0, buf1) = splitAt n (x:buf)
266 src = Prelude.reverse src0 ++ xs
267 gobuf s buf1 src (return pst1)
268 PR.Stop n b -> do
269 assert (n <= length (x:buf)) (return ())
270 let (src0, buf1) = splitAt n (x:buf)
271 src = Prelude.reverse src0 ++ xs
272 return (Zipper [] buf1 src s, Right b)
273 PR.Error err ->
274 return (Zipper [] (x:buf) xs s, Left err)
276-- The case when checkpoints exist
277-- XXX code duplication alert!
278parse pstep initial extract (Zipper (cp:cps) ls rs stream) =
279 case rs of
280 [] -> go 0 stream ls initial
281 _ -> gobuf 0 stream ls rs initial
283 where
285 {-# INLINE go #-}
286 go cnt st buf !acc =
287 let stop = do
288 pst <- acc
289 r <- try $ extract pst
290 return $ case r of
291 Left (e :: PR.ParseError) ->
292 (nil, Left (displayException e))
293 Right b -> (nil, Right b)
294 single x = yieldk x K.nil
295 yieldk x r = do
296 acc1 <- acc >>= \b -> pstep b x
297 let cnt1 = cnt + 1
298 case acc1 of
299 -- PR.Yield 0 pst1 -> go SPEC s [] pst1
300 PR.Yield n pst1 -> do
301 assert (n <= length (x:buf)) (return ())
302 go cnt1 r (x:buf) (return pst1)
303 PR.YieldB n pst1 -> do
304 assert (n <= length (x:buf)) (return ())
305 let src0 = Prelude.take n (x:buf)
306 src = Prelude.reverse src0
307 gobuf (cnt1 - n) r [] src (return pst1)
308 PR.Skip 0 pst1 -> go cnt1 r (x:buf) (return pst1)
309 PR.Skip n pst1 -> do
310 assert (n <= length (x:buf)) (return ())
311 let (src0, buf1) = splitAt n (x:buf)
312 src = Prelude.reverse src0
313 assert (cnt1 - n >= 0) (return ())
314 gobuf (cnt1 - n) r buf1 src (return pst1)
315 PR.Stop n b -> do
316 assert (n <= length (x:buf)) (return ())
317 let (src0, buf1) = splitAt n (x:buf)
318 src = Prelude.reverse src0
319 assert (cp + cnt1 - n >= 0) (return ())
320 return ( Zipper (cp + cnt1 - n : cps) buf1 src r
321 , Right b
322 )
323 PR.Error err ->
324 return ( Zipper (cp + cnt1 : cps) (x:buf) [] r
325 , Left err
326 )
327 in K.foldStream defState yieldk single stop st
329 gobuf cnt s buf [] !pst = go cnt s buf pst
330 gobuf cnt s buf (x:xs) !pst = do
331 r <- pst
332 pRes <- pstep r x
333 let cnt1 = cnt + 1
334 case pRes of
335 -- PR.Yield 0 pst1 -> go SPEC s [] pst1
336 PR.Yield n pst1 -> do
337 assert (n <= length (x:buf)) (return ())
338 gobuf cnt1 s (x:buf) xs (return pst1)
339 PR.YieldB n pst1 -> do
340 assert (n <= length (x:buf)) (return ())
341 let src0 = Prelude.take n (x:buf)
342 src = Prelude.reverse src0
343 gobuf (cnt1 - n) s [] (src ++ xs) (return pst1)
344 PR.Skip 0 pst1 -> gobuf cnt1 s (x:buf) xs (return pst1)
345 PR.Skip n pst1 -> do
346 assert (n <= length (x:buf)) (return ())
347 let (src0, buf1) = splitAt n (x:buf)
348 src = Prelude.reverse src0 ++ xs
349 assert (cnt1 - n >= 0) (return ())
350 gobuf (cnt1 - n) s buf1 src (return pst1)
351 PR.Stop n b -> do
352 assert (n <= length (x:buf)) (return ())
353 let (src0, buf1) = splitAt n (x:buf)
354 src = Prelude.reverse src0 ++ xs
355 assert (cp + cnt1 - n >= 0) (return ())
356 return (Zipper (cp + cnt1 - n : cps) buf1 src s, Right b)
357 PR.Error err ->
358 return (Zipper (cp + cnt1 : cps) (x:buf) xs s, Left err)