git.iko.soy/streamly
on 206e353 help sign in
streamly/src/Streamly/Internal/Data/Array/Stream/Foreign.hs
692 lines 26 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Array.Stream.Foreign
3-- Copyright : (c) 2019 Composewell Technologies
4-- License : BSD3-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9-- Combinators to efficiently manipulate streams of immutable arrays.
10--
11module Streamly.Internal.Data.Array.Stream.Foreign
12 (
13 -- * Creation
14 arraysOf
16 -- * Flattening to elements
17 , concat
18 , concatRev
19 , interpose
20 , interposeSuffix
21 , intercalateSuffix
22 , unlines
24 -- * Elimination
25 , fold
26 -- , parse
27 , parseD
28 , foldMany
29 , toArray
31 -- * Compaction
32 , lpackArraysChunksOf
33#if !defined(mingw32_HOST_OS)
34 , groupIOVecsOf
35#endif
36 , compact
38 -- * Splitting
39 , splitOn
40 , splitOnSuffix
41 )
42where
44#include "inline.hs"
46import Control.Exception (assert)
47import Control.Monad.Catch (MonadThrow, throwM)
48import Control.Monad.IO.Class (MonadIO(..))
49import Data.Word (Word8)
50import Foreign.ForeignPtr (withForeignPtr, touchForeignPtr)
51import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
52import Foreign.Ptr (minusPtr, plusPtr, castPtr)
53import Foreign.Storable (Storable(..))
54import Fusion.Plugin.Types (Fuse(..))
55import GHC.Exts (SpecConstrAnnotation(..))
56import GHC.ForeignPtr (ForeignPtr(..))
57import GHC.Ptr (Ptr(..))
58import GHC.Types (SPEC(..))
59import Prelude hiding (null, last, (!!), read, concat, unlines)
61#if !defined(mingw32_HOST_OS)
62import Streamly.Internal.FileSystem.FDIO (IOVec(..))
63#endif
64import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
65import Streamly.Internal.Data.Fold.Type (Fold(..))
66import Streamly.Internal.Data.Parser (ParseError(..))
67import Streamly.Internal.Data.Stream.Serial (SerialT)
68import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)
69import Streamly.Internal.Data.SVar (adaptState, defState)
71import qualified Streamly.Internal.Data.Array.Foreign as A
72import qualified Streamly.Internal.Data.Array.Foreign as Array
73import qualified Streamly.Internal.Data.Array.Foreign.Type as A
74import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
75import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS
76import qualified Streamly.Internal.Data.Array.Stream.Fold.Foreign as ASF
77import qualified Streamly.Internal.Data.Fold.Type as FL
78import qualified Streamly.Internal.Data.Parser as PR
79import qualified Streamly.Internal.Data.Parser.ParserD as PRD
80import qualified Streamly.Internal.Data.Stream.IsStream as S
81import qualified Streamly.Internal.Data.Stream.StreamD as D
83-------------------------------------------------------------------------------
84-- Generation
85-------------------------------------------------------------------------------
87-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
88-- @n@ elements each.
89--
90-- > arraysOf n = Stream.chunksOf n (Array.writeN n)
91--
92-- /Pre-release/
93{-# INLINE arraysOf #-}
94arraysOf :: (IsStream t, MonadIO m, Storable a)
95 => Int -> t m a -> t m (Array a)
96arraysOf n str = D.fromStreamD $ A.arraysOf n (D.toStreamD str)
98-------------------------------------------------------------------------------
99-- Append
100-------------------------------------------------------------------------------
102-- XXX efficiently compare two streams of arrays. Two streams can have chunks
103-- of different sizes, we can handle that in the stream comparison abstraction.
104-- This could be useful e.g. to fast compare whether two files differ.
106-- | Convert a stream of arrays into a stream of their elements.
107--
108-- Same as the following but more efficient:
109--
110-- > concat = Stream.unfoldMany Array.read
111--
112-- @since 0.7.0
113{-# INLINE concat #-}
114concat :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a
115-- concat m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)
116-- concat m = D.fromStreamD $ D.concatMap A.toStreamD (D.toStreamD m)
117concat m = D.fromStreamD $ D.unfoldMany A.read (D.toStreamD m)
119-- | Convert a stream of arrays into a stream of their elements reversing the
120-- contents of each array before flattening.
121--
122-- > concatRev = Stream.unfoldMany Array.readRev
123--
124-- @since 0.7.0
125{-# INLINE concatRev #-}
126concatRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a
127concatRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)
129-------------------------------------------------------------------------------
130-- Intersperse and append
131-------------------------------------------------------------------------------
133-- | Flatten a stream of arrays after inserting the given element between
134-- arrays.
135--
136-- /Pre-release/
137{-# INLINE interpose #-}
138interpose :: (MonadIO m, IsStream t, Storable a) => a -> t m (Array a) -> t m a
139interpose x = S.interpose x A.read
141{-# INLINE intercalateSuffix #-}
142intercalateSuffix :: (MonadIO m, IsStream t, Storable a)
143 => Array a -> t m (Array a) -> t m a
144intercalateSuffix arr = S.intercalateSuffix arr A.read
146-- | Flatten a stream of arrays appending the given element after each
147-- array.
148--
149-- @since 0.7.0
150{-# INLINE interposeSuffix #-}
151interposeSuffix :: (MonadIO m, IsStream t, Storable a)
152 => a -> t m (Array a) -> t m a
153-- interposeSuffix x = D.fromStreamD . A.unlines x . D.toStreamD
154interposeSuffix x = S.interposeSuffix x A.read
156data FlattenState s a =
157 OuterLoop s
158 | InnerLoop s !(ForeignPtr a) !(Ptr a) !(Ptr a)
160{-# INLINE_NORMAL unlines #-}
161unlines :: forall m a. (MonadIO m, Storable a)
162 => a -> D.Stream m (Array a) -> D.Stream m a
163unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
164 where
165 {-# INLINE_LATE step' #-}
166 step' gst (OuterLoop st) = do
167 r <- step (adaptState gst) st
168 return $ case r of
169 D.Yield Array{..} s ->
170 let p = unsafeForeignPtrToPtr aStart
171 in D.Skip (InnerLoop s aStart p aEnd)
172 D.Skip s -> D.Skip (OuterLoop s)
173 D.Stop -> D.Stop
175 step' _ (InnerLoop st _ p end) | p == end =
176 return $ D.Yield sep $ OuterLoop st
178 step' _ (InnerLoop st startf p end) = do
179 x <- liftIO $ do
180 r <- peek p
181 touchForeignPtr startf
182 return r
183 return $ D.Yield x (InnerLoop st startf
184 (p `plusPtr` sizeOf (undefined :: a)) end)
186-------------------------------------------------------------------------------
187-- Compact
188-------------------------------------------------------------------------------
190{-# INLINE_NORMAL packArraysChunksOf #-}
191packArraysChunksOf :: (MonadIO m, Storable a)
192 => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
193packArraysChunksOf n str =
194 D.map A.unsafeFreeze $ AS.packArraysChunksOf n $ D.map A.unsafeThaw str
196-- XXX instead of writing two different versions of this operation, we should
197-- write it as a pipe.
198{-# INLINE_NORMAL lpackArraysChunksOf #-}
199lpackArraysChunksOf :: (MonadIO m, Storable a)
200 => Int -> Fold m (Array a) () -> Fold m (Array a) ()
201lpackArraysChunksOf n fld =
202 FL.map A.unsafeThaw $ AS.lpackArraysChunksOf n (FL.map A.unsafeFreeze fld)
204#if !defined(mingw32_HOST_OS)
206-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
207-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
208-- each array and a maximum of @maxEntries@ entries in each array.
209--
210-- @since 0.7.0
211{-# INLINE_NORMAL groupIOVecsOf #-}
212groupIOVecsOf :: MonadIO m
213 => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
214groupIOVecsOf n maxIOVLen str =
215 D.map A.unsafeFreeze
216 $ AS.groupIOVecsOf n maxIOVLen
217 $ D.map A.unsafeThaw str
218#endif
220-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
221-- maximum specified size in bytes.
222--
223-- @since 0.7.0
224{-# INLINE compact #-}
225compact :: (MonadIO m, Storable a)
226 => Int -> SerialT m (Array a) -> SerialT m (Array a)
227compact n xs = D.fromStreamD $ packArraysChunksOf n (D.toStreamD xs)
229-------------------------------------------------------------------------------
230-- Split
231-------------------------------------------------------------------------------
233data SplitState s arr
234 = Initial s
235 | Buffering s arr
236 | Splitting s arr
237 | Yielding arr (SplitState s arr)
238 | Finishing
240-- | Split a stream of arrays on a given separator byte, dropping the separator
241-- and coalescing all the arrays between two separators into a single array.
242--
243-- @since 0.7.0
244{-# INLINE_NORMAL _splitOn #-}
245_splitOn
246 :: MonadIO m
247 => Word8
248 -> D.Stream m (Array Word8)
249 -> D.Stream m (Array Word8)
250_splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
252 where
254 {-# INLINE_LATE step' #-}
255 step' gst (Initial st) = do
256 r <- step gst st
257 case r of
258 D.Yield arr s -> do
259 (arr1, marr2) <- A.breakOn byte arr
260 return $ case marr2 of
261 Nothing -> D.Skip (Buffering s arr1)
262 Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
263 D.Skip s -> return $ D.Skip (Initial s)
264 D.Stop -> return D.Stop
266 step' gst (Buffering st buf) = do
267 r <- step gst st
268 case r of
269 D.Yield arr s -> do
270 (arr1, marr2) <- A.breakOn byte arr
271 buf' <- A.spliceTwo buf arr1
272 return $ case marr2 of
273 Nothing -> D.Skip (Buffering s buf')
274 Just x -> D.Skip (Yielding buf' (Splitting s x))
275 D.Skip s -> return $ D.Skip (Buffering s buf)
276 D.Stop -> return $
277 if A.byteLength buf == 0
278 then D.Stop
279 else D.Skip (Yielding buf Finishing)
281 step' _ (Splitting st buf) = do
282 (arr1, marr2) <- A.breakOn byte buf
283 return $ case marr2 of
284 Nothing -> D.Skip $ Buffering st arr1
285 Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
287 step' _ (Yielding arr next) = return $ D.Yield arr next
288 step' _ Finishing = return D.Stop
290-- | Split a stream of arrays on a given separator byte, dropping the separator
291-- and coalescing all the arrays between two separators into a single array.
292--
293-- @since 0.7.0
294{-# INLINE splitOn #-}
295splitOn
296 :: (IsStream t, MonadIO m)
297 => Word8
298 -> t m (Array Word8)
299 -> t m (Array Word8)
300splitOn byte s =
301 D.fromStreamD $ D.splitInnerBy (A.breakOn byte) A.spliceTwo $ D.toStreamD s
303{-# INLINE splitOnSuffix #-}
304splitOnSuffix
305 :: (IsStream t, MonadIO m)
306 => Word8
307 -> t m (Array Word8)
308 -> t m (Array Word8)
309-- splitOn byte s = D.fromStreamD $ A.splitOn byte $ D.toStreamD s
310splitOnSuffix byte s =
311 D.fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.spliceTwo $ D.toStreamD s
313-------------------------------------------------------------------------------
314-- Elimination
315-------------------------------------------------------------------------------
317-- When we have to take an array partially, take the last part of the array.
318{-# INLINE takeArrayListRev #-}
319takeArrayListRev :: forall a. Storable a => Int -> [Array a] -> [Array a]
320takeArrayListRev = go
322 where
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]
337-- When we have to take an array partially, take the last part of the array in
338-- the first split.
339{-# INLINE splitAtArrayListRev #-}
340splitAtArrayListRev :: forall a. Storable a =>
341 Int -> [Array a] -> ([Array a],[Array a])
342splitAtArrayListRev 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)
363-------------------------------------------------------------------------------
364-- Fold to a single Array
365-------------------------------------------------------------------------------
367-- XXX Both of these implementations of splicing seem to perform equally well.
368-- We need to perform benchmarks over a range of sizes though.
370-- CAUTION! length must more than equal to lengths of all the arrays in the
371-- stream.
372{-# INLINE spliceArraysLenUnsafe #-}
373spliceArraysLenUnsafe :: (MonadIO m, Storable a)
374 => Int -> SerialT m (MA.Array a) -> m (MA.Array a)
375spliceArraysLenUnsafe len buffered = do
376 arr <- liftIO $ MA.newArray len
377 end <- S.foldlM' writeArr (return $ MA.aEnd arr) buffered
378 return $ arr {MA.aEnd = end}
380 where
382 writeArr dst (MA.Array as ae _) =
383 liftIO $ withForeignPtr as $ \src -> do
384 let count = ae `minusPtr` src
385 A.memcpy (castPtr dst) (castPtr src) count
386 return $ dst `plusPtr` count
388{-# INLINE _spliceArrays #-}
389_spliceArrays :: (MonadIO m, Storable a)
390 => SerialT m (Array a) -> m (Array a)
391_spliceArrays s = do
392 buffered <- S.foldr S.cons S.nil s
393 len <- S.sum (S.map Array.length buffered)
394 arr <- liftIO $ MA.newArray len
395 end <- S.foldlM' writeArr (return $ MA.aEnd arr) s
396 return $ A.unsafeFreeze $ arr {MA.aEnd = end}
398 where
400 writeArr dst (Array as ae) =
401 liftIO $ withForeignPtr as $ \src -> do
402 let count = ae `minusPtr` src
403 A.memcpy (castPtr dst) (castPtr src) count
404 return $ dst `plusPtr` count
406{-# INLINE _spliceArraysBuffered #-}
407_spliceArraysBuffered :: (MonadIO m, Storable a)
408 => SerialT m (Array a) -> m (Array a)
409_spliceArraysBuffered s = do
410 buffered <- S.foldr S.cons S.nil s
411 len <- S.sum (S.map Array.length buffered)
412 A.unsafeFreeze <$> spliceArraysLenUnsafe len (S.map A.unsafeThaw s)
414{-# INLINE spliceArraysRealloced #-}
415spliceArraysRealloced :: forall m a. (MonadIO m, Storable a)
416 => SerialT m (Array a) -> m (Array a)
417spliceArraysRealloced s = do
418 let idst = liftIO $ MA.newArray (A.bytesToElemCount (undefined :: a)
419 (A.mkChunkSizeKB 4))
421 arr <- S.foldlM' MA.spliceWithDoubling idst (S.map A.unsafeThaw s)
422 liftIO $ A.unsafeFreeze <$> MA.shrinkToFit arr
424-- XXX This should just be "fold A.write"
425--
426-- | Given a stream of arrays, splice them all together to generate a single
427-- array. The stream must be /finite/.
428--
429-- @since 0.7.0
430{-# INLINE toArray #-}
431toArray :: (MonadIO m, Storable a) => SerialT m (Array a) -> m (Array a)
432toArray = spliceArraysRealloced
433-- spliceArrays = _spliceArraysBuffered
435-- exponentially increasing sizes of the chunks upto the max limit.
436-- XXX this will be easier to implement with parsers/terminating folds
437-- With this we should be able to reduce the number of chunks/allocations.
438-- The reallocation/copy based toArray can also be implemented using this.
439--
440{-
441{-# INLINE toArraysInRange #-}
442toArraysInRange :: (IsStream t, MonadIO m, Storable a)
443 => Int -> Int -> Fold m (Array a) b -> Fold m a b
444toArraysInRange low high (Fold step initial extract) =
445-}
447{-
448-- | Fold the input to a pure buffered stream (List) of arrays.
449{-# INLINE _toArraysOf #-}
450_toArraysOf :: (MonadIO m, Storable a)
451 => Int -> Fold m a (SerialT Identity (Array a))
452_toArraysOf n = FL.chunksOf n (A.writeNF n) FL.toStream
453-}
455-------------------------------------------------------------------------------
456-- Parsing
457-------------------------------------------------------------------------------
459-- GHC parser does not accept {-# ANN type [] NoSpecConstr #-}, so we need
460-- to make a newtype.
461{-# ANN type List NoSpecConstr #-}
462newtype List a = List {getList :: [a]}
464{-# INLINE_NORMAL parseD #-}
465parseD ::
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))
470parseD (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
477 where
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)
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
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 #-}
535parse ::
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))
540parse p s = fmap D.fromStreamD <$> parseD p (D.toStreamD s)
541-}
543-- | Fold an array stream using the supplied array stream 'Fold'.
544--
545-- /Pre-release/
546--
547{-# INLINE fold #-}
548fold :: (MonadIO m, MonadThrow m, Storable a) =>
549 ASF.Fold m a b -> SerialT m (A.Array a) -> m b
550fold (ASF.Fold p) s = fst <$> parseD p (D.toStreamD s)
552{-# ANN type ParseChunksState Fuse #-}
553data 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)
560{-# INLINE_NORMAL foldManyD #-}
561foldManyD
562 :: (MonadThrow m, Storable a)
563 => ASF.Fold m a b
564 -> D.Stream m (Array a)
565 -> D.Stream m b
566foldManyD (ASF.Fold (PRD.Parser pstep initial extract)) (D.Stream step state) =
567 D.Stream stepOuter (ParseChunksInit [] state)
569 where
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
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
600 -- XXX we just discard any leftover input at the end
601 stepOuter _ (ParseChunksInitLeftOver _) = return D.Stop
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)
647 -- go back to stream processing mode
648 stepOuter _ (ParseChunksBuf [] s buf pst) =
649 return $ D.Skip $ ParseChunksStream s buf pst
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
678 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next
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 #-}
687foldMany
688 :: (IsStream t, MonadThrow m, Storable a)
689 => ASF.Fold m a b
690 -> t m (Array a)
691 -> t m b
692foldMany p m = D.fromStreamD $ foldManyD p (D.toStreamD m)