git.iko.soy/streamly
on cc44d97 help sign in
streamly/src/Streamly/Internal/Data/Stream/StreamD/Nesting.hs
2424 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"
149#include "ArrayMacros.h"
151import Control.Exception (assert)
152import Control.Monad.Catch (MonadThrow, throwM)
153import Control.Monad.IO.Class (MonadIO(..))
154import Data.Bits (shiftR, shiftL, (.|.), (.&.))
155#if __GLASGOW_HASKELL__ >= 801
156import Data.Functor.Identity ( Identity )
157#endif
158import Data.Word (Word32)
159import Foreign.Storable (Storable(..))
160import Fusion.Plugin.Types (Fuse(..))
161import GHC.Types (SPEC(..))
163import Streamly.Internal.Data.Array.Foreign.Type (Array(..))
164import Streamly.Internal.Data.Fold.Step (Step(..))
165import Streamly.Internal.Data.Fold.Type (Fold(..))
166import Streamly.Internal.Data.Parser (ParseError(..))
167import Streamly.Internal.Data.Refold.Type (Refold(..))
168import Streamly.Internal.Data.SVar.Type (adaptState)
169import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
170import Streamly.Internal.Data.Unfold.Type (Unfold(..))
172import qualified Streamly.Internal.Data.Array.Foreign.Type as A
173import qualified Streamly.Internal.Data.Fold as FL
174import qualified Streamly.Internal.Data.Parser as PR
175import qualified Streamly.Internal.Data.Parser.ParserD as PRD
176import qualified Streamly.Internal.Data.Ring.Foreign as RB
178import Streamly.Internal.Data.Stream.StreamD.Type
180import Prelude hiding (concatMap, mapM, zipWith)
182------------------------------------------------------------------------------
183-- Appending
184------------------------------------------------------------------------------
186data AppendState s1 s2 = AppendFirst s1 | AppendSecond s2
188-- Note that this could be much faster compared to the CPS stream. However, as
189-- the number of streams being composed increases this may become expensive.
190-- Need to see where the breaking point is between the two.
191--
192{-# INLINE_NORMAL append #-}
193append :: Monad m => Stream m a -> Stream m a -> Stream m a
194append (Stream step1 state1) (Stream step2 state2) =
195 Stream step (AppendFirst state1)
197 where
199 {-# INLINE_LATE step #-}
200 step gst (AppendFirst st) = do
201 r <- step1 gst st
202 return $ case r of
203 Yield a s -> Yield a (AppendFirst s)
204 Skip s -> Skip (AppendFirst s)
205 Stop -> Skip (AppendSecond state2)
207 step gst (AppendSecond st) = do
208 r <- step2 gst st
209 return $ case r of
210 Yield a s -> Yield a (AppendSecond s)
211 Skip s -> Skip (AppendSecond s)
212 Stop -> Stop
214------------------------------------------------------------------------------
215-- Interleaving
216------------------------------------------------------------------------------
218data InterleaveState s1 s2 = InterleaveFirst s1 s2 | InterleaveSecond s1 s2
219 | InterleaveSecondOnly s2 | InterleaveFirstOnly s1
221{-# INLINE_NORMAL interleave #-}
222interleave :: Monad m => Stream m a -> Stream m a -> Stream m a
223interleave (Stream step1 state1) (Stream step2 state2) =
224 Stream step (InterleaveFirst state1 state2)
226 where
228 {-# INLINE_LATE step #-}
229 step gst (InterleaveFirst st1 st2) = do
230 r <- step1 gst st1
231 return $ case r of
232 Yield a s -> Yield a (InterleaveSecond s st2)
233 Skip s -> Skip (InterleaveFirst s st2)
234 Stop -> Skip (InterleaveSecondOnly st2)
236 step gst (InterleaveSecond st1 st2) = do
237 r <- step2 gst st2
238 return $ case r of
239 Yield a s -> Yield a (InterleaveFirst st1 s)
240 Skip s -> Skip (InterleaveSecond st1 s)
241 Stop -> Skip (InterleaveFirstOnly st1)
243 step gst (InterleaveFirstOnly st1) = do
244 r <- step1 gst st1
245 return $ case r of
246 Yield a s -> Yield a (InterleaveFirstOnly s)
247 Skip s -> Skip (InterleaveFirstOnly s)
248 Stop -> Stop
250 step gst (InterleaveSecondOnly st2) = do
251 r <- step2 gst st2
252 return $ case r of
253 Yield a s -> Yield a (InterleaveSecondOnly s)
254 Skip s -> Skip (InterleaveSecondOnly s)
255 Stop -> Stop
257{-# INLINE_NORMAL interleaveMin #-}
258interleaveMin :: Monad m => Stream m a -> Stream m a -> Stream m a
259interleaveMin (Stream step1 state1) (Stream step2 state2) =
260 Stream step (InterleaveFirst state1 state2)
262 where
264 {-# INLINE_LATE step #-}
265 step gst (InterleaveFirst st1 st2) = do
266 r <- step1 gst st1
267 return $ case r of
268 Yield a s -> Yield a (InterleaveSecond s st2)
269 Skip s -> Skip (InterleaveFirst s st2)
270 Stop -> Stop
272 step gst (InterleaveSecond st1 st2) = do
273 r <- step2 gst st2
274 return $ case r of
275 Yield a s -> Yield a (InterleaveFirst st1 s)
276 Skip s -> Skip (InterleaveSecond st1 s)
277 Stop -> Stop
279 step _ (InterleaveFirstOnly _) = undefined
280 step _ (InterleaveSecondOnly _) = undefined
282{-# INLINE_NORMAL interleaveSuffix #-}
283interleaveSuffix :: Monad m => Stream m a -> Stream m a -> Stream m a
284interleaveSuffix (Stream step1 state1) (Stream step2 state2) =
285 Stream step (InterleaveFirst state1 state2)
287 where
289 {-# INLINE_LATE step #-}
290 step gst (InterleaveFirst st1 st2) = do
291 r <- step1 gst st1
292 return $ case r of
293 Yield a s -> Yield a (InterleaveSecond s st2)
294 Skip s -> Skip (InterleaveFirst s st2)
295 Stop -> Stop
297 step gst (InterleaveSecond st1 st2) = do
298 r <- step2 gst st2
299 return $ case r of
300 Yield a s -> Yield a (InterleaveFirst st1 s)
301 Skip s -> Skip (InterleaveSecond st1 s)
302 Stop -> Skip (InterleaveFirstOnly st1)
304 step gst (InterleaveFirstOnly st1) = do
305 r <- step1 gst st1
306 return $ case r of
307 Yield a s -> Yield a (InterleaveFirstOnly s)
308 Skip s -> Skip (InterleaveFirstOnly s)
309 Stop -> Stop
311 step _ (InterleaveSecondOnly _) = undefined
313data InterleaveInfixState s1 s2 a
314 = InterleaveInfixFirst s1 s2
315 | InterleaveInfixSecondBuf s1 s2
316 | InterleaveInfixSecondYield s1 s2 a
317 | InterleaveInfixFirstYield s1 s2 a
318 | InterleaveInfixFirstOnly s1
320{-# INLINE_NORMAL interleaveInfix #-}
321interleaveInfix :: Monad m => Stream m a -> Stream m a -> Stream m a
322interleaveInfix (Stream step1 state1) (Stream step2 state2) =
323 Stream step (InterleaveInfixFirst state1 state2)
325 where
327 {-# INLINE_LATE step #-}
328 step gst (InterleaveInfixFirst st1 st2) = do
329 r <- step1 gst st1
330 return $ case r of
331 Yield a s -> Yield a (InterleaveInfixSecondBuf s st2)
332 Skip s -> Skip (InterleaveInfixFirst s st2)
333 Stop -> Stop
335 step gst (InterleaveInfixSecondBuf st1 st2) = do
336 r <- step2 gst st2
337 return $ case r of
338 Yield a s -> Skip (InterleaveInfixSecondYield st1 s a)
339 Skip s -> Skip (InterleaveInfixSecondBuf st1 s)
340 Stop -> Skip (InterleaveInfixFirstOnly st1)
342 step gst (InterleaveInfixSecondYield st1 st2 x) = do
343 r <- step1 gst st1
344 return $ case r of
345 Yield a s -> Yield x (InterleaveInfixFirstYield s st2 a)
346 Skip s -> Skip (InterleaveInfixSecondYield s st2 x)
347 Stop -> Stop
349 step _ (InterleaveInfixFirstYield st1 st2 x) = do
350 return $ Yield x (InterleaveInfixSecondBuf st1 st2)
352 step gst (InterleaveInfixFirstOnly st1) = do
353 r <- step1 gst st1
354 return $ case r of
355 Yield a s -> Yield a (InterleaveInfixFirstOnly s)
356 Skip s -> Skip (InterleaveInfixFirstOnly s)
357 Stop -> Stop
359------------------------------------------------------------------------------
360-- Scheduling
361------------------------------------------------------------------------------
363{-# INLINE_NORMAL roundRobin #-}
364roundRobin :: Monad m => Stream m a -> Stream m a -> Stream m a
365roundRobin (Stream step1 state1) (Stream step2 state2) =
366 Stream step (InterleaveFirst state1 state2)
368 where
370 {-# INLINE_LATE step #-}
371 step gst (InterleaveFirst st1 st2) = do
372 r <- step1 gst st1
373 return $ case r of
374 Yield a s -> Yield a (InterleaveSecond s st2)
375 Skip s -> Skip (InterleaveSecond s st2)
376 Stop -> Skip (InterleaveSecondOnly st2)
378 step gst (InterleaveSecond st1 st2) = do
379 r <- step2 gst st2
380 return $ case r of
381 Yield a s -> Yield a (InterleaveFirst st1 s)
382 Skip s -> Skip (InterleaveFirst st1 s)
383 Stop -> Skip (InterleaveFirstOnly st1)
385 step gst (InterleaveSecondOnly st2) = do
386 r <- step2 gst st2
387 return $ case r of
388 Yield a s -> Yield a (InterleaveSecondOnly s)
389 Skip s -> Skip (InterleaveSecondOnly s)
390 Stop -> Stop
392 step gst (InterleaveFirstOnly st1) = do
393 r <- step1 gst st1
394 return $ case r of
395 Yield a s -> Yield a (InterleaveFirstOnly s)
396 Skip s -> Skip (InterleaveFirstOnly s)
397 Stop -> Stop
399------------------------------------------------------------------------------
400-- Zipping
401------------------------------------------------------------------------------
403{-# INLINE_NORMAL zipWithM #-}
404zipWithM :: Monad m
405 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
406zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)
407 where
408 {-# INLINE_LATE step #-}
409 step gst (sa, sb, Nothing) = do
410 r <- stepa (adaptState gst) sa
411 return $
412 case r of
413 Yield x sa' -> Skip (sa', sb, Just x)
414 Skip sa' -> Skip (sa', sb, Nothing)
415 Stop -> Stop
417 step gst (sa, sb, Just x) = do
418 r <- stepb (adaptState gst) sb
419 case r of
420 Yield y sb' -> do
421 z <- f x y
422 return $ Yield z (sa, sb', Nothing)
423 Skip sb' -> return $ Skip (sa, sb', Just x)
424 Stop -> return Stop
426#if __GLASGOW_HASKELL__ >= 801
427{-# RULES "zipWithM xs xs"
428 forall f xs. zipWithM @Identity f xs xs = mapM (\x -> f x x) xs #-}
429#endif
431{-# INLINE zipWith #-}
432zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
433zipWith f = zipWithM (\a b -> return (f a b))
435------------------------------------------------------------------------------
436-- Merging
437------------------------------------------------------------------------------
439{-# INLINE_NORMAL mergeByM #-}
440mergeByM
441 :: (Monad m)
442 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
443mergeByM cmp (Stream stepa ta) (Stream stepb tb) =
444 Stream step (Just ta, Just tb, Nothing, Nothing)
445 where
446 {-# INLINE_LATE step #-}
448 -- one of the values is missing, and the corresponding stream is running
449 step gst (Just sa, sb, Nothing, b) = do
450 r <- stepa gst sa
451 return $ case r of
452 Yield a sa' -> Skip (Just sa', sb, Just a, b)
453 Skip sa' -> Skip (Just sa', sb, Nothing, b)
454 Stop -> Skip (Nothing, sb, Nothing, b)
456 step gst (sa, Just sb, a, Nothing) = do
457 r <- stepb gst sb
458 return $ case r of
459 Yield b sb' -> Skip (sa, Just sb', a, Just b)
460 Skip sb' -> Skip (sa, Just sb', a, Nothing)
461 Stop -> Skip (sa, Nothing, a, Nothing)
463 -- both the values are available
464 step _ (sa, sb, Just a, Just b) = do
465 res <- cmp a b
466 return $ case res of
467 GT -> Yield b (sa, sb, Just a, Nothing)
468 _ -> Yield a (sa, sb, Nothing, Just b)
470 -- one of the values is missing, corresponding stream is done
471 step _ (Nothing, sb, Nothing, Just b) =
472 return $ Yield b (Nothing, sb, Nothing, Nothing)
474 step _ (sa, Nothing, Just a, Nothing) =
475 return $ Yield a (sa, Nothing, Nothing, Nothing)
477 step _ (Nothing, Nothing, Nothing, Nothing) = return Stop
479{-# INLINE mergeBy #-}
480mergeBy
481 :: (Monad m)
482 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
483mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
485------------------------------------------------------------------------------
486-- Combine N Streams - unfoldMany
487------------------------------------------------------------------------------
489data ConcatUnfoldInterleaveState o i =
490 ConcatUnfoldInterleaveOuter o [i]
491 | ConcatUnfoldInterleaveInner o [i]
492 | ConcatUnfoldInterleaveInnerL [i] [i]
493 | ConcatUnfoldInterleaveInnerR [i] [i]
495-- XXX use arrays to store state instead of lists?
496--
497-- XXX In general we can use different scheduling strategies e.g. how to
498-- schedule the outer vs inner loop or assigning weights to different streams
499-- or outer and inner loops.
501-- After a yield, switch to the next stream. Do not switch streams on Skip.
502-- Yield from outer stream switches to the inner stream.
503--
504-- There are two choices here, (1) exhaust the outer stream first and then
505-- start yielding from the inner streams, this is much simpler to implement,
506-- (2) yield at least one element from an inner stream before going back to
507-- outer stream and opening the next stream from it.
508--
509-- Ideally, we need some scheduling bias to inner streams vs outer stream.
510-- Maybe we can configure the behavior.
511--
512-- XXX Instead of using "concatPairsWith wSerial" we can implement an N-way
513-- interleaving CPS combinator which behaves like unfoldManyInterleave. Intead
514-- of pairing up the streams We just need to go yielding one element from each
515-- stream and storing the remaining streams and then keep doing rounds through
516-- those in a round robin fashion. This would be much like wAsync.
517--
518-- | This does not pair streams like concatPairsWith, instead, it goes through
519-- each stream one by one and yields one element from each stream. After it
520-- goes to the last stream it reverses the traversal to come back to the first
521-- stream yielding elements from each stream on its way back to the first
522-- stream and so on.
523--
524-- >>> input = Stream.fromList [[1,1],[2,2],[3,3],[4,4],[5,5]]
525-- >>> Stream.toList $ Stream.unfoldManyInterleave Unfold.fromList input
526-- [1,2,3,4,5,5,4,3,2,1]
527--
528-- Note that this is order of magnitude more efficient than "concatPairsWith
529-- wSerial"
530{-# INLINE_NORMAL unfoldManyInterleave #-}
531unfoldManyInterleave :: Monad m => Unfold m a b -> Stream m a -> Stream m b
532unfoldManyInterleave (Unfold istep inject) (Stream ostep ost) =
533 Stream step (ConcatUnfoldInterleaveOuter ost [])
535 where
537 {-# INLINE_LATE step #-}
538 step gst (ConcatUnfoldInterleaveOuter o ls) = do
539 r <- ostep (adaptState gst) o
540 case r of
541 Yield a o' -> do
542 i <- inject a
543 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
544 Skip o' -> return $ Skip (ConcatUnfoldInterleaveOuter o' ls)
545 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
547 step _ (ConcatUnfoldInterleaveInner _ []) = undefined
548 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
549 r <- istep st
550 return $ case r of
551 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
552 Skip s -> Skip (ConcatUnfoldInterleaveInner o (s:ls))
553 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
555 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
556 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
557 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
559 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
560 r <- istep st
561 return $ case r of
562 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
563 Skip s -> Skip (ConcatUnfoldInterleaveInnerL (s:ls) rs)
564 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
566 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
567 step _ (ConcatUnfoldInterleaveInnerR ls []) =
568 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
570 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
571 r <- istep st
572 return $ case r of
573 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
574 Skip s -> Skip (ConcatUnfoldInterleaveInnerR ls (s:rs))
575 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
577-- XXX In general we can use different scheduling strategies e.g. how to
578-- schedule the outer vs inner loop or assigning weights to different streams
579-- or outer and inner loops.
580--
581-- This could be inefficient if the tasks are too small.
582--
583-- Compared to unfoldManyInterleave this one switches streams on Skips.
584--
585{-# INLINE_NORMAL unfoldManyRoundRobin #-}
586unfoldManyRoundRobin :: Monad m => Unfold m a b -> Stream m a -> Stream m b
587unfoldManyRoundRobin (Unfold istep inject) (Stream ostep ost) =
588 Stream step (ConcatUnfoldInterleaveOuter ost [])
589 where
590 {-# INLINE_LATE step #-}
591 step gst (ConcatUnfoldInterleaveOuter o ls) = do
592 r <- ostep (adaptState gst) o
593 case r of
594 Yield a o' -> do
595 i <- inject a
596 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
597 Skip o' -> return $ Skip (ConcatUnfoldInterleaveInner o' ls)
598 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
600 step _ (ConcatUnfoldInterleaveInner o []) =
601 return $ Skip (ConcatUnfoldInterleaveOuter o [])
603 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
604 r <- istep st
605 return $ case r of
606 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
607 Skip s -> Skip (ConcatUnfoldInterleaveOuter o (s:ls))
608 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
610 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
611 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
612 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
614 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
615 r <- istep st
616 return $ case r of
617 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
618 Skip s -> Skip (ConcatUnfoldInterleaveInnerL ls (s:rs))
619 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
621 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
622 step _ (ConcatUnfoldInterleaveInnerR ls []) =
623 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
625 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
626 r <- istep st
627 return $ case r of
628 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
629 Skip s -> Skip (ConcatUnfoldInterleaveInnerR (s:ls) rs)
630 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
632------------------------------------------------------------------------------
633-- Combine N Streams - interpose
634------------------------------------------------------------------------------
636{-# ANN type InterposeSuffixState Fuse #-}
637data InterposeSuffixState s1 i1 =
638 InterposeSuffixFirst s1
639 -- | InterposeSuffixFirstYield s1 i1
640 | InterposeSuffixFirstInner s1 i1
641 | InterposeSuffixSecond s1
643-- Note that if an unfolded layer turns out to be nil we still emit the
644-- separator effect. An alternate behavior could be to emit the separator
645-- effect only if at least one element has been yielded by the unfolding.
646-- However, that becomes a bit complicated, so we have chosen the former
647-- behvaior for now.
648{-# INLINE_NORMAL interposeSuffix #-}
649interposeSuffix
650 :: Monad m
651 => m c -> Unfold m b c -> Stream m b -> Stream m c
652interposeSuffix
653 action
654 (Unfold istep1 inject1) (Stream step1 state1) =
655 Stream step (InterposeSuffixFirst state1)
657 where
659 {-# INLINE_LATE step #-}
660 step gst (InterposeSuffixFirst s1) = do
661 r <- step1 (adaptState gst) s1
662 case r of
663 Yield a s -> do
664 i <- inject1 a
665 i `seq` return (Skip (InterposeSuffixFirstInner s i))
666 -- i `seq` return (Skip (InterposeSuffixFirstYield s i))
667 Skip s -> return $ Skip (InterposeSuffixFirst s)
668 Stop -> return Stop
670 {-
671 step _ (InterposeSuffixFirstYield s1 i1) = do
672 r <- istep1 i1
673 return $ case r of
674 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
675 Skip i' -> Skip (InterposeSuffixFirstYield s1 i')
676 Stop -> Skip (InterposeSuffixFirst s1)
677 -}
679 step _ (InterposeSuffixFirstInner s1 i1) = do
680 r <- istep1 i1
681 return $ case r of
682 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
683 Skip i' -> Skip (InterposeSuffixFirstInner s1 i')
684 Stop -> Skip (InterposeSuffixSecond s1)
686 step _ (InterposeSuffixSecond s1) = do
687 r <- action
688 return $ Yield r (InterposeSuffixFirst s1)
690{-# ANN type InterposeState Fuse #-}
691data InterposeState s1 i1 a =
692 InterposeFirst s1
693 -- | InterposeFirstYield s1 i1
694 | InterposeFirstInner s1 i1
695 | InterposeFirstInject s1
696 -- | InterposeFirstBuf s1 i1
697 | InterposeSecondYield s1 i1
698 -- -- | InterposeSecondYield s1 i1 a
699 -- -- | InterposeFirstResume s1 i1 a
701-- Note that this only interposes the pure values, we may run many effects to
702-- generate those values as some effects may not generate anything (Skip).
703{-# INLINE_NORMAL interpose #-}
704interpose :: Monad m => m c -> Unfold m b c -> Stream m b -> Stream m c
705interpose
706 action
707 (Unfold istep1 inject1) (Stream step1 state1) =
708 Stream step (InterposeFirst state1)
710 where
712 {-# INLINE_LATE step #-}
713 step gst (InterposeFirst s1) = do
714 r <- step1 (adaptState gst) s1
715 case r of
716 Yield a s -> do
717 i <- inject1 a
718 i `seq` return (Skip (InterposeFirstInner s i))
719 -- i `seq` return (Skip (InterposeFirstYield s i))
720 Skip s -> return $ Skip (InterposeFirst s)
721 Stop -> return Stop
723 {-
724 step _ (InterposeFirstYield s1 i1) = do
725 r <- istep1 i1
726 return $ case r of
727 Yield x i' -> Yield x (InterposeFirstInner s1 i')
728 Skip i' -> Skip (InterposeFirstYield s1 i')
729 Stop -> Skip (InterposeFirst s1)
730 -}
732 step _ (InterposeFirstInner s1 i1) = do
733 r <- istep1 i1
734 return $ case r of
735 Yield x i' -> Yield x (InterposeFirstInner s1 i')
736 Skip i' -> Skip (InterposeFirstInner s1 i')
737 Stop -> Skip (InterposeFirstInject s1)
739 step gst (InterposeFirstInject s1) = do
740 r <- step1 (adaptState gst) s1
741 case r of
742 Yield a s -> do
743 i <- inject1 a
744 -- i `seq` return (Skip (InterposeFirstBuf s i))
745 i `seq` return (Skip (InterposeSecondYield s i))
746 Skip s -> return $ Skip (InterposeFirstInject s)
747 Stop -> return Stop
749 {-
750 step _ (InterposeFirstBuf s1 i1) = do
751 r <- istep1 i1
752 return $ case r of
753 Yield x i' -> Skip (InterposeSecondYield s1 i' x)
754 Skip i' -> Skip (InterposeFirstBuf s1 i')
755 Stop -> Stop
756 -}
758 {-
759 step _ (InterposeSecondYield s1 i1 v) = do
760 r <- action
761 return $ Yield r (InterposeFirstResume s1 i1 v)
762 -}
763 step _ (InterposeSecondYield s1 i1) = do
764 r <- action
765 return $ Yield r (InterposeFirstInner s1 i1)
767 {-
768 step _ (InterposeFirstResume s1 i1 v) = do
769 return $ Yield v (InterposeFirstInner s1 i1)
770 -}
772------------------------------------------------------------------------------
773-- Combine N Streams - intercalate
774------------------------------------------------------------------------------
776data ICUState s1 s2 i1 i2 =
777 ICUFirst s1 s2
778 | ICUSecond s1 s2
779 | ICUSecondOnly s2
780 | ICUFirstOnly s1
781 | ICUFirstInner s1 s2 i1
782 | ICUSecondInner s1 s2 i2
783 | ICUFirstOnlyInner s1 i1
784 | ICUSecondOnlyInner s2 i2
786-- | Interleave streams (full streams, not the elements) unfolded from two
787-- input streams and concat. Stop when the first stream stops. If the second
788-- stream ends before the first one then first stream still keeps running alone
789-- without any interleaving with the second stream.
790--
791-- [a1, a2, ... an] [b1, b2 ...]
792-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
793-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
794-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
795--
796{-# INLINE_NORMAL gintercalateSuffix #-}
797gintercalateSuffix
798 :: Monad m
799 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
800gintercalateSuffix
801 (Unfold istep1 inject1) (Stream step1 state1)
802 (Unfold istep2 inject2) (Stream step2 state2) =
803 Stream step (ICUFirst state1 state2)
805 where
807 {-# INLINE_LATE step #-}
808 step gst (ICUFirst s1 s2) = do
809 r <- step1 (adaptState gst) s1
810 case r of
811 Yield a s -> do
812 i <- inject1 a
813 i `seq` return (Skip (ICUFirstInner s s2 i))
814 Skip s -> return $ Skip (ICUFirst s s2)
815 Stop -> return Stop
817 step gst (ICUFirstOnly s1) = do
818 r <- step1 (adaptState gst) s1
819 case r of
820 Yield a s -> do
821 i <- inject1 a
822 i `seq` return (Skip (ICUFirstOnlyInner s i))
823 Skip s -> return $ Skip (ICUFirstOnly s)
824 Stop -> return Stop
826 step _ (ICUFirstInner s1 s2 i1) = do
827 r <- istep1 i1
828 return $ case r of
829 Yield x i' -> Yield x (ICUFirstInner s1 s2 i')
830 Skip i' -> Skip (ICUFirstInner s1 s2 i')
831 Stop -> Skip (ICUSecond s1 s2)
833 step _ (ICUFirstOnlyInner s1 i1) = do
834 r <- istep1 i1
835 return $ case r of
836 Yield x i' -> Yield x (ICUFirstOnlyInner s1 i')
837 Skip i' -> Skip (ICUFirstOnlyInner s1 i')
838 Stop -> Skip (ICUFirstOnly s1)
840 step gst (ICUSecond s1 s2) = do
841 r <- step2 (adaptState gst) s2
842 case r of
843 Yield a s -> do
844 i <- inject2 a
845 i `seq` return (Skip (ICUSecondInner s1 s i))
846 Skip s -> return $ Skip (ICUSecond s1 s)
847 Stop -> return $ Skip (ICUFirstOnly s1)
849 step _ (ICUSecondInner s1 s2 i2) = do
850 r <- istep2 i2
851 return $ case r of
852 Yield x i' -> Yield x (ICUSecondInner s1 s2 i')
853 Skip i' -> Skip (ICUSecondInner s1 s2 i')
854 Stop -> Skip (ICUFirst s1 s2)
856 step _ (ICUSecondOnly _s2) = undefined
857 step _ (ICUSecondOnlyInner _s2 _i2) = undefined
859data ICALState s1 s2 i1 i2 a =
860 ICALFirst s1 s2
861 -- | ICALFirstYield s1 s2 i1
862 | ICALFirstInner s1 s2 i1
863 | ICALFirstOnly s1
864 | ICALFirstOnlyInner s1 i1
865 | ICALSecondInject s1 s2
866 | ICALFirstInject s1 s2 i2
867 -- | ICALFirstBuf s1 s2 i1 i2
868 | ICALSecondInner s1 s2 i1 i2
869 -- -- | ICALSecondInner s1 s2 i1 i2 a
870 -- -- | ICALFirstResume s1 s2 i1 i2 a
872-- | Interleave streams (full streams, not the elements) unfolded from two
873-- input streams and concat. Stop when the first stream stops. If the second
874-- stream ends before the first one then first stream still keeps running alone
875-- without any interleaving with the second stream.
876--
877-- [a1, a2, ... an] [b1, b2 ...]
878-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
879-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
880-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
881--
882{-# INLINE_NORMAL gintercalate #-}
883gintercalate
884 :: Monad m
885 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
886gintercalate
887 (Unfold istep1 inject1) (Stream step1 state1)
888 (Unfold istep2 inject2) (Stream step2 state2) =
889 Stream step (ICALFirst state1 state2)
891 where
893 {-# INLINE_LATE step #-}
894 step gst (ICALFirst s1 s2) = do
895 r <- step1 (adaptState gst) s1
896 case r of
897 Yield a s -> do
898 i <- inject1 a
899 i `seq` return (Skip (ICALFirstInner s s2 i))
900 -- i `seq` return (Skip (ICALFirstYield s s2 i))
901 Skip s -> return $ Skip (ICALFirst s s2)
902 Stop -> return Stop
904 {-
905 step _ (ICALFirstYield s1 s2 i1) = do
906 r <- istep1 i1
907 return $ case r of
908 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
909 Skip i' -> Skip (ICALFirstYield s1 s2 i')
910 Stop -> Skip (ICALFirst s1 s2)
911 -}
913 step _ (ICALFirstInner s1 s2 i1) = do
914 r <- istep1 i1
915 return $ case r of
916 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
917 Skip i' -> Skip (ICALFirstInner s1 s2 i')
918 Stop -> Skip (ICALSecondInject s1 s2)
920 step gst (ICALFirstOnly s1) = do
921 r <- step1 (adaptState gst) s1
922 case r of
923 Yield a s -> do
924 i <- inject1 a
925 i `seq` return (Skip (ICALFirstOnlyInner s i))
926 Skip s -> return $ Skip (ICALFirstOnly s)
927 Stop -> return Stop
929 step _ (ICALFirstOnlyInner s1 i1) = do
930 r <- istep1 i1
931 return $ case r of
932 Yield x i' -> Yield x (ICALFirstOnlyInner s1 i')
933 Skip i' -> Skip (ICALFirstOnlyInner s1 i')
934 Stop -> Skip (ICALFirstOnly s1)
936 -- We inject the second stream even before checking if the first stream
937 -- would yield any more elements. There is no clear choice whether we
938 -- should do this before or after that. Doing it after may make the state
939 -- machine a bit simpler though.
940 step gst (ICALSecondInject s1 s2) = do
941 r <- step2 (adaptState gst) s2
942 case r of
943 Yield a s -> do
944 i <- inject2 a
945 i `seq` return (Skip (ICALFirstInject s1 s i))
946 Skip s -> return $ Skip (ICALSecondInject s1 s)
947 Stop -> return $ Skip (ICALFirstOnly s1)
949 step gst (ICALFirstInject s1 s2 i2) = do
950 r <- step1 (adaptState gst) s1
951 case r of
952 Yield a s -> do
953 i <- inject1 a
954 i `seq` return (Skip (ICALSecondInner s s2 i i2))
955 -- i `seq` return (Skip (ICALFirstBuf s s2 i i2))
956 Skip s -> return $ Skip (ICALFirstInject s s2 i2)
957 Stop -> return Stop
959 {-
960 step _ (ICALFirstBuf s1 s2 i1 i2) = do
961 r <- istep1 i1
962 return $ case r of
963 Yield x i' -> Skip (ICALSecondInner s1 s2 i' i2 x)
964 Skip i' -> Skip (ICALFirstBuf s1 s2 i' i2)
965 Stop -> Stop
967 step _ (ICALSecondInner s1 s2 i1 i2 v) = do
968 r <- istep2 i2
969 return $ case r of
970 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i' v)
971 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i' v)
972 Stop -> Skip (ICALFirstResume s1 s2 i1 i2 v)
973 -}
975 step _ (ICALSecondInner s1 s2 i1 i2) = do
976 r <- istep2 i2
977 return $ case r of
978 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i')
979 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i')
980 Stop -> Skip (ICALFirstInner s1 s2 i1)
981 -- Stop -> Skip (ICALFirstResume s1 s2 i1 i2)
983 {-
984 step _ (ICALFirstResume s1 s2 i1 i2 x) = do
985 return $ Yield x (ICALFirstInner s1 s2 i1 i2)
986 -}
988------------------------------------------------------------------------------
989-- Folding
990------------------------------------------------------------------------------
992{-# ANN type FIterState Fuse #-}
993data FIterState s f m a b
994 = FIterInit s f
995 | forall fs. FIterStream s (fs -> a -> m (FL.Step fs b)) fs (fs -> m b)
996 | FIterYield b (FIterState s f m a b)
997 | FIterStop
999{-# INLINE_NORMAL foldIterateM #-}
1000foldIterateM ::
1001 Monad m => (b -> m (FL.Fold m a b)) -> m b -> Stream m a -> Stream m b
1002foldIterateM func seed0 (Stream step state) =
1003 Stream stepOuter (FIterInit state seed0)
1005 where
1007 {-# INLINE iterStep #-}
1008 iterStep from st fstep extract = do
1009 res <- from
1010 return
1011 $ Skip
1012 $ case res of
1013 FL.Partial fs -> FIterStream st fstep fs extract
1014 FL.Done fb -> FIterYield fb $ FIterInit st (return fb)
1016 {-# INLINE_LATE stepOuter #-}
1017 stepOuter _ (FIterInit st seed) = do
1018 (FL.Fold fstep initial extract) <- seed >>= func
1019 iterStep initial st fstep extract
1020 stepOuter gst (FIterStream st fstep fs extract) = do
1021 r <- step (adaptState gst) st
1022 case r of
1023 Yield x s -> do
1024 iterStep (fstep fs x) s fstep extract
1025 Skip s -> return $ Skip $ FIterStream s fstep fs extract
1026 Stop -> do
1027 b <- extract fs
1028 return $ Skip $ FIterYield b FIterStop
1029 stepOuter _ (FIterYield a next) = return $ Yield a next
1030 stepOuter _ FIterStop = return Stop
1032{-# ANN type CIterState Fuse #-}
1033data CIterState s f fs b
1034 = CIterInit s f
1035 | CIterConsume s fs
1036 | CIterYield b (CIterState s f fs b)
1037 | CIterStop
1039-- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
1040-- much more efficient due to stream fusion.
1042-- /Internal/
1043{-# INLINE_NORMAL refoldIterateM #-}
1044refoldIterateM ::
1045 Monad m => Refold m b a b -> m b -> Stream m a -> Stream m b
1046refoldIterateM (Refold fstep finject fextract) initial (Stream step state) =
1047 Stream stepOuter (CIterInit state initial)
1049 where
1051 {-# INLINE iterStep #-}
1052 iterStep st action = do
1053 res <- action
1054 return
1055 $ Skip
1056 $ case res of
1057 FL.Partial fs -> CIterConsume st fs
1058 FL.Done fb -> CIterYield fb $ CIterInit st (return fb)
1060 {-# INLINE_LATE stepOuter #-}
1061 stepOuter _ (CIterInit st action) = do
1062 iterStep st (action >>= finject)
1063 stepOuter gst (CIterConsume st fs) = do
1064 r <- step (adaptState gst) st
1065 case r of
1066 Yield x s -> iterStep s (fstep fs x)
1067 Skip s -> return $ Skip $ CIterConsume s fs
1068 Stop -> do
1069 b <- fextract fs
1070 return $ Skip $ CIterYield b CIterStop
1071 stepOuter _ (CIterYield a next) = return $ Yield a next
1072 stepOuter _ CIterStop = return Stop
1074-- "n" elements at the end are dropped by the fold.
1075{-# INLINE sliceBy #-}
1076sliceBy :: Monad m => Fold m a Int -> Int -> Refold m (Int, Int) a (Int, Int)
1077sliceBy (Fold step1 initial1 extract1) n = Refold step inject extract
1079 where
1081 inject (i, len) = do
1082 r <- initial1
1083 return $ case r of
1084 Partial s -> Partial $ Tuple' (i + len + n) s
1085 Done l -> Done (i, l)
1087 step (Tuple' i s) x = do
1088 r <- step1 s x
1089 return $ case r of
1090 Partial s1 -> Partial $ Tuple' i s1
1091 Done len -> Done (i, len)
1093 extract (Tuple' i s) = (i,) <$> extract1 s
1095{-# INLINE sliceOnSuffix #-}
1096sliceOnSuffix :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
1097sliceOnSuffix predicate =
1098 -- Scan the stream with the given refold
1099 refoldIterateM
1100 (sliceBy (FL.takeEndBy_ predicate FL.length) 1)
1101 (return (-1, 0))
1103------------------------------------------------------------------------------
1104-- Parsing
1105------------------------------------------------------------------------------
1107{-# ANN type ParseChunksState Fuse #-}
1108data ParseChunksState x inpBuf st pst =
1109 ParseChunksInit inpBuf st
1110 | ParseChunksInitLeftOver inpBuf
1111 | ParseChunksStream st inpBuf !pst
1112 | ParseChunksBuf inpBuf st inpBuf !pst
1113 | ParseChunksYield x (ParseChunksState x inpBuf st pst)
1115{-# INLINE_NORMAL parseMany #-}
1116parseMany
1117 :: MonadThrow m
1118 => PRD.Parser m a b
1119 -> Stream m a
1120 -> Stream m b
1121parseMany (PRD.Parser pstep initial extract) (Stream step state) =
1122 Stream stepOuter (ParseChunksInit [] state)
1124 where
1126 {-# INLINE_LATE stepOuter #-}
1127 -- Buffer is empty, get the first element from the stream, initialize the
1128 -- fold and then go to stream processing loop.
1129 stepOuter gst (ParseChunksInit [] st) = do
1130 r <- step (adaptState gst) st
1131 case r of
1132 Yield x s -> do
1133 res <- initial
1134 case res of
1135 PRD.IPartial ps ->
1136 return $ Skip $ ParseChunksBuf [x] s [] ps
1137 PRD.IDone pb ->
1138 let next = ParseChunksInit [x] s
1139 in return $ Skip $ ParseChunksYield pb next
1140 PRD.IError err -> throwM $ ParseError err
1141 Skip s -> return $ Skip $ ParseChunksInit [] s
1142 Stop -> return Stop
1144 -- Buffer is not empty, go to buffered processing loop
1145 stepOuter _ (ParseChunksInit src st) = do
1146 res <- initial
1147 case res of
1148 PRD.IPartial ps ->
1149 return $ Skip $ ParseChunksBuf src st [] ps
1150 PRD.IDone pb ->
1151 let next = ParseChunksInit src st
1152 in return $ Skip $ ParseChunksYield pb next
1153 PRD.IError err -> throwM $ ParseError err
1155 -- XXX we just discard any leftover input at the end
1156 stepOuter _ (ParseChunksInitLeftOver _) = return Stop
1158 -- Buffer is empty, process elements from the stream
1159 stepOuter gst (ParseChunksStream st buf pst) = do
1160 r <- step (adaptState gst) st
1161 case r of
1162 Yield x s -> do
1163 pRes <- pstep pst x
1164 case pRes of
1165 PR.Partial 0 pst1 ->
1166 return $ Skip $ ParseChunksStream s [] pst1
1167 PR.Partial n pst1 -> do
1168 assert (n <= length (x:buf)) (return ())
1169 let src0 = Prelude.take n (x:buf)
1170 src = Prelude.reverse src0
1171 return $ Skip $ ParseChunksBuf src s [] pst1
1172 PR.Continue 0 pst1 ->
1173 return $ Skip $ ParseChunksStream s (x:buf) pst1
1174 PR.Continue n pst1 -> do
1175 assert (n <= length (x:buf)) (return ())
1176 let (src0, buf1) = splitAt n (x:buf)
1177 src = Prelude.reverse src0
1178 return $ Skip $ ParseChunksBuf src s buf1 pst1
1179 PR.Done 0 b -> do
1180 return $ Skip $
1181 ParseChunksYield b (ParseChunksInit [] s)
1182 PR.Done n b -> do
1183 assert (n <= length (x:buf)) (return ())
1184 let src = Prelude.reverse (Prelude.take n (x:buf))
1185 return $ Skip $
1186 ParseChunksYield b (ParseChunksInit src s)
1187 PR.Error err -> throwM $ ParseError err
1188 Skip s -> return $ Skip $ ParseChunksStream s buf pst
1189 Stop -> do
1190 b <- extract pst
1191 let src = Prelude.reverse buf
1192 return $ Skip $
1193 ParseChunksYield b (ParseChunksInitLeftOver src)
1195 -- go back to stream processing mode
1196 stepOuter _ (ParseChunksBuf [] s buf pst) =
1197 return $ Skip $ ParseChunksStream s buf pst
1199 -- buffered processing loop
1200 stepOuter _ (ParseChunksBuf (x:xs) s buf pst) = do
1201 pRes <- pstep pst x
1202 case pRes of
1203 PR.Partial 0 pst1 ->
1204 return $ Skip $ ParseChunksBuf xs s [] pst1
1205 PR.Partial n pst1 -> do
1206 assert (n <= length (x:buf)) (return ())
1207 let src0 = Prelude.take n (x:buf)
1208 src = Prelude.reverse src0 ++ xs
1209 return $ Skip $ ParseChunksBuf src s [] pst1
1210 PR.Continue 0 pst1 ->
1211 return $ Skip $ ParseChunksBuf xs s (x:buf) pst1
1212 PR.Continue n pst1 -> do
1213 assert (n <= length (x:buf)) (return ())
1214 let (src0, buf1) = splitAt n (x:buf)
1215 src = Prelude.reverse src0 ++ xs
1216 return $ Skip $ ParseChunksBuf src s buf1 pst1
1217 PR.Done 0 b ->
1218 return $ Skip $ ParseChunksYield b (ParseChunksInit xs s)
1219 PR.Done n b -> do
1220 assert (n <= length (x:buf)) (return ())
1221 let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs
1222 return $ Skip $ ParseChunksYield b (ParseChunksInit src s)
1223 PR.Error err -> throwM $ ParseError err
1225 stepOuter _ (ParseChunksYield a next) = return $ Yield a next
1227{-# ANN type ConcatParseState Fuse #-}
1228data ConcatParseState b inpBuf st p m a =
1229 ConcatParseInit inpBuf st p
1230 | ConcatParseInitLeftOver inpBuf
1231 | forall s. ConcatParseStream st inpBuf (s -> a -> m (PRD.Step s b)) s (s -> m b)
1232 | forall s. ConcatParseBuf inpBuf st inpBuf (s -> a -> m (PRD.Step s b)) s (s -> m b)
1233 | ConcatParseYield b (ConcatParseState b inpBuf st p m a)
1235{-# INLINE_NORMAL parseIterate #-}
1236parseIterate
1237 :: MonadThrow m
1238 => (b -> PRD.Parser m a b)
1239 -> b
1240 -> Stream m a
1241 -> Stream m b
1242parseIterate func seed (Stream step state) =
1243 Stream stepOuter (ConcatParseInit [] state (func seed))
1245 where
1247 {-# INLINE_LATE stepOuter #-}
1248 -- Buffer is empty, go to stream processing loop
1249 stepOuter _ (ConcatParseInit [] st (PRD.Parser pstep initial extract)) = do
1250 res <- initial
1251 case res of
1252 PRD.IPartial ps ->
1253 return $ Skip $ ConcatParseStream st [] pstep ps extract
1254 PRD.IDone pb ->
1255 let next = ConcatParseInit [] st (func pb)
1256 in return $ Skip $ ConcatParseYield pb next
1257 PRD.IError err -> throwM $ ParseError err
1259 -- Buffer is not empty, go to buffered processing loop
1260 stepOuter _ (ConcatParseInit src st
1261 (PRD.Parser pstep initial extract)) = do
1262 res <- initial
1263 case res of
1264 PRD.IPartial ps ->
1265 return $ Skip $ ConcatParseBuf src st [] pstep ps extract
1266 PRD.IDone pb ->
1267 let next = ConcatParseInit src st (func pb)
1268 in return $ Skip $ ConcatParseYield pb next
1269 PRD.IError err -> throwM $ ParseError err
1271 -- XXX we just discard any leftover input at the end
1272 stepOuter _ (ConcatParseInitLeftOver _) = return Stop
1274 -- Buffer is empty process elements from the stream
1275 stepOuter gst (ConcatParseStream st buf pstep pst extract) = do
1276 r <- step (adaptState gst) st
1277 case r of
1278 Yield x s -> do
1279 pRes <- pstep pst x
1280 case pRes of
1281 PR.Partial 0 pst1 ->
1282 return $ Skip $ ConcatParseStream s [] pstep pst1 extract
1283 PR.Partial n pst1 -> do
1284 assert (n <= length (x:buf)) (return ())
1285 let src0 = Prelude.take n (x:buf)
1286 src = Prelude.reverse src0
1287 return $ Skip $ ConcatParseBuf src s [] pstep pst1 extract
1288 -- PR.Continue 0 pst1 ->
1289 -- return $ Skip $ ConcatParseStream s (x:buf) pst1
1290 PR.Continue n pst1 -> do
1291 assert (n <= length (x:buf)) (return ())
1292 let (src0, buf1) = splitAt n (x:buf)
1293 src = Prelude.reverse src0
1294 return $ Skip $ ConcatParseBuf src s buf1 pstep pst1 extract
1295 -- XXX Specialize for Stop 0 common case?
1296 PR.Done n b -> do
1297 assert (n <= length (x:buf)) (return ())
1298 let src = Prelude.reverse (Prelude.take n (x:buf))
1299 return $ Skip $
1300 ConcatParseYield b (ConcatParseInit src s (func b))
1301 PR.Error err -> throwM $ ParseError err
1302 Skip s -> return $ Skip $ ConcatParseStream s buf pstep pst extract
1303 Stop -> do
1304 b <- extract pst
1305 let src = Prelude.reverse buf
1306 return $ Skip $ ConcatParseYield b (ConcatParseInitLeftOver src)
1308 -- go back to stream processing mode
1309 stepOuter _ (ConcatParseBuf [] s buf pstep ps extract) =
1310 return $ Skip $ ConcatParseStream s buf pstep ps extract
1312 -- buffered processing loop
1313 stepOuter _ (ConcatParseBuf (x:xs) s buf pstep pst extract) = do
1314 pRes <- pstep pst x
1315 case pRes of
1316 PR.Partial 0 pst1 ->
1317 return $ Skip $ ConcatParseBuf xs s [] pstep pst1 extract
1318 PR.Partial n pst1 -> do
1319 assert (n <= length (x:buf)) (return ())
1320 let src0 = Prelude.take n (x:buf)
1321 src = Prelude.reverse src0 ++ xs
1322 return $ Skip $ ConcatParseBuf src s [] pstep pst1 extract
1323 -- PR.Continue 0 pst1 -> return $ Skip $ ConcatParseBuf xs s (x:buf) pst1
1324 PR.Continue n pst1 -> do
1325 assert (n <= length (x:buf)) (return ())
1326 let (src0, buf1) = splitAt n (x:buf)
1327 src = Prelude.reverse src0 ++ xs
1328 return $ Skip $ ConcatParseBuf src s buf1 pstep pst1 extract
1329 -- XXX Specialize for Stop 0 common case?
1330 PR.Done n b -> do
1331 assert (n <= length (x:buf)) (return ())
1332 let src = Prelude.reverse (Prelude.take n (x:buf)) ++ xs
1333 return $ Skip $ ConcatParseYield b
1334 (ConcatParseInit src s (func b))
1335 PR.Error err -> throwM $ ParseError err
1337 stepOuter _ (ConcatParseYield a next) = return $ Yield a next
1339------------------------------------------------------------------------------
1340-- Grouping
1341------------------------------------------------------------------------------
1343data GroupByState st fs a b
1344 = GroupingInit st
1345 | GroupingDo st !fs
1346 | GroupingInitWith st !a
1347 | GroupingDoWith st !fs !a
1348 | GroupingYield !b (GroupByState st fs a b)
1349 | GroupingDone
1351{-# INLINE_NORMAL groupsBy #-}
1352groupsBy :: Monad m
1353 => (a -> a -> Bool)
1354 -> Fold m a b
1355 -> Stream m a
1356 -> Stream m b
1358groupsBy eq fld = parseMany (PRD.groupBy eq fld)
1360groupsBy cmp (Fold fstep initial done) (Stream step state) =
1361 Stream stepOuter (GroupingInit state)
1363 where
1365 {-# INLINE_LATE stepOuter #-}
1366 stepOuter _ (GroupingInit st) = do
1367 -- XXX Note that if the stream stops without yielding a single element
1368 -- in the group we discard the "initial" effect.
1369 res <- initial
1370 return
1371 $ case res of
1372 FL.Partial s -> Skip $ GroupingDo st s
1373 FL.Done b -> Yield b $ GroupingInit st
1374 stepOuter gst (GroupingDo st fs) = do
1375 res <- step (adaptState gst) st
1376 case res of
1377 Yield x s -> do
1378 r <- fstep fs x
1379 case r of
1380 FL.Partial fs1 -> go SPEC x s fs1
1381 FL.Done b -> return $ Yield b (GroupingInit s)
1382 Skip s -> return $ Skip $ GroupingDo s fs
1383 Stop -> return Stop
1385 where
1387 go !_ prev stt !acc = do
1388 res <- step (adaptState gst) stt
1389 case res of
1390 Yield x s -> do
1391 if cmp x prev
1392 then do
1393 r <- fstep acc x
1394 case r of
1395 FL.Partial fs1 -> go SPEC prev s fs1
1396 FL.Done b -> return $ Yield b (GroupingInit s)
1397 else do
1398 r <- done acc
1399 return $ Yield r (GroupingInitWith s x)
1400 Skip s -> go SPEC prev s acc
1401 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1402 stepOuter _ (GroupingInitWith st x) = do
1403 res <- initial
1404 return
1405 $ case res of
1406 FL.Partial s -> Skip $ GroupingDoWith st s x
1407 FL.Done b -> Yield b $ GroupingInitWith st x
1408 stepOuter gst (GroupingDoWith st fs prev) = do
1409 res <- fstep fs prev
1410 case res of
1411 FL.Partial fs1 -> go SPEC st fs1
1412 FL.Done b -> return $ Yield b (GroupingInit st)
1414 where
1416 -- XXX code duplicated from the previous equation
1417 go !_ stt !acc = do
1418 res <- step (adaptState gst) stt
1419 case res of
1420 Yield x s -> do
1421 if cmp x prev
1422 then do
1423 r <- fstep acc x
1424 case r of
1425 FL.Partial fs1 -> go SPEC s fs1
1426 FL.Done b -> return $ Yield b (GroupingInit s)
1427 else do
1428 r <- done acc
1429 return $ Yield r (GroupingInitWith s x)
1430 Skip s -> go SPEC s acc
1431 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1432 stepOuter _ (GroupingYield _ _) = error "groupsBy: Unreachable"
1433 stepOuter _ GroupingDone = return Stop
1435{-# INLINE_NORMAL groupsRollingBy #-}
1436groupsRollingBy :: Monad m
1437 => (a -> a -> Bool)
1438 -> Fold m a b
1439 -> Stream m a
1440 -> Stream m b
1442groupsRollingBy eq fld = parseMany (PRD.groupByRolling eq fld)
1444groupsRollingBy cmp (Fold fstep initial done) (Stream step state) =
1445 Stream stepOuter (GroupingInit state)
1447 where
1449 {-# INLINE_LATE stepOuter #-}
1450 stepOuter _ (GroupingInit st) = do
1451 -- XXX Note that if the stream stops without yielding a single element
1452 -- in the group we discard the "initial" effect.
1453 res <- initial
1454 return
1455 $ case res of
1456 FL.Partial fs -> Skip $ GroupingDo st fs
1457 FL.Done fb -> Yield fb $ GroupingInit st
1458 stepOuter gst (GroupingDo st fs) = do
1459 res <- step (adaptState gst) st
1460 case res of
1461 Yield x s -> do
1462 r <- fstep fs x
1463 case r of
1464 FL.Partial fs1 -> go SPEC x s fs1
1465 FL.Done fb -> return $ Yield fb (GroupingInit s)
1466 Skip s -> return $ Skip $ GroupingDo s fs
1467 Stop -> return Stop
1469 where
1471 go !_ prev stt !acc = do
1472 res <- step (adaptState gst) stt
1473 case res of
1474 Yield x s -> do
1475 if cmp prev x
1476 then do
1477 r <- fstep acc x
1478 case r of
1479 FL.Partial fs1 -> go SPEC x s fs1
1480 FL.Done b -> return $ Yield b (GroupingInit s)
1481 else do
1482 r <- done acc
1483 return $ Yield r (GroupingInitWith s x)
1484 Skip s -> go SPEC prev s acc
1485 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1486 stepOuter _ (GroupingInitWith st x) = do
1487 res <- initial
1488 return
1489 $ case res of
1490 FL.Partial s -> Skip $ GroupingDoWith st s x
1491 FL.Done b -> Yield b $ GroupingInitWith st x
1492 stepOuter gst (GroupingDoWith st fs previous) = do
1493 res <- fstep fs previous
1494 case res of
1495 FL.Partial s -> go SPEC previous st s
1496 FL.Done b -> return $ Yield b (GroupingInit st)
1498 where
1500 -- XXX GHC: groupsBy has one less parameter in this go loop and it
1501 -- fuses. However, groupsRollingBy does not fuse, removing the prev
1502 -- parameter makes it fuse. Something needs to be fixed in GHC. The
1503 -- workaround for this is noted in the comments below.
1504 go !_ prev !stt !acc = do
1505 res <- step (adaptState gst) stt
1506 case res of
1507 Yield x s -> do
1508 if cmp prev x
1509 then do
1510 r <- fstep acc x
1511 case r of
1512 FL.Partial fs1 -> go SPEC x s fs1
1513 FL.Done b -> return $ Yield b (GroupingInit st)
1514 else do
1515 {-
1516 r <- done acc
1517 return $ Yield r (GroupingInitWith s x)
1518 -}
1519 -- The code above does not let groupBy fuse. We use the
1520 -- alternative code below instead. Instead of jumping
1521 -- to GroupingInitWith state, we unroll the code of
1522 -- GroupingInitWith state here to help GHC with stream
1523 -- fusion.
1524 result <- initial
1525 r <- done acc
1526 return
1527 $ Yield r
1528 $ case result of
1529 FL.Partial fsi -> GroupingDoWith s fsi x
1530 FL.Done b -> GroupingYield b (GroupingInit s)
1531 Skip s -> go SPEC prev s acc
1532 Stop -> done acc >>= \r -> return $ Yield r GroupingDone
1533 stepOuter _ (GroupingYield r next) = return $ Yield r next
1534 stepOuter _ GroupingDone = return Stop
1536------------------------------------------------------------------------------
1537-- Splitting - by a predicate
1538------------------------------------------------------------------------------
1540data WordsByState st fs b
1541 = WordsByInit st
1542 | WordsByDo st !fs
1543 | WordsByDone
1544 | WordsByYield !b (WordsByState st fs b)
1546{-# INLINE_NORMAL wordsBy #-}
1547wordsBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1548wordsBy predicate (Fold fstep initial done) (Stream step state) =
1549 Stream stepOuter (WordsByInit state)
1551 where
1553 {-# INLINE_LATE stepOuter #-}
1554 stepOuter _ (WordsByInit st) = do
1555 res <- initial
1556 return
1557 $ case res of
1558 FL.Partial s -> Skip $ WordsByDo st s
1559 FL.Done b -> Yield b (WordsByInit st)
1561 stepOuter gst (WordsByDo st fs) = do
1562 res <- step (adaptState gst) st
1563 case res of
1564 Yield x s -> do
1565 if predicate x
1566 then do
1567 resi <- initial
1568 return
1569 $ case resi of
1570 FL.Partial fs1 -> Skip $ WordsByDo s fs1
1571 FL.Done b -> Yield b (WordsByInit s)
1572 else do
1573 r <- fstep fs x
1574 case r of
1575 FL.Partial fs1 -> go SPEC s fs1
1576 FL.Done b -> return $ Yield b (WordsByInit s)
1577 Skip s -> return $ Skip $ WordsByDo s fs
1578 Stop -> return Stop
1580 where
1582 go !_ stt !acc = do
1583 res <- step (adaptState gst) stt
1584 case res of
1585 Yield x s -> do
1586 if predicate x
1587 then do
1588 {-
1589 r <- done acc
1590 return $ Yield r (WordsByInit s)
1591 -}
1592 -- The above code does not fuse well. Need to check why
1593 -- GHC is not able to simplify it well. Using the code
1594 -- below, instead of jumping through the WordsByInit
1595 -- state always, we directly go to WordsByDo state in
1596 -- the common case of Partial.
1597 resi <- initial
1598 r <- done acc
1599 return
1600 $ Yield r
1601 $ case resi of
1602 FL.Partial fs1 -> WordsByDo s fs1
1603 FL.Done b -> WordsByYield b (WordsByInit s)
1604 else do
1605 r <- fstep acc x
1606 case r of
1607 FL.Partial fs1 -> go SPEC s fs1
1608 FL.Done b -> return $ Yield b (WordsByInit s)
1609 Skip s -> go SPEC s acc
1610 Stop -> done acc >>= \r -> return $ Yield r WordsByDone
1612 stepOuter _ WordsByDone = return Stop
1614 stepOuter _ (WordsByYield b next) = return $ Yield b next
1616------------------------------------------------------------------------------
1617-- Splitting on a sequence
1618------------------------------------------------------------------------------
1620-- String search algorithms:
1621-- http://www-igm.univ-mlv.fr/~lecroq/string/index.html
1624-- TODO can we unify the splitting operations using a splitting configuration
1625-- like in the split package.
1627data SplitStyle = Infix | Suffix | Prefix deriving (Eq, Show)
1628data SplitOptions = SplitOptions
1629 { style :: SplitStyle
1630 , withSep :: Bool -- ^ keep the separators in output
1631 -- , compact :: Bool -- ^ treat multiple consecutive separators as one
1632 -- , trimHead :: Bool -- ^ drop blank at head
1633 -- , trimTail :: Bool -- ^ drop blank at tail
1637-- XXX using "fs" as the last arg in Constructors may simplify the code a bit,
1638-- because we can use the constructor directly without having to create "jump"
1639-- functions.
1640{-# ANN type SplitOnSeqState Fuse #-}
1641data SplitOnSeqState rb rh ck w fs s b x =
1642 SplitOnSeqInit
1643 | SplitOnSeqYield b (SplitOnSeqState rb rh ck w fs s b x)
1644 | SplitOnSeqDone
1646 | SplitOnSeqEmpty !fs s
1648 | SplitOnSeqSingle !fs s x
1650 | SplitOnSeqWordInit !fs s
1651 | SplitOnSeqWordLoop !w s !fs
1652 | SplitOnSeqWordDone Int !fs !w
1654 | SplitOnSeqKRInit Int !fs s rb !rh
1655 | SplitOnSeqKRLoop fs s rb !rh !ck
1656 | SplitOnSeqKRCheck fs s rb !rh
1657 | SplitOnSeqKRDone Int !fs rb !rh
1659 | SplitOnSeqReinit (fs -> SplitOnSeqState rb rh ck w fs s b x)
1661{-# INLINE_NORMAL splitOnSeq #-}
1662splitOnSeq
1663 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1664 => Array a
1665 -> Fold m a b
1666 -> Stream m a
1667 -> Stream m b
1668splitOnSeq patArr (Fold fstep initial done) (Stream step state) =
1669 Stream stepOuter SplitOnSeqInit
1671 where
1673 patLen = A.length patArr
1674 maxIndex = patLen - 1
1675 elemBits = SIZE_OF(a) * 8
1677 -- For word pattern case
1678 wordMask :: Word
1679 wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1681 elemMask :: Word
1682 elemMask = (1 `shiftL` elemBits) - 1
1684 wordPat :: Word
1685 wordPat = wordMask .&. A.foldl' addToWord 0 patArr
1687 addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1689 -- For Rabin-Karp search
1690 k = 2891336453 :: Word32
1691 coeff = k ^ patLen
1693 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1695 deltaCksum cksum old new =
1696 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1698 -- XXX shall we use a random starting hash or 1 instead of 0?
1699 patHash = A.foldl' addCksum 0 patArr
1701 skip = return . Skip
1703 nextAfterInit nextGen stepRes =
1704 case stepRes of
1705 FL.Partial s -> nextGen s
1706 FL.Done b -> SplitOnSeqYield b (SplitOnSeqReinit nextGen)
1708 {-# INLINE yieldProceed #-}
1709 yieldProceed nextGen fs =
1710 initial >>= skip . SplitOnSeqYield fs . nextAfterInit nextGen
1712 {-# INLINE_LATE stepOuter #-}
1713 stepOuter _ SplitOnSeqInit = do
1714 res <- initial
1715 case res of
1716 FL.Partial acc ->
1717 if patLen == 0
1718 then return $ Skip $ SplitOnSeqEmpty acc state
1719 else if patLen == 1
1720 then do
1721 pat <- liftIO $ A.unsafeIndexIO 0 patArr
1722 return $ Skip $ SplitOnSeqSingle acc state pat
1723 else if SIZE_OF(a) * patLen
1724 <= sizeOf (undefined :: Word)
1725 then return $ Skip $ SplitOnSeqWordInit acc state
1726 else do
1727 (rb, rhead) <- liftIO $ RB.new patLen
1728 skip $ SplitOnSeqKRInit 0 acc state rb rhead
1729 FL.Done b -> skip $ SplitOnSeqYield b SplitOnSeqInit
1731 stepOuter _ (SplitOnSeqYield x next) = return $ Yield x next
1733 ---------------------------
1734 -- Checkpoint
1735 ---------------------------
1737 stepOuter _ (SplitOnSeqReinit nextGen) =
1738 initial >>= skip . nextAfterInit nextGen
1740 ---------------------------
1741 -- Empty pattern
1742 ---------------------------
1744 stepOuter gst (SplitOnSeqEmpty acc st) = do
1745 res <- step (adaptState gst) st
1746 case res of
1747 Yield x s -> do
1748 r <- fstep acc x
1749 b1 <-
1750 case r of
1751 FL.Partial acc1 -> done acc1
1752 FL.Done b -> return b
1753 let jump c = SplitOnSeqEmpty c s
1754 in yieldProceed jump b1
1755 Skip s -> skip (SplitOnSeqEmpty acc s)
1756 Stop -> return Stop
1758 -----------------
1759 -- Done
1760 -----------------
1762 stepOuter _ SplitOnSeqDone = return Stop
1764 -----------------
1765 -- Single Pattern
1766 -----------------
1768 stepOuter gst (SplitOnSeqSingle fs st pat) = do
1769 res <- step (adaptState gst) st
1770 case res of
1771 Yield x s -> do
1772 let jump c = SplitOnSeqSingle c s pat
1773 if pat == x
1774 then done fs >>= yieldProceed jump
1775 else do
1776 r <- fstep fs x
1777 case r of
1778 FL.Partial fs1 -> skip $ jump fs1
1779 FL.Done b -> yieldProceed jump b
1780 Skip s -> return $ Skip $ SplitOnSeqSingle fs s pat
1781 Stop -> do
1782 r <- done fs
1783 return $ Skip $ SplitOnSeqYield r SplitOnSeqDone
1785 ---------------------------
1786 -- Short Pattern - Shift Or
1787 ---------------------------
1789 stepOuter _ (SplitOnSeqWordDone 0 fs _) = do
1790 r <- done fs
1791 skip $ SplitOnSeqYield r SplitOnSeqDone
1792 stepOuter _ (SplitOnSeqWordDone n fs wrd) = do
1793 let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
1794 r <- fstep fs (toEnum $ fromIntegral old)
1795 case r of
1796 FL.Partial fs1 -> skip $ SplitOnSeqWordDone (n - 1) fs1 wrd
1797 FL.Done b -> do
1798 let jump c = SplitOnSeqWordDone (n - 1) c wrd
1799 yieldProceed jump b
1801 stepOuter gst (SplitOnSeqWordInit fs st0) =
1802 go SPEC 0 0 st0
1804 where
1806 {-# INLINE go #-}
1807 go !_ !idx !wrd !st = do
1808 res <- step (adaptState gst) st
1809 case res of
1810 Yield x s -> do
1811 let wrd1 = addToWord wrd x
1812 if idx == maxIndex
1813 then do
1814 if wrd1 .&. wordMask == wordPat
1815 then do
1816 let jump c = SplitOnSeqWordInit c s
1817 done fs >>= yieldProceed jump
1818 else skip $ SplitOnSeqWordLoop wrd1 s fs
1819 else go SPEC (idx + 1) wrd1 s
1820 Skip s -> go SPEC idx wrd s
1821 Stop -> do
1822 if idx /= 0
1823 then skip $ SplitOnSeqWordDone idx fs wrd
1824 else do
1825 r <- done fs
1826 skip $ SplitOnSeqYield r SplitOnSeqDone
1828 stepOuter gst (SplitOnSeqWordLoop wrd0 st0 fs0) =
1829 go SPEC wrd0 st0 fs0
1831 where
1833 {-# INLINE go #-}
1834 go !_ !wrd !st !fs = do
1835 res <- step (adaptState gst) st
1836 case res of
1837 Yield x s -> do
1838 let jump c = SplitOnSeqWordInit c s
1839 wrd1 = addToWord wrd x
1840 old = (wordMask .&. wrd)
1841 `shiftR` (elemBits * (patLen - 1))
1842 r <- fstep fs (toEnum $ fromIntegral old)
1843 case r of
1844 FL.Partial fs1 -> do
1845 if wrd1 .&. wordMask == wordPat
1846 then done fs1 >>= yieldProceed jump
1847 else go SPEC wrd1 s fs1
1848 FL.Done b -> yieldProceed jump b
1849 Skip s -> go SPEC wrd s fs
1850 Stop -> skip $ SplitOnSeqWordDone patLen fs wrd
1852 -------------------------------
1853 -- General Pattern - Karp Rabin
1854 -------------------------------
1856 stepOuter gst (SplitOnSeqKRInit idx fs st rb rh) = do
1857 res <- step (adaptState gst) st
1858 case res of
1859 Yield x s -> do
1860 rh1 <- liftIO $ RB.unsafeInsert rb rh x
1861 if idx == maxIndex
1862 then do
1863 let fld = RB.unsafeFoldRing (RB.ringBound rb)
1864 let !ringHash = fld addCksum 0 rb
1865 if ringHash == patHash
1866 then skip $ SplitOnSeqKRCheck fs s rb rh1
1867 else skip $ SplitOnSeqKRLoop fs s rb rh1 ringHash
1868 else skip $ SplitOnSeqKRInit (idx + 1) fs s rb rh1
1869 Skip s -> skip $ SplitOnSeqKRInit idx fs s rb rh
1870 Stop -> do
1871 skip $ SplitOnSeqKRDone idx fs rb (RB.startOf rb)
1873 -- XXX The recursive "go" is more efficient than the state based recursion
1874 -- code commented out below. Perhaps its more efficient because of
1875 -- factoring out "rb" outside the loop.
1876 --
1877 stepOuter gst (SplitOnSeqKRLoop fs0 st0 rb rh0 cksum0) =
1878 go SPEC fs0 st0 rh0 cksum0
1880 where
1882 go !_ !fs !st !rh !cksum = do
1883 res <- step (adaptState gst) st
1884 case res of
1885 Yield x s -> do
1886 old <- liftIO $ peek rh
1887 let cksum1 = deltaCksum cksum old x
1888 r <- fstep fs old
1889 case r of
1890 FL.Partial fs1 -> do
1891 rh1 <- liftIO (RB.unsafeInsert rb rh x)
1892 if cksum1 == patHash
1893 then skip $ SplitOnSeqKRCheck fs1 s rb rh1
1894 else go SPEC fs1 s rh1 cksum1
1895 FL.Done b -> do
1896 let rst = RB.startOf rb
1897 jump c = SplitOnSeqKRInit 0 c s rb rst
1898 yieldProceed jump b
1899 Skip s -> go SPEC fs s rh cksum
1900 Stop -> skip $ SplitOnSeqKRDone patLen fs rb rh
1902 -- XXX The following code is 5 times slower compared to the recursive loop
1903 -- based code above. Need to investigate why. One possibility is that the
1904 -- go loop above does not thread around the ring buffer (rb). This code may
1905 -- be causing the state to bloat and getting allocated on each iteration.
1906 -- We can check the cmm/asm code to confirm. If so a good GHC solution to
1907 -- such problem is needed. One way to avoid this could be to use unboxed
1908 -- mutable state?
1909 {-
1910 stepOuter gst (SplitOnSeqKRLoop fs st rb rh cksum) = do
1911 res <- step (adaptState gst) st
1912 case res of
1913 Yield x s -> do
1914 old <- liftIO $ peek rh
1915 let cksum1 = deltaCksum cksum old x
1916 fs1 <- fstep fs old
1917 if (cksum1 == patHash)
1918 then do
1919 r <- done fs1
1920 skip $ SplitOnSeqYield r $ SplitOnSeqKRInit 0 s rb rh
1921 else do
1922 rh1 <- liftIO (RB.unsafeInsert rb rh x)
1923 skip $ SplitOnSeqKRLoop fs1 s rb rh1 cksum1
1924 Skip s -> skip $ SplitOnSeqKRLoop fs s rb rh cksum
1925 Stop -> skip $ SplitOnSeqKRDone patLen fs rb rh
1926 -}
1928 stepOuter _ (SplitOnSeqKRCheck fs st rb rh) = do
1929 if RB.unsafeEqArray rb rh patArr
1930 then do
1931 r <- done fs
1932 let rst = RB.startOf rb
1933 jump c = SplitOnSeqKRInit 0 c st rb rst
1934 yieldProceed jump r
1935 else skip $ SplitOnSeqKRLoop fs st rb rh patHash
1937 stepOuter _ (SplitOnSeqKRDone 0 fs _ _) = do
1938 r <- done fs
1939 skip $ SplitOnSeqYield r SplitOnSeqDone
1940 stepOuter _ (SplitOnSeqKRDone n fs rb rh) = do
1941 old <- liftIO $ peek rh
1942 let rh1 = RB.advance rb rh
1943 r <- fstep fs old
1944 case r of
1945 FL.Partial fs1 -> skip $ SplitOnSeqKRDone (n - 1) fs1 rb rh1
1946 FL.Done b -> do
1947 let jump c = SplitOnSeqKRDone (n - 1) c rb rh1
1948 yieldProceed jump b
1950{-# ANN type SplitOnSuffixSeqState Fuse #-}
1951data SplitOnSuffixSeqState rb rh ck w fs s b x =
1952 SplitOnSuffixSeqInit
1953 | SplitOnSuffixSeqYield b (SplitOnSuffixSeqState rb rh ck w fs s b x)
1954 | SplitOnSuffixSeqDone
1956 | SplitOnSuffixSeqEmpty !fs s
1958 | SplitOnSuffixSeqSingleInit !fs s x
1959 | SplitOnSuffixSeqSingle !fs s x
1961 | SplitOnSuffixSeqWordInit !fs s
1962 | SplitOnSuffixSeqWordLoop !w s !fs
1963 | SplitOnSuffixSeqWordDone Int !fs !w
1965 | SplitOnSuffixSeqKRInit Int !fs s rb !rh
1966 | SplitOnSuffixSeqKRInit1 !fs s rb !rh
1967 | SplitOnSuffixSeqKRLoop fs s rb !rh !ck
1968 | SplitOnSuffixSeqKRCheck fs s rb !rh
1969 | SplitOnSuffixSeqKRDone Int !fs rb !rh
1971 | SplitOnSuffixSeqReinit
1972 (fs -> SplitOnSuffixSeqState rb rh ck w fs s b x)
1974{-# INLINE_NORMAL splitOnSuffixSeq #-}
1975splitOnSuffixSeq
1976 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1977 => Bool
1978 -> Array a
1979 -> Fold m a b
1980 -> Stream m a
1981 -> Stream m b
1982splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =
1983 Stream stepOuter SplitOnSuffixSeqInit
1985 where
1987 patLen = A.length patArr
1988 maxIndex = patLen - 1
1989 elemBits = SIZE_OF(a) * 8
1991 -- For word pattern case
1992 wordMask :: Word
1993 wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1995 elemMask :: Word
1996 elemMask = (1 `shiftL` elemBits) - 1
1998 wordPat :: Word
1999 wordPat = wordMask .&. A.foldl' addToWord 0 patArr
2001 addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
2003 nextAfterInit nextGen stepRes =
2004 case stepRes of
2005 FL.Partial s -> nextGen s
2006 FL.Done b ->
2007 SplitOnSuffixSeqYield b (SplitOnSuffixSeqReinit nextGen)
2009 {-# INLINE yieldProceed #-}
2010 yieldProceed nextGen fs =
2011 initial >>= skip . SplitOnSuffixSeqYield fs . nextAfterInit nextGen
2013 -- For single element pattern case
2014 {-# INLINE processYieldSingle #-}
2015 processYieldSingle pat x s fs = do
2016 let jump c = SplitOnSuffixSeqSingleInit c s pat
2017 if pat == x
2018 then do
2019 r <- if withSep then fstep fs x else return $ FL.Partial fs
2020 b1 <-
2021 case r of
2022 FL.Partial fs1 -> done fs1
2023 FL.Done b -> return b
2024 yieldProceed jump b1
2025 else do
2026 r <- fstep fs x
2027 case r of
2028 FL.Partial fs1 -> skip $ SplitOnSuffixSeqSingle fs1 s pat
2029 FL.Done b -> yieldProceed jump b
2031 -- For Rabin-Karp search
2032 k = 2891336453 :: Word32
2033 coeff = k ^ patLen
2035 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
2037 deltaCksum cksum old new =
2038 addCksum cksum new - coeff * fromIntegral (fromEnum old)
2040 -- XXX shall we use a random starting hash or 1 instead of 0?
2041 patHash = A.foldl' addCksum 0 patArr
2043 skip = return . Skip
2045 {-# INLINE_LATE stepOuter #-}
2046 stepOuter _ SplitOnSuffixSeqInit = do
2047 res <- initial
2048 case res of
2049 FL.Partial fs ->
2050 if patLen == 0
2051 then skip $ SplitOnSuffixSeqEmpty fs state
2052 else if patLen == 1
2053 then do
2054 pat <- liftIO $ A.unsafeIndexIO 0 patArr
2055 skip $ SplitOnSuffixSeqSingleInit fs state pat
2056 else if SIZE_OF(a) * patLen
2057 <= sizeOf (undefined :: Word)
2058 then skip $ SplitOnSuffixSeqWordInit fs state
2059 else do
2060 (rb, rhead) <- liftIO $ RB.new patLen
2061 skip $ SplitOnSuffixSeqKRInit 0 fs state rb rhead
2062 FL.Done fb -> skip $ SplitOnSuffixSeqYield fb SplitOnSuffixSeqInit
2064 stepOuter _ (SplitOnSuffixSeqYield x next) = return $ Yield x next
2066 ---------------------------
2067 -- Reinit
2068 ---------------------------
2070 stepOuter _ (SplitOnSuffixSeqReinit nextGen) =
2071 initial >>= skip . nextAfterInit nextGen
2073 ---------------------------
2074 -- Empty pattern
2075 ---------------------------
2077 stepOuter gst (SplitOnSuffixSeqEmpty acc st) = do
2078 res <- step (adaptState gst) st
2079 case res of
2080 Yield x s -> do
2081 let jump c = SplitOnSuffixSeqEmpty c s
2082 r <- fstep acc x
2083 b1 <-
2084 case r of
2085 FL.Partial fs -> done fs
2086 FL.Done b -> return b
2087 yieldProceed jump b1
2088 Skip s -> skip (SplitOnSuffixSeqEmpty acc s)
2089 Stop -> return Stop
2091 -----------------
2092 -- Done
2093 -----------------
2095 stepOuter _ SplitOnSuffixSeqDone = return Stop
2097 -----------------
2098 -- Single Pattern
2099 -----------------
2101 stepOuter gst (SplitOnSuffixSeqSingleInit fs st pat) = do
2102 res <- step (adaptState gst) st
2103 case res of
2104 Yield x s -> processYieldSingle pat x s fs
2105 Skip s -> skip $ SplitOnSuffixSeqSingleInit fs s pat
2106 Stop -> return Stop
2108 stepOuter gst (SplitOnSuffixSeqSingle fs st pat) = do
2109 res <- step (adaptState gst) st
2110 case res of
2111 Yield x s -> processYieldSingle pat x s fs
2112 Skip s -> skip $ SplitOnSuffixSeqSingle fs s pat
2113 Stop -> do
2114 r <- done fs
2115 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2117 ---------------------------
2118 -- Short Pattern - Shift Or
2119 ---------------------------
2121 stepOuter _ (SplitOnSuffixSeqWordDone 0 fs _) = do
2122 r <- done fs
2123 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2124 stepOuter _ (SplitOnSuffixSeqWordDone n fs wrd) = do
2125 let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
2126 r <- fstep fs (toEnum $ fromIntegral old)
2127 case r of
2128 FL.Partial fs1 -> skip $ SplitOnSuffixSeqWordDone (n - 1) fs1 wrd
2129 FL.Done b -> do
2130 let jump c = SplitOnSuffixSeqWordDone (n - 1) c wrd
2131 yieldProceed jump b
2133 stepOuter gst (SplitOnSuffixSeqWordInit fs0 st0) = do
2134 res <- step (adaptState gst) st0
2135 case res of
2136 Yield x s -> do
2137 let wrd = addToWord 0 x
2138 r <- if withSep then fstep fs0 x else return $ FL.Partial fs0
2139 case r of
2140 FL.Partial fs1 -> go SPEC 1 wrd s fs1
2141 FL.Done b -> do
2142 let jump c = SplitOnSuffixSeqWordInit c s
2143 yieldProceed jump b
2144 Skip s -> skip (SplitOnSuffixSeqWordInit fs0 s)
2145 Stop -> return Stop
2147 where
2149 {-# INLINE go #-}
2150 go !_ !idx !wrd !st !fs = do
2151 res <- step (adaptState gst) st
2152 case res of
2153 Yield x s -> do
2154 let jump c = SplitOnSuffixSeqWordInit c s
2155 let wrd1 = addToWord wrd x
2156 r <- if withSep then fstep fs x else return $ FL.Partial fs
2157 case r of
2158 FL.Partial fs1 ->
2159 if idx /= maxIndex
2160 then go SPEC (idx + 1) wrd1 s fs1
2161 else if wrd1 .&. wordMask /= wordPat
2162 then skip $ SplitOnSuffixSeqWordLoop wrd1 s fs1
2163 else do done fs >>= yieldProceed jump
2164 FL.Done b -> yieldProceed jump b
2165 Skip s -> go SPEC idx wrd s fs
2166 Stop -> skip $ SplitOnSuffixSeqWordDone idx fs wrd
2168 stepOuter gst (SplitOnSuffixSeqWordLoop wrd0 st0 fs0) =
2169 go SPEC wrd0 st0 fs0
2171 where
2173 {-# INLINE go #-}
2174 go !_ !wrd !st !fs = do
2175 res <- step (adaptState gst) st
2176 case res of
2177 Yield x s -> do
2178 let jump c = SplitOnSuffixSeqWordInit c s
2179 wrd1 = addToWord wrd x
2180 old = (wordMask .&. wrd)
2181 `shiftR` (elemBits * (patLen - 1))
2182 r <-
2183 if withSep
2184 then fstep fs x
2185 else fstep fs (toEnum $ fromIntegral old)
2186 case r of
2187 FL.Partial fs1 ->
2188 if wrd1 .&. wordMask == wordPat
2189 then done fs1 >>= yieldProceed jump
2190 else go SPEC wrd1 s fs1
2191 FL.Done b -> yieldProceed jump b
2192 Skip s -> go SPEC wrd s fs
2193 Stop ->
2194 if wrd .&. wordMask == wordPat
2195 then return Stop
2196 else if withSep
2197 then do
2198 r <- done fs
2199 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2200 else skip $ SplitOnSuffixSeqWordDone patLen fs wrd
2202 -------------------------------
2203 -- General Pattern - Karp Rabin
2204 -------------------------------
2206 stepOuter gst (SplitOnSuffixSeqKRInit idx0 fs st0 rb rh0) = do
2207 res <- step (adaptState gst) st0
2208 case res of
2209 Yield x s -> do
2210 rh1 <- liftIO $ RB.unsafeInsert rb rh0 x
2211 r <- if withSep then fstep fs x else return $ FL.Partial fs
2212 case r of
2213 FL.Partial fs1 ->
2214 skip $ SplitOnSuffixSeqKRInit1 fs1 s rb rh1
2215 FL.Done b -> do
2216 let rst = RB.startOf rb
2217 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2218 yieldProceed jump b
2219 Skip s -> skip $ SplitOnSuffixSeqKRInit idx0 fs s rb rh0
2220 Stop -> return Stop
2222 stepOuter gst (SplitOnSuffixSeqKRInit1 fs0 st0 rb rh0) = do
2223 go SPEC 1 rh0 st0 fs0
2225 where
2227 go !_ !idx !rh st !fs = do
2228 res <- step (adaptState gst) st
2229 case res of
2230 Yield x s -> do
2231 rh1 <- liftIO (RB.unsafeInsert rb rh x)
2232 r <- if withSep then fstep fs x else return $ FL.Partial fs
2233 case r of
2234 FL.Partial fs1 ->
2235 if idx /= maxIndex
2236 then go SPEC (idx + 1) rh1 s fs1
2237 else skip $
2238 let fld = RB.unsafeFoldRing (RB.ringBound rb)
2239 !ringHash = fld addCksum 0 rb
2240 in if ringHash == patHash
2241 then SplitOnSuffixSeqKRCheck fs1 s rb rh1
2242 else SplitOnSuffixSeqKRLoop
2243 fs1 s rb rh1 ringHash
2244 FL.Done b -> do
2245 let rst = RB.startOf rb
2246 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2247 yieldProceed jump b
2248 Skip s -> go SPEC idx rh s fs
2249 Stop -> do
2250 -- do not issue a blank segment when we end at pattern
2251 if (idx == maxIndex) && RB.unsafeEqArray rb rh patArr
2252 then return Stop
2253 else if withSep
2254 then do
2255 r <- done fs
2256 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2257 else skip $ SplitOnSuffixSeqKRDone idx fs rb (RB.startOf rb)
2259 stepOuter gst (SplitOnSuffixSeqKRLoop fs0 st0 rb rh0 cksum0) =
2260 go SPEC fs0 st0 rh0 cksum0
2262 where
2264 go !_ !fs !st !rh !cksum = do
2265 res <- step (adaptState gst) st
2266 case res of
2267 Yield x s -> do
2268 old <- liftIO $ peek rh
2269 rh1 <- liftIO (RB.unsafeInsert rb rh x)
2270 let cksum1 = deltaCksum cksum old x
2271 r <- if withSep then fstep fs x else fstep fs old
2272 case r of
2273 FL.Partial fs1 ->
2274 if cksum1 /= patHash
2275 then go SPEC fs1 s rh1 cksum1
2276 else skip $ SplitOnSuffixSeqKRCheck fs1 s rb rh1
2277 FL.Done b -> do
2278 let rst = RB.startOf rb
2279 jump c = SplitOnSuffixSeqKRInit 0 c s rb rst
2280 yieldProceed jump b
2281 Skip s -> go SPEC fs s rh cksum
2282 Stop ->
2283 if RB.unsafeEqArray rb rh patArr
2284 then return Stop
2285 else if withSep
2286 then do
2287 r <- done fs
2288 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2289 else skip $ SplitOnSuffixSeqKRDone patLen fs rb rh
2291 stepOuter _ (SplitOnSuffixSeqKRCheck fs st rb rh) = do
2292 if RB.unsafeEqArray rb rh patArr
2293 then do
2294 r <- done fs
2295 let rst = RB.startOf rb
2296 jump c = SplitOnSuffixSeqKRInit 0 c st rb rst
2297 yieldProceed jump r
2298 else skip $ SplitOnSuffixSeqKRLoop fs st rb rh patHash
2300 stepOuter _ (SplitOnSuffixSeqKRDone 0 fs _ _) = do
2301 r <- done fs
2302 skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone
2303 stepOuter _ (SplitOnSuffixSeqKRDone n fs rb rh) = do
2304 old <- liftIO $ peek rh
2305 let rh1 = RB.advance rb rh
2306 r <- fstep fs old
2307 case r of
2308 FL.Partial fs1 -> skip $ SplitOnSuffixSeqKRDone (n - 1) fs1 rb rh1
2309 FL.Done b -> do
2310 let jump c = SplitOnSuffixSeqKRDone (n - 1) c rb rh1
2311 yieldProceed jump b
2313------------------------------------------------------------------------------
2314-- Nested Container Transformation
2315------------------------------------------------------------------------------
2317{-# ANN type SplitState Fuse #-}
2318data SplitState s arr
2319 = SplitInitial s
2320 | SplitBuffering s arr
2321 | SplitSplitting s arr
2322 | SplitYielding arr (SplitState s arr)
2323 | SplitFinishing
2325-- XXX An alternative approach would be to use a partial fold (Fold m a b) to
2326-- split using a splitBy like combinator. The Fold would consume upto the
2327-- separator and return any leftover which can then be fed to the next fold.
2329-- We can revisit this once we have partial folds/parsers.
2331-- | Performs infix separator style splitting.
2332{-# INLINE_NORMAL splitInnerBy #-}
2333splitInnerBy
2334 :: Monad m
2335 => (f a -> m (f a, Maybe (f a))) -- splitter
2336 -> (f a -> f a -> m (f a)) -- joiner
2337 -> Stream m (f a)
2338 -> Stream m (f a)
2339splitInnerBy splitter joiner (Stream step1 state1) =
2340 Stream step (SplitInitial state1)
2342 where
2344 {-# INLINE_LATE step #-}
2345 step gst (SplitInitial st) = do
2346 r <- step1 gst st
2347 case r of
2348 Yield x s -> do
2349 (x1, mx2) <- splitter x
2350 return $ case mx2 of
2351 Nothing -> Skip (SplitBuffering s x1)
2352 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
2353 Skip s -> return $ Skip (SplitInitial s)
2354 Stop -> return Stop
2356 step gst (SplitBuffering st buf) = do
2357 r <- step1 gst st
2358 case r of
2359 Yield x s -> do
2360 (x1, mx2) <- splitter x
2361 buf' <- joiner buf x1
2362 return $ case mx2 of
2363 Nothing -> Skip (SplitBuffering s buf')
2364 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
2365 Skip s -> return $ Skip (SplitBuffering s buf)
2366 Stop -> return $ Skip (SplitYielding buf SplitFinishing)
2368 step _ (SplitSplitting st buf) = do
2369 (x1, mx2) <- splitter buf
2370 return $ case mx2 of
2371 Nothing -> Skip $ SplitBuffering st x1
2372 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
2374 step _ (SplitYielding x next) = return $ Yield x next
2375 step _ SplitFinishing = return Stop
2377-- | Performs infix separator style splitting.
2378{-# INLINE_NORMAL splitInnerBySuffix #-}
2379splitInnerBySuffix
2380 :: (Monad m, Eq (f a), Monoid (f a))
2381 => (f a -> m (f a, Maybe (f a))) -- splitter
2382 -> (f a -> f a -> m (f a)) -- joiner
2383 -> Stream m (f a)
2384 -> Stream m (f a)
2385splitInnerBySuffix splitter joiner (Stream step1 state1) =
2386 Stream step (SplitInitial state1)
2388 where
2390 {-# INLINE_LATE step #-}
2391 step gst (SplitInitial st) = do
2392 r <- step1 gst st
2393 case r of
2394 Yield x s -> do
2395 (x1, mx2) <- splitter x
2396 return $ case mx2 of
2397 Nothing -> Skip (SplitBuffering s x1)
2398 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
2399 Skip s -> return $ Skip (SplitInitial s)
2400 Stop -> return Stop
2402 step gst (SplitBuffering st buf) = do
2403 r <- step1 gst st
2404 case r of
2405 Yield x s -> do
2406 (x1, mx2) <- splitter x
2407 buf' <- joiner buf x1
2408 return $ case mx2 of
2409 Nothing -> Skip (SplitBuffering s buf')
2410 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
2411 Skip s -> return $ Skip (SplitBuffering s buf)
2412 Stop -> return $
2413 if buf == mempty
2414 then Stop
2415 else Skip (SplitYielding buf SplitFinishing)
2417 step _ (SplitSplitting st buf) = do
2418 (x1, mx2) <- splitter buf
2419 return $ case mx2 of
2420 Nothing -> Skip $ SplitBuffering st x1
2421 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
2423 step _ (SplitYielding x next) = return $ Yield x next
2424 step _ SplitFinishing = return Stop