@@ -0,0 +1,380 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE ExistentialQuantification #-}
3+{-# LANGUAGE BangPatterns #-}
4+{-# LANGUAGE RankNTypes #-}
5+{-# LANGUAGE ScopedTypeVariables #-}
6+
7+#include "inline.hs"
8+
9+-- |
10+-- Module : Streamly.Internal.Data.Zipper
11+-- Copyright : (c) 2020 Composewell Technologies
12+-- License : BSD3
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.
51+
52+module Streamly.Internal.Data.Zipper
53+ (
54+ Zipper (..)
55+
56+ -- * Construction
57+ , nil
58+ , fromStream
59+ , fromList
60+
61+ -- * Checkpointing
62+ , checkpoint
63+ , release
64+ , restore
65+
66+ -- * Parsing
67+ , parse
68+ )
69+where
70+
71+import Control.Exception (assert, Exception(..))
72+import Control.Monad.Catch (MonadCatch, try, SomeException)
73+-- import GHC.Types (SPEC(..))
74+import Prelude hiding (splitAt)
75+
76+import Streamly.Internal.Data.Stream.StreamK.Type (Stream(..))
77+import Streamly.Internal.Data.SVar (defState)
78+
79+import qualified Streamly.Internal.Data.Parser.Types as PR
80+import qualified Streamly.Internal.Data.Stream.StreamK as K
81+
82+-------------------------------------------------------------------------------
83+-- Zipper type
84+-------------------------------------------------------------------------------
85+
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
89+-- @rights@.
90+--
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
99+-- released.
100+--
101+-- /Internal/
102+--
103+data Zipper m a = Zipper [Int] [a] [a] (Stream m a)
104+
105+-------------------------------------------------------------------------------
106+-- Construction
107+-------------------------------------------------------------------------------
108+
109+-- | Empty zipper.
110+--
111+-- /Internal/
112+--
113+{-# INLINE nil #-}
114+nil :: Zipper m a
115+nil = Zipper [] [] [] K.nil
116+
117+-- | Create a zipper from a stream.
118+--
119+-- /Internal/
120+--
121+{-# INLINE fromStream #-}
122+fromStream :: Stream m a -> Zipper m a
123+fromStream = Zipper [] [] []
124+
125+-- | Create a zipper from a list.
126+--
127+-- /Internal/
128+--
129+{-# INLINE fromList #-}
130+fromList :: [a] -> Zipper m a
131+fromList xs = Zipper [] [] xs K.nil
132+
133+-- | Add a checkpoint to the Zipper so that we do not release the buffer beyond
134+-- the checkpoint.
135+--
136+-- /Internal/
137+--
138+{-# INLINE checkpoint #-}
139+checkpoint :: Zipper m a -> Zipper m a
140+checkpoint (Zipper cps xs ys stream) = Zipper (0:cps) xs ys stream
141+
142+-- | Release the latest checkpoint, releases any values held by the checkpoint.
143+--
144+-- /Internal/
145+--
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) $
151+ case cps of
152+ [] -> assert (n == length xs) $ Zipper [] [] ys stream
153+ cp:rest -> Zipper ((cp + n) : rest) xs ys stream
154+
155+-- | Rewind to restore the cursor to the latest checkpoint.
156+--
157+-- /Internal/
158+--
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
167+
168+-------------------------------------------------------------------------------
169+-- Parse driver for a Zipper
170+-------------------------------------------------------------------------------
171+
172+-- Inlined definition. Without the inline "serially/parser/take" benchmark
173+-- degrades and splitParse does not fuse. Even using "inline" at the callsite
174+-- does not help.
175+{-# INLINE splitAt #-}
176+splitAt :: Int -> [a] -> ([a],[a])
177+splitAt n ls
178+ | n <= 0 = ([], ls)
179+ | otherwise = splitAt' n ls
180+ where
181+ splitAt' :: Int -> [a] -> ([a], [a])
182+ splitAt' _ [] = ([], [])
183+ splitAt' 1 (x:xs) = ([x], xs)
184+ splitAt' m (x:xs) = (x:xs', xs'')
185+ where
186+ (xs', xs'') = splitAt' (m - 1) xs
187+
188+-- | Parse a stream zipper using a parser's @step@, @initial@ and @extract@
189+-- functions.
190+--
191+{-# INLINE_NORMAL parse #-}
192+parse
193+ :: MonadCatch m
194+ => (s -> a -> m (PR.Step s b))
195+ -> m s
196+ -> (s -> m b)
197+ -> Zipper m a
198+ -> m (Zipper m a, Either String b)
199+
200+-- The case when no checkpoints exist
201+parse pstep initial extract (Zipper [] ls rs stream) =
202+ case rs of
203+ [] -> go stream ls initial
204+ _ -> gobuf stream ls rs initial
205+
206+ where
207+
208+ {-# INLINE go #-}
209+ go st buf !acc =
210+ let stop = do
211+ pst <- acc
212+ r <- try $ extract pst
213+ return $ case r of
214+ Left (e :: SomeException) ->
215+ (nil, Left (displayException e))
216+ Right b -> (nil, Right b)
217+ single x = do
218+ acc1 <- acc >>= \b -> pstep b x
219+ case acc1 of
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)
230+ PR.Stop n b -> do
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)
235+ PR.Error err ->
236+ return (Zipper [] (x:buf) [] K.nil, Left err)
237+ yieldk x r = do
238+ acc1 <- acc >>= \b -> pstep b x
239+ case acc1 of
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)
250+ PR.Stop n b -> do
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)
255+ PR.Error err ->
256+ return (Zipper [] (x:buf) [] r, Left err)
257+ in K.foldStream defState yieldk single stop st
258+
259+ gobuf s buf [] !pst = go s buf pst
260+ gobuf s buf (x:xs) !pst = do
261+ r <- pst
262+ pRes <- pstep r x
263+ case pRes of
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)
274+ PR.Stop n b -> do
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)
279+ PR.Error err ->
280+ return (Zipper [] (x:buf) xs s, Left err)
281+
282+-- The case when checkpoints exist
283+-- XXX code duplication alert!
284+parse pstep initial extract (Zipper (cp:cps) ls rs stream) =
285+ case rs of
286+ [] -> go 0 stream ls initial
287+ _ -> gobuf 0 stream ls rs initial
288+
289+ where
290+
291+ {-# INLINE go #-}
292+ go cnt st buf !acc =
293+ let stop = do
294+ pst <- acc
295+ r <- try $ extract pst
296+ return $ case r of
297+ Left (e :: SomeException) ->
298+ (nil, Left (displayException e))
299+ Right b -> (nil, Right b)
300+ single x = do
301+ acc1 <- acc >>= \b -> pstep b x
302+ let cnt1 = cnt + 1
303+ case acc1 of
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)
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 K.nil
321+ , Right b
322+ )
323+ PR.Error err ->
324+ return ( Zipper (cp + cnt1 : cps) (x:buf) [] K.nil
325+ , Left err
326+ )
327+ yieldk x r = do
328+ acc1 <- acc >>= \b -> pstep b x
329+ let cnt1 = cnt + 1
330+ case acc1 of
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)
342+ PR.Stop n b -> do
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
348+ , Right b
349+ )
350+ PR.Error err ->
351+ return ( Zipper (cp + cnt1 : cps) (x:buf) [] r
352+ , Left err
353+ )
354+ in K.foldStream defState yieldk single stop st
355+
356+ gobuf cnt s buf [] !pst = go cnt s buf pst
357+ gobuf cnt s buf (x:xs) !pst = do
358+ r <- pst
359+ pRes <- pstep r x
360+ let cnt1 = cnt + 1
361+ case pRes of
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)
373+ PR.Stop n b -> do
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)
379+ PR.Error err ->
380+ return (Zipper (cp + cnt1 : cps) (x:buf) xs s, Left err)