git.iko.soy/streamly
on d2cde55 help sign in
streamly/src/Streamly/Internal/Data/Stream/StreamD/Nesting.hs
2423 lines 90 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 )
146where
148#include "inline.hs"
150import Control.Exception (assert)
151import Control.Monad.Catch (MonadThrow, throwM)
152import Control.Monad.IO.Class (MonadIO(..))
153import Data.Bits (shiftR, shiftL, (.|.), (.&.))
154#if __GLASGOW_HASKELL__ >= 801
155import Data.Functor.Identity ( Identity )
156#endif
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#if __GLASGOW_HASKELL__ >= 801
426{-# RULES "zipWithM xs xs"
427 forall f xs. zipWithM @Identity f xs xs = mapM (\x -> f x x) xs #-}
428#endif
430{-# INLINE zipWith #-}
431zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
432zipWith f = zipWithM (\a b -> return (f a b))
434------------------------------------------------------------------------------
435-- Merging
436------------------------------------------------------------------------------
438{-# INLINE_NORMAL mergeByM #-}
439mergeByM
440 :: (Monad m)
441 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
442mergeByM cmp (Stream stepa ta) (Stream stepb tb) =
443 Stream step (Just ta, Just tb, Nothing, Nothing)
444 where
445 {-# INLINE_LATE step #-}
447 -- one of the values is missing, and the corresponding stream is running
448 step gst (Just sa, sb, Nothing, b) = do
449 r <- stepa gst sa
450 return $ case r of
451 Yield a sa' -> Skip (Just sa', sb, Just a, b)
452 Skip sa' -> Skip (Just sa', sb, Nothing, b)
453 Stop -> Skip (Nothing, sb, Nothing, b)
455 step gst (sa, Just sb, a, Nothing) = do
456 r <- stepb gst sb
457 return $ case r of
458 Yield b sb' -> Skip (sa, Just sb', a, Just b)
459 Skip sb' -> Skip (sa, Just sb', a, Nothing)
460 Stop -> Skip (sa, Nothing, a, Nothing)
462 -- both the values are available
463 step _ (sa, sb, Just a, Just b) = do
464 res <- cmp a b
465 return $ case res of
466 GT -> Yield b (sa, sb, Just a, Nothing)
467 _ -> Yield a (sa, sb, Nothing, Just b)
469 -- one of the values is missing, corresponding stream is done
470 step _ (Nothing, sb, Nothing, Just b) =
471 return $ Yield b (Nothing, sb, Nothing, Nothing)
473 step _ (sa, Nothing, Just a, Nothing) =
474 return $ Yield a (sa, Nothing, Nothing, Nothing)
476 step _ (Nothing, Nothing, Nothing, Nothing) = return Stop
478{-# INLINE mergeBy #-}
479mergeBy
480 :: (Monad m)
481 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
482mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
484------------------------------------------------------------------------------
485-- Combine N Streams - unfoldMany
486------------------------------------------------------------------------------
488data ConcatUnfoldInterleaveState o i =
489 ConcatUnfoldInterleaveOuter o [i]
490 | ConcatUnfoldInterleaveInner o [i]
491 | ConcatUnfoldInterleaveInnerL [i] [i]
492 | ConcatUnfoldInterleaveInnerR [i] [i]
494-- XXX use arrays to store state instead of lists?
495--
496-- XXX In general we can use different scheduling strategies e.g. how to
497-- schedule the outer vs inner loop or assigning weights to different streams
498-- or outer and inner loops.
500-- After a yield, switch to the next stream. Do not switch streams on Skip.
501-- Yield from outer stream switches to the inner stream.
502--
503-- There are two choices here, (1) exhaust the outer stream first and then
504-- start yielding from the inner streams, this is much simpler to implement,
505-- (2) yield at least one element from an inner stream before going back to
506-- outer stream and opening the next stream from it.
507--
508-- Ideally, we need some scheduling bias to inner streams vs outer stream.
509-- Maybe we can configure the behavior.
510--
511-- XXX Instead of using "concatPairsWith wSerial" we can implement an N-way
512-- interleaving CPS combinator which behaves like unfoldManyInterleave. Intead
513-- of pairing up the streams We just need to go yielding one element from each
514-- stream and storing the remaining streams and then keep doing rounds through
515-- those in a round robin fashion. This would be much like wAsync.
516--
517-- | This does not pair streams like concatPairsWith, instead, it goes through
518-- each stream one by one and yields one element from each stream. After it
519-- goes to the last stream it reverses the traversal to come back to the first
520-- stream yielding elements from each stream on its way back to the first
521-- stream and so on.
522--
523-- >>> input = Stream.fromList [[1,1],[2,2],[3,3],[4,4],[5,5]]
524-- >>> Stream.toList $ Stream.unfoldManyInterleave Unfold.fromList input
525-- [1,2,3,4,5,5,4,3,2,1]
526--
527-- Note that this is order of magnitude more efficient than "concatPairsWith
528-- wSerial"
529{-# INLINE_NORMAL unfoldManyInterleave #-}
530unfoldManyInterleave :: Monad m => Unfold m a b -> Stream m a -> Stream m b
531unfoldManyInterleave (Unfold istep inject) (Stream ostep ost) =
532 Stream step (ConcatUnfoldInterleaveOuter ost [])
534 where
536 {-# INLINE_LATE step #-}
537 step gst (ConcatUnfoldInterleaveOuter o ls) = do
538 r <- ostep (adaptState gst) o
539 case r of
540 Yield a o' -> do
541 i <- inject a
542 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
543 Skip o' -> return $ Skip (ConcatUnfoldInterleaveOuter o' ls)
544 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
546 step _ (ConcatUnfoldInterleaveInner _ []) = undefined
547 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
548 r <- istep st
549 return $ case r of
550 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
551 Skip s -> Skip (ConcatUnfoldInterleaveInner o (s:ls))
552 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
554 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
555 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
556 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
558 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
559 r <- istep st
560 return $ case r of
561 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
562 Skip s -> Skip (ConcatUnfoldInterleaveInnerL (s:ls) rs)
563 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
565 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
566 step _ (ConcatUnfoldInterleaveInnerR ls []) =
567 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
569 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
570 r <- istep st
571 return $ case r of
572 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
573 Skip s -> Skip (ConcatUnfoldInterleaveInnerR ls (s:rs))
574 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
576-- XXX In general we can use different scheduling strategies e.g. how to
577-- schedule the outer vs inner loop or assigning weights to different streams
578-- or outer and inner loops.
579--
580-- This could be inefficient if the tasks are too small.
581--
582-- Compared to unfoldManyInterleave this one switches streams on Skips.
583--
584{-# INLINE_NORMAL unfoldManyRoundRobin #-}
585unfoldManyRoundRobin :: Monad m => Unfold m a b -> Stream m a -> Stream m b
586unfoldManyRoundRobin (Unfold istep inject) (Stream ostep ost) =
587 Stream step (ConcatUnfoldInterleaveOuter ost [])
588 where
589 {-# INLINE_LATE step #-}
590 step gst (ConcatUnfoldInterleaveOuter o ls) = do
591 r <- ostep (adaptState gst) o
592 case r of
593 Yield a o' -> do
594 i <- inject a
595 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
596 Skip o' -> return $ Skip (ConcatUnfoldInterleaveInner o' ls)
597 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
599 step _ (ConcatUnfoldInterleaveInner o []) =
600 return $ Skip (ConcatUnfoldInterleaveOuter o [])
602 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
603 r <- istep st
604 return $ case r of
605 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
606 Skip s -> Skip (ConcatUnfoldInterleaveOuter o (s:ls))
607 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
609 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
610 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
611 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
613 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
614 r <- istep st
615 return $ case r of
616 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
617 Skip s -> Skip (ConcatUnfoldInterleaveInnerL ls (s:rs))
618 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
620 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
621 step _ (ConcatUnfoldInterleaveInnerR ls []) =
622 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
624 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
625 r <- istep st
626 return $ case r of
627 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
628 Skip s -> Skip (ConcatUnfoldInterleaveInnerR (s:ls) rs)
629 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
631------------------------------------------------------------------------------
632-- Combine N Streams - interpose
633------------------------------------------------------------------------------
635{-# ANN type InterposeSuffixState Fuse #-}
636data InterposeSuffixState s1 i1 =
637 InterposeSuffixFirst s1
638 -- | InterposeSuffixFirstYield s1 i1
639 | InterposeSuffixFirstInner s1 i1
640 | InterposeSuffixSecond s1
642-- Note that if an unfolded layer turns out to be nil we still emit the
643-- separator effect. An alternate behavior could be to emit the separator
644-- effect only if at least one element has been yielded by the unfolding.
645-- However, that becomes a bit complicated, so we have chosen the former
646-- behvaior for now.
647{-# INLINE_NORMAL interposeSuffix #-}
648interposeSuffix
649 :: Monad m
650 => m c -> Unfold m b c -> Stream m b -> Stream m c
651interposeSuffix
652 action
653 (Unfold istep1 inject1) (Stream step1 state1) =
654 Stream step (InterposeSuffixFirst state1)
656 where
658 {-# INLINE_LATE step #-}
659 step gst (InterposeSuffixFirst s1) = do
660 r <- step1 (adaptState gst) s1
661 case r of
662 Yield a s -> do
663 i <- inject1 a
664 i `seq` return (Skip (InterposeSuffixFirstInner s i))
665 -- i `seq` return (Skip (InterposeSuffixFirstYield s i))
666 Skip s -> return $ Skip (InterposeSuffixFirst s)
667 Stop -> return Stop
669 {-
670 step _ (InterposeSuffixFirstYield s1 i1) = do
671 r <- istep1 i1
672 return $ case r of
673 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
674 Skip i' -> Skip (InterposeSuffixFirstYield s1 i')
675 Stop -> Skip (InterposeSuffixFirst s1)
676 -}
678 step _ (InterposeSuffixFirstInner s1 i1) = do
679 r <- istep1 i1
680 return $ case r of
681 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
682 Skip i' -> Skip (InterposeSuffixFirstInner s1 i')
683 Stop -> Skip (InterposeSuffixSecond s1)
685 step _ (InterposeSuffixSecond s1) = do
686 r <- action
687 return $ Yield r (InterposeSuffixFirst s1)
689{-# ANN type InterposeState Fuse #-}
690data InterposeState s1 i1 a =
691 InterposeFirst s1
692 -- | InterposeFirstYield s1 i1
693 | InterposeFirstInner s1 i1
694 | InterposeFirstInject s1
695 -- | InterposeFirstBuf s1 i1
696 | InterposeSecondYield s1 i1
697 -- -- | InterposeSecondYield s1 i1 a
698 -- -- | InterposeFirstResume s1 i1 a
700-- Note that this only interposes the pure values, we may run many effects to
701-- generate those values as some effects may not generate anything (Skip).
702{-# INLINE_NORMAL interpose #-}
703interpose :: Monad m => m c -> Unfold m b c -> Stream m b -> Stream m c
704interpose
705 action
706 (Unfold istep1 inject1) (Stream step1 state1) =
707 Stream step (InterposeFirst state1)
709 where
711 {-# INLINE_LATE step #-}
712 step gst (InterposeFirst s1) = do
713 r <- step1 (adaptState gst) s1
714 case r of
715 Yield a s -> do
716 i <- inject1 a
717 i `seq` return (Skip (InterposeFirstInner s i))
718 -- i `seq` return (Skip (InterposeFirstYield s i))
719 Skip s -> return $ Skip (InterposeFirst s)
720 Stop -> return Stop
722 {-
723 step _ (InterposeFirstYield s1 i1) = do
724 r <- istep1 i1
725 return $ case r of
726 Yield x i' -> Yield x (InterposeFirstInner s1 i')
727 Skip i' -> Skip (InterposeFirstYield s1 i')
728 Stop -> Skip (InterposeFirst s1)
729 -}
731 step _ (InterposeFirstInner s1 i1) = do
732 r <- istep1 i1
733 return $ case r of
734 Yield x i' -> Yield x (InterposeFirstInner s1 i')
735 Skip i' -> Skip (InterposeFirstInner s1 i')
736 Stop -> Skip (InterposeFirstInject s1)
738 step gst (InterposeFirstInject s1) = do
739 r <- step1 (adaptState gst) s1
740 case r of
741 Yield a s -> do
742 i <- inject1 a
743 -- i `seq` return (Skip (InterposeFirstBuf s i))
744 i `seq` return (Skip (InterposeSecondYield s i))
745 Skip s -> return $ Skip (InterposeFirstInject s)
746 Stop -> return Stop
748 {-
749 step _ (InterposeFirstBuf s1 i1) = do
750 r <- istep1 i1
751 return $ case r of
752 Yield x i' -> Skip (InterposeSecondYield s1 i' x)
753 Skip i' -> Skip (InterposeFirstBuf s1 i')
754 Stop -> Stop
755 -}
757 {-
758 step _ (InterposeSecondYield s1 i1 v) = do
759 r <- action
760 return $ Yield r (InterposeFirstResume s1 i1 v)
761 -}
762 step _ (InterposeSecondYield s1 i1) = do
763 r <- action
764 return $ Yield r (InterposeFirstInner s1 i1)
766 {-
767 step _ (InterposeFirstResume s1 i1 v) = do
768 return $ Yield v (InterposeFirstInner s1 i1)
769 -}
771------------------------------------------------------------------------------
772-- Combine N Streams - intercalate
773------------------------------------------------------------------------------
775data ICUState s1 s2 i1 i2 =
776 ICUFirst s1 s2
777 | ICUSecond s1 s2
778 | ICUSecondOnly s2
779 | ICUFirstOnly s1
780 | ICUFirstInner s1 s2 i1
781 | ICUSecondInner s1 s2 i2
782 | ICUFirstOnlyInner s1 i1
783 | ICUSecondOnlyInner s2 i2
785-- | Interleave streams (full streams, not the elements) unfolded from two
786-- input streams and concat. Stop when the first stream stops. If the second
787-- stream ends before the first one then first stream still keeps running alone
788-- without any interleaving with the second stream.
789--
790-- [a1, a2, ... an] [b1, b2 ...]
791-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
792-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
793-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
794--
795{-# INLINE_NORMAL gintercalateSuffix #-}
796gintercalateSuffix
797 :: Monad m
798 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
799gintercalateSuffix
800 (Unfold istep1 inject1) (Stream step1 state1)
801 (Unfold istep2 inject2) (Stream step2 state2) =
802 Stream step (ICUFirst state1 state2)
804 where
806 {-# INLINE_LATE step #-}
807 step gst (ICUFirst s1 s2) = do
808 r <- step1 (adaptState gst) s1
809 case r of
810 Yield a s -> do
811 i <- inject1 a
812 i `seq` return (Skip (ICUFirstInner s s2 i))
813 Skip s -> return $ Skip (ICUFirst s s2)
814 Stop -> return Stop
816 step gst (ICUFirstOnly s1) = do
817 r <- step1 (adaptState gst) s1
818 case r of
819 Yield a s -> do
820 i <- inject1 a
821 i `seq` return (Skip (ICUFirstOnlyInner s i))
822 Skip s -> return $ Skip (ICUFirstOnly s)
823 Stop -> return Stop
825 step _ (ICUFirstInner s1 s2 i1) = do
826 r <- istep1 i1
827 return $ case r of
828 Yield x i' -> Yield x (ICUFirstInner s1 s2 i')
829 Skip i' -> Skip (ICUFirstInner s1 s2 i')
830 Stop -> Skip (ICUSecond s1 s2)
832 step _ (ICUFirstOnlyInner s1 i1) = do
833 r <- istep1 i1
834 return $ case r of
835 Yield x i' -> Yield x (ICUFirstOnlyInner s1 i')
836 Skip i' -> Skip (ICUFirstOnlyInner s1 i')
837 Stop -> Skip (ICUFirstOnly s1)
839 step gst (ICUSecond s1 s2) = do
840 r <- step2 (adaptState gst) s2
841 case r of
842 Yield a s -> do
843 i <- inject2 a
844 i `seq` return (Skip (ICUSecondInner s1 s i))
845 Skip s -> return $ Skip (ICUSecond s1 s)
846 Stop -> return $ Skip (ICUFirstOnly s1)
848 step _ (ICUSecondInner s1 s2 i2) = do
849 r <- istep2 i2
850 return $ case r of
851 Yield x i' -> Yield x (ICUSecondInner s1 s2 i')
852 Skip i' -> Skip (ICUSecondInner s1 s2 i')
853 Stop -> Skip (ICUFirst s1 s2)
855 step _ (ICUSecondOnly _s2) = undefined
856 step _ (ICUSecondOnlyInner _s2 _i2) = undefined
858data ICALState s1 s2 i1 i2 a =
859 ICALFirst s1 s2
860 -- | ICALFirstYield s1 s2 i1
861 | ICALFirstInner s1 s2 i1
862 | ICALFirstOnly s1
863 | ICALFirstOnlyInner s1 i1
864 | ICALSecondInject s1 s2
865 | ICALFirstInject s1 s2 i2
866 -- | ICALFirstBuf s1 s2 i1 i2
867 | ICALSecondInner s1 s2 i1 i2
868 -- -- | ICALSecondInner s1 s2 i1 i2 a
869 -- -- | ICALFirstResume s1 s2 i1 i2 a
871-- | Interleave streams (full streams, not the elements) unfolded from two
872-- input streams and concat. Stop when the first stream stops. If the second
873-- stream ends before the first one then first stream still keeps running alone
874-- without any interleaving with the second stream.
875--
876-- [a1, a2, ... an] [b1, b2 ...]
877-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
878-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
879-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
880--
881{-# INLINE_NORMAL gintercalate #-}
882gintercalate
883 :: Monad m
884 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
885gintercalate
886 (Unfold istep1 inject1) (Stream step1 state1)
887 (Unfold istep2 inject2) (Stream step2 state2) =
888 Stream step (ICALFirst state1 state2)
890 where
892 {-# INLINE_LATE step #-}
893 step gst (ICALFirst s1 s2) = do
894 r <- step1 (adaptState gst) s1
895 case r of
896 Yield a s -> do
897 i <- inject1 a
898 i `seq` return (Skip (ICALFirstInner s s2 i))
899 -- i `seq` return (Skip (ICALFirstYield s s2 i))
900 Skip s -> return $ Skip (ICALFirst s s2)
901 Stop -> return Stop
903 {-
904 step _ (ICALFirstYield s1 s2 i1) = do
905 r <- istep1 i1
906 return $ case r of
907 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
908 Skip i' -> Skip (ICALFirstYield s1 s2 i')
909 Stop -> Skip (ICALFirst s1 s2)
910 -}
912 step _ (ICALFirstInner s1 s2 i1) = do
913 r <- istep1 i1
914 return $ case r of
915 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
916 Skip i' -> Skip (ICALFirstInner s1 s2 i')
917 Stop -> Skip (ICALSecondInject s1 s2)
919 step gst (ICALFirstOnly s1) = do
920 r <- step1 (adaptState gst) s1
921 case r of
922 Yield a s -> do
923 i <- inject1 a
924 i `seq` return (Skip (ICALFirstOnlyInner s i))
925 Skip s -> return $ Skip (ICALFirstOnly s)
926 Stop -> return Stop
928 step _ (ICALFirstOnlyInner s1 i1) = do
929 r <- istep1 i1
930 return $ case r of
931 Yield x i' -> Yield x (ICALFirstOnlyInner s1 i')
932 Skip i' -> Skip (ICALFirstOnlyInner s1 i')
933 Stop -> Skip (ICALFirstOnly s1)
935 -- We inject the second stream even before checking if the first stream
936 -- would yield any more elements. There is no clear choice whether we
937 -- should do this before or after that. Doing it after may make the state
938 -- machine a bit simpler though.
939 step gst (ICALSecondInject s1 s2) = do
940 r <- step2 (adaptState gst) s2
941 case r of
942 Yield a s -> do
943 i <- inject2 a
944 i `seq` return (Skip (ICALFirstInject s1 s i))
945 Skip s -> return $ Skip (ICALSecondInject s1 s)
946 Stop -> return $ Skip (ICALFirstOnly s1)
948 step gst (ICALFirstInject s1 s2 i2) = do
949 r <- step1 (adaptState gst) s1
950 case r of
951 Yield a s -> do
952 i <- inject1 a
953 i `seq` return (Skip (ICALSecondInner s s2 i i2))
954 -- i `seq` return (Skip (ICALFirstBuf s s2 i i2))
955 Skip s -> return $ Skip (ICALFirstInject s s2 i2)
956 Stop -> return Stop
958 {-
959 step _ (ICALFirstBuf s1 s2 i1 i2) = do
960 r <- istep1 i1
961 return $ case r of
962 Yield x i' -> Skip (ICALSecondInner s1 s2 i' i2 x)
963 Skip i' -> Skip (ICALFirstBuf s1 s2 i' i2)
964 Stop -> Stop
966 step _ (ICALSecondInner s1 s2 i1 i2 v) = do
967 r <- istep2 i2
968 return $ case r of
969 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i' v)
970 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i' v)
971 Stop -> Skip (ICALFirstResume s1 s2 i1 i2 v)
972 -}
974 step _ (ICALSecondInner s1 s2 i1 i2) = do
975 r <- istep2 i2
976 return $ case r of
977 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i')
978 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i')
979 Stop -> Skip (ICALFirstInner s1 s2 i1)
980 -- Stop -> Skip (ICALFirstResume s1 s2 i1 i2)
982 {-
983 step _ (ICALFirstResume s1 s2 i1 i2 x) = do
984 return $ Yield x (ICALFirstInner s1 s2 i1 i2)
985 -}
987------------------------------------------------------------------------------
988-- Folding
989------------------------------------------------------------------------------
991{-# ANN type FIterState Fuse #-}
992data FIterState s f m a b
993 = FIterInit s f
994 | forall fs. FIterStream s (fs -> a -> m (FL.Step fs b)) fs (fs -> m b)
995 | FIterYield b (FIterState s f m a b)
996 | FIterStop
998{-# INLINE_NORMAL foldIterateM #-}
999foldIterateM ::
1000 Monad m => (b -> m (FL.Fold m a b)) -> m b -> Stream m a -> Stream m b
1001foldIterateM func seed0 (Stream step state) =
1002 Stream stepOuter (FIterInit state seed0)
1004 where
1006 {-# INLINE iterStep #-}
1007 iterStep from st fstep extract = do
1008 res <- from
1009 return
1010 $ Skip
1011 $ case res of
1012 FL.Partial fs -> FIterStream st fstep fs extract
1013 FL.Done fb -> FIterYield fb $ FIterInit st (return fb)
1015 {-# INLINE_LATE stepOuter #-}
1016 stepOuter _ (FIterInit st seed) = do
1017 (FL.Fold fstep initial extract) <- seed >>= func
1018 iterStep initial st fstep extract
1019 stepOuter gst (FIterStream st fstep fs extract) = do
1020 r <- step (adaptState gst) st
1021 case r of
1022 Yield x s -> do
1023 iterStep (fstep fs x) s fstep extract
1024 Skip s -> return $ Skip $ FIterStream s fstep fs extract
1025 Stop -> do
1026 b <- extract fs
1027 return $ Skip $ FIterYield b FIterStop
1028 stepOuter _ (FIterYield a next) = return $ Yield a next
1029 stepOuter _ FIterStop = return Stop
1031{-# ANN type CIterState Fuse #-}
1032data CIterState s f fs b
1033 = CIterInit s f
1034 | CIterConsume s fs
1035 | CIterYield b (CIterState s f fs b)
1036 | CIterStop
1038-- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
1039-- much more efficient due to stream fusion.
1041-- /Internal/
1042{-# INLINE_NORMAL refoldIterateM #-}
1043refoldIterateM ::
1044 Monad m => Refold m b a b -> m b -> Stream m a -> Stream m b
1045refoldIterateM (Refold fstep finject fextract) initial (Stream step state) =
1046 Stream stepOuter (CIterInit state initial)
1048 where
1050 {-# INLINE iterStep #-}
1051 iterStep st action = do
1052 res <- action
1053 return
1054 $ Skip
1055 $ case res of
1056 FL.Partial fs -> CIterConsume st fs
1057 FL.Done fb -> CIterYield fb $ CIterInit st (return fb)
1059 {-# INLINE_LATE stepOuter #-}
1060 stepOuter _ (CIterInit st action) = do
1061 iterStep st (action >>= finject)
1062 stepOuter gst (CIterConsume st fs) = do
1063 r <- step (adaptState gst) st
1064 case r of
1065 Yield x s -> iterStep s (fstep fs x)
1066 Skip s -> return $ Skip $ CIterConsume s fs
1067 Stop -> do
1068 b <- fextract fs
1069 return $ Skip $ CIterYield b CIterStop
1070 stepOuter _ (CIterYield a next) = return $ Yield a next
1071 stepOuter _ CIterStop = return Stop
1073-- "n" elements at the end are dropped by the fold.
1074{-# INLINE sliceBy #-}
1075sliceBy :: Monad m => Fold m a Int -> Int -> Refold m (Int, Int) a (Int, Int)
1076sliceBy (Fold step1 initial1 extract1) n = Refold step inject extract
1078 where
1080 inject (i, len) = do
1081 r <- initial1
1082 return $ case r of
1083 Partial s -> Partial $ Tuple' (i + len + n) s
1084 Done l -> Done (i, l)
1086 step (Tuple' i s) x = do
1087 r <- step1 s x
1088 return $ case r of
1089 Partial s1 -> Partial $ Tuple' i s1
1090 Done len -> Done (i, len)
1092 extract (Tuple' i s) = (i,) <$> extract1 s
1094{-# INLINE sliceOnSuffix #-}
1095sliceOnSuffix :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
1096sliceOnSuffix predicate =
1097 -- Scan the stream with the given refold
1098 refoldIterateM
1099 (sliceBy (FL.takeEndBy_ predicate FL.length) 1)
1100 (return (-1, 0))
1102------------------------------------------------------------------------------
1103-- Parsing
1104------------------------------------------------------------------------------
1106{-# ANN type ParseChunksState Fuse #-}
1107data ParseChunksState x inpBuf st pst =
1108 ParseChunksInit inpBuf st
1109 | ParseChunksInitLeftOver inpBuf
1110 | ParseChunksStream st inpBuf !pst
1111 | ParseChunksBuf inpBuf st inpBuf !pst
1112 | ParseChunksYield x (ParseChunksState x inpBuf st pst)
1114{-# INLINE_NORMAL parseMany #-}
1115parseMany
1116 :: MonadThrow m
1117 => PRD.Parser m a b
1118 -> Stream m a
1119 -> Stream m b
1120parseMany (PRD.Parser pstep initial extract) (Stream step state) =
1121 Stream stepOuter (ParseChunksInit [] state)
1123 where
1125 {-# INLINE_LATE stepOuter #-}
1126 -- Buffer is empty, get the first element from the stream, initialize the
1127 -- fold and then go to stream processing loop.
1128 stepOuter gst (ParseChunksInit [] st) = do
1129 r <- step (adaptState gst) st
1130 case r of
1131 Yield x s -> do
1132 res <- initial
1133 case res of
1134 PRD.IPartial ps ->
1135 return $ Skip $ ParseChunksBuf [x] s [] ps
1136 PRD.IDone pb ->
1137 let next = ParseChunksInit [x] s
1138 in return $ Skip $ ParseChunksYield pb next
1139 PRD.IError err -> throwM $ ParseError err
1140 Skip s -> return $ Skip $ ParseChunksInit [] s
1141 Stop -> return Stop
1143 -- Buffer is not empty, go to buffered processing loop
1144 stepOuter _ (ParseChunksInit src st) = do
1145 res <- initial
1146 case res of
1147 PRD.IPartial ps ->
1148 return $ Skip $ ParseChunksBuf src st [] ps
1149 PRD.IDone pb ->
1150 let next = ParseChunksInit src st
1151 in return $ Skip $ ParseChunksYield pb next
1152 PRD.IError err -> throwM $ ParseError err
1154 -- XXX we just discard any leftover input at the end
1155 stepOuter _ (ParseChunksInitLeftOver _) = return Stop
1157 -- Buffer is empty, process elements from the stream
1158 stepOuter gst (ParseChunksStream st buf pst) = do
1159 r <- step (adaptState gst) st
1160 case r of
1161 Yield x s -> do
1162 pRes <- pstep pst x
1163 case pRes of
1164 PR.Partial 0 pst1 ->
1165 return $ Skip $ ParseChunksStream s [] pst1
1166 PR.Partial n pst1 -> do
1167 assert (n <= length (x:buf)) (return ())
1168 let src0 = Prelude.take n (x:buf)
1169 src = Prelude.reverse src0
1170 return $ Skip $ ParseChunksBuf src s [] pst1
1171 PR.Continue 0 pst1 ->
1172 return $ Skip $ ParseChunksStream s (x:buf) pst1
1173 PR.Continue n pst1 -> do
1174 assert (n <= length (x:buf)) (return ())
1175 let (src0, buf1) = splitAt n (x:buf)
1176 src = Prelude.reverse src0
1177 return $ Skip $ ParseChunksBuf src s buf1 pst1
1178 PR.Done 0 b -> do
1179 return $ Skip $
1180 ParseChunksYield b (ParseChunksInit [] s)
1181 PR.Done n b -> do
1182 assert (n <= length (x:buf)) (return ())
1183 let src = Prelude.reverse (Prelude.take n (x:buf))
1184 return $ Skip $
1185 ParseChunksYield b (ParseChunksInit src s)
1186 PR.Error err -> throwM $ ParseError err
1187 Skip s -> return $ Skip $ ParseChunksStream s buf pst
1188 Stop -> do
1189 b <- extract pst
1190 let src = Prelude.reverse buf
1191 return $ Skip $
1192 ParseChunksYield b (ParseChunksInitLeftOver src)
1194 -- go back to stream processing mode
1195 stepOuter _ (ParseChunksBuf [] s buf pst) =
1196 return $ Skip $ ParseChunksStream s buf pst
1198 -- buffered processing loop
1199 stepOuter _ (ParseChunksBuf (x:xs) s buf pst) = do
1200 pRes <- pstep pst x
1201 case pRes of
1202 PR.Partial 0 pst1 ->
1203 return $ Skip $ ParseChunksBuf xs s [] pst1
1204 PR.Partial n pst1 -> do
1205 assert (n <= length (x:buf)) (return ())
1206 let src0 = Prelude.take n (x:buf)
1207 src = Prelude.reverse src0 ++ xs
1208 return $ Skip $ ParseChunksBuf src s [] pst1
1209 PR.Continue 0 pst1 ->
1210 return $ Skip $ ParseChunksBuf xs s (x:buf) pst1
1211 PR.Continue n pst1 -> do
1212 assert (n <= length (x:buf)) (return ())
1213 let (src0, buf1) = splitAt n (x:buf)
1214 src = Prelude.reverse src0 ++ xs
1215 return $ Skip $ ParseChunksBuf src s buf1 pst1
1216 PR.Done 0 b ->
1217 return $ Skip $ ParseChunksYield b (ParseChunksInit xs s)
1218 PR.Done n b -> do
1219 assert (n <= length (x:buf)) (return ())
1220 let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs
1221 return $ Skip $ ParseChunksYield b (ParseChunksInit src s)
1222 PR.Error err -> throwM $ ParseError err
1224 stepOuter _ (ParseChunksYield a next) = return $ Yield a next
1226{-# ANN type ConcatParseState Fuse #-}
1227data ConcatParseState b inpBuf st p m a =
1228 ConcatParseInit inpBuf st p
1229 | ConcatParseInitLeftOver inpBuf
1230 | forall s. ConcatParseStream st inpBuf (s -> a -> m (PRD.Step s b)) s (s -> m b)
1231 | forall s. ConcatParseBuf inpBuf st inpBuf (s -> a -> m (PRD.Step s b)) s (s -> m b)
1232 | ConcatParseYield b (ConcatParseState b inpBuf st p m a)
1234{-# INLINE_NORMAL parseIterate #-}
1235parseIterate
1236 :: MonadThrow m
1237 => (b -> PRD.Parser m a b)
1238 -> b
1239 -> Stream m a
1240 -> Stream m b
1241parseIterate func seed (Stream step state) =
1242 Stream stepOuter (ConcatParseInit [] state (func seed))
1244 where
1246 {-# INLINE_LATE stepOuter #-}
1247 -- Buffer is empty, go to stream processing loop
1248 stepOuter _ (ConcatParseInit [] st (PRD.Parser pstep initial extract)) = do
1249 res <- initial
1250 case res of
1251 PRD.IPartial ps ->
1252 return $ Skip $ ConcatParseStream st [] pstep ps extract
1253 PRD.IDone pb ->
1254 let next = ConcatParseInit [] st (func pb)
1255 in return $ Skip $ ConcatParseYield pb next
1256 PRD.IError err -> throwM $ ParseError err
1258 -- Buffer is not empty, go to buffered processing loop
1259 stepOuter _ (ConcatParseInit src st
1260 (PRD.Parser pstep initial extract)) = do
1261 res <- initial
1262 case res of
1263 PRD.IPartial ps ->
1264 return $ Skip $ ConcatParseBuf src st [] pstep ps extract
1265 PRD.IDone pb ->
1266 let next = ConcatParseInit src st (func pb)
1267 in return $ Skip $ ConcatParseYield pb next
1268 PRD.IError err -> throwM $ ParseError err
1270 -- XXX we just discard any leftover input at the end
1271 stepOuter _ (ConcatParseInitLeftOver _) = return Stop
1273 -- Buffer is empty process elements from the stream
1274 stepOuter gst (ConcatParseStream st buf pstep pst extract) = do
1275 r <- step (adaptState gst) st
1276 case r of
1277 Yield x s -> do
1278 pRes <- pstep pst x
1279 case pRes of
1280 PR.Partial 0 pst1 ->
1281 return $ Skip $ ConcatParseStream s [] pstep pst1 extract
1282 PR.Partial n pst1 -> do
1283 assert (n <= length (x:buf)) (return ())
1284 let src0 = Prelude.take n (x:buf)
1285 src = Prelude.reverse src0
1286 return $ Skip $ ConcatParseBuf src s [] pstep pst1 extract
1287 -- PR.Continue 0 pst1 ->
1288 -- return $ Skip $ ConcatParseStream s (x:buf) pst1
1289 PR.Continue n pst1 -> do
1290 assert (n <= length (x:buf)) (return ())
1291 let (src0, buf1) = splitAt n (x:buf)
1292 src = Prelude.reverse src0
1293 return $ Skip $ ConcatParseBuf src s buf1 pstep pst1 extract
1294 -- XXX Specialize for Stop 0 common case?
1295 PR.Done n b -> do
1296 assert (n <= length (x:buf)) (return ())
1297 let src = Prelude.reverse (Prelude.take n (x:buf))
1298 return $ Skip $
1299 ConcatParseYield b (ConcatParseInit src s (func b))
1300 PR.Error err -> throwM $ ParseError err
1301 Skip s -> return $ Skip $ ConcatParseStream s buf pstep pst extract
1302 Stop -> do
1303 b <- extract pst
1304 let src = Prelude.reverse buf
1305 return $ Skip $ ConcatParseYield b (ConcatParseInitLeftOver src)
1307 -- go back to stream processing mode
1308 stepOuter _ (ConcatParseBuf [] s buf pstep ps extract) =
1309 return $ Skip $ ConcatParseStream s buf pstep ps extract
1311 -- buffered processing loop
1312 stepOuter _ (ConcatParseBuf (x:xs) s buf pstep pst extract) = do
1313 pRes <- pstep pst x
1314 case pRes of
1315 PR.Partial 0 pst1 ->
1316 return $ Skip $ ConcatParseBuf xs s [] pstep pst1 extract
1317 PR.Partial n pst1 -> do
1318 assert (n <= length (x:buf)) (return ())
1319 let src0 = Prelude.take n (x:buf)
1320 src = Prelude.reverse src0 ++ xs
1321 return $ Skip $ ConcatParseBuf src s [] pstep pst1 extract
1322 -- PR.Continue 0 pst1 -> return $ Skip $ ConcatParseBuf xs s (x:buf) pst1
1323 PR.Continue n pst1 -> do
1324 assert (n <= length (x:buf)) (return ())
1325 let (src0, buf1) = splitAt n (x:buf)
1326 src = Prelude.reverse src0 ++ xs
1327 return $ Skip $ ConcatParseBuf src s buf1 pstep pst1 extract
1328 -- XXX Specialize for Stop 0 common case?
1329 PR.Done n b -> do
1330 assert (n <= length (x:buf)) (return ())
1331 let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs
1332 return $ Skip $ ConcatParseYield b
1333 (ConcatParseInit src s (func b))
1334 PR.Error err -> throwM $ ParseError err
1336 stepOuter _ (ConcatParseYield a next) = return $ Yield a next
1338------------------------------------------------------------------------------
1339-- Grouping
1340------------------------------------------------------------------------------
1342data GroupByState st fs a b
1343 = GroupingInit st
1344 | GroupingDo st !fs
1345 | GroupingInitWith st !a
1346 | GroupingDoWith st !fs !a
1347 | GroupingYield !b (GroupByState st fs a b)
1348 | GroupingDone
1350{-# INLINE_NORMAL groupsBy #-}
1351groupsBy :: Monad m
1352 => (a -> a -> Bool)
1353 -> Fold m a b
1354 -> Stream m a
1355 -> Stream m b
1357groupsBy eq fld = parseMany (PRD.groupBy eq fld)
1359groupsBy cmp (Fold fstep initial done) (Stream step state) =
1360 Stream stepOuter (GroupingInit state)
1362 where
1364 {-# INLINE_LATE stepOuter #-}
1365 stepOuter _ (GroupingInit st) = do
1366 -- XXX Note that if the stream stops without yielding a single element
1367 -- in the group we discard the "initial" effect.
1368 res <- initial
1369 return
1370 $ case res of
1371 FL.Partial s -> Skip $ GroupingDo st s
1372 FL.Done b -> Yield b $ GroupingInit st
1373 stepOuter gst (GroupingDo st fs) = do
1374 res <- step (adaptState gst) st
1375 case res of
1376 Yield x s -> do
1377 r <- fstep fs x
1378 case r of
1379 FL.Partial fs1 -> go SPEC x s fs1
1380 FL.Done b -> return $ Yield b (GroupingInit s)
1381 Skip s -> return $ Skip $ GroupingDo s fs
1382 Stop -> return Stop
1384 where
1386 go !_ prev stt !acc = do
1387 res <- step (adaptState gst) stt
1388 case res of
1389 Yield x s -> do
1390 if cmp x prev
1391 then do
1392 r <- fstep acc x
1393 case r of
1394 FL.Partial fs1 -> go SPEC prev s fs1
1395 FL.Done b -> return $ Yield b (GroupingInit s)
1396 else do
1397 r <- done acc
1398 return $ Yield r (GroupingInitWith s x)
1399 Skip s -> go SPEC prev s acc
1400 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1401 stepOuter _ (GroupingInitWith st x) = do
1402 res <- initial
1403 return
1404 $ case res of
1405 FL.Partial s -> Skip $ GroupingDoWith st s x
1406 FL.Done b -> Yield b $ GroupingInitWith st x
1407 stepOuter gst (GroupingDoWith st fs prev) = do
1408 res <- fstep fs prev
1409 case res of
1410 FL.Partial fs1 -> go SPEC st fs1
1411 FL.Done b -> return $ Yield b (GroupingInit st)
1413 where
1415 -- XXX code duplicated from the previous equation
1416 go !_ stt !acc = do
1417 res <- step (adaptState gst) stt
1418 case res of
1419 Yield x s -> do
1420 if cmp x prev
1421 then do
1422 r <- fstep acc x
1423 case r of
1424 FL.Partial fs1 -> go SPEC s fs1
1425 FL.Done b -> return $ Yield b (GroupingInit s)
1426 else do
1427 r <- done acc
1428 return $ Yield r (GroupingInitWith s x)
1429 Skip s -> go SPEC s acc
1430 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1431 stepOuter _ (GroupingYield _ _) = error "groupsBy: Unreachable"
1432 stepOuter _ GroupingDone = return Stop
1434{-# INLINE_NORMAL groupsRollingBy #-}
1435groupsRollingBy :: Monad m
1436 => (a -> a -> Bool)
1437 -> Fold m a b
1438 -> Stream m a
1439 -> Stream m b
1441groupsRollingBy eq fld = parseMany (PRD.groupByRolling eq fld)
1443groupsRollingBy cmp (Fold fstep initial done) (Stream step state) =
1444 Stream stepOuter (GroupingInit state)
1446 where
1448 {-# INLINE_LATE stepOuter #-}
1449 stepOuter _ (GroupingInit st) = do
1450 -- XXX Note that if the stream stops without yielding a single element
1451 -- in the group we discard the "initial" effect.
1452 res <- initial
1453 return
1454 $ case res of
1455 FL.Partial fs -> Skip $ GroupingDo st fs
1456 FL.Done fb -> Yield fb $ GroupingInit st
1457 stepOuter gst (GroupingDo st fs) = do
1458 res <- step (adaptState gst) st
1459 case res of
1460 Yield x s -> do
1461 r <- fstep fs x
1462 case r of
1463 FL.Partial fs1 -> go SPEC x s fs1
1464 FL.Done fb -> return $ Yield fb (GroupingInit s)
1465 Skip s -> return $ Skip $ GroupingDo s fs
1466 Stop -> return Stop
1468 where
1470 go !_ prev stt !acc = do
1471 res <- step (adaptState gst) stt
1472 case res of
1473 Yield x s -> do
1474 if cmp prev x
1475 then do
1476 r <- fstep acc x
1477 case r of
1478 FL.Partial fs1 -> go SPEC x s fs1
1479 FL.Done b -> return $ Yield b (GroupingInit s)
1480 else do
1481 r <- done acc
1482 return $ Yield r (GroupingInitWith s x)
1483 Skip s -> go SPEC prev s acc
1484 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1485 stepOuter _ (GroupingInitWith st x) = do
1486 res <- initial
1487 return
1488 $ case res of
1489 FL.Partial s -> Skip $ GroupingDoWith st s x
1490 FL.Done b -> Yield b $ GroupingInitWith st x
1491 stepOuter gst (GroupingDoWith st fs previous) = do
1492 res <- fstep fs previous
1493 case res of
1494 FL.Partial s -> go SPEC previous st s
1495 FL.Done b -> return $ Yield b (GroupingInit st)
1497 where
1499 -- XXX GHC: groupsBy has one less parameter in this go loop and it
1500 -- fuses. However, groupsRollingBy does not fuse, removing the prev
1501 -- parameter makes it fuse. Something needs to be fixed in GHC. The
1502 -- workaround for this is noted in the comments below.
1503 go !_ prev !stt !acc = do
1504 res <- step (adaptState gst) stt
1505 case res of
1506 Yield x s -> do
1507 if cmp prev x
1508 then do
1509 r <- fstep acc x
1510 case r of
1511 FL.Partial fs1 -> go SPEC x s fs1
1512 FL.Done b -> return $ Yield b (GroupingInit st)
1513 else do
1514 {-
1515 r <- done acc
1516 return $ Yield r (GroupingInitWith s x)
1517 -}
1518 -- The code above does not let groupBy fuse. We use the
1519 -- alternative code below instead. Instead of jumping
1520 -- to GroupingInitWith state, we unroll the code of
1521 -- GroupingInitWith state here to help GHC with stream
1522 -- fusion.
1523 result <- initial
1524 r <- done acc
1525 return
1526 $ Yield r
1527 $ case result of
1528 FL.Partial fsi -> GroupingDoWith s fsi x
1529 FL.Done b -> GroupingYield b (GroupingInit s)
1530 Skip s -> go SPEC prev s acc
1531 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1532 stepOuter _ (GroupingYield r next) = return $ Yield r next
1533 stepOuter _ GroupingDone = return Stop
1535------------------------------------------------------------------------------
1536-- Splitting - by a predicate
1537------------------------------------------------------------------------------
1539data WordsByState st fs b
1540 = WordsByInit st
1541 | WordsByDo st !fs
1542 | WordsByDone
1543 | WordsByYield !b (WordsByState st fs b)
1545{-# INLINE_NORMAL wordsBy #-}
1546wordsBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1547wordsBy predicate (Fold fstep initial done) (Stream step state) =
1548 Stream stepOuter (WordsByInit state)
1550 where
1552 {-# INLINE_LATE stepOuter #-}
1553 stepOuter _ (WordsByInit st) = do
1554 res <- initial
1555 return
1556 $ case res of
1557 FL.Partial s -> Skip $ WordsByDo st s
1558 FL.Done b -> Yield b (WordsByInit st)
1560 stepOuter gst (WordsByDo st fs) = do
1561 res <- step (adaptState gst) st
1562 case res of
1563 Yield x s -> do
1564 if predicate x
1565 then do
1566 resi <- initial
1567 return
1568 $ case resi of
1569 FL.Partial fs1 -> Skip $ WordsByDo s fs1
1570 FL.Done b -> Yield b (WordsByInit s)
1571 else do
1572 r <- fstep fs x
1573 case r of
1574 FL.Partial fs1 -> go SPEC s fs1
1575 FL.Done b -> return $ Yield b (WordsByInit s)
1576 Skip s -> return $ Skip $ WordsByDo s fs
1577 Stop -> return Stop
1579 where
1581 go !_ stt !acc = do
1582 res <- step (adaptState gst) stt
1583 case res of
1584 Yield x s -> do
1585 if predicate x
1586 then do
1587 {-
1588 r <- done acc
1589 return $ Yield r (WordsByInit s)
1590 -}
1591 -- The above code does not fuse well. Need to check why
1592 -- GHC is not able to simplify it well. Using the code
1593 -- below, instead of jumping through the WordsByInit
1594 -- state always, we directly go to WordsByDo state in
1595 -- the common case of Partial.
1596 resi <- initial
1597 r <- done acc
1598 return
1599 $ Yield r
1600 $ case resi of
1601 FL.Partial fs1 -> WordsByDo s fs1
1602 FL.Done b -> WordsByYield b (WordsByInit s)
1603 else do
1604 r <- fstep acc x
1605 case r of
1606 FL.Partial fs1 -> go SPEC s fs1
1607 FL.Done b -> return $ Yield b (WordsByInit s)
1608 Skip s -> go SPEC s acc
1609 Stop -> done acc >>= \r -> return $ Yield r WordsByDone
1611 stepOuter _ WordsByDone = return Stop
1613 stepOuter _ (WordsByYield b next) = return $ Yield b next
1615------------------------------------------------------------------------------
1616-- Splitting on a sequence
1617------------------------------------------------------------------------------
1619-- String search algorithms:
1620-- http://www-igm.univ-mlv.fr/~lecroq/string/index.html
1623-- TODO can we unify the splitting operations using a splitting configuration
1624-- like in the split package.
1626data SplitStyle = Infix | Suffix | Prefix deriving (Eq, Show)
1627data SplitOptions = SplitOptions
1628 { style :: SplitStyle
1629 , withSep :: Bool -- ^ keep the separators in output
1630 -- , compact :: Bool -- ^ treat multiple consecutive separators as one
1631 -- , trimHead :: Bool -- ^ drop blank at head
1632 -- , trimTail :: Bool -- ^ drop blank at tail
1636-- XXX using "fs" as the last arg in Constructors may simplify the code a bit,
1637-- because we can use the constructor directly without having to create "jump"
1638-- functions.
1639{-# ANN type SplitOnSeqState Fuse #-}
1640data SplitOnSeqState rb rh ck w fs s b x =
1641 SplitOnSeqInit
1642 | SplitOnSeqYield b (SplitOnSeqState rb rh ck w fs s b x)
1643 | SplitOnSeqDone
1645 | SplitOnSeqEmpty !fs s
1647 | SplitOnSeqSingle !fs s x
1649 | SplitOnSeqWordInit !fs s
1650 | SplitOnSeqWordLoop !w s !fs
1651 | SplitOnSeqWordDone Int !fs !w
1653 | SplitOnSeqKRInit Int !fs s rb !rh
1654 | SplitOnSeqKRLoop fs s rb !rh !ck
1655 | SplitOnSeqKRCheck fs s rb !rh
1656 | SplitOnSeqKRDone Int !fs rb !rh
1658 | SplitOnSeqReinit (fs -> SplitOnSeqState rb rh ck w fs s b x)
1660{-# INLINE_NORMAL splitOnSeq #-}
1661splitOnSeq
1662 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1663 => Array a
1664 -> Fold m a b
1665 -> Stream m a
1666 -> Stream m b
1667splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
1668 Stream stepOuter SplitOnSeqInit
1670 where
1672 patLen = A.length patArr
1673 maxIndex = patLen - 1
1674 elemBits = sizeOf (undefined :: a) * 8
1676 -- For word pattern case
1677 wordMask :: Word
1678 wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1680 elemMask :: Word
1681 elemMask = (1 `shiftL` elemBits) - 1
1683 wordPat :: Word
1684 wordPat = wordMask .&. A.foldl' addToWord 0 patArr
1686 addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1688 -- For Rabin-Karp search
1689 k = 2891336453 :: Word32
1690 coeff = k ^ patLen
1692 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1694 deltaCksum cksum old new =
1695 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1697 -- XXX shall we use a random starting hash or 1 instead of 0?
1698 patHash = A.foldl' addCksum 0 patArr
1700 skip = return . Skip
1702 nextAfterInit nextGen stepRes =
1703 case stepRes of
1704 FL.Partial s -> nextGen s
1705 FL.Done b -> SplitOnSeqYield b (SplitOnSeqReinit nextGen)
1707 {-# INLINE yieldProceed #-}
1708 yieldProceed nextGen fs =
1709 initial >>= skip . SplitOnSeqYield fs . nextAfterInit nextGen
1711 {-# INLINE_LATE stepOuter #-}
1712 stepOuter _ SplitOnSeqInit = do
1713 res <- initial
1714 case res of
1715 FL.Partial acc ->
1716 if patLen == 0
1717 then return $ Skip $ SplitOnSeqEmpty acc state
1718 else if patLen == 1
1719 then do
1720 pat <- liftIO $ A.unsafeIndexIO patArr 0
1721 return $ Skip $ SplitOnSeqSingle acc state pat
1722 else if sizeOf (undefined :: a) * patLen
1723 <= sizeOf (undefined :: Word)
1724 then return $ Skip $ SplitOnSeqWordInit acc state
1725 else do
1726 (rb, rhead) <- liftIO $ RB.new patLen
1727 skip $ SplitOnSeqKRInit 0 acc state rb rhead
1728 FL.Done b -> skip $ SplitOnSeqYield b SplitOnSeqInit
1730 stepOuter _ (SplitOnSeqYield x next) = return $ Yield x next
1732 ---------------------------
1733 -- Checkpoint
1734 ---------------------------
1736 stepOuter _ (SplitOnSeqReinit nextGen) =
1737 initial >>= skip . nextAfterInit nextGen
1739 ---------------------------
1740 -- Empty pattern
1741 ---------------------------
1743 stepOuter gst (SplitOnSeqEmpty acc st) = do
1744 res <- step (adaptState gst) st
1745 case res of
1746 Yield x s -> do
1747 r <- fstep acc x
1748 b1 <-
1749 case r of
1750 FL.Partial acc1 -> done acc1
1751 FL.Done b -> return b
1752 let jump c = SplitOnSeqEmpty c s
1753 in yieldProceed jump b1
1754 Skip s -> skip (SplitOnSeqEmpty acc s)
1755 Stop -> return Stop
1757 -----------------
1758 -- Done
1759 -----------------
1761 stepOuter _ SplitOnSeqDone = return Stop
1763 -----------------
1764 -- Single Pattern
1765 -----------------
1767 stepOuter gst (SplitOnSeqSingle fs st pat) = do
1768 res <- step (adaptState gst) st
1769 case res of
1770 Yield x s -> do
1771 let jump c = SplitOnSeqSingle c s pat
1772 if pat == x
1773 then done fs >>= yieldProceed jump
1774 else do
1775 r <- fstep fs x
1776 case r of
1777 FL.Partial fs1 -> skip $ jump fs1
1778 FL.Done b -> yieldProceed jump b
1779 Skip s -> return $ Skip $ SplitOnSeqSingle fs s pat
1780 Stop -> do
1781 r <- done fs
1782 return $ Skip $ SplitOnSeqYield r SplitOnSeqDone
1784 ---------------------------
1785 -- Short Pattern - Shift Or
1786 ---------------------------
1788 stepOuter _ (SplitOnSeqWordDone 0 fs _) = do
1789 r <- done fs
1790 skip $ SplitOnSeqYield r SplitOnSeqDone
1791 stepOuter _ (SplitOnSeqWordDone n fs wrd) = do
1792 let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
1793 r <- fstep fs (toEnum $ fromIntegral old)
1794 case r of
1795 FL.Partial fs1 -> skip $ SplitOnSeqWordDone (n - 1) fs1 wrd
1796 FL.Done b -> do
1797 let jump c = SplitOnSeqWordDone (n - 1) c wrd
1798 yieldProceed jump b
1800 stepOuter gst (SplitOnSeqWordInit fs st0) =
1801 go SPEC 0 0 st0
1803 where
1805 {-# INLINE go #-}
1806 go !_ !idx !wrd !st = do
1807 res <- step (adaptState gst) st
1808 case res of
1809 Yield x s -> do
1810 let wrd1 = addToWord wrd x
1811 if idx == maxIndex
1812 then do
1813 if wrd1 .&. wordMask == wordPat
1814 then do
1815 let jump c = SplitOnSeqWordInit c s
1816 done fs >>= yieldProceed jump
1817 else skip $ SplitOnSeqWordLoop wrd1 s fs
1818 else go SPEC (idx + 1) wrd1 s
1819 Skip s -> go SPEC idx wrd s
1820 Stop -> do
1821 if idx /= 0
1822 then skip $ SplitOnSeqWordDone idx fs wrd
1823 else do
1824 r <- done fs
1825 skip $ SplitOnSeqYield r SplitOnSeqDone
1827 stepOuter gst (SplitOnSeqWordLoop wrd0 st0 fs0) =
1828 go SPEC wrd0 st0 fs0
1830 where
1832 {-# INLINE go #-}
1833 go !_ !wrd !st !fs = do
1834 res <- step (adaptState gst) st
1835 case res of
1836 Yield x s -> do
1837 let jump c = SplitOnSeqWordInit c s
1838 wrd1 = addToWord wrd x
1839 old = (wordMask .&. wrd)
1840 `shiftR` (elemBits * (patLen - 1))
1841 r <- fstep fs (toEnum $ fromIntegral old)
1842 case r of
1843 FL.Partial fs1 -> do
1844 if wrd1 .&. wordMask == wordPat
1845 then done fs1 >>= yieldProceed jump
1846 else go SPEC wrd1 s fs1
1847 FL.Done b -> yieldProceed jump b
1848 Skip s -> go SPEC wrd s fs
1849 Stop -> skip $ SplitOnSeqWordDone patLen fs wrd
1851 -------------------------------
1852 -- General Pattern - Karp Rabin
1853 -------------------------------
1855 stepOuter gst (SplitOnSeqKRInit idx fs st rb rh) = do
1856 res <- step (adaptState gst) st
1857 case res of
1858 Yield x s -> do
1859 rh1 <- liftIO $ RB.unsafeInsert rb rh x
1860 if idx == maxIndex
1861 then do
1862 let fld = RB.unsafeFoldRing (RB.ringBound rb)
1863 let !ringHash = fld addCksum 0 rb
1864 if ringHash == patHash
1865 then skip $ SplitOnSeqKRCheck fs s rb rh1
1866 else skip $ SplitOnSeqKRLoop fs s rb rh1 ringHash
1867 else skip $ SplitOnSeqKRInit (idx + 1) fs s rb rh1
1868 Skip s -> skip $ SplitOnSeqKRInit idx fs s rb rh
1869 Stop -> do
1870 skip $ SplitOnSeqKRDone idx fs rb (RB.startOf rb)
1872 -- XXX The recursive "go" is more efficient than the state based recursion
1873 -- code commented out below. Perhaps its more efficient because of
1874 -- factoring out "rb" outside the loop.
1875 --
1876 stepOuter gst (SplitOnSeqKRLoop fs0 st0 rb rh0 cksum0) =
1877 go SPEC fs0 st0 rh0 cksum0
1879 where
1881 go !_ !fs !st !rh !cksum = do
1882 res <- step (adaptState gst) st
1883 case res of
1884 Yield x s -> do
1885 old <- liftIO $ peek rh
1886 let cksum1 = deltaCksum cksum old x
1887 r <- fstep fs old
1888 case r of
1889 FL.Partial fs1 -> do
1890 rh1 <- liftIO (RB.unsafeInsert rb rh x)
1891 if cksum1 == patHash
1892 then skip $ SplitOnSeqKRCheck fs1 s rb rh1
1893 else go SPEC fs1 s rh1 cksum1
1894 FL.Done b -> do
1895 let rst = RB.startOf rb
1896 jump c = SplitOnSeqKRInit 0 c s rb rst
1897 yieldProceed jump b
1898 Skip s -> go SPEC fs s rh cksum
1899 Stop -> skip $ SplitOnSeqKRDone patLen fs rb rh
1901 -- XXX The following code is 5 times slower compared to the recursive loop
1902 -- based code above. Need to investigate why. One possibility is that the
1903 -- go loop above does not thread around the ring buffer (rb). This code may
1904 -- be causing the state to bloat and getting allocated on each iteration.
1905 -- We can check the cmm/asm code to confirm. If so a good GHC solution to
1906 -- such problem is needed. One way to avoid this could be to use unboxed
1907 -- mutable state?
1908 {-
1909 stepOuter gst (SplitOnSeqKRLoop fs st rb rh cksum) = do
1910 res <- step (adaptState gst) st
1911 case res of
1912 Yield x s -> do
1913 old <- liftIO $ peek rh
1914 let cksum1 = deltaCksum cksum old x
1915 fs1 <- fstep fs old
1916 if (cksum1 == patHash)
1917 then do
1918 r <- done fs1
1919 skip $ SplitOnSeqYield r $ SplitOnSeqKRInit 0 s rb rh
1920 else do
1921 rh1 <- liftIO (RB.unsafeInsert rb rh x)
1922 skip $ SplitOnSeqKRLoop fs1 s rb rh1 cksum1
1923 Skip s -> skip $ SplitOnSeqKRLoop fs s rb rh cksum
1924 Stop -> skip $ SplitOnSeqKRDone patLen fs rb rh
1925 -}
1927 stepOuter _ (SplitOnSeqKRCheck fs st rb rh) = do
1928 if RB.unsafeEqArray rb rh patArr
1929 then do
1930 r <- done fs
1931 let rst = RB.startOf rb
1932 jump c = SplitOnSeqKRInit 0 c st rb rst
1933 yieldProceed jump r
1934 else skip $ SplitOnSeqKRLoop fs st rb rh patHash
1936 stepOuter _ (SplitOnSeqKRDone 0 fs _ _) = do
1937 r <- done fs
1938 skip $ SplitOnSeqYield r SplitOnSeqDone
1939 stepOuter _ (SplitOnSeqKRDone n fs rb rh) = do
1940 old <- liftIO $ peek rh
1941 let rh1 = RB.advance rb rh
1942 r <- fstep fs old
1943 case r of
1944 FL.Partial fs1 -> skip $ SplitOnSeqKRDone (n - 1) fs1 rb rh1
1945 FL.Done b -> do
1946 let jump c = SplitOnSeqKRDone (n - 1) c rb rh1
1947 yieldProceed jump b
1949{-# ANN type SplitOnSuffixSeqState Fuse #-}
1950data SplitOnSuffixSeqState rb rh ck w fs s b x =
1951 SplitOnSuffixSeqInit
1952 | SplitOnSuffixSeqYield b (SplitOnSuffixSeqState rb rh ck w fs s b x)
1953 | SplitOnSuffixSeqDone
1955 | SplitOnSuffixSeqEmpty !fs s
1957 | SplitOnSuffixSeqSingleInit !fs s x
1958 | SplitOnSuffixSeqSingle !fs s x
1960 | SplitOnSuffixSeqWordInit !fs s
1961 | SplitOnSuffixSeqWordLoop !w s !fs
1962 | SplitOnSuffixSeqWordDone Int !fs !w
1964 | SplitOnSuffixSeqKRInit Int !fs s rb !rh
1965 | SplitOnSuffixSeqKRInit1 !fs s rb !rh
1966 | SplitOnSuffixSeqKRLoop fs s rb !rh !ck
1967 | SplitOnSuffixSeqKRCheck fs s rb !rh
1968 | SplitOnSuffixSeqKRDone Int !fs rb !rh
1970 | SplitOnSuffixSeqReinit
1971 (fs -> SplitOnSuffixSeqState rb rh ck w fs s b x)
1973{-# INLINE_NORMAL splitOnSuffixSeq #-}
1974splitOnSuffixSeq
1975 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1976 => Bool
1977 -> Array a
1978 -> Fold m a b
1979 -> Stream m a
1980 -> Stream m b
1981splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
1982 Stream stepOuter SplitOnSuffixSeqInit
1984 where
1986 patLen = A.length patArr
1987 maxIndex = patLen - 1
1988 elemBits = sizeOf (undefined :: a) * 8
1990 -- For word pattern case
1991 wordMask :: Word
1992 wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1994 elemMask :: Word
1995 elemMask = (1 `shiftL` elemBits) - 1
1997 wordPat :: Word
1998 wordPat = wordMask .&. A.foldl' addToWord 0 patArr
2000 addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
2002 nextAfterInit nextGen stepRes =
2003 case stepRes of
2004 FL.Partial s -> nextGen s
2005 FL.Done b ->
2006 SplitOnSuffixSeqYield b (SplitOnSuffixSeqReinit nextGen)
2008 {-# INLINE yieldProceed #-}
2009 yieldProceed nextGen fs =
2010 initial >>= skip . SplitOnSuffixSeqYield fs . nextAfterInit nextGen
2012 -- For single element pattern case
2013 {-# INLINE processYieldSingle #-}
2014 processYieldSingle pat x s fs = do
2015 let jump c = SplitOnSuffixSeqSingleInit c s pat
2016 if pat == x
2017 then do
2018 r <- if withSep then fstep fs x else return $ FL.Partial fs
2019 b1 <-
2020 case r of
2021 FL.Partial fs1 -> done fs1
2022 FL.Done b -> return b
2023 yieldProceed jump b1
2024 else do
2025 r <- fstep fs x
2026 case r of
2027 FL.Partial fs1 -> skip $ SplitOnSuffixSeqSingle fs1 s pat
2028 FL.Done b -> yieldProceed jump b
2030 -- For Rabin-Karp search
2031 k = 2891336453 :: Word32
2032 coeff = k ^ patLen
2034 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
2036 deltaCksum cksum old new =
2037 addCksum cksum new - coeff * fromIntegral (fromEnum old)
2039 -- XXX shall we use a random starting hash or 1 instead of 0?
2040 patHash = A.foldl' addCksum 0 patArr
2042 skip = return . Skip
2044 {-# INLINE_LATE stepOuter #-}
2045 stepOuter _ SplitOnSuffixSeqInit = do
2046 res <- initial
2047 case res of
2048 FL.Partial fs ->
2049 if patLen == 0
2050 then skip $ SplitOnSuffixSeqEmpty fs state
2051 else if patLen == 1
2052 then do
2053 pat <- liftIO $ A.unsafeIndexIO patArr 0
2054 skip $ SplitOnSuffixSeqSingleInit fs state pat
2055 else if sizeOf (undefined :: a) * patLen
2056 <= sizeOf (undefined :: Word)
2057 then skip $ SplitOnSuffixSeqWordInit fs state
2058 else do
2059 (rb, rhead) <- liftIO $ RB.new patLen
2060 skip $ SplitOnSuffixSeqKRInit 0 fs state rb rhead
2061 FL.Done fb -> skip $ SplitOnSuffixSeqYield fb SplitOnSuffixSeqInit
2063 stepOuter _ (SplitOnSuffixSeqYield x next) = return $ Yield x next
2065 ---------------------------
2066 -- Reinit
2067 ---------------------------
2069 stepOuter _ (SplitOnSuffixSeqReinit nextGen) =
2070 initial >>= skip . nextAfterInit nextGen
2072 ---------------------------
2073 -- Empty pattern
2074 ---------------------------
2076 stepOuter gst (SplitOnSuffixSeqEmpty acc st) = do
2077 res <- step (adaptState gst) st
2078 case res of
2079 Yield x s -> do
2080 let jump c = SplitOnSuffixSeqEmpty c s
2081 r <- fstep acc x
2082 b1 <-
2083 case r of
2084 FL.Partial fs -> done fs
2085 FL.Done b -> return b
2086 yieldProceed jump b1
2087 Skip s -> skip (SplitOnSuffixSeqEmpty acc s)
2088 Stop -> return Stop
2090 -----------------
2091 -- Done
2092 -----------------
2094 stepOuter _ SplitOnSuffixSeqDone = return Stop
2096 -----------------
2097 -- Single Pattern
2098 -----------------
2100 stepOuter gst (SplitOnSuffixSeqSingleInit fs st pat) = do
2101 res <- step (adaptState gst) st
2102 case res of
2103 Yield x s -> processYieldSingle pat x s fs
2104 Skip s -> skip $ SplitOnSuffixSeqSingleInit fs s pat
2105 Stop -> return Stop
2107 stepOuter gst (SplitOnSuffixSeqSingle fs st pat) = do
2108 res <- step (adaptState gst) st
2109 case res of
2110 Yield x s -> processYieldSingle pat x s fs
2111 Skip s -> skip $ SplitOnSuffixSeqSingle fs s pat
2112 Stop -> do
2113 r <- done fs
2114 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2116 ---------------------------
2117 -- Short Pattern - Shift Or
2118 ---------------------------
2120 stepOuter _ (SplitOnSuffixSeqWordDone 0 fs _) = do
2121 r <- done fs
2122 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2123 stepOuter _ (SplitOnSuffixSeqWordDone n fs wrd) = do
2124 let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
2125 r <- fstep fs (toEnum $ fromIntegral old)
2126 case r of
2127 FL.Partial fs1 -> skip $ SplitOnSuffixSeqWordDone (n - 1) fs1 wrd
2128 FL.Done b -> do
2129 let jump c = SplitOnSuffixSeqWordDone (n - 1) c wrd
2130 yieldProceed jump b
2132 stepOuter gst (SplitOnSuffixSeqWordInit fs0 st0) = do
2133 res <- step (adaptState gst) st0
2134 case res of
2135 Yield x s -> do
2136 let wrd = addToWord 0 x
2137 r <- if withSep then fstep fs0 x else return $ FL.Partial fs0
2138 case r of
2139 FL.Partial fs1 -> go SPEC 1 wrd s fs1
2140 FL.Done b -> do
2141 let jump c = SplitOnSuffixSeqWordInit c s
2142 yieldProceed jump b
2143 Skip s -> skip (SplitOnSuffixSeqWordInit fs0 s)
2144 Stop -> return Stop
2146 where
2148 {-# INLINE go #-}
2149 go !_ !idx !wrd !st !fs = do
2150 res <- step (adaptState gst) st
2151 case res of
2152 Yield x s -> do
2153 let jump c = SplitOnSuffixSeqWordInit c s
2154 let wrd1 = addToWord wrd x
2155 r <- if withSep then fstep fs x else return $ FL.Partial fs
2156 case r of
2157 FL.Partial fs1 ->
2158 if idx /= maxIndex
2159 then go SPEC (idx + 1) wrd1 s fs1
2160 else if wrd1 .&. wordMask /= wordPat
2161 then skip $ SplitOnSuffixSeqWordLoop wrd1 s fs1
2162 else do done fs >>= yieldProceed jump
2163 FL.Done b -> yieldProceed jump b
2164 Skip s -> go SPEC idx wrd s fs
2165 Stop -> skip $ SplitOnSuffixSeqWordDone idx fs wrd
2167 stepOuter gst (SplitOnSuffixSeqWordLoop wrd0 st0 fs0) =
2168 go SPEC wrd0 st0 fs0
2170 where
2172 {-# INLINE go #-}
2173 go !_ !wrd !st !fs = do
2174 res <- step (adaptState gst) st
2175 case res of
2176 Yield x s -> do
2177 let jump c = SplitOnSuffixSeqWordInit c s
2178 wrd1 = addToWord wrd x
2179 old = (wordMask .&. wrd)
2180 `shiftR` (elemBits * (patLen - 1))
2181 r <-
2182 if withSep
2183 then fstep fs x
2184 else fstep fs (toEnum $ fromIntegral old)
2185 case r of
2186 FL.Partial fs1 ->
2187 if wrd1 .&. wordMask == wordPat
2188 then done fs1 >>= yieldProceed jump
2189 else go SPEC wrd1 s fs1
2190 FL.Done b -> yieldProceed jump b
2191 Skip s -> go SPEC wrd s fs
2192 Stop ->
2193 if wrd .&. wordMask == wordPat
2194 then return Stop
2195 else if withSep
2196 then do
2197 r <- done fs
2198 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2199 else skip $ SplitOnSuffixSeqWordDone patLen fs wrd
2201 -------------------------------
2202 -- General Pattern - Karp Rabin
2203 -------------------------------
2205 stepOuter gst (SplitOnSuffixSeqKRInit idx0 fs st0 rb rh0) = do
2206 res <- step (adaptState gst) st0
2207 case res of
2208 Yield x s -> do
2209 rh1 <- liftIO $ RB.unsafeInsert rb rh0 x
2210 r <- if withSep then fstep fs x else return $ FL.Partial fs
2211 case r of
2212 FL.Partial fs1 ->
2213 skip $ SplitOnSuffixSeqKRInit1 fs1 s rb rh1
2214 FL.Done b -> do
2215 let rst = RB.startOf rb
2216 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2217 yieldProceed jump b
2218 Skip s -> skip $ SplitOnSuffixSeqKRInit idx0 fs s rb rh0
2219 Stop -> return Stop
2221 stepOuter gst (SplitOnSuffixSeqKRInit1 fs0 st0 rb rh0) = do
2222 go SPEC 1 rh0 st0 fs0
2224 where
2226 go !_ !idx !rh st !fs = do
2227 res <- step (adaptState gst) st
2228 case res of
2229 Yield x s -> do
2230 rh1 <- liftIO (RB.unsafeInsert rb rh x)
2231 r <- if withSep then fstep fs x else return $ FL.Partial fs
2232 case r of
2233 FL.Partial fs1 ->
2234 if idx /= maxIndex
2235 then go SPEC (idx + 1) rh1 s fs1
2236 else skip $
2237 let fld = RB.unsafeFoldRing (RB.ringBound rb)
2238 !ringHash = fld addCksum 0 rb
2239 in if ringHash == patHash
2240 then SplitOnSuffixSeqKRCheck fs1 s rb rh1
2241 else SplitOnSuffixSeqKRLoop
2242 fs1 s rb rh1 ringHash
2243 FL.Done b -> do
2244 let rst = RB.startOf rb
2245 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2246 yieldProceed jump b
2247 Skip s -> go SPEC idx rh s fs
2248 Stop -> do
2249 -- do not issue a blank segment when we end at pattern
2250 if (idx == maxIndex) && RB.unsafeEqArray rb rh patArr
2251 then return Stop
2252 else if withSep
2253 then do
2254 r <- done fs
2255 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2256 else skip $ SplitOnSuffixSeqKRDone idx fs rb (RB.startOf rb)
2258 stepOuter gst (SplitOnSuffixSeqKRLoop fs0 st0 rb rh0 cksum0) =
2259 go SPEC fs0 st0 rh0 cksum0
2261 where
2263 go !_ !fs !st !rh !cksum = do
2264 res <- step (adaptState gst) st
2265 case res of
2266 Yield x s -> do
2267 old <- liftIO $ peek rh
2268 rh1 <- liftIO (RB.unsafeInsert rb rh x)
2269 let cksum1 = deltaCksum cksum old x
2270 r <- if withSep then fstep fs x else fstep fs old
2271 case r of
2272 FL.Partial fs1 ->
2273 if cksum1 /= patHash
2274 then go SPEC fs1 s rh1 cksum1
2275 else skip $ SplitOnSuffixSeqKRCheck fs1 s rb rh1
2276 FL.Done b -> do
2277 let rst = RB.startOf rb
2278 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2279 yieldProceed jump b
2280 Skip s -> go SPEC fs s rh cksum
2281 Stop ->
2282 if RB.unsafeEqArray rb rh patArr
2283 then return Stop
2284 else if withSep
2285 then do
2286 r <- done fs
2287 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2288 else skip $ SplitOnSuffixSeqKRDone patLen fs rb rh
2290 stepOuter _ (SplitOnSuffixSeqKRCheck fs st rb rh) = do
2291 if RB.unsafeEqArray rb rh patArr
2292 then do
2293 r <- done fs
2294 let rst = RB.startOf rb
2295 jump c = SplitOnSuffixSeqKRInit 0 c st rb rst
2296 yieldProceed jump r
2297 else skip $ SplitOnSuffixSeqKRLoop fs st rb rh patHash
2299 stepOuter _ (SplitOnSuffixSeqKRDone 0 fs _ _) = do
2300 r <- done fs
2301 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2302 stepOuter _ (SplitOnSuffixSeqKRDone n fs rb rh) = do
2303 old <- liftIO $ peek rh
2304 let rh1 = RB.advance rb rh
2305 r <- fstep fs old
2306 case r of
2307 FL.Partial fs1 -> skip $ SplitOnSuffixSeqKRDone (n - 1) fs1 rb rh1
2308 FL.Done b -> do
2309 let jump c = SplitOnSuffixSeqKRDone (n - 1) c rb rh1
2310 yieldProceed jump b
2312------------------------------------------------------------------------------
2313-- Nested Container Transformation
2314------------------------------------------------------------------------------
2316{-# ANN type SplitState Fuse #-}
2317data SplitState s arr
2318 = SplitInitial s
2319 | SplitBuffering s arr
2320 | SplitSplitting s arr
2321 | SplitYielding arr (SplitState s arr)
2322 | SplitFinishing
2324-- XXX An alternative approach would be to use a partial fold (Fold m a b) to
2325-- split using a splitBy like combinator. The Fold would consume upto the
2326-- separator and return any leftover which can then be fed to the next fold.
2328-- We can revisit this once we have partial folds/parsers.
2330-- | Performs infix separator style splitting.
2331{-# INLINE_NORMAL splitInnerBy #-}
2332splitInnerBy
2333 :: Monad m
2334 => (f a -> m (f a, Maybe (f a))) -- splitter
2335 -> (f a -> f a -> m (f a)) -- joiner
2336 -> Stream m (f a)
2337 -> Stream m (f a)
2338splitInnerBy splitter joiner (Stream step1 state1) =
2339 Stream step (SplitInitial state1)
2341 where
2343 {-# INLINE_LATE step #-}
2344 step gst (SplitInitial st) = do
2345 r <- step1 gst st
2346 case r of
2347 Yield x s -> do
2348 (x1, mx2) <- splitter x
2349 return $ case mx2 of
2350 Nothing -> Skip (SplitBuffering s x1)
2351 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
2352 Skip s -> return $ Skip (SplitInitial s)
2353 Stop -> return Stop
2355 step gst (SplitBuffering st buf) = do
2356 r <- step1 gst st
2357 case r of
2358 Yield x s -> do
2359 (x1, mx2) <- splitter x
2360 buf' <- joiner buf x1
2361 return $ case mx2 of
2362 Nothing -> Skip (SplitBuffering s buf')
2363 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
2364 Skip s -> return $ Skip (SplitBuffering s buf)
2365 Stop -> return $ Skip (SplitYielding buf SplitFinishing)
2367 step _ (SplitSplitting st buf) = do
2368 (x1, mx2) <- splitter buf
2369 return $ case mx2 of
2370 Nothing -> Skip $ SplitBuffering st x1
2371 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
2373 step _ (SplitYielding x next) = return $ Yield x next
2374 step _ SplitFinishing = return Stop
2376-- | Performs infix separator style splitting.
2377{-# INLINE_NORMAL splitInnerBySuffix #-}
2378splitInnerBySuffix
2379 :: (Monad m, Eq (f a), Monoid (f a))
2380 => (f a -> m (f a, Maybe (f a))) -- splitter
2381 -> (f a -> f a -> m (f a)) -- joiner
2382 -> Stream m (f a)
2383 -> Stream m (f a)
2384splitInnerBySuffix splitter joiner (Stream step1 state1) =
2385 Stream step (SplitInitial state1)
2387 where
2389 {-# INLINE_LATE step #-}
2390 step gst (SplitInitial st) = do
2391 r <- step1 gst st
2392 case r of
2393 Yield x s -> do
2394 (x1, mx2) <- splitter x
2395 return $ case mx2 of
2396 Nothing -> Skip (SplitBuffering s x1)
2397 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
2398 Skip s -> return $ Skip (SplitInitial s)
2399 Stop -> return Stop
2401 step gst (SplitBuffering st buf) = do
2402 r <- step1 gst st
2403 case r of
2404 Yield x s -> do
2405 (x1, mx2) <- splitter x
2406 buf' <- joiner buf x1
2407 return $ case mx2 of
2408 Nothing -> Skip (SplitBuffering s buf')
2409 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
2410 Skip s -> return $ Skip (SplitBuffering s buf)
2411 Stop -> return $
2412 if buf == mempty
2413 then Stop
2414 else Skip (SplitYielding buf SplitFinishing)
2416 step _ (SplitSplitting st buf) = do
2417 (x1, mx2) <- splitter buf
2418 return $ case mx2 of
2419 Nothing -> Skip $ SplitBuffering st x1
2420 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
2422 step _ (SplitYielding x next) = return $ Yield x next
2423 step _ SplitFinishing = return Stop