git.iko.soy/streamly
on master help sign in
Add array streams folds
206e353Harendra Kumar
src/Streamly/Internal/Data/Array/Stream/Fold/Foreign.hsnew file+293
compare with master
unifiedsplit
@@ -0,0 +1,293 @@
1+-- |
2+-- Module : Streamly.Internal.Data.Array.Stream.Fold.Foreign
3+-- Copyright : (c) 2021 Composewell Technologies
4+-- License : BSD-3-Clause
5+-- Maintainer : streamly@composewell.com
6+-- Stability : experimental
7+-- Portability : GHC
8+--
9+-- Fold a stream of foreign arrays. @Fold m a b@ in this module works
10+-- on a stream of "Array a" and produces an output of type @b@.
11+--
12+-- Though @Fold m a b@ in this module works on a stream of @Array a@ it is
13+-- different from @Data.Fold m (Array a) b@. While the latter works on arrays
14+-- as a whole treating them as atomic elements, the folds in this module can
15+-- work on the stream of arrays as if it is an element stream with all the
16+-- arrays coalesced together. This module allows adapting the element stream
17+-- folds in Data.Fold to correctly work on an array stream as if it is an
18+-- element stream. For example:
19+--
20+-- >>> import qualified Streamly.Data.Fold as Fold
21+-- >>> import qualified Streamly.Internal.Data.Array.Stream.Foreign as ArrayStream
22+-- >>> import qualified Streamly.Internal.Data.Array.Stream.Fold.Foreign as ArrayFold
23+-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream (arraysOf)
24+-- >>> import qualified Streamly.Prelude as Stream
25+--
26+-- >>> ArrayStream.fold (ArrayFold.fromFold (Fold.take 7 Fold.toList)) $ Stream.arraysOf 5 $ Stream.fromList "hello world"
27+-- "hello w"
28+--
29+module Streamly.Internal.Data.Array.Stream.Fold.Foreign
30+ (
31+ Fold (..)
32+
33+ -- * Construction
34+ , fromFold
35+ , fromArrayFold
36+
37+ -- * Mapping
38+ , rmapM
39+
40+ -- * Applicative
41+ , yield
42+ , yieldM
43+ , serialWith
44+
45+ -- * Monad
46+ , concatMap
47+
48+ -- * Transformation
49+ , take
50+ )
51+where
52+
53+import Control.Applicative (liftA2)
54+import Control.Exception (assert)
55+import Control.Monad.Catch (MonadThrow)
56+import Control.Monad.IO.Class (MonadIO(..))
57+import Foreign.ForeignPtr (touchForeignPtr)
58+import Foreign.Ptr (minusPtr, plusPtr)
59+import Foreign.Storable (Storable(..))
60+import GHC.ForeignPtr (ForeignPtr(..))
61+import GHC.Ptr (Ptr(..))
62+import GHC.Types (SPEC(..))
63+import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
64+import Streamly.Internal.Data.Parser.ParserD (Initial(..), Step(..))
65+import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
66+
67+import qualified Streamly.Internal.Data.Array.Foreign as Array
68+import qualified Streamly.Internal.Data.Fold as Fold
69+import qualified Streamly.Internal.Data.Parser.ParserD as ParserD
70+import qualified Streamly.Internal.Data.Parser.ParserD.Type as ParserD
71+
72+import Prelude hiding (concatMap, take)
73+
74+-- | Array stream fold.
75+--
76+-- An array stream fold is basically an array stream "Parser" that does not
77+-- fail. In case of array stream folds the count in 'Partial', 'Continue' and
78+-- 'Done' is a count of elements that includes the leftover element count in
79+-- the array that is currently being processed by the parser. If none of the
80+-- elements is consumed by the parser the count is at least the whole array
81+-- length. If the whole array is consumed by the parser then the count will be
82+-- 0.
83+--
84+-- /Pre-release/
85+--
86+newtype Fold m a b = Fold (ParserD.Parser m (Array a) b)
87+
88+-------------------------------------------------------------------------------
89+-- Constructing array stream folds from element folds and parsers
90+-------------------------------------------------------------------------------
91+
92+-- | Convert an element 'Fold' into an array stream fold.
93+--
94+-- /Pre-release/
95+{-# INLINE fromFold #-}
96+fromFold :: forall m a b. (MonadIO m, Storable a) =>
97+ Fold.Fold m a b -> Fold m a b
98+fromFold (Fold.Fold fstep finitial fextract) =
99+ Fold (ParserD.Parser step initial fextract)
100+
101+ where
102+
103+ initial = do
104+ res <- finitial
105+ return
106+ $ case res of
107+ Fold.Partial s1 -> IPartial s1
108+ Fold.Done b -> IDone b
109+
110+ step s (Array fp@(ForeignPtr start _) end) = do
111+ goArray SPEC (Ptr start) s
112+
113+ where
114+
115+ goArray !_ !cur !fs | cur >= end = do
116+ assert (cur == end) (return ())
117+ liftIO $ touchForeignPtr fp
118+ return $ Partial 0 fs
119+ goArray !_ !cur !fs = do
120+ x <- liftIO $ peek cur
121+ res <- fstep fs x
122+ let elemSize = sizeOf (undefined :: a)
123+ next = cur `plusPtr` elemSize
124+ case res of
125+ Fold.Done b ->
126+ return $ Done ((end `minusPtr` next) `div` elemSize) b
127+ Fold.Partial fs1 ->
128+ goArray SPEC next fs1
129+
130+-- | Adapt an array stream fold.
131+--
132+-- /Pre-release/
133+{-# INLINE fromArrayFold #-}
134+fromArrayFold :: forall m a b. (MonadIO m) =>
135+ Fold.Fold m (Array a) b -> Fold m a b
136+fromArrayFold f = Fold $ ParserD.fromFold f
137+
138+-------------------------------------------------------------------------------
139+-- Functor
140+-------------------------------------------------------------------------------
141+
142+-- | Maps a function over the result of fold.
143+--
144+-- /Pre-release/
145+instance Functor m => Functor (Fold m a) where
146+ {-# INLINE fmap #-}
147+ fmap f (Fold p) = Fold $ fmap f p
148+
149+-- | Map a monadic function on the output of a fold.
150+--
151+-- /Pre-release/
152+{-# INLINE rmapM #-}
153+rmapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
154+rmapM f (Fold p) = Fold $ ParserD.rmapM f p
155+
156+-------------------------------------------------------------------------------
157+-- Sequential applicative
158+-------------------------------------------------------------------------------
159+
160+-- | A fold that always yields a pure value without consuming any input.
161+--
162+-- /Pre-release/
163+--
164+{-# INLINE yield #-}
165+yield :: Monad m => b -> Fold m a b
166+yield = Fold . ParserD.yield
167+
168+-- | A fold that always yields the result of an effectful action without
169+-- consuming any input.
170+--
171+-- /Pre-release/
172+--
173+{-# INLINE yieldM #-}
174+yieldM :: Monad m => m b -> Fold m a b
175+yieldM = Fold . ParserD.yieldM
176+
177+-- | Applies two folds sequentially on the input stream and combines their
178+-- results using the supplied function.
179+--
180+-- /Pre-release/
181+{-# INLINE serial_ #-}
182+serial_ :: MonadThrow m => Fold m x a -> Fold m x b -> Fold m x b
183+serial_ (Fold p1) (Fold p2) = Fold $ ParserD.noErrorUnsafeSplit_ p1 p2
184+
185+-- | Applies two folds sequentially on the input stream and combines their
186+-- results using the supplied function.
187+--
188+-- /Pre-release/
189+{-# INLINE serialWith #-}
190+serialWith :: MonadThrow m
191+ => (a -> b -> c) -> Fold m x a -> Fold m x b -> Fold m x c
192+serialWith f (Fold p1) (Fold p2) =
193+ Fold $ ParserD.noErrorUnsafeSplitWith f p1 p2
194+
195+-- | 'Applicative' form of 'serialWith'.
196+-- > (<*>) = serialWith id
197+instance MonadThrow m => Applicative (Fold m a) where
198+ {-# INLINE pure #-}
199+ pure = yield
200+
201+ {-# INLINE (<*>) #-}
202+ (<*>) = serialWith id
203+
204+ {-# INLINE (*>) #-}
205+ (*>) = serial_
206+
207+#if MIN_VERSION_base(4,10,0)
208+ {-# INLINE liftA2 #-}
209+ liftA2 f x = (<*>) (fmap f x)
210+#endif
211+
212+-------------------------------------------------------------------------------
213+-- Monad
214+-------------------------------------------------------------------------------
215+
216+-- | Applies a fold on the input stream, generates the next fold from the
217+-- output of the previously applied fold and then applies that fold.
218+--
219+-- /Pre-release/
220+--
221+{-# INLINE concatMap #-}
222+concatMap :: MonadThrow m =>
223+ (b -> Fold m a c) -> Fold m a b -> Fold m a c
224+concatMap func (Fold p) =
225+ Fold $ ParserD.noErrorUnsafeConcatMap (\x -> let Fold y = func x in y) p
226+
227+-- | Monad instance applies folds sequentially. Next fold can depend on the
228+-- output of the previous fold. See 'concatMap'.
229+--
230+-- > (>>=) = flip concatMap
231+instance MonadThrow m => Monad (Fold m a) where
232+ {-# INLINE return #-}
233+ return = pure
234+
235+ {-# INLINE (>>=) #-}
236+ (>>=) = flip concatMap
237+
238+ {-# INLINE (>>) #-}
239+ (>>) = (*>)
240+
241+-------------------------------------------------------------------------------
242+-- Array to Array folds
243+-------------------------------------------------------------------------------
244+
245+{-# INLINE take #-}
246+take :: forall m a b. (Monad m, Storable a) => Int -> Fold m a b -> Fold m a b
247+take n (Fold (ParserD.Parser step1 initial1 extract1)) =
248+ Fold $ ParserD.Parser step initial extract
249+
250+ where
251+
252+ initial = do
253+ res <- initial1
254+ case res of
255+ IPartial s ->
256+ if n > 0
257+ then return $ IPartial $ Tuple' n s
258+ else IDone <$> extract1 s
259+ IDone b -> return $ IDone b
260+ IError err -> return $ IError err
261+
262+ step (Tuple' i r) arr = do
263+ let len = Array.length arr
264+ if len <= i
265+ then do
266+ res <- step1 r arr
267+ case res of
268+ Partial j s ->
269+ if (i + j - len) > 0
270+ then return $ Partial j (Tuple' (i + j - len) s)
271+ else Done j <$> extract1 s
272+ Continue j s ->
273+ if (i + j - len) > 0
274+ then return $ Continue j (Tuple' (i + j - len) s)
275+ else Done j <$> extract1 s
276+ Done j b -> return $ Done j b
277+ Error err -> return $ Error err
278+ else do
279+ let !(Array (ForeignPtr start contents) _) = arr
280+ sz = sizeOf (undefined :: a)
281+ end = Ptr start `plusPtr` (i * sz)
282+ arr1 = Array (ForeignPtr start contents) end
283+ remaining = len - i
284+ res <- step1 r arr1
285+ case res of
286+ Partial 0 s -> Done remaining <$> extract1 s
287+ Partial j s -> return $ Partial (remaining + j) (Tuple' j s)
288+ Continue 0 s -> Done remaining <$> extract1 s
289+ Continue j s -> return $ Continue (remaining + j) (Tuple' j s)
290+ Done j b -> return $ Done (remaining + j) b
291+ Error err -> return $ Error err
292+
293+ extract (Tuple' _ r) = extract1 r
src/Streamly/Internal/Data/Array/Stream/Foreign.hs+310−5
@@ -22,6 +22,10 @@ module Streamly.Internal.Data.Array.Stream.Foreign
22 , unlines
22 , unlines
23
23
24 -- * Elimination
24 -- * Elimination
25+ , fold
26+ -- , parse
27+ , parseD
28+ , foldMany
25 , toArray
29 , toArray
26
30
27 -- * Compaction
31 -- * Compaction
@@ -39,30 +43,40 @@ where
39
43
40#include "inline.hs"
44#include "inline.hs"
41
45
46+import Control.Exception (assert)
47+import Control.Monad.Catch (MonadThrow, throwM)
42import Control.Monad.IO.Class (MonadIO(..))
48import Control.Monad.IO.Class (MonadIO(..))
43import Data.Word (Word8)
49import Data.Word (Word8)
44import Foreign.ForeignPtr (withForeignPtr, touchForeignPtr)
50import Foreign.ForeignPtr (withForeignPtr, touchForeignPtr)
45import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
51import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
46import Foreign.Ptr (minusPtr, plusPtr, castPtr)
52import Foreign.Ptr (minusPtr, plusPtr, castPtr)
47import Foreign.Storable (Storable(..))
53import Foreign.Storable (Storable(..))
54+import Fusion.Plugin.Types (Fuse(..))
55+import GHC.Exts (SpecConstrAnnotation(..))
48import GHC.ForeignPtr (ForeignPtr(..))
56import GHC.ForeignPtr (ForeignPtr(..))
49import GHC.Ptr (Ptr(..))
57import GHC.Ptr (Ptr(..))
58+import GHC.Types (SPEC(..))
50−import Prelude hiding (length, null, last, map, (!!), read, concat, unlines)
59+import Prelude hiding (null, last, (!!), read, concat, unlines)
51
60
52#if !defined(mingw32_HOST_OS)
61#if !defined(mingw32_HOST_OS)
53import Streamly.Internal.FileSystem.FDIO (IOVec(..))
62import Streamly.Internal.FileSystem.FDIO (IOVec(..))
54#endif
63#endif
55−import Streamly.Internal.Data.Array.Foreign.Type (Array(..), length)
64+import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
56import Streamly.Internal.Data.Fold.Type (Fold(..))
65import Streamly.Internal.Data.Fold.Type (Fold(..))
66+import Streamly.Internal.Data.Parser (ParseError(..))
57import Streamly.Internal.Data.Stream.Serial (SerialT)
67import Streamly.Internal.Data.Stream.Serial (SerialT)
58import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)
68import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)
59−import Streamly.Internal.Data.SVar (adaptState)
69+import Streamly.Internal.Data.SVar (adaptState, defState)
60
70
61import qualified Streamly.Internal.Data.Array.Foreign as A
71import qualified Streamly.Internal.Data.Array.Foreign as A
72+import qualified Streamly.Internal.Data.Array.Foreign as Array
62import qualified Streamly.Internal.Data.Array.Foreign.Type as A
73import qualified Streamly.Internal.Data.Array.Foreign.Type as A
63import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
74import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
64import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS
75import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS
76+import qualified Streamly.Internal.Data.Array.Stream.Fold.Foreign as ASF
65import qualified Streamly.Internal.Data.Fold.Type as FL
77import qualified Streamly.Internal.Data.Fold.Type as FL
78+import qualified Streamly.Internal.Data.Parser as PR
79+import qualified Streamly.Internal.Data.Parser.ParserD as PRD
66import qualified Streamly.Internal.Data.Stream.IsStream as S
80import qualified Streamly.Internal.Data.Stream.IsStream as S
67import qualified Streamly.Internal.Data.Stream.StreamD as D
81import qualified Streamly.Internal.Data.Stream.StreamD as D
68
82
@@ -300,6 +314,56 @@ splitOnSuffix byte s =
300-- Elimination
314-- Elimination
301-------------------------------------------------------------------------------
315-------------------------------------------------------------------------------
302
316
317+-- When we have to take an array partially, take the last part of the array.
318+{-# INLINE takeArrayListRev #-}
319+takeArrayListRev :: forall a. Storable a => Int -> [Array a] -> [Array a]
320+takeArrayListRev = go
321+
322+ where
323+
324+ go _ [] = []
325+ go n _ | n <= 0 = []
326+ go n (x:xs) =
327+ let len = Array.length x
328+ in if n > len
329+ then x : go (n - len) xs
330+ else if n == len
331+ then [x]
332+ else let !(Array (ForeignPtr _ contents) end) = x
333+ sz = sizeOf (undefined :: a)
334+ !(Ptr start) = end `plusPtr` negate (n * sz)
335+ in [Array (ForeignPtr start contents) end]
336+
337+-- When we have to take an array partially, take the last part of the array in
338+-- the first split.
339+{-# INLINE splitAtArrayListRev #-}
340+splitAtArrayListRev :: forall a. Storable a =>
341+ Int -> [Array a] -> ([Array a],[Array a])
342+splitAtArrayListRev n ls
343+ | n <= 0 = ([], ls)
344+ | otherwise = go n ls
345+ where
346+ go :: Int -> [Array a] -> ([Array a], [Array a])
347+ go _ [] = ([], [])
348+ go m (x:xs) =
349+ let len = Array.length x
350+ (xs', xs'') = go (m - len) xs
351+ in if m > len
352+ then (x:xs', xs'')
353+ else if m == len
354+ then ([x],xs)
355+ else let !(Array (ForeignPtr start contents) end) = x
356+ sz = sizeOf (undefined :: a)
357+ end1 = end `plusPtr` negate (m * sz)
358+ arr2 = Array (ForeignPtr start contents) end1
359+ !(Ptr addrEnd1) = end1
360+ arr1 = Array (ForeignPtr addrEnd1 contents) end
361+ in ([arr1], arr2:xs)
362+
363+-------------------------------------------------------------------------------
364+-- Fold to a single Array
365+-------------------------------------------------------------------------------
366+
303-- XXX Both of these implementations of splicing seem to perform equally well.
367-- XXX Both of these implementations of splicing seem to perform equally well.
304-- We need to perform benchmarks over a range of sizes though.
368-- We need to perform benchmarks over a range of sizes though.
305
369
@@ -326,7 +390,7 @@ _spliceArrays :: (MonadIO m, Storable a)
326 => SerialT m (Array a) -> m (Array a)
390 => SerialT m (Array a) -> m (Array a)
327_spliceArrays s = do
391_spliceArrays s = do
328 buffered <- S.foldr S.cons S.nil s
392 buffered <- S.foldr S.cons S.nil s
329− len <- S.sum (S.map length buffered)
393+ len <- S.sum (S.map Array.length buffered)
330 arr <- liftIO $ MA.newArray len
394 arr <- liftIO $ MA.newArray len
331 end <- S.foldlM' writeArr (return $ MA.aEnd arr) s
395 end <- S.foldlM' writeArr (return $ MA.aEnd arr) s
332 return $ A.unsafeFreeze $ arr {MA.aEnd = end}
396 return $ A.unsafeFreeze $ arr {MA.aEnd = end}
@@ -344,7 +408,7 @@ _spliceArraysBuffered :: (MonadIO m, Storable a)
344 => SerialT m (Array a) -> m (Array a)
408 => SerialT m (Array a) -> m (Array a)
345_spliceArraysBuffered s = do
409_spliceArraysBuffered s = do
346 buffered <- S.foldr S.cons S.nil s
410 buffered <- S.foldr S.cons S.nil s
347− len <- S.sum (S.map length buffered)
411+ len <- S.sum (S.map Array.length buffered)
348 A.unsafeFreeze <$> spliceArraysLenUnsafe len (S.map A.unsafeThaw s)
412 A.unsafeFreeze <$> spliceArraysLenUnsafe len (S.map A.unsafeThaw s)
349
413
350{-# INLINE spliceArraysRealloced #-}
414{-# INLINE spliceArraysRealloced #-}
@@ -357,6 +421,8 @@ spliceArraysRealloced s = do
357 arr <- S.foldlM' MA.spliceWithDoubling idst (S.map A.unsafeThaw s)
421 arr <- S.foldlM' MA.spliceWithDoubling idst (S.map A.unsafeThaw s)
358 liftIO $ A.unsafeFreeze <$> MA.shrinkToFit arr
422 liftIO $ A.unsafeFreeze <$> MA.shrinkToFit arr
359
423
424+-- XXX This should just be "fold A.write"
425+--
360-- | Given a stream of arrays, splice them all together to generate a single
426-- | Given a stream of arrays, splice them all together to generate a single
361-- array. The stream must be /finite/.
427-- array. The stream must be /finite/.
362--
428--
@@ -385,3 +451,242 @@ _toArraysOf :: (MonadIO m, Storable a)
385 => Int -> Fold m a (SerialT Identity (Array a))
451 => Int -> Fold m a (SerialT Identity (Array a))
386_toArraysOf n = FL.chunksOf n (A.writeNF n) FL.toStream
452_toArraysOf n = FL.chunksOf n (A.writeNF n) FL.toStream
387-}
453-}
454+
455+-------------------------------------------------------------------------------
456+-- Parsing
457+-------------------------------------------------------------------------------
458+
459+-- GHC parser does not accept {-# ANN type [] NoSpecConstr #-}, so we need
460+-- to make a newtype.
461+{-# ANN type List NoSpecConstr #-}
462+newtype List a = List {getList :: [a]}
463+
464+{-# INLINE_NORMAL parseD #-}
465+parseD ::
466+ forall m a b. (MonadIO m, MonadThrow m, Storable a)
467+ => PRD.Parser m (Array a) b
468+ -> D.Stream m (Array.Array a)
469+ -> m (b, D.Stream m (Array.Array a))
470+parseD (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do
471+ res <- initial
472+ case res of
473+ PRD.IPartial s -> go SPEC state (List []) s
474+ PRD.IDone b -> return (b, stream)
475+ PRD.IError err -> throwM $ ParseError err
476+
477+ where
478+
479+ -- "backBuf" contains last few items in the stream that we may have to
480+ -- backtrack to.
481+ --
482+ -- XXX currently we are using a dumb list based approach for backtracking
483+ -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
484+ -- That will allow us more efficient random back and forth movement.
485+ {-# INLINE go #-}
486+ go !_ st backBuf !pst = do
487+ r <- step defState st
488+ case r of
489+ D.Yield x s -> gobuf SPEC [x] s backBuf pst
490+ D.Skip s -> go SPEC s backBuf pst
491+ D.Stop -> do
492+ b <- extract pst
493+ return (b, D.nil)
494+
495+ gobuf !_ [] s backBuf !pst = go SPEC s backBuf pst
496+ gobuf !_ (x:xs) s backBuf !pst = do
497+ pRes <- pstep pst x
498+ case pRes of
499+ PR.Partial 0 pst1 ->
500+ gobuf SPEC xs s (List []) pst1
501+ PR.Partial n pst1 -> do
502+ assert
503+ (n <= sum (map Array.length (x:getList backBuf)))
504+ (return ())
505+ let src0 = takeArrayListRev n (x:getList backBuf)
506+ src = Prelude.reverse src0 ++ xs
507+ gobuf SPEC src s (List []) pst1
508+ PR.Continue 0 pst1 ->
509+ gobuf SPEC xs s (List (x:getList backBuf)) pst1
510+ PR.Continue n pst1 -> do
511+ assert
512+ (n <= sum (map Array.length (x:getList backBuf)))
513+ (return ())
514+ let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)
515+ src = Prelude.reverse src0 ++ xs
516+ gobuf SPEC src s (List buf1) pst1
517+ PR.Done 0 b ->
518+ return (b, D.Stream step s)
519+ PR.Done n b -> do
520+ assert
521+ (n <= sum (map Array.length (x:getList backBuf)))
522+ (return ())
523+ let src0 = takeArrayListRev n (x:getList backBuf)
524+ src = Prelude.reverse src0 ++ xs
525+ return (b, D.append (D.fromList src) (D.Stream step s))
526+ PR.Error err -> throwM $ ParseError err
527+
528+{-
529+-- | Parse an array stream using the supplied 'Parser'. Returns the parse
530+-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
531+--
532+-- /Internal/
533+--
534+{-# INLINE parse #-}
535+parse ::
536+ (MonadIO m, MonadThrow m, Storable a)
537+ => PRD.Parser m a b
538+ -> SerialT m (A.Array a)
539+ -> m (b, SerialT m (A.Array a))
540+parse p s = fmap D.fromStreamD <$> parseD p (D.toStreamD s)
541+-}
542+
543+-- | Fold an array stream using the supplied array stream 'Fold'.
544+--
545+-- /Pre-release/
546+--
547+{-# INLINE fold #-}
548+fold :: (MonadIO m, MonadThrow m, Storable a) =>
549+ ASF.Fold m a b -> SerialT m (A.Array a) -> m b
550+fold (ASF.Fold p) s = fst <$> parseD p (D.toStreamD s)
551+
552+{-# ANN type ParseChunksState Fuse #-}
553+data ParseChunksState x inpBuf st pst =
554+ ParseChunksInit inpBuf st
555+ | ParseChunksInitLeftOver inpBuf
556+ | ParseChunksStream st inpBuf !pst
557+ | ParseChunksBuf inpBuf st inpBuf !pst
558+ | ParseChunksYield x (ParseChunksState x inpBuf st pst)
559+
560+{-# INLINE_NORMAL foldManyD #-}
561+foldManyD
562+ :: (MonadThrow m, Storable a)
563+ => ASF.Fold m a b
564+ -> D.Stream m (Array a)
565+ -> D.Stream m b
566+foldManyD (ASF.Fold (PRD.Parser pstep initial extract)) (D.Stream step state) =
567+ D.Stream stepOuter (ParseChunksInit [] state)
568+
569+ where
570+
571+ {-# INLINE_LATE stepOuter #-}
572+ -- Buffer is empty, get the first element from the stream, initialize the
573+ -- fold and then go to stream processing loop.
574+ stepOuter gst (ParseChunksInit [] st) = do
575+ r <- step (adaptState gst) st
576+ case r of
577+ D.Yield x s -> do
578+ res <- initial
579+ case res of
580+ PRD.IPartial ps ->
581+ return $ D.Skip $ ParseChunksBuf [x] s [] ps
582+ PRD.IDone pb ->
583+ let next = ParseChunksInit [x] s
584+ in return $ D.Skip $ ParseChunksYield pb next
585+ PRD.IError err -> throwM $ ParseError err
586+ D.Skip s -> return $ D.Skip $ ParseChunksInit [] s
587+ D.Stop -> return D.Stop
588+
589+ -- Buffer is not empty, go to buffered processing loop
590+ stepOuter _ (ParseChunksInit src st) = do
591+ res <- initial
592+ case res of
593+ PRD.IPartial ps ->
594+ return $ D.Skip $ ParseChunksBuf src st [] ps
595+ PRD.IDone pb ->
596+ let next = ParseChunksInit src st
597+ in return $ D.Skip $ ParseChunksYield pb next
598+ PRD.IError err -> throwM $ ParseError err
599+
600+ -- XXX we just discard any leftover input at the end
601+ stepOuter _ (ParseChunksInitLeftOver _) = return D.Stop
602+
603+ -- Buffer is empty, process elements from the stream
604+ stepOuter gst (ParseChunksStream st backBuf pst) = do
605+ r <- step (adaptState gst) st
606+ case r of
607+ D.Yield x s -> do
608+ pRes <- pstep pst x
609+ case pRes of
610+ PR.Partial 0 pst1 ->
611+ return $ D.Skip $ ParseChunksStream s [] pst1
612+ PR.Partial n pst1 -> do
613+ assert
614+ (n <= sum (map Array.length (x:backBuf)))
615+ (return ())
616+ let src0 = takeArrayListRev n (x:backBuf)
617+ src = Prelude.reverse src0
618+ return $ D.Skip $ ParseChunksBuf src s [] pst1
619+ PR.Continue 0 pst1 ->
620+ return $ D.Skip $ ParseChunksStream s (x:backBuf) pst1
621+ PR.Continue n pst1 -> do
622+ assert
623+ (n <= sum (map Array.length (x:backBuf)))
624+ (return ())
625+ let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
626+ src = Prelude.reverse src0
627+ return $ D.Skip $ ParseChunksBuf src s buf1 pst1
628+ PR.Done 0 b -> do
629+ return $ D.Skip $
630+ ParseChunksYield b (ParseChunksInit [] s)
631+ PR.Done n b -> do
632+ assert
633+ (n <= sum (map Array.length (x:backBuf)))
634+ (return ())
635+ let src0 = takeArrayListRev n (x:backBuf)
636+ let src = Prelude.reverse src0
637+ return $ D.Skip $
638+ ParseChunksYield b (ParseChunksInit src s)
639+ PR.Error err -> throwM $ ParseError err
640+ D.Skip s -> return $ D.Skip $ ParseChunksStream s backBuf pst
641+ D.Stop -> do
642+ b <- extract pst
643+ let src = Prelude.reverse backBuf
644+ return $ D.Skip $
645+ ParseChunksYield b (ParseChunksInitLeftOver src)
646+
647+ -- go back to stream processing mode
648+ stepOuter _ (ParseChunksBuf [] s buf pst) =
649+ return $ D.Skip $ ParseChunksStream s buf pst
650+
651+ -- buffered processing loop
652+ stepOuter _ (ParseChunksBuf (x:xs) s backBuf pst) = do
653+ pRes <- pstep pst x
654+ case pRes of
655+ PR.Partial 0 pst1 ->
656+ return $ D.Skip $ ParseChunksBuf xs s [] pst1
657+ PR.Partial n pst1 -> do
658+ assert (n <= sum (map Array.length (x:backBuf))) (return ())
659+ let src0 = takeArrayListRev n (x:backBuf)
660+ src = Prelude.reverse src0 ++ xs
661+ return $ D.Skip $ ParseChunksBuf src s [] pst1
662+ PR.Continue 0 pst1 ->
663+ return $ D.Skip $ ParseChunksBuf xs s (x:backBuf) pst1
664+ PR.Continue n pst1 -> do
665+ assert (n <= sum (map Array.length (x:backBuf))) (return ())
666+ let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
667+ src = Prelude.reverse src0 ++ xs
668+ return $ D.Skip $ ParseChunksBuf src s buf1 pst1
669+ PR.Done 0 b ->
670+ return $ D.Skip $ ParseChunksYield b (ParseChunksInit xs s)
671+ PR.Done n b -> do
672+ assert (n <= sum (map Array.length (x:backBuf))) (return ())
673+ let src0 = takeArrayListRev n (x:backBuf)
674+ src = Prelude.reverse src0 ++ xs
675+ return $ D.Skip $ ParseChunksYield b (ParseChunksInit src s)
676+ PR.Error err -> throwM $ ParseError err
677+
678+ stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next
679+
680+-- | Apply an array stream 'Fold' repeatedly on an array stream and emit the
681+-- fold outputs in the output stream.
682+--
683+-- See "Streamly.Prelude.foldMany" for more details.
684+--
685+-- /Pre-release/
686+{-# INLINE foldMany #-}
687+foldMany
688+ :: (IsStream t, MonadThrow m, Storable a)
689+ => ASF.Fold m a b
690+ -> t m (Array a)
691+ -> t m b
692+foldMany p m = D.fromStreamD $ foldManyD p (D.toStreamD m)
src/Streamly/Internal/Data/Parser/ParserD.hs+1
@@ -16,6 +16,7 @@ module Streamly.Internal.Data.Parser.ParserD
16 , ParseError (..)
16 , ParseError (..)
17 , Step (..)
17 , Step (..)
18 , Initial (..)
18 , Initial (..)
19+ , rmapM
19
20
20 -- First order parsers
21 -- First order parsers
21 -- * Accumulators
22 -- * Accumulators
src/Streamly/Internal/Data/Parser/ParserD/Type.hs+161−9
@@ -115,6 +115,7 @@ module Streamly.Internal.Data.Parser.ParserD.Type
115 , Step (..)
115 , Step (..)
116 , Parser (..)
116 , Parser (..)
117 , ParseError (..)
117 , ParseError (..)
118+ , rmapM
118
119
119 , yield
120 , yield
120 , yieldM
121 , yieldM
@@ -129,13 +130,15 @@ module Streamly.Internal.Data.Parser.ParserD.Type
129 , alt
130 , alt
130 , concatMap
131 , concatMap
131
132
133+ , noErrorUnsafeSplit_
132 , noErrorUnsafeSplitWith
134 , noErrorUnsafeSplitWith
135+ , noErrorUnsafeConcatMap
133 )
136 )
134where
137where
135
138
136import Control.Applicative (Alternative(..), liftA2)
139import Control.Applicative (Alternative(..), liftA2)
137import Control.Exception (assert, Exception(..))
140import Control.Exception (assert, Exception(..))
138−import Control.Monad (MonadPlus(..))
141+import Control.Monad (MonadPlus(..), (>=>))
139import Control.Monad.Catch (MonadCatch, try, throwM, MonadThrow)
142import Control.Monad.Catch (MonadCatch, try, throwM, MonadThrow)
140import Data.Bifunctor (Bifunctor(..))
143import Data.Bifunctor (Bifunctor(..))
141import Fusion.Plugin.Types (Fuse(..))
144import Fusion.Plugin.Types (Fuse(..))
@@ -264,6 +267,18 @@ instance Functor (Step s) where
264 fmap f (Done n b) = Done n (f b)
267 fmap f (Done n b) = Done n (f b)
265 fmap _ (Error err) = Error err
268 fmap _ (Error err) = Error err
266
269
270+-- | Map a monadic function over the result @b@ in @Step s b@.
271+--
272+-- /Internal/
273+{-# INLINE mapMStep #-}
274+mapMStep :: Applicative m => (a -> m b) -> Step s a -> m (Step s b)
275+mapMStep f res =
276+ case res of
277+ Partial n s -> pure $ Partial n s
278+ Done n b -> Done n <$> f b
279+ Continue n s -> pure $ Continue n s
280+ Error err -> pure $ Error err
281+
267-- | A parser is a fold that can fail and is represented as @Parser step
282-- | A parser is a fold that can fail and is represented as @Parser step
268-- initial extract@. Before we drive a parser we call the @initial@ action to
283-- initial extract@. Before we drive a parser we call the @initial@ action to
269-- retrieve the initial state of the fold. The parser driver invokes @step@
284-- retrieve the initial state of the fold. The parser driver invokes @step@
@@ -303,6 +318,27 @@ instance Functor m => Functor (Parser m a) where
303 step s b = fmap2 f (step1 s b)
318 step s b = fmap2 f (step1 s b)
304 fmap2 g = fmap (fmap g)
319 fmap2 g = fmap (fmap g)
305
320
321+------------------------------------------------------------------------------
322+-- Mapping on the output
323+------------------------------------------------------------------------------
324+
325+-- | Map a monadic function on the output of a parser.
326+--
327+-- /Pre-release/
328+{-# INLINE rmapM #-}
329+rmapM :: Monad m => (b -> m c) -> Parser m a b -> Parser m a c
330+rmapM f (Parser step initial extract) = Parser step1 initial1 (extract >=> f)
331+
332+ where
333+
334+ initial1 = do
335+ res <- initial
336+ case res of
337+ IPartial x -> return $ IPartial x
338+ IDone a -> IDone <$> f a
339+ IError err -> return $ IError err
340+ step1 s a = step s a >>= mapMStep f
341+
306-- | See 'Streamly.Internal.Data.Parser.yield'.
342-- | See 'Streamly.Internal.Data.Parser.yield'.
307--
343--
308-- /Pre-release/
344-- /Pre-release/
@@ -454,7 +490,6 @@ noErrorUnsafeSplitWith func (Parser stepL initialL extractL)
454 IError err -> error $ "noErrorUnsafeSplitWith: cannot use a "
490 IError err -> error $ "noErrorUnsafeSplitWith: cannot use a "
455 ++ "failing parser. Parser failed with: " ++ err
491 ++ "failing parser. Parser failed with: " ++ err
456
492
457−
458{-# ANN type SeqAState Fuse #-}
493{-# ANN type SeqAState Fuse #-}
459data SeqAState sl sr = SeqAL sl | SeqAR sr
494data SeqAState sl sr = SeqAL sl | SeqAR sr
460
495
@@ -484,19 +519,75 @@ split_ (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
484
519
485 -- Note: For the composed parse to terminate, the left parser has to be
520 -- Note: For the composed parse to terminate, the left parser has to be
486 -- a terminating parser returning a Done at some point.
521 -- a terminating parser returning a Done at some point.
487− step (SeqAL st) a =
522+ step (SeqAL st) a = do
523+ -- Important: Do not use Applicative here. Applicative somehow caused
524+ -- the right action to run many times, not sure why though.
525+ resL <- stepL st a
488− (\resL initR -> case resL of
526+ case resL of
489 -- Note: this leads to buffering even if we are not in an
527 -- Note: this leads to buffering even if we are not in an
490 -- Alternative composition.
528 -- Alternative composition.
491− Partial n s -> Continue n (SeqAL s)
529+ Partial n s -> return $ Continue n (SeqAL s)
492− Continue n s -> Continue n (SeqAL s)
530+ Continue n s -> return $ Continue n (SeqAL s)
493 Done n _ -> do
531 Done n _ -> do
532+ initR <- initialR
494− case initR of
533+ return $ case initR of
495 IPartial s -> Continue n (SeqAR s)
534 IPartial s -> Continue n (SeqAR s)
496 IDone b -> Done n b
535 IDone b -> Done n b
497 IError err -> Error err
536 IError err -> Error err
498− -- XXX should we sequence initialR monadically?
537+ Error err -> return $ Error err
538+
539+ step (SeqAR st) a =
540+ (\case
541+ Partial n s -> Partial n (SeqAR s)
542+ Continue n s -> Continue n (SeqAR s)
543+ Done n b -> Done n b
499− Error err -> Error err) <$> stepL st a <*> initialR
544+ Error err -> Error err) <$> stepR st a
545+
546+ extract (SeqAR sR) = extractR sR
547+ extract (SeqAL sL) = do
548+ _ <- extractL sL
549+ res <- initialR
550+ case res of
551+ IPartial sR -> extractR sR
552+ IDone rR -> return rR
553+ IError err -> throwM $ ParseError err
554+
555+{-# INLINE noErrorUnsafeSplit_ #-}
556+noErrorUnsafeSplit_ :: MonadThrow m => Parser m x a -> Parser m x b -> Parser m x b
557+noErrorUnsafeSplit_ (Parser stepL initialL extractL) (Parser stepR initialR extractR) =
558+ Parser step initial extract
559+
560+ where
561+
562+ initial = do
563+ resL <- initialL
564+ case resL of
565+ IPartial sl -> return $ IPartial $ SeqAL sl
566+ IDone _ -> do
567+ resR <- initialR
568+ return $ case resR of
569+ IPartial sr -> IPartial $ SeqAR sr
570+ IDone br -> IDone br
571+ IError err -> IError err
572+ IError err -> return $ IError err
573+
574+ -- Note: For the composed parse to terminate, the left parser has to be
575+ -- a terminating parser returning a Done at some point.
576+ step (SeqAL st) a = do
577+ -- Important: Please do not use Applicative here. Applicative somehow
578+ -- caused the next action to run many times in the "tar" parsing code,
579+ -- not sure why though.
580+ resL <- stepL st a
581+ case resL of
582+ Partial n s -> return $ Partial n (SeqAL s)
583+ Continue n s -> return $ Continue n (SeqAL s)
584+ Done n _ -> do
585+ initR <- initialR
586+ return $ case initR of
587+ IPartial s -> Partial n (SeqAR s)
588+ IDone b -> Done n b
589+ IError err -> Error err
590+ Error err -> return $ Error err
500
591
501 step (SeqAR st) a =
592 step (SeqAR st) a =
502 (\case
593 (\case
@@ -923,6 +1014,64 @@ concatMap func (Parser stepL initialL extractL) = Parser step initial extract
923 extract (ConcatParseR _ s extractR) = extractR s
1014 extract (ConcatParseR _ s extractR) = extractR s
924 extract (ConcatParseL sL) = extractL sL >>= extractP . func
1015 extract (ConcatParseL sL) = extractL sL >>= extractP . func
925
1016
1017+{-# INLINE noErrorUnsafeConcatMap #-}
1018+noErrorUnsafeConcatMap :: MonadThrow m =>
1019+ (b -> Parser m a c) -> Parser m a b -> Parser m a c
1020+noErrorUnsafeConcatMap func (Parser stepL initialL extractL) =
1021+ Parser step initial extract
1022+
1023+ where
1024+
1025+ {-# INLINE initializeR #-}
1026+ initializeR (Parser stepR initialR extractR) = do
1027+ resR <- initialR
1028+ return $ case resR of
1029+ IPartial sr -> IPartial $ ConcatParseR stepR sr extractR
1030+ IDone br -> IDone br
1031+ IError err -> IError err
1032+
1033+ initial = do
1034+ res <- initialL
1035+ case res of
1036+ IPartial s -> return $ IPartial $ ConcatParseL s
1037+ IDone b -> initializeR (func b)
1038+ IError err -> return $ IError err
1039+
1040+ {-# INLINE initializeRL #-}
1041+ initializeRL n (Parser stepR initialR extractR) = do
1042+ resR <- initialR
1043+ return $ case resR of
1044+ IPartial sr -> Partial n $ ConcatParseR stepR sr extractR
1045+ IDone br -> Done n br
1046+ IError err -> Error err
1047+
1048+ step (ConcatParseL st) a = do
1049+ r <- stepL st a
1050+ case r of
1051+ Partial n s -> return $ Partial n (ConcatParseL s)
1052+ Continue n s -> return $ Continue n (ConcatParseL s)
1053+ Done n b -> initializeRL n (func b)
1054+ Error err -> return $ Error err
1055+
1056+ step (ConcatParseR stepR st extractR) a = do
1057+ r <- stepR st a
1058+ return $ case r of
1059+ Partial n s -> Partial n $ ConcatParseR stepR s extractR
1060+ Continue n s -> Continue n $ ConcatParseR stepR s extractR
1061+ Done n b -> Done n b
1062+ Error err -> Error err
1063+
1064+ {-# INLINE extractP #-}
1065+ extractP (Parser _ initialR extractR) = do
1066+ res <- initialR
1067+ case res of
1068+ IPartial s -> extractR s
1069+ IDone b -> return b
1070+ IError err -> throwM $ ParseError err
1071+
1072+ extract (ConcatParseR _ s extractR) = extractR s
1073+ extract (ConcatParseL sL) = extractL sL >>= extractP . func
1074+
926-- Note: The monad instance has quadratic performance complexity. It works fine
1075-- Note: The monad instance has quadratic performance complexity. It works fine
927-- for small number of compositions but for a scalable implementation we need a
1076-- for small number of compositions but for a scalable implementation we need a
928-- CPS version.
1077-- CPS version.
@@ -936,6 +1085,9 @@ instance MonadThrow m => Monad (Parser m a) where
936 {-# INLINE (>>=) #-}
1085 {-# INLINE (>>=) #-}
937 (>>=) = flip concatMap
1086 (>>=) = flip concatMap
938
1087
1088+ {-# INLINE (>>) #-}
1089+ (>>) = (*>)
1090+
939-- | See documentation of 'Streamly.Internal.Data.Parser.ParserK.Type.Parser'.
1091-- | See documentation of 'Streamly.Internal.Data.Parser.ParserK.Type.Parser'.
940--
1092--
941instance MonadCatch m => MonadPlus (Parser m a) where
1093instance MonadCatch m => MonadPlus (Parser m a) where
streamly.cabal+1
@@ -481,6 +481,7 @@ library
481 , Streamly.Internal.Data.SmallArray
481 , Streamly.Internal.Data.SmallArray
482 , Streamly.Internal.Data.Array.Stream.Mut.Foreign
482 , Streamly.Internal.Data.Array.Stream.Mut.Foreign
483 , Streamly.Internal.Data.Array.Stream.Foreign
483 , Streamly.Internal.Data.Array.Stream.Foreign
484+ , Streamly.Internal.Data.Array.Stream.Fold.Foreign
484
485
485 -- Memory storage
486 -- Memory storage
486 , Streamly.Internal.Ring.Foreign
487 , Streamly.Internal.Ring.Foreign