1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE ExistentialQuantification #-}
5{-# LANGUAGE FlexibleContexts #-}
6{-# LANGUAGE FlexibleInstances #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE PatternSynonyms #-}
9{-# LANGUAGE RecordWildCards #-}
10{-# LANGUAGE ScopedTypeVariables #-}
11{-# LANGUAGE ViewPatterns #-}
12{-# LANGUAGE RankNTypes #-}
13{-# LANGUAGE MagicHash #-}
15#if __GLASGOW_HASKELL__ >= 801
16{-# LANGUAGE TypeApplications #-}
22-- Module : Streamly.Streams.StreamD
23-- Copyright : (c) 2018 Harendra Kumar
24-- Copyright : (c) Roman Leshchinskiy 2008-2010
25-- Copyright : (c) The University of Glasgow, 2009
26-- Copyright : (c) Bjoern Hoehrmann 2008-2009
29-- Maintainer : streamly@composewell.com
30-- Stability : experimental
33-- Direct style re-implementation of CPS style stream in StreamK module. The
34-- symbol or suffix 'D' in this module denotes the "Direct" style. GHC is able
35-- to INLINE and fuse direct style better, providing better performance than
39-- import qualified Streamly.Streams.StreamD as D
42-- Some of the functions in this file have been adapted from the vector
43-- library, https://hackage.haskell.org/package/vector.
45module Streamly.Streams.StreamD
65 -- ** Specialized Generation
66 -- | Generate a monadic stream from a seed.
77 , enumerateFromStepIntegral
78 , enumerateFromIntegral
79 , enumerateFromThenIntegral
80 , enumerateFromToIntegral
81 , enumerateFromThenToIntegral
83 , enumerateFromStepNum
86 , enumerateFromToFractional
87 , enumerateFromThenToFractional
90 -- | Transform an input structure into a stream.
91 -- | Direct style stream does not support @fromFoldable@.
119 -- ** Specialized Folds
140 -- ** Flattening nested streams
143 , ConcatMapUState (..)
145 , ConcatUnfoldInterleaveState (..)
146 , concatUnfoldInterleave
147 , concatUnfoldRoundrobin
150 , InterleaveState(..)
155 , roundRobin -- interleaveFair?/ParallelFair
188 -- | Transform a stream into another type.
205 -- ** By folding (scans)
248 , intersperseSuffixBySpan
272 -- * Transformation comprehensions
284 -- * UTF8 Encoding / Decoding transformations.
292 , resumeDecodeUtf8Either
294 , decodeUtf8ArraysLenient
298import Control.Exception (Exception, SomeException)
299import Control.Monad (void)
300import Control.Monad.Catch (MonadCatch)
301import Control.Monad.IO.Class (MonadIO(..))
302import Control.Monad.Reader (ReaderT)
303import Control.Monad.State.Strict (StateT)
304import Control.Monad.Trans (MonadTrans(lift))
305import Data.Bits (shiftR, shiftL, (.|.), (.&.))
306import Data.Functor.Identity (Identity(..))
307import Data.Maybe (fromJust, isJust)
308import Data.Word (Word32)
309import Foreign.Ptr (Ptr)
310import Foreign.Storable (Storable(..))
311import GHC.Base (assert, unsafeChr, ord)
312import GHC.IO.Encoding.Failure (isSurrogate)
313import GHC.ForeignPtr (ForeignPtr (..))
314import GHC.Ptr (Ptr (..))
315import GHC.Types (SPEC(..))
316import GHC.Word (Word8(..))
317import System.IO.Unsafe (unsafePerformIO)
319 hiding (map, mapM, mapM_, repeat, foldr, last, take, filter,
320 takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,
321 notElem, null, head, tail, zipWith, lookup, foldr1, sequence,
322 (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat,
325import qualified Control.Monad.Catch as MC
326import qualified Control.Monad.Reader as Reader
327import qualified Control.Monad.State.Strict as State
329import Streamly.Internal.Memory.Array.Types (Array(..))
330import Streamly.Internal.Data.Fold.Types (Fold(..))
331import Streamly.Internal.Data.Pipe.Types (Pipe(..), PipeState(..))
332import Streamly.Internal.Data.SVar (MonadAsync, defState, adaptState)
333import Streamly.Internal.Data.Unfold.Types (Unfold(..))
334import Streamly.Internal.Data.Strict (Tuple'(..))
336import Streamly.Internal.Data.Stream.StreamD.Type
338import qualified Streamly.Internal.Data.Pipe.Types as Pipe
339import qualified Streamly.Internal.Memory.Array.Types as A
340import qualified Streamly.Memory.Ring as RB
341import qualified Streamly.Streams.StreamK as K
343import Foreign.Ptr (plusPtr)
344import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
345import Foreign.ForeignPtr (touchForeignPtr)
347------------------------------------------------------------------------------
349------------------------------------------------------------------------------
351-- | An empty 'Stream'.
352{-# INLINE_NORMAL nil #-}
353nil :: Monad m => Stream m a
354nil = Stream (\_ _ -> return Stop) ()
356-- | An empty 'Stream' with a side effect.
357{-# INLINE_NORMAL nilM #-}
358nilM :: Monad m => m b -> Stream m a
359nilM m = Stream (\_ _ -> m >> return Stop) ()
361{-# INLINE_NORMAL consM #-}
362consM :: Monad m => m a -> Stream m a -> Stream m a
363consM m (Stream step state) = Stream step1 Nothing
365 {-# INLINE_LATE step1 #-}
366 step1 _ Nothing = m >>= \x -> return $ Yield x (Just state)
367 step1 gst (Just st) = do
371 Yield a s -> Yield a (Just s)
372 Skip s -> Skip (Just s)
375-- XXX implement in terms of consM?
376-- cons x = consM (return x)
378-- | Can fuse but has O(n^2) complexity.
379{-# INLINE_NORMAL cons #-}
380cons :: Monad m => a -> Stream m a -> Stream m a
381cons x (Stream step state) = Stream step1 Nothing
383 {-# INLINE_LATE step1 #-}
384 step1 _ Nothing = return $ Yield x (Just state)
385 step1 gst (Just st) = do
389 Yield a s -> Yield a (Just s)
390 Skip s -> Skip (Just s)
393-------------------------------------------------------------------------------
395-------------------------------------------------------------------------------
397-- Does not fuse, has the same performance as the StreamK version.
398{-# INLINE_NORMAL uncons #-}
399uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
400uncons (UnStream step state) = go state
403 r <- step defState st
405 Yield x s -> return $ Just (x, Stream step s)
407 Stop -> return Nothing
409------------------------------------------------------------------------------
410-- Generation by unfold
411------------------------------------------------------------------------------
413{-# INLINE_NORMAL unfoldrM #-}
414unfoldrM :: Monad m => (s -> m (Maybe (a, s))) -> s -> Stream m a
415unfoldrM next state = Stream step state
417 {-# INLINE_LATE step #-}
421 Just (x, s) -> Yield x s
424{-# INLINE_LATE unfoldr #-}
425unfoldr :: Monad m => (s -> Maybe (a, s)) -> s -> Stream m a
426unfoldr f = unfoldrM (return . f)
428-- | Convert an 'Unfold' into a 'Stream' by supplying it a seed.
430{-# INLINE_NORMAL unfold #-}
431unfold :: Monad m => Unfold m a b -> a -> Stream m b
432unfold (Unfold ustep inject) seed = Stream step Nothing
434 {-# INLINE_LATE step #-}
435 step _ Nothing = inject seed >>= return . Skip . Just
436 step _ (Just st) = do
439 Yield x s -> Yield x (Just s)
440 Skip s -> Skip (Just s)
443------------------------------------------------------------------------------
444-- Specialized Generation
445------------------------------------------------------------------------------
447repeatM :: Monad m => m a -> Stream m a
448repeatM x = Stream (\_ _ -> x >>= \r -> return $ Yield r ()) ()
450repeat :: Monad m => a -> Stream m a
451repeat x = Stream (\_ _ -> return $ Yield x ()) ()
453{-# INLINE_NORMAL replicateM #-}
454replicateM :: forall m a. Monad m => Int -> m a -> Stream m a
455replicateM n p = Stream step n
457 {-# INLINE_LATE step #-}
459 | i <= 0 = return Stop
462 return $ Yield x (i - 1)
464{-# INLINE_NORMAL replicate #-}
465replicate :: Monad m => Int -> a -> Stream m a
466replicate n x = replicateM n (return x)
468-- This would not work properly for floats, therefore we put an Integral
470-- | Can be used to enumerate unbounded integrals. This does not check for
471-- overflow or underflow for bounded integrals.
472{-# INLINE_NORMAL enumerateFromStepIntegral #-}
473enumerateFromStepIntegral :: (Integral a, Monad m) => a -> a -> Stream m a
474enumerateFromStepIntegral from stride =
475 from `seq` stride `seq` Stream step from
477 {-# INLINE_LATE step #-}
478 step _ !x = return $ Yield x $! (x + stride)
480-- We are assuming that "to" is constrained by the type to be within
482{-# INLINE enumerateFromToIntegral #-}
483enumerateFromToIntegral :: (Monad m, Integral a) => a -> a -> Stream m a
484enumerateFromToIntegral from to =
485 takeWhile (<= to) $ enumerateFromStepIntegral from 1
487{-# INLINE enumerateFromIntegral #-}
488enumerateFromIntegral :: (Monad m, Integral a, Bounded a) => a -> Stream m a
489enumerateFromIntegral from = enumerateFromToIntegral from maxBound
491data EnumState a = EnumInit | EnumYield a a a | EnumStop
493{-# INLINE_NORMAL enumerateFromThenToIntegralUp #-}
494enumerateFromThenToIntegralUp
495 :: (Monad m, Integral a)
496 => a -> a -> a -> Stream m a
497enumerateFromThenToIntegralUp from next to = Stream step EnumInit
499 {-# INLINE_LATE step #-}
505 else Yield from EnumStop
506 else -- from <= next <= to
507 let stride = next - from
508 in Skip $ EnumYield from stride (to - stride)
510 step _ (EnumYield x stride toMinus) =
513 then Yield x EnumStop
514 else Yield x $ EnumYield (x + stride) stride toMinus
516 step _ EnumStop = return Stop
518{-# INLINE_NORMAL enumerateFromThenToIntegralDn #-}
519enumerateFromThenToIntegralDn
520 :: (Monad m, Integral a)
521 => a -> a -> a -> Stream m a
522enumerateFromThenToIntegralDn from next to = Stream step EnumInit
524 {-# INLINE_LATE step #-}
526 return $ if to > next
529 else Yield from EnumStop
530 else -- from >= next >= to
531 let stride = next - from
532 in Skip $ EnumYield from stride (to - stride)
534 step _ (EnumYield x stride toMinus) =
537 then Yield x EnumStop
538 else Yield x $ EnumYield (x + stride) stride toMinus
540 step _ EnumStop = return Stop
542{-# INLINE_NORMAL enumerateFromThenToIntegral #-}
543enumerateFromThenToIntegral
544 :: (Monad m, Integral a)
545 => a -> a -> a -> Stream m a
546enumerateFromThenToIntegral from next to
547 | next >= from = enumerateFromThenToIntegralUp from next to
548 | otherwise = enumerateFromThenToIntegralDn from next to
550{-# INLINE_NORMAL enumerateFromThenIntegral #-}
551enumerateFromThenIntegral
552 :: (Monad m, Integral a, Bounded a)
553 => a -> a -> Stream m a
554enumerateFromThenIntegral from next =
556 then enumerateFromThenToIntegralUp from next maxBound
557 else enumerateFromThenToIntegralDn from next minBound
559-- For floating point numbers if the increment is less than the precision then
560-- it just gets lost. Therefore we cannot always increment it correctly by just
562-- 9007199254740992 + 1 + 1 :: Double => 9.007199254740992e15
563-- 9007199254740992 + 2 :: Double => 9.007199254740994e15
565-- Instead we accumulate the increment counter and compute the increment
566-- everytime before adding it to the starting number.
568-- This works for Integrals as well as floating point numbers, but
569-- enumerateFromStepIntegral is faster for integrals.
570{-# INLINE_NORMAL enumerateFromStepNum #-}
571enumerateFromStepNum :: (Monad m, Num a) => a -> a -> Stream m a
572enumerateFromStepNum from stride = Stream step 0
574 {-# INLINE_LATE step #-}
575 step _ !i = return $ (Yield $! (from + i * stride)) $! (i + 1)
577{-# INLINE_NORMAL numFrom #-}
578numFrom :: (Monad m, Num a) => a -> Stream m a
579numFrom from = enumerateFromStepNum from 1
581{-# INLINE_NORMAL numFromThen #-}
582numFromThen :: (Monad m, Num a) => a -> a -> Stream m a
583numFromThen from next = enumerateFromStepNum from (next - from)
585-- We cannot write a general function for Num. The only way to write code
586-- portable between the two is to use a 'Real' constraint and convert between
587-- Fractional and Integral using fromRational which is horribly slow.
588{-# INLINE_NORMAL enumerateFromToFractional #-}
589enumerateFromToFractional
590 :: (Monad m, Fractional a, Ord a)
591 => a -> a -> Stream m a
592enumerateFromToFractional from to =
593 takeWhile (<= to + 1 / 2) $ enumerateFromStepNum from 1
595{-# INLINE_NORMAL enumerateFromThenToFractional #-}
596enumerateFromThenToFractional
597 :: (Monad m, Fractional a, Ord a)
598 => a -> a -> a -> Stream m a
599enumerateFromThenToFractional from next to =
600 takeWhile predicate $ numFromThen from next
602 mid = (next - from) / 2
603 predicate | next >= from = (<= to + mid)
604 | otherwise = (>= to + mid)
606-------------------------------------------------------------------------------
607-- Generation by Conversion
608-------------------------------------------------------------------------------
610{-# INLINE_NORMAL fromIndicesM #-}
611fromIndicesM :: Monad m => (Int -> m a) -> Stream m a
612fromIndicesM gen = Stream step 0
614 {-# INLINE_LATE step #-}
617 return $ Yield x (i + 1)
619{-# INLINE fromIndices #-}
620fromIndices :: Monad m => (Int -> a) -> Stream m a
621fromIndices gen = fromIndicesM (return . gen)
623{-# INLINE_NORMAL generateM #-}
624generateM :: Monad m => Int -> (Int -> m a) -> Stream m a
625generateM n gen = n `seq` Stream step 0
627 {-# INLINE_LATE step #-}
628 step _ i | i < n = do
630 return $ Yield x (i + 1)
631 | otherwise = return Stop
633{-# INLINE generate #-}
634generate :: Monad m => Int -> (Int -> a) -> Stream m a
635generate n gen = generateM n (return . gen)
637-- XXX we need the MonadAsync constraint because of a rewrite rule.
638-- | Convert a list of monadic actions to a 'Stream'
639{-# INLINE_LATE fromListM #-}
640fromListM :: MonadAsync m => [m a] -> Stream m a
641fromListM = Stream step
643 {-# INLINE_LATE step #-}
644 step _ (m:ms) = m >>= \x -> return $ Yield x ms
645 step _ [] = return Stop
647{-# INLINE toStreamD #-}
648toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a
649toStreamD = fromStreamK . K.toStream
651{-# INLINE_NORMAL hoist #-}
652hoist :: Monad n => (forall x. m x -> n x) -> Stream m a -> Stream n a
653hoist f (Stream step state) = (Stream step' state)
656 r <- f $ step (adaptState gst) st
658 Yield x s -> Yield x s
662{-# INLINE_NORMAL generally #-}
663generally :: Monad m => Stream Identity a -> Stream m a
664generally = hoist (return . runIdentity)
666{-# INLINE_NORMAL liftInner #-}
667liftInner :: (Monad m, MonadTrans t, Monad (t m))
668 => Stream m a -> Stream (t m) a
669liftInner (Stream step state) = Stream step' state
672 r <- lift $ step (adaptState gst) st
674 Yield x s -> Yield x s
678{-# INLINE_NORMAL runReaderT #-}
679runReaderT :: Monad m => s -> Stream (ReaderT s m) a -> Stream m a
680runReaderT sval (Stream step state) = Stream step' state
683 r <- Reader.runReaderT (step (adaptState gst) st) sval
685 Yield x s -> Yield x s
689{-# INLINE_NORMAL evalStateT #-}
690evalStateT :: Monad m => s -> Stream (StateT s m) a -> Stream m a
691evalStateT sval (Stream step state) = Stream step' (state, sval)
693 step' gst (st, sv) = do
694 (r, sv') <- State.runStateT (step (adaptState gst) st) sv
696 Yield x s -> Yield x (s, sv')
697 Skip s -> Skip (s, sv')
700{-# INLINE_NORMAL runStateT #-}
701runStateT :: Monad m => s -> Stream (StateT s m) a -> Stream m (s, a)
702runStateT sval (Stream step state) = Stream step' (state, sval)
704 step' gst (st, sv) = do
705 (r, sv') <- State.runStateT (step (adaptState gst) st) sv
707 Yield x s -> Yield (sv', x) (s, sv')
708 Skip s -> Skip (s, sv')
711------------------------------------------------------------------------------
712-- Elimination by Folds
713------------------------------------------------------------------------------
715------------------------------------------------------------------------------
717------------------------------------------------------------------------------
719{-# INLINE_NORMAL foldr1 #-}
720foldr1 :: Monad m => (a -> a -> a) -> Stream m a -> m (Maybe a)
724 Nothing -> return Nothing
725 Just (h, t) -> fmap Just (foldr f h t)
727------------------------------------------------------------------------------
729------------------------------------------------------------------------------
731{-# INLINE_NORMAL foldlT #-}
732foldlT :: (Monad m, Monad (s m), MonadTrans s)
733 => (s m b -> a -> s m b) -> s m b -> Stream m a -> s m b
734foldlT fstep begin (Stream step state) = go SPEC begin state
737 r <- lift $ step defState st
739 Yield x s -> go SPEC (fstep acc x) s
740 Skip s -> go SPEC acc s
743-- Note, this is going to have horrible performance, because of the nature of
744-- the stream type (i.e. direct stream vs CPS). Its only for reference, it is
745-- likely be practically unusable.
746{-# INLINE_NORMAL foldlS #-}
748 => (Stream m b -> a -> Stream m b) -> Stream m b -> Stream m a -> Stream m b
749foldlS fstep begin (Stream step state) = Stream step' (Left (state, begin))
751 step' gst (Left (st, acc)) = do
752 r <- step (adaptState gst) st
754 Yield x s -> Skip (Left (s, fstep acc x))
755 Skip s -> Skip (Left (s, acc))
756 Stop -> Skip (Right acc)
758 step' gst (Right (Stream stp stt)) = do
759 r <- stp (adaptState gst) stt
761 Yield x s -> Yield x (Right (Stream stp s))
762 Skip s -> Skip (Right (Stream stp s))
765------------------------------------------------------------------------------
767------------------------------------------------------------------------------
769-- | Run a streaming composition, discard the results.
770{-# INLINE_LATE drain #-}
771drain :: Monad m => Stream m a -> m ()
772-- drain = foldrM (\_ xs -> xs) (return ())
773drain (Stream step state) = go SPEC state
776 r <- step defState st
778 Yield _ s -> go SPEC s
782{-# INLINE_NORMAL null #-}
783null :: Monad m => Stream m a -> m Bool
784null m = foldrM (\_ _ -> return False) (return True) m
786{-# INLINE_NORMAL head #-}
787head :: Monad m => Stream m a -> m (Maybe a)
788head m = foldrM (\x _ -> return (Just x)) (return Nothing) m
790-- Does not fuse, has the same performance as the StreamK version.
791{-# INLINE_NORMAL tail #-}
792tail :: Monad m => Stream m a -> m (Maybe (Stream m a))
793tail (UnStream step state) = go state
796 r <- step defState st
798 Yield _ s -> return (Just $ Stream step s)
800 Stop -> return Nothing
802-- XXX will it fuse? need custom impl?
803{-# INLINE_NORMAL last #-}
804last :: Monad m => Stream m a -> m (Maybe a)
805last = foldl' (\_ y -> Just y) Nothing
807{-# INLINE_NORMAL elem #-}
808elem :: (Monad m, Eq a) => a -> Stream m a -> m Bool
809-- elem e m = foldrM (\x xs -> if x == e then return True else xs) (return False) m
810elem e (Stream step state) = go state
813 r <- step defState st
816 | x == e -> return True
821{-# INLINE_NORMAL notElem #-}
822notElem :: (Monad m, Eq a) => a -> Stream m a -> m Bool
823notElem e s = fmap not (elem e s)
825{-# INLINE_NORMAL all #-}
826all :: Monad m => (a -> Bool) -> Stream m a -> m Bool
827-- all p m = foldrM (\x xs -> if p x then xs else return False) (return True) m
828all p (Stream step state) = go state
831 r <- step defState st
835 | otherwise -> return False
839{-# INLINE_NORMAL any #-}
840any :: Monad m => (a -> Bool) -> Stream m a -> m Bool
841-- any p m = foldrM (\x xs -> if p x then return True else xs) (return False) m
842any p (Stream step state) = go state
845 r <- step defState st
853{-# INLINE_NORMAL maximum #-}
854maximum :: (Monad m, Ord a) => Stream m a -> m (Maybe a)
855maximum (Stream step state) = go Nothing state
858 r <- step defState st
860 Yield x s -> go (Just x) s
861 Skip s -> go Nothing s
862 Stop -> return Nothing
863 go (Just acc) st = do
864 r <- step defState st
867 | acc <= x -> go (Just x) s
868 | otherwise -> go (Just acc) s
869 Skip s -> go (Just acc) s
870 Stop -> return (Just acc)
872{-# INLINE_NORMAL maximumBy #-}
873maximumBy :: Monad m => (a -> a -> Ordering) -> Stream m a -> m (Maybe a)
874maximumBy cmp (Stream step state) = go Nothing state
877 r <- step defState st
879 Yield x s -> go (Just x) s
880 Skip s -> go Nothing s
881 Stop -> return Nothing
882 go (Just acc) st = do
883 r <- step defState st
885 Yield x s -> case cmp acc x of
886 GT -> go (Just acc) s
888 Skip s -> go (Just acc) s
889 Stop -> return (Just acc)
891{-# INLINE_NORMAL minimum #-}
892minimum :: (Monad m, Ord a) => Stream m a -> m (Maybe a)
893minimum (Stream step state) = go Nothing state
896 r <- step defState st
898 Yield x s -> go (Just x) s
899 Skip s -> go Nothing s
900 Stop -> return Nothing
901 go (Just acc) st = do
902 r <- step defState st
905 | acc <= x -> go (Just acc) s
906 | otherwise -> go (Just x) s
907 Skip s -> go (Just acc) s
908 Stop -> return (Just acc)
910{-# INLINE_NORMAL minimumBy #-}
911minimumBy :: Monad m => (a -> a -> Ordering) -> Stream m a -> m (Maybe a)
912minimumBy cmp (Stream step state) = go Nothing state
915 r <- step defState st
917 Yield x s -> go (Just x) s
918 Skip s -> go Nothing s
919 Stop -> return Nothing
920 go (Just acc) st = do
921 r <- step defState st
923 Yield x s -> case cmp acc x of
926 Skip s -> go (Just acc) s
927 Stop -> return (Just acc)
929{-# INLINE_NORMAL (!!) #-}
930(!!) :: (Monad m) => Stream m a -> Int -> m (Maybe a)
931(Stream step state) !! i = go i state
934 r <- step defState st
936 Yield x s | n < 0 -> return Nothing
937 | n == 0 -> return $ Just x
938 | otherwise -> go (n - 1) s
940 Stop -> return Nothing
942{-# INLINE_NORMAL lookup #-}
943lookup :: (Monad m, Eq a) => a -> Stream m (a, b) -> m (Maybe b)
944lookup e m = foldrM (\(a, b) xs -> if e == a then return (Just b) else xs)
947{-# INLINE_NORMAL findM #-}
948findM :: Monad m => (a -> m Bool) -> Stream m a -> m (Maybe a)
949findM p m = foldrM (\x xs -> p x >>= \r -> if r then return (Just x) else xs)
953find :: Monad m => (a -> Bool) -> Stream m a -> m (Maybe a)
954find p = findM (return . p)
956{-# INLINE_NORMAL findIndices #-}
957findIndices :: Monad m => (a -> Bool) -> Stream m a -> Stream m Int
958findIndices p (Stream step state) = Stream step' (state, 0)
960 {-# INLINE_LATE step' #-}
961 step' gst (st, i) = i `seq` do
962 r <- step (adaptState gst) st
964 Yield x s -> if p x then Yield i (s, i+1) else Skip (s, i+1)
965 Skip s -> Skip (s, i)
968{-# INLINE toListRev #-}
969toListRev :: Monad m => Stream m a -> m [a]
970toListRev = foldl' (flip (:)) []
972-- We can implement reverse as:
974-- > reverse = foldlS (flip cons) nil
976-- However, this implementation is unusable because of the horrible performance
977-- of cons. So we just convert it to a list first and then stream from the
980-- XXX Maybe we can use an Array instead of a list here?
981{-# INLINE_NORMAL reverse #-}
982reverse :: Monad m => Stream m a -> Stream m a
983reverse m = Stream step Nothing
985 {-# INLINE_LATE step #-}
988 return $ Skip (Just xs)
989 step _ (Just (x:xs)) = return $ Yield x (Just xs)
990 step _ (Just []) = return Stop
992-- Much faster reverse for Storables
993{-# INLINE_NORMAL reverse' #-}
994reverse' :: forall m a. (MonadIO m, Storable a) => Stream m a -> Stream m a
996-- This commented implementation copies the whole stream into one single array
997-- and then streams from that array, this is 3-4x faster than the chunked code
998-- that follows. Though this could be problematic due to unbounded large
999-- allocations. We need to figure out why the chunked code is slower and if we
1000-- can optimize the chunked code to work as fast as this one. It may be a
1002import Foreign.ForeignPtr (touchForeignPtr)
1003import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
1004import Foreign.Ptr (Ptr, plusPtr)
1005reverse' m = Stream step Nothing
1007 {-# INLINE_LATE step #-}
1009 arr <- A.fromStreamD m
1010 let p = aEnd arr `plusPtr` negate (sizeOf (undefined :: a))
1011 return $ Skip $ Just (aStart arr, p)
1013 step _ (Just (start, p)) | p < unsafeForeignPtrToPtr start = return Stop
1015 step _ (Just (start, p)) = do
1016 let !x = A.unsafeInlineIO $ do
1018 touchForeignPtr start
1020 next = p `plusPtr` negate (sizeOf (undefined :: a))
1021 return $ Yield x (Just (start, next))
1028 $ A.fromStreamDArraysOf A.defaultChunkSize m
1031------------------------------------------------------------------------------
1032-- Grouping/Splitting
1033------------------------------------------------------------------------------
1035{-# INLINE_NORMAL splitSuffixBy' #-}
1036splitSuffixBy' :: Monad m
1037 => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1038splitSuffixBy' predicate f (Stream step state) =
1039 Stream (stepOuter f) (Just state)
1043 {-# INLINE_LATE stepOuter #-}
1044 stepOuter (Fold fstep initial done) gst (Just st) = do
1045 res <- step (adaptState gst) st
1051 then done acc' >>= \val -> return $ Yield val (Just s)
1054 Skip s -> return $ Skip $ Just s
1060 res <- step (adaptState gst) stt
1065 then done acc' >>= \val -> return $ Yield val (Just s)
1067 Skip s -> go SPEC s acc
1068 Stop -> done acc >>= \val -> return $ Yield val Nothing
1070 stepOuter _ _ Nothing = return Stop
1072{-# INLINE_NORMAL groupsBy #-}
1078groupsBy cmp f (Stream step state) = Stream (stepOuter f) (Just state, Nothing)
1082 {-# INLINE_LATE stepOuter #-}
1083 stepOuter (Fold fstep initial done) gst (Just st, Nothing) = do
1084 res <- step (adaptState gst) st
1091 Skip s -> return $ Skip $ (Just s, Nothing)
1096 go !_ prev stt !acc = do
1097 res <- step (adaptState gst) stt
1104 else done acc >>= \r -> return $ Yield r (Just s, Just x)
1105 Skip s -> go SPEC prev s acc
1106 Stop -> done acc >>= \r -> return $ Yield r (Nothing, Nothing)
1108 stepOuter (Fold fstep initial done) gst (Just st, Just prev) = do
1110 acc' <- fstep acc prev
1115 -- XXX code duplicated from the previous equation
1117 res <- step (adaptState gst) stt
1124 else done acc >>= \r -> return $ Yield r (Just s, Just x)
1125 Skip s -> go SPEC s acc
1126 Stop -> done acc >>= \r -> return $ Yield r (Nothing, Nothing)
1128 stepOuter _ _ (Nothing,_) = return Stop
1130{-# INLINE_NORMAL groupsRollingBy #-}
1131groupsRollingBy :: Monad m
1136groupsRollingBy cmp f (Stream step state) =
1137 Stream (stepOuter f) (Just state, Nothing)
1140 {-# INLINE_LATE stepOuter #-}
1141 stepOuter (Fold fstep initial done) gst (Just st, Nothing) = do
1142 res <- step (adaptState gst) st
1149 Skip s -> return $ Skip $ (Just s, Nothing)
1153 go !_ prev stt !acc = do
1154 res <- step (adaptState gst) stt
1162 done acc >>= \r -> return $ Yield r (Just s, Just x)
1163 Skip s -> go SPEC prev s acc
1164 Stop -> done acc >>= \r -> return $ Yield r (Nothing, Nothing)
1166 stepOuter (Fold fstep initial done) gst (Just st, Just prev') = do
1168 acc' <- fstep acc prev'
1169 go SPEC prev' st acc'
1172 go !_ prevv stt !acc = do
1173 res <- step (adaptState gst) stt
1180 else done acc >>= \r -> return $ Yield r (Just s, Just x)
1181 Skip s -> go SPEC prevv s acc
1182 Stop -> done acc >>= \r -> return $ Yield r (Nothing, Nothing)
1184 stepOuter _ _ (Nothing, _) = return Stop
1186{-# INLINE_NORMAL splitBy #-}
1187splitBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1188splitBy predicate f (Stream step state) = Stream (step' f) (Just state)
1192 {-# INLINE_LATE step' #-}
1193 step' (Fold fstep initial done) gst (Just st) = initial >>= go SPEC st
1198 res <- step (adaptState gst) stt
1202 then done acc >>= \r -> return $ Yield r (Just s)
1206 Skip s -> go SPEC s acc
1207 Stop -> done acc >>= \r -> return $ Yield r Nothing
1209 step' _ _ Nothing = return Stop
1211-- XXX requires -funfolding-use-threshold=150 in lines-unlines benchmark
1212{-# INLINE_NORMAL splitSuffixBy #-}
1213splitSuffixBy :: Monad m
1214 => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1215splitSuffixBy predicate f (Stream step state) = Stream (step' f) (Just state)
1219 {-# INLINE_LATE step' #-}
1220 step' (Fold fstep initial done) gst (Just st) = do
1221 res <- step (adaptState gst) st
1226 then done acc >>= \val -> return $ Yield val (Just s)
1231 Skip s -> return $ Skip $ Just s
1237 res <- step (adaptState gst) stt
1241 then done acc >>= \r -> return $ Yield r (Just s)
1245 Skip s -> go SPEC s acc
1246 Stop -> done acc >>= \r -> return $ Yield r Nothing
1248 step' _ _ Nothing = return Stop
1250{-# INLINE_NORMAL wordsBy #-}
1251wordsBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1252wordsBy predicate f (Stream step state) = Stream (stepOuter f) (Just state)
1256 {-# INLINE_LATE stepOuter #-}
1257 stepOuter (Fold fstep initial done) gst (Just st) = do
1258 res <- step (adaptState gst) st
1262 then return $ Skip (Just s)
1268 Skip s -> return $ Skip $ Just s
1274 res <- step (adaptState gst) stt
1278 then done acc >>= \r -> return $ Yield r (Just s)
1282 Skip s -> go SPEC s acc
1283 Stop -> done acc >>= \r -> return $ Yield r Nothing
1285 stepOuter _ _ Nothing = return Stop
1287-- String search algorithms:
1288-- http://www-igm.univ-mlv.fr/~lecroq/string/index.html
1291-- TODO can we unify the splitting operations using a splitting configuration
1292-- like in the split package.
1294data SplitStyle = Infix | Suffix | Prefix deriving (Eq, Show)
1296data SplitOptions = SplitOptions
1297 { style :: SplitStyle
1298 , withSep :: Bool -- ^ keep the separators in output
1299 -- , compact :: Bool -- ^ treat multiple consecutive separators as one
1300 -- , trimHead :: Bool -- ^ drop blank at head
1301 -- , trimTail :: Bool -- ^ drop blank at tail
1305data SplitOnState s a =
1310 | GO_KARP_RABIN s !(RB.Ring a) !(Ptr a)
1313{-# INLINE_NORMAL splitOn #-}
1315 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1320splitOn patArr@Array{..} (Fold fstep initial done) (Stream step state) =
1321 Stream stepOuter GO_START
1325 patLen = A.length patArr
1326 maxIndex = patLen - 1
1327 elemBits = sizeOf (undefined :: a) * 8
1329 {-# INLINE_LATE stepOuter #-}
1330 stepOuter _ GO_START =
1332 then return $ Skip $ GO_EMPTY_PAT state
1335 r <- liftIO $ (A.unsafeIndexIO patArr 0)
1336 return $ Skip $ GO_SINGLE_PAT state r
1337 else if sizeOf (undefined :: a) * patLen
1338 <= sizeOf (undefined :: Word)
1339 then return $ Skip $ GO_SHORT_PAT state
1341 (rb, rhead) <- liftIO $ RB.new patLen
1342 return $ Skip $ GO_KARP_RABIN state rb rhead
1344 stepOuter gst (GO_SINGLE_PAT stt pat) = initial >>= go SPEC stt
1349 res <- step (adaptState gst) st
1355 return $ Yield r (GO_SINGLE_PAT s pat)
1356 else fstep acc x >>= go SPEC s
1357 Skip s -> go SPEC s acc
1358 Stop -> done acc >>= \r -> return $ Yield r GO_DONE
1360 stepOuter gst (GO_SHORT_PAT stt) = initial >>= go0 SPEC 0 (0 :: Word) stt
1365 mask = (1 `shiftL` (elemBits * patLen)) - 1
1367 addToWord wrd a = (wrd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1370 patWord = mask .&. A.foldl' addToWord 0 patArr
1372 go0 !_ !idx wrd st !acc = do
1373 res <- step (adaptState gst) st
1376 let wrd' = addToWord wrd x
1379 if wrd' .&. mask == patWord
1382 return $ Yield r (GO_SHORT_PAT s)
1383 else go1 SPEC wrd' s acc
1384 else go0 SPEC (idx + 1) wrd' s acc
1385 Skip s -> go0 SPEC idx wrd s acc
1388 then go2 wrd idx acc
1390 done acc' >>= \r -> return $ Yield r GO_DONE
1393 go1 !_ wrd st !acc = do
1394 res <- step (adaptState gst) st
1397 let wrd' = addToWord wrd x
1398 old = (mask .&. wrd) `shiftR` (elemBits * (patLen - 1))
1399 acc' <- fstep acc (toEnum $ fromIntegral old)
1400 if wrd' .&. mask == patWord
1401 then done acc' >>= \r -> return $ Yield r (GO_SHORT_PAT s)
1402 else go1 SPEC wrd' s acc'
1403 Skip s -> go1 SPEC wrd s acc
1405 acc' <- go2 wrd patLen acc
1406 done acc' >>= \r -> return $ Yield r GO_DONE
1408 go2 !wrd !n !acc | n > 0 = do
1409 let old = (mask .&. wrd) `shiftR` (elemBits * (n - 1))
1410 fstep acc (toEnum $ fromIntegral old) >>= go2 wrd (n - 1)
1411 go2 _ _ acc = return acc
1413 stepOuter gst (GO_KARP_RABIN stt rb rhead) = do
1414 initial >>= go0 SPEC 0 rhead stt
1418 k = 2891336453 :: Word32
1420 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1421 deltaCksum cksum old new =
1422 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1424 -- XXX shall we use a random starting hash or 1 instead of 0?
1425 patHash = A.foldl' addCksum 0 patArr
1428 go0 !_ !idx !rh st !acc = do
1429 res <- step (adaptState gst) st
1432 rh' <- liftIO $ RB.unsafeInsert rb rh x
1435 let fold = RB.unsafeFoldRing (RB.ringBound rb)
1436 let !ringHash = fold addCksum 0 rb
1437 if ringHash == patHash
1438 then go2 SPEC ringHash rh' s acc
1439 else go1 SPEC ringHash rh' s acc
1440 else go0 SPEC (idx + 1) rh' s acc
1441 Skip s -> go0 SPEC idx rh s acc
1443 !acc' <- if idx /= 0
1444 then RB.unsafeFoldRingM rh fstep acc rb
1446 done acc' >>= \r -> return $ Yield r GO_DONE
1448 -- XXX Theoretically this code can do 4 times faster if GHC generates
1449 -- optimal code. If we use just "(cksum' == patHash)" condition it goes
1450 -- 4x faster, as soon as we add the "RB.unsafeEqArray rb v" condition
1451 -- the generated code changes drastically and becomes 4x slower. Need
1452 -- to investigate what is going on with GHC.
1454 go1 !_ !cksum !rh st !acc = do
1455 res <- step (adaptState gst) st
1458 old <- liftIO $ peek rh
1459 let cksum' = deltaCksum cksum old x
1460 acc' <- fstep acc old
1462 if (cksum' == patHash)
1464 rh' <- liftIO (RB.unsafeInsert rb rh x)
1465 go2 SPEC cksum' rh' s acc'
1467 rh' <- liftIO (RB.unsafeInsert rb rh x)
1468 go1 SPEC cksum' rh' s acc'
1469 Skip s -> go1 SPEC cksum rh s acc
1471 acc' <- RB.unsafeFoldRingFullM rh fstep acc rb
1472 done acc' >>= \r -> return $ Yield r GO_DONE
1474 go2 !_ !cksum' !rh' s !acc' = do
1475 if RB.unsafeEqArray rb rh' patArr
1478 return $ Yield r (GO_KARP_RABIN s rb rhead)
1479 else go1 SPEC cksum' rh' s acc'
1481 stepOuter gst (GO_EMPTY_PAT st) = do
1482 res <- step (adaptState gst) st
1487 done acc' >>= \r -> return $ Yield r (GO_EMPTY_PAT s)
1488 Skip s -> return $ Skip (GO_EMPTY_PAT s)
1491 stepOuter _ GO_DONE = return Stop
1493{-# INLINE_NORMAL splitSuffixOn #-}
1495 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1501splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)
1502 (Stream step state) =
1503 Stream stepOuter GO_START
1507 patLen = A.length patArr
1508 maxIndex = patLen - 1
1509 elemBits = sizeOf (undefined :: a) * 8
1511 {-# INLINE_LATE stepOuter #-}
1512 stepOuter _ GO_START =
1514 then return $ Skip $ GO_EMPTY_PAT state
1517 r <- liftIO $ (A.unsafeIndexIO patArr 0)
1518 return $ Skip $ GO_SINGLE_PAT state r
1519 else if sizeOf (undefined :: a) * patLen
1520 <= sizeOf (undefined :: Word)
1521 then return $ Skip $ GO_SHORT_PAT state
1523 (rb, rhead) <- liftIO $ RB.new patLen
1524 return $ Skip $ GO_KARP_RABIN state rb rhead
1526 stepOuter gst (GO_SINGLE_PAT stt pat) = do
1527 -- This first part is the only difference between splitOn and
1529 -- If the last element is a separator do not issue a blank segment.
1530 res <- step (adaptState gst) stt
1536 acc' <- if withSep then fstep acc x else return acc
1537 done acc' >>= \r -> return $ Yield r (GO_SINGLE_PAT s pat)
1538 else fstep acc x >>= go SPEC s
1539 Skip s -> return $ Skip $ (GO_SINGLE_PAT s pat)
1544 -- This is identical for splitOn and splitSuffixOn
1546 res <- step (adaptState gst) st
1551 acc' <- if withSep then fstep acc x else return acc
1553 return $ Yield r (GO_SINGLE_PAT s pat)
1554 else fstep acc x >>= go SPEC s
1555 Skip s -> go SPEC s acc
1556 Stop -> done acc >>= \r -> return $ Yield r GO_DONE
1558 stepOuter gst (GO_SHORT_PAT stt) = do
1560 -- Call "initial" only if the stream yields an element, otherwise we
1561 -- may call "initial" but never yield anything. initial may produce a
1562 -- side effect, therefore we will end up doing and discard a side
1567 res <- step (adaptState gst) stt
1571 let wrd' = addToWord wrd x
1572 acc' <- if withSep then fstep acc x else return acc
1575 if wrd' .&. mask == patWord
1576 then done acc' >>= \r -> return $ Yield r (GO_SHORT_PAT s)
1577 else go0 SPEC (idx + 1) wrd' s acc'
1578 else go0 SPEC (idx + 1) wrd' s acc'
1579 Skip s -> return $ Skip (GO_SHORT_PAT s)
1585 mask = (1 `shiftL` (elemBits * patLen)) - 1
1587 addToWord wrd a = (wrd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1590 patWord = mask .&. A.foldl' addToWord 0 patArr
1592 go0 !_ !idx wrd st !acc = do
1593 res <- step (adaptState gst) st
1596 let wrd' = addToWord wrd x
1597 acc' <- if withSep then fstep acc x else return acc
1600 if wrd' .&. mask == patWord
1603 return $ Yield r (GO_SHORT_PAT s)
1604 else go1 SPEC wrd' s acc'
1605 else go0 SPEC (idx + 1) wrd' s acc'
1606 Skip s -> go0 SPEC idx wrd s acc
1608 if (idx == maxIndex) && (wrd .&. mask == patWord)
1611 acc' <- if idx /= 0 && not withSep
1612 then go2 wrd idx acc
1614 done acc' >>= \r -> return $ Yield r GO_DONE
1617 go1 !_ wrd st !acc = do
1618 res <- step (adaptState gst) st
1621 let wrd' = addToWord wrd x
1622 old = (mask .&. wrd) `shiftR` (elemBits * (patLen - 1))
1625 else fstep acc (toEnum $ fromIntegral old)
1626 if wrd' .&. mask == patWord
1627 then done acc' >>= \r -> return $ Yield r (GO_SHORT_PAT s)
1628 else go1 SPEC wrd' s acc'
1629 Skip s -> go1 SPEC wrd s acc
1631 -- If the last sequence is a separator do not issue a blank
1633 if wrd .&. mask == patWord
1638 else go2 wrd patLen acc
1639 done acc' >>= \r -> return $ Yield r GO_DONE
1641 go2 !wrd !n !acc | n > 0 = do
1642 let old = (mask .&. wrd) `shiftR` (elemBits * (n - 1))
1643 fstep acc (toEnum $ fromIntegral old) >>= go2 wrd (n - 1)
1644 go2 _ _ acc = return acc
1646 stepOuter gst (GO_KARP_RABIN stt rb rhead) = do
1648 res <- step (adaptState gst) stt
1652 acc' <- if withSep then fstep acc x else return acc
1653 rh' <- liftIO (RB.unsafeInsert rb rhead x)
1656 let fold = RB.unsafeFoldRing (RB.ringBound rb)
1657 let !ringHash = fold addCksum 0 rb
1658 if ringHash == patHash
1659 then go2 SPEC ringHash rh' s acc'
1660 else go0 SPEC (idx + 1) rh' s acc'
1661 else go0 SPEC (idx + 1) rh' s acc'
1662 Skip s -> return $ Skip (GO_KARP_RABIN s rb rhead)
1667 k = 2891336453 :: Word32
1669 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1670 deltaCksum cksum old new =
1671 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1673 -- XXX shall we use a random starting hash or 1 instead of 0?
1674 patHash = A.foldl' addCksum 0 patArr
1677 go0 !_ !idx !rh st !acc = do
1678 res <- step (adaptState gst) st
1681 acc' <- if withSep then fstep acc x else return acc
1682 rh' <- liftIO (RB.unsafeInsert rb rh x)
1685 let fold = RB.unsafeFoldRing (RB.ringBound rb)
1686 let !ringHash = fold addCksum 0 rb
1687 if ringHash == patHash
1688 then go2 SPEC ringHash rh' s acc'
1689 else go1 SPEC ringHash rh' s acc'
1690 else go0 SPEC (idx + 1) rh' s acc'
1691 Skip s -> go0 SPEC idx rh s acc
1693 -- do not issue a blank segment when we end at pattern
1694 if (idx == maxIndex) && RB.unsafeEqArray rb rh patArr
1697 !acc' <- if idx /= 0 && not withSep
1698 then RB.unsafeFoldRingM rh fstep acc rb
1700 done acc' >>= \r -> return $ Yield r GO_DONE
1702 -- XXX Theoretically this code can do 4 times faster if GHC generates
1703 -- optimal code. If we use just "(cksum' == patHash)" condition it goes
1704 -- 4x faster, as soon as we add the "RB.unsafeEqArray rb v" condition
1705 -- the generated code changes drastically and becomes 4x slower. Need
1706 -- to investigate what is going on with GHC.
1708 go1 !_ !cksum !rh st !acc = do
1709 res <- step (adaptState gst) st
1712 old <- liftIO $ peek rh
1713 let cksum' = deltaCksum cksum old x
1718 if (cksum' == patHash)
1720 rh' <- liftIO (RB.unsafeInsert rb rh x)
1721 go2 SPEC cksum' rh' s acc'
1723 rh' <- liftIO (RB.unsafeInsert rb rh x)
1724 go1 SPEC cksum' rh' s acc'
1725 Skip s -> go1 SPEC cksum rh s acc
1727 if RB.unsafeEqArray rb rh patArr
1732 else RB.unsafeFoldRingFullM rh fstep acc rb
1733 done acc' >>= \r -> return $ Yield r GO_DONE
1735 go2 !_ !cksum' !rh' s !acc' = do
1736 if RB.unsafeEqArray rb rh' patArr
1739 return $ Yield r (GO_KARP_RABIN s rb rhead)
1740 else go1 SPEC cksum' rh' s acc'
1742 stepOuter gst (GO_EMPTY_PAT st) = do
1743 res <- step (adaptState gst) st
1748 done acc' >>= \r -> return $ Yield r (GO_EMPTY_PAT s)
1749 Skip s -> return $ Skip (GO_EMPTY_PAT s)
1752 stepOuter _ GO_DONE = return Stop
1754data SplitState s arr
1756 | SplitBuffering s arr
1757 | SplitSplitting s arr
1758 | SplitYielding arr (SplitState s arr)
1761-- XXX An alternative approach would be to use a partial fold (Fold m a b) to
1762-- split using a splitBy like combinator. The Fold would consume upto the
1763-- separator and return any leftover which can then be fed to the next fold.
1765-- We can revisit this once we have partial folds/parsers.
1767-- | Performs infix separator style splitting.
1768{-# INLINE_NORMAL splitInnerBy #-}
1771 => (f a -> m (f a, Maybe (f a))) -- splitter
1772 -> (f a -> f a -> m (f a)) -- joiner
1775splitInnerBy splitter joiner (Stream step1 state1) =
1776 (Stream step (SplitInitial state1))
1780 {-# INLINE_LATE step #-}
1781 step gst (SplitInitial st) = do
1785 (x1, mx2) <- splitter x
1786 return $ case mx2 of
1787 Nothing -> Skip (SplitBuffering s x1)
1788 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
1789 Skip s -> return $ Skip (SplitInitial s)
1790 Stop -> return $ Stop
1792 step gst (SplitBuffering st buf) = do
1796 (x1, mx2) <- splitter x
1797 buf' <- joiner buf x1
1798 return $ case mx2 of
1799 Nothing -> Skip (SplitBuffering s buf')
1800 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
1801 Skip s -> return $ Skip (SplitBuffering s buf)
1802 Stop -> return $ Skip (SplitYielding buf SplitFinishing)
1804 step _ (SplitSplitting st buf) = do
1805 (x1, mx2) <- splitter buf
1806 return $ case mx2 of
1807 Nothing -> Skip $ SplitBuffering st x1
1808 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
1810 step _ (SplitYielding x next) = return $ Yield x next
1811 step _ SplitFinishing = return $ Stop
1813-- | Performs infix separator style splitting.
1814{-# INLINE_NORMAL splitInnerBySuffix #-}
1816 :: (Monad m, Eq (f a), Monoid (f a))
1817 => (f a -> m (f a, Maybe (f a))) -- splitter
1818 -> (f a -> f a -> m (f a)) -- joiner
1821splitInnerBySuffix splitter joiner (Stream step1 state1) =
1822 (Stream step (SplitInitial state1))
1826 {-# INLINE_LATE step #-}
1827 step gst (SplitInitial st) = do
1831 (x1, mx2) <- splitter x
1832 return $ case mx2 of
1833 Nothing -> Skip (SplitBuffering s x1)
1834 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
1835 Skip s -> return $ Skip (SplitInitial s)
1836 Stop -> return $ Stop
1838 step gst (SplitBuffering st buf) = do
1842 (x1, mx2) <- splitter x
1843 buf' <- joiner buf x1
1844 return $ case mx2 of
1845 Nothing -> Skip (SplitBuffering s buf')
1846 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
1847 Skip s -> return $ Skip (SplitBuffering s buf)
1851 else Skip (SplitYielding buf SplitFinishing)
1853 step _ (SplitSplitting st buf) = do
1854 (x1, mx2) <- splitter buf
1855 return $ case mx2 of
1856 Nothing -> Skip $ SplitBuffering st x1
1857 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
1859 step _ (SplitYielding x next) = return $ Yield x next
1860 step _ SplitFinishing = return $ Stop
1862------------------------------------------------------------------------------
1864------------------------------------------------------------------------------
1866{-# INLINE_NORMAL isPrefixOf #-}
1867isPrefixOf :: (Eq a, Monad m) => Stream m a -> Stream m a -> m Bool
1868isPrefixOf (Stream stepa ta) (Stream stepb tb) = go (ta, tb, Nothing)
1870 go (sa, sb, Nothing) = do
1871 r <- stepa defState sa
1873 Yield x sa' -> go (sa', sb, Just x)
1874 Skip sa' -> go (sa', sb, Nothing)
1877 go (sa, sb, Just x) = do
1878 r <- stepb defState sb
1882 then go (sa, sb', Nothing)
1884 Skip sb' -> go (sa, sb', Just x)
1885 Stop -> return False
1887{-# INLINE_NORMAL isSubsequenceOf #-}
1888isSubsequenceOf :: (Eq a, Monad m) => Stream m a -> Stream m a -> m Bool
1889isSubsequenceOf (Stream stepa ta) (Stream stepb tb) = go (ta, tb, Nothing)
1891 go (sa, sb, Nothing) = do
1892 r <- stepa defState sa
1894 Yield x sa' -> go (sa', sb, Just x)
1895 Skip sa' -> go (sa', sb, Nothing)
1898 go (sa, sb, Just x) = do
1899 r <- stepb defState sb
1903 then go (sa, sb', Nothing)
1904 else go (sa, sb', Just x)
1905 Skip sb' -> go (sa, sb', Just x)
1906 Stop -> return False
1908{-# INLINE_NORMAL stripPrefix #-}
1911 => Stream m a -> Stream m a -> m (Maybe (Stream m a))
1912stripPrefix (Stream stepa ta) (Stream stepb tb) = go (ta, tb, Nothing)
1914 go (sa, sb, Nothing) = do
1915 r <- stepa defState sa
1917 Yield x sa' -> go (sa', sb, Just x)
1918 Skip sa' -> go (sa', sb, Nothing)
1919 Stop -> return $ Just (Stream stepb sb)
1921 go (sa, sb, Just x) = do
1922 r <- stepb defState sb
1926 then go (sa, sb', Nothing)
1928 Skip sb' -> go (sa, sb', Just x)
1929 Stop -> return Nothing
1931------------------------------------------------------------------------------
1933------------------------------------------------------------------------------
1935-- | Execute a monadic action for each element of the 'Stream'
1936{-# INLINE_NORMAL mapM_ #-}
1937mapM_ :: Monad m => (a -> m b) -> Stream m a -> m ()
1938mapM_ m = drain . mapM m
1940-------------------------------------------------------------------------------
1941-- Stream transformations using Unfolds
1942-------------------------------------------------------------------------------
1944-- Define a unique structure to use in inspection testing
1945data ConcatMapUState o i =
1947 | ConcatMapUInner o i
1949-- | @concatMapU unfold stream@ uses @unfold@ to map the input stream elements
1950-- to streams and then flattens the generated streams into a single output
1953-- This is like 'concatMap' but uses an unfold with an explicit state to
1954-- generate the stream instead of a 'Stream' type generator. This allows better
1955-- optimization via fusion. This can be many times more efficient than
1958{-# INLINE_NORMAL concatMapU #-}
1959concatMapU :: Monad m => Unfold m a b -> Stream m a -> Stream m b
1960concatMapU (Unfold istep inject) (Stream ostep ost) =
1961 Stream step (ConcatMapUOuter ost)
1963 {-# INLINE_LATE step #-}
1964 step gst (ConcatMapUOuter o) = do
1965 r <- ostep (adaptState gst) o
1969 i `seq` return (Skip (ConcatMapUInner o' i))
1970 Skip o' -> return $ Skip (ConcatMapUOuter o')
1971 Stop -> return $ Stop
1973 step _ (ConcatMapUInner o i) = do
1976 Yield x i' -> Yield x (ConcatMapUInner o i')
1977 Skip i' -> Skip (ConcatMapUInner o i')
1978 Stop -> Skip (ConcatMapUOuter o)
1980data ConcatUnfoldInterleaveState o i =
1981 ConcatUnfoldInterleaveOuter o [i]
1982 | ConcatUnfoldInterleaveInner o [i]
1983 | ConcatUnfoldInterleaveInnerL [i] [i]
1984 | ConcatUnfoldInterleaveInnerR [i] [i]
1986-- XXX use arrays to store state instead of lists.
1987-- XXX In general we can use different scheduling strategies e.g. how to
1988-- schedule the outer vs inner loop or assigning weights to different streams
1989-- or outer and inner loops.
1991-- After a yield, switch to the next stream. Do not switch streams on Skip.
1992-- Yield from outer stream switches to the inner stream.
1994-- There are two choices here, (1) exhaust the outer stream first and then
1995-- start yielding from the inner streams, this is much simpler to implement,
1996-- (2) yield at least one element from an inner stream before going back to
1997-- outer stream and opening the next stream from it.
1999-- Ideally, we need some scheduling bias to inner streams vs outer stream.
2000-- Maybe we can configure the behavior.
2002{-# INLINE_NORMAL concatUnfoldInterleave #-}
2003concatUnfoldInterleave :: Monad m => Unfold m a b -> Stream m a -> Stream m b
2004concatUnfoldInterleave (Unfold istep inject) (Stream ostep ost) =
2005 Stream step (ConcatUnfoldInterleaveOuter ost [])
2007 {-# INLINE_LATE step #-}
2008 step gst (ConcatUnfoldInterleaveOuter o ls) = do
2009 r <- ostep (adaptState gst) o
2013 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
2014 Skip o' -> return $ Skip (ConcatUnfoldInterleaveOuter o' ls)
2015 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
2017 step _ (ConcatUnfoldInterleaveInner _ []) = undefined
2018 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
2021 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
2022 Skip s -> Skip (ConcatUnfoldInterleaveInner o (s:ls))
2023 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
2025 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
2026 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
2027 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
2029 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
2032 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
2033 Skip s -> Skip (ConcatUnfoldInterleaveInnerL (s:ls) rs)
2034 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
2036 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
2037 step _ (ConcatUnfoldInterleaveInnerR ls []) =
2038 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
2040 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
2043 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
2044 Skip s -> Skip (ConcatUnfoldInterleaveInnerR ls (s:rs))
2045 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
2047-- XXX In general we can use different scheduling strategies e.g. how to
2048-- schedule the outer vs inner loop or assigning weights to different streams
2049-- or outer and inner loops.
2051-- This could be inefficient if the tasks are too small.
2053-- Compared to concatUnfoldInterleave this one switches streams on Skips.
2055{-# INLINE_NORMAL concatUnfoldRoundrobin #-}
2056concatUnfoldRoundrobin :: Monad m => Unfold m a b -> Stream m a -> Stream m b
2057concatUnfoldRoundrobin (Unfold istep inject) (Stream ostep ost) =
2058 Stream step (ConcatUnfoldInterleaveOuter ost [])
2060 {-# INLINE_LATE step #-}
2061 step gst (ConcatUnfoldInterleaveOuter o ls) = do
2062 r <- ostep (adaptState gst) o
2066 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
2067 Skip o' -> return $ Skip (ConcatUnfoldInterleaveInner o' ls)
2068 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
2070 step _ (ConcatUnfoldInterleaveInner o []) =
2071 return $ Skip (ConcatUnfoldInterleaveOuter o [])
2073 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
2076 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
2077 Skip s -> Skip (ConcatUnfoldInterleaveOuter o (s:ls))
2078 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
2080 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
2081 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
2082 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
2084 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
2087 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
2088 Skip s -> Skip (ConcatUnfoldInterleaveInnerL ls (s:rs))
2089 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
2091 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
2092 step _ (ConcatUnfoldInterleaveInnerR ls []) =
2093 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
2095 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
2098 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
2099 Skip s -> Skip (ConcatUnfoldInterleaveInnerR (s:ls) rs)
2100 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
2102data AppendState s1 s2 = AppendFirst s1 | AppendSecond s2
2104-- Note that this could be much faster compared to the CPS stream. However, as
2105-- the number of streams being composed increases this may become expensive.
2106-- Need to see where the breaking point is between the two.
2108{-# INLINE_NORMAL append #-}
2109append :: Monad m => Stream m a -> Stream m a -> Stream m a
2110append (Stream step1 state1) (Stream step2 state2) =
2111 Stream step (AppendFirst state1)
2115 {-# INLINE_LATE step #-}
2116 step gst (AppendFirst st) = do
2119 Yield a s -> Yield a (AppendFirst s)
2120 Skip s -> Skip (AppendFirst s)
2121 Stop -> Skip (AppendSecond state2)
2123 step gst (AppendSecond st) = do
2126 Yield a s -> Yield a (AppendSecond s)
2127 Skip s -> Skip (AppendSecond s)
2130data InterleaveState s1 s2 = InterleaveFirst s1 s2 | InterleaveSecond s1 s2
2131 | InterleaveSecondOnly s2 | InterleaveFirstOnly s1
2133{-# INLINE_NORMAL interleave #-}
2134interleave :: Monad m => Stream m a -> Stream m a -> Stream m a
2135interleave (Stream step1 state1) (Stream step2 state2) =
2136 Stream step (InterleaveFirst state1 state2)
2140 {-# INLINE_LATE step #-}
2141 step gst (InterleaveFirst st1 st2) = do
2144 Yield a s -> Yield a (InterleaveSecond s st2)
2145 Skip s -> Skip (InterleaveFirst s st2)
2146 Stop -> Skip (InterleaveSecondOnly st2)
2148 step gst (InterleaveSecond st1 st2) = do
2151 Yield a s -> Yield a (InterleaveFirst st1 s)
2152 Skip s -> Skip (InterleaveSecond st1 s)
2153 Stop -> Skip (InterleaveFirstOnly st1)
2155 step gst (InterleaveFirstOnly st1) = do
2158 Yield a s -> Yield a (InterleaveFirstOnly s)
2159 Skip s -> Skip (InterleaveFirstOnly s)
2162 step gst (InterleaveSecondOnly st2) = do
2165 Yield a s -> Yield a (InterleaveSecondOnly s)
2166 Skip s -> Skip (InterleaveSecondOnly s)
2169{-# INLINE_NORMAL interleaveMin #-}
2170interleaveMin :: Monad m => Stream m a -> Stream m a -> Stream m a
2171interleaveMin (Stream step1 state1) (Stream step2 state2) =
2172 Stream step (InterleaveFirst state1 state2)
2176 {-# INLINE_LATE step #-}
2177 step gst (InterleaveFirst st1 st2) = do
2180 Yield a s -> Yield a (InterleaveSecond s st2)
2181 Skip s -> Skip (InterleaveFirst s st2)
2184 step gst (InterleaveSecond st1 st2) = do
2187 Yield a s -> Yield a (InterleaveFirst st1 s)
2188 Skip s -> Skip (InterleaveSecond st1 s)
2191 step _ (InterleaveFirstOnly _) = undefined
2192 step _ (InterleaveSecondOnly _) = undefined
2194{-# INLINE_NORMAL interleaveSuffix #-}
2195interleaveSuffix :: Monad m => Stream m a -> Stream m a -> Stream m a
2196interleaveSuffix (Stream step1 state1) (Stream step2 state2) =
2197 Stream step (InterleaveFirst state1 state2)
2201 {-# INLINE_LATE step #-}
2202 step gst (InterleaveFirst st1 st2) = do
2205 Yield a s -> Yield a (InterleaveSecond s st2)
2206 Skip s -> Skip (InterleaveFirst s st2)
2209 step gst (InterleaveSecond st1 st2) = do
2212 Yield a s -> Yield a (InterleaveFirst st1 s)
2213 Skip s -> Skip (InterleaveSecond st1 s)
2214 Stop -> Skip (InterleaveFirstOnly st1)
2216 step gst (InterleaveFirstOnly st1) = do
2219 Yield a s -> Yield a (InterleaveFirstOnly s)
2220 Skip s -> Skip (InterleaveFirstOnly s)
2223 step _ (InterleaveSecondOnly _) = undefined
2225data InterleaveInfixState s1 s2 a
2226 = InterleaveInfixFirst s1 s2
2227 | InterleaveInfixSecondBuf s1 s2
2228 | InterleaveInfixSecondYield s1 s2 a
2229 | InterleaveInfixFirstYield s1 s2 a
2230 | InterleaveInfixFirstOnly s1
2232{-# INLINE_NORMAL interleaveInfix #-}
2233interleaveInfix :: Monad m => Stream m a -> Stream m a -> Stream m a
2234interleaveInfix (Stream step1 state1) (Stream step2 state2) =
2235 Stream step (InterleaveInfixFirst state1 state2)
2239 {-# INLINE_LATE step #-}
2240 step gst (InterleaveInfixFirst st1 st2) = do
2243 Yield a s -> Yield a (InterleaveInfixSecondBuf s st2)
2244 Skip s -> Skip (InterleaveInfixFirst s st2)
2247 step gst (InterleaveInfixSecondBuf st1 st2) = do
2250 Yield a s -> Skip (InterleaveInfixSecondYield st1 s a)
2251 Skip s -> Skip (InterleaveInfixSecondBuf st1 s)
2252 Stop -> Skip (InterleaveInfixFirstOnly st1)
2254 step gst (InterleaveInfixSecondYield st1 st2 x) = do
2257 Yield a s -> Yield x (InterleaveInfixFirstYield s st2 a)
2258 Skip s -> Skip (InterleaveInfixSecondYield s st2 x)
2261 step _ (InterleaveInfixFirstYield st1 st2 x) = do
2262 return $ Yield x (InterleaveInfixSecondBuf st1 st2)
2264 step gst (InterleaveInfixFirstOnly st1) = do
2267 Yield a s -> Yield a (InterleaveInfixFirstOnly s)
2268 Skip s -> Skip (InterleaveInfixFirstOnly s)
2271{-# INLINE_NORMAL roundRobin #-}
2272roundRobin :: Monad m => Stream m a -> Stream m a -> Stream m a
2273roundRobin (Stream step1 state1) (Stream step2 state2) =
2274 Stream step (InterleaveFirst state1 state2)
2278 {-# INLINE_LATE step #-}
2279 step gst (InterleaveFirst st1 st2) = do
2282 Yield a s -> Yield a (InterleaveSecond s st2)
2283 Skip s -> Skip (InterleaveSecond s st2)
2284 Stop -> Skip (InterleaveSecondOnly st2)
2286 step gst (InterleaveSecond st1 st2) = do
2289 Yield a s -> Yield a (InterleaveFirst st1 s)
2290 Skip s -> Skip (InterleaveFirst st1 s)
2291 Stop -> Skip (InterleaveFirstOnly st1)
2293 step gst (InterleaveSecondOnly st2) = do
2296 Yield a s -> Yield a (InterleaveSecondOnly s)
2297 Skip s -> Skip (InterleaveSecondOnly s)
2300 step gst (InterleaveFirstOnly st1) = do
2303 Yield a s -> Yield a (InterleaveFirstOnly s)
2304 Skip s -> Skip (InterleaveFirstOnly s)
2307data ICUState s1 s2 i1 i2 =
2312 | ICUFirstInner s1 s2 i1
2313 | ICUSecondInner s1 s2 i2
2314 | ICUFirstOnlyInner s1 i1
2315 | ICUSecondOnlyInner s2 i2
2317-- | Interleave streams (full streams, not the elements) unfolded from two
2318-- input streams and concat. Stop when the first stream stops. If the second
2319-- stream ends before the first one then first stream still keeps running alone
2320-- without any interleaving with the second stream.
2322-- [a1, a2, ... an] [b1, b2 ...]
2323-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
2324-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
2325-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
2327{-# INLINE_NORMAL gintercalateSuffix #-}
2330 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
2332 (Unfold istep1 inject1) (Stream step1 state1)
2333 (Unfold istep2 inject2) (Stream step2 state2) =
2334 Stream step (ICUFirst state1 state2)
2338 {-# INLINE_LATE step #-}
2339 step gst (ICUFirst s1 s2) = do
2340 r <- step1 (adaptState gst) s1
2344 i `seq` return (Skip (ICUFirstInner s s2 i))
2345 Skip s -> return $ Skip (ICUFirst s s2)
2348 step gst (ICUFirstOnly s1) = do
2349 r <- step1 (adaptState gst) s1
2353 i `seq` return (Skip (ICUFirstOnlyInner s i))
2354 Skip s -> return $ Skip (ICUFirstOnly s)
2357 step _ (ICUFirstInner s1 s2 i1) = do
2360 Yield x i' -> Yield x (ICUFirstInner s1 s2 i')
2361 Skip i' -> Skip (ICUFirstInner s1 s2 i')
2362 Stop -> Skip (ICUSecond s1 s2)
2364 step _ (ICUFirstOnlyInner s1 i1) = do
2367 Yield x i' -> Yield x (ICUFirstOnlyInner s1 i')
2368 Skip i' -> Skip (ICUFirstOnlyInner s1 i')
2369 Stop -> Skip (ICUFirstOnly s1)
2371 step gst (ICUSecond s1 s2) = do
2372 r <- step2 (adaptState gst) s2
2376 i `seq` return (Skip (ICUSecondInner s1 s i))
2377 Skip s -> return $ Skip (ICUSecond s1 s)
2378 Stop -> return $ Skip (ICUFirstOnly s1)
2380 step _ (ICUSecondInner s1 s2 i2) = do
2383 Yield x i' -> Yield x (ICUSecondInner s1 s2 i')
2384 Skip i' -> Skip (ICUSecondInner s1 s2 i')
2385 Stop -> Skip (ICUFirst s1 s2)
2387 step _ (ICUSecondOnly _s2) = undefined
2388 step _ (ICUSecondOnlyInner _s2 _i2) = undefined
2390data InterposeSuffixState s1 i1 =
2391 InterposeSuffixFirst s1
2392 -- | InterposeSuffixFirstYield s1 i1
2393 | InterposeSuffixFirstInner s1 i1
2394 | InterposeSuffixSecond s1
2396-- Note that if an unfolded layer turns out to be nil we still emit the
2397-- separator effect. An alternate behavior could be to emit the separator
2398-- effect only if at least one element has been yielded by the unfolding.
2399-- However, that becomes a bit complicated, so we have chosen the former
2401{-# INLINE_NORMAL interposeSuffix #-}
2404 => m c -> Unfold m b c -> Stream m b -> Stream m c
2407 (Unfold istep1 inject1) (Stream step1 state1) =
2408 Stream step (InterposeSuffixFirst state1)
2412 {-# INLINE_LATE step #-}
2413 step gst (InterposeSuffixFirst s1) = do
2414 r <- step1 (adaptState gst) s1
2418 i `seq` return (Skip (InterposeSuffixFirstInner s i))
2419 -- i `seq` return (Skip (InterposeSuffixFirstYield s i))
2420 Skip s -> return $ Skip (InterposeSuffixFirst s)
2424 step _ (InterposeSuffixFirstYield s1 i1) = do
2427 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
2428 Skip i' -> Skip (InterposeSuffixFirstYield s1 i')
2429 Stop -> Skip (InterposeSuffixFirst s1)
2432 step _ (InterposeSuffixFirstInner s1 i1) = do
2435 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
2436 Skip i' -> Skip (InterposeSuffixFirstInner s1 i')
2437 Stop -> Skip (InterposeSuffixSecond s1)
2439 step _ (InterposeSuffixSecond s1) = do
2441 return $ Yield r (InterposeSuffixFirst s1)
2443data ICALState s1 s2 i1 i2 a =
2445 -- | ICALFirstYield s1 s2 i1
2446 | ICALFirstInner s1 s2 i1
2448 | ICALFirstOnlyInner s1 i1
2449 | ICALSecondInject s1 s2
2450 | ICALFirstInject s1 s2 i2
2451 -- | ICALFirstBuf s1 s2 i1 i2
2452 | ICALSecondInner s1 s2 i1 i2
2453 -- -- | ICALSecondInner s1 s2 i1 i2 a
2454 -- -- | ICALFirstResume s1 s2 i1 i2 a
2456-- | Interleave streams (full streams, not the elements) unfolded from two
2457-- input streams and concat. Stop when the first stream stops. If the second
2458-- stream ends before the first one then first stream still keeps running alone
2459-- without any interleaving with the second stream.
2461-- [a1, a2, ... an] [b1, b2 ...]
2462-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
2463-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
2464-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
2466{-# INLINE_NORMAL gintercalate #-}
2469 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
2471 (Unfold istep1 inject1) (Stream step1 state1)
2472 (Unfold istep2 inject2) (Stream step2 state2) =
2473 Stream step (ICALFirst state1 state2)
2477 {-# INLINE_LATE step #-}
2478 step gst (ICALFirst s1 s2) = do
2479 r <- step1 (adaptState gst) s1
2483 i `seq` return (Skip (ICALFirstInner s s2 i))
2484 -- i `seq` return (Skip (ICALFirstYield s s2 i))
2485 Skip s -> return $ Skip (ICALFirst s s2)
2489 step _ (ICALFirstYield s1 s2 i1) = do
2492 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
2493 Skip i' -> Skip (ICALFirstYield s1 s2 i')
2494 Stop -> Skip (ICALFirst s1 s2)
2497 step _ (ICALFirstInner s1 s2 i1) = do
2500 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
2501 Skip i' -> Skip (ICALFirstInner s1 s2 i')
2502 Stop -> Skip (ICALSecondInject s1 s2)
2504 step gst (ICALFirstOnly s1) = do
2505 r <- step1 (adaptState gst) s1
2509 i `seq` return (Skip (ICALFirstOnlyInner s i))
2510 Skip s -> return $ Skip (ICALFirstOnly s)
2513 step _ (ICALFirstOnlyInner s1 i1) = do
2516 Yield x i' -> Yield x (ICALFirstOnlyInner s1 i')
2517 Skip i' -> Skip (ICALFirstOnlyInner s1 i')
2518 Stop -> Skip (ICALFirstOnly s1)
2520 -- We inject the second stream even before checking if the first stream
2521 -- would yield any more elements. There is no clear choice whether we
2522 -- should do this before or after that. Doing it after may make the state
2523 -- machine a bit simpler though.
2524 step gst (ICALSecondInject s1 s2) = do
2525 r <- step2 (adaptState gst) s2
2529 i `seq` return (Skip (ICALFirstInject s1 s i))
2530 Skip s -> return $ Skip (ICALSecondInject s1 s)
2531 Stop -> return $ Skip (ICALFirstOnly s1)
2533 step gst (ICALFirstInject s1 s2 i2) = do
2534 r <- step1 (adaptState gst) s1
2538 i `seq` return (Skip (ICALSecondInner s s2 i i2))
2539 -- i `seq` return (Skip (ICALFirstBuf s s2 i i2))
2540 Skip s -> return $ Skip (ICALFirstInject s s2 i2)
2544 step _ (ICALFirstBuf s1 s2 i1 i2) = do
2547 Yield x i' -> Skip (ICALSecondInner s1 s2 i' i2 x)
2548 Skip i' -> Skip (ICALFirstBuf s1 s2 i' i2)
2551 step _ (ICALSecondInner s1 s2 i1 i2 v) = do
2554 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i' v)
2555 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i' v)
2556 Stop -> Skip (ICALFirstResume s1 s2 i1 i2 v)
2559 step _ (ICALSecondInner s1 s2 i1 i2) = do
2562 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i')
2563 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i')
2564 Stop -> Skip (ICALFirstInner s1 s2 i1)
2565 -- Stop -> Skip (ICALFirstResume s1 s2 i1 i2)
2568 step _ (ICALFirstResume s1 s2 i1 i2 x) = do
2569 return $ Yield x (ICALFirstInner s1 s2 i1 i2)
2572data InterposeState s1 i1 a =
2574 -- | InterposeFirstYield s1 i1
2575 | InterposeFirstInner s1 i1
2576 | InterposeFirstInject s1
2577 -- | InterposeFirstBuf s1 i1
2578 | InterposeSecondYield s1 i1
2579 -- -- | InterposeSecondYield s1 i1 a
2580 -- -- | InterposeFirstResume s1 i1 a
2582-- Note that this only interposes the pure values, we may run many effects to
2583-- generate those values as some effects may not generate anything (Skip).
2584{-# INLINE_NORMAL interpose #-}
2585interpose :: Monad m => m c -> Unfold m b c -> Stream m b -> Stream m c
2588 (Unfold istep1 inject1) (Stream step1 state1) =
2589 Stream step (InterposeFirst state1)
2593 {-# INLINE_LATE step #-}
2594 step gst (InterposeFirst s1) = do
2595 r <- step1 (adaptState gst) s1
2599 i `seq` return (Skip (InterposeFirstInner s i))
2600 -- i `seq` return (Skip (InterposeFirstYield s i))
2601 Skip s -> return $ Skip (InterposeFirst s)
2605 step _ (InterposeFirstYield s1 i1) = do
2608 Yield x i' -> Yield x (InterposeFirstInner s1 i')
2609 Skip i' -> Skip (InterposeFirstYield s1 i')
2610 Stop -> Skip (InterposeFirst s1)
2613 step _ (InterposeFirstInner s1 i1) = do
2616 Yield x i' -> Yield x (InterposeFirstInner s1 i')
2617 Skip i' -> Skip (InterposeFirstInner s1 i')
2618 Stop -> Skip (InterposeFirstInject s1)
2620 step gst (InterposeFirstInject s1) = do
2621 r <- step1 (adaptState gst) s1
2625 -- i `seq` return (Skip (InterposeFirstBuf s i))
2626 i `seq` return (Skip (InterposeSecondYield s i))
2627 Skip s -> return $ Skip (InterposeFirstInject s)
2631 step _ (InterposeFirstBuf s1 i1) = do
2634 Yield x i' -> Skip (InterposeSecondYield s1 i' x)
2635 Skip i' -> Skip (InterposeFirstBuf s1 i')
2640 step _ (InterposeSecondYield s1 i1 v) = do
2642 return $ Yield r (InterposeFirstResume s1 i1 v)
2644 step _ (InterposeSecondYield s1 i1) = do
2646 return $ Yield r (InterposeFirstInner s1 i1)
2649 step _ (InterposeFirstResume s1 i1 v) = do
2650 return $ Yield v (InterposeFirstInner s1 i1)
2653------------------------------------------------------------------------------
2655------------------------------------------------------------------------------
2657data GbracketState s1 s2 v
2659 | GBracketNormal s1 v
2660 | GBracketException s2
2662-- | The most general bracketing and exception combinator. All other
2663-- combinators can be expressed in terms of this combinator. This can also be
2664-- used for cases which are not covered by the standard combinators.
2668{-# INLINE_NORMAL gbracket #-}
2672 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
2673 -> (c -> m d) -- ^ after, on normal stop
2674 -> (c -> e -> Stream m b) -- ^ on exception
2675 -> (c -> Stream m b) -- ^ stream generator
2677gbracket bef exc aft fexc fnormal =
2678 Stream step GBracketInit
2682 {-# INLINE_LATE step #-}
2683 step _ GBracketInit = do
2685 return $ Skip $ GBracketNormal (fnormal r) r
2687 step gst (GBracketNormal (UnStream step1 st) v) = do
2688 res <- exc $ step1 gst st
2690 Right r -> case r of
2692 return $ Yield x (GBracketNormal (Stream step1 s) v)
2693 Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)
2694 Stop -> aft v >> return Stop
2695 Left e -> return $ Skip (GBracketException (fexc v e))
2696 step gst (GBracketException (UnStream step1 st)) = do
2699 Yield x s -> return $ Yield x (GBracketException (Stream step1 s))
2700 Skip s -> return $ Skip (GBracketException (Stream step1 s))
2703-- | Run a side effect before the stream yields its first element.
2704{-# INLINE_NORMAL before #-}
2705before :: Monad m => m b -> Stream m a -> Stream m a
2706before action (Stream step state) = Stream step' Nothing
2710 {-# INLINE_LATE step' #-}
2711 step' _ Nothing = action >> return (Skip (Just state))
2713 step' gst (Just st) = do
2716 Yield x s -> return $ Yield x (Just s)
2717 Skip s -> return $ Skip (Just s)
2720-- | Run a side effect whenever the stream stops normally.
2721{-# INLINE_NORMAL after #-}
2722after :: Monad m => m b -> Stream m a -> Stream m a
2723after action (Stream step state) = Stream step' state
2727 {-# INLINE_LATE step' #-}
2731 Yield x s -> return $ Yield x s
2732 Skip s -> return $ Skip s
2733 Stop -> action >> return Stop
2735-- XXX These combinators are expensive due to the call to
2736-- onException/handle/try on each step. Therefore, when possible, they should
2737-- be called in an outer loop where we perform less iterations. For example, we
2738-- cannot call them on each iteration in a char stream, instead we can call
2739-- them when doing an IO on an array.
2741-- XXX For high performance error checks in busy streams we may need another
2742-- Error constructor in step.
2744-- | Run a side effect whenever the stream aborts due to an exception. The
2745-- exception is not caught, simply rethrown.
2746{-# INLINE_NORMAL onException #-}
2747onException :: MonadCatch m => m b -> Stream m a -> Stream m a
2748onException action str =
2749 gbracket (return ()) MC.try return
2750 (\_ (e :: MC.SomeException) -> nilM (action >> MC.throwM e))
2753{-# INLINE_NORMAL _onException #-}
2754_onException :: MonadCatch m => m b -> Stream m a -> Stream m a
2755_onException action (Stream step state) = Stream step' state
2759 {-# INLINE_LATE step' #-}
2761 res <- step gst st `MC.onException` action
2763 Yield x s -> return $ Yield x s
2764 Skip s -> return $ Skip s
2767-- XXX bracket is like concatMap, it generates a stream and then flattens it.
2768-- Like concatMap it has 10x worse performance compared to linear fused
2771-- | Run the first action before the stream starts and remember its output,
2772-- generate a stream using the output, run the second action providing the
2773-- remembered value as an argument whenever the stream ends normally or due to
2775{-# INLINE_NORMAL bracket #-}
2776bracket :: MonadCatch m => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
2777bracket bef aft bet =
2778 gbracket bef MC.try aft
2779 (\a (e :: SomeException) -> nilM (aft a >> MC.throwM e)) bet
2781data BracketState s v = BracketInit | BracketRun s v
2783{-# INLINE_NORMAL _bracket #-}
2784_bracket :: MonadCatch m => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
2785_bracket bef aft bet = Stream step' BracketInit
2789 {-# INLINE_LATE step' #-}
2790 step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))
2792 -- NOTE: It is important to use UnStream instead of the Stream pattern
2793 -- here, otherwise we get huge perf degradation, see note in concatMap.
2794 step' gst (BracketRun (UnStream step state) v) = do
2795 -- res <- step gst state `MC.onException` aft v
2796 res <- MC.try $ step gst state
2798 Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop
2799 Right r -> case r of
2800 Yield x s -> return $ Yield x (BracketRun (Stream step s) v)
2801 Skip s -> return $ Skip (BracketRun (Stream step s) v)
2802 Stop -> aft v >> return Stop
2804-- | Run a side effect whenever the stream stops normally or aborts due to an
2806{-# INLINE finally #-}
2807finally :: MonadCatch m => m b -> Stream m a -> Stream m a
2808-- finally action xs = after action $ onException action xs
2809finally action xs = bracket (return ()) (\_ -> action) (const xs)
2811-- | When evaluating a stream if an exception occurs, stream evaluation aborts
2812-- and the specified exception handler is run with the exception as argument.
2813{-# INLINE_NORMAL handle #-}
2814handle :: (MonadCatch m, Exception e)
2815 => (e -> Stream m a) -> Stream m a -> Stream m a
2817 gbracket (return ()) MC.try return (\_ e -> f e) (\_ -> str)
2819{-# INLINE_NORMAL _handle #-}
2820_handle :: (MonadCatch m, Exception e)
2821 => (e -> Stream m a) -> Stream m a -> Stream m a
2822_handle f (Stream step state) = Stream step' (Left state)
2826 {-# INLINE_LATE step' #-}
2827 step' gst (Left st) = do
2828 res <- MC.try $ step gst st
2830 Left e -> return $ Skip $ Right (f e)
2831 Right r -> case r of
2832 Yield x s -> return $ Yield x (Left s)
2833 Skip s -> return $ Skip (Left s)
2836 step' gst (Right (UnStream step1 st)) = do
2839 Yield x s -> return $ Yield x (Right (Stream step1 s))
2840 Skip s -> return $ Skip (Right (Stream step1 s))
2843-------------------------------------------------------------------------------
2844-- General transformation
2845-------------------------------------------------------------------------------
2847{-# INLINE_NORMAL transform #-}
2848transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
2849transform (Pipe pstep1 pstep2 pstate) (Stream step state) =
2850 Stream step' (Consume pstate, state)
2854 {-# INLINE_LATE step' #-}
2856 step' gst (Consume pst, st) = pst `seq` do
2857 r <- step (adaptState gst) st
2862 Pipe.Yield b pst' -> return $ Yield b (pst', s)
2863 Pipe.Continue pst' -> return $ Skip (pst', s)
2864 Skip s -> return $ Skip (Consume pst, s)
2867 step' _ (Produce pst, st) = pst `seq` do
2870 Pipe.Yield b pst' -> return $ Yield b (pst', st)
2871 Pipe.Continue pst' -> return $ Skip (pst', st)
2873------------------------------------------------------------------------------
2874-- Transformation by Folding (Scans)
2875------------------------------------------------------------------------------
2877------------------------------------------------------------------------------
2879------------------------------------------------------------------------------
2881-- XXX Is a prescan useful, discarding the last step does not sound useful? I
2882-- am not sure about the utility of this function, so this is implemented but
2883-- not exposed. We can expose it if someone provides good reasons why this is
2886-- XXX We have to execute the stream one step ahead to know that we are at the
2887-- last step. The vector implementation of prescan executes the last fold step
2888-- but does not yield the result. This means we have executed the effect but
2889-- discarded value. This does not sound right. In this implementation we are
2890-- not executing the last fold step.
2891{-# INLINE_NORMAL prescanlM' #-}
2892prescanlM' :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> Stream m b
2893prescanlM' f mz (Stream step state) = Stream step' (state, mz)
2895 {-# INLINE_LATE step' #-}
2896 step' gst (st, prev) = do
2897 r <- step (adaptState gst) st
2901 return $ Yield acc (s, f acc x)
2902 Skip s -> return $ Skip (s, prev)
2905{-# INLINE prescanl' #-}
2906prescanl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
2907prescanl' f z = prescanlM' (\a b -> return (f a b)) (return z)
2909------------------------------------------------------------------------------
2910-- Monolithic postscans (postscan followed by a map)
2911------------------------------------------------------------------------------
2913-- The performance of a modular postscan followed by a map seems to be
2914-- equivalent to this monolithic scan followed by map therefore we may not need
2915-- this implementation. We just have it for performance comparison and in case
2916-- modular version does not perform well in some situation.
2918{-# INLINE_NORMAL postscanlMx' #-}
2919postscanlMx' :: Monad m
2920 => (x -> a -> m x) -> m x -> (x -> m b) -> Stream m a -> Stream m b
2921postscanlMx' fstep begin done (Stream step state) = do
2922 Stream step' (state, begin)
2924 {-# INLINE_LATE step' #-}
2925 step' gst (st, acc) = do
2926 r <- step (adaptState gst) st
2932 v `seq` y `seq` return (Yield v (s, return y))
2933 Skip s -> return $ Skip (s, acc)
2936{-# INLINE_NORMAL postscanlx' #-}
2937postscanlx' :: Monad m
2938 => (x -> a -> x) -> x -> (x -> b) -> Stream m a -> Stream m b
2939postscanlx' fstep begin done s =
2940 postscanlMx' (\b a -> return (fstep b a)) (return begin) (return . done) s
2942-- XXX do we need consM strict to evaluate the begin value?
2943{-# INLINE scanlMx' #-}
2945 => (x -> a -> m x) -> m x -> (x -> m b) -> Stream m a -> Stream m b
2946scanlMx' fstep begin done s =
2947 (begin >>= \x -> x `seq` done x) `consM` postscanlMx' fstep begin done s
2949{-# INLINE scanlx' #-}
2951 => (x -> a -> x) -> x -> (x -> b) -> Stream m a -> Stream m b
2952scanlx' fstep begin done s =
2953 scanlMx' (\b a -> return (fstep b a)) (return begin) (return . done) s
2955------------------------------------------------------------------------------
2957------------------------------------------------------------------------------
2959{-# INLINE_NORMAL postscanlM' #-}
2960postscanlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
2961postscanlM' fstep begin (Stream step state) =
2962 begin `seq` Stream step' (state, begin)
2964 {-# INLINE_LATE step' #-}
2965 step' gst (st, acc) = acc `seq` do
2966 r <- step (adaptState gst) st
2970 y `seq` return (Yield y (s, y))
2971 Skip s -> return $ Skip (s, acc)
2974{-# INLINE_NORMAL postscanl' #-}
2975postscanl' :: Monad m => (a -> b -> a) -> a -> Stream m b -> Stream m a
2976postscanl' f = postscanlM' (\a b -> return (f a b))
2978{-# INLINE_NORMAL postscanlM #-}
2979postscanlM :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
2980postscanlM fstep begin (Stream step state) = Stream step' (state, begin)
2982 {-# INLINE_LATE step' #-}
2983 step' gst (st, acc) = do
2984 r <- step (adaptState gst) st
2988 return (Yield y (s, y))
2989 Skip s -> return $ Skip (s, acc)
2992{-# INLINE_NORMAL postscanl #-}
2993postscanl :: Monad m => (a -> b -> a) -> a -> Stream m b -> Stream m a
2994postscanl f = postscanlM (\a b -> return (f a b))
2996{-# INLINE_NORMAL scanlM' #-}
2997scanlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
2998scanlM' fstep begin s = begin `seq` (begin `cons` postscanlM' fstep begin s)
3000{-# INLINE scanl' #-}
3001scanl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
3002scanl' f = scanlM' (\a b -> return (f a b))
3004{-# INLINE_NORMAL scanlM #-}
3005scanlM :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
3006scanlM fstep begin s = begin `cons` postscanlM fstep begin s
3009scanl :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
3010scanl f = scanlM (\a b -> return (f a b))
3012{-# INLINE_NORMAL scanl1M #-}
3013scanl1M :: Monad m => (a -> a -> m a) -> Stream m a -> Stream m a
3014scanl1M fstep (Stream step state) = Stream step' (state, Nothing)
3016 {-# INLINE_LATE step' #-}
3017 step' gst (st, Nothing) = do
3020 Yield x s -> return $ Yield x (s, Just x)
3021 Skip s -> return $ Skip (s, Nothing)
3024 step' gst (st, Just acc) = do
3029 return $ Yield z (s, Just z)
3030 Skip s -> return $ Skip (s, Just acc)
3033{-# INLINE scanl1 #-}
3034scanl1 :: Monad m => (a -> a -> a) -> Stream m a -> Stream m a
3035scanl1 f = scanl1M (\x y -> return (f x y))
3037{-# INLINE_NORMAL scanl1M' #-}
3038scanl1M' :: Monad m => (a -> a -> m a) -> Stream m a -> Stream m a
3039scanl1M' fstep (Stream step state) = Stream step' (state, Nothing)
3041 {-# INLINE_LATE step' #-}
3042 step' gst (st, Nothing) = do
3045 Yield x s -> x `seq` return $ Yield x (s, Just x)
3046 Skip s -> return $ Skip (s, Nothing)
3049 step' gst (st, Just acc) = acc `seq` do
3054 z `seq` return $ Yield z (s, Just z)
3055 Skip s -> return $ Skip (s, Just acc)
3058{-# INLINE scanl1' #-}
3059scanl1' :: Monad m => (a -> a -> a) -> Stream m a -> Stream m a
3060scanl1' f = scanl1M' (\x y -> return (f x y))
3063tap :: Monad m => Fold m a b -> Stream m a -> Stream m a
3064tap (Fold fstep initial extract) (Stream step state) = Stream step' Nothing
3068 step' _ Nothing = do
3070 return $ Skip (Just (r, state))
3072 step' gst (Just (acc, st)) = do
3077 return $ Yield x (Just (acc', s))
3078 Skip s -> return $ Skip (Just (acc, s))
3083-------------------------------------------------------------------------------
3085-------------------------------------------------------------------------------
3087{-# INLINE_NORMAL takeWhileM #-}
3088takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
3089takeWhileM f (Stream step state) = Stream step' state
3091 {-# INLINE_LATE step' #-}
3097 return $ if b then Yield x s else Stop
3098 Skip s -> return $ Skip s
3101{-# INLINE takeWhile #-}
3102takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
3103takeWhile f = takeWhileM (return . f)
3105{-# INLINE_NORMAL drop #-}
3106drop :: Monad m => Int -> Stream m a -> Stream m a
3107drop n (Stream step state) = Stream step' (state, Just n)
3109 {-# INLINE_LATE step' #-}
3110 step' gst (st, Just i)
3115 Yield _ s -> Skip (s, Just (i - 1))
3116 Skip s -> Skip (s, Just i)
3118 | otherwise = return $ Skip (st, Nothing)
3120 step' gst (st, Nothing) = do
3124 Yield x s -> Yield x (s, Nothing)
3125 Skip s -> Skip (s, Nothing)
3128data DropWhileState s a
3130 | DropWhileYield a s
3133{-# INLINE_NORMAL dropWhileM #-}
3134dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
3135dropWhileM f (Stream step state) = Stream step' (DropWhileDrop state)
3137 {-# INLINE_LATE step' #-}
3138 step' gst (DropWhileDrop st) = do
3144 then return $ Skip (DropWhileDrop s)
3145 else return $ Skip (DropWhileYield x s)
3146 Skip s -> return $ Skip (DropWhileDrop s)
3149 step' gst (DropWhileNext st) = do
3152 Yield x s -> return $ Skip (DropWhileYield x s)
3153 Skip s -> return $ Skip (DropWhileNext s)
3156 step' _ (DropWhileYield x st) = return $ Yield x (DropWhileNext st)
3158{-# INLINE dropWhile #-}
3159dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
3160dropWhile f = dropWhileM (return . f)
3162{-# INLINE_NORMAL filterM #-}
3163filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
3164filterM f (Stream step state) = Stream step' state
3166 {-# INLINE_LATE step' #-}
3175 Skip s -> return $ Skip s
3178{-# INLINE filter #-}
3179filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
3180filter f = filterM (return . f)
3182{-# INLINE_NORMAL uniq #-}
3183uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
3184uniq (Stream step state) = Stream step' (Nothing, state)
3186 {-# INLINE_LATE step' #-}
3187 step' gst (Nothing, st) = do
3190 Yield x s -> return $ Yield x (Just x, s)
3191 Skip s -> return $ Skip (Nothing, s)
3193 step' gst (Just x, st) = do
3196 Yield y s | x == y -> return $ Skip (Just x, s)
3197 | otherwise -> return $ Yield y (Just y, s)
3198 Skip s -> return $ Skip (Just x, s)
3201------------------------------------------------------------------------------
3202-- Transformation by Mapping
3203------------------------------------------------------------------------------
3205{-# INLINE_NORMAL sequence #-}
3206sequence :: Monad m => Stream m (m a) -> Stream m a
3207sequence (Stream step state) = Stream step' state
3209 {-# INLINE_LATE step' #-}
3211 r <- step (adaptState gst) st
3213 Yield x s -> x >>= \a -> return (Yield a s)
3214 Skip s -> return $ Skip s
3217------------------------------------------------------------------------------
3219------------------------------------------------------------------------------
3221data LoopState x s = FirstYield s
3222 | InterspersingYield s
3225{-# INLINE_NORMAL intersperseM #-}
3226intersperseM :: Monad m => m a -> Stream m a -> Stream m a
3227intersperseM m (Stream step state) = Stream step' (FirstYield state)
3229 {-# INLINE_LATE step' #-}
3230 step' gst (FirstYield st) = do
3234 Yield x s -> Skip (YieldAndCarry x s)
3235 Skip s -> Skip (FirstYield s)
3238 step' gst (InterspersingYield st) = do
3243 return $ Yield a (YieldAndCarry x s)
3244 Skip s -> return $ Skip $ InterspersingYield s
3247 step' _ (YieldAndCarry x st) = return $ Yield x (InterspersingYield st)
3252 | SuffixYield a (SuffixState s a)
3254{-# INLINE_NORMAL intersperseSuffix #-}
3255intersperseSuffix :: forall m a. Monad m => m a -> Stream m a -> Stream m a
3256intersperseSuffix action (Stream step state) = Stream step' (SuffixElem state)
3258 {-# INLINE_LATE step' #-}
3259 step' gst (SuffixElem st) = do
3262 Yield x s -> Skip (SuffixYield x (SuffixSuffix s))
3263 Skip s -> Skip (SuffixElem s)
3266 step' _ (SuffixSuffix st) = do
3267 action >>= \r -> return $ Skip (SuffixYield r (SuffixElem st))
3269 step' _ (SuffixYield x next) = return $ Yield x next
3271data SuffixSpanState s a
3272 = SuffixSpanElem s Int
3273 | SuffixSpanSuffix s
3274 | SuffixSpanYield a (SuffixSpanState s a)
3278-- | intersperse after every n items
3279{-# INLINE_NORMAL intersperseSuffixBySpan #-}
3280intersperseSuffixBySpan :: forall m a. Monad m
3281 => Int -> m a -> Stream m a -> Stream m a
3282intersperseSuffixBySpan n action (Stream step state) =
3283 Stream step' (SuffixSpanElem state n)
3285 {-# INLINE_LATE step' #-}
3286 step' gst (SuffixSpanElem st i) | i > 0 = do
3289 Yield x s -> Skip (SuffixSpanYield x (SuffixSpanElem s (i - 1)))
3290 Skip s -> Skip (SuffixSpanElem s i)
3291 Stop -> if i == n then Stop else Skip SuffixSpanLast
3292 step' _ (SuffixSpanElem st _) = return $ Skip (SuffixSpanSuffix st)
3294 step' _ (SuffixSpanSuffix st) = do
3295 action >>= \r -> return $ Skip (SuffixSpanYield r (SuffixSpanElem st n))
3297 step' _ (SuffixSpanLast) = do
3298 action >>= \r -> return $ Skip (SuffixSpanYield r SuffixSpanStop)
3300 step' _ (SuffixSpanYield x next) = return $ Yield x next
3302 step' _ (SuffixSpanStop) = return Stop
3304{-# INLINE intersperse #-}
3305intersperse :: Monad m => a -> Stream m a -> Stream m a
3306intersperse a = intersperseM (return a)
3308{-# INLINE_NORMAL insertBy #-}
3309insertBy :: Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
3310insertBy cmp a (Stream step state) = Stream step' (state, False, Nothing)
3312 {-# INLINE_LATE step' #-}
3313 step' gst (st, False, _) = do
3316 Yield x s -> case cmp a x of
3317 GT -> return $ Yield x (s, False, Nothing)
3318 _ -> return $ Yield a (s, True, Just x)
3319 Skip s -> return $ Skip (s, False, Nothing)
3320 Stop -> return $ Yield a (st, True, Nothing)
3322 step' _ (_, True, Nothing) = return Stop
3324 step' gst (st, True, Just prev) = do
3327 Yield x s -> return $ Yield prev (s, True, Just x)
3328 Skip s -> return $ Skip (s, True, Just prev)
3329 Stop -> return $ Yield prev (st, True, Nothing)
3331------------------------------------------------------------------------------
3333------------------------------------------------------------------------------
3335{-# INLINE_NORMAL deleteBy #-}
3336deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
3337deleteBy eq x (Stream step state) = Stream step' (state, False)
3339 {-# INLINE_LATE step' #-}
3340 step' gst (st, False) = do
3343 Yield y s -> return $
3344 if eq x y then Skip (s, True) else Yield y (s, False)
3345 Skip s -> return $ Skip (s, False)
3348 step' gst (st, True) = do
3351 Yield y s -> return $ Yield y (s, True)
3352 Skip s -> return $ Skip (s, True)
3355------------------------------------------------------------------------------
3356-- Transformation by Map and Filter
3357------------------------------------------------------------------------------
3359-- XXX Will this always fuse properly?
3360{-# INLINE_NORMAL mapMaybe #-}
3361mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
3362mapMaybe f = fmap fromJust . filter isJust . map f
3364{-# INLINE_NORMAL mapMaybeM #-}
3365mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Stream m a -> Stream m b
3366mapMaybeM f = fmap fromJust . filter isJust . mapM f
3368------------------------------------------------------------------------------
3370------------------------------------------------------------------------------
3372{-# INLINE_NORMAL indexed #-}
3373indexed :: Monad m => Stream m a -> Stream m (Int, a)
3374indexed (Stream step state) = Stream step' (state, 0)
3376 {-# INLINE_LATE step' #-}
3377 step' gst (st, i) = i `seq` do
3378 r <- step (adaptState gst) st
3380 Yield x s -> return $ Yield (i, x) (s, i+1)
3381 Skip s -> return $ Skip (s, i)
3384{-# INLINE_NORMAL indexedR #-}
3385indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
3386indexedR m (Stream step state) = Stream step' (state, m)
3388 {-# INLINE_LATE step' #-}
3389 step' gst (st, i) = i `seq` do
3390 r <- step (adaptState gst) st
3392 Yield x s -> let i' = i - 1
3393 in return $ Yield (i, x) (s, i')
3394 Skip s -> return $ Skip (s, i)
3397{-# INLINE_NORMAL zipWithM #-}
3399 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
3400zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)
3402 {-# INLINE_LATE step #-}
3403 step gst (sa, sb, Nothing) = do
3404 r <- stepa (adaptState gst) sa
3407 Yield x sa' -> Skip (sa', sb, Just x)
3408 Skip sa' -> Skip (sa', sb, Nothing)
3411 step gst (sa, sb, Just x) = do
3412 r <- stepb (adaptState gst) sb
3416 return $ Yield z (sa, sb', Nothing)
3417 Skip sb' -> return $ Skip (sa, sb', Just x)
3420#if __GLASGOW_HASKELL__ >= 801
3421{-# RULES "zipWithM xs xs"
3422 forall f xs. zipWithM @Identity f xs xs = mapM (\x -> f x x) xs #-}
3425{-# INLINE zipWith #-}
3426zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
3427zipWith f = zipWithM (\a b -> return (f a b))
3429------------------------------------------------------------------------------
3431------------------------------------------------------------------------------
3433{-# INLINE_NORMAL mergeByM #-}
3436 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
3437mergeByM cmp (Stream stepa ta) (Stream stepb tb) =
3438 Stream step (Just ta, Just tb, Nothing, Nothing)
3440 {-# INLINE_LATE step #-}
3442 -- one of the values is missing, and the corresponding stream is running
3443 step gst (Just sa, sb, Nothing, b) = do
3446 Yield a sa' -> Skip (Just sa', sb, Just a, b)
3447 Skip sa' -> Skip (Just sa', sb, Nothing, b)
3448 Stop -> Skip (Nothing, sb, Nothing, b)
3450 step gst (sa, Just sb, a, Nothing) = do
3453 Yield b sb' -> Skip (sa, Just sb', a, Just b)
3454 Skip sb' -> Skip (sa, Just sb', a, Nothing)
3455 Stop -> Skip (sa, Nothing, a, Nothing)
3457 -- both the values are available
3458 step _ (sa, sb, Just a, Just b) = do
3460 return $ case res of
3461 GT -> Yield b (sa, sb, Just a, Nothing)
3462 _ -> Yield a (sa, sb, Nothing, Just b)
3464 -- one of the values is missing, corresponding stream is done
3465 step _ (Nothing, sb, Nothing, Just b) =
3466 return $ Yield b (Nothing, sb, Nothing, Nothing)
3468 step _ (sa, Nothing, Just a, Nothing) =
3469 return $ Yield a (sa, Nothing, Nothing, Nothing)
3471 step _ (Nothing, Nothing, Nothing, Nothing) = return Stop
3473{-# INLINE mergeBy #-}
3476 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
3477mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
3479------------------------------------------------------------------------------
3480-- Transformation comprehensions
3481------------------------------------------------------------------------------
3483{-# INLINE_NORMAL the #-}
3484the :: (Eq a, Monad m) => Stream m a -> m (Maybe a)
3485the (Stream step state) = go state
3488 r <- step defState st
3490 Yield x s -> go' x s
3492 Stop -> return Nothing
3494 r <- step defState st
3496 Yield x s | x == n -> go' n s
3497 | otherwise -> return Nothing
3499 Stop -> return (Just n)
3501--------------------------------------------------------------------------------
3502-- UTF8 Encoding / Decoding
3503--------------------------------------------------------------------------------
3505-- UTF-8 primitives, Lifted from GHC.IO.Encoding.UTF8.
3508ord2 :: Char -> WList
3509ord2 c = assert (n >= 0x80 && n <= 0x07ff) (WCons x1 (WCons x2 WNil))
3512 x1 = fromIntegral $ (n `shiftR` 6) + 0xC0
3513 x2 = fromIntegral $ (n .&. 0x3F) + 0x80
3516ord3 :: Char -> WList
3517ord3 c = assert (n >= 0x0800 && n <= 0xffff) (WCons x1 (WCons x2 (WCons x3 WNil)))
3520 x1 = fromIntegral $ (n `shiftR` 12) + 0xE0
3521 x2 = fromIntegral $ ((n `shiftR` 6) .&. 0x3F) + 0x80
3522 x3 = fromIntegral $ (n .&. 0x3F) + 0x80
3525ord4 :: Char -> WList
3526ord4 c = assert (n >= 0x10000) (WCons x1 (WCons x2 (WCons x3 (WCons x4 WNil))))
3529 x1 = fromIntegral $ (n `shiftR` 18) + 0xF0
3530 x2 = fromIntegral $ ((n `shiftR` 12) .&. 0x3F) + 0x80
3531 x3 = fromIntegral $ ((n `shiftR` 6) .&. 0x3F) + 0x80
3532 x4 = fromIntegral $ (n .&. 0x3F) + 0x80
3534data CodingFailureMode
3535 = TransliterateCodingFailure
3536 | ErrorOnCodingFailure
3539{-# INLINE replacementChar #-}
3540replacementChar :: Char
3541replacementChar = '\xFFFD'
3543-- Int helps in cheaper conversion from Int to Char
3545type DecodeState = Word8
3547-- See http://bjoern.hoehrmann.de/utf-8/decoder/dfa/ for details.
3549{-# INLINE runFold #-}
3550runFold :: (Monad m) => Fold m a b -> Stream m a -> m b
3551runFold (Fold step begin done) = foldlMx' step begin done
3553-- XXX Use names decodeSuccess = 0, decodeFailure = 12
3555decodeTable :: [Word8]
3557 -- The first part of the table maps bytes to character classes that
3558 -- to reduce the size of the transition table and create bitmasks.
3559 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,
3560 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,
3561 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,
3562 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,
3563 1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1, 9,9,9,9,9,9,9,9,9,9,9,9,9,9,9,9,
3564 7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7, 7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,
3565 8,8,2,2,2,2,2,2,2,2,2,2,2,2,2,2, 2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,
3566 10,3,3,3,3,3,3,3,3,3,3,3,3,4,3,3, 11,6,6,6,5,8,8,8,8,8,8,8,8,8,8,8,
3568 -- The second part is a transition table that maps a combination
3569 -- of a state of the automaton and a character class to a state.
3570 0,12,24,36,60,96,84,12,12,12,48,72, 12,12,12,12,12,12,12,12,12,12,12,12,
3571 12, 0,12,12,12,12,12, 0,12, 0,12,12, 12,24,12,12,12,12,12,24,12,24,12,12,
3572 12,12,12,12,12,12,12,24,12,12,12,12, 12,24,12,12,12,12,12,12,12,24,12,12,
3573 12,12,12,12,12,12,12,36,12,36,12,12, 12,36,12,12,12,12,12,36,12,36,12,12,
3574 12,36,12,12,12,12,12,12,12,12,12,12
3577utf8d :: A.Array Word8
3580 -- Aligning to cacheline makes a barely noticeable difference
3581 -- XXX currently alignment is not implemented for unmanaged allocation
3582 $ runFold (A.writeNAlignedUnmanaged 64 (length decodeTable))
3583 (fromList decodeTable)
3585-- | Return element at the specified index without checking the bounds.
3586-- and without touching the foreign ptr.
3587{-# INLINE_NORMAL unsafePeekElemOff #-}
3588unsafePeekElemOff :: forall a. Storable a => Ptr a -> Int -> a
3589unsafePeekElemOff p i = let !x = A.unsafeInlineIO $ peekElemOff p i in x
3591-- decode is split into two separate cases to avoid branching instructions.
3592-- From the higher level flow we already know which case we are in so we can
3593-- call the appropriate decode function.
3595-- When the state is 0
3596{-# INLINE decode0 #-}
3597decode0 :: Ptr Word8 -> Word8 -> Tuple' DecodeState CodePoint
3599 let !t = table `unsafePeekElemOff` fromIntegral byte
3600 !codep' = (0xff `shiftR` (fromIntegral t)) .&. fromIntegral byte
3601 !state' = table `unsafePeekElemOff` (256 + fromIntegral t)
3602 in assert ((byte > 0x7f || error showByte)
3603 && (state' /= 0 || error (showByte ++ showTable)))
3604 (Tuple' state' codep')
3609 let !(Ptr addr) = table
3610 end = table `plusPtr` 364
3611 in A.Array (ForeignPtr addr undefined) end end :: A.Array Word8
3612 showByte = "Streamly: decode0: byte: " ++ show byte
3613 showTable = " table: " ++ show utf8table
3615-- When the state is not 0
3616{-# INLINE decode1 #-}
3622 -> Tuple' DecodeState CodePoint
3623decode1 table state codep byte =
3624 -- Remember codep is Int type!
3625 -- Can it be unsafe to convert the resulting Int to Char?
3626 let !t = table `unsafePeekElemOff` fromIntegral byte
3627 !codep' = (fromIntegral byte .&. 0x3f) .|. (codep `shiftL` 6)
3628 !state' = table `unsafePeekElemOff`
3629 (256 + fromIntegral state + fromIntegral t)
3630 in assert (codep' <= 0x10FFFF
3631 || error (showByte ++ showState state codep))
3632 (Tuple' state' codep')
3636 let !(Ptr addr) = table
3637 end = table `plusPtr` 364
3638 in A.Array (ForeignPtr addr undefined) end end :: A.Array Word8
3639 showByte = "Streamly: decode1: byte: " ++ show byte
3641 " state: " ++ show st ++
3642 " codepoint: " ++ show cp ++
3643 " table: " ++ show utf8table
3645-- We can divide the errors in three general categories:
3646-- * A non-starter was encountered in a begin state
3647-- * A starter was encountered without completing a codepoint
3648-- * The last codepoint was not complete (input underflow)
3650data DecodeError = DecodeError !DecodeState !CodePoint deriving Show
3653 = FreshPointDecodeInit s
3654 | FreshPointDecodeInit1 s Word8
3655 | FreshPointDecodeFirst s Word8
3656 | FreshPointDecoding s !DecodeState !CodePoint
3657 | YieldAndContinue a (FreshPoint s a)
3660-- XXX Add proper error messages
3661-- XXX Implement this in terms of decodeUtf8Either
3662{-# INLINE_NORMAL decodeUtf8With #-}
3663decodeUtf8With :: Monad m => CodingFailureMode -> Stream m Word8 -> Stream m Char
3664decodeUtf8With cfm (Stream step state) =
3665 let Array p _ _ = utf8d
3666 !ptr = (unsafeForeignPtrToPtr p)
3667 in Stream (step' ptr) (FreshPointDecodeInit state)
3669 {-# INLINE transliterateOrError #-}
3670 transliterateOrError e s =
3672 ErrorOnCodingFailure -> error e
3673 TransliterateCodingFailure -> YieldAndContinue replacementChar s
3674 {-# INLINE inputUnderflow #-}
3677 ErrorOnCodingFailure ->
3678 error "Streamly.Streams.StreamD.decodeUtf8With: Input Underflow"
3679 TransliterateCodingFailure -> YieldAndContinue replacementChar Done
3680 {-# INLINE_LATE step' #-}
3681 step' _ gst (FreshPointDecodeInit st) = do
3682 r <- step (adaptState gst) st
3684 Yield x s -> Skip (FreshPointDecodeInit1 s x)
3685 Skip s -> Skip (FreshPointDecodeInit s)
3688 step' _ _ (FreshPointDecodeInit1 st x) = do
3689 -- Note: It is important to use a ">" instead of a "<=" test
3690 -- here for GHC to generate code layout for default branch
3691 -- prediction for the common case. This is fragile and might
3692 -- change with the compiler versions, we need a more reliable
3693 -- "likely" primitive to control branch predication.
3696 return $ Skip $ YieldAndContinue
3697 (unsafeChr (fromIntegral x))
3698 (FreshPointDecodeInit st)
3699 -- Using a separate state here generates a jump to a
3700 -- separate code block in the core which seems to perform
3701 -- slightly better for the non-ascii case.
3702 True -> return $ Skip $ FreshPointDecodeFirst st x
3704 -- XXX should we merge it with FreshPointDecodeInit1?
3705 step' table _ (FreshPointDecodeFirst st x) = do
3706 let (Tuple' sv cp) = decode0 table x
3711 transliterateOrError
3712 "Streamly.Streams.StreamD.decodeUtf8With: Invalid UTF8 codepoint encountered"
3713 (FreshPointDecodeInit st)
3714 0 -> error "unreachable state"
3715 _ -> Skip (FreshPointDecoding st sv cp)
3717 -- We recover by trying the new byte x a starter of a new codepoint.
3718 -- XXX need to use the same recovery in array decoding routine as well
3719 step' table gst (FreshPointDecoding st statePtr codepointPtr) = do
3720 r <- step (adaptState gst) st
3723 let (Tuple' sv cp) = decode1 table statePtr codepointPtr x
3726 0 -> Skip $ YieldAndContinue (unsafeChr cp)
3727 (FreshPointDecodeInit s)
3730 transliterateOrError
3731 "Streamly.Streams.StreamD.decodeUtf8With: Invalid UTF8 codepoint encountered"
3732 (FreshPointDecodeInit1 s x)
3733 _ -> Skip (FreshPointDecoding s sv cp)
3734 Skip s -> return $ Skip (FreshPointDecoding s statePtr codepointPtr)
3735 Stop -> return $ Skip inputUnderflow
3737 step' _ _ (YieldAndContinue c s) = return $ Yield c s
3738 step' _ _ Done = return Stop
3740{-# INLINE decodeUtf8 #-}
3741decodeUtf8 :: Monad m => Stream m Word8 -> Stream m Char
3742decodeUtf8 = decodeUtf8With ErrorOnCodingFailure
3744{-# INLINE decodeUtf8Lenient #-}
3745decodeUtf8Lenient :: Monad m => Stream m Word8 -> Stream m Char
3746decodeUtf8Lenient = decodeUtf8With TransliterateCodingFailure
3748{-# INLINE_NORMAL resumeDecodeUtf8Either #-}
3749resumeDecodeUtf8Either
3754 -> Stream m (Either DecodeError Char)
3755resumeDecodeUtf8Either dst codep (Stream step state) =
3756 let Array p _ _ = utf8d
3757 !ptr = (unsafeForeignPtrToPtr p)
3760 then FreshPointDecodeInit state
3761 else FreshPointDecoding state dst codep
3762 in Stream (step' ptr) stt
3764 {-# INLINE_LATE step' #-}
3765 step' _ gst (FreshPointDecodeInit st) = do
3766 r <- step (adaptState gst) st
3768 Yield x s -> Skip (FreshPointDecodeInit1 s x)
3769 Skip s -> Skip (FreshPointDecodeInit s)
3772 step' _ _ (FreshPointDecodeInit1 st x) = do
3773 -- Note: It is important to use a ">" instead of a "<=" test
3774 -- here for GHC to generate code layout for default branch
3775 -- prediction for the common case. This is fragile and might
3776 -- change with the compiler versions, we need a more reliable
3777 -- "likely" primitive to control branch predication.
3780 return $ Skip $ YieldAndContinue
3781 (Right $ unsafeChr (fromIntegral x))
3782 (FreshPointDecodeInit st)
3783 -- Using a separate state here generates a jump to a
3784 -- separate code block in the core which seems to perform
3785 -- slightly better for the non-ascii case.
3786 True -> return $ Skip $ FreshPointDecodeFirst st x
3788 -- XXX should we merge it with FreshPointDecodeInit1?
3789 step' table _ (FreshPointDecodeFirst st x) = do
3790 let (Tuple' sv cp) = decode0 table x
3794 Skip $ YieldAndContinue (Left $ DecodeError 0 (fromIntegral x))
3795 (FreshPointDecodeInit st)
3796 0 -> error "unreachable state"
3797 _ -> Skip (FreshPointDecoding st sv cp)
3799 -- We recover by trying the new byte x a starter of a new codepoint.
3800 -- XXX need to use the same recovery in array decoding routine as well
3801 step' table gst (FreshPointDecoding st statePtr codepointPtr) = do
3802 r <- step (adaptState gst) st
3805 let (Tuple' sv cp) = decode1 table statePtr codepointPtr x
3808 0 -> Skip $ YieldAndContinue (Right $ unsafeChr cp)
3809 (FreshPointDecodeInit s)
3811 Skip $ YieldAndContinue (Left $ DecodeError statePtr codepointPtr)
3812 (FreshPointDecodeInit1 s x)
3813 _ -> Skip (FreshPointDecoding s sv cp)
3814 Skip s -> return $ Skip (FreshPointDecoding s statePtr codepointPtr)
3815 Stop -> return $ Skip $ YieldAndContinue (Left $ DecodeError statePtr codepointPtr) Done
3817 step' _ _ (YieldAndContinue c s) = return $ Yield c s
3818 step' _ _ Done = return Stop
3820{-# INLINE_NORMAL decodeUtf8Either #-}
3821decodeUtf8Either :: Monad m
3822 => Stream m Word8 -> Stream m (Either DecodeError Char)
3823decodeUtf8Either = resumeDecodeUtf8Either 0 0
3825data FlattenState s a
3826 = OuterLoop s !(Maybe (DecodeState, CodePoint))
3827 | InnerLoopDecodeInit s (ForeignPtr a) !(Ptr a) !(Ptr a)
3828 | InnerLoopDecodeFirst s (ForeignPtr a) !(Ptr a) !(Ptr a) Word8
3829 | InnerLoopDecoding s (ForeignPtr a) !(Ptr a) !(Ptr a)
3830 !DecodeState !CodePoint
3831 | YAndC !Char (FlattenState s a) -- These constructors can be
3832 -- encoded in the FreshPoint
3833 -- type, I prefer to keep these
3834 -- flat even though that means
3835 -- coming up with new names
3838-- The normal decodeUtf8 above should fuse with flattenArrays
3839-- to create this exact code but it doesn't for some reason, as of now this
3840-- remains the fastest way I could figure out to decodeUtf8.
3842-- XXX Add Proper error messages
3843{-# INLINE_NORMAL decodeUtf8ArraysWith #-}
3844decodeUtf8ArraysWith ::
3846 => CodingFailureMode
3847 -> Stream m (A.Array Word8)
3849decodeUtf8ArraysWith cfm (Stream step state) =
3850 let Array p _ _ = utf8d
3851 !ptr = (unsafeForeignPtrToPtr p)
3852 in Stream (step' ptr) (OuterLoop state Nothing)
3854 {-# INLINE transliterateOrError #-}
3855 transliterateOrError e s =
3857 ErrorOnCodingFailure -> error e
3858 TransliterateCodingFailure -> YAndC replacementChar s
3859 {-# INLINE inputUnderflow #-}
3862 ErrorOnCodingFailure ->
3864 "Streamly.Streams.StreamD.decodeUtf8ArraysWith: Input Underflow"
3865 TransliterateCodingFailure -> YAndC replacementChar D
3866 {-# INLINE_LATE step' #-}
3867 step' _ gst (OuterLoop st Nothing) = do
3868 r <- step (adaptState gst) st
3871 Yield A.Array {..} s ->
3872 let p = unsafeForeignPtrToPtr aStart
3873 in Skip (InnerLoopDecodeInit s aStart p aEnd)
3874 Skip s -> Skip (OuterLoop s Nothing)
3876 step' _ gst (OuterLoop st dst@(Just (ds, cp))) = do
3877 r <- step (adaptState gst) st
3880 Yield A.Array {..} s ->
3881 let p = unsafeForeignPtrToPtr aStart
3882 in Skip (InnerLoopDecoding s aStart p aEnd ds cp)
3883 Skip s -> Skip (OuterLoop s dst)
3884 Stop -> Skip inputUnderflow
3885 step' _ _ (InnerLoopDecodeInit st startf p end)
3887 liftIO $ touchForeignPtr startf
3888 return $ Skip $ OuterLoop st Nothing
3889 step' _ _ (InnerLoopDecodeInit st startf p end) = do
3890 x <- liftIO $ peek p
3891 -- Note: It is important to use a ">" instead of a "<=" test here for
3892 -- GHC to generate code layout for default branch prediction for the
3893 -- common case. This is fragile and might change with the compiler
3894 -- versions, we need a more reliable "likely" primitive to control
3895 -- branch predication.
3898 return $ Skip $ YAndC
3899 (unsafeChr (fromIntegral x))
3900 (InnerLoopDecodeInit st startf (p `plusPtr` 1) end)
3901 -- Using a separate state here generates a jump to a separate code
3902 -- block in the core which seems to perform slightly better for the
3904 True -> return $ Skip $ InnerLoopDecodeFirst st startf p end x
3906 step' table _ (InnerLoopDecodeFirst st startf p end x) = do
3907 let (Tuple' sv cp) = decode0 table x
3912 transliterateOrError
3913 "Streamly.Streams.StreamD.decodeUtf8ArraysWith: Invalid UTF8 codepoint encountered"
3914 (InnerLoopDecodeInit st startf (p `plusPtr` 1) end)
3915 0 -> error "unreachable state"
3916 _ -> Skip (InnerLoopDecoding st startf (p `plusPtr` 1) end sv cp)
3917 step' _ _ (InnerLoopDecoding st startf p end sv cp)
3919 liftIO $ touchForeignPtr startf
3920 return $ Skip $ OuterLoop st (Just (sv, cp))
3921 step' table _ (InnerLoopDecoding st startf p end statePtr codepointPtr) = do
3922 x <- liftIO $ peek p
3923 let (Tuple' sv cp) = decode1 table statePtr codepointPtr x
3930 (InnerLoopDecodeInit st startf (p `plusPtr` 1) end)
3933 transliterateOrError
3934 "Streamly.Streams.StreamD.decodeUtf8ArraysWith: Invalid UTF8 codepoint encountered"
3935 (InnerLoopDecodeInit st startf (p `plusPtr` 1) end)
3936 _ -> Skip (InnerLoopDecoding st startf (p `plusPtr` 1) end sv cp)
3937 step' _ _ (YAndC c s) = return $ Yield c s
3938 step' _ _ D = return Stop
3940{-# INLINE decodeUtf8Arrays #-}
3943 => Stream m (A.Array Word8)
3945decodeUtf8Arrays = decodeUtf8ArraysWith ErrorOnCodingFailure
3947{-# INLINE decodeUtf8ArraysLenient #-}
3948decodeUtf8ArraysLenient ::
3950 => Stream m (A.Array Word8)
3952decodeUtf8ArraysLenient = decodeUtf8ArraysWith TransliterateCodingFailure
3954data WList = WCons !Word8 !WList | WNil
3956data EncodeState s = EncodeState s !WList
3958-- More yield points improve performance, but I am not sure if they can cause
3959-- too much code bloat or some trouble with fusion. So keeping only two yield
3960-- points for now, one for the ascii chars (fast path) and one for all other
3961-- paths (slow path).
3962{-# INLINE_NORMAL encodeUtf8 #-}
3963encodeUtf8 :: Monad m => Stream m Char -> Stream m Word8
3964encodeUtf8 (Stream step state) = Stream step' (EncodeState state WNil)
3966 {-# INLINE_LATE step' #-}
3967 step' gst (EncodeState st WNil) = do
3968 r <- step (adaptState gst) st
3975 Yield (fromIntegral x) (EncodeState s WNil)
3976 | x <= 0x7FF -> Skip (EncodeState s (ord2 c))
3980 "Streamly.Streams.StreamD.encodeUtf8: Encountered a surrogate"
3981 else Skip (EncodeState s (ord3 c))
3982 | otherwise -> Skip (EncodeState s (ord4 c))
3983 Skip s -> Skip (EncodeState s WNil)
3985 step' _ (EncodeState s (WCons x xs)) = return $ Yield x (EncodeState s xs)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list