@@ -0,0 +1,380 @@
2+{-# LANGUAGE ExistentialQuantification #-}
3+{-# LANGUAGE BangPatterns #-}
4+{-# LANGUAGE RankNTypes #-}
5+{-# LANGUAGE ScopedTypeVariables #-}
10+-- Module : Streamly.Internal.Data.Zipper
11+-- Copyright : (c) 2020 Composewell Technologies
13+-- Maintainer : streamly@composewell.com
14+-- Stability : experimental
17+-- A special zipper derived from a stream to facilitate stream parsing.
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.
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.
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,
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
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
49+-- See https://hackage.haskell.org/package/zipper for an existing type class.
50+-- We need a monadic one.
52+module Streamly.Internal.Data.Zipper
71+import Control.Exception (assert, Exception(..))
72+import Control.Monad.Catch (MonadCatch, try, SomeException)
73+-- import GHC.Types (SPEC(..))
74+import Prelude hiding (splitAt)
76+import Streamly.Internal.Data.Stream.StreamK.Type (Stream(..))
77+import Streamly.Internal.Data.SVar (defState)
79+import qualified Streamly.Internal.Data.Parser.Types as PR
80+import qualified Streamly.Internal.Data.Stream.StreamK as K
82+-------------------------------------------------------------------------------
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. @tail@
88+-- is a stream that is used to generate more data if the cursor moves past
91+-- @checkpoints@ is a stack of checkpoints. A new checkpoint is created by a
92+-- @checkpoint@ operation. A checkpoint consists of a count that tracks how
93+-- many elements we have yielded after the checkpoint was taken. We need this
94+-- information to backtrack to the checkpoint. If we enter a nested
95+-- alternative we add another checkpoint in the stack. When we exit an
96+-- alternative we call a @release@ on the checkpoint. The @release@ removes the
97+-- checkpoint from the stack and adds its element count to the previous
98+-- checkpoint in the stack. When the last checkpoint is removed the buffer is
103+data Zipper m a = Zipper [Int] [a] [a] (Stream m a)
105+-------------------------------------------------------------------------------
107+-------------------------------------------------------------------------------
115+nil = Zipper [] [] [] K.nil
117+-- | Create a zipper from a stream.
121+{-# INLINE fromStream #-}
122+fromStream :: Stream m a -> Zipper m a
123+fromStream = Zipper [] [] []
125+-- | Create a zipper from a list.
129+{-# INLINE fromList #-}
130+fromList :: [a] -> Zipper m a
131+fromList xs = Zipper [] [] xs K.nil
133+-- | Add a checkpoint to the Zipper so that we do not release the buffer beyond
138+{-# INLINE checkpoint #-}
139+checkpoint :: Zipper m a -> Zipper m a
140+checkpoint (Zipper cps xs ys stream) = Zipper (0:cps) xs ys stream
142+-- | Release the latest checkpoint, releases any values held by the checkpoint.
146+{-# INLINE release #-}
147+release :: Zipper m a -> Zipper m a
148+release (Zipper [] _ _ _) = error "Bug: release, no checkpoint exists!"
149+release (Zipper (n:cps) xs ys stream) =
150+ assert (n <= length xs) $
152+ [] -> assert (n == length xs) $ Zipper [] [] ys stream
153+ cp:rest -> Zipper ((cp + n) : rest) xs ys stream
155+-- | Rewind to restore the cursor to the latest checkpoint.
159+{-# INLINE restore #-}
160+restore :: Zipper m a -> Zipper m a
161+restore (Zipper [] _ _ _) = error "Bug: restore, no checkpoint exists!"
162+restore (Zipper (n:cps) xs ys stream) =
163+ assert (n <= length xs) $
164+ let (src0, buf1) = splitAt n xs
165+ src = Prelude.reverse src0
166+ in Zipper cps buf1 (src ++ ys) stream
168+-------------------------------------------------------------------------------
169+-- Parse driver for a Zipper
170+-------------------------------------------------------------------------------
172+-- Inlined definition. Without the inline "serially/parser/take" benchmark
173+-- degrades and splitParse does not fuse. Even using "inline" at the callsite
175+{-# INLINE splitAt #-}
176+splitAt :: Int -> [a] -> ([a],[a])
179+ | otherwise = splitAt' n ls
181+ splitAt' :: Int -> [a] -> ([a], [a])
182+ splitAt' _ [] = ([], [])
183+ splitAt' 1 (x:xs) = ([x], xs)
184+ splitAt' m (x:xs) = (x:xs', xs'')
186+ (xs', xs'') = splitAt' (m - 1) xs
188+-- | Parse a stream zipper using a parser's @step@, @initial@ and @extract@
191+{-# INLINE_NORMAL parse #-}
194+ => (s -> a -> m (PR.Step s b))
198+ -> m (Zipper m a, Either String b)
200+-- The case when no checkpoints exist
201+parse pstep initial extract (Zipper [] ls rs stream) =
203+ [] -> go stream ls initial
204+ _ -> gobuf stream ls rs initial
212+ r <- try $ extract pst
214+ Left (e :: SomeException) ->
215+ (nil, Left (displayException e))
216+ Right b -> (nil, Right b)
218+ acc1 <- acc >>= \b -> pstep b x
220+ -- PR.Yield 0 pst1 -> go SPEC s [] pst1
221+ PR.Yield n pst1 -> do
222+ assert (n <= length (x:buf)) (return ())
223+ go K.nil (Prelude.take n (x:buf)) (return pst1)
224+ PR.Skip 0 pst1 -> go K.nil (x:buf) (return pst1)
225+ PR.Skip n pst1 -> do
226+ assert (n <= length (x:buf)) (return ())
227+ let (src0, buf1) = splitAt n (x:buf)
228+ src = Prelude.reverse src0
229+ gobuf K.nil buf1 src (return pst1)
231+ assert (n <= length (x:buf)) (return ())
232+ let (src0, buf1) = splitAt n (x:buf)
233+ src = Prelude.reverse src0
234+ return (Zipper [] buf1 src K.nil, Right b)
236+ return (Zipper [] (x:buf) [] K.nil, Left err)
238+ acc1 <- acc >>= \b -> pstep b x
240+ -- PR.Yield 0 pst1 -> go SPEC s [] pst1
241+ PR.Yield n pst1 -> do
242+ assert (n <= length (x:buf)) (return ())
243+ go r (Prelude.take n (x:buf)) (return pst1)
244+ PR.Skip 0 pst1 -> go r (x:buf) (return pst1)
245+ PR.Skip n pst1 -> do
246+ assert (n <= length (x:buf)) (return ())
247+ let (src0, buf1) = splitAt n (x:buf)
248+ src = Prelude.reverse src0
249+ gobuf r buf1 src (return pst1)
251+ assert (n <= length (x:buf)) (return ())
252+ let (src0, buf1) = splitAt n (x:buf)
253+ src = Prelude.reverse src0
254+ return (Zipper [] buf1 src r, Right b)
256+ return (Zipper [] (x:buf) [] r, Left err)
257+ in K.foldStream defState yieldk single stop st
259+ gobuf s buf [] !pst = go s buf pst
260+ gobuf s buf (x:xs) !pst = do
264+ -- PR.Yield 0 pst1 -> go SPEC s [] pst1
265+ PR.Yield n pst1 -> do
266+ assert (n <= length (x:buf)) (return ())
267+ gobuf s (Prelude.take n (x:buf)) xs (return pst1)
268+ PR.Skip 0 pst1 -> gobuf s (x:buf) xs (return pst1)
269+ PR.Skip n pst1 -> do
270+ assert (n <= length (x:buf)) (return ())
271+ let (src0, buf1) = splitAt n (x:buf)
272+ src = Prelude.reverse src0 ++ xs
273+ gobuf s buf1 src (return pst1)
275+ assert (n <= length (x:buf)) (return ())
276+ let (src0, buf1) = splitAt n (x:buf)
277+ src = Prelude.reverse src0 ++ xs
278+ return (Zipper [] buf1 src s, Right b)
280+ return (Zipper [] (x:buf) xs s, Left err)
282+-- The case when checkpoints exist
283+-- XXX code duplication alert!
284+parse pstep initial extract (Zipper (cp:cps) ls rs stream) =
286+ [] -> go 0 stream ls initial
287+ _ -> gobuf 0 stream ls rs initial
292+ go cnt st buf !acc =
295+ r <- try $ extract pst
297+ Left (e :: SomeException) ->
298+ (nil, Left (displayException e))
299+ Right b -> (nil, Right b)
301+ acc1 <- acc >>= \b -> pstep b x
304+ -- PR.Yield 0 pst1 -> go SPEC s [] pst1
305+ PR.Yield n pst1 -> do
306+ assert (n <= length (x:buf)) (return ())
307+ go cnt1 K.nil (x:buf) (return pst1)
308+ PR.Skip 0 pst1 -> go cnt1 K.nil (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) K.nil buf1 src (return pst1)
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 K.nil
324+ return ( Zipper (cp + cnt1 : cps) (x:buf) [] K.nil
328+ acc1 <- acc >>= \b -> pstep b x
331+ -- PR.Yield 0 pst1 -> go SPEC s [] pst1
332+ PR.Yield n pst1 -> do
333+ assert (n <= length (x:buf)) (return ())
334+ go cnt1 r (x:buf) (return pst1)
335+ PR.Skip 0 pst1 -> go cnt1 r (x:buf) (return pst1)
336+ PR.Skip n pst1 -> do
337+ assert (n <= length (x:buf)) (return ())
338+ let (src0, buf1) = splitAt n (x:buf)
339+ src = Prelude.reverse src0
340+ assert (cnt1 - n >= 0) (return ())
341+ gobuf (cnt1 - n) r buf1 src (return pst1)
343+ assert (n <= length (x:buf)) (return ())
344+ let (src0, buf1) = splitAt n (x:buf)
345+ src = Prelude.reverse src0
346+ assert (cp + cnt1 - n >= 0) (return ())
347+ return ( Zipper (cp + cnt1 - n : cps) buf1 src r
351+ return ( Zipper (cp + cnt1 : cps) (x:buf) [] r
354+ in K.foldStream defState yieldk single stop st
356+ gobuf cnt s buf [] !pst = go cnt s buf pst
357+ gobuf cnt s buf (x:xs) !pst = do
362+ -- PR.Yield 0 pst1 -> go SPEC s [] pst1
363+ PR.Yield n pst1 -> do
364+ assert (n <= length (x:buf)) (return ())
365+ gobuf cnt1 s (x:buf) xs (return pst1)
366+ PR.Skip 0 pst1 -> gobuf cnt1 s (x:buf) xs (return pst1)
367+ PR.Skip n pst1 -> do
368+ assert (n <= length (x:buf)) (return ())
369+ let (src0, buf1) = splitAt n (x:buf)
370+ src = Prelude.reverse src0 ++ xs
371+ assert (cnt1 - n >= 0) (return ())
372+ gobuf (cnt1 - n) s buf1 src (return pst1)
374+ assert (n <= length (x:buf)) (return ())
375+ let (src0, buf1) = splitAt n (x:buf)
376+ src = Prelude.reverse src0 ++ xs
377+ assert (cp + cnt1 - n >= 0) (return ())
378+ return (Zipper (cp + cnt1 - n : cps) buf1 src s, Right b)
380+ return (Zipper (cp + cnt1 : cps) (x:buf) xs s, Left err)