d80d5e0 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement fromParserK

d80d5e0Harendra Kumar6 years ago

src/Streamly/Internal/Data/Parser/ParserD.hs+39−9

compare with master

unifiedsplit

@@ -199,22 +199,52 @@ fromFold (Fold fstep finitial fextract) = Parser step finitial fextract

199{-# INLINE_LATE toParserK #-}

200toParserK :: MonadCatch m => Parser m a b -> K.Parser m a b

201toParserK (Parser step initial extract) =

202− K.MkParser $ \inp yieldk ->

203− Z.parse step initial extract inp >>= yieldk

202+ K.MkParser $ K.parse step initial extract

204−

203+

204+-- XXX The CPS style parsers use a zipper buffering the data, if a parserD is

205+-- driving the parserK then it would also be buffering the same data. For

206+-- ParserD, instead of maintaining the buffer in the common driver, each parser

207+-- can have its own buffering and we can return the unconsumed buffer in the

208+-- end. That way the zipper is maintained by the parser. If the parser fails

209+-- then it has to return all of the input. It is anyway maintained by

210+-- intermediate level parsers in a composition, so the only difference would be

211+-- that even the leaf levels parsers would do it. If we abstract the zipper

212+-- maintainance then it may not be too unwieldy.

213+--

205-- | Convert a CPS style 'K.Parser' to a direct style 'Parser'.

214-- | Convert a CPS style 'K.Parser' to a direct style 'Parser'.

206--

215--

207-- /Unimplemented/

216-- /Unimplemented/

208--

217--

209−{-# NOINLINE fromParserK #-}

218+{-# INLINE_LATE fromParserK #-}

210−fromParserK :: Monad m => K.Parser m a b -> Parser m a b

219+fromParserK :: MonadThrow m => K.Parser m a b -> Parser m a b

211−fromParserK _ = Parser step initial extract

220+fromParserK parser = Parser step initial extract

212

221

213 where

222 where

214

223

215− initial = return ()

224+ initial = return Nothing

225+

226+ step Nothing a = do

227+ let yieldk _ (K.Done b) =

228+ -- let leftover = 0 -- XXX get it from the zipper

229+ return $ K.Stop b

216− step () _ = error "fromParserK: unimplemented"

230+ yieldk _ K.Continue = error "Bug: fromParserK: got Continue"

217− extract () = error "fromParserK: unimplemented"

231+ yieldk _ (K.Error e) = return $ K.Failed e

232+ r <- K.runParser parser (Z.append a Z.nil) yieldk

233+ case r of

234+ K.Stop b -> return $ Done 0 b

235+ K.Failed e -> return $ Error e

236+ -- XXX we can use the zipper state here to pare down the driver

237+ -- buffer

238+ K.Pause cont -> return $ Partial 0 $ Just cont

239+ step (Just cont) a = do

240+ r <- cont (Just a)

241+ case r of

242+ K.Stop b -> return $ Done 0 b

243+ K.Failed e -> return $ Error e

244+ K.Pause cont1 -> return $ Partial 0 $ Just cont1

245+

246+ extract Nothing = throwM $ ParseError "end of input"

247+ extract (Just cont) = K.extractParse cont

218

248

219#ifndef DISABLE_FUSION

249#ifndef DISABLE_FUSION

220{-# RULES "fromParserK/toParserK fusion" [2]

250{-# RULES "fromParserK/toParserK fusion" [2]

src/Streamly/Internal/Data/Parser/ParserK/Types.hs+289−33

@@ -20,31 +20,98 @@

21module Streamly.Internal.Data.Parser.ParserK.Types

22 (

23+ Parse (..)

24+ , Driver (..)

23− Parser (..)

25+ , Parser (..)

26+ , extractParse

24 , yield

27 , yield

25 , die

28 , die

29+

30+ -- * Parsing

31+ , parse

26 )

32 )

27where

33where

29−import Control.Monad (MonadPlus(..), ap)

30import Control.Applicative (Alternative(..))

35import Control.Applicative (Alternative(..))

31−import Streamly.Internal.Data.Zipper (Zipper)

36+import Control.Exception (assert, Exception(..))

32−import qualified Streamly.Internal.Data.Zipper as Z

37+import Control.Monad (MonadPlus(..), ap)

33−

38+import Control.Monad.Catch (MonadCatch, MonadThrow(..), try)

34#if MIN_VERSION_base(4,9,0)

39#if MIN_VERSION_base(4,9,0)

35import qualified Control.Monad.Fail as Fail

40import qualified Control.Monad.Fail as Fail

36#endif

41#endif

42+#if !(MIN_VERSION_base(4,10,0))

43+import Data.Semigroup ((<>))

44+#endif

45+import Streamly.Internal.Data.Zipper (Zipper (..))

46+import Streamly.Internal.Data.SVar (defState)

47+import Prelude hiding (splitAt)

48+

49+import qualified Streamly.Internal.Data.Zipper as Z

50+import qualified Streamly.Internal.Data.Parser.ParserD.Types as PD

51+import qualified Streamly.Internal.Data.Stream.StreamK as K

53+-- | The parse driver result. The driver may stop with a final result, pause

54+-- with a continuation to resume, or fail with an error.

55+--

56+-- /Internal/

57+--

58+data Driver m a r =

59+ Stop r

60+ -- XXX we can use a "resume" and a "stop" continuations instead of Maybe.

61+ -- measure if that works any better.

62+ | Pause (Maybe a -> m (Driver m a r))

63+ | Failed String

64+

65+instance Functor m => Functor (Driver m a) where

66+ fmap f (Stop r) = Stop (f r)

67+ fmap f (Pause yld) = Pause (fmap (fmap f) . yld)

68+ fmap _ (Failed e) = Failed e

69+

70+-- The parser's result.

71+--

72+-- /Internal/

73+--

74+data Parse b =

75+ Done !b -- Done, no more input needed

76+ | Continue -- No result, need more input

77+ | Error !String -- Failed

78+

79+instance Functor Parse where

80+ fmap f (Done b) = Done (f b)

81+ fmap _ Continue = Continue

82+ fmap _ (Error e) = Error e

83+

84+-- | A continuation passing style parser representation. The parser is supplied

85+-- with a 'Zipper' with an initial input, and a continuation to consume the

86+-- parse result. The Zipper is threaded around as a state and maintains the

87+-- parser backtracking buffer.

88+--

38−newtype Parser m a b =

89+newtype Parser m a b = MkParser

39− MkParser { runParser :: forall r.

90+ { runParser :: forall r.

40− Zipper m a -> ((Zipper m a, Either String b) -> m r) -> m r }

91+ Zipper m a

92+ -> (Zipper m a -> Parse b -> m (Driver m a r))

93+ -> m (Driver m a r)

94+ }

96+-- | When there is no more input to feed, extract the result from the Parser.

97+--

98+-- /Internal/

99+--

100+extractParse :: MonadThrow m => (Maybe a -> m (Driver m a b)) -> m b

101+extractParse cont = do

102+ r <- cont Nothing

103+ case r of

104+ Stop b -> return b

105+ Pause _ -> error "Bug: extractParse got Pause"

106+ Failed e -> throwM $ PD.ParseError e

107+

42−-- | Maps a function over the output of the fold.

108+-- | Maps a function over the output of the parser.

43--

109--

44instance Functor m => Functor (Parser m a) where

110instance Functor m => Functor (Parser m a) where

45 {-# INLINE fmap #-}

111 {-# INLINE fmap #-}

46− fmap f parser =

112+ fmap f parser = MkParser $ \zipper yieldk ->

47− MkParser $ \inp yieldk -> runParser parser inp (yieldk . fmap (fmap f))

113+ let yld z res = yieldk z (fmap f res)

114+ in runParser parser zipper yld

115

49-- | See 'Streamly.Internal.Data.Parser.yield'.

116-- | See 'Streamly.Internal.Data.Parser.yield'.

50--

117--

@@ -52,7 +119,7 @@ instance Functor m => Functor (Parser m a) where

52--

119--

53{-# INLINE yield #-}

120{-# INLINE yield #-}

54yield :: b -> Parser m a b

121yield :: b -> Parser m a b

55−yield b = MkParser (\inp yieldk -> yieldk (inp, Right b))

122+yield b = MkParser (\zipper yieldk -> yieldk zipper (Done b))

123

57-------------------------------------------------------------------------------

124-------------------------------------------------------------------------------

58-- Sequential applicative

125-- Sequential applicative

@@ -70,18 +137,19 @@ instance Monad m => Applicative (Parser m a) where

70 (<*>) = ap

137 (<*>) = ap

138

72 {-# INLINE (*>) #-}

139 {-# INLINE (*>) #-}

73− m1 *> m2 = MkParser $ \inp yieldk ->

140+ m1 *> m2 = MkParser $ \zipper yieldk ->

74− let yield1 (z, b) = case b of

75− Right _ -> runParser m2 z yieldk

141+ let yield1 z (Done _) = runParser m2 z yieldk

76− Left err -> runParser (die err) z yieldk

142+ yield1 z Continue = yieldk z Continue

143+ yield1 z (Error e) = yieldk z (Error e)

77− in runParser m1 inp yield1

144+ in runParser m1 zipper yield1

145

79 {-# INLINE (<*) #-}

146 {-# INLINE (<*) #-}

80− m1 <* m2 = MkParser $ \inp yieldk ->

147+ m1 <* m2 = MkParser $ \zipper yieldk ->

81− let yield1 (z, b) = case b of

148+ let

82− Right _ -> runParser m2 z (\(z1, _) -> yieldk (z1, b))

149+ yield1 z (Done b) = runParser m2 z $ \z1 _ -> yieldk z1 (Done b)

83− Left err -> runParser (die err) z yieldk

150+ yield1 z Continue = yieldk z Continue

151+ yield1 z (Error e) = yieldk z (Error e)

84− in runParser m1 inp yield1

152+ in runParser m1 zipper yield1

153

86-- | See 'Streamly.Internal.Data.Parser.die'.

154-- | See 'Streamly.Internal.Data.Parser.die'.

87--

155--

@@ -89,7 +157,7 @@ instance Monad m => Applicative (Parser m a) where

89--

157--

90{-# INLINE die #-}

158{-# INLINE die #-}

91die :: String -> Parser m a b

159die :: String -> Parser m a b

92−die err = MkParser (\z yieldk -> yieldk (z, Left err))

160+die err = MkParser (\z yieldk -> yieldk z (Error err))

161

94-- | Monad composition can be used for lookbehind parsers, we can make the

162-- | Monad composition can be used for lookbehind parsers, we can make the

95-- future parses depend on the previously parsed values.

163-- future parses depend on the previously parsed values.

@@ -138,11 +206,12 @@ instance Monad m => Monad (Parser m a) where

138 return = pure

206 return = pure

139

207

140 {-# INLINE (>>=) #-}

208 {-# INLINE (>>=) #-}

141− m >>= k = MkParser $ \inp yieldk ->

209+ m >>= k = MkParser $ \zipper yieldk ->

142− let yield1 (z, b) = case b of

210+ let

143− Right x -> runParser (k x) z yieldk

211+ yield1 z (Done b) = runParser (k b) z yieldk

144− Left err -> runParser (die err) z yieldk

212+ yield1 z Continue = yieldk z Continue

213+ yield1 z (Error e) = yieldk z (Error e)

145− in runParser m inp yield1

214+ in runParser m zipper yield1

146

215

147#if !(MIN_VERSION_base(4,13,0))

216#if !(MIN_VERSION_base(4,13,0))

148 -- This is redefined instead of just being Fail.fail to be

217 -- This is redefined instead of just being Fail.fail to be

@@ -175,11 +244,11 @@ instance Monad m => Alternative (Parser m a) where

175 empty = die "empty"

244 empty = die "empty"

176

245

177 {-# INLINE (<|>) #-}

246 {-# INLINE (<|>) #-}

178− m1 <|> m2 = MkParser $ \inp yieldk ->

247+ m1 <|> m2 = MkParser $ \zipper yieldk ->

179− let yield1 (z, b) = case b of

180− Right _ -> yieldk (Z.release z, b)

248+ let yield1 z (Done b) = yieldk (Z.release z) (Done b)

249+ yield1 z Continue = yieldk z Continue

181− Left _ -> runParser m2 (Z.restore z) yieldk

250+ yield1 z (Error _) = runParser m2 (Z.restore z) yieldk

182− in runParser m1 (Z.checkpoint inp) yield1

251+ in runParser m1 (Z.checkpoint zipper) yield1

183

252

184 -- some and many are implemented here instead of using default definitions

253 -- some and many are implemented here instead of using default definitions

185 -- so that we can use INLINE on them. It gives 50% performance improvement.

254 -- so that we can use INLINE on them. It gives 50% performance improvement.

@@ -211,3 +280,190 @@ instance Monad m => MonadPlus (Parser m a) where

211

280

212 {-# INLINE mplus #-}

281 {-# INLINE mplus #-}

213 mplus = (<|>)

282 mplus = (<|>)

283+

284+-------------------------------------------------------------------------------

285+-- Parse driver for a Zipper

286+-------------------------------------------------------------------------------

287+

288+-- Inlined definition. Without the inline "serially/parser/take" benchmark

289+-- degrades and splitParse does not fuse. Even using "inline" at the callsite

290+-- does not help.

291+{-# INLINE splitAt #-}

292+splitAt :: Int -> [a] -> ([a],[a])

293+splitAt n ls

294+ | n <= 0 = ([], ls)

295+ | otherwise = splitAt' n ls

296+ where

297+ splitAt' :: Int -> [a] -> ([a], [a])

298+ splitAt' _ [] = ([], [])

299+ splitAt' 1 (x:xs) = ([x], xs)

300+ splitAt' m (x:xs) = (x:xs', xs'')

301+ where

302+ (xs', xs'') = splitAt' (m - 1) xs

303+

304+-- | Parse a stream zipper using a direct style parser's @step@, @initial@ and

305+-- @extract@ functions.

306+--

307+{-# INLINE_NORMAL parse #-}

308+parse

309+ :: MonadCatch m

310+ => (s -> a -> m (PD.Step s b))

311+ -> m s

312+ -> (s -> m b)

313+ -> Zipper m a

314+ -> (Zipper m a -> Parse b -> m (Driver m a r))

315+ -> m (Driver m a r)

316+

317+-- The case when no checkpoints exist

318+parse pstep initial extract (Zipper [] backward forward stream) cont =

319+ case forward of

320+ -- Stop if stream is nil

321+ [] -> goStream stream backward initial

322+ _ -> goBuf stream backward forward initial

323+

324+ where

325+

326+ -- Pauses if no input was generated by the stream and returns a

327+ -- continuation. If the continuation is called with Nothing then it stops

328+ -- and returns the result.

329+ {-# INLINE goStream #-}

330+ goStream st back !acc =

331+ -- XXX We do not need to wrap the element in a stream we can process it

332+ -- directly.

333+ let parseCont (Just x) = goStream (K.yield x) back acc

334+ parseCont Nothing = do

335+ pst <- acc

336+ r <- try $ extract pst

337+ case r of

338+ Left (e :: PD.ParseError) ->

339+ cont Z.nil (Error (displayException e))

340+ Right b -> cont Z.nil (Done b)

341+ stop = return $ Pause parseCont

342+ single x = yieldk x K.nil

343+ yieldk x r = do

344+ acc1 <- acc >>= \b -> pstep b x

345+ case acc1 of

346+ PD.Partial 0 pst1 -> goStream r [] (return pst1)

347+ PD.Partial n pst1 -> do

348+ assert (n <= length (x:back)) (return ())

349+ let src0 = Prelude.take n (x:back)

350+ src = Prelude.reverse src0

351+ goBuf r [] src (return pst1)

352+ PD.Continue 0 pst1 -> goStream r (x:back) (return pst1)

353+ PD.Continue n pst1 -> do

354+ assert (n <= length (x:back)) (return ())

355+ let (src0, buf1) = splitAt n (x:back)

356+ src = Prelude.reverse src0

357+ goBuf r buf1 src (return pst1)

358+ PD.Done n b -> do

359+ assert (n <= length (x:back)) (return ())

360+ let (src0, buf1) = splitAt n (x:back)

361+ src = Prelude.reverse src0

362+ cont (Zipper [] buf1 src r) (Done b)

363+ PD.Error err -> cont (Zipper [] (x:back) [] r) (Error err)

364+ in K.foldStream defState yieldk single stop st

365+

366+ goBuf s back [] !pst = goStream s back pst

367+ goBuf s back (x:xs) !pst = do

368+ r <- pst

369+ pRes <- pstep r x

370+ case pRes of

371+ PD.Partial 0 pst1 ->

372+ goBuf s [] xs (return pst1)

373+ PD.Partial n pst1 -> do

374+ assert (n <= length (x:back)) (return ())

375+ let src0 = Prelude.take n (x:back)

376+ src = Prelude.reverse src0

377+ goBuf s [] (src ++ xs) (return pst1)

378+ PD.Continue 0 pst1 -> goBuf s (x:back) xs (return pst1)

379+ PD.Continue n pst1 -> do

380+ assert (n <= length (x:back)) (return ())

381+ let (src0, buf1) = splitAt n (x:back)

382+ src = Prelude.reverse src0 ++ xs

383+ goBuf s buf1 src (return pst1)

384+ PD.Done n b -> do

385+ assert (n <= length (x:back)) (return ())

386+ let (src0, buf1) = splitAt n (x:back)

387+ src = Prelude.reverse src0 ++ xs

388+ cont (Zipper [] buf1 src s) (Done b)

389+ PD.Error err -> cont (Zipper [] (x:back) xs s) (Error err)

390+

391+-- The case when checkpoints exist

392+-- XXX code duplication alert!

393+parse pstep initial extract (Zipper (cp:cps) backward forward stream) cont =

394+ case forward of

395+ [] -> goStream 0 stream backward initial

396+ _ -> goBuf 0 stream backward forward initial

397+

398+ where

399+

400+ {-# INLINE goStream #-}

401+ goStream cnt st back !acc =

402+ let parseCont (Just x) = goStream cnt (K.yield x) back acc

403+ parseCont Nothing = do

404+ pst <- acc

405+ r <- try $ extract pst

406+ case r of

407+ Left (e :: PD.ParseError) ->

408+ cont Z.nil (Error (displayException e))

409+ Right b -> cont Z.nil (Done b)

410+ stop = return $ Pause parseCont

411+ single x = yieldk x K.nil

412+ yieldk x r = do

413+ acc1 <- acc >>= \b -> pstep b x

414+ let cnt1 = cnt + 1

415+ case acc1 of

416+ PD.Partial 0 pst1 ->

417+ goStream cnt1 r [] (return pst1)

418+ PD.Partial n pst1 -> do

419+ assert (n <= length (x:back)) (return ())

420+ let src0 = Prelude.take n (x:back)

421+ src = Prelude.reverse src0

422+ goBuf (cnt1 - n) r [] src (return pst1)

423+ PD.Continue 0 pst1 -> goStream cnt1 r (x:back) (return pst1)

424+ PD.Continue n pst1 -> do

425+ assert (n <= length (x:back)) (return ())

426+ let (src0, buf1) = splitAt n (x:back)

427+ src = Prelude.reverse src0

428+ assert (cnt1 - n >= 0) (return ())

429+ goBuf (cnt1 - n) r buf1 src (return pst1)

430+ PD.Done n b -> do

431+ assert (n <= length (x:back)) (return ())

432+ let (src0, buf1) = splitAt n (x:back)

433+ src = Prelude.reverse src0

434+ assert (cp + cnt1 - n >= 0) (return ())

435+ cont (Zipper (cp + cnt1 - n : cps) buf1 src r)

436+ (Done b)

437+ PD.Error err ->

438+ cont (Zipper (cp + cnt1 : cps) (x:back) [] r)

439+ (Error err)

440+ in K.foldStream defState yieldk single stop st

441+

442+ goBuf cnt s back [] !pst = goStream cnt s back pst

443+ goBuf cnt s back (x:xs) !pst = do

444+ r <- pst

445+ pRes <- pstep r x

446+ let cnt1 = cnt + 1

447+ case pRes of

448+ PD.Partial 0 pst1 ->

449+ goBuf cnt1 s [] xs (return pst1)

450+ PD.Partial n pst1 -> do

451+ assert (n <= length (x:back)) (return ())

452+ let src0 = Prelude.take n (x:back)

453+ src = Prelude.reverse src0

454+ goBuf (cnt1 - n) s [] (src ++ xs) (return pst1)

455+ PD.Continue 0 pst1 -> goBuf cnt1 s (x:back) xs (return pst1)

456+ PD.Continue n pst1 -> do

457+ assert (n <= length (x:back)) (return ())

458+ let (src0, buf1) = splitAt n (x:back)

459+ src = Prelude.reverse src0 ++ xs

460+ assert (cnt1 - n >= 0) (return ())

461+ goBuf (cnt1 - n) s buf1 src (return pst1)

462+ PD.Done n b -> do

463+ assert (n <= length (x:back)) (return ())

464+ let (src0, buf1) = splitAt n (x:back)

465+ src = Prelude.reverse src0 ++ xs

466+ assert (cp + cnt1 - n >= 0) (return ())

467+ cont (Zipper (cp + cnt1 - n : cps) buf1 src s) (Done b)

468+ PD.Error err ->

469+ cont (Zipper (cp + cnt1 : cps) (x:back) xs s) (Error err)

src/Streamly/Internal/Data/Zipper.hs+41−183

@@ -52,25 +52,24 @@ module Streamly.Internal.Data.Zipper

52 , fromStream

53 , fromList

55+ -- * Updating

56+ , append

57+

55 -- * Checkpointing

58 -- * Checkpointing

56 , checkpoint

59 , checkpoint

57 , release

60 , release

58 , restore

61 , restore

59−

60− -- * Parsing

61− , parse

62 )

63where

65−import Control.Exception (assert, Exception(..))

65+import Control.Exception (assert)

66−import Control.Monad.Catch (MonadCatch, try)

66+#if !(MIN_VERSION_base(4,13,0))

67−-- import GHC.Types (SPEC(..))

67+import Data.Semigroup ((<>))

68+#endif

68import Prelude hiding (splitAt)

69import Prelude hiding (splitAt)

70import Streamly.Internal.Data.Stream.StreamK.Type (Stream(..))

71import Streamly.Internal.Data.Stream.StreamK.Type (Stream(..))

71−import Streamly.Internal.Data.SVar (defState)

73−import qualified Streamly.Internal.Data.Parser.ParserD.Types as PR

74import qualified Streamly.Internal.Data.Stream.StreamK as K

73import qualified Streamly.Internal.Data.Stream.StreamK as K

76-------------------------------------------------------------------------------

75-------------------------------------------------------------------------------

@@ -95,7 +94,11 @@ import qualified Streamly.Internal.Data.Stream.StreamK as K

95--

94--

96-- /Internal/

95-- /Internal/

97--

96--

98−data Zipper m a = Zipper [Int] [a] [a] (Stream m a)

97+data Zipper m a = Zipper

98+ [Int] -- checkpoints

99+ [a] -- past buffered inputs (for backtracking)

100+ [a] -- future buffered inputs (created by backtracking)

101+ (Stream m a) -- stream input to use after buffer is over

102

100-------------------------------------------------------------------------------

103-------------------------------------------------------------------------------

101-- Construction

104-- Construction

@@ -125,6 +128,23 @@ fromStream = Zipper [] [] []

125fromList :: [a] -> Zipper m a

128fromList :: [a] -> Zipper m a

126fromList xs = Zipper [] [] xs K.nil

129fromList xs = Zipper [] [] xs K.nil

127

130

131+-------------------------------------------------------------------------------

132+-- Updating

133+-------------------------------------------------------------------------------

134+

135+-- | Add an element at the tail of the Zipper.

136+--

137+-- /Internal/

138+--

139+{-# INLINE append #-}

140+append :: a -> Zipper m a -> Zipper m a

141+append x (Zipper checkpoints backward forward stream) =

142+ Zipper checkpoints backward forward (stream <> K.yield x)

143+

144+-------------------------------------------------------------------------------

145+-- Checkpointing

146+-------------------------------------------------------------------------------

147+

128-- | Add a checkpoint to the Zipper so that we do not release the buffer beyond

148-- | Add a checkpoint to the Zipper so that we do not release the buffer beyond

129-- the checkpoint.

149-- the checkpoint.

130--

150--

@@ -149,23 +169,8 @@ release (Zipper (n:cps) xs ys stream) =

149 [] -> assert (n == length xs) $ Zipper [] [] ys stream

169 [] -> assert (n == length xs) $ Zipper [] [] ys stream

150 cp:rest -> Zipper ((cp + n) : rest) xs ys stream

170 cp:rest -> Zipper ((cp + n) : rest) xs ys stream

151

171

152−-- | Rewind to restore the cursor to the latest checkpoint.

172+-- XXX recheck this, and unify with the definition in StreamD and ParserK

153−--

154−-- /Internal/

155--

173--

156−{-# INLINE restore #-}

157−restore :: Zipper m a -> Zipper m a

158−restore (Zipper [] _ _ _) = error "Bug: restore, no checkpoint exists!"

159−restore (Zipper (n:cps) xs ys stream) =

160− assert (n <= length xs) $

161− let (src0, buf1) = splitAt n xs

162− src = Prelude.reverse src0

163− in Zipper cps buf1 (src ++ ys) stream

164−

165−-------------------------------------------------------------------------------

166−-- Parse driver for a Zipper

167−-------------------------------------------------------------------------------

168−

169-- Inlined definition. Without the inline "serially/parser/take" benchmark

174-- Inlined definition. Without the inline "serially/parser/take" benchmark

170-- degrades and splitParse does not fuse. Even using "inline" at the callsite

175-- degrades and splitParse does not fuse. Even using "inline" at the callsite

171-- does not help.

176-- does not help.

@@ -182,162 +187,15 @@ splitAt n ls

182 where

187 where

183 (xs', xs'') = splitAt' (m - 1) xs

188 (xs', xs'') = splitAt' (m - 1) xs

184

189

185−-- | Parse a stream zipper using a parser's @step@, @initial@ and @extract@

190+-- | Rewind to restore the cursor to the latest checkpoint.

186−-- functions.

187--

191--

188−{-# INLINE_NORMAL parse #-}

192+-- /Internal/

189−parse

193+--

190− :: MonadCatch m

194+{-# INLINE restore #-}

191− => (s -> a -> m (PR.Step s b))

192− -> m s

193− -> (s -> m b)

194− -> Zipper m a

195+restore :: Zipper m a -> Zipper m a

195− -> m (Zipper m a, Either String b)

196+restore (Zipper [] _ _ _) = error "Bug: restore, no checkpoint exists!"

196−

197+restore (Zipper (n:cps) xs ys stream) =

197−-- The case when no checkpoints exist

198−parse pstep initial extract (Zipper [] ls rs stream) =

199− case rs of

200− [] -> go stream ls initial

201− _ -> gobuf stream ls rs initial

202−

203− where

204−

205− {-# INLINE go #-}

206− go st buf !acc =

207− let stop = do

208− pst <- acc

209− r <- try $ extract pst

210− return $ case r of

211− Left (e :: PR.ParseError) ->

212− (nil, Left (displayException e))

213− Right b -> (nil, Right b)

214− single x = yieldk x K.nil

215− yieldk x r = do

216− acc1 <- acc >>= \b -> pstep b x

217− case acc1 of

218− PR.Partial 0 pst1 -> go r [] (return pst1)

219− PR.Partial n pst1 -> do

220− assert (n <= length (x:buf)) (return ())

221− let src0 = Prelude.take n (x:buf)

222− src = Prelude.reverse src0

223− gobuf r [] src (return pst1)

224− PR.Continue 0 pst1 -> go r (x:buf) (return pst1)

225− PR.Continue 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 r buf1 src (return pst1)

230− PR.Done 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 r, Right b)

235− PR.Error err ->

236− return (Zipper [] (x:buf) [] r, Left err)

237− in K.foldStream defState yieldk single stop st

238−

239− gobuf s buf [] !pst = go s buf pst

240− gobuf s buf (x:xs) !pst = do

241− r <- pst

242− pRes <- pstep r x

243− case pRes of

244− PR.Partial 0 pst1 ->

245− gobuf s [] xs (return pst1)

246− PR.Partial n pst1 -> do

247− assert (n <= length (x:buf)) (return ())

248− let src0 = Prelude.take n (x:buf)

249− src = Prelude.reverse src0

250− gobuf s [] (src ++ xs) (return pst1)

251− PR.Continue 0 pst1 -> gobuf s (x:buf) xs (return pst1)

252− PR.Continue n pst1 -> do

253− assert (n <= length (x:buf)) (return ())

254− let (src0, buf1) = splitAt n (x:buf)

255− src = Prelude.reverse src0 ++ xs

256− gobuf s buf1 src (return pst1)

257− PR.Done n b -> do

258− assert (n <= length (x:buf)) (return ())

259− let (src0, buf1) = splitAt n (x:buf)

260− src = Prelude.reverse src0 ++ xs

261− return (Zipper [] buf1 src s, Right b)

262− PR.Error err ->

263− return (Zipper [] (x:buf) xs s, Left err)

264−

265−-- The case when checkpoints exist

266−-- XXX code duplication alert!

267−parse pstep initial extract (Zipper (cp:cps) ls rs stream) =

268− case rs of

269− [] -> go 0 stream ls initial

270− _ -> gobuf 0 stream ls rs initial

271−

272− where

273−

274− {-# INLINE go #-}

275− go cnt st buf !acc =

276− let stop = do

277− pst <- acc

278− r <- try $ extract pst

279− return $ case r of

280− Left (e :: PR.ParseError) ->

281− (nil, Left (displayException e))

282− Right b -> (nil, Right b)

283− single x = yieldk x K.nil

284− yieldk x r = do

285− acc1 <- acc >>= \b -> pstep b x

286− let cnt1 = cnt + 1

287− case acc1 of

288− PR.Partial 0 pst1 ->

289− go cnt1 r [] (return pst1)

290− PR.Partial n pst1 -> do

291− assert (n <= length (x:buf)) (return ())

292− let src0 = Prelude.take n (x:buf)

293− src = Prelude.reverse src0

294− gobuf (cnt1 - n) r [] src (return pst1)

295− PR.Continue 0 pst1 -> go cnt1 r (x:buf) (return pst1)

296− PR.Continue n pst1 -> do

297− assert (n <= length (x:buf)) (return ())

298− let (src0, buf1) = splitAt n (x:buf)

299− src = Prelude.reverse src0

300− assert (cnt1 - n >= 0) (return ())

301− gobuf (cnt1 - n) r buf1 src (return pst1)

302− PR.Done n b -> do

303− assert (n <= length (x:buf)) (return ())

304− let (src0, buf1) = splitAt n (x:buf)

305− src = Prelude.reverse src0

306− assert (cp + cnt1 - n >= 0) (return ())

307− return ( Zipper (cp + cnt1 - n : cps) buf1 src r

308− , Right b

309− )

310− PR.Error err ->

311− return ( Zipper (cp + cnt1 : cps) (x:buf) [] r

312− , Left err

313− )

314− in K.foldStream defState yieldk single stop st

315−

316− gobuf cnt s buf [] !pst = go cnt s buf pst

317− gobuf cnt s buf (x:xs) !pst = do

318− r <- pst

319− pRes <- pstep r x

320− let cnt1 = cnt + 1

321− case pRes of

322− PR.Partial 0 pst1 ->

323− gobuf cnt1 s [] xs (return pst1)

324− PR.Partial n pst1 -> do

325− assert (n <= length (x:buf)) (return ())

326− let src0 = Prelude.take n (x:buf)

327− src = Prelude.reverse src0

328− gobuf (cnt1 - n) s [] (src ++ xs) (return pst1)

329− PR.Continue 0 pst1 -> gobuf cnt1 s (x:buf) xs (return pst1)

330− PR.Continue n pst1 -> do

331− assert (n <= length (x:buf)) (return ())

332− let (src0, buf1) = splitAt n (x:buf)

333− src = Prelude.reverse src0 ++ xs

334− assert (cnt1 - n >= 0) (return ())

335− gobuf (cnt1 - n) s buf1 src (return pst1)

336− PR.Done n b -> do

337− assert (n <= length (x:buf)) (return ())

198+ assert (n <= length xs) $

338− let (src0, buf1) = splitAt n (x:buf)

199+ let (src0, buf1) = splitAt n xs

339− src = Prelude.reverse src0 ++ xs

200+ src = Prelude.reverse src0

340− assert (cp + cnt1 - n >= 0) (return ())

341− return (Zipper (cp + cnt1 - n : cps) buf1 src s, Right b)

201+ in Zipper cps buf1 (src ++ ys) stream

342− PR.Error err ->

343− return (Zipper (cp + cnt1 : cps) (x:buf) xs s, Left err)

src/Streamly/Internal/Prelude.hs+7−4

@@ -1368,11 +1368,14 @@ parseD (PRD.Parser step initial extract) = P.parselMx' step initial extract

1368{-# INLINE parseK #-}

1369parseK :: MonadThrow m => PRK.Parser m a b -> SerialT m a -> m b

1370parseK parser xs = do

1371+ let yieldk _ (PRK.Done b) = return $ PRK.Stop b

1372+ yieldk _ PRK.Continue = error "Bug: fromParserK: got Continue"

1373+ yieldk _ (PRK.Error e) = return $ PRK.Failed e

1371− r <- PRK.runParser parser (ZR.fromStream $ K.toStream xs)

1374+ r <- PRK.runParser parser (ZR.fromStream $ K.toStream xs) yieldk

1372− (\(_, b) -> return b)

1373 case r of

1375 case r of

1376+ PRK.Stop b -> return b

1374− Left err -> throwM (PRD.ParseError err)

1377+ PRK.Failed err -> throwM (PRD.ParseError err)

1375− Right b -> return b

1378+ PRK.Pause cont -> PRK.extractParse cont

1376

1379

1377-- | Parse a stream using the supplied 'Parser'.

1380-- | Parse a stream using the supplied 'Parser'.

1378--

1381--