git.iko.soy/streamly
on aaa775b help sign in
streamly/core/src/Streamly/Internal/Data/Stream/StreamD/Nesting.hs
2466 lines 92 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.StreamD.Nesting
3-- Copyright : (c) 2018 Composewell Technologies
4-- (c) Roman Leshchinskiy 2008-2010
5-- License : BSD-3-Clause
6-- Maintainer : streamly@composewell.com
7-- Stability : experimental
8-- Portability : GHC
9--
10-- This module contains transformations involving multiple streams, unfolds or
11-- folds. There are two types of transformations generational or eliminational.
12-- Generational transformations are like the "Generate" module but they
13-- generate a stream by combining streams instead of elements. Eliminational
14-- transformations are like the "Eliminate" module but they transform a stream
15-- by eliminating parts of the stream instead of eliminating the whole stream.
16--
17-- These combinators involve transformation, generation, elimination so can be
18-- classified under any of those.
19--
20-- Ultimately these operations should be supported by Unfolds, Pipes and Folds,
21-- and this module may become redundant.
23-- The zipWithM combinator in this module has been adapted from the vector
24-- package (c) Roman Leshchinskiy.
25--
26module Streamly.Internal.Data.Stream.StreamD.Nesting
27 (
28 -- * Generate
29 -- | Combining streams to generate streams.
31 -- ** Combine Two Streams
32 -- | Functions ending in the shape:
33 --
34 -- @t m a -> t m a -> t m a@.
36 -- *** Appending
37 -- | Append a stream after another. A special case of concatMap or
38 -- unfoldMany.
39 AppendState(..)
40 , append
42 -- *** Interleaving
43 -- | Interleave elements from two streams alternately. A special case of
44 -- unfoldManyInterleave.
45 , InterleaveState(..)
46 , interleave
47 , interleaveMin
48 , interleaveSuffix
49 , interleaveInfix
51 -- *** Scheduling
52 -- | Execute streams alternately irrespective of whether they generate
53 -- elements or not. Note 'interleave' would execute a stream until it
54 -- yields an element. A special case of unfoldManyRoundRobin.
55 , roundRobin -- interleaveFair?/ParallelFair
57 -- *** Zipping
58 -- | Zip corresponding elements of two streams.
59 , zipWith
60 , zipWithM
62 -- *** Merging
63 -- | Interleave elements from two streams based on a condition.
64 , mergeBy
65 , mergeByM
67 -- ** Combine N Streams
68 -- | Functions generally ending in these shapes:
69 --
70 -- @
71 -- concat: f (t m a) -> t m a
72 -- concatMap: (a -> t m b) -> t m a -> t m b
73 -- unfoldMany: Unfold m a b -> t m a -> t m b
74 -- @
76 -- *** ConcatMap
77 -- | Generate streams by mapping a stream generator on each element of an
78 -- input stream, append the resulting streams and flatten.
79 , concatMap
80 , concatMapM
82 -- *** ConcatUnfold
83 -- | Generate streams by using an unfold on each element of an input
84 -- stream, append the resulting streams and flatten. A special case of
85 -- gintercalate.
86 , unfoldMany
87 , ConcatUnfoldInterleaveState (..)
88 , unfoldManyInterleave
89 , unfoldManyRoundRobin
91 -- *** Interpose
92 -- | Like unfoldMany but intersperses an effect between the streams. A
93 -- special case of gintercalate.
94 , interpose
95 , interposeSuffix
97 -- *** Intercalate
98 -- | Like unfoldMany but intersperses streams from another source between
99 -- the streams from the first source.
100 , gintercalate
101 , gintercalateSuffix
103 -- * Eliminate
104 -- | Folding and Parsing chunks of streams to eliminate nested streams.
105 -- Functions generally ending in these shapes:
106 --
107 -- @
108 -- f (Fold m a b) -> t m a -> t m b
109 -- f (Parser m a b) -> t m a -> t m b
110 -- @
112 -- ** Folding
113 -- | Apply folds on a stream.
114 , foldMany
115 , refoldMany
116 , foldIterateM
117 , refoldIterateM
119 -- ** Parsing
120 -- | Parsing is opposite to flattening. 'parseMany' is dual to concatMap or
121 -- unfoldMany. concatMap generates a stream from single values in a
122 -- stream and flattens, parseMany does the opposite of flattening by
123 -- splitting the stream and then folds each such split to single value in
124 -- the output stream.
125 , parseMany
126 , parseIterate
128 -- ** Grouping
129 -- | Group segments of a stream and fold. Special case of parsing.
130 , chunksOf
131 , groupsBy
132 , groupsRollingBy
134 -- ** Splitting
135 -- | A special case of parsing.
136 , wordsBy
137 , splitOnSeq
138 , splitOnSuffixSeq
139 , sliceOnSuffix
141 -- * Transform (Nested Containers)
142 -- | Opposite to compact in ArrayStream
143 , splitInnerBy
144 , splitInnerBySuffix
145 , intersectBySorted
146 )
147where
149#include "inline.hs"
150#include "ArrayMacros.h"
152import Control.Exception (assert)
153import Control.Monad.Catch (MonadThrow, throwM)
154import Control.Monad.IO.Class (MonadIO(..))
155import Data.Bits (shiftR, shiftL, (.|.), (.&.))
156import Data.Functor.Identity ( Identity )
157import Data.Word (Word32)
158import Foreign.Storable (Storable(..))
159import Fusion.Plugin.Types (Fuse(..))
160import GHC.Types (SPEC(..))
162import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
163import Streamly.Internal.Data.Fold.Step (Step(..))
164import Streamly.Internal.Data.Fold.Type (Fold(..))
165import Streamly.Internal.Data.Parser (ParseError(..))
166import Streamly.Internal.Data.Refold.Type (Refold(..))
167import Streamly.Internal.Data.SVar.Type (adaptState)
168import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
169import Streamly.Internal.Data.Unfold.Type (Unfold(..))
171import qualified Streamly.Internal.Data.Array.Foreign.Type as A
172import qualified Streamly.Internal.Data.Fold as FL
173import qualified Streamly.Internal.Data.Parser as PR
174import qualified Streamly.Internal.Data.Parser.ParserD as PRD
175import qualified Streamly.Internal.Data.Ring.Foreign as RB
177import Streamly.Internal.Data.Stream.StreamD.Type
179import Prelude hiding (concatMap, mapM, zipWith)
181------------------------------------------------------------------------------
182-- Appending
183------------------------------------------------------------------------------
185data AppendState s1 s2 = AppendFirst s1 | AppendSecond s2
187-- Note that this could be much faster compared to the CPS stream. However, as
188-- the number of streams being composed increases this may become expensive.
189-- Need to see where the breaking point is between the two.
190--
191{-# INLINE_NORMAL append #-}
192append :: Monad m => Stream m a -> Stream m a -> Stream m a
193append (Stream step1 state1) (Stream step2 state2) =
194 Stream step (AppendFirst state1)
196 where
198 {-# INLINE_LATE step #-}
199 step gst (AppendFirst st) = do
200 r <- step1 gst st
201 return $ case r of
202 Yield a s -> Yield a (AppendFirst s)
203 Skip s -> Skip (AppendFirst s)
204 Stop -> Skip (AppendSecond state2)
206 step gst (AppendSecond st) = do
207 r <- step2 gst st
208 return $ case r of
209 Yield a s -> Yield a (AppendSecond s)
210 Skip s -> Skip (AppendSecond s)
211 Stop -> Stop
213------------------------------------------------------------------------------
214-- Interleaving
215------------------------------------------------------------------------------
217data InterleaveState s1 s2 = InterleaveFirst s1 s2 | InterleaveSecond s1 s2
218 | InterleaveSecondOnly s2 | InterleaveFirstOnly s1
220{-# INLINE_NORMAL interleave #-}
221interleave :: Monad m => Stream m a -> Stream m a -> Stream m a
222interleave (Stream step1 state1) (Stream step2 state2) =
223 Stream step (InterleaveFirst state1 state2)
225 where
227 {-# INLINE_LATE step #-}
228 step gst (InterleaveFirst st1 st2) = do
229 r <- step1 gst st1
230 return $ case r of
231 Yield a s -> Yield a (InterleaveSecond s st2)
232 Skip s -> Skip (InterleaveFirst s st2)
233 Stop -> Skip (InterleaveSecondOnly st2)
235 step gst (InterleaveSecond st1 st2) = do
236 r <- step2 gst st2
237 return $ case r of
238 Yield a s -> Yield a (InterleaveFirst st1 s)
239 Skip s -> Skip (InterleaveSecond st1 s)
240 Stop -> Skip (InterleaveFirstOnly st1)
242 step gst (InterleaveFirstOnly st1) = do
243 r <- step1 gst st1
244 return $ case r of
245 Yield a s -> Yield a (InterleaveFirstOnly s)
246 Skip s -> Skip (InterleaveFirstOnly s)
247 Stop -> Stop
249 step gst (InterleaveSecondOnly st2) = do
250 r <- step2 gst st2
251 return $ case r of
252 Yield a s -> Yield a (InterleaveSecondOnly s)
253 Skip s -> Skip (InterleaveSecondOnly s)
254 Stop -> Stop
256{-# INLINE_NORMAL interleaveMin #-}
257interleaveMin :: Monad m => Stream m a -> Stream m a -> Stream m a
258interleaveMin (Stream step1 state1) (Stream step2 state2) =
259 Stream step (InterleaveFirst state1 state2)
261 where
263 {-# INLINE_LATE step #-}
264 step gst (InterleaveFirst st1 st2) = do
265 r <- step1 gst st1
266 return $ case r of
267 Yield a s -> Yield a (InterleaveSecond s st2)
268 Skip s -> Skip (InterleaveFirst s st2)
269 Stop -> Stop
271 step gst (InterleaveSecond st1 st2) = do
272 r <- step2 gst st2
273 return $ case r of
274 Yield a s -> Yield a (InterleaveFirst st1 s)
275 Skip s -> Skip (InterleaveSecond st1 s)
276 Stop -> Stop
278 step _ (InterleaveFirstOnly _) = undefined
279 step _ (InterleaveSecondOnly _) = undefined
281{-# INLINE_NORMAL interleaveSuffix #-}
282interleaveSuffix :: Monad m => Stream m a -> Stream m a -> Stream m a
283interleaveSuffix (Stream step1 state1) (Stream step2 state2) =
284 Stream step (InterleaveFirst state1 state2)
286 where
288 {-# INLINE_LATE step #-}
289 step gst (InterleaveFirst st1 st2) = do
290 r <- step1 gst st1
291 return $ case r of
292 Yield a s -> Yield a (InterleaveSecond s st2)
293 Skip s -> Skip (InterleaveFirst s st2)
294 Stop -> Stop
296 step gst (InterleaveSecond st1 st2) = do
297 r <- step2 gst st2
298 return $ case r of
299 Yield a s -> Yield a (InterleaveFirst st1 s)
300 Skip s -> Skip (InterleaveSecond st1 s)
301 Stop -> Skip (InterleaveFirstOnly st1)
303 step gst (InterleaveFirstOnly st1) = do
304 r <- step1 gst st1
305 return $ case r of
306 Yield a s -> Yield a (InterleaveFirstOnly s)
307 Skip s -> Skip (InterleaveFirstOnly s)
308 Stop -> Stop
310 step _ (InterleaveSecondOnly _) = undefined
312data InterleaveInfixState s1 s2 a
313 = InterleaveInfixFirst s1 s2
314 | InterleaveInfixSecondBuf s1 s2
315 | InterleaveInfixSecondYield s1 s2 a
316 | InterleaveInfixFirstYield s1 s2 a
317 | InterleaveInfixFirstOnly s1
319{-# INLINE_NORMAL interleaveInfix #-}
320interleaveInfix :: Monad m => Stream m a -> Stream m a -> Stream m a
321interleaveInfix (Stream step1 state1) (Stream step2 state2) =
322 Stream step (InterleaveInfixFirst state1 state2)
324 where
326 {-# INLINE_LATE step #-}
327 step gst (InterleaveInfixFirst st1 st2) = do
328 r <- step1 gst st1
329 return $ case r of
330 Yield a s -> Yield a (InterleaveInfixSecondBuf s st2)
331 Skip s -> Skip (InterleaveInfixFirst s st2)
332 Stop -> Stop
334 step gst (InterleaveInfixSecondBuf st1 st2) = do
335 r <- step2 gst st2
336 return $ case r of
337 Yield a s -> Skip (InterleaveInfixSecondYield st1 s a)
338 Skip s -> Skip (InterleaveInfixSecondBuf st1 s)
339 Stop -> Skip (InterleaveInfixFirstOnly st1)
341 step gst (InterleaveInfixSecondYield st1 st2 x) = do
342 r <- step1 gst st1
343 return $ case r of
344 Yield a s -> Yield x (InterleaveInfixFirstYield s st2 a)
345 Skip s -> Skip (InterleaveInfixSecondYield s st2 x)
346 Stop -> Stop
348 step _ (InterleaveInfixFirstYield st1 st2 x) = do
349 return $ Yield x (InterleaveInfixSecondBuf st1 st2)
351 step gst (InterleaveInfixFirstOnly st1) = do
352 r <- step1 gst st1
353 return $ case r of
354 Yield a s -> Yield a (InterleaveInfixFirstOnly s)
355 Skip s -> Skip (InterleaveInfixFirstOnly s)
356 Stop -> Stop
358------------------------------------------------------------------------------
359-- Scheduling
360------------------------------------------------------------------------------
362{-# INLINE_NORMAL roundRobin #-}
363roundRobin :: Monad m => Stream m a -> Stream m a -> Stream m a
364roundRobin (Stream step1 state1) (Stream step2 state2) =
365 Stream step (InterleaveFirst state1 state2)
367 where
369 {-# INLINE_LATE step #-}
370 step gst (InterleaveFirst st1 st2) = do
371 r <- step1 gst st1
372 return $ case r of
373 Yield a s -> Yield a (InterleaveSecond s st2)
374 Skip s -> Skip (InterleaveSecond s st2)
375 Stop -> Skip (InterleaveSecondOnly st2)
377 step gst (InterleaveSecond st1 st2) = do
378 r <- step2 gst st2
379 return $ case r of
380 Yield a s -> Yield a (InterleaveFirst st1 s)
381 Skip s -> Skip (InterleaveFirst st1 s)
382 Stop -> Skip (InterleaveFirstOnly st1)
384 step gst (InterleaveSecondOnly st2) = do
385 r <- step2 gst st2
386 return $ case r of
387 Yield a s -> Yield a (InterleaveSecondOnly s)
388 Skip s -> Skip (InterleaveSecondOnly s)
389 Stop -> Stop
391 step gst (InterleaveFirstOnly st1) = do
392 r <- step1 gst st1
393 return $ case r of
394 Yield a s -> Yield a (InterleaveFirstOnly s)
395 Skip s -> Skip (InterleaveFirstOnly s)
396 Stop -> Stop
398------------------------------------------------------------------------------
399-- Zipping
400------------------------------------------------------------------------------
402{-# INLINE_NORMAL zipWithM #-}
403zipWithM :: Monad m
404 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
405zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)
406 where
407 {-# INLINE_LATE step #-}
408 step gst (sa, sb, Nothing) = do
409 r <- stepa (adaptState gst) sa
410 return $
411 case r of
412 Yield x sa' -> Skip (sa', sb, Just x)
413 Skip sa' -> Skip (sa', sb, Nothing)
414 Stop -> Stop
416 step gst (sa, sb, Just x) = do
417 r <- stepb (adaptState gst) sb
418 case r of
419 Yield y sb' -> do
420 z <- f x y
421 return $ Yield z (sa, sb', Nothing)
422 Skip sb' -> return $ Skip (sa, sb', Just x)
423 Stop -> return Stop
425{-# RULES "zipWithM xs xs"
426 forall f xs. zipWithM @Identity f xs xs = mapM (\x -> f x x) xs #-}
428{-# INLINE zipWith #-}
429zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
430zipWith f = zipWithM (\a b -> return (f a b))
432------------------------------------------------------------------------------
433-- Merging
434------------------------------------------------------------------------------
436{-# INLINE_NORMAL mergeByM #-}
437mergeByM
438 :: (Monad m)
439 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
440mergeByM cmp (Stream stepa ta) (Stream stepb tb) =
441 Stream step (Just ta, Just tb, Nothing, Nothing)
442 where
443 {-# INLINE_LATE step #-}
445 -- one of the values is missing, and the corresponding stream is running
446 step gst (Just sa, sb, Nothing, b) = do
447 r <- stepa gst sa
448 return $ case r of
449 Yield a sa' -> Skip (Just sa', sb, Just a, b)
450 Skip sa' -> Skip (Just sa', sb, Nothing, b)
451 Stop -> Skip (Nothing, sb, Nothing, b)
453 step gst (sa, Just sb, a, Nothing) = do
454 r <- stepb gst sb
455 return $ case r of
456 Yield b sb' -> Skip (sa, Just sb', a, Just b)
457 Skip sb' -> Skip (sa, Just sb', a, Nothing)
458 Stop -> Skip (sa, Nothing, a, Nothing)
460 -- both the values are available
461 step _ (sa, sb, Just a, Just b) = do
462 res <- cmp a b
463 return $ case res of
464 GT -> Yield b (sa, sb, Just a, Nothing)
465 _ -> Yield a (sa, sb, Nothing, Just b)
467 -- one of the values is missing, corresponding stream is done
468 step _ (Nothing, sb, Nothing, Just b) =
469 return $ Yield b (Nothing, sb, Nothing, Nothing)
471 step _ (sa, Nothing, Just a, Nothing) =
472 return $ Yield a (sa, Nothing, Nothing, Nothing)
474 step _ (Nothing, Nothing, Nothing, Nothing) = return Stop
476{-# INLINE mergeBy #-}
477mergeBy
478 :: (Monad m)
479 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
480mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
482-------------------------------------------------------------------------------
483-- Intersection of sorted streams
484-------------------------------------------------------------------------------
486-- Assuming the streams are sorted in ascending order
487{-# INLINE_NORMAL intersectBySorted #-}
488intersectBySorted :: Monad m
489 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
490intersectBySorted cmp (Stream stepa ta) (Stream stepb tb) =
491 Stream step
492 ( ta -- left stream state
493 , tb -- right stream state
494 , Nothing -- left value
495 , Nothing -- right value
496 )
498 where
500 {-# INLINE_LATE step #-}
501 -- step 1, fetch the first value
502 step gst (sa, sb, Nothing, b) = do
503 r <- stepa gst sa
504 return $ case r of
505 Yield a sa' -> Skip (sa', sb, Just a, b) -- step 2/3
506 Skip sa' -> Skip (sa', sb, Nothing, b)
507 Stop -> Stop
509 -- step 2, fetch the second value
510 step gst (sa, sb, a@(Just _), Nothing) = do
511 r <- stepb gst sb
512 return $ case r of
513 Yield b sb' -> Skip (sa, sb', a, Just b) -- step 3
514 Skip sb' -> Skip (sa, sb', a, Nothing)
515 Stop -> Stop
517 -- step 3, compare the two values
518 step _ (sa, sb, Just a, Just b) = do
519 let res = cmp a b
520 return $ case res of
521 GT -> Skip (sa, sb, Just a, Nothing) -- step 2
522 LT -> Skip (sa, sb, Nothing, Just b) -- step 1
523 EQ -> Yield a (sa, sb, Nothing, Just b) -- step 1
525------------------------------------------------------------------------------
526-- Combine N Streams - unfoldMany
527------------------------------------------------------------------------------
529data ConcatUnfoldInterleaveState o i =
530 ConcatUnfoldInterleaveOuter o [i]
531 | ConcatUnfoldInterleaveInner o [i]
532 | ConcatUnfoldInterleaveInnerL [i] [i]
533 | ConcatUnfoldInterleaveInnerR [i] [i]
535-- XXX use arrays to store state instead of lists?
536--
537-- XXX In general we can use different scheduling strategies e.g. how to
538-- schedule the outer vs inner loop or assigning weights to different streams
539-- or outer and inner loops.
541-- After a yield, switch to the next stream. Do not switch streams on Skip.
542-- Yield from outer stream switches to the inner stream.
543--
544-- There are two choices here, (1) exhaust the outer stream first and then
545-- start yielding from the inner streams, this is much simpler to implement,
546-- (2) yield at least one element from an inner stream before going back to
547-- outer stream and opening the next stream from it.
548--
549-- Ideally, we need some scheduling bias to inner streams vs outer stream.
550-- Maybe we can configure the behavior.
551--
552-- XXX Instead of using "concatPairsWith wSerial" we can implement an N-way
553-- interleaving CPS combinator which behaves like unfoldManyInterleave. Intead
554-- of pairing up the streams We just need to go yielding one element from each
555-- stream and storing the remaining streams and then keep doing rounds through
556-- those in a round robin fashion. This would be much like wAsync.
557--
558-- | This does not pair streams like concatPairsWith, instead, it goes through
559-- each stream one by one and yields one element from each stream. After it
560-- goes to the last stream it reverses the traversal to come back to the first
561-- stream yielding elements from each stream on its way back to the first
562-- stream and so on.
563--
564-- >>> input = Stream.fromList [[1,1],[2,2],[3,3],[4,4],[5,5]]
565-- >>> Stream.toList $ Stream.unfoldManyInterleave Unfold.fromList input
566-- [1,2,3,4,5,5,4,3,2,1]
567--
568-- Note that this is order of magnitude more efficient than "concatPairsWith
569-- wSerial"
570{-# INLINE_NORMAL unfoldManyInterleave #-}
571unfoldManyInterleave :: Monad m => Unfold m a b -> Stream m a -> Stream m b
572unfoldManyInterleave (Unfold istep inject) (Stream ostep ost) =
573 Stream step (ConcatUnfoldInterleaveOuter ost [])
575 where
577 {-# INLINE_LATE step #-}
578 step gst (ConcatUnfoldInterleaveOuter o ls) = do
579 r <- ostep (adaptState gst) o
580 case r of
581 Yield a o' -> do
582 i <- inject a
583 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
584 Skip o' -> return $ Skip (ConcatUnfoldInterleaveOuter o' ls)
585 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
587 step _ (ConcatUnfoldInterleaveInner _ []) = undefined
588 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
589 r <- istep st
590 return $ case r of
591 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
592 Skip s -> Skip (ConcatUnfoldInterleaveInner o (s:ls))
593 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
595 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
596 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
597 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
599 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
600 r <- istep st
601 return $ case r of
602 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
603 Skip s -> Skip (ConcatUnfoldInterleaveInnerL (s:ls) rs)
604 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
606 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
607 step _ (ConcatUnfoldInterleaveInnerR ls []) =
608 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
610 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
611 r <- istep st
612 return $ case r of
613 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
614 Skip s -> Skip (ConcatUnfoldInterleaveInnerR ls (s:rs))
615 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
617-- XXX In general we can use different scheduling strategies e.g. how to
618-- schedule the outer vs inner loop or assigning weights to different streams
619-- or outer and inner loops.
620--
621-- This could be inefficient if the tasks are too small.
622--
623-- Compared to unfoldManyInterleave this one switches streams on Skips.
624--
625{-# INLINE_NORMAL unfoldManyRoundRobin #-}
626unfoldManyRoundRobin :: Monad m => Unfold m a b -> Stream m a -> Stream m b
627unfoldManyRoundRobin (Unfold istep inject) (Stream ostep ost) =
628 Stream step (ConcatUnfoldInterleaveOuter ost [])
629 where
630 {-# INLINE_LATE step #-}
631 step gst (ConcatUnfoldInterleaveOuter o ls) = do
632 r <- ostep (adaptState gst) o
633 case r of
634 Yield a o' -> do
635 i <- inject a
636 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
637 Skip o' -> return $ Skip (ConcatUnfoldInterleaveInner o' ls)
638 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
640 step _ (ConcatUnfoldInterleaveInner o []) =
641 return $ Skip (ConcatUnfoldInterleaveOuter o [])
643 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
644 r <- istep st
645 return $ case r of
646 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
647 Skip s -> Skip (ConcatUnfoldInterleaveOuter o (s:ls))
648 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
650 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
651 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
652 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
654 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
655 r <- istep st
656 return $ case r of
657 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
658 Skip s -> Skip (ConcatUnfoldInterleaveInnerL ls (s:rs))
659 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
661 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
662 step _ (ConcatUnfoldInterleaveInnerR ls []) =
663 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
665 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
666 r <- istep st
667 return $ case r of
668 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
669 Skip s -> Skip (ConcatUnfoldInterleaveInnerR (s:ls) rs)
670 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
672------------------------------------------------------------------------------
673-- Combine N Streams - interpose
674------------------------------------------------------------------------------
676{-# ANN type InterposeSuffixState Fuse #-}
677data InterposeSuffixState s1 i1 =
678 InterposeSuffixFirst s1
679 -- | InterposeSuffixFirstYield s1 i1
680 | InterposeSuffixFirstInner s1 i1
681 | InterposeSuffixSecond s1
683-- Note that if an unfolded layer turns out to be nil we still emit the
684-- separator effect. An alternate behavior could be to emit the separator
685-- effect only if at least one element has been yielded by the unfolding.
686-- However, that becomes a bit complicated, so we have chosen the former
687-- behvaior for now.
688{-# INLINE_NORMAL interposeSuffix #-}
689interposeSuffix
690 :: Monad m
691 => m c -> Unfold m b c -> Stream m b -> Stream m c
692interposeSuffix
693 action
694 (Unfold istep1 inject1) (Stream step1 state1) =
695 Stream step (InterposeSuffixFirst state1)
697 where
699 {-# INLINE_LATE step #-}
700 step gst (InterposeSuffixFirst s1) = do
701 r <- step1 (adaptState gst) s1
702 case r of
703 Yield a s -> do
704 i <- inject1 a
705 i `seq` return (Skip (InterposeSuffixFirstInner s i))
706 -- i `seq` return (Skip (InterposeSuffixFirstYield s i))
707 Skip s -> return $ Skip (InterposeSuffixFirst s)
708 Stop -> return Stop
710 {-
711 step _ (InterposeSuffixFirstYield s1 i1) = do
712 r <- istep1 i1
713 return $ case r of
714 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
715 Skip i' -> Skip (InterposeSuffixFirstYield s1 i')
716 Stop -> Skip (InterposeSuffixFirst s1)
717 -}
719 step _ (InterposeSuffixFirstInner s1 i1) = do
720 r <- istep1 i1
721 return $ case r of
722 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
723 Skip i' -> Skip (InterposeSuffixFirstInner s1 i')
724 Stop -> Skip (InterposeSuffixSecond s1)
726 step _ (InterposeSuffixSecond s1) = do
727 r <- action
728 return $ Yield r (InterposeSuffixFirst s1)
730{-# ANN type InterposeState Fuse #-}
731data InterposeState s1 i1 a =
732 InterposeFirst s1
733 -- | InterposeFirstYield s1 i1
734 | InterposeFirstInner s1 i1
735 | InterposeFirstInject s1
736 -- | InterposeFirstBuf s1 i1
737 | InterposeSecondYield s1 i1
738 -- -- | InterposeSecondYield s1 i1 a
739 -- -- | InterposeFirstResume s1 i1 a
741-- Note that this only interposes the pure values, we may run many effects to
742-- generate those values as some effects may not generate anything (Skip).
743{-# INLINE_NORMAL interpose #-}
744interpose :: Monad m => m c -> Unfold m b c -> Stream m b -> Stream m c
745interpose
746 action
747 (Unfold istep1 inject1) (Stream step1 state1) =
748 Stream step (InterposeFirst state1)
750 where
752 {-# INLINE_LATE step #-}
753 step gst (InterposeFirst s1) = do
754 r <- step1 (adaptState gst) s1
755 case r of
756 Yield a s -> do
757 i <- inject1 a
758 i `seq` return (Skip (InterposeFirstInner s i))
759 -- i `seq` return (Skip (InterposeFirstYield s i))
760 Skip s -> return $ Skip (InterposeFirst s)
761 Stop -> return Stop
763 {-
764 step _ (InterposeFirstYield s1 i1) = do
765 r <- istep1 i1
766 return $ case r of
767 Yield x i' -> Yield x (InterposeFirstInner s1 i')
768 Skip i' -> Skip (InterposeFirstYield s1 i')
769 Stop -> Skip (InterposeFirst s1)
770 -}
772 step _ (InterposeFirstInner s1 i1) = do
773 r <- istep1 i1
774 return $ case r of
775 Yield x i' -> Yield x (InterposeFirstInner s1 i')
776 Skip i' -> Skip (InterposeFirstInner s1 i')
777 Stop -> Skip (InterposeFirstInject s1)
779 step gst (InterposeFirstInject s1) = do
780 r <- step1 (adaptState gst) s1
781 case r of
782 Yield a s -> do
783 i <- inject1 a
784 -- i `seq` return (Skip (InterposeFirstBuf s i))
785 i `seq` return (Skip (InterposeSecondYield s i))
786 Skip s -> return $ Skip (InterposeFirstInject s)
787 Stop -> return Stop
789 {-
790 step _ (InterposeFirstBuf s1 i1) = do
791 r <- istep1 i1
792 return $ case r of
793 Yield x i' -> Skip (InterposeSecondYield s1 i' x)
794 Skip i' -> Skip (InterposeFirstBuf s1 i')
795 Stop -> Stop
796 -}
798 {-
799 step _ (InterposeSecondYield s1 i1 v) = do
800 r <- action
801 return $ Yield r (InterposeFirstResume s1 i1 v)
802 -}
803 step _ (InterposeSecondYield s1 i1) = do
804 r <- action
805 return $ Yield r (InterposeFirstInner s1 i1)
807 {-
808 step _ (InterposeFirstResume s1 i1 v) = do
809 return $ Yield v (InterposeFirstInner s1 i1)
810 -}
812------------------------------------------------------------------------------
813-- Combine N Streams - intercalate
814------------------------------------------------------------------------------
816data ICUState s1 s2 i1 i2 =
817 ICUFirst s1 s2
818 | ICUSecond s1 s2
819 | ICUSecondOnly s2
820 | ICUFirstOnly s1
821 | ICUFirstInner s1 s2 i1
822 | ICUSecondInner s1 s2 i2
823 | ICUFirstOnlyInner s1 i1
824 | ICUSecondOnlyInner s2 i2
826-- | Interleave streams (full streams, not the elements) unfolded from two
827-- input streams and concat. Stop when the first stream stops. If the second
828-- stream ends before the first one then first stream still keeps running alone
829-- without any interleaving with the second stream.
830--
831-- [a1, a2, ... an] [b1, b2 ...]
832-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
833-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
834-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
835--
836{-# INLINE_NORMAL gintercalateSuffix #-}
837gintercalateSuffix
838 :: Monad m
839 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
840gintercalateSuffix
841 (Unfold istep1 inject1) (Stream step1 state1)
842 (Unfold istep2 inject2) (Stream step2 state2) =
843 Stream step (ICUFirst state1 state2)
845 where
847 {-# INLINE_LATE step #-}
848 step gst (ICUFirst s1 s2) = do
849 r <- step1 (adaptState gst) s1
850 case r of
851 Yield a s -> do
852 i <- inject1 a
853 i `seq` return (Skip (ICUFirstInner s s2 i))
854 Skip s -> return $ Skip (ICUFirst s s2)
855 Stop -> return Stop
857 step gst (ICUFirstOnly s1) = do
858 r <- step1 (adaptState gst) s1
859 case r of
860 Yield a s -> do
861 i <- inject1 a
862 i `seq` return (Skip (ICUFirstOnlyInner s i))
863 Skip s -> return $ Skip (ICUFirstOnly s)
864 Stop -> return Stop
866 step _ (ICUFirstInner s1 s2 i1) = do
867 r <- istep1 i1
868 return $ case r of
869 Yield x i' -> Yield x (ICUFirstInner s1 s2 i')
870 Skip i' -> Skip (ICUFirstInner s1 s2 i')
871 Stop -> Skip (ICUSecond s1 s2)
873 step _ (ICUFirstOnlyInner s1 i1) = do
874 r <- istep1 i1
875 return $ case r of
876 Yield x i' -> Yield x (ICUFirstOnlyInner s1 i')
877 Skip i' -> Skip (ICUFirstOnlyInner s1 i')
878 Stop -> Skip (ICUFirstOnly s1)
880 step gst (ICUSecond s1 s2) = do
881 r <- step2 (adaptState gst) s2
882 case r of
883 Yield a s -> do
884 i <- inject2 a
885 i `seq` return (Skip (ICUSecondInner s1 s i))
886 Skip s -> return $ Skip (ICUSecond s1 s)
887 Stop -> return $ Skip (ICUFirstOnly s1)
889 step _ (ICUSecondInner s1 s2 i2) = do
890 r <- istep2 i2
891 return $ case r of
892 Yield x i' -> Yield x (ICUSecondInner s1 s2 i')
893 Skip i' -> Skip (ICUSecondInner s1 s2 i')
894 Stop -> Skip (ICUFirst s1 s2)
896 step _ (ICUSecondOnly _s2) = undefined
897 step _ (ICUSecondOnlyInner _s2 _i2) = undefined
899data ICALState s1 s2 i1 i2 a =
900 ICALFirst s1 s2
901 -- | ICALFirstYield s1 s2 i1
902 | ICALFirstInner s1 s2 i1
903 | ICALFirstOnly s1
904 | ICALFirstOnlyInner s1 i1
905 | ICALSecondInject s1 s2
906 | ICALFirstInject s1 s2 i2
907 -- | ICALFirstBuf s1 s2 i1 i2
908 | ICALSecondInner s1 s2 i1 i2
909 -- -- | ICALSecondInner s1 s2 i1 i2 a
910 -- -- | ICALFirstResume s1 s2 i1 i2 a
912-- | Interleave streams (full streams, not the elements) unfolded from two
913-- input streams and concat. Stop when the first stream stops. If the second
914-- stream ends before the first one then first stream still keeps running alone
915-- without any interleaving with the second stream.
916--
917-- [a1, a2, ... an] [b1, b2 ...]
918-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
919-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
920-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
921--
922{-# INLINE_NORMAL gintercalate #-}
923gintercalate
924 :: Monad m
925 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
926gintercalate
927 (Unfold istep1 inject1) (Stream step1 state1)
928 (Unfold istep2 inject2) (Stream step2 state2) =
929 Stream step (ICALFirst state1 state2)
931 where
933 {-# INLINE_LATE step #-}
934 step gst (ICALFirst s1 s2) = do
935 r <- step1 (adaptState gst) s1
936 case r of
937 Yield a s -> do
938 i <- inject1 a
939 i `seq` return (Skip (ICALFirstInner s s2 i))
940 -- i `seq` return (Skip (ICALFirstYield s s2 i))
941 Skip s -> return $ Skip (ICALFirst s s2)
942 Stop -> return Stop
944 {-
945 step _ (ICALFirstYield s1 s2 i1) = do
946 r <- istep1 i1
947 return $ case r of
948 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
949 Skip i' -> Skip (ICALFirstYield s1 s2 i')
950 Stop -> Skip (ICALFirst s1 s2)
951 -}
953 step _ (ICALFirstInner s1 s2 i1) = do
954 r <- istep1 i1
955 return $ case r of
956 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
957 Skip i' -> Skip (ICALFirstInner s1 s2 i')
958 Stop -> Skip (ICALSecondInject s1 s2)
960 step gst (ICALFirstOnly s1) = do
961 r <- step1 (adaptState gst) s1
962 case r of
963 Yield a s -> do
964 i <- inject1 a
965 i `seq` return (Skip (ICALFirstOnlyInner s i))
966 Skip s -> return $ Skip (ICALFirstOnly s)
967 Stop -> return Stop
969 step _ (ICALFirstOnlyInner s1 i1) = do
970 r <- istep1 i1
971 return $ case r of
972 Yield x i' -> Yield x (ICALFirstOnlyInner s1 i')
973 Skip i' -> Skip (ICALFirstOnlyInner s1 i')
974 Stop -> Skip (ICALFirstOnly s1)
976 -- We inject the second stream even before checking if the first stream
977 -- would yield any more elements. There is no clear choice whether we
978 -- should do this before or after that. Doing it after may make the state
979 -- machine a bit simpler though.
980 step gst (ICALSecondInject s1 s2) = do
981 r <- step2 (adaptState gst) s2
982 case r of
983 Yield a s -> do
984 i <- inject2 a
985 i `seq` return (Skip (ICALFirstInject s1 s i))
986 Skip s -> return $ Skip (ICALSecondInject s1 s)
987 Stop -> return $ Skip (ICALFirstOnly s1)
989 step gst (ICALFirstInject s1 s2 i2) = do
990 r <- step1 (adaptState gst) s1
991 case r of
992 Yield a s -> do
993 i <- inject1 a
994 i `seq` return (Skip (ICALSecondInner s s2 i i2))
995 -- i `seq` return (Skip (ICALFirstBuf s s2 i i2))
996 Skip s -> return $ Skip (ICALFirstInject s s2 i2)
997 Stop -> return Stop
999 {-
1000 step _ (ICALFirstBuf s1 s2 i1 i2) = do
1001 r <- istep1 i1
1002 return $ case r of
1003 Yield x i' -> Skip (ICALSecondInner s1 s2 i' i2 x)
1004 Skip i' -> Skip (ICALFirstBuf s1 s2 i' i2)
1005 Stop -> Stop
1007 step _ (ICALSecondInner s1 s2 i1 i2 v) = do
1008 r <- istep2 i2
1009 return $ case r of
1010 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i' v)
1011 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i' v)
1012 Stop -> Skip (ICALFirstResume s1 s2 i1 i2 v)
1013 -}
1015 step _ (ICALSecondInner s1 s2 i1 i2) = do
1016 r <- istep2 i2
1017 return $ case r of
1018 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i')
1019 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i')
1020 Stop -> Skip (ICALFirstInner s1 s2 i1)
1021 -- Stop -> Skip (ICALFirstResume s1 s2 i1 i2)
1023 {-
1024 step _ (ICALFirstResume s1 s2 i1 i2 x) = do
1025 return $ Yield x (ICALFirstInner s1 s2 i1 i2)
1026 -}
1028------------------------------------------------------------------------------
1029-- Folding
1030------------------------------------------------------------------------------
1032{-# ANN type FIterState Fuse #-}
1033data FIterState s f m a b
1034 = FIterInit s f
1035 | forall fs. FIterStream s (fs -> a -> m (FL.Step fs b)) fs (fs -> m b)
1036 | FIterYield b (FIterState s f m a b)
1037 | FIterStop
1039{-# INLINE_NORMAL foldIterateM #-}
1040foldIterateM ::
1041 Monad m => (b -> m (FL.Fold m a b)) -> m b -> Stream m a -> Stream m b
1042foldIterateM func seed0 (Stream step state) =
1043 Stream stepOuter (FIterInit state seed0)
1045 where
1047 {-# INLINE iterStep #-}
1048 iterStep from st fstep extract = do
1049 res <- from
1050 return
1051 $ Skip
1052 $ case res of
1053 FL.Partial fs -> FIterStream st fstep fs extract
1054 FL.Done fb -> FIterYield fb $ FIterInit st (return fb)
1056 {-# INLINE_LATE stepOuter #-}
1057 stepOuter _ (FIterInit st seed) = do
1058 (FL.Fold fstep initial extract) <- seed >>= func
1059 iterStep initial st fstep extract
1060 stepOuter gst (FIterStream st fstep fs extract) = do
1061 r <- step (adaptState gst) st
1062 case r of
1063 Yield x s -> do
1064 iterStep (fstep fs x) s fstep extract
1065 Skip s -> return $ Skip $ FIterStream s fstep fs extract
1066 Stop -> do
1067 b <- extract fs
1068 return $ Skip $ FIterYield b FIterStop
1069 stepOuter _ (FIterYield a next) = return $ Yield a next
1070 stepOuter _ FIterStop = return Stop
1072{-# ANN type CIterState Fuse #-}
1073data CIterState s f fs b
1074 = CIterInit s f
1075 | CIterConsume s fs
1076 | CIterYield b (CIterState s f fs b)
1077 | CIterStop
1079-- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
1080-- much more efficient due to stream fusion.
1082-- /Internal/
1083{-# INLINE_NORMAL refoldIterateM #-}
1084refoldIterateM ::
1085 Monad m => Refold m b a b -> m b -> Stream m a -> Stream m b
1086refoldIterateM (Refold fstep finject fextract) initial (Stream step state) =
1087 Stream stepOuter (CIterInit state initial)
1089 where
1091 {-# INLINE iterStep #-}
1092 iterStep st action = do
1093 res <- action
1094 return
1095 $ Skip
1096 $ case res of
1097 FL.Partial fs -> CIterConsume st fs
1098 FL.Done fb -> CIterYield fb $ CIterInit st (return fb)
1100 {-# INLINE_LATE stepOuter #-}
1101 stepOuter _ (CIterInit st action) = do
1102 iterStep st (action >>= finject)
1103 stepOuter gst (CIterConsume st fs) = do
1104 r <- step (adaptState gst) st
1105 case r of
1106 Yield x s -> iterStep s (fstep fs x)
1107 Skip s -> return $ Skip $ CIterConsume s fs
1108 Stop -> do
1109 b <- fextract fs
1110 return $ Skip $ CIterYield b CIterStop
1111 stepOuter _ (CIterYield a next) = return $ Yield a next
1112 stepOuter _ CIterStop = return Stop
1114-- "n" elements at the end are dropped by the fold.
1115{-# INLINE sliceBy #-}
1116sliceBy :: Monad m => Fold m a Int -> Int -> Refold m (Int, Int) a (Int, Int)
1117sliceBy (Fold step1 initial1 extract1) n = Refold step inject extract
1119 where
1121 inject (i, len) = do
1122 r <- initial1
1123 return $ case r of
1124 Partial s -> Partial $ Tuple' (i + len + n) s
1125 Done l -> Done (i, l)
1127 step (Tuple' i s) x = do
1128 r <- step1 s x
1129 return $ case r of
1130 Partial s1 -> Partial $ Tuple' i s1
1131 Done len -> Done (i, len)
1133 extract (Tuple' i s) = (i,) <$> extract1 s
1135{-# INLINE sliceOnSuffix #-}
1136sliceOnSuffix :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
1137sliceOnSuffix predicate =
1138 -- Scan the stream with the given refold
1139 refoldIterateM
1140 (sliceBy (FL.takeEndBy_ predicate FL.length) 1)
1141 (return (-1, 0))
1143------------------------------------------------------------------------------
1144-- Parsing
1145------------------------------------------------------------------------------
1147{-# ANN type ParseChunksState Fuse #-}
1148data ParseChunksState x inpBuf st pst =
1149 ParseChunksInit inpBuf st
1150 | ParseChunksInitLeftOver inpBuf
1151 | ParseChunksStream st inpBuf !pst
1152 | ParseChunksBuf inpBuf st inpBuf !pst
1153 | ParseChunksYield x (ParseChunksState x inpBuf st pst)
1155-- XXX This is in fact parseMany1 (a la foldMany1). Do we need a parseMany as
1156-- well?
1157{-# INLINE_NORMAL parseMany #-}
1158parseMany
1159 :: MonadThrow m
1160 => PRD.Parser m a b
1161 -> Stream m a
1162 -> Stream m b
1163parseMany (PRD.Parser pstep initial extract) (Stream step state) =
1164 Stream stepOuter (ParseChunksInit [] state)
1166 where
1168 {-# INLINE_LATE stepOuter #-}
1169 -- Buffer is empty, get the first element from the stream, initialize the
1170 -- fold and then go to stream processing loop.
1171 stepOuter gst (ParseChunksInit [] st) = do
1172 r <- step (adaptState gst) st
1173 case r of
1174 Yield x s -> do
1175 res <- initial
1176 case res of
1177 PRD.IPartial ps ->
1178 return $ Skip $ ParseChunksBuf [x] s [] ps
1179 PRD.IDone pb ->
1180 let next = ParseChunksInit [x] s
1181 in return $ Skip $ ParseChunksYield pb next
1182 PRD.IError err -> throwM $ ParseError err
1183 Skip s -> return $ Skip $ ParseChunksInit [] s
1184 Stop -> return Stop
1186 -- Buffer is not empty, go to buffered processing loop
1187 stepOuter _ (ParseChunksInit src st) = do
1188 res <- initial
1189 case res of
1190 PRD.IPartial ps ->
1191 return $ Skip $ ParseChunksBuf src st [] ps
1192 PRD.IDone pb ->
1193 let next = ParseChunksInit src st
1194 in return $ Skip $ ParseChunksYield pb next
1195 PRD.IError err -> throwM $ ParseError err
1197 -- XXX we just discard any leftover input at the end
1198 stepOuter _ (ParseChunksInitLeftOver _) = return Stop
1200 -- Buffer is empty, process elements from the stream
1201 stepOuter gst (ParseChunksStream st buf pst) = do
1202 r <- step (adaptState gst) st
1203 case r of
1204 Yield x s -> do
1205 pRes <- pstep pst x
1206 case pRes of
1207 PR.Partial 0 pst1 ->
1208 return $ Skip $ ParseChunksStream s [] pst1
1209 PR.Partial n pst1 -> do
1210 assert (n <= length (x:buf)) (return ())
1211 let src0 = Prelude.take n (x:buf)
1212 src = Prelude.reverse src0
1213 return $ Skip $ ParseChunksBuf src s [] pst1
1214 PR.Continue 0 pst1 ->
1215 return $ Skip $ ParseChunksStream s (x:buf) pst1
1216 PR.Continue n pst1 -> do
1217 assert (n <= length (x:buf)) (return ())
1218 let (src0, buf1) = splitAt n (x:buf)
1219 src = Prelude.reverse src0
1220 return $ Skip $ ParseChunksBuf src s buf1 pst1
1221 PR.Done 0 b -> do
1222 return $ Skip $
1223 ParseChunksYield b (ParseChunksInit [] s)
1224 PR.Done n b -> do
1225 assert (n <= length (x:buf)) (return ())
1226 let src = Prelude.reverse (Prelude.take n (x:buf))
1227 return $ Skip $
1228 ParseChunksYield b (ParseChunksInit src s)
1229 PR.Error err -> throwM $ ParseError err
1230 Skip s -> return $ Skip $ ParseChunksStream s buf pst
1231 Stop -> do
1232 b <- extract pst
1233 let src = Prelude.reverse buf
1234 return $ Skip $
1235 ParseChunksYield b (ParseChunksInitLeftOver src)
1237 -- go back to stream processing mode
1238 stepOuter _ (ParseChunksBuf [] s buf pst) =
1239 return $ Skip $ ParseChunksStream s buf pst
1241 -- buffered processing loop
1242 stepOuter _ (ParseChunksBuf (x:xs) s buf pst) = do
1243 pRes <- pstep pst x
1244 case pRes of
1245 PR.Partial 0 pst1 ->
1246 return $ Skip $ ParseChunksBuf xs s [] pst1
1247 PR.Partial n pst1 -> do
1248 assert (n <= length (x:buf)) (return ())
1249 let src0 = Prelude.take n (x:buf)
1250 src = Prelude.reverse src0 ++ xs
1251 return $ Skip $ ParseChunksBuf src s [] pst1
1252 PR.Continue 0 pst1 ->
1253 return $ Skip $ ParseChunksBuf xs s (x:buf) pst1
1254 PR.Continue n pst1 -> do
1255 assert (n <= length (x:buf)) (return ())
1256 let (src0, buf1) = splitAt n (x:buf)
1257 src = Prelude.reverse src0 ++ xs
1258 return $ Skip $ ParseChunksBuf src s buf1 pst1
1259 PR.Done 0 b ->
1260 return $ Skip $ ParseChunksYield b (ParseChunksInit xs s)
1261 PR.Done n b -> do
1262 assert (n <= length (x:buf)) (return ())
1263 let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs
1264 return $ Skip $ ParseChunksYield b (ParseChunksInit src s)
1265 PR.Error err -> throwM $ ParseError err
1267 stepOuter _ (ParseChunksYield a next) = return $ Yield a next
1269{-# ANN type ConcatParseState Fuse #-}
1270data ConcatParseState b inpBuf st p m a =
1271 ConcatParseInit inpBuf st p
1272 | ConcatParseInitLeftOver inpBuf
1273 | forall s. ConcatParseStream st inpBuf (s -> a -> m (PRD.Step s b)) s (s -> m b)
1274 | forall s. ConcatParseBuf inpBuf st inpBuf (s -> a -> m (PRD.Step s b)) s (s -> m b)
1275 | ConcatParseYield b (ConcatParseState b inpBuf st p m a)
1277{-# INLINE_NORMAL parseIterate #-}
1278parseIterate
1279 :: MonadThrow m
1280 => (b -> PRD.Parser m a b)
1281 -> b
1282 -> Stream m a
1283 -> Stream m b
1284parseIterate func seed (Stream step state) =
1285 Stream stepOuter (ConcatParseInit [] state (func seed))
1287 where
1289 {-# INLINE_LATE stepOuter #-}
1290 -- Buffer is empty, go to stream processing loop
1291 stepOuter _ (ConcatParseInit [] st (PRD.Parser pstep initial extract)) = do
1292 res <- initial
1293 case res of
1294 PRD.IPartial ps ->
1295 return $ Skip $ ConcatParseStream st [] pstep ps extract
1296 PRD.IDone pb ->
1297 let next = ConcatParseInit [] st (func pb)
1298 in return $ Skip $ ConcatParseYield pb next
1299 PRD.IError err -> throwM $ ParseError err
1301 -- Buffer is not empty, go to buffered processing loop
1302 stepOuter _ (ConcatParseInit src st
1303 (PRD.Parser pstep initial extract)) = do
1304 res <- initial
1305 case res of
1306 PRD.IPartial ps ->
1307 return $ Skip $ ConcatParseBuf src st [] pstep ps extract
1308 PRD.IDone pb ->
1309 let next = ConcatParseInit src st (func pb)
1310 in return $ Skip $ ConcatParseYield pb next
1311 PRD.IError err -> throwM $ ParseError err
1313 -- XXX we just discard any leftover input at the end
1314 stepOuter _ (ConcatParseInitLeftOver _) = return Stop
1316 -- Buffer is empty process elements from the stream
1317 stepOuter gst (ConcatParseStream st buf pstep pst extract) = do
1318 r <- step (adaptState gst) st
1319 case r of
1320 Yield x s -> do
1321 pRes <- pstep pst x
1322 case pRes of
1323 PR.Partial 0 pst1 ->
1324 return $ Skip $ ConcatParseStream s [] pstep pst1 extract
1325 PR.Partial n pst1 -> do
1326 assert (n <= length (x:buf)) (return ())
1327 let src0 = Prelude.take n (x:buf)
1328 src = Prelude.reverse src0
1329 return $ Skip $ ConcatParseBuf src s [] pstep pst1 extract
1330 -- PR.Continue 0 pst1 ->
1331 -- return $ Skip $ ConcatParseStream s (x:buf) pst1
1332 PR.Continue n pst1 -> do
1333 assert (n <= length (x:buf)) (return ())
1334 let (src0, buf1) = splitAt n (x:buf)
1335 src = Prelude.reverse src0
1336 return $ Skip $ ConcatParseBuf src s buf1 pstep pst1 extract
1337 -- XXX Specialize for Stop 0 common case?
1338 PR.Done n b -> do
1339 assert (n <= length (x:buf)) (return ())
1340 let src = Prelude.reverse (Prelude.take n (x:buf))
1341 return $ Skip $
1342 ConcatParseYield b (ConcatParseInit src s (func b))
1343 PR.Error err -> throwM $ ParseError err
1344 Skip s -> return $ Skip $ ConcatParseStream s buf pstep pst extract
1345 Stop -> do
1346 b <- extract pst
1347 let src = Prelude.reverse buf
1348 return $ Skip $ ConcatParseYield b (ConcatParseInitLeftOver src)
1350 -- go back to stream processing mode
1351 stepOuter _ (ConcatParseBuf [] s buf pstep ps extract) =
1352 return $ Skip $ ConcatParseStream s buf pstep ps extract
1354 -- buffered processing loop
1355 stepOuter _ (ConcatParseBuf (x:xs) s buf pstep pst extract) = do
1356 pRes <- pstep pst x
1357 case pRes of
1358 PR.Partial 0 pst1 ->
1359 return $ Skip $ ConcatParseBuf xs s [] pstep pst1 extract
1360 PR.Partial n pst1 -> do
1361 assert (n <= length (x:buf)) (return ())
1362 let src0 = Prelude.take n (x:buf)
1363 src = Prelude.reverse src0 ++ xs
1364 return $ Skip $ ConcatParseBuf src s [] pstep pst1 extract
1365 -- PR.Continue 0 pst1 -> return $ Skip $ ConcatParseBuf xs s (x:buf) pst1
1366 PR.Continue n pst1 -> do
1367 assert (n <= length (x:buf)) (return ())
1368 let (src0, buf1) = splitAt n (x:buf)
1369 src = Prelude.reverse src0 ++ xs
1370 return $ Skip $ ConcatParseBuf src s buf1 pstep pst1 extract
1371 -- XXX Specialize for Stop 0 common case?
1372 PR.Done n b -> do
1373 assert (n <= length (x:buf)) (return ())
1374 let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs
1375 return $ Skip $ ConcatParseYield b
1376 (ConcatParseInit src s (func b))
1377 PR.Error err -> throwM $ ParseError err
1379 stepOuter _ (ConcatParseYield a next) = return $ Yield a next
1381------------------------------------------------------------------------------
1382-- Grouping
1383------------------------------------------------------------------------------
1385data GroupByState st fs a b
1386 = GroupingInit st
1387 | GroupingDo st !fs
1388 | GroupingInitWith st !a
1389 | GroupingDoWith st !fs !a
1390 | GroupingYield !b (GroupByState st fs a b)
1391 | GroupingDone
1393{-# INLINE_NORMAL groupsBy #-}
1394groupsBy :: Monad m
1395 => (a -> a -> Bool)
1396 -> Fold m a b
1397 -> Stream m a
1398 -> Stream m b
1400groupsBy eq fld = parseMany (PRD.groupBy eq fld)
1402groupsBy cmp (Fold fstep initial done) (Stream step state) =
1403 Stream stepOuter (GroupingInit state)
1405 where
1407 {-# INLINE_LATE stepOuter #-}
1408 stepOuter _ (GroupingInit st) = do
1409 -- XXX Note that if the stream stops without yielding a single element
1410 -- in the group we discard the "initial" effect.
1411 res <- initial
1412 return
1413 $ case res of
1414 FL.Partial s -> Skip $ GroupingDo st s
1415 FL.Done b -> Yield b $ GroupingInit st
1416 stepOuter gst (GroupingDo st fs) = do
1417 res <- step (adaptState gst) st
1418 case res of
1419 Yield x s -> do
1420 r <- fstep fs x
1421 case r of
1422 FL.Partial fs1 -> go SPEC x s fs1
1423 FL.Done b -> return $ Yield b (GroupingInit s)
1424 Skip s -> return $ Skip $ GroupingDo s fs
1425 Stop -> return Stop
1427 where
1429 go !_ prev stt !acc = do
1430 res <- step (adaptState gst) stt
1431 case res of
1432 Yield x s -> do
1433 if cmp x prev
1434 then do
1435 r <- fstep acc x
1436 case r of
1437 FL.Partial fs1 -> go SPEC prev s fs1
1438 FL.Done b -> return $ Yield b (GroupingInit s)
1439 else do
1440 r <- done acc
1441 return $ Yield r (GroupingInitWith s x)
1442 Skip s -> go SPEC prev s acc
1443 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1444 stepOuter _ (GroupingInitWith st x) = do
1445 res <- initial
1446 return
1447 $ case res of
1448 FL.Partial s -> Skip $ GroupingDoWith st s x
1449 FL.Done b -> Yield b $ GroupingInitWith st x
1450 stepOuter gst (GroupingDoWith st fs prev) = do
1451 res <- fstep fs prev
1452 case res of
1453 FL.Partial fs1 -> go SPEC st fs1
1454 FL.Done b -> return $ Yield b (GroupingInit st)
1456 where
1458 -- XXX code duplicated from the previous equation
1459 go !_ stt !acc = do
1460 res <- step (adaptState gst) stt
1461 case res of
1462 Yield x s -> do
1463 if cmp x prev
1464 then do
1465 r <- fstep acc x
1466 case r of
1467 FL.Partial fs1 -> go SPEC s fs1
1468 FL.Done b -> return $ Yield b (GroupingInit s)
1469 else do
1470 r <- done acc
1471 return $ Yield r (GroupingInitWith s x)
1472 Skip s -> go SPEC s acc
1473 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1474 stepOuter _ (GroupingYield _ _) = error "groupsBy: Unreachable"
1475 stepOuter _ GroupingDone = return Stop
1477{-# INLINE_NORMAL groupsRollingBy #-}
1478groupsRollingBy :: Monad m
1479 => (a -> a -> Bool)
1480 -> Fold m a b
1481 -> Stream m a
1482 -> Stream m b
1484groupsRollingBy eq fld = parseMany (PRD.groupByRolling eq fld)
1486groupsRollingBy cmp (Fold fstep initial done) (Stream step state) =
1487 Stream stepOuter (GroupingInit state)
1489 where
1491 {-# INLINE_LATE stepOuter #-}
1492 stepOuter _ (GroupingInit st) = do
1493 -- XXX Note that if the stream stops without yielding a single element
1494 -- in the group we discard the "initial" effect.
1495 res <- initial
1496 return
1497 $ case res of
1498 FL.Partial fs -> Skip $ GroupingDo st fs
1499 FL.Done fb -> Yield fb $ GroupingInit st
1500 stepOuter gst (GroupingDo st fs) = do
1501 res <- step (adaptState gst) st
1502 case res of
1503 Yield x s -> do
1504 r <- fstep fs x
1505 case r of
1506 FL.Partial fs1 -> go SPEC x s fs1
1507 FL.Done fb -> return $ Yield fb (GroupingInit s)
1508 Skip s -> return $ Skip $ GroupingDo s fs
1509 Stop -> return Stop
1511 where
1513 go !_ prev stt !acc = do
1514 res <- step (adaptState gst) stt
1515 case res of
1516 Yield x s -> do
1517 if cmp prev x
1518 then do
1519 r <- fstep acc x
1520 case r of
1521 FL.Partial fs1 -> go SPEC x s fs1
1522 FL.Done b -> return $ Yield b (GroupingInit s)
1523 else do
1524 r <- done acc
1525 return $ Yield r (GroupingInitWith s x)
1526 Skip s -> go SPEC prev s acc
1527 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1528 stepOuter _ (GroupingInitWith st x) = do
1529 res <- initial
1530 return
1531 $ case res of
1532 FL.Partial s -> Skip $ GroupingDoWith st s x
1533 FL.Done b -> Yield b $ GroupingInitWith st x
1534 stepOuter gst (GroupingDoWith st fs previous) = do
1535 res <- fstep fs previous
1536 case res of
1537 FL.Partial s -> go SPEC previous st s
1538 FL.Done b -> return $ Yield b (GroupingInit st)
1540 where
1542 -- XXX GHC: groupsBy has one less parameter in this go loop and it
1543 -- fuses. However, groupsRollingBy does not fuse, removing the prev
1544 -- parameter makes it fuse. Something needs to be fixed in GHC. The
1545 -- workaround for this is noted in the comments below.
1546 go !_ prev !stt !acc = do
1547 res <- step (adaptState gst) stt
1548 case res of
1549 Yield x s -> do
1550 if cmp prev x
1551 then do
1552 r <- fstep acc x
1553 case r of
1554 FL.Partial fs1 -> go SPEC x s fs1
1555 FL.Done b -> return $ Yield b (GroupingInit st)
1556 else do
1557 {-
1558 r <- done acc
1559 return $ Yield r (GroupingInitWith s x)
1560 -}
1561 -- The code above does not let groupBy fuse. We use the
1562 -- alternative code below instead. Instead of jumping
1563 -- to GroupingInitWith state, we unroll the code of
1564 -- GroupingInitWith state here to help GHC with stream
1565 -- fusion.
1566 result <- initial
1567 r <- done acc
1568 return
1569 $ Yield r
1570 $ case result of
1571 FL.Partial fsi -> GroupingDoWith s fsi x
1572 FL.Done b -> GroupingYield b (GroupingInit s)
1573 Skip s -> go SPEC prev s acc
1574 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1575 stepOuter _ (GroupingYield r next) = return $ Yield r next
1576 stepOuter _ GroupingDone = return Stop
1578------------------------------------------------------------------------------
1579-- Splitting - by a predicate
1580------------------------------------------------------------------------------
1582data WordsByState st fs b
1583 = WordsByInit st
1584 | WordsByDo st !fs
1585 | WordsByDone
1586 | WordsByYield !b (WordsByState st fs b)
1588{-# INLINE_NORMAL wordsBy #-}
1589wordsBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1590wordsBy predicate (Fold fstep initial done) (Stream step state) =
1591 Stream stepOuter (WordsByInit state)
1593 where
1595 {-# INLINE_LATE stepOuter #-}
1596 stepOuter _ (WordsByInit st) = do
1597 res <- initial
1598 return
1599 $ case res of
1600 FL.Partial s -> Skip $ WordsByDo st s
1601 FL.Done b -> Yield b (WordsByInit st)
1603 stepOuter gst (WordsByDo st fs) = do
1604 res <- step (adaptState gst) st
1605 case res of
1606 Yield x s -> do
1607 if predicate x
1608 then do
1609 resi <- initial
1610 return
1611 $ case resi of
1612 FL.Partial fs1 -> Skip $ WordsByDo s fs1
1613 FL.Done b -> Yield b (WordsByInit s)
1614 else do
1615 r <- fstep fs x
1616 case r of
1617 FL.Partial fs1 -> go SPEC s fs1
1618 FL.Done b -> return $ Yield b (WordsByInit s)
1619 Skip s -> return $ Skip $ WordsByDo s fs
1620 Stop -> return Stop
1622 where
1624 go !_ stt !acc = do
1625 res <- step (adaptState gst) stt
1626 case res of
1627 Yield x s -> do
1628 if predicate x
1629 then do
1630 {-
1631 r <- done acc
1632 return $ Yield r (WordsByInit s)
1633 -}
1634 -- The above code does not fuse well. Need to check why
1635 -- GHC is not able to simplify it well. Using the code
1636 -- below, instead of jumping through the WordsByInit
1637 -- state always, we directly go to WordsByDo state in
1638 -- the common case of Partial.
1639 resi <- initial
1640 r <- done acc
1641 return
1642 $ Yield r
1643 $ case resi of
1644 FL.Partial fs1 -> WordsByDo s fs1
1645 FL.Done b -> WordsByYield b (WordsByInit s)
1646 else do
1647 r <- fstep acc x
1648 case r of
1649 FL.Partial fs1 -> go SPEC s fs1
1650 FL.Done b -> return $ Yield b (WordsByInit s)
1651 Skip s -> go SPEC s acc
1652 Stop -> done acc >>= \r -> return $ Yield r WordsByDone
1654 stepOuter _ WordsByDone = return Stop
1656 stepOuter _ (WordsByYield b next) = return $ Yield b next
1658------------------------------------------------------------------------------
1659-- Splitting on a sequence
1660------------------------------------------------------------------------------
1662-- String search algorithms:
1663-- http://www-igm.univ-mlv.fr/~lecroq/string/index.html
1666-- TODO can we unify the splitting operations using a splitting configuration
1667-- like in the split package.
1669data SplitStyle = Infix | Suffix | Prefix deriving (Eq, Show)
1670data SplitOptions = SplitOptions
1671 { style :: SplitStyle
1672 , withSep :: Bool -- ^ keep the separators in output
1673 -- , compact :: Bool -- ^ treat multiple consecutive separators as one
1674 -- , trimHead :: Bool -- ^ drop blank at head
1675 -- , trimTail :: Bool -- ^ drop blank at tail
1679-- XXX using "fs" as the last arg in Constructors may simplify the code a bit,
1680-- because we can use the constructor directly without having to create "jump"
1681-- functions.
1682{-# ANN type SplitOnSeqState Fuse #-}
1683data SplitOnSeqState rb rh ck w fs s b x =
1684 SplitOnSeqInit
1685 | SplitOnSeqYield b (SplitOnSeqState rb rh ck w fs s b x)
1686 | SplitOnSeqDone
1688 | SplitOnSeqEmpty !fs s
1690 | SplitOnSeqSingle !fs s x
1692 | SplitOnSeqWordInit !fs s
1693 | SplitOnSeqWordLoop !w s !fs
1694 | SplitOnSeqWordDone Int !fs !w
1696 | SplitOnSeqKRInit Int !fs s rb !rh
1697 | SplitOnSeqKRLoop fs s rb !rh !ck
1698 | SplitOnSeqKRCheck fs s rb !rh
1699 | SplitOnSeqKRDone Int !fs rb !rh
1701 | SplitOnSeqReinit (fs -> SplitOnSeqState rb rh ck w fs s b x)
1703{-# INLINE_NORMAL splitOnSeq #-}
1704splitOnSeq
1705 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1706 => Array a
1707 -> Fold m a b
1708 -> Stream m a
1709 -> Stream m b
1710splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
1711 Stream stepOuter SplitOnSeqInit
1713 where
1715 patLen = A.length patArr
1716 maxIndex = patLen - 1
1717 elemBits = SIZE_OF(a) * 8
1719 -- For word pattern case
1720 wordMask :: Word
1721 wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1723 elemMask :: Word
1724 elemMask = (1 `shiftL` elemBits) - 1
1726 wordPat :: Word
1727 wordPat = wordMask .&. A.foldl' addToWord 0 patArr
1729 addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1731 -- For Rabin-Karp search
1732 k = 2891336453 :: Word32
1733 coeff = k ^ patLen
1735 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1737 deltaCksum cksum old new =
1738 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1740 -- XXX shall we use a random starting hash or 1 instead of 0?
1741 patHash = A.foldl' addCksum 0 patArr
1743 skip = return . Skip
1745 nextAfterInit nextGen stepRes =
1746 case stepRes of
1747 FL.Partial s -> nextGen s
1748 FL.Done b -> SplitOnSeqYield b (SplitOnSeqReinit nextGen)
1750 {-# INLINE yieldProceed #-}
1751 yieldProceed nextGen fs =
1752 initial >>= skip . SplitOnSeqYield fs . nextAfterInit nextGen
1754 {-# INLINE_LATE stepOuter #-}
1755 stepOuter _ SplitOnSeqInit = do
1756 res <- initial
1757 case res of
1758 FL.Partial acc ->
1759 if patLen == 0
1760 then return $ Skip $ SplitOnSeqEmpty acc state
1761 else if patLen == 1
1762 then do
1763 pat <- liftIO $ A.unsafeIndexIO 0 patArr
1764 return $ Skip $ SplitOnSeqSingle acc state pat
1765 else if SIZE_OF(a) * patLen
1766 <= sizeOf (undefined :: Word)
1767 then return $ Skip $ SplitOnSeqWordInit acc state
1768 else do
1769 (rb, rhead) <- liftIO $ RB.new patLen
1770 skip $ SplitOnSeqKRInit 0 acc state rb rhead
1771 FL.Done b -> skip $ SplitOnSeqYield b SplitOnSeqInit
1773 stepOuter _ (SplitOnSeqYield x next) = return $ Yield x next
1775 ---------------------------
1776 -- Checkpoint
1777 ---------------------------
1779 stepOuter _ (SplitOnSeqReinit nextGen) =
1780 initial >>= skip . nextAfterInit nextGen
1782 ---------------------------
1783 -- Empty pattern
1784 ---------------------------
1786 stepOuter gst (SplitOnSeqEmpty acc st) = do
1787 res <- step (adaptState gst) st
1788 case res of
1789 Yield x s -> do
1790 r <- fstep acc x
1791 b1 <-
1792 case r of
1793 FL.Partial acc1 -> done acc1
1794 FL.Done b -> return b
1795 let jump c = SplitOnSeqEmpty c s
1796 in yieldProceed jump b1
1797 Skip s -> skip (SplitOnSeqEmpty acc s)
1798 Stop -> return Stop
1800 -----------------
1801 -- Done
1802 -----------------
1804 stepOuter _ SplitOnSeqDone = return Stop
1806 -----------------
1807 -- Single Pattern
1808 -----------------
1810 stepOuter gst (SplitOnSeqSingle fs st pat) = do
1811 res <- step (adaptState gst) st
1812 case res of
1813 Yield x s -> do
1814 let jump c = SplitOnSeqSingle c s pat
1815 if pat == x
1816 then done fs >>= yieldProceed jump
1817 else do
1818 r <- fstep fs x
1819 case r of
1820 FL.Partial fs1 -> skip $ jump fs1
1821 FL.Done b -> yieldProceed jump b
1822 Skip s -> return $ Skip $ SplitOnSeqSingle fs s pat
1823 Stop -> do
1824 r <- done fs
1825 return $ Skip $ SplitOnSeqYield r SplitOnSeqDone
1827 ---------------------------
1828 -- Short Pattern - Shift Or
1829 ---------------------------
1831 stepOuter _ (SplitOnSeqWordDone 0 fs _) = do
1832 r <- done fs
1833 skip $ SplitOnSeqYield r SplitOnSeqDone
1834 stepOuter _ (SplitOnSeqWordDone n fs wrd) = do
1835 let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
1836 r <- fstep fs (toEnum $ fromIntegral old)
1837 case r of
1838 FL.Partial fs1 -> skip $ SplitOnSeqWordDone (n - 1) fs1 wrd
1839 FL.Done b -> do
1840 let jump c = SplitOnSeqWordDone (n - 1) c wrd
1841 yieldProceed jump b
1843 stepOuter gst (SplitOnSeqWordInit fs st0) =
1844 go SPEC 0 0 st0
1846 where
1848 {-# INLINE go #-}
1849 go !_ !idx !wrd !st = do
1850 res <- step (adaptState gst) st
1851 case res of
1852 Yield x s -> do
1853 let wrd1 = addToWord wrd x
1854 if idx == maxIndex
1855 then do
1856 if wrd1 .&. wordMask == wordPat
1857 then do
1858 let jump c = SplitOnSeqWordInit c s
1859 done fs >>= yieldProceed jump
1860 else skip $ SplitOnSeqWordLoop wrd1 s fs
1861 else go SPEC (idx + 1) wrd1 s
1862 Skip s -> go SPEC idx wrd s
1863 Stop -> do
1864 if idx /= 0
1865 then skip $ SplitOnSeqWordDone idx fs wrd
1866 else do
1867 r <- done fs
1868 skip $ SplitOnSeqYield r SplitOnSeqDone
1870 stepOuter gst (SplitOnSeqWordLoop wrd0 st0 fs0) =
1871 go SPEC wrd0 st0 fs0
1873 where
1875 {-# INLINE go #-}
1876 go !_ !wrd !st !fs = do
1877 res <- step (adaptState gst) st
1878 case res of
1879 Yield x s -> do
1880 let jump c = SplitOnSeqWordInit c s
1881 wrd1 = addToWord wrd x
1882 old = (wordMask .&. wrd)
1883 `shiftR` (elemBits * (patLen - 1))
1884 r <- fstep fs (toEnum $ fromIntegral old)
1885 case r of
1886 FL.Partial fs1 -> do
1887 if wrd1 .&. wordMask == wordPat
1888 then done fs1 >>= yieldProceed jump
1889 else go SPEC wrd1 s fs1
1890 FL.Done b -> yieldProceed jump b
1891 Skip s -> go SPEC wrd s fs
1892 Stop -> skip $ SplitOnSeqWordDone patLen fs wrd
1894 -------------------------------
1895 -- General Pattern - Karp Rabin
1896 -------------------------------
1898 stepOuter gst (SplitOnSeqKRInit idx fs st rb rh) = do
1899 res <- step (adaptState gst) st
1900 case res of
1901 Yield x s -> do
1902 rh1 <- liftIO $ RB.unsafeInsert rb rh x
1903 if idx == maxIndex
1904 then do
1905 let fld = RB.unsafeFoldRing (RB.ringBound rb)
1906 let !ringHash = fld addCksum 0 rb
1907 if ringHash == patHash
1908 then skip $ SplitOnSeqKRCheck fs s rb rh1
1909 else skip $ SplitOnSeqKRLoop fs s rb rh1 ringHash
1910 else skip $ SplitOnSeqKRInit (idx + 1) fs s rb rh1
1911 Skip s -> skip $ SplitOnSeqKRInit idx fs s rb rh
1912 Stop -> do
1913 skip $ SplitOnSeqKRDone idx fs rb (RB.startOf rb)
1915 -- XXX The recursive "go" is more efficient than the state based recursion
1916 -- code commented out below. Perhaps its more efficient because of
1917 -- factoring out "rb" outside the loop.
1918 --
1919 stepOuter gst (SplitOnSeqKRLoop fs0 st0 rb rh0 cksum0) =
1920 go SPEC fs0 st0 rh0 cksum0
1922 where
1924 go !_ !fs !st !rh !cksum = do
1925 res <- step (adaptState gst) st
1926 case res of
1927 Yield x s -> do
1928 old <- liftIO $ peek rh
1929 let cksum1 = deltaCksum cksum old x
1930 r <- fstep fs old
1931 case r of
1932 FL.Partial fs1 -> do
1933 rh1 <- liftIO (RB.unsafeInsert rb rh x)
1934 if cksum1 == patHash
1935 then skip $ SplitOnSeqKRCheck fs1 s rb rh1
1936 else go SPEC fs1 s rh1 cksum1
1937 FL.Done b -> do
1938 let rst = RB.startOf rb
1939 jump c = SplitOnSeqKRInit 0 c s rb rst
1940 yieldProceed jump b
1941 Skip s -> go SPEC fs s rh cksum
1942 Stop -> skip $ SplitOnSeqKRDone patLen fs rb rh
1944 -- XXX The following code is 5 times slower compared to the recursive loop
1945 -- based code above. Need to investigate why. One possibility is that the
1946 -- go loop above does not thread around the ring buffer (rb). This code may
1947 -- be causing the state to bloat and getting allocated on each iteration.
1948 -- We can check the cmm/asm code to confirm. If so a good GHC solution to
1949 -- such problem is needed. One way to avoid this could be to use unboxed
1950 -- mutable state?
1951 {-
1952 stepOuter gst (SplitOnSeqKRLoop fs st rb rh cksum) = do
1953 res <- step (adaptState gst) st
1954 case res of
1955 Yield x s -> do
1956 old <- liftIO $ peek rh
1957 let cksum1 = deltaCksum cksum old x
1958 fs1 <- fstep fs old
1959 if (cksum1 == patHash)
1960 then do
1961 r <- done fs1
1962 skip $ SplitOnSeqYield r $ SplitOnSeqKRInit 0 s rb rh
1963 else do
1964 rh1 <- liftIO (RB.unsafeInsert rb rh x)
1965 skip $ SplitOnSeqKRLoop fs1 s rb rh1 cksum1
1966 Skip s -> skip $ SplitOnSeqKRLoop fs s rb rh cksum
1967 Stop -> skip $ SplitOnSeqKRDone patLen fs rb rh
1968 -}
1970 stepOuter _ (SplitOnSeqKRCheck fs st rb rh) = do
1971 if RB.unsafeEqArray rb rh patArr
1972 then do
1973 r <- done fs
1974 let rst = RB.startOf rb
1975 jump c = SplitOnSeqKRInit 0 c st rb rst
1976 yieldProceed jump r
1977 else skip $ SplitOnSeqKRLoop fs st rb rh patHash
1979 stepOuter _ (SplitOnSeqKRDone 0 fs _ _) = do
1980 r <- done fs
1981 skip $ SplitOnSeqYield r SplitOnSeqDone
1982 stepOuter _ (SplitOnSeqKRDone n fs rb rh) = do
1983 old <- liftIO $ peek rh
1984 let rh1 = RB.advance rb rh
1985 r <- fstep fs old
1986 case r of
1987 FL.Partial fs1 -> skip $ SplitOnSeqKRDone (n - 1) fs1 rb rh1
1988 FL.Done b -> do
1989 let jump c = SplitOnSeqKRDone (n - 1) c rb rh1
1990 yieldProceed jump b
1992{-# ANN type SplitOnSuffixSeqState Fuse #-}
1993data SplitOnSuffixSeqState rb rh ck w fs s b x =
1994 SplitOnSuffixSeqInit
1995 | SplitOnSuffixSeqYield b (SplitOnSuffixSeqState rb rh ck w fs s b x)
1996 | SplitOnSuffixSeqDone
1998 | SplitOnSuffixSeqEmpty !fs s
2000 | SplitOnSuffixSeqSingleInit !fs s x
2001 | SplitOnSuffixSeqSingle !fs s x
2003 | SplitOnSuffixSeqWordInit !fs s
2004 | SplitOnSuffixSeqWordLoop !w s !fs
2005 | SplitOnSuffixSeqWordDone Int !fs !w
2007 | SplitOnSuffixSeqKRInit Int !fs s rb !rh
2008 | SplitOnSuffixSeqKRInit1 !fs s rb !rh
2009 | SplitOnSuffixSeqKRLoop fs s rb !rh !ck
2010 | SplitOnSuffixSeqKRCheck fs s rb !rh
2011 | SplitOnSuffixSeqKRDone Int !fs rb !rh
2013 | SplitOnSuffixSeqReinit
2014 (fs -> SplitOnSuffixSeqState rb rh ck w fs s b x)
2016{-# INLINE_NORMAL splitOnSuffixSeq #-}
2017splitOnSuffixSeq
2018 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
2019 => Bool
2020 -> Array a
2021 -> Fold m a b
2022 -> Stream m a
2023 -> Stream m b
2024splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
2025 Stream stepOuter SplitOnSuffixSeqInit
2027 where
2029 patLen = A.length patArr
2030 maxIndex = patLen - 1
2031 elemBits = SIZE_OF(a) * 8
2033 -- For word pattern case
2034 wordMask :: Word
2035 wordMask = (1 `shiftL` (elemBits * patLen)) - 1
2037 elemMask :: Word
2038 elemMask = (1 `shiftL` elemBits) - 1
2040 wordPat :: Word
2041 wordPat = wordMask .&. A.foldl' addToWord 0 patArr
2043 addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
2045 nextAfterInit nextGen stepRes =
2046 case stepRes of
2047 FL.Partial s -> nextGen s
2048 FL.Done b ->
2049 SplitOnSuffixSeqYield b (SplitOnSuffixSeqReinit nextGen)
2051 {-# INLINE yieldProceed #-}
2052 yieldProceed nextGen fs =
2053 initial >>= skip . SplitOnSuffixSeqYield fs . nextAfterInit nextGen
2055 -- For single element pattern case
2056 {-# INLINE processYieldSingle #-}
2057 processYieldSingle pat x s fs = do
2058 let jump c = SplitOnSuffixSeqSingleInit c s pat
2059 if pat == x
2060 then do
2061 r <- if withSep then fstep fs x else return $ FL.Partial fs
2062 b1 <-
2063 case r of
2064 FL.Partial fs1 -> done fs1
2065 FL.Done b -> return b
2066 yieldProceed jump b1
2067 else do
2068 r <- fstep fs x
2069 case r of
2070 FL.Partial fs1 -> skip $ SplitOnSuffixSeqSingle fs1 s pat
2071 FL.Done b -> yieldProceed jump b
2073 -- For Rabin-Karp search
2074 k = 2891336453 :: Word32
2075 coeff = k ^ patLen
2077 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
2079 deltaCksum cksum old new =
2080 addCksum cksum new - coeff * fromIntegral (fromEnum old)
2082 -- XXX shall we use a random starting hash or 1 instead of 0?
2083 patHash = A.foldl' addCksum 0 patArr
2085 skip = return . Skip
2087 {-# INLINE_LATE stepOuter #-}
2088 stepOuter _ SplitOnSuffixSeqInit = do
2089 res <- initial
2090 case res of
2091 FL.Partial fs ->
2092 if patLen == 0
2093 then skip $ SplitOnSuffixSeqEmpty fs state
2094 else if patLen == 1
2095 then do
2096 pat <- liftIO $ A.unsafeIndexIO 0 patArr
2097 skip $ SplitOnSuffixSeqSingleInit fs state pat
2098 else if SIZE_OF(a) * patLen
2099 <= sizeOf (undefined :: Word)
2100 then skip $ SplitOnSuffixSeqWordInit fs state
2101 else do
2102 (rb, rhead) <- liftIO $ RB.new patLen
2103 skip $ SplitOnSuffixSeqKRInit 0 fs state rb rhead
2104 FL.Done fb -> skip $ SplitOnSuffixSeqYield fb SplitOnSuffixSeqInit
2106 stepOuter _ (SplitOnSuffixSeqYield x next) = return $ Yield x next
2108 ---------------------------
2109 -- Reinit
2110 ---------------------------
2112 stepOuter _ (SplitOnSuffixSeqReinit nextGen) =
2113 initial >>= skip . nextAfterInit nextGen
2115 ---------------------------
2116 -- Empty pattern
2117 ---------------------------
2119 stepOuter gst (SplitOnSuffixSeqEmpty acc st) = do
2120 res <- step (adaptState gst) st
2121 case res of
2122 Yield x s -> do
2123 let jump c = SplitOnSuffixSeqEmpty c s
2124 r <- fstep acc x
2125 b1 <-
2126 case r of
2127 FL.Partial fs -> done fs
2128 FL.Done b -> return b
2129 yieldProceed jump b1
2130 Skip s -> skip (SplitOnSuffixSeqEmpty acc s)
2131 Stop -> return Stop
2133 -----------------
2134 -- Done
2135 -----------------
2137 stepOuter _ SplitOnSuffixSeqDone = return Stop
2139 -----------------
2140 -- Single Pattern
2141 -----------------
2143 stepOuter gst (SplitOnSuffixSeqSingleInit fs st pat) = do
2144 res <- step (adaptState gst) st
2145 case res of
2146 Yield x s -> processYieldSingle pat x s fs
2147 Skip s -> skip $ SplitOnSuffixSeqSingleInit fs s pat
2148 Stop -> return Stop
2150 stepOuter gst (SplitOnSuffixSeqSingle fs st pat) = do
2151 res <- step (adaptState gst) st
2152 case res of
2153 Yield x s -> processYieldSingle pat x s fs
2154 Skip s -> skip $ SplitOnSuffixSeqSingle fs s pat
2155 Stop -> do
2156 r <- done fs
2157 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2159 ---------------------------
2160 -- Short Pattern - Shift Or
2161 ---------------------------
2163 stepOuter _ (SplitOnSuffixSeqWordDone 0 fs _) = do
2164 r <- done fs
2165 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2166 stepOuter _ (SplitOnSuffixSeqWordDone n fs wrd) = do
2167 let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
2168 r <- fstep fs (toEnum $ fromIntegral old)
2169 case r of
2170 FL.Partial fs1 -> skip $ SplitOnSuffixSeqWordDone (n - 1) fs1 wrd
2171 FL.Done b -> do
2172 let jump c = SplitOnSuffixSeqWordDone (n - 1) c wrd
2173 yieldProceed jump b
2175 stepOuter gst (SplitOnSuffixSeqWordInit fs0 st0) = do
2176 res <- step (adaptState gst) st0
2177 case res of
2178 Yield x s -> do
2179 let wrd = addToWord 0 x
2180 r <- if withSep then fstep fs0 x else return $ FL.Partial fs0
2181 case r of
2182 FL.Partial fs1 -> go SPEC 1 wrd s fs1
2183 FL.Done b -> do
2184 let jump c = SplitOnSuffixSeqWordInit c s
2185 yieldProceed jump b
2186 Skip s -> skip (SplitOnSuffixSeqWordInit fs0 s)
2187 Stop -> return Stop
2189 where
2191 {-# INLINE go #-}
2192 go !_ !idx !wrd !st !fs = do
2193 res <- step (adaptState gst) st
2194 case res of
2195 Yield x s -> do
2196 let jump c = SplitOnSuffixSeqWordInit c s
2197 let wrd1 = addToWord wrd x
2198 r <- if withSep then fstep fs x else return $ FL.Partial fs
2199 case r of
2200 FL.Partial fs1 ->
2201 if idx /= maxIndex
2202 then go SPEC (idx + 1) wrd1 s fs1
2203 else if wrd1 .&. wordMask /= wordPat
2204 then skip $ SplitOnSuffixSeqWordLoop wrd1 s fs1
2205 else do done fs >>= yieldProceed jump
2206 FL.Done b -> yieldProceed jump b
2207 Skip s -> go SPEC idx wrd s fs
2208 Stop -> skip $ SplitOnSuffixSeqWordDone idx fs wrd
2210 stepOuter gst (SplitOnSuffixSeqWordLoop wrd0 st0 fs0) =
2211 go SPEC wrd0 st0 fs0
2213 where
2215 {-# INLINE go #-}
2216 go !_ !wrd !st !fs = do
2217 res <- step (adaptState gst) st
2218 case res of
2219 Yield x s -> do
2220 let jump c = SplitOnSuffixSeqWordInit c s
2221 wrd1 = addToWord wrd x
2222 old = (wordMask .&. wrd)
2223 `shiftR` (elemBits * (patLen - 1))
2224 r <-
2225 if withSep
2226 then fstep fs x
2227 else fstep fs (toEnum $ fromIntegral old)
2228 case r of
2229 FL.Partial fs1 ->
2230 if wrd1 .&. wordMask == wordPat
2231 then done fs1 >>= yieldProceed jump
2232 else go SPEC wrd1 s fs1
2233 FL.Done b -> yieldProceed jump b
2234 Skip s -> go SPEC wrd s fs
2235 Stop ->
2236 if wrd .&. wordMask == wordPat
2237 then return Stop
2238 else if withSep
2239 then do
2240 r <- done fs
2241 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2242 else skip $ SplitOnSuffixSeqWordDone patLen fs wrd
2244 -------------------------------
2245 -- General Pattern - Karp Rabin
2246 -------------------------------
2248 stepOuter gst (SplitOnSuffixSeqKRInit idx0 fs st0 rb rh0) = do
2249 res <- step (adaptState gst) st0
2250 case res of
2251 Yield x s -> do
2252 rh1 <- liftIO $ RB.unsafeInsert rb rh0 x
2253 r <- if withSep then fstep fs x else return $ FL.Partial fs
2254 case r of
2255 FL.Partial fs1 ->
2256 skip $ SplitOnSuffixSeqKRInit1 fs1 s rb rh1
2257 FL.Done b -> do
2258 let rst = RB.startOf rb
2259 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2260 yieldProceed jump b
2261 Skip s -> skip $ SplitOnSuffixSeqKRInit idx0 fs s rb rh0
2262 Stop -> return Stop
2264 stepOuter gst (SplitOnSuffixSeqKRInit1 fs0 st0 rb rh0) = do
2265 go SPEC 1 rh0 st0 fs0
2267 where
2269 go !_ !idx !rh st !fs = do
2270 res <- step (adaptState gst) st
2271 case res of
2272 Yield x s -> do
2273 rh1 <- liftIO (RB.unsafeInsert rb rh x)
2274 r <- if withSep then fstep fs x else return $ FL.Partial fs
2275 case r of
2276 FL.Partial fs1 ->
2277 if idx /= maxIndex
2278 then go SPEC (idx + 1) rh1 s fs1
2279 else skip $
2280 let fld = RB.unsafeFoldRing (RB.ringBound rb)
2281 !ringHash = fld addCksum 0 rb
2282 in if ringHash == patHash
2283 then SplitOnSuffixSeqKRCheck fs1 s rb rh1
2284 else SplitOnSuffixSeqKRLoop
2285 fs1 s rb rh1 ringHash
2286 FL.Done b -> do
2287 let rst = RB.startOf rb
2288 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2289 yieldProceed jump b
2290 Skip s -> go SPEC idx rh s fs
2291 Stop -> do
2292 -- do not issue a blank segment when we end at pattern
2293 if (idx == maxIndex) && RB.unsafeEqArray rb rh patArr
2294 then return Stop
2295 else if withSep
2296 then do
2297 r <- done fs
2298 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2299 else skip $ SplitOnSuffixSeqKRDone idx fs rb (RB.startOf rb)
2301 stepOuter gst (SplitOnSuffixSeqKRLoop fs0 st0 rb rh0 cksum0) =
2302 go SPEC fs0 st0 rh0 cksum0
2304 where
2306 go !_ !fs !st !rh !cksum = do
2307 res <- step (adaptState gst) st
2308 case res of
2309 Yield x s -> do
2310 old <- liftIO $ peek rh
2311 rh1 <- liftIO (RB.unsafeInsert rb rh x)
2312 let cksum1 = deltaCksum cksum old x
2313 r <- if withSep then fstep fs x else fstep fs old
2314 case r of
2315 FL.Partial fs1 ->
2316 if cksum1 /= patHash
2317 then go SPEC fs1 s rh1 cksum1
2318 else skip $ SplitOnSuffixSeqKRCheck fs1 s rb rh1
2319 FL.Done b -> do
2320 let rst = RB.startOf rb
2321 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2322 yieldProceed jump b
2323 Skip s -> go SPEC fs s rh cksum
2324 Stop ->
2325 if RB.unsafeEqArray rb rh patArr
2326 then return Stop
2327 else if withSep
2328 then do
2329 r <- done fs
2330 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2331 else skip $ SplitOnSuffixSeqKRDone patLen fs rb rh
2333 stepOuter _ (SplitOnSuffixSeqKRCheck fs st rb rh) = do
2334 if RB.unsafeEqArray rb rh patArr
2335 then do
2336 r <- done fs
2337 let rst = RB.startOf rb
2338 jump c = SplitOnSuffixSeqKRInit 0 c st rb rst
2339 yieldProceed jump r
2340 else skip $ SplitOnSuffixSeqKRLoop fs st rb rh patHash
2342 stepOuter _ (SplitOnSuffixSeqKRDone 0 fs _ _) = do
2343 r <- done fs
2344 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2345 stepOuter _ (SplitOnSuffixSeqKRDone n fs rb rh) = do
2346 old <- liftIO $ peek rh
2347 let rh1 = RB.advance rb rh
2348 r <- fstep fs old
2349 case r of
2350 FL.Partial fs1 -> skip $ SplitOnSuffixSeqKRDone (n - 1) fs1 rb rh1
2351 FL.Done b -> do
2352 let jump c = SplitOnSuffixSeqKRDone (n - 1) c rb rh1
2353 yieldProceed jump b
2355------------------------------------------------------------------------------
2356-- Nested Container Transformation
2357------------------------------------------------------------------------------
2359{-# ANN type SplitState Fuse #-}
2360data SplitState s arr
2361 = SplitInitial s
2362 | SplitBuffering s arr
2363 | SplitSplitting s arr
2364 | SplitYielding arr (SplitState s arr)
2365 | SplitFinishing
2367-- XXX An alternative approach would be to use a partial fold (Fold m a b) to
2368-- split using a splitBy like combinator. The Fold would consume upto the
2369-- separator and return any leftover which can then be fed to the next fold.
2371-- We can revisit this once we have partial folds/parsers.
2373-- | Performs infix separator style splitting.
2374{-# INLINE_NORMAL splitInnerBy #-}
2375splitInnerBy
2376 :: Monad m
2377 => (f a -> m (f a, Maybe (f a))) -- splitter
2378 -> (f a -> f a -> m (f a)) -- joiner
2379 -> Stream m (f a)
2380 -> Stream m (f a)
2381splitInnerBy splitter joiner (Stream step1 state1) =
2382 Stream step (SplitInitial state1)
2384 where
2386 {-# INLINE_LATE step #-}
2387 step gst (SplitInitial st) = do
2388 r <- step1 gst st
2389 case r of
2390 Yield x s -> do
2391 (x1, mx2) <- splitter x
2392 return $ case mx2 of
2393 Nothing -> Skip (SplitBuffering s x1)
2394 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
2395 Skip s -> return $ Skip (SplitInitial s)
2396 Stop -> return Stop
2398 step gst (SplitBuffering st buf) = do
2399 r <- step1 gst st
2400 case r of
2401 Yield x s -> do
2402 (x1, mx2) <- splitter x
2403 buf' <- joiner buf x1
2404 return $ case mx2 of
2405 Nothing -> Skip (SplitBuffering s buf')
2406 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
2407 Skip s -> return $ Skip (SplitBuffering s buf)
2408 Stop -> return $ Skip (SplitYielding buf SplitFinishing)
2410 step _ (SplitSplitting st buf) = do
2411 (x1, mx2) <- splitter buf
2412 return $ case mx2 of
2413 Nothing -> Skip $ SplitBuffering st x1
2414 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
2416 step _ (SplitYielding x next) = return $ Yield x next
2417 step _ SplitFinishing = return Stop
2419-- | Performs infix separator style splitting.
2420{-# INLINE_NORMAL splitInnerBySuffix #-}
2421splitInnerBySuffix
2422 :: (Monad m, Eq (f a), Monoid (f a))
2423 => (f a -> m (f a, Maybe (f a))) -- splitter
2424 -> (f a -> f a -> m (f a)) -- joiner
2425 -> Stream m (f a)
2426 -> Stream m (f a)
2427splitInnerBySuffix splitter joiner (Stream step1 state1) =
2428 Stream step (SplitInitial state1)
2430 where
2432 {-# INLINE_LATE step #-}
2433 step gst (SplitInitial st) = do
2434 r <- step1 gst st
2435 case r of
2436 Yield x s -> do
2437 (x1, mx2) <- splitter x
2438 return $ case mx2 of
2439 Nothing -> Skip (SplitBuffering s x1)
2440 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
2441 Skip s -> return $ Skip (SplitInitial s)
2442 Stop -> return Stop
2444 step gst (SplitBuffering st buf) = do
2445 r <- step1 gst st
2446 case r of
2447 Yield x s -> do
2448 (x1, mx2) <- splitter x
2449 buf' <- joiner buf x1
2450 return $ case mx2 of
2451 Nothing -> Skip (SplitBuffering s buf')
2452 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
2453 Skip s -> return $ Skip (SplitBuffering s buf)
2454 Stop -> return $
2455 if buf == mempty
2456 then Stop
2457 else Skip (SplitYielding buf SplitFinishing)
2459 step _ (SplitSplitting st buf) = do
2460 (x1, mx2) <- splitter buf
2461 return $ case mx2 of
2462 Nothing -> Skip $ SplitBuffering st x1
2463 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
2465 step _ (SplitYielding x next) = return $ Yield x next
2466 step _ SplitFinishing = return Stop