git.iko.soy/streamly
on 03218ad help sign in
streamly/src/Streamly/Internal/Data/Array/Stream/Foreign.hs
703 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 , fold_
27 -- , parse
28 , parseD
29 , foldMany
30 , toArray
32 -- * Compaction
33 , lpackArraysChunksOf
34#if !defined(mingw32_HOST_OS)
35 , groupIOVecsOf
36#endif
37 , compact
39 -- * Splitting
40 , splitOn
41 , splitOnSuffix
42 )
43where
45#include "inline.hs"
47import Data.Bifunctor (second)
48import Control.Exception (assert)
49import Control.Monad.Catch (MonadThrow, throwM)
50import Control.Monad.IO.Class (MonadIO(..))
51import Data.Word (Word8)
52import Foreign.ForeignPtr (withForeignPtr, touchForeignPtr)
53import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
54import Foreign.Ptr (minusPtr, plusPtr, castPtr)
55import Foreign.Storable (Storable(..))
56import Fusion.Plugin.Types (Fuse(..))
57import GHC.Exts (SpecConstrAnnotation(..))
58import GHC.ForeignPtr (ForeignPtr(..))
59import GHC.Ptr (Ptr(..))
60import GHC.Types (SPEC(..))
61import Prelude hiding (null, last, (!!), read, concat, unlines)
63#if !defined(mingw32_HOST_OS)
64import Streamly.Internal.FileSystem.FDIO (IOVec(..))
65#endif
66import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
67import Streamly.Internal.Data.Fold.Type (Fold(..))
68import Streamly.Internal.Data.Parser (ParseError(..))
69import Streamly.Internal.Data.Stream.Serial (SerialT)
70import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)
71import Streamly.Internal.Data.SVar (adaptState, defState)
73import qualified Streamly.Internal.Data.Array.Foreign as A
74import qualified Streamly.Internal.Data.Array.Foreign as Array
75import qualified Streamly.Internal.Data.Array.Foreign.Type as A
76import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
77import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS
78import qualified Streamly.Internal.Data.Array.Stream.Fold.Foreign as ASF
79import qualified Streamly.Internal.Data.Fold.Type as FL
80import qualified Streamly.Internal.Data.Parser as PR
81import qualified Streamly.Internal.Data.Parser.ParserD as PRD
82import qualified Streamly.Internal.Data.Stream.IsStream as S
83import qualified Streamly.Internal.Data.Stream.StreamD as D
85-------------------------------------------------------------------------------
86-- Generation
87-------------------------------------------------------------------------------
89-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
90-- @n@ elements each.
91--
92-- > arraysOf n = Stream.chunksOf n (Array.writeN n)
93--
94-- /Pre-release/
95{-# INLINE arraysOf #-}
96arraysOf :: (IsStream t, MonadIO m, Storable a)
97 => Int -> t m a -> t m (Array a)
98arraysOf n str = D.fromStreamD $ A.arraysOf n (D.toStreamD str)
100-------------------------------------------------------------------------------
101-- Append
102-------------------------------------------------------------------------------
104-- XXX efficiently compare two streams of arrays. Two streams can have chunks
105-- of different sizes, we can handle that in the stream comparison abstraction.
106-- This could be useful e.g. to fast compare whether two files differ.
108-- | Convert a stream of arrays into a stream of their elements.
109--
110-- Same as the following but more efficient:
111--
112-- > concat = Stream.unfoldMany Array.read
113--
114-- @since 0.7.0
115{-# INLINE concat #-}
116concat :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a
117-- concat m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)
118-- concat m = D.fromStreamD $ D.concatMap A.toStreamD (D.toStreamD m)
119concat m = D.fromStreamD $ D.unfoldMany A.read (D.toStreamD m)
121-- | Convert a stream of arrays into a stream of their elements reversing the
122-- contents of each array before flattening.
123--
124-- > concatRev = Stream.unfoldMany Array.readRev
125--
126-- @since 0.7.0
127{-# INLINE concatRev #-}
128concatRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a
129concatRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)
131-------------------------------------------------------------------------------
132-- Intersperse and append
133-------------------------------------------------------------------------------
135-- | Flatten a stream of arrays after inserting the given element between
136-- arrays.
137--
138-- /Pre-release/
139{-# INLINE interpose #-}
140interpose :: (MonadIO m, IsStream t, Storable a) => a -> t m (Array a) -> t m a
141interpose x = S.interpose x A.read
143{-# INLINE intercalateSuffix #-}
144intercalateSuffix :: (MonadIO m, IsStream t, Storable a)
145 => Array a -> t m (Array a) -> t m a
146intercalateSuffix = S.intercalateSuffix A.read
148-- | Flatten a stream of arrays appending the given element after each
149-- array.
150--
151-- @since 0.7.0
152{-# INLINE interposeSuffix #-}
153interposeSuffix :: (MonadIO m, IsStream t, Storable a)
154 => a -> t m (Array a) -> t m a
155-- interposeSuffix x = D.fromStreamD . A.unlines x . D.toStreamD
156interposeSuffix x = S.interposeSuffix x A.read
158data FlattenState s a =
159 OuterLoop s
160 | InnerLoop s !(ForeignPtr a) !(Ptr a) !(Ptr a)
162{-# INLINE_NORMAL unlines #-}
163unlines :: forall m a. (MonadIO m, Storable a)
164 => a -> D.Stream m (Array a) -> D.Stream m a
165unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)
166 where
167 {-# INLINE_LATE step' #-}
168 step' gst (OuterLoop st) = do
169 r <- step (adaptState gst) st
170 return $ case r of
171 D.Yield Array{..} s ->
172 let p = unsafeForeignPtrToPtr aStart
173 in D.Skip (InnerLoop s aStart p aEnd)
174 D.Skip s -> D.Skip (OuterLoop s)
175 D.Stop -> D.Stop
177 step' _ (InnerLoop st _ p end) | p == end =
178 return $ D.Yield sep $ OuterLoop st
180 step' _ (InnerLoop st startf p end) = do
181 x <- liftIO $ do
182 r <- peek p
183 touchForeignPtr startf
184 return r
185 return $ D.Yield x (InnerLoop st startf
186 (p `plusPtr` sizeOf (undefined :: a)) end)
188-------------------------------------------------------------------------------
189-- Compact
190-------------------------------------------------------------------------------
192{-# INLINE_NORMAL packArraysChunksOf #-}
193packArraysChunksOf :: (MonadIO m, Storable a)
194 => Int -> D.Stream m (Array a) -> D.Stream m (Array a)
195packArraysChunksOf n str =
196 D.map A.unsafeFreeze $ AS.packArraysChunksOf n $ D.map A.unsafeThaw str
198-- XXX instead of writing two different versions of this operation, we should
199-- write it as a pipe.
200{-# INLINE_NORMAL lpackArraysChunksOf #-}
201lpackArraysChunksOf :: (MonadIO m, Storable a)
202 => Int -> Fold m (Array a) () -> Fold m (Array a) ()
203lpackArraysChunksOf n fld =
204 FL.map A.unsafeThaw $ AS.lpackArraysChunksOf n (FL.map A.unsafeFreeze fld)
206#if !defined(mingw32_HOST_OS)
208-- | @groupIOVecsOf maxBytes maxEntries@ groups arrays in the incoming stream
209-- to create a stream of 'IOVec' arrays with a maximum of @maxBytes@ bytes in
210-- each array and a maximum of @maxEntries@ entries in each array.
211--
212-- @since 0.7.0
213{-# INLINE_NORMAL groupIOVecsOf #-}
214groupIOVecsOf :: MonadIO m
215 => Int -> Int -> D.Stream m (Array a) -> D.Stream m (Array IOVec)
216groupIOVecsOf n maxIOVLen str =
217 D.map A.unsafeFreeze
218 $ AS.groupIOVecsOf n maxIOVLen
219 $ D.map A.unsafeThaw str
220#endif
222-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a
223-- maximum specified size in bytes.
224--
225-- @since 0.7.0
226{-# INLINE compact #-}
227compact :: (MonadIO m, Storable a)
228 => Int -> SerialT m (Array a) -> SerialT m (Array a)
229compact n xs = D.fromStreamD $ packArraysChunksOf n (D.toStreamD xs)
231-------------------------------------------------------------------------------
232-- Split
233-------------------------------------------------------------------------------
235data SplitState s arr
236 = Initial s
237 | Buffering s arr
238 | Splitting s arr
239 | Yielding arr (SplitState s arr)
240 | Finishing
242-- | Split a stream of arrays on a given separator byte, dropping the separator
243-- and coalescing all the arrays between two separators into a single array.
244--
245-- @since 0.7.0
246{-# INLINE_NORMAL _splitOn #-}
247_splitOn
248 :: MonadIO m
249 => Word8
250 -> D.Stream m (Array Word8)
251 -> D.Stream m (Array Word8)
252_splitOn byte (D.Stream step state) = D.Stream step' (Initial state)
254 where
256 {-# INLINE_LATE step' #-}
257 step' gst (Initial st) = do
258 r <- step gst st
259 case r of
260 D.Yield arr s -> do
261 (arr1, marr2) <- A.breakOn byte arr
262 return $ case marr2 of
263 Nothing -> D.Skip (Buffering s arr1)
264 Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))
265 D.Skip s -> return $ D.Skip (Initial s)
266 D.Stop -> return D.Stop
268 step' gst (Buffering st buf) = do
269 r <- step gst st
270 case r of
271 D.Yield arr s -> do
272 (arr1, marr2) <- A.breakOn byte arr
273 buf' <- A.spliceTwo buf arr1
274 return $ case marr2 of
275 Nothing -> D.Skip (Buffering s buf')
276 Just x -> D.Skip (Yielding buf' (Splitting s x))
277 D.Skip s -> return $ D.Skip (Buffering s buf)
278 D.Stop -> return $
279 if A.byteLength buf == 0
280 then D.Stop
281 else D.Skip (Yielding buf Finishing)
283 step' _ (Splitting st buf) = do
284 (arr1, marr2) <- A.breakOn byte buf
285 return $ case marr2 of
286 Nothing -> D.Skip $ Buffering st arr1
287 Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)
289 step' _ (Yielding arr next) = return $ D.Yield arr next
290 step' _ Finishing = return D.Stop
292-- | Split a stream of arrays on a given separator byte, dropping the separator
293-- and coalescing all the arrays between two separators into a single array.
294--
295-- @since 0.7.0
296{-# INLINE splitOn #-}
297splitOn
298 :: (IsStream t, MonadIO m)
299 => Word8
300 -> t m (Array Word8)
301 -> t m (Array Word8)
302splitOn byte s =
303 D.fromStreamD $ D.splitInnerBy (A.breakOn byte) A.spliceTwo $ D.toStreamD s
305{-# INLINE splitOnSuffix #-}
306splitOnSuffix
307 :: (IsStream t, MonadIO m)
308 => Word8
309 -> t m (Array Word8)
310 -> t m (Array Word8)
311-- splitOn byte s = D.fromStreamD $ A.splitOn byte $ D.toStreamD s
312splitOnSuffix byte s =
313 D.fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.spliceTwo $ D.toStreamD s
315-------------------------------------------------------------------------------
316-- Elimination
317-------------------------------------------------------------------------------
319-- When we have to take an array partially, take the last part of the array.
320{-# INLINE takeArrayListRev #-}
321takeArrayListRev :: forall a. Storable a => Int -> [Array a] -> [Array a]
322takeArrayListRev = go
324 where
326 go _ [] = []
327 go n _ | n <= 0 = []
328 go n (x:xs) =
329 let len = Array.length x
330 in if n > len
331 then x : go (n - len) xs
332 else if n == len
333 then [x]
334 else let !(Array (ForeignPtr _ contents) end) = x
335 sz = sizeOf (undefined :: a)
336 !(Ptr start) = end `plusPtr` negate (n * sz)
337 in [Array (ForeignPtr start contents) end]
339-- When we have to take an array partially, take the last part of the array in
340-- the first split.
341{-# INLINE splitAtArrayListRev #-}
342splitAtArrayListRev :: forall a. Storable a =>
343 Int -> [Array a] -> ([Array a],[Array a])
344splitAtArrayListRev n ls
345 | n <= 0 = ([], ls)
346 | otherwise = go n ls
347 where
348 go :: Int -> [Array a] -> ([Array a], [Array a])
349 go _ [] = ([], [])
350 go m (x:xs) =
351 let len = Array.length x
352 (xs', xs'') = go (m - len) xs
353 in if m > len
354 then (x:xs', xs'')
355 else if m == len
356 then ([x],xs)
357 else let !(Array (ForeignPtr start contents) end) = x
358 sz = sizeOf (undefined :: a)
359 end1 = end `plusPtr` negate (m * sz)
360 arr2 = Array (ForeignPtr start contents) end1
361 !(Ptr addrEnd1) = end1
362 arr1 = Array (ForeignPtr addrEnd1 contents) end
363 in ([arr1], arr2:xs)
365-------------------------------------------------------------------------------
366-- Fold to a single Array
367-------------------------------------------------------------------------------
369-- XXX Both of these implementations of splicing seem to perform equally well.
370-- We need to perform benchmarks over a range of sizes though.
372-- CAUTION! length must more than equal to lengths of all the arrays in the
373-- stream.
374{-# INLINE spliceArraysLenUnsafe #-}
375spliceArraysLenUnsafe :: (MonadIO m, Storable a)
376 => Int -> SerialT m (MA.Array a) -> m (MA.Array a)
377spliceArraysLenUnsafe len buffered = do
378 arr <- liftIO $ MA.newArray len
379 end <- S.foldlM' writeArr (return $ MA.aEnd arr) buffered
380 return $ arr {MA.aEnd = end}
382 where
384 writeArr dst (MA.Array as ae _) =
385 liftIO $ withForeignPtr as $ \src -> do
386 let count = ae `minusPtr` src
387 A.memcpy (castPtr dst) (castPtr src) count
388 return $ dst `plusPtr` count
390{-# INLINE _spliceArrays #-}
391_spliceArrays :: (MonadIO m, Storable a)
392 => SerialT m (Array a) -> m (Array a)
393_spliceArrays s = do
394 buffered <- S.foldr S.cons S.nil s
395 len <- S.sum (S.map Array.length buffered)
396 arr <- liftIO $ MA.newArray len
397 end <- S.foldlM' writeArr (return $ MA.aEnd arr) s
398 return $ A.unsafeFreeze $ arr {MA.aEnd = end}
400 where
402 writeArr dst (Array as ae) =
403 liftIO $ withForeignPtr as $ \src -> do
404 let count = ae `minusPtr` src
405 A.memcpy (castPtr dst) (castPtr src) count
406 return $ dst `plusPtr` count
408{-# INLINE _spliceArraysBuffered #-}
409_spliceArraysBuffered :: (MonadIO m, Storable a)
410 => SerialT m (Array a) -> m (Array a)
411_spliceArraysBuffered s = do
412 buffered <- S.foldr S.cons S.nil s
413 len <- S.sum (S.map Array.length buffered)
414 A.unsafeFreeze <$> spliceArraysLenUnsafe len (S.map A.unsafeThaw s)
416{-# INLINE spliceArraysRealloced #-}
417spliceArraysRealloced :: forall m a. (MonadIO m, Storable a)
418 => SerialT m (Array a) -> m (Array a)
419spliceArraysRealloced s = do
420 let idst = liftIO $ MA.newArray (A.bytesToElemCount (undefined :: a)
421 (A.mkChunkSizeKB 4))
423 arr <- S.foldlM' MA.spliceWithDoubling idst (S.map A.unsafeThaw s)
424 liftIO $ A.unsafeFreeze <$> MA.shrinkToFit arr
426-- XXX This should just be "fold A.write"
427--
428-- | Given a stream of arrays, splice them all together to generate a single
429-- array. The stream must be /finite/.
430--
431-- @since 0.7.0
432{-# INLINE toArray #-}
433toArray :: (MonadIO m, Storable a) => SerialT m (Array a) -> m (Array a)
434toArray = spliceArraysRealloced
435-- spliceArrays = _spliceArraysBuffered
437-- exponentially increasing sizes of the chunks upto the max limit.
438-- XXX this will be easier to implement with parsers/terminating folds
439-- With this we should be able to reduce the number of chunks/allocations.
440-- The reallocation/copy based toArray can also be implemented using this.
441--
442{-
443{-# INLINE toArraysInRange #-}
444toArraysInRange :: (IsStream t, MonadIO m, Storable a)
445 => Int -> Int -> Fold m (Array a) b -> Fold m a b
446toArraysInRange low high (Fold step initial extract) =
447-}
449{-
450-- | Fold the input to a pure buffered stream (List) of arrays.
451{-# INLINE _toArraysOf #-}
452_toArraysOf :: (MonadIO m, Storable a)
453 => Int -> Fold m a (SerialT Identity (Array a))
454_toArraysOf n = FL.chunksOf n (A.writeNF n) FL.toStream
455-}
457-------------------------------------------------------------------------------
458-- Parsing
459-------------------------------------------------------------------------------
461-- GHC parser does not accept {-# ANN type [] NoSpecConstr #-}, so we need
462-- to make a newtype.
463{-# ANN type List NoSpecConstr #-}
464newtype List a = List {getList :: [a]}
466{-# INLINE_NORMAL parseD #-}
467parseD ::
468 forall m a b. (MonadIO m, MonadThrow m, Storable a)
469 => PRD.Parser m (Array a) b
470 -> D.Stream m (Array.Array a)
471 -> m (b, D.Stream m (Array.Array a))
472parseD (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do
473 res <- initial
474 case res of
475 PRD.IPartial s -> go SPEC state (List []) s
476 PRD.IDone b -> return (b, stream)
477 PRD.IError err -> throwM $ ParseError err
479 where
481 -- "backBuf" contains last few items in the stream that we may have to
482 -- backtrack to.
483 --
484 -- XXX currently we are using a dumb list based approach for backtracking
485 -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
486 -- That will allow us more efficient random back and forth movement.
487 {-# INLINE go #-}
488 go !_ st backBuf !pst = do
489 r <- step defState st
490 case r of
491 D.Yield x s -> gobuf SPEC [x] s backBuf pst
492 D.Skip s -> go SPEC s backBuf pst
493 D.Stop -> do
494 b <- extract pst
495 return (b, D.nil)
497 gobuf !_ [] s backBuf !pst = go SPEC s backBuf pst
498 gobuf !_ (x:xs) s backBuf !pst = do
499 pRes <- pstep pst x
500 case pRes of
501 PR.Partial 0 pst1 ->
502 gobuf SPEC xs s (List []) pst1
503 PR.Partial n pst1 -> do
504 assert
505 (n <= sum (map Array.length (x:getList backBuf)))
506 (return ())
507 let src0 = takeArrayListRev n (x:getList backBuf)
508 src = Prelude.reverse src0 ++ xs
509 gobuf SPEC src s (List []) pst1
510 PR.Continue 0 pst1 ->
511 gobuf SPEC xs s (List (x:getList backBuf)) pst1
512 PR.Continue n pst1 -> do
513 assert
514 (n <= sum (map Array.length (x:getList backBuf)))
515 (return ())
516 let (src0, buf1) = splitAtArrayListRev n (x:getList backBuf)
517 src = Prelude.reverse src0 ++ xs
518 gobuf SPEC src s (List buf1) pst1
519 PR.Done 0 b ->
520 return (b, D.Stream step s)
521 PR.Done n b -> do
522 assert
523 (n <= sum (map Array.length (x:getList backBuf)))
524 (return ())
525 let src0 = takeArrayListRev n (x:getList backBuf)
526 src = Prelude.reverse src0 ++ xs
527 return (b, D.append (D.fromList src) (D.Stream step s))
528 PR.Error err -> throwM $ ParseError err
530{-
531-- | Parse an array stream using the supplied 'Parser'. Returns the parse
532-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
533--
534-- /Internal/
535--
536{-# INLINE parse #-}
537parse ::
538 (MonadIO m, MonadThrow m, Storable a)
539 => PRD.Parser m a b
540 -> SerialT m (A.Array a)
541 -> m (b, SerialT m (A.Array a))
542parse p s = fmap D.fromStreamD <$> parseD p (D.toStreamD s)
543-}
545-- | Fold an array stream using the supplied array stream 'Fold'.
546--
547-- /Pre-release/
548--
549{-# INLINE fold #-}
550fold :: (MonadIO m, MonadThrow m, Storable a) =>
551 ASF.Fold m a b -> SerialT m (A.Array a) -> m b
552fold (ASF.Fold p) s = fst <$> parseD p (D.toStreamD s)
554-- | Like 'fold' but also returns the remaining stream.
555--
556-- /Pre-release/
557--
558{-# INLINE fold_ #-}
559fold_ :: (MonadIO m, MonadThrow m, Storable a) =>
560 ASF.Fold m a b -> SerialT m (A.Array a) -> m (b, SerialT m (A.Array a))
561fold_ (ASF.Fold p) s = second D.fromStreamD <$> parseD p (D.toStreamD s)
563{-# ANN type ParseChunksState Fuse #-}
564data ParseChunksState x inpBuf st pst =
565 ParseChunksInit inpBuf st
566 | ParseChunksInitLeftOver inpBuf
567 | ParseChunksStream st inpBuf !pst
568 | ParseChunksBuf inpBuf st inpBuf !pst
569 | ParseChunksYield x (ParseChunksState x inpBuf st pst)
571{-# INLINE_NORMAL foldManyD #-}
572foldManyD
573 :: (MonadThrow m, Storable a)
574 => ASF.Fold m a b
575 -> D.Stream m (Array a)
576 -> D.Stream m b
577foldManyD (ASF.Fold (PRD.Parser pstep initial extract)) (D.Stream step state) =
578 D.Stream stepOuter (ParseChunksInit [] state)
580 where
582 {-# INLINE_LATE stepOuter #-}
583 -- Buffer is empty, get the first element from the stream, initialize the
584 -- fold and then go to stream processing loop.
585 stepOuter gst (ParseChunksInit [] st) = do
586 r <- step (adaptState gst) st
587 case r of
588 D.Yield x s -> do
589 res <- initial
590 case res of
591 PRD.IPartial ps ->
592 return $ D.Skip $ ParseChunksBuf [x] s [] ps
593 PRD.IDone pb ->
594 let next = ParseChunksInit [x] s
595 in return $ D.Skip $ ParseChunksYield pb next
596 PRD.IError err -> throwM $ ParseError err
597 D.Skip s -> return $ D.Skip $ ParseChunksInit [] s
598 D.Stop -> return D.Stop
600 -- Buffer is not empty, go to buffered processing loop
601 stepOuter _ (ParseChunksInit src st) = do
602 res <- initial
603 case res of
604 PRD.IPartial ps ->
605 return $ D.Skip $ ParseChunksBuf src st [] ps
606 PRD.IDone pb ->
607 let next = ParseChunksInit src st
608 in return $ D.Skip $ ParseChunksYield pb next
609 PRD.IError err -> throwM $ ParseError err
611 -- XXX we just discard any leftover input at the end
612 stepOuter _ (ParseChunksInitLeftOver _) = return D.Stop
614 -- Buffer is empty, process elements from the stream
615 stepOuter gst (ParseChunksStream st backBuf pst) = do
616 r <- step (adaptState gst) st
617 case r of
618 D.Yield x s -> do
619 pRes <- pstep pst x
620 case pRes of
621 PR.Partial 0 pst1 ->
622 return $ D.Skip $ ParseChunksStream s [] pst1
623 PR.Partial n pst1 -> do
624 assert
625 (n <= sum (map Array.length (x:backBuf)))
626 (return ())
627 let src0 = takeArrayListRev n (x:backBuf)
628 src = Prelude.reverse src0
629 return $ D.Skip $ ParseChunksBuf src s [] pst1
630 PR.Continue 0 pst1 ->
631 return $ D.Skip $ ParseChunksStream s (x:backBuf) pst1
632 PR.Continue n pst1 -> do
633 assert
634 (n <= sum (map Array.length (x:backBuf)))
635 (return ())
636 let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
637 src = Prelude.reverse src0
638 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
639 PR.Done 0 b -> do
640 return $ D.Skip $
641 ParseChunksYield b (ParseChunksInit [] s)
642 PR.Done n b -> do
643 assert
644 (n <= sum (map Array.length (x:backBuf)))
645 (return ())
646 let src0 = takeArrayListRev n (x:backBuf)
647 let src = Prelude.reverse src0
648 return $ D.Skip $
649 ParseChunksYield b (ParseChunksInit src s)
650 PR.Error err -> throwM $ ParseError err
651 D.Skip s -> return $ D.Skip $ ParseChunksStream s backBuf pst
652 D.Stop -> do
653 b <- extract pst
654 let src = Prelude.reverse backBuf
655 return $ D.Skip $
656 ParseChunksYield b (ParseChunksInitLeftOver src)
658 -- go back to stream processing mode
659 stepOuter _ (ParseChunksBuf [] s buf pst) =
660 return $ D.Skip $ ParseChunksStream s buf pst
662 -- buffered processing loop
663 stepOuter _ (ParseChunksBuf (x:xs) s backBuf pst) = do
664 pRes <- pstep pst x
665 case pRes of
666 PR.Partial 0 pst1 ->
667 return $ D.Skip $ ParseChunksBuf xs s [] pst1
668 PR.Partial n pst1 -> do
669 assert (n <= sum (map Array.length (x:backBuf))) (return ())
670 let src0 = takeArrayListRev n (x:backBuf)
671 src = Prelude.reverse src0 ++ xs
672 return $ D.Skip $ ParseChunksBuf src s [] pst1
673 PR.Continue 0 pst1 ->
674 return $ D.Skip $ ParseChunksBuf xs s (x:backBuf) pst1
675 PR.Continue n pst1 -> do
676 assert (n <= sum (map Array.length (x:backBuf))) (return ())
677 let (src0, buf1) = splitAtArrayListRev n (x:backBuf)
678 src = Prelude.reverse src0 ++ xs
679 return $ D.Skip $ ParseChunksBuf src s buf1 pst1
680 PR.Done 0 b ->
681 return $ D.Skip $ ParseChunksYield b (ParseChunksInit xs s)
682 PR.Done n b -> do
683 assert (n <= sum (map Array.length (x:backBuf))) (return ())
684 let src0 = takeArrayListRev n (x:backBuf)
685 src = Prelude.reverse src0 ++ xs
686 return $ D.Skip $ ParseChunksYield b (ParseChunksInit src s)
687 PR.Error err -> throwM $ ParseError err
689 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next
691-- | Apply an array stream 'Fold' repeatedly on an array stream and emit the
692-- fold outputs in the output stream.
693--
694-- See "Streamly.Prelude.foldMany" for more details.
695--
696-- /Pre-release/
697{-# INLINE foldMany #-}
698foldMany
699 :: (IsStream t, MonadThrow m, Storable a)
700 => ASF.Fold m a b
701 -> t m (Array a)
702 -> t m b
703foldMany p m = D.fromStreamD $ foldManyD p (D.toStreamD m)