@@ -128,6 +128,9 @@ module Streamly.Internal.Data.Stream.StreamD
128128 , foldlMx'
129129 , runFold
130130
131+ , parselMx'
132+ , parseChunks
133+
131134 -- ** Specialized Folds
132135 , tap
133136 , tapOffsetEvery
@@ -320,7 +323,7 @@ where
320323
321324import Control.Concurrent (killThread, myThreadId, takeMVar, threadDelay)
322325import Control.Exception
323− (Exception, SomeException, AsyncException, fromException, mask_)
326+ (assert, Exception, SomeException, AsyncException, fromException, mask_)
324327import Control.Monad (void, when, forever)
325328import Control.Monad.Catch (MonadCatch, throwM)
326329import Control.Monad.IO.Class (MonadIO(..))
@@ -343,13 +346,14 @@ import Prelude
343346 takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,
344347 notElem, null, head, tail, zipWith, lookup, foldr1, sequence,
345348 (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat,
346− reverse, iterate)
349+ reverse, iterate, splitAt)
347350
348351import qualified Control.Monad.Catch as MC
349352import qualified Control.Monad.Reader as Reader
350353import qualified Control.Monad.State.Strict as State
351354import qualified Prelude
352355
356+import Fusion.Plugin.Types (Fuse(..))
353357import Streamly.Internal.Mutable.Prim.Var
354358 (Prim, Var, readVar, newVar, modifyVar')
355359import Streamly.Internal.Data.Time.Units
@@ -358,6 +362,7 @@ import Streamly.Internal.Data.Time.Units
358362import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
359363import Streamly.Internal.Memory.Array.Types (Array(..))
360364import Streamly.Internal.Data.Fold.Types (Fold(..))
365+import Streamly.Internal.Data.Parse.Types (Parse(..))
361366import Streamly.Internal.Data.Pipe.Types (Pipe(..), PipeState(..))
362367import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
363368import Streamly.Internal.Data.Time.Units
@@ -374,6 +379,7 @@ import qualified Streamly.Internal.Memory.Array.Types as A
374379import qualified Streamly.Internal.Data.Fold as FL
375380import qualified Streamly.Memory.Ring as RB
376381import qualified Streamly.Internal.Data.Stream.StreamK as K
382+import qualified Streamly.Internal.Data.Parse.Types as PR
377383
378384------------------------------------------------------------------------------
379385-- Construction
@@ -942,6 +948,184 @@ foldlS fstep begin (Stream step state) = Stream step' (Left (state, begin))
942948 Skip s -> Skip (Right (Stream stp s))
943949 Stop -> Stop
944950
951+------------------------------------------------------------------------------
952+-- Parses
953+------------------------------------------------------------------------------
954+
955+-- Inlined definition. Without the inline "serially/parser/take" benchmark
956+-- degrades and parseChunks does not fuse. Even using "inline" at the callsite
957+-- does not help.
958+{-# INLINE splitAt #-}
959+splitAt :: Int -> [a] -> ([a],[a])
960+splitAt n ls
961+ | n <= 0 = ([], ls)
962+ | otherwise = splitAt' n ls
963+ where
964+ splitAt' :: Int -> [a] -> ([a], [a])
965+ splitAt' _ [] = ([], [])
966+ splitAt' 1 (x:xs) = ([x], xs)
967+ splitAt' m (x:xs) = (x:xs', xs'')
968+ where
969+ (xs', xs'') = splitAt' (m - 1) xs
970+
971+-- | Run a 'Parse' over a stream.
972+{-# INLINE_NORMAL parselMx' #-}
973+parselMx'
974+ :: Monad m
975+ => (s -> a -> m (PR.Step s))
976+ -> m s
977+ -> (s -> m (Either String (Int, b)))
978+ -> Stream m a
979+ -> m (Either String b)
980+parselMx' pstep initial extract (Stream step state) = do
981+ initial >>= go SPEC state []
982+
983+ where
984+
985+ -- XXX currently we are using a dumb list based approach for backtracking
986+ -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
987+ -- That will allow us more efficient random back and forth movement.
988+ {-# INLINE go #-}
989+ go !_ st buf !pst = do
990+ r <- step defState st
991+ case r of
992+ Yield x s -> do
993+ pRes <- pstep pst x
994+ case pRes of
995+ -- PR.Keep 0 pst1 -> go SPEC s [] pst1
996+ PR.Keep n pst1 -> do
997+ assert (n <= length (x:buf)) (return ())
998+ go SPEC s (Prelude.take n (x:buf)) pst1
999+ PR.Back n pst1 -> do
1000+ assert (n <= length (x:buf)) (return ())
1001+ let (src0, buf1) = splitAt n (x:buf)
1002+ src = Prelude.reverse src0
1003+ gobuf SPEC s buf1 src pst1
1004+ PR.Hold pst1 -> go SPEC s (x:buf) pst1
1005+ PR.Halt pst1 -> fmap (fmap snd) (extract pst1)
1006+ Skip s -> go SPEC s buf pst
1007+ Stop -> fmap (fmap snd) (extract pst)
1008+
1009+ gobuf !_ s buf [] !pst = go SPEC s buf pst
1010+ gobuf !_ s buf (x:xs) !pst = do
1011+ pRes <- pstep pst x
1012+ case pRes of
1013+ -- PR.Keep 0 pst1 -> go SPEC s [] pst1
1014+ PR.Keep n pst1 -> do
1015+ assert (n <= length (x:buf)) (return ())
1016+ gobuf SPEC s (Prelude.take n (x:buf)) xs pst1
1017+ PR.Back n pst1 -> do
1018+ assert (n <= length (x:buf)) (return ())
1019+ let (src0, buf1) = splitAt n (x:buf)
1020+ src = Prelude.reverse src0 ++ xs
1021+ gobuf SPEC s buf1 src pst1
1022+ PR.Hold pst1 -> gobuf SPEC s (x:buf) xs pst1
1023+ PR.Halt pst1 -> fmap (fmap snd) (extract pst1)
1024+
1025+------------------------------------------------------------------------------
1026+-- Repeated parsing
1027+------------------------------------------------------------------------------
1028+
1029+{-# ANN type ParseChunksState Fuse #-}
1030+data ParseChunksState x inpBuf st pst =
1031+ ParseChunksInit inpBuf st
1032+ | ParseChunksInitLeftOver inpBuf
1033+ | ParseChunksStream st inpBuf pst
1034+ | ParseChunksBuf inpBuf st inpBuf pst
1035+ | ParseChunksYield x (ParseChunksState x inpBuf st pst)
1036+
1037+{-# INLINE_NORMAL parseChunks #-}
1038+parseChunks
1039+ :: Monad m
1040+ => Parse m a b
1041+ -> Stream m a
1042+ -> Stream m b
1043+parseChunks (Parse pstep initial extract) (Stream step state) =
1044+ Stream stepOuter (ParseChunksInit [] state)
1045+
1046+ where
1047+
1048+ {-# INLINE_LATE stepOuter #-}
1049+ -- Buffer is empty, go to stream processing loop
1050+ stepOuter _ (ParseChunksInit [] st) = do
1051+ initial >>= return . Skip . ParseChunksStream st []
1052+
1053+ -- Buffer is not empty, go to buffered processing loop
1054+ stepOuter _ (ParseChunksInit src st) = do
1055+ initial >>= return . Skip . ParseChunksBuf src st []
1056+
1057+ -- XXX we just discard any leftover input at the end
1058+ stepOuter _ (ParseChunksInitLeftOver _) = return Stop
1059+
1060+ -- Buffer is empty process elements from the stream
1061+ stepOuter gst (ParseChunksStream st buf pst) = do
1062+ r <- step (adaptState gst) st
1063+ case r of
1064+ Yield x s -> do
1065+ pRes <- pstep pst x
1066+ case pRes of
1067+ -- PR.Keep 0 pst1 -> go SPEC s [] pst1
1068+ PR.Keep n pst1 -> do
1069+ assert (n <= length (x:buf)) (return ())
1070+ let buf1 = Prelude.take n (x:buf)
1071+ return $ Skip $ ParseChunksStream s buf1 pst1
1072+ PR.Back n pst1 -> do
1073+ assert (n <= length (x:buf)) (return ())
1074+ let (src0, buf1) = splitAt n (x:buf)
1075+ src = Prelude.reverse src0
1076+ return $ Skip $ ParseChunksBuf src s buf1 pst1
1077+ PR.Hold pst1 ->
1078+ return $ Skip $ ParseChunksStream s (x:buf) pst1
1079+ PR.Halt pst1 -> do
1080+ res <- extract pst1
1081+ case res of
1082+ Left err -> error err
1083+ -- XXX (0,a) common case
1084+ Right (n, a) -> do
1085+ assert (n <= length (x:buf)) (return ())
1086+ let src = Prelude.reverse (Prelude.take n (x:buf))
1087+ return $ Skip $ ParseChunksYield a (ParseChunksInit src s)
1088+ Skip s -> return $ Skip $ ParseChunksStream s buf pst
1089+ Stop -> do
1090+ res <- extract pst
1091+ case res of
1092+ Left err -> error err
1093+ Right (n, a) -> do
1094+ let src = Prelude.reverse (Prelude.take n (buf))
1095+ return $ Skip $ ParseChunksYield a (ParseChunksInitLeftOver src)
1096+
1097+ -- go back to stream processing mode
1098+ stepOuter _ (ParseChunksBuf [] s buf pst) =
1099+ return $ Skip $ ParseChunksStream s buf pst
1100+
1101+ -- buffered processing loop
1102+ stepOuter _ (ParseChunksBuf (x:xs) s buf pst) = do
1103+ pRes <- pstep pst x
1104+ case pRes of
1105+ -- PR.Keep 0 pst1 ->
1106+ PR.Keep n pst1 -> do
1107+ assert (n <= length (x:buf)) (return ())
1108+ let buf1 = Prelude.take n (x:buf)
1109+ return $ Skip $ ParseChunksBuf xs s buf1 pst1
1110+ PR.Back n pst1 -> do
1111+ assert (n <= length (x:buf)) (return ())
1112+ let (src0, buf1) = splitAt n (x:buf)
1113+ src = Prelude.reverse src0 ++ xs
1114+ return $ Skip $ ParseChunksBuf src s buf1 pst1
1115+ PR.Hold pst1 ->
1116+ return $ Skip $ ParseChunksBuf xs s (x:buf) pst1
1117+ PR.Halt pst1 -> do
1118+ res <- extract pst1
1119+ case res of
1120+ Left err -> error err
1121+ -- XXX have (0,a) case
1122+ Right (n, a) -> do
1123+ assert (n <= length (x:buf)) (return ())
1124+ let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs
1125+ return $ Skip $ ParseChunksYield a (ParseChunksInit src s)
1126+
1127+ stepOuter _ (ParseChunksYield a next) = return $ Yield a next
1128+
9451129------------------------------------------------------------------------------
9461130-- Specialized Folds
9471131------------------------------------------------------------------------------