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.Internal.Data.Stream.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.Internal.Data.Stream.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.Internal.Data.Stream.StreamD
50#if __GLASGOW_HASKELL__ >= 800
51 , Stream (Stream, UnStream)
71 -- ** Specialized Generation
72 -- | Generate a monadic stream from a seed.
85 , enumerateFromStepIntegral
86 , enumerateFromIntegral
87 , enumerateFromThenIntegral
88 , enumerateFromToIntegral
89 , enumerateFromThenToIntegral
91 , enumerateFromStepNum
94 , enumerateFromToFractional
95 , enumerateFromThenToFractional
98 -- | Transform an input structure into a stream.
99 -- | Direct style stream does not support @fromFoldable@.
128 -- ** Specialized Folds
153 -- ** Flattening nested streams
156 , ConcatMapUState (..)
158 , ConcatUnfoldInterleaveState (..)
159 , concatUnfoldInterleave
160 , concatUnfoldRoundrobin
163 , InterleaveState(..)
168 , roundRobin -- interleaveFair?/ParallelFair
201 -- | Transform a stream into another type.
218 -- ** By folding (scans)
261 , intersperseSuffixBySpan
287 -- * Transformation comprehensions
299 -- * Concurrent Application
306import Control.Concurrent (killThread, myThreadId, takeMVar, threadDelay)
307import Control.Exception
308 (Exception, SomeException, AsyncException, fromException)
309import Control.Monad (void, when)
310import Control.Monad.Catch (MonadCatch, throwM)
311import Control.Monad.IO.Class (MonadIO(..))
312import Control.Monad.Reader (ReaderT)
313import Control.Monad.State.Strict (StateT)
314import Control.Monad.Trans (MonadTrans(lift))
315import Data.Bits (shiftR, shiftL, (.|.), (.&.))
316import Data.Functor.Identity (Identity(..))
317import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef, modifyIORef')
318import Data.Maybe (fromJust, isJust, isNothing)
319import Data.Word (Word32)
320import Foreign.Ptr (Ptr)
321import Foreign.Storable (Storable(..))
322import GHC.Types (SPEC(..))
323import System.Mem (performMajorGC)
325 hiding (map, mapM, mapM_, repeat, foldr, last, take, filter,
326 takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,
327 notElem, null, head, tail, zipWith, lookup, foldr1, sequence,
328 (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat,
331import qualified Control.Monad.Catch as MC
332import qualified Control.Monad.Reader as Reader
333import qualified Control.Monad.State.Strict as State
334import qualified Prelude
336import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
337import Streamly.Internal.Memory.Array.Types (Array(..))
338import Streamly.Internal.Data.Fold.Types (Fold(..))
339import Streamly.Internal.Data.Pipe.Types (Pipe(..), PipeState(..))
340import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
341import Streamly.Internal.Data.Unfold.Types (Unfold(..))
343import Streamly.Internal.Data.Stream.StreamD.Type
344import Streamly.Internal.Data.SVar
345import Streamly.Internal.Data.Stream.SVar (fromConsumer, pushToFold)
347import qualified Streamly.Internal.Data.Pipe.Types as Pipe
348import qualified Streamly.Internal.Memory.Array.Types as A
349import qualified Streamly.Internal.Data.Fold as FL
350import qualified Streamly.Memory.Ring as RB
351import qualified Streamly.Internal.Data.Stream.StreamK as K
353------------------------------------------------------------------------------
355------------------------------------------------------------------------------
357-- | An empty 'Stream'.
358{-# INLINE_NORMAL nil #-}
359nil :: Monad m => Stream m a
360nil = Stream (\_ _ -> return Stop) ()
362-- | An empty 'Stream' with a side effect.
363{-# INLINE_NORMAL nilM #-}
364nilM :: Monad m => m b -> Stream m a
365nilM m = Stream (\_ _ -> m >> return Stop) ()
367{-# INLINE_NORMAL consM #-}
368consM :: Monad m => m a -> Stream m a -> Stream m a
369consM m (Stream step state) = Stream step1 Nothing
371 {-# INLINE_LATE step1 #-}
372 step1 _ Nothing = m >>= \x -> return $ Yield x (Just state)
373 step1 gst (Just st) = do
377 Yield a s -> Yield a (Just s)
378 Skip s -> Skip (Just s)
381-- XXX implement in terms of consM?
382-- cons x = consM (return x)
384-- | Can fuse but has O(n^2) complexity.
385{-# INLINE_NORMAL cons #-}
386cons :: Monad m => a -> Stream m a -> Stream m a
387cons x (Stream step state) = Stream step1 Nothing
389 {-# INLINE_LATE step1 #-}
390 step1 _ Nothing = return $ Yield x (Just state)
391 step1 gst (Just st) = do
395 Yield a s -> Yield a (Just s)
396 Skip s -> Skip (Just s)
399-------------------------------------------------------------------------------
401-------------------------------------------------------------------------------
403-- Does not fuse, has the same performance as the StreamK version.
404{-# INLINE_NORMAL uncons #-}
405uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
406uncons (UnStream step state) = go state
409 r <- step defState st
411 Yield x s -> return $ Just (x, Stream step s)
413 Stop -> return Nothing
415------------------------------------------------------------------------------
416-- Generation by unfold
417------------------------------------------------------------------------------
419{-# INLINE_NORMAL unfoldrM #-}
420unfoldrM :: Monad m => (s -> m (Maybe (a, s))) -> s -> Stream m a
421unfoldrM next state = Stream step state
423 {-# INLINE_LATE step #-}
427 Just (x, s) -> Yield x s
430{-# INLINE_LATE unfoldr #-}
431unfoldr :: Monad m => (s -> Maybe (a, s)) -> s -> Stream m a
432unfoldr f = unfoldrM (return . f)
434-- | Convert an 'Unfold' into a 'Stream' by supplying it a seed.
436{-# INLINE_NORMAL unfold #-}
437unfold :: Monad m => Unfold m a b -> a -> Stream m b
438unfold (Unfold ustep inject) seed = Stream step Nothing
440 {-# INLINE_LATE step #-}
441 step _ Nothing = inject seed >>= return . Skip . Just
442 step _ (Just st) = do
445 Yield x s -> Yield x (Just s)
446 Skip s -> Skip (Just s)
449------------------------------------------------------------------------------
450-- Specialized Generation
451------------------------------------------------------------------------------
453{-# INLINE_NORMAL repeatM #-}
454repeatM :: Monad m => m a -> Stream m a
455repeatM x = Stream (\_ _ -> x >>= \r -> return $ Yield r ()) ()
457{-# INLINE_NORMAL repeat #-}
458repeat :: Monad m => a -> Stream m a
459repeat x = Stream (\_ _ -> return $ Yield x ()) ()
461{-# INLINE_NORMAL iterateM #-}
462iterateM :: Monad m => (a -> m a) -> m a -> Stream m a
463iterateM step = Stream (\_ st -> st >>= \x -> return $ Yield x (step x))
465{-# INLINE_NORMAL iterate #-}
466iterate :: Monad m => (a -> a) -> a -> Stream m a
467iterate step st = iterateM (return . step) (return st)
469{-# INLINE_NORMAL replicateM #-}
470replicateM :: forall m a. Monad m => Int -> m a -> Stream m a
471replicateM n p = Stream step n
473 {-# INLINE_LATE step #-}
475 | i <= 0 = return Stop
478 return $ Yield x (i - 1)
480{-# INLINE_NORMAL replicate #-}
481replicate :: Monad m => Int -> a -> Stream m a
482replicate n x = replicateM n (return x)
484-- This would not work properly for floats, therefore we put an Integral
486-- | Can be used to enumerate unbounded integrals. This does not check for
487-- overflow or underflow for bounded integrals.
488{-# INLINE_NORMAL enumerateFromStepIntegral #-}
489enumerateFromStepIntegral :: (Integral a, Monad m) => a -> a -> Stream m a
490enumerateFromStepIntegral from stride =
491 from `seq` stride `seq` Stream step from
493 {-# INLINE_LATE step #-}
494 step _ !x = return $ Yield x $! (x + stride)
496-- We are assuming that "to" is constrained by the type to be within
498{-# INLINE enumerateFromToIntegral #-}
499enumerateFromToIntegral :: (Monad m, Integral a) => a -> a -> Stream m a
500enumerateFromToIntegral from to =
501 takeWhile (<= to) $ enumerateFromStepIntegral from 1
503{-# INLINE enumerateFromIntegral #-}
504enumerateFromIntegral :: (Monad m, Integral a, Bounded a) => a -> Stream m a
505enumerateFromIntegral from = enumerateFromToIntegral from maxBound
507data EnumState a = EnumInit | EnumYield a a a | EnumStop
509{-# INLINE_NORMAL enumerateFromThenToIntegralUp #-}
510enumerateFromThenToIntegralUp
511 :: (Monad m, Integral a)
512 => a -> a -> a -> Stream m a
513enumerateFromThenToIntegralUp from next to = Stream step EnumInit
515 {-# INLINE_LATE step #-}
521 else Yield from EnumStop
522 else -- from <= next <= to
523 let stride = next - from
524 in Skip $ EnumYield from stride (to - stride)
526 step _ (EnumYield x stride toMinus) =
529 then Yield x EnumStop
530 else Yield x $ EnumYield (x + stride) stride toMinus
532 step _ EnumStop = return Stop
534{-# INLINE_NORMAL enumerateFromThenToIntegralDn #-}
535enumerateFromThenToIntegralDn
536 :: (Monad m, Integral a)
537 => a -> a -> a -> Stream m a
538enumerateFromThenToIntegralDn from next to = Stream step EnumInit
540 {-# INLINE_LATE step #-}
542 return $ if to > next
545 else Yield from EnumStop
546 else -- from >= next >= to
547 let stride = next - from
548 in Skip $ EnumYield from stride (to - stride)
550 step _ (EnumYield x stride toMinus) =
553 then Yield x EnumStop
554 else Yield x $ EnumYield (x + stride) stride toMinus
556 step _ EnumStop = return Stop
558{-# INLINE_NORMAL enumerateFromThenToIntegral #-}
559enumerateFromThenToIntegral
560 :: (Monad m, Integral a)
561 => a -> a -> a -> Stream m a
562enumerateFromThenToIntegral from next to
563 | next >= from = enumerateFromThenToIntegralUp from next to
564 | otherwise = enumerateFromThenToIntegralDn from next to
566{-# INLINE_NORMAL enumerateFromThenIntegral #-}
567enumerateFromThenIntegral
568 :: (Monad m, Integral a, Bounded a)
569 => a -> a -> Stream m a
570enumerateFromThenIntegral from next =
572 then enumerateFromThenToIntegralUp from next maxBound
573 else enumerateFromThenToIntegralDn from next minBound
575-- For floating point numbers if the increment is less than the precision then
576-- it just gets lost. Therefore we cannot always increment it correctly by just
578-- 9007199254740992 + 1 + 1 :: Double => 9.007199254740992e15
579-- 9007199254740992 + 2 :: Double => 9.007199254740994e15
581-- Instead we accumulate the increment counter and compute the increment
582-- every time before adding it to the starting number.
584-- This works for Integrals as well as floating point numbers, but
585-- enumerateFromStepIntegral is faster for integrals.
586{-# INLINE_NORMAL enumerateFromStepNum #-}
587enumerateFromStepNum :: (Monad m, Num a) => a -> a -> Stream m a
588enumerateFromStepNum from stride = Stream step 0
590 {-# INLINE_LATE step #-}
591 step _ !i = return $ (Yield $! (from + i * stride)) $! (i + 1)
593{-# INLINE_NORMAL numFrom #-}
594numFrom :: (Monad m, Num a) => a -> Stream m a
595numFrom from = enumerateFromStepNum from 1
597{-# INLINE_NORMAL numFromThen #-}
598numFromThen :: (Monad m, Num a) => a -> a -> Stream m a
599numFromThen from next = enumerateFromStepNum from (next - from)
601-- We cannot write a general function for Num. The only way to write code
602-- portable between the two is to use a 'Real' constraint and convert between
603-- Fractional and Integral using fromRational which is horribly slow.
604{-# INLINE_NORMAL enumerateFromToFractional #-}
605enumerateFromToFractional
606 :: (Monad m, Fractional a, Ord a)
607 => a -> a -> Stream m a
608enumerateFromToFractional from to =
609 takeWhile (<= to + 1 / 2) $ enumerateFromStepNum from 1
611{-# INLINE_NORMAL enumerateFromThenToFractional #-}
612enumerateFromThenToFractional
613 :: (Monad m, Fractional a, Ord a)
614 => a -> a -> a -> Stream m a
615enumerateFromThenToFractional from next to =
616 takeWhile predicate $ numFromThen from next
618 mid = (next - from) / 2
619 predicate | next >= from = (<= to + mid)
620 | otherwise = (>= to + mid)
622-------------------------------------------------------------------------------
623-- Generation by Conversion
624-------------------------------------------------------------------------------
626{-# INLINE_NORMAL fromIndicesM #-}
627fromIndicesM :: Monad m => (Int -> m a) -> Stream m a
628fromIndicesM gen = Stream step 0
630 {-# INLINE_LATE step #-}
633 return $ Yield x (i + 1)
635{-# INLINE fromIndices #-}
636fromIndices :: Monad m => (Int -> a) -> Stream m a
637fromIndices gen = fromIndicesM (return . gen)
639{-# INLINE_NORMAL generateM #-}
640generateM :: Monad m => Int -> (Int -> m a) -> Stream m a
641generateM n gen = n `seq` Stream step 0
643 {-# INLINE_LATE step #-}
644 step _ i | i < n = do
646 return $ Yield x (i + 1)
647 | otherwise = return Stop
649{-# INLINE generate #-}
650generate :: Monad m => Int -> (Int -> a) -> Stream m a
651generate n gen = generateM n (return . gen)
653-- XXX we need the MonadAsync constraint because of a rewrite rule.
654-- | Convert a list of monadic actions to a 'Stream'
655{-# INLINE_LATE fromListM #-}
656fromListM :: MonadAsync m => [m a] -> Stream m a
657fromListM = Stream step
659 {-# INLINE_LATE step #-}
660 step _ (m:ms) = m >>= \x -> return $ Yield x ms
661 step _ [] = return Stop
663{-# INLINE toStreamD #-}
664toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a
665toStreamD = fromStreamK . K.toStream
667-------------------------------------------------------------------------------
668-- Generation from SVar
669-------------------------------------------------------------------------------
671data FromSVarState t m a =
673 | FromSVarRead (SVar t m a)
674 | FromSVarLoop (SVar t m a) [ChildEvent a]
675 | FromSVarDone (SVar t m a)
677{-# INLINE_NORMAL fromSVar #-}
678fromSVar :: (MonadAsync m) => SVar t m a -> Stream m a
679fromSVar svar = Stream step FromSVarInit
682 {-# INLINE_LATE step #-}
683 step _ FromSVarInit = do
684 ref <- liftIO $ newIORef ()
685 _ <- liftIO $ mkWeakIORef ref hook
686 -- when this copy of svar gets garbage collected "ref" will get
687 -- garbage collected and our GC hook will be called.
688 let sv = svar{svarRef = Just ref}
689 return $ Skip (FromSVarRead sv)
693 {-# NOINLINE hook #-}
695 when (svarInspectMode svar) $ do
696 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
698 printSVar svar "SVar Garbage Collected"
700 -- If there are any SVars referenced by this SVar a GC will prompt
701 -- them to be cleaned up quickly.
702 when (svarInspectMode svar) performMajorGC
704 step _ (FromSVarRead sv) = do
705 list <- readOutputQ sv
706 -- Reversing the output is important to guarantee that we process the
707 -- outputs in the same order as they were generated by the constituent
709 return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
711 step _ (FromSVarLoop sv []) = do
712 done <- postProcess sv
713 return $ Skip $ if done
714 then (FromSVarDone sv)
715 else (FromSVarRead sv)
717 step _ (FromSVarLoop sv (ev : es)) = do
719 ChildYield a -> return $ Yield a (FromSVarLoop sv es)
720 ChildStop tid e -> do
724 stop <- shouldStop tid
727 liftIO (cleanupSVar sv)
728 return $ Skip (FromSVarDone sv)
729 else return $ Skip (FromSVarLoop sv es)
731 case fromException ex of
733 return $ Skip (FromSVarLoop sv es)
734 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
738 case svarStopStyle sv of
739 StopNone -> return False
740 StopAny -> return True
742 sid <- liftIO $ readIORef (svarStopBy sv)
743 return $ if tid == sid then True else False
745 step _ (FromSVarDone sv) = do
746 when (svarInspectMode sv) $ do
747 t <- liftIO $ getTime Monotonic
748 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
749 liftIO $ printSVar sv "SVar Done"
752-------------------------------------------------------------------------------
753-- Process events received by a fold consumer from a stream producer
754-------------------------------------------------------------------------------
756{-# INLINE_NORMAL fromProducer #-}
757fromProducer :: (MonadAsync m) => SVar t m a -> Stream m a
758fromProducer svar = Stream step (FromSVarRead svar)
761 {-# INLINE_LATE step #-}
762 step _ (FromSVarRead sv) = do
763 list <- readOutputQ sv
764 -- Reversing the output is important to guarantee that we process the
765 -- outputs in the same order as they were generated by the constituent
767 return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
769 step _ (FromSVarLoop sv []) = return $ Skip $ FromSVarRead sv
770 step _ (FromSVarLoop sv (ev : es)) = do
772 ChildYield a -> return $ Yield a (FromSVarLoop sv es)
773 ChildStop tid e -> do
777 sendStopToProducer sv
778 return $ Skip (FromSVarDone sv)
779 Just _ -> error "Bug: fromProducer: received exception"
781 step _ (FromSVarDone sv) = do
782 when (svarInspectMode sv) $ do
783 t <- liftIO $ getTime Monotonic
784 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
785 liftIO $ printSVar sv "SVar Done"
788 step _ FromSVarInit = undefined
790-------------------------------------------------------------------------------
791-- Hoisting the inner monad
792-------------------------------------------------------------------------------
794{-# INLINE_NORMAL hoist #-}
795hoist :: Monad n => (forall x. m x -> n x) -> Stream m a -> Stream n a
796hoist f (Stream step state) = (Stream step' state)
798 {-# INLINE_LATE step' #-}
800 r <- f $ step (adaptState gst) st
802 Yield x s -> Yield x s
806{-# INLINE generally #-}
807generally :: Monad m => Stream Identity a -> Stream m a
808generally = hoist (return . runIdentity)
810{-# INLINE_NORMAL liftInner #-}
811liftInner :: (Monad m, MonadTrans t, Monad (t m))
812 => Stream m a -> Stream (t m) a
813liftInner (Stream step state) = Stream step' state
815 {-# INLINE_LATE step' #-}
817 r <- lift $ step (adaptState gst) st
819 Yield x s -> Yield x s
823{-# INLINE_NORMAL runReaderT #-}
824runReaderT :: Monad m => s -> Stream (ReaderT s m) a -> Stream m a
825runReaderT sval (Stream step state) = Stream step' state
827 {-# INLINE_LATE step' #-}
829 r <- Reader.runReaderT (step (adaptState gst) st) sval
831 Yield x s -> Yield x s
835{-# INLINE_NORMAL evalStateT #-}
836evalStateT :: Monad m => s -> Stream (StateT s m) a -> Stream m a
837evalStateT sval (Stream step state) = Stream step' (state, sval)
839 {-# INLINE_LATE step' #-}
840 step' gst (st, sv) = do
841 (r, sv') <- State.runStateT (step (adaptState gst) st) sv
843 Yield x s -> Yield x (s, sv')
844 Skip s -> Skip (s, sv')
847{-# INLINE_NORMAL runStateT #-}
848runStateT :: Monad m => s -> Stream (StateT s m) a -> Stream m (s, a)
849runStateT sval (Stream step state) = Stream step' (state, sval)
851 {-# INLINE_LATE step' #-}
852 step' gst (st, sv) = do
853 (r, sv') <- State.runStateT (step (adaptState gst) st) sv
855 Yield x s -> Yield (sv', x) (s, sv')
856 Skip s -> Skip (s, sv')
859------------------------------------------------------------------------------
860-- Elimination by Folds
861------------------------------------------------------------------------------
863------------------------------------------------------------------------------
865------------------------------------------------------------------------------
867{-# INLINE_NORMAL foldr1 #-}
868foldr1 :: Monad m => (a -> a -> a) -> Stream m a -> m (Maybe a)
872 Nothing -> return Nothing
873 Just (h, t) -> fmap Just (foldr f h t)
875------------------------------------------------------------------------------
877------------------------------------------------------------------------------
879{-# INLINE_NORMAL foldlT #-}
880foldlT :: (Monad m, Monad (s m), MonadTrans s)
881 => (s m b -> a -> s m b) -> s m b -> Stream m a -> s m b
882foldlT fstep begin (Stream step state) = go SPEC begin state
885 r <- lift $ step defState st
887 Yield x s -> go SPEC (fstep acc x) s
888 Skip s -> go SPEC acc s
891-- Note, this is going to have horrible performance, because of the nature of
892-- the stream type (i.e. direct stream vs CPS). Its only for reference, it is
893-- likely be practically unusable.
894{-# INLINE_NORMAL foldlS #-}
896 => (Stream m b -> a -> Stream m b) -> Stream m b -> Stream m a -> Stream m b
897foldlS fstep begin (Stream step state) = Stream step' (Left (state, begin))
899 step' gst (Left (st, acc)) = do
900 r <- step (adaptState gst) st
902 Yield x s -> Skip (Left (s, fstep acc x))
903 Skip s -> Skip (Left (s, acc))
904 Stop -> Skip (Right acc)
906 step' gst (Right (Stream stp stt)) = do
907 r <- stp (adaptState gst) stt
909 Yield x s -> Yield x (Right (Stream stp s))
910 Skip s -> Skip (Right (Stream stp s))
913------------------------------------------------------------------------------
915------------------------------------------------------------------------------
917-- | Run a streaming composition, discard the results.
918{-# INLINE_LATE drain #-}
919drain :: Monad m => Stream m a -> m ()
920-- drain = foldrM (\_ xs -> xs) (return ())
921drain (Stream step state) = go SPEC state
924 r <- step defState st
926 Yield _ s -> go SPEC s
930{-# INLINE_NORMAL null #-}
931null :: Monad m => Stream m a -> m Bool
932null m = foldrM (\_ _ -> return False) (return True) m
934{-# INLINE_NORMAL head #-}
935head :: Monad m => Stream m a -> m (Maybe a)
936head m = foldrM (\x _ -> return (Just x)) (return Nothing) m
938{-# INLINE_NORMAL headElse #-}
939headElse :: Monad m => a -> Stream m a -> m a
940headElse a m = foldrM (\x _ -> return x) (return a) m
942-- Does not fuse, has the same performance as the StreamK version.
943{-# INLINE_NORMAL tail #-}
944tail :: Monad m => Stream m a -> m (Maybe (Stream m a))
945tail (UnStream step state) = go state
948 r <- step defState st
950 Yield _ s -> return (Just $ Stream step s)
952 Stop -> return Nothing
954-- XXX will it fuse? need custom impl?
955{-# INLINE_NORMAL last #-}
956last :: Monad m => Stream m a -> m (Maybe a)
957last = foldl' (\_ y -> Just y) Nothing
959{-# INLINE_NORMAL elem #-}
960elem :: (Monad m, Eq a) => a -> Stream m a -> m Bool
961-- elem e m = foldrM (\x xs -> if x == e then return True else xs) (return False) m
962elem e (Stream step state) = go state
965 r <- step defState st
968 | x == e -> return True
973{-# INLINE_NORMAL notElem #-}
974notElem :: (Monad m, Eq a) => a -> Stream m a -> m Bool
975notElem e s = fmap not (elem e s)
977{-# INLINE_NORMAL all #-}
978all :: Monad m => (a -> Bool) -> Stream m a -> m Bool
979-- all p m = foldrM (\x xs -> if p x then xs else return False) (return True) m
980all p (Stream step state) = go state
983 r <- step defState st
987 | otherwise -> return False
991{-# INLINE_NORMAL any #-}
992any :: Monad m => (a -> Bool) -> Stream m a -> m Bool
993-- any p m = foldrM (\x xs -> if p x then return True else xs) (return False) m
994any p (Stream step state) = go state
997 r <- step defState st
1000 | p x -> return True
1003 Stop -> return False
1005{-# INLINE_NORMAL maximum #-}
1006maximum :: (Monad m, Ord a) => Stream m a -> m (Maybe a)
1007maximum (Stream step state) = go Nothing state
1010 r <- step defState st
1012 Yield x s -> go (Just x) s
1013 Skip s -> go Nothing s
1014 Stop -> return Nothing
1015 go (Just acc) st = do
1016 r <- step defState st
1019 | acc <= x -> go (Just x) s
1020 | otherwise -> go (Just acc) s
1021 Skip s -> go (Just acc) s
1022 Stop -> return (Just acc)
1024{-# INLINE_NORMAL maximumBy #-}
1025maximumBy :: Monad m => (a -> a -> Ordering) -> Stream m a -> m (Maybe a)
1026maximumBy cmp (Stream step state) = go Nothing state
1029 r <- step defState st
1031 Yield x s -> go (Just x) s
1032 Skip s -> go Nothing s
1033 Stop -> return Nothing
1034 go (Just acc) st = do
1035 r <- step defState st
1037 Yield x s -> case cmp acc x of
1038 GT -> go (Just acc) s
1040 Skip s -> go (Just acc) s
1041 Stop -> return (Just acc)
1043{-# INLINE_NORMAL minimum #-}
1044minimum :: (Monad m, Ord a) => Stream m a -> m (Maybe a)
1045minimum (Stream step state) = go Nothing state
1048 r <- step defState st
1050 Yield x s -> go (Just x) s
1051 Skip s -> go Nothing s
1052 Stop -> return Nothing
1053 go (Just acc) st = do
1054 r <- step defState st
1057 | acc <= x -> go (Just acc) s
1058 | otherwise -> go (Just x) s
1059 Skip s -> go (Just acc) s
1060 Stop -> return (Just acc)
1062{-# INLINE_NORMAL minimumBy #-}
1063minimumBy :: Monad m => (a -> a -> Ordering) -> Stream m a -> m (Maybe a)
1064minimumBy cmp (Stream step state) = go Nothing state
1067 r <- step defState st
1069 Yield x s -> go (Just x) s
1070 Skip s -> go Nothing s
1071 Stop -> return Nothing
1072 go (Just acc) st = do
1073 r <- step defState st
1075 Yield x s -> case cmp acc x of
1077 _ -> go (Just acc) s
1078 Skip s -> go (Just acc) s
1079 Stop -> return (Just acc)
1081{-# INLINE_NORMAL (!!) #-}
1082(!!) :: (Monad m) => Stream m a -> Int -> m (Maybe a)
1083(Stream step state) !! i = go i state
1086 r <- step defState st
1088 Yield x s | n < 0 -> return Nothing
1089 | n == 0 -> return $ Just x
1090 | otherwise -> go (n - 1) s
1092 Stop -> return Nothing
1094{-# INLINE_NORMAL lookup #-}
1095lookup :: (Monad m, Eq a) => a -> Stream m (a, b) -> m (Maybe b)
1096lookup e m = foldrM (\(a, b) xs -> if e == a then return (Just b) else xs)
1099{-# INLINE_NORMAL findM #-}
1100findM :: Monad m => (a -> m Bool) -> Stream m a -> m (Maybe a)
1101findM p m = foldrM (\x xs -> p x >>= \r -> if r then return (Just x) else xs)
1105find :: Monad m => (a -> Bool) -> Stream m a -> m (Maybe a)
1106find p = findM (return . p)
1108{-# INLINE_NORMAL findIndices #-}
1109findIndices :: Monad m => (a -> Bool) -> Stream m a -> Stream m Int
1110findIndices p (Stream step state) = Stream step' (state, 0)
1112 {-# INLINE_LATE step' #-}
1113 step' gst (st, i) = i `seq` do
1114 r <- step (adaptState gst) st
1116 Yield x s -> if p x then Yield i (s, i+1) else Skip (s, i+1)
1117 Skip s -> Skip (s, i)
1120{-# INLINE toListRev #-}
1121toListRev :: Monad m => Stream m a -> m [a]
1122toListRev = foldl' (flip (:)) []
1124-- We can implement reverse as:
1126-- > reverse = foldlS (flip cons) nil
1128-- However, this implementation is unusable because of the horrible performance
1129-- of cons. So we just convert it to a list first and then stream from the
1132-- XXX Maybe we can use an Array instead of a list here?
1133{-# INLINE_NORMAL reverse #-}
1134reverse :: Monad m => Stream m a -> Stream m a
1135reverse m = Stream step Nothing
1137 {-# INLINE_LATE step #-}
1140 return $ Skip (Just xs)
1141 step _ (Just (x:xs)) = return $ Yield x (Just xs)
1142 step _ (Just []) = return Stop
1144-- Much faster reverse for Storables
1145{-# INLINE_NORMAL reverse' #-}
1146reverse' :: forall m a. (MonadIO m, Storable a) => Stream m a -> Stream m a
1148-- This commented implementation copies the whole stream into one single array
1149-- and then streams from that array, this is 3-4x faster than the chunked code
1150-- that follows. Though this could be problematic due to unbounded large
1151-- allocations. We need to figure out why the chunked code is slower and if we
1152-- can optimize the chunked code to work as fast as this one. It may be a
1154import Foreign.ForeignPtr (touchForeignPtr)
1155import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
1156import Foreign.Ptr (Ptr, plusPtr)
1157reverse' m = Stream step Nothing
1159 {-# INLINE_LATE step #-}
1161 arr <- A.fromStreamD m
1162 let p = aEnd arr `plusPtr` negate (sizeOf (undefined :: a))
1163 return $ Skip $ Just (aStart arr, p)
1165 step _ (Just (start, p)) | p < unsafeForeignPtrToPtr start = return Stop
1167 step _ (Just (start, p)) = do
1168 let !x = A.unsafeInlineIO $ do
1170 touchForeignPtr start
1172 next = p `plusPtr` negate (sizeOf (undefined :: a))
1173 return $ Yield x (Just (start, next))
1180 $ A.fromStreamDArraysOf A.defaultChunkSize m
1183------------------------------------------------------------------------------
1184-- Grouping/Splitting
1185------------------------------------------------------------------------------
1187{-# INLINE_NORMAL splitSuffixBy' #-}
1188splitSuffixBy' :: Monad m
1189 => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1190splitSuffixBy' predicate f (Stream step state) =
1191 Stream (stepOuter f) (Just state)
1195 {-# INLINE_LATE stepOuter #-}
1196 stepOuter (Fold fstep initial done) gst (Just st) = do
1197 res <- step (adaptState gst) st
1203 then done acc' >>= \val -> return $ Yield val (Just s)
1206 Skip s -> return $ Skip $ Just s
1212 res <- step (adaptState gst) stt
1217 then done acc' >>= \val -> return $ Yield val (Just s)
1219 Skip s -> go SPEC s acc
1220 Stop -> done acc >>= \val -> return $ Yield val Nothing
1222 stepOuter _ _ Nothing = return Stop
1224{-# INLINE_NORMAL groupsBy #-}
1230groupsBy cmp f (Stream step state) = Stream (stepOuter f) (Just state, Nothing)
1234 {-# INLINE_LATE stepOuter #-}
1235 stepOuter (Fold fstep initial done) gst (Just st, Nothing) = do
1236 res <- step (adaptState gst) st
1243 Skip s -> return $ Skip $ (Just s, Nothing)
1248 go !_ prev stt !acc = do
1249 res <- step (adaptState gst) stt
1256 else done acc >>= \r -> return $ Yield r (Just s, Just x)
1257 Skip s -> go SPEC prev s acc
1258 Stop -> done acc >>= \r -> return $ Yield r (Nothing, Nothing)
1260 stepOuter (Fold fstep initial done) gst (Just st, Just prev) = do
1262 acc' <- fstep acc prev
1267 -- XXX code duplicated from the previous equation
1269 res <- step (adaptState gst) stt
1276 else done acc >>= \r -> return $ Yield r (Just s, Just x)
1277 Skip s -> go SPEC s acc
1278 Stop -> done acc >>= \r -> return $ Yield r (Nothing, Nothing)
1280 stepOuter _ _ (Nothing,_) = return Stop
1282{-# INLINE_NORMAL groupsRollingBy #-}
1283groupsRollingBy :: Monad m
1288groupsRollingBy cmp f (Stream step state) =
1289 Stream (stepOuter f) (Just state, Nothing)
1292 {-# INLINE_LATE stepOuter #-}
1293 stepOuter (Fold fstep initial done) gst (Just st, Nothing) = do
1294 res <- step (adaptState gst) st
1301 Skip s -> return $ Skip $ (Just s, Nothing)
1305 go !_ prev stt !acc = do
1306 res <- step (adaptState gst) stt
1314 done acc >>= \r -> return $ Yield r (Just s, Just x)
1315 Skip s -> go SPEC prev s acc
1316 Stop -> done acc >>= \r -> return $ Yield r (Nothing, Nothing)
1318 stepOuter (Fold fstep initial done) gst (Just st, Just prev') = do
1320 acc' <- fstep acc prev'
1321 go SPEC prev' st acc'
1324 go !_ prevv stt !acc = do
1325 res <- step (adaptState gst) stt
1332 else done acc >>= \r -> return $ Yield r (Just s, Just x)
1333 Skip s -> go SPEC prevv s acc
1334 Stop -> done acc >>= \r -> return $ Yield r (Nothing, Nothing)
1336 stepOuter _ _ (Nothing, _) = return Stop
1338{-# INLINE_NORMAL splitBy #-}
1339splitBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1340splitBy predicate f (Stream step state) = Stream (step' f) (Just state)
1344 {-# INLINE_LATE step' #-}
1345 step' (Fold fstep initial done) gst (Just st) = initial >>= go SPEC st
1350 res <- step (adaptState gst) stt
1354 then done acc >>= \r -> return $ Yield r (Just s)
1358 Skip s -> go SPEC s acc
1359 Stop -> done acc >>= \r -> return $ Yield r Nothing
1361 step' _ _ Nothing = return Stop
1363-- XXX requires -funfolding-use-threshold=150 in lines-unlines benchmark
1364{-# INLINE_NORMAL splitSuffixBy #-}
1365splitSuffixBy :: Monad m
1366 => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1367splitSuffixBy predicate f (Stream step state) = Stream (step' f) (Just state)
1371 {-# INLINE_LATE step' #-}
1372 step' (Fold fstep initial done) gst (Just st) = do
1373 res <- step (adaptState gst) st
1378 then done acc >>= \val -> return $ Yield val (Just s)
1383 Skip s -> return $ Skip $ Just s
1389 res <- step (adaptState gst) stt
1393 then done acc >>= \r -> return $ Yield r (Just s)
1397 Skip s -> go SPEC s acc
1398 Stop -> done acc >>= \r -> return $ Yield r Nothing
1400 step' _ _ Nothing = return Stop
1402{-# INLINE_NORMAL wordsBy #-}
1403wordsBy :: Monad m => (a -> Bool) -> Fold m a b -> Stream m a -> Stream m b
1404wordsBy predicate f (Stream step state) = Stream (stepOuter f) (Just state)
1408 {-# INLINE_LATE stepOuter #-}
1409 stepOuter (Fold fstep initial done) gst (Just st) = do
1410 res <- step (adaptState gst) st
1414 then return $ Skip (Just s)
1420 Skip s -> return $ Skip $ Just s
1426 res <- step (adaptState gst) stt
1430 then done acc >>= \r -> return $ Yield r (Just s)
1434 Skip s -> go SPEC s acc
1435 Stop -> done acc >>= \r -> return $ Yield r Nothing
1437 stepOuter _ _ Nothing = return Stop
1439-- String search algorithms:
1440-- http://www-igm.univ-mlv.fr/~lecroq/string/index.html
1443-- TODO can we unify the splitting operations using a splitting configuration
1444-- like in the split package.
1446data SplitStyle = Infix | Suffix | Prefix deriving (Eq, Show)
1448data SplitOptions = SplitOptions
1449 { style :: SplitStyle
1450 , withSep :: Bool -- ^ keep the separators in output
1451 -- , compact :: Bool -- ^ treat multiple consecutive separators as one
1452 -- , trimHead :: Bool -- ^ drop blank at head
1453 -- , trimTail :: Bool -- ^ drop blank at tail
1457data SplitOnState s a =
1462 | GO_KARP_RABIN s !(RB.Ring a) !(Ptr a)
1465{-# INLINE_NORMAL splitOn #-}
1467 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1472splitOn patArr@Array{..} (Fold fstep initial done) (Stream step state) =
1473 Stream stepOuter GO_START
1477 patLen = A.length patArr
1478 maxIndex = patLen - 1
1479 elemBits = sizeOf (undefined :: a) * 8
1481 {-# INLINE_LATE stepOuter #-}
1482 stepOuter _ GO_START =
1484 then return $ Skip $ GO_EMPTY_PAT state
1487 r <- liftIO $ (A.unsafeIndexIO patArr 0)
1488 return $ Skip $ GO_SINGLE_PAT state r
1489 else if sizeOf (undefined :: a) * patLen
1490 <= sizeOf (undefined :: Word)
1491 then return $ Skip $ GO_SHORT_PAT state
1493 (rb, rhead) <- liftIO $ RB.new patLen
1494 return $ Skip $ GO_KARP_RABIN state rb rhead
1496 stepOuter gst (GO_SINGLE_PAT stt pat) = initial >>= go SPEC stt
1501 res <- step (adaptState gst) st
1507 return $ Yield r (GO_SINGLE_PAT s pat)
1508 else fstep acc x >>= go SPEC s
1509 Skip s -> go SPEC s acc
1510 Stop -> done acc >>= \r -> return $ Yield r GO_DONE
1512 stepOuter gst (GO_SHORT_PAT stt) = initial >>= go0 SPEC 0 (0 :: Word) stt
1517 mask = (1 `shiftL` (elemBits * patLen)) - 1
1519 addToWord wrd a = (wrd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1522 patWord = mask .&. A.foldl' addToWord 0 patArr
1524 go0 !_ !idx wrd st !acc = do
1525 res <- step (adaptState gst) st
1528 let wrd' = addToWord wrd x
1531 if wrd' .&. mask == patWord
1534 return $ Yield r (GO_SHORT_PAT s)
1535 else go1 SPEC wrd' s acc
1536 else go0 SPEC (idx + 1) wrd' s acc
1537 Skip s -> go0 SPEC idx wrd s acc
1540 then go2 wrd idx acc
1542 done acc' >>= \r -> return $ Yield r GO_DONE
1545 go1 !_ wrd st !acc = do
1546 res <- step (adaptState gst) st
1549 let wrd' = addToWord wrd x
1550 old = (mask .&. wrd) `shiftR` (elemBits * (patLen - 1))
1551 acc' <- fstep acc (toEnum $ fromIntegral old)
1552 if wrd' .&. mask == patWord
1553 then done acc' >>= \r -> return $ Yield r (GO_SHORT_PAT s)
1554 else go1 SPEC wrd' s acc'
1555 Skip s -> go1 SPEC wrd s acc
1557 acc' <- go2 wrd patLen acc
1558 done acc' >>= \r -> return $ Yield r GO_DONE
1560 go2 !wrd !n !acc | n > 0 = do
1561 let old = (mask .&. wrd) `shiftR` (elemBits * (n - 1))
1562 fstep acc (toEnum $ fromIntegral old) >>= go2 wrd (n - 1)
1563 go2 _ _ acc = return acc
1565 stepOuter gst (GO_KARP_RABIN stt rb rhead) = do
1566 initial >>= go0 SPEC 0 rhead stt
1570 k = 2891336453 :: Word32
1572 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1573 deltaCksum cksum old new =
1574 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1576 -- XXX shall we use a random starting hash or 1 instead of 0?
1577 patHash = A.foldl' addCksum 0 patArr
1580 go0 !_ !idx !rh st !acc = do
1581 res <- step (adaptState gst) st
1584 rh' <- liftIO $ RB.unsafeInsert rb rh x
1587 let fold = RB.unsafeFoldRing (RB.ringBound rb)
1588 let !ringHash = fold addCksum 0 rb
1589 if ringHash == patHash
1590 then go2 SPEC ringHash rh' s acc
1591 else go1 SPEC ringHash rh' s acc
1592 else go0 SPEC (idx + 1) rh' s acc
1593 Skip s -> go0 SPEC idx rh s acc
1595 !acc' <- if idx /= 0
1596 then RB.unsafeFoldRingM rh fstep acc rb
1598 done acc' >>= \r -> return $ Yield r GO_DONE
1600 -- XXX Theoretically this code can do 4 times faster if GHC generates
1601 -- optimal code. If we use just "(cksum' == patHash)" condition it goes
1602 -- 4x faster, as soon as we add the "RB.unsafeEqArray rb v" condition
1603 -- the generated code changes drastically and becomes 4x slower. Need
1604 -- to investigate what is going on with GHC.
1606 go1 !_ !cksum !rh st !acc = do
1607 res <- step (adaptState gst) st
1610 old <- liftIO $ peek rh
1611 let cksum' = deltaCksum cksum old x
1612 acc' <- fstep acc old
1614 if (cksum' == patHash)
1616 rh' <- liftIO (RB.unsafeInsert rb rh x)
1617 go2 SPEC cksum' rh' s acc'
1619 rh' <- liftIO (RB.unsafeInsert rb rh x)
1620 go1 SPEC cksum' rh' s acc'
1621 Skip s -> go1 SPEC cksum rh s acc
1623 acc' <- RB.unsafeFoldRingFullM rh fstep acc rb
1624 done acc' >>= \r -> return $ Yield r GO_DONE
1626 go2 !_ !cksum' !rh' s !acc' = do
1627 if RB.unsafeEqArray rb rh' patArr
1630 return $ Yield r (GO_KARP_RABIN s rb rhead)
1631 else go1 SPEC cksum' rh' s acc'
1633 stepOuter gst (GO_EMPTY_PAT st) = do
1634 res <- step (adaptState gst) st
1639 done acc' >>= \r -> return $ Yield r (GO_EMPTY_PAT s)
1640 Skip s -> return $ Skip (GO_EMPTY_PAT s)
1643 stepOuter _ GO_DONE = return Stop
1645{-# INLINE_NORMAL splitSuffixOn #-}
1647 :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)
1653splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)
1654 (Stream step state) =
1655 Stream stepOuter GO_START
1659 patLen = A.length patArr
1660 maxIndex = patLen - 1
1661 elemBits = sizeOf (undefined :: a) * 8
1663 {-# INLINE_LATE stepOuter #-}
1664 stepOuter _ GO_START =
1666 then return $ Skip $ GO_EMPTY_PAT state
1669 r <- liftIO $ (A.unsafeIndexIO patArr 0)
1670 return $ Skip $ GO_SINGLE_PAT state r
1671 else if sizeOf (undefined :: a) * patLen
1672 <= sizeOf (undefined :: Word)
1673 then return $ Skip $ GO_SHORT_PAT state
1675 (rb, rhead) <- liftIO $ RB.new patLen
1676 return $ Skip $ GO_KARP_RABIN state rb rhead
1678 stepOuter gst (GO_SINGLE_PAT stt pat) = do
1679 -- This first part is the only difference between splitOn and
1681 -- If the last element is a separator do not issue a blank segment.
1682 res <- step (adaptState gst) stt
1688 acc' <- if withSep then fstep acc x else return acc
1689 done acc' >>= \r -> return $ Yield r (GO_SINGLE_PAT s pat)
1690 else fstep acc x >>= go SPEC s
1691 Skip s -> return $ Skip $ (GO_SINGLE_PAT s pat)
1696 -- This is identical for splitOn and splitSuffixOn
1698 res <- step (adaptState gst) st
1703 acc' <- if withSep then fstep acc x else return acc
1705 return $ Yield r (GO_SINGLE_PAT s pat)
1706 else fstep acc x >>= go SPEC s
1707 Skip s -> go SPEC s acc
1708 Stop -> done acc >>= \r -> return $ Yield r GO_DONE
1710 stepOuter gst (GO_SHORT_PAT stt) = do
1712 -- Call "initial" only if the stream yields an element, otherwise we
1713 -- may call "initial" but never yield anything. initial may produce a
1714 -- side effect, therefore we will end up doing and discard a side
1719 res <- step (adaptState gst) stt
1723 let wrd' = addToWord wrd x
1724 acc' <- if withSep then fstep acc x else return acc
1727 if wrd' .&. mask == patWord
1728 then done acc' >>= \r -> return $ Yield r (GO_SHORT_PAT s)
1729 else go0 SPEC (idx + 1) wrd' s acc'
1730 else go0 SPEC (idx + 1) wrd' s acc'
1731 Skip s -> return $ Skip (GO_SHORT_PAT s)
1737 mask = (1 `shiftL` (elemBits * patLen)) - 1
1739 addToWord wrd a = (wrd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1742 patWord = mask .&. A.foldl' addToWord 0 patArr
1744 go0 !_ !idx wrd st !acc = do
1745 res <- step (adaptState gst) st
1748 let wrd' = addToWord wrd x
1749 acc' <- if withSep then fstep acc x else return acc
1752 if wrd' .&. mask == patWord
1755 return $ Yield r (GO_SHORT_PAT s)
1756 else go1 SPEC wrd' s acc'
1757 else go0 SPEC (idx + 1) wrd' s acc'
1758 Skip s -> go0 SPEC idx wrd s acc
1760 if (idx == maxIndex) && (wrd .&. mask == patWord)
1763 acc' <- if idx /= 0 && not withSep
1764 then go2 wrd idx acc
1766 done acc' >>= \r -> return $ Yield r GO_DONE
1769 go1 !_ wrd st !acc = do
1770 res <- step (adaptState gst) st
1773 let wrd' = addToWord wrd x
1774 old = (mask .&. wrd) `shiftR` (elemBits * (patLen - 1))
1777 else fstep acc (toEnum $ fromIntegral old)
1778 if wrd' .&. mask == patWord
1779 then done acc' >>= \r -> return $ Yield r (GO_SHORT_PAT s)
1780 else go1 SPEC wrd' s acc'
1781 Skip s -> go1 SPEC wrd s acc
1783 -- If the last sequence is a separator do not issue a blank
1785 if wrd .&. mask == patWord
1790 else go2 wrd patLen acc
1791 done acc' >>= \r -> return $ Yield r GO_DONE
1793 go2 !wrd !n !acc | n > 0 = do
1794 let old = (mask .&. wrd) `shiftR` (elemBits * (n - 1))
1795 fstep acc (toEnum $ fromIntegral old) >>= go2 wrd (n - 1)
1796 go2 _ _ acc = return acc
1798 stepOuter gst (GO_KARP_RABIN stt rb rhead) = do
1800 res <- step (adaptState gst) stt
1804 acc' <- if withSep then fstep acc x else return acc
1805 rh' <- liftIO (RB.unsafeInsert rb rhead x)
1808 let fold = RB.unsafeFoldRing (RB.ringBound rb)
1809 let !ringHash = fold addCksum 0 rb
1810 if ringHash == patHash
1811 then go2 SPEC ringHash rh' s acc'
1812 else go0 SPEC (idx + 1) rh' s acc'
1813 else go0 SPEC (idx + 1) rh' s acc'
1814 Skip s -> return $ Skip (GO_KARP_RABIN s rb rhead)
1819 k = 2891336453 :: Word32
1821 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1822 deltaCksum cksum old new =
1823 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1825 -- XXX shall we use a random starting hash or 1 instead of 0?
1826 patHash = A.foldl' addCksum 0 patArr
1829 go0 !_ !idx !rh st !acc = do
1830 res <- step (adaptState gst) st
1833 acc' <- if withSep then fstep acc x else return acc
1834 rh' <- liftIO (RB.unsafeInsert rb rh x)
1837 let fold = RB.unsafeFoldRing (RB.ringBound rb)
1838 let !ringHash = fold addCksum 0 rb
1839 if ringHash == patHash
1840 then go2 SPEC ringHash rh' s acc'
1841 else go1 SPEC ringHash rh' s acc'
1842 else go0 SPEC (idx + 1) rh' s acc'
1843 Skip s -> go0 SPEC idx rh s acc
1845 -- do not issue a blank segment when we end at pattern
1846 if (idx == maxIndex) && RB.unsafeEqArray rb rh patArr
1849 !acc' <- if idx /= 0 && not withSep
1850 then RB.unsafeFoldRingM rh fstep acc rb
1852 done acc' >>= \r -> return $ Yield r GO_DONE
1854 -- XXX Theoretically this code can do 4 times faster if GHC generates
1855 -- optimal code. If we use just "(cksum' == patHash)" condition it goes
1856 -- 4x faster, as soon as we add the "RB.unsafeEqArray rb v" condition
1857 -- the generated code changes drastically and becomes 4x slower. Need
1858 -- to investigate what is going on with GHC.
1860 go1 !_ !cksum !rh st !acc = do
1861 res <- step (adaptState gst) st
1864 old <- liftIO $ peek rh
1865 let cksum' = deltaCksum cksum old x
1870 if (cksum' == patHash)
1872 rh' <- liftIO (RB.unsafeInsert rb rh x)
1873 go2 SPEC cksum' rh' s acc'
1875 rh' <- liftIO (RB.unsafeInsert rb rh x)
1876 go1 SPEC cksum' rh' s acc'
1877 Skip s -> go1 SPEC cksum rh s acc
1879 if RB.unsafeEqArray rb rh patArr
1884 else RB.unsafeFoldRingFullM rh fstep acc rb
1885 done acc' >>= \r -> return $ Yield r GO_DONE
1887 go2 !_ !cksum' !rh' s !acc' = do
1888 if RB.unsafeEqArray rb rh' patArr
1891 return $ Yield r (GO_KARP_RABIN s rb rhead)
1892 else go1 SPEC cksum' rh' s acc'
1894 stepOuter gst (GO_EMPTY_PAT st) = do
1895 res <- step (adaptState gst) st
1900 done acc' >>= \r -> return $ Yield r (GO_EMPTY_PAT s)
1901 Skip s -> return $ Skip (GO_EMPTY_PAT s)
1904 stepOuter _ GO_DONE = return Stop
1906data SplitState s arr
1908 | SplitBuffering s arr
1909 | SplitSplitting s arr
1910 | SplitYielding arr (SplitState s arr)
1913-- XXX An alternative approach would be to use a partial fold (Fold m a b) to
1914-- split using a splitBy like combinator. The Fold would consume upto the
1915-- separator and return any leftover which can then be fed to the next fold.
1917-- We can revisit this once we have partial folds/parsers.
1919-- | Performs infix separator style splitting.
1920{-# INLINE_NORMAL splitInnerBy #-}
1923 => (f a -> m (f a, Maybe (f a))) -- splitter
1924 -> (f a -> f a -> m (f a)) -- joiner
1927splitInnerBy splitter joiner (Stream step1 state1) =
1928 (Stream step (SplitInitial state1))
1932 {-# INLINE_LATE step #-}
1933 step gst (SplitInitial st) = do
1937 (x1, mx2) <- splitter x
1938 return $ case mx2 of
1939 Nothing -> Skip (SplitBuffering s x1)
1940 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
1941 Skip s -> return $ Skip (SplitInitial s)
1942 Stop -> return $ Stop
1944 step gst (SplitBuffering st buf) = do
1948 (x1, mx2) <- splitter x
1949 buf' <- joiner buf x1
1950 return $ case mx2 of
1951 Nothing -> Skip (SplitBuffering s buf')
1952 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
1953 Skip s -> return $ Skip (SplitBuffering s buf)
1954 Stop -> return $ Skip (SplitYielding buf SplitFinishing)
1956 step _ (SplitSplitting st buf) = do
1957 (x1, mx2) <- splitter buf
1958 return $ case mx2 of
1959 Nothing -> Skip $ SplitBuffering st x1
1960 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
1962 step _ (SplitYielding x next) = return $ Yield x next
1963 step _ SplitFinishing = return $ Stop
1965-- | Performs infix separator style splitting.
1966{-# INLINE_NORMAL splitInnerBySuffix #-}
1968 :: (Monad m, Eq (f a), Monoid (f a))
1969 => (f a -> m (f a, Maybe (f a))) -- splitter
1970 -> (f a -> f a -> m (f a)) -- joiner
1973splitInnerBySuffix splitter joiner (Stream step1 state1) =
1974 (Stream step (SplitInitial state1))
1978 {-# INLINE_LATE step #-}
1979 step gst (SplitInitial st) = do
1983 (x1, mx2) <- splitter x
1984 return $ case mx2 of
1985 Nothing -> Skip (SplitBuffering s x1)
1986 Just x2 -> Skip (SplitYielding x1 (SplitSplitting s x2))
1987 Skip s -> return $ Skip (SplitInitial s)
1988 Stop -> return $ Stop
1990 step gst (SplitBuffering st buf) = do
1994 (x1, mx2) <- splitter x
1995 buf' <- joiner buf x1
1996 return $ case mx2 of
1997 Nothing -> Skip (SplitBuffering s buf')
1998 Just x2 -> Skip (SplitYielding buf' (SplitSplitting s x2))
1999 Skip s -> return $ Skip (SplitBuffering s buf)
2003 else Skip (SplitYielding buf SplitFinishing)
2005 step _ (SplitSplitting st buf) = do
2006 (x1, mx2) <- splitter buf
2007 return $ case mx2 of
2008 Nothing -> Skip $ SplitBuffering st x1
2009 Just x2 -> Skip $ SplitYielding x1 (SplitSplitting st x2)
2011 step _ (SplitYielding x next) = return $ Yield x next
2012 step _ SplitFinishing = return $ Stop
2014------------------------------------------------------------------------------
2016------------------------------------------------------------------------------
2018{-# INLINE_NORMAL isPrefixOf #-}
2019isPrefixOf :: (Eq a, Monad m) => Stream m a -> Stream m a -> m Bool
2020isPrefixOf (Stream stepa ta) (Stream stepb tb) = go (ta, tb, Nothing)
2022 go (sa, sb, Nothing) = do
2023 r <- stepa defState sa
2025 Yield x sa' -> go (sa', sb, Just x)
2026 Skip sa' -> go (sa', sb, Nothing)
2029 go (sa, sb, Just x) = do
2030 r <- stepb defState sb
2034 then go (sa, sb', Nothing)
2036 Skip sb' -> go (sa, sb', Just x)
2037 Stop -> return False
2039{-# INLINE_NORMAL isSubsequenceOf #-}
2040isSubsequenceOf :: (Eq a, Monad m) => Stream m a -> Stream m a -> m Bool
2041isSubsequenceOf (Stream stepa ta) (Stream stepb tb) = go (ta, tb, Nothing)
2043 go (sa, sb, Nothing) = do
2044 r <- stepa defState sa
2046 Yield x sa' -> go (sa', sb, Just x)
2047 Skip sa' -> go (sa', sb, Nothing)
2050 go (sa, sb, Just x) = do
2051 r <- stepb defState sb
2055 then go (sa, sb', Nothing)
2056 else go (sa, sb', Just x)
2057 Skip sb' -> go (sa, sb', Just x)
2058 Stop -> return False
2060{-# INLINE_NORMAL stripPrefix #-}
2063 => Stream m a -> Stream m a -> m (Maybe (Stream m a))
2064stripPrefix (Stream stepa ta) (Stream stepb tb) = go (ta, tb, Nothing)
2066 go (sa, sb, Nothing) = do
2067 r <- stepa defState sa
2069 Yield x sa' -> go (sa', sb, Just x)
2070 Skip sa' -> go (sa', sb, Nothing)
2071 Stop -> return $ Just (Stream stepb sb)
2073 go (sa, sb, Just x) = do
2074 r <- stepb defState sb
2078 then go (sa, sb', Nothing)
2080 Skip sb' -> go (sa, sb', Just x)
2081 Stop -> return Nothing
2083------------------------------------------------------------------------------
2085------------------------------------------------------------------------------
2087-- | Execute a monadic action for each element of the 'Stream'
2088{-# INLINE_NORMAL mapM_ #-}
2089mapM_ :: Monad m => (a -> m b) -> Stream m a -> m ()
2090mapM_ m = drain . mapM m
2092-------------------------------------------------------------------------------
2093-- Stream transformations using Unfolds
2094-------------------------------------------------------------------------------
2096-- Define a unique structure to use in inspection testing
2097data ConcatMapUState o i =
2099 | ConcatMapUInner o i
2101-- | @concatMapU unfold stream@ uses @unfold@ to map the input stream elements
2102-- to streams and then flattens the generated streams into a single output
2105-- This is like 'concatMap' but uses an unfold with an explicit state to
2106-- generate the stream instead of a 'Stream' type generator. This allows better
2107-- optimization via fusion. This can be many times more efficient than
2110{-# INLINE_NORMAL concatMapU #-}
2111concatMapU :: Monad m => Unfold m a b -> Stream m a -> Stream m b
2112concatMapU (Unfold istep inject) (Stream ostep ost) =
2113 Stream step (ConcatMapUOuter ost)
2115 {-# INLINE_LATE step #-}
2116 step gst (ConcatMapUOuter o) = do
2117 r <- ostep (adaptState gst) o
2121 i `seq` return (Skip (ConcatMapUInner o' i))
2122 Skip o' -> return $ Skip (ConcatMapUOuter o')
2123 Stop -> return $ Stop
2125 step _ (ConcatMapUInner o i) = do
2128 Yield x i' -> Yield x (ConcatMapUInner o i')
2129 Skip i' -> Skip (ConcatMapUInner o i')
2130 Stop -> Skip (ConcatMapUOuter o)
2132data ConcatUnfoldInterleaveState o i =
2133 ConcatUnfoldInterleaveOuter o [i]
2134 | ConcatUnfoldInterleaveInner o [i]
2135 | ConcatUnfoldInterleaveInnerL [i] [i]
2136 | ConcatUnfoldInterleaveInnerR [i] [i]
2138-- XXX use arrays to store state instead of lists.
2139-- XXX In general we can use different scheduling strategies e.g. how to
2140-- schedule the outer vs inner loop or assigning weights to different streams
2141-- or outer and inner loops.
2143-- After a yield, switch to the next stream. Do not switch streams on Skip.
2144-- Yield from outer stream switches to the inner stream.
2146-- There are two choices here, (1) exhaust the outer stream first and then
2147-- start yielding from the inner streams, this is much simpler to implement,
2148-- (2) yield at least one element from an inner stream before going back to
2149-- outer stream and opening the next stream from it.
2151-- Ideally, we need some scheduling bias to inner streams vs outer stream.
2152-- Maybe we can configure the behavior.
2154{-# INLINE_NORMAL concatUnfoldInterleave #-}
2155concatUnfoldInterleave :: Monad m => Unfold m a b -> Stream m a -> Stream m b
2156concatUnfoldInterleave (Unfold istep inject) (Stream ostep ost) =
2157 Stream step (ConcatUnfoldInterleaveOuter ost [])
2159 {-# INLINE_LATE step #-}
2160 step gst (ConcatUnfoldInterleaveOuter o ls) = do
2161 r <- ostep (adaptState gst) o
2165 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
2166 Skip o' -> return $ Skip (ConcatUnfoldInterleaveOuter o' ls)
2167 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
2169 step _ (ConcatUnfoldInterleaveInner _ []) = undefined
2170 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
2173 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
2174 Skip s -> Skip (ConcatUnfoldInterleaveInner o (s:ls))
2175 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
2177 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
2178 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
2179 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
2181 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
2184 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
2185 Skip s -> Skip (ConcatUnfoldInterleaveInnerL (s:ls) rs)
2186 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
2188 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
2189 step _ (ConcatUnfoldInterleaveInnerR ls []) =
2190 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
2192 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
2195 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
2196 Skip s -> Skip (ConcatUnfoldInterleaveInnerR ls (s:rs))
2197 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
2199-- XXX In general we can use different scheduling strategies e.g. how to
2200-- schedule the outer vs inner loop or assigning weights to different streams
2201-- or outer and inner loops.
2203-- This could be inefficient if the tasks are too small.
2205-- Compared to concatUnfoldInterleave this one switches streams on Skips.
2207{-# INLINE_NORMAL concatUnfoldRoundrobin #-}
2208concatUnfoldRoundrobin :: Monad m => Unfold m a b -> Stream m a -> Stream m b
2209concatUnfoldRoundrobin (Unfold istep inject) (Stream ostep ost) =
2210 Stream step (ConcatUnfoldInterleaveOuter ost [])
2212 {-# INLINE_LATE step #-}
2213 step gst (ConcatUnfoldInterleaveOuter o ls) = do
2214 r <- ostep (adaptState gst) o
2218 i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))
2219 Skip o' -> return $ Skip (ConcatUnfoldInterleaveInner o' ls)
2220 Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
2222 step _ (ConcatUnfoldInterleaveInner o []) =
2223 return $ Skip (ConcatUnfoldInterleaveOuter o [])
2225 step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do
2228 Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))
2229 Skip s -> Skip (ConcatUnfoldInterleaveOuter o (s:ls))
2230 Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)
2232 step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop
2233 step _ (ConcatUnfoldInterleaveInnerL [] rs) =
2234 return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)
2236 step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do
2239 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))
2240 Skip s -> Skip (ConcatUnfoldInterleaveInnerL ls (s:rs))
2241 Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)
2243 step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop
2244 step _ (ConcatUnfoldInterleaveInnerR ls []) =
2245 return $ Skip (ConcatUnfoldInterleaveInnerL ls [])
2247 step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do
2250 Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)
2251 Skip s -> Skip (ConcatUnfoldInterleaveInnerR (s:ls) rs)
2252 Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)
2254data AppendState s1 s2 = AppendFirst s1 | AppendSecond s2
2256-- Note that this could be much faster compared to the CPS stream. However, as
2257-- the number of streams being composed increases this may become expensive.
2258-- Need to see where the breaking point is between the two.
2260{-# INLINE_NORMAL append #-}
2261append :: Monad m => Stream m a -> Stream m a -> Stream m a
2262append (Stream step1 state1) (Stream step2 state2) =
2263 Stream step (AppendFirst state1)
2267 {-# INLINE_LATE step #-}
2268 step gst (AppendFirst st) = do
2271 Yield a s -> Yield a (AppendFirst s)
2272 Skip s -> Skip (AppendFirst s)
2273 Stop -> Skip (AppendSecond state2)
2275 step gst (AppendSecond st) = do
2278 Yield a s -> Yield a (AppendSecond s)
2279 Skip s -> Skip (AppendSecond s)
2282data InterleaveState s1 s2 = InterleaveFirst s1 s2 | InterleaveSecond s1 s2
2283 | InterleaveSecondOnly s2 | InterleaveFirstOnly s1
2285{-# INLINE_NORMAL interleave #-}
2286interleave :: Monad m => Stream m a -> Stream m a -> Stream m a
2287interleave (Stream step1 state1) (Stream step2 state2) =
2288 Stream step (InterleaveFirst state1 state2)
2292 {-# INLINE_LATE step #-}
2293 step gst (InterleaveFirst st1 st2) = do
2296 Yield a s -> Yield a (InterleaveSecond s st2)
2297 Skip s -> Skip (InterleaveFirst s st2)
2298 Stop -> Skip (InterleaveSecondOnly st2)
2300 step gst (InterleaveSecond st1 st2) = do
2303 Yield a s -> Yield a (InterleaveFirst st1 s)
2304 Skip s -> Skip (InterleaveSecond st1 s)
2305 Stop -> Skip (InterleaveFirstOnly st1)
2307 step gst (InterleaveFirstOnly st1) = do
2310 Yield a s -> Yield a (InterleaveFirstOnly s)
2311 Skip s -> Skip (InterleaveFirstOnly s)
2314 step gst (InterleaveSecondOnly st2) = do
2317 Yield a s -> Yield a (InterleaveSecondOnly s)
2318 Skip s -> Skip (InterleaveSecondOnly s)
2321{-# INLINE_NORMAL interleaveMin #-}
2322interleaveMin :: Monad m => Stream m a -> Stream m a -> Stream m a
2323interleaveMin (Stream step1 state1) (Stream step2 state2) =
2324 Stream step (InterleaveFirst state1 state2)
2328 {-# INLINE_LATE step #-}
2329 step gst (InterleaveFirst st1 st2) = do
2332 Yield a s -> Yield a (InterleaveSecond s st2)
2333 Skip s -> Skip (InterleaveFirst s st2)
2336 step gst (InterleaveSecond st1 st2) = do
2339 Yield a s -> Yield a (InterleaveFirst st1 s)
2340 Skip s -> Skip (InterleaveSecond st1 s)
2343 step _ (InterleaveFirstOnly _) = undefined
2344 step _ (InterleaveSecondOnly _) = undefined
2346{-# INLINE_NORMAL interleaveSuffix #-}
2347interleaveSuffix :: Monad m => Stream m a -> Stream m a -> Stream m a
2348interleaveSuffix (Stream step1 state1) (Stream step2 state2) =
2349 Stream step (InterleaveFirst state1 state2)
2353 {-# INLINE_LATE step #-}
2354 step gst (InterleaveFirst st1 st2) = do
2357 Yield a s -> Yield a (InterleaveSecond s st2)
2358 Skip s -> Skip (InterleaveFirst s st2)
2361 step gst (InterleaveSecond st1 st2) = do
2364 Yield a s -> Yield a (InterleaveFirst st1 s)
2365 Skip s -> Skip (InterleaveSecond st1 s)
2366 Stop -> Skip (InterleaveFirstOnly st1)
2368 step gst (InterleaveFirstOnly st1) = do
2371 Yield a s -> Yield a (InterleaveFirstOnly s)
2372 Skip s -> Skip (InterleaveFirstOnly s)
2375 step _ (InterleaveSecondOnly _) = undefined
2377data InterleaveInfixState s1 s2 a
2378 = InterleaveInfixFirst s1 s2
2379 | InterleaveInfixSecondBuf s1 s2
2380 | InterleaveInfixSecondYield s1 s2 a
2381 | InterleaveInfixFirstYield s1 s2 a
2382 | InterleaveInfixFirstOnly s1
2384{-# INLINE_NORMAL interleaveInfix #-}
2385interleaveInfix :: Monad m => Stream m a -> Stream m a -> Stream m a
2386interleaveInfix (Stream step1 state1) (Stream step2 state2) =
2387 Stream step (InterleaveInfixFirst state1 state2)
2391 {-# INLINE_LATE step #-}
2392 step gst (InterleaveInfixFirst st1 st2) = do
2395 Yield a s -> Yield a (InterleaveInfixSecondBuf s st2)
2396 Skip s -> Skip (InterleaveInfixFirst s st2)
2399 step gst (InterleaveInfixSecondBuf st1 st2) = do
2402 Yield a s -> Skip (InterleaveInfixSecondYield st1 s a)
2403 Skip s -> Skip (InterleaveInfixSecondBuf st1 s)
2404 Stop -> Skip (InterleaveInfixFirstOnly st1)
2406 step gst (InterleaveInfixSecondYield st1 st2 x) = do
2409 Yield a s -> Yield x (InterleaveInfixFirstYield s st2 a)
2410 Skip s -> Skip (InterleaveInfixSecondYield s st2 x)
2413 step _ (InterleaveInfixFirstYield st1 st2 x) = do
2414 return $ Yield x (InterleaveInfixSecondBuf st1 st2)
2416 step gst (InterleaveInfixFirstOnly st1) = do
2419 Yield a s -> Yield a (InterleaveInfixFirstOnly s)
2420 Skip s -> Skip (InterleaveInfixFirstOnly s)
2423{-# INLINE_NORMAL roundRobin #-}
2424roundRobin :: Monad m => Stream m a -> Stream m a -> Stream m a
2425roundRobin (Stream step1 state1) (Stream step2 state2) =
2426 Stream step (InterleaveFirst state1 state2)
2430 {-# INLINE_LATE step #-}
2431 step gst (InterleaveFirst st1 st2) = do
2434 Yield a s -> Yield a (InterleaveSecond s st2)
2435 Skip s -> Skip (InterleaveSecond s st2)
2436 Stop -> Skip (InterleaveSecondOnly st2)
2438 step gst (InterleaveSecond st1 st2) = do
2441 Yield a s -> Yield a (InterleaveFirst st1 s)
2442 Skip s -> Skip (InterleaveFirst st1 s)
2443 Stop -> Skip (InterleaveFirstOnly st1)
2445 step gst (InterleaveSecondOnly st2) = do
2448 Yield a s -> Yield a (InterleaveSecondOnly s)
2449 Skip s -> Skip (InterleaveSecondOnly s)
2452 step gst (InterleaveFirstOnly st1) = do
2455 Yield a s -> Yield a (InterleaveFirstOnly s)
2456 Skip s -> Skip (InterleaveFirstOnly s)
2459data ICUState s1 s2 i1 i2 =
2464 | ICUFirstInner s1 s2 i1
2465 | ICUSecondInner s1 s2 i2
2466 | ICUFirstOnlyInner s1 i1
2467 | ICUSecondOnlyInner s2 i2
2469-- | Interleave streams (full streams, not the elements) unfolded from two
2470-- input streams and concat. Stop when the first stream stops. If the second
2471-- stream ends before the first one then first stream still keeps running alone
2472-- without any interleaving with the second stream.
2474-- [a1, a2, ... an] [b1, b2 ...]
2475-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
2476-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
2477-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
2479{-# INLINE_NORMAL gintercalateSuffix #-}
2482 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
2484 (Unfold istep1 inject1) (Stream step1 state1)
2485 (Unfold istep2 inject2) (Stream step2 state2) =
2486 Stream step (ICUFirst state1 state2)
2490 {-# INLINE_LATE step #-}
2491 step gst (ICUFirst s1 s2) = do
2492 r <- step1 (adaptState gst) s1
2496 i `seq` return (Skip (ICUFirstInner s s2 i))
2497 Skip s -> return $ Skip (ICUFirst s s2)
2500 step gst (ICUFirstOnly s1) = do
2501 r <- step1 (adaptState gst) s1
2505 i `seq` return (Skip (ICUFirstOnlyInner s i))
2506 Skip s -> return $ Skip (ICUFirstOnly s)
2509 step _ (ICUFirstInner s1 s2 i1) = do
2512 Yield x i' -> Yield x (ICUFirstInner s1 s2 i')
2513 Skip i' -> Skip (ICUFirstInner s1 s2 i')
2514 Stop -> Skip (ICUSecond s1 s2)
2516 step _ (ICUFirstOnlyInner s1 i1) = do
2519 Yield x i' -> Yield x (ICUFirstOnlyInner s1 i')
2520 Skip i' -> Skip (ICUFirstOnlyInner s1 i')
2521 Stop -> Skip (ICUFirstOnly s1)
2523 step gst (ICUSecond s1 s2) = do
2524 r <- step2 (adaptState gst) s2
2528 i `seq` return (Skip (ICUSecondInner s1 s i))
2529 Skip s -> return $ Skip (ICUSecond s1 s)
2530 Stop -> return $ Skip (ICUFirstOnly s1)
2532 step _ (ICUSecondInner s1 s2 i2) = do
2535 Yield x i' -> Yield x (ICUSecondInner s1 s2 i')
2536 Skip i' -> Skip (ICUSecondInner s1 s2 i')
2537 Stop -> Skip (ICUFirst s1 s2)
2539 step _ (ICUSecondOnly _s2) = undefined
2540 step _ (ICUSecondOnlyInner _s2 _i2) = undefined
2542data InterposeSuffixState s1 i1 =
2543 InterposeSuffixFirst s1
2544 -- | InterposeSuffixFirstYield s1 i1
2545 | InterposeSuffixFirstInner s1 i1
2546 | InterposeSuffixSecond s1
2548-- Note that if an unfolded layer turns out to be nil we still emit the
2549-- separator effect. An alternate behavior could be to emit the separator
2550-- effect only if at least one element has been yielded by the unfolding.
2551-- However, that becomes a bit complicated, so we have chosen the former
2553{-# INLINE_NORMAL interposeSuffix #-}
2556 => m c -> Unfold m b c -> Stream m b -> Stream m c
2559 (Unfold istep1 inject1) (Stream step1 state1) =
2560 Stream step (InterposeSuffixFirst state1)
2564 {-# INLINE_LATE step #-}
2565 step gst (InterposeSuffixFirst s1) = do
2566 r <- step1 (adaptState gst) s1
2570 i `seq` return (Skip (InterposeSuffixFirstInner s i))
2571 -- i `seq` return (Skip (InterposeSuffixFirstYield s i))
2572 Skip s -> return $ Skip (InterposeSuffixFirst s)
2576 step _ (InterposeSuffixFirstYield s1 i1) = do
2579 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
2580 Skip i' -> Skip (InterposeSuffixFirstYield s1 i')
2581 Stop -> Skip (InterposeSuffixFirst s1)
2584 step _ (InterposeSuffixFirstInner s1 i1) = do
2587 Yield x i' -> Yield x (InterposeSuffixFirstInner s1 i')
2588 Skip i' -> Skip (InterposeSuffixFirstInner s1 i')
2589 Stop -> Skip (InterposeSuffixSecond s1)
2591 step _ (InterposeSuffixSecond s1) = do
2593 return $ Yield r (InterposeSuffixFirst s1)
2595data ICALState s1 s2 i1 i2 a =
2597 -- | ICALFirstYield s1 s2 i1
2598 | ICALFirstInner s1 s2 i1
2600 | ICALFirstOnlyInner s1 i1
2601 | ICALSecondInject s1 s2
2602 | ICALFirstInject s1 s2 i2
2603 -- | ICALFirstBuf s1 s2 i1 i2
2604 | ICALSecondInner s1 s2 i1 i2
2605 -- -- | ICALSecondInner s1 s2 i1 i2 a
2606 -- -- | ICALFirstResume s1 s2 i1 i2 a
2608-- | Interleave streams (full streams, not the elements) unfolded from two
2609-- input streams and concat. Stop when the first stream stops. If the second
2610-- stream ends before the first one then first stream still keeps running alone
2611-- without any interleaving with the second stream.
2613-- [a1, a2, ... an] [b1, b2 ...]
2614-- => [streamA1, streamA2, ... streamAn] [streamB1, streamB2, ...]
2615-- => [streamA1, streamB1, streamA2...StreamAn, streamBn]
2616-- => [a11, a12, ...a1j, b11, b12, ...b1k, a21, a22, ...]
2618{-# INLINE_NORMAL gintercalate #-}
2621 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
2623 (Unfold istep1 inject1) (Stream step1 state1)
2624 (Unfold istep2 inject2) (Stream step2 state2) =
2625 Stream step (ICALFirst state1 state2)
2629 {-# INLINE_LATE step #-}
2630 step gst (ICALFirst s1 s2) = do
2631 r <- step1 (adaptState gst) s1
2635 i `seq` return (Skip (ICALFirstInner s s2 i))
2636 -- i `seq` return (Skip (ICALFirstYield s s2 i))
2637 Skip s -> return $ Skip (ICALFirst s s2)
2641 step _ (ICALFirstYield s1 s2 i1) = do
2644 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
2645 Skip i' -> Skip (ICALFirstYield s1 s2 i')
2646 Stop -> Skip (ICALFirst s1 s2)
2649 step _ (ICALFirstInner s1 s2 i1) = do
2652 Yield x i' -> Yield x (ICALFirstInner s1 s2 i')
2653 Skip i' -> Skip (ICALFirstInner s1 s2 i')
2654 Stop -> Skip (ICALSecondInject s1 s2)
2656 step gst (ICALFirstOnly s1) = do
2657 r <- step1 (adaptState gst) s1
2661 i `seq` return (Skip (ICALFirstOnlyInner s i))
2662 Skip s -> return $ Skip (ICALFirstOnly s)
2665 step _ (ICALFirstOnlyInner s1 i1) = do
2668 Yield x i' -> Yield x (ICALFirstOnlyInner s1 i')
2669 Skip i' -> Skip (ICALFirstOnlyInner s1 i')
2670 Stop -> Skip (ICALFirstOnly s1)
2672 -- We inject the second stream even before checking if the first stream
2673 -- would yield any more elements. There is no clear choice whether we
2674 -- should do this before or after that. Doing it after may make the state
2675 -- machine a bit simpler though.
2676 step gst (ICALSecondInject s1 s2) = do
2677 r <- step2 (adaptState gst) s2
2681 i `seq` return (Skip (ICALFirstInject s1 s i))
2682 Skip s -> return $ Skip (ICALSecondInject s1 s)
2683 Stop -> return $ Skip (ICALFirstOnly s1)
2685 step gst (ICALFirstInject s1 s2 i2) = do
2686 r <- step1 (adaptState gst) s1
2690 i `seq` return (Skip (ICALSecondInner s s2 i i2))
2691 -- i `seq` return (Skip (ICALFirstBuf s s2 i i2))
2692 Skip s -> return $ Skip (ICALFirstInject s s2 i2)
2696 step _ (ICALFirstBuf s1 s2 i1 i2) = do
2699 Yield x i' -> Skip (ICALSecondInner s1 s2 i' i2 x)
2700 Skip i' -> Skip (ICALFirstBuf s1 s2 i' i2)
2703 step _ (ICALSecondInner s1 s2 i1 i2 v) = do
2706 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i' v)
2707 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i' v)
2708 Stop -> Skip (ICALFirstResume s1 s2 i1 i2 v)
2711 step _ (ICALSecondInner s1 s2 i1 i2) = do
2714 Yield x i' -> Yield x (ICALSecondInner s1 s2 i1 i')
2715 Skip i' -> Skip (ICALSecondInner s1 s2 i1 i')
2716 Stop -> Skip (ICALFirstInner s1 s2 i1)
2717 -- Stop -> Skip (ICALFirstResume s1 s2 i1 i2)
2720 step _ (ICALFirstResume s1 s2 i1 i2 x) = do
2721 return $ Yield x (ICALFirstInner s1 s2 i1 i2)
2724data InterposeState s1 i1 a =
2726 -- | InterposeFirstYield s1 i1
2727 | InterposeFirstInner s1 i1
2728 | InterposeFirstInject s1
2729 -- | InterposeFirstBuf s1 i1
2730 | InterposeSecondYield s1 i1
2731 -- -- | InterposeSecondYield s1 i1 a
2732 -- -- | InterposeFirstResume s1 i1 a
2734-- Note that this only interposes the pure values, we may run many effects to
2735-- generate those values as some effects may not generate anything (Skip).
2736{-# INLINE_NORMAL interpose #-}
2737interpose :: Monad m => m c -> Unfold m b c -> Stream m b -> Stream m c
2740 (Unfold istep1 inject1) (Stream step1 state1) =
2741 Stream step (InterposeFirst state1)
2745 {-# INLINE_LATE step #-}
2746 step gst (InterposeFirst s1) = do
2747 r <- step1 (adaptState gst) s1
2751 i `seq` return (Skip (InterposeFirstInner s i))
2752 -- i `seq` return (Skip (InterposeFirstYield s i))
2753 Skip s -> return $ Skip (InterposeFirst s)
2757 step _ (InterposeFirstYield s1 i1) = do
2760 Yield x i' -> Yield x (InterposeFirstInner s1 i')
2761 Skip i' -> Skip (InterposeFirstYield s1 i')
2762 Stop -> Skip (InterposeFirst s1)
2765 step _ (InterposeFirstInner s1 i1) = do
2768 Yield x i' -> Yield x (InterposeFirstInner s1 i')
2769 Skip i' -> Skip (InterposeFirstInner s1 i')
2770 Stop -> Skip (InterposeFirstInject s1)
2772 step gst (InterposeFirstInject s1) = do
2773 r <- step1 (adaptState gst) s1
2777 -- i `seq` return (Skip (InterposeFirstBuf s i))
2778 i `seq` return (Skip (InterposeSecondYield s i))
2779 Skip s -> return $ Skip (InterposeFirstInject s)
2783 step _ (InterposeFirstBuf s1 i1) = do
2786 Yield x i' -> Skip (InterposeSecondYield s1 i' x)
2787 Skip i' -> Skip (InterposeFirstBuf s1 i')
2792 step _ (InterposeSecondYield s1 i1 v) = do
2794 return $ Yield r (InterposeFirstResume s1 i1 v)
2796 step _ (InterposeSecondYield s1 i1) = do
2798 return $ Yield r (InterposeFirstInner s1 i1)
2801 step _ (InterposeFirstResume s1 i1 v) = do
2802 return $ Yield v (InterposeFirstInner s1 i1)
2805------------------------------------------------------------------------------
2807------------------------------------------------------------------------------
2809data GbracketState s1 s2 v
2811 | GBracketNormal s1 v
2812 | GBracketException s2
2814-- | The most general bracketing and exception combinator. All other
2815-- combinators can be expressed in terms of this combinator. This can also be
2816-- used for cases which are not covered by the standard combinators.
2820{-# INLINE_NORMAL gbracket #-}
2824 -> (forall s. m s -> m (Either e s)) -- ^ try (exception handling)
2825 -> (c -> m d) -- ^ after, on normal stop
2826 -> (c -> e -> Stream m b) -- ^ on exception
2827 -> (c -> Stream m b) -- ^ stream generator
2829gbracket bef exc aft fexc fnormal =
2830 Stream step GBracketInit
2834 {-# INLINE_LATE step #-}
2835 step _ GBracketInit = do
2837 return $ Skip $ GBracketNormal (fnormal r) r
2839 step gst (GBracketNormal (UnStream step1 st) v) = do
2840 res <- exc $ step1 gst st
2842 Right r -> case r of
2844 return $ Yield x (GBracketNormal (Stream step1 s) v)
2845 Skip s -> return $ Skip (GBracketNormal (Stream step1 s) v)
2846 Stop -> aft v >> return Stop
2847 Left e -> return $ Skip (GBracketException (fexc v e))
2848 step gst (GBracketException (UnStream step1 st)) = do
2851 Yield x s -> return $ Yield x (GBracketException (Stream step1 s))
2852 Skip s -> return $ Skip (GBracketException (Stream step1 s))
2855-- | Run a side effect before the stream yields its first element.
2856{-# INLINE_NORMAL before #-}
2857before :: Monad m => m b -> Stream m a -> Stream m a
2858before action (Stream step state) = Stream step' Nothing
2862 {-# INLINE_LATE step' #-}
2863 step' _ Nothing = action >> return (Skip (Just state))
2865 step' gst (Just st) = do
2868 Yield x s -> return $ Yield x (Just s)
2869 Skip s -> return $ Skip (Just s)
2872-- | Run a side effect whenever the stream stops normally.
2873{-# INLINE_NORMAL after #-}
2874after :: Monad m => m b -> Stream m a -> Stream m a
2875after action (Stream step state) = Stream step' state
2879 {-# INLINE_LATE step' #-}
2883 Yield x s -> return $ Yield x s
2884 Skip s -> return $ Skip s
2885 Stop -> action >> return Stop
2887-- XXX These combinators are expensive due to the call to
2888-- onException/handle/try on each step. Therefore, when possible, they should
2889-- be called in an outer loop where we perform less iterations. For example, we
2890-- cannot call them on each iteration in a char stream, instead we can call
2891-- them when doing an IO on an array.
2893-- XXX For high performance error checks in busy streams we may need another
2894-- Error constructor in step.
2896-- | Run a side effect whenever the stream aborts due to an exception. The
2897-- exception is not caught, simply rethrown.
2898{-# INLINE_NORMAL onException #-}
2899onException :: MonadCatch m => m b -> Stream m a -> Stream m a
2900onException action str =
2901 gbracket (return ()) MC.try return
2902 (\_ (e :: MC.SomeException) -> nilM (action >> MC.throwM e))
2905{-# INLINE_NORMAL _onException #-}
2906_onException :: MonadCatch m => m b -> Stream m a -> Stream m a
2907_onException action (Stream step state) = Stream step' state
2911 {-# INLINE_LATE step' #-}
2913 res <- step gst st `MC.onException` action
2915 Yield x s -> return $ Yield x s
2916 Skip s -> return $ Skip s
2919-- XXX bracket is like concatMap, it generates a stream and then flattens it.
2920-- Like concatMap it has 10x worse performance compared to linear fused
2923-- | Run the first action before the stream starts and remember its output,
2924-- generate a stream using the output, run the second action providing the
2925-- remembered value as an argument whenever the stream ends normally or due to
2927{-# INLINE_NORMAL bracket #-}
2928bracket :: MonadCatch m => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
2929bracket bef aft bet =
2930 gbracket bef MC.try aft
2931 (\a (e :: SomeException) -> nilM (aft a >> MC.throwM e)) bet
2933data BracketState s v = BracketInit | BracketRun s v
2935{-# INLINE_NORMAL _bracket #-}
2936_bracket :: MonadCatch m => m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
2937_bracket bef aft bet = Stream step' BracketInit
2941 {-# INLINE_LATE step' #-}
2942 step' _ BracketInit = bef >>= \x -> return (Skip (BracketRun (bet x) x))
2944 -- NOTE: It is important to use UnStream instead of the Stream pattern
2945 -- here, otherwise we get huge perf degradation, see note in concatMap.
2946 step' gst (BracketRun (UnStream step state) v) = do
2947 -- res <- step gst state `MC.onException` aft v
2948 res <- MC.try $ step gst state
2950 Left (e :: SomeException) -> aft v >> MC.throwM e >> return Stop
2951 Right r -> case r of
2952 Yield x s -> return $ Yield x (BracketRun (Stream step s) v)
2953 Skip s -> return $ Skip (BracketRun (Stream step s) v)
2954 Stop -> aft v >> return Stop
2956-- | Run a side effect whenever the stream stops normally or aborts due to an
2958{-# INLINE finally #-}
2959finally :: MonadCatch m => m b -> Stream m a -> Stream m a
2960-- finally action xs = after action $ onException action xs
2961finally action xs = bracket (return ()) (\_ -> action) (const xs)
2963-- | When evaluating a stream if an exception occurs, stream evaluation aborts
2964-- and the specified exception handler is run with the exception as argument.
2965{-# INLINE_NORMAL handle #-}
2966handle :: (MonadCatch m, Exception e)
2967 => (e -> Stream m a) -> Stream m a -> Stream m a
2969 gbracket (return ()) MC.try return (\_ e -> f e) (\_ -> str)
2971{-# INLINE_NORMAL _handle #-}
2972_handle :: (MonadCatch m, Exception e)
2973 => (e -> Stream m a) -> Stream m a -> Stream m a
2974_handle f (Stream step state) = Stream step' (Left state)
2978 {-# INLINE_LATE step' #-}
2979 step' gst (Left st) = do
2980 res <- MC.try $ step gst st
2982 Left e -> return $ Skip $ Right (f e)
2983 Right r -> case r of
2984 Yield x s -> return $ Yield x (Left s)
2985 Skip s -> return $ Skip (Left s)
2988 step' gst (Right (UnStream step1 st)) = do
2991 Yield x s -> return $ Yield x (Right (Stream step1 s))
2992 Skip s -> return $ Skip (Right (Stream step1 s))
2995-------------------------------------------------------------------------------
2996-- General transformation
2997-------------------------------------------------------------------------------
2999{-# INLINE_NORMAL transform #-}
3000transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
3001transform (Pipe pstep1 pstep2 pstate) (Stream step state) =
3002 Stream step' (Consume pstate, state)
3006 {-# INLINE_LATE step' #-}
3008 step' gst (Consume pst, st) = pst `seq` do
3009 r <- step (adaptState gst) st
3014 Pipe.Yield b pst' -> return $ Yield b (pst', s)
3015 Pipe.Continue pst' -> return $ Skip (pst', s)
3016 Skip s -> return $ Skip (Consume pst, s)
3019 step' _ (Produce pst, st) = pst `seq` do
3022 Pipe.Yield b pst' -> return $ Yield b (pst', st)
3023 Pipe.Continue pst' -> return $ Skip (pst', st)
3025------------------------------------------------------------------------------
3026-- Transformation by Folding (Scans)
3027------------------------------------------------------------------------------
3029------------------------------------------------------------------------------
3031------------------------------------------------------------------------------
3033-- XXX Is a prescan useful, discarding the last step does not sound useful? I
3034-- am not sure about the utility of this function, so this is implemented but
3035-- not exposed. We can expose it if someone provides good reasons why this is
3038-- XXX We have to execute the stream one step ahead to know that we are at the
3039-- last step. The vector implementation of prescan executes the last fold step
3040-- but does not yield the result. This means we have executed the effect but
3041-- discarded value. This does not sound right. In this implementation we are
3042-- not executing the last fold step.
3043{-# INLINE_NORMAL prescanlM' #-}
3044prescanlM' :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> Stream m b
3045prescanlM' f mz (Stream step state) = Stream step' (state, mz)
3047 {-# INLINE_LATE step' #-}
3048 step' gst (st, prev) = do
3049 r <- step (adaptState gst) st
3053 return $ Yield acc (s, f acc x)
3054 Skip s -> return $ Skip (s, prev)
3057{-# INLINE prescanl' #-}
3058prescanl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
3059prescanl' f z = prescanlM' (\a b -> return (f a b)) (return z)
3061------------------------------------------------------------------------------
3062-- Monolithic postscans (postscan followed by a map)
3063------------------------------------------------------------------------------
3065-- The performance of a modular postscan followed by a map seems to be
3066-- equivalent to this monolithic scan followed by map therefore we may not need
3067-- this implementation. We just have it for performance comparison and in case
3068-- modular version does not perform well in some situation.
3070{-# INLINE_NORMAL postscanlMx' #-}
3071postscanlMx' :: Monad m
3072 => (x -> a -> m x) -> m x -> (x -> m b) -> Stream m a -> Stream m b
3073postscanlMx' fstep begin done (Stream step state) = do
3074 Stream step' (state, begin)
3076 {-# INLINE_LATE step' #-}
3077 step' gst (st, acc) = do
3078 r <- step (adaptState gst) st
3084 v `seq` y `seq` return (Yield v (s, return y))
3085 Skip s -> return $ Skip (s, acc)
3088{-# INLINE_NORMAL postscanlx' #-}
3089postscanlx' :: Monad m
3090 => (x -> a -> x) -> x -> (x -> b) -> Stream m a -> Stream m b
3091postscanlx' fstep begin done s =
3092 postscanlMx' (\b a -> return (fstep b a)) (return begin) (return . done) s
3094-- XXX do we need consM strict to evaluate the begin value?
3095{-# INLINE scanlMx' #-}
3097 => (x -> a -> m x) -> m x -> (x -> m b) -> Stream m a -> Stream m b
3098scanlMx' fstep begin done s =
3099 (begin >>= \x -> x `seq` done x) `consM` postscanlMx' fstep begin done s
3101{-# INLINE scanlx' #-}
3103 => (x -> a -> x) -> x -> (x -> b) -> Stream m a -> Stream m b
3104scanlx' fstep begin done s =
3105 scanlMx' (\b a -> return (fstep b a)) (return begin) (return . done) s
3107------------------------------------------------------------------------------
3109------------------------------------------------------------------------------
3111{-# INLINE_NORMAL postscanlM' #-}
3112postscanlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
3113postscanlM' fstep begin (Stream step state) =
3114 begin `seq` Stream step' (state, begin)
3116 {-# INLINE_LATE step' #-}
3117 step' gst (st, acc) = acc `seq` do
3118 r <- step (adaptState gst) st
3122 y `seq` return (Yield y (s, y))
3123 Skip s -> return $ Skip (s, acc)
3126{-# INLINE_NORMAL postscanl' #-}
3127postscanl' :: Monad m => (a -> b -> a) -> a -> Stream m b -> Stream m a
3128postscanl' f = postscanlM' (\a b -> return (f a b))
3130{-# INLINE_NORMAL postscanlM #-}
3131postscanlM :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
3132postscanlM fstep begin (Stream step state) = Stream step' (state, begin)
3134 {-# INLINE_LATE step' #-}
3135 step' gst (st, acc) = do
3136 r <- step (adaptState gst) st
3140 return (Yield y (s, y))
3141 Skip s -> return $ Skip (s, acc)
3144{-# INLINE_NORMAL postscanl #-}
3145postscanl :: Monad m => (a -> b -> a) -> a -> Stream m b -> Stream m a
3146postscanl f = postscanlM (\a b -> return (f a b))
3148{-# INLINE_NORMAL scanlM' #-}
3149scanlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
3150scanlM' fstep begin s = begin `seq` (begin `cons` postscanlM' fstep begin s)
3152{-# INLINE scanl' #-}
3153scanl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
3154scanl' f = scanlM' (\a b -> return (f a b))
3156{-# INLINE_NORMAL scanlM #-}
3157scanlM :: Monad m => (b -> a -> m b) -> b -> Stream m a -> Stream m b
3158scanlM fstep begin s = begin `cons` postscanlM fstep begin s
3161scanl :: Monad m => (b -> a -> b) -> b -> Stream m a -> Stream m b
3162scanl f = scanlM (\a b -> return (f a b))
3164{-# INLINE_NORMAL scanl1M #-}
3165scanl1M :: Monad m => (a -> a -> m a) -> Stream m a -> Stream m a
3166scanl1M fstep (Stream step state) = Stream step' (state, Nothing)
3168 {-# INLINE_LATE step' #-}
3169 step' gst (st, Nothing) = do
3172 Yield x s -> return $ Yield x (s, Just x)
3173 Skip s -> return $ Skip (s, Nothing)
3176 step' gst (st, Just acc) = do
3181 return $ Yield z (s, Just z)
3182 Skip s -> return $ Skip (s, Just acc)
3185{-# INLINE scanl1 #-}
3186scanl1 :: Monad m => (a -> a -> a) -> Stream m a -> Stream m a
3187scanl1 f = scanl1M (\x y -> return (f x y))
3189{-# INLINE_NORMAL scanl1M' #-}
3190scanl1M' :: Monad m => (a -> a -> m a) -> Stream m a -> Stream m a
3191scanl1M' fstep (Stream step state) = Stream step' (state, Nothing)
3193 {-# INLINE_LATE step' #-}
3194 step' gst (st, Nothing) = do
3197 Yield x s -> x `seq` return $ Yield x (s, Just x)
3198 Skip s -> return $ Skip (s, Nothing)
3201 step' gst (st, Just acc) = acc `seq` do
3206 z `seq` return $ Yield z (s, Just z)
3207 Skip s -> return $ Skip (s, Just acc)
3210{-# INLINE scanl1' #-}
3211scanl1' :: Monad m => (a -> a -> a) -> Stream m a -> Stream m a
3212scanl1' f = scanl1M' (\x y -> return (f x y))
3215tap :: Monad m => Fold m a b -> Stream m a -> Stream m a
3216tap (Fold fstep initial extract) (Stream step state) = Stream step' Nothing
3220 step' _ Nothing = do
3222 return $ Skip (Just (r, state))
3224 step' gst (Just (acc, st)) = acc `seq` do
3229 return $ Yield x (Just (acc', s))
3230 Skip s -> return $ Skip (Just (acc, s))
3235{-# INLINE_NORMAL tapRate #-}
3237 (MonadAsync m, MonadCatch m)
3242tapRate samplingRate action (Stream step state) = Stream step' Nothing
3244 {-# NOINLINE loop #-}
3245 loop countRef prev = do
3248 (do liftIO $ threadDelay (round $ samplingRate * 1000000)
3249 i <- liftIO $ readIORef countRef
3250 let !diff = i - prev
3253 (\(e :: AsyncException) -> do
3254 i <- liftIO $ readIORef countRef
3255 let !diff = i - prev
3257 throwM (MC.toException e))
3260 {-# INLINE_LATE step' #-}
3261 step' _ Nothing = do
3262 countRef <- liftIO $ newIORef 0
3263 tid <- fork $ loop countRef 0
3264 ref <- liftIO $ newIORef ()
3265 _ <- liftIO $ mkWeakIORef ref (killThread tid)
3266 return $ Skip (Just (countRef, tid, state, ref))
3268 step' gst (Just (countRef, tid, st, ref)) = do
3272 liftIO $ modifyIORef' countRef (+ 1)
3273 return $ Yield x (Just (countRef, tid, s, ref))
3274 Skip s -> return $ Skip (Just (countRef, tid, s, ref))
3276 liftIO $ killThread tid
3280-------------------------------------------------------------------------------
3282-------------------------------------------------------------------------------
3284{-# INLINE_NORMAL takeWhileM #-}
3285takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
3286takeWhileM f (Stream step state) = Stream step' state
3288 {-# INLINE_LATE step' #-}
3294 return $ if b then Yield x s else Stop
3295 Skip s -> return $ Skip s
3298{-# INLINE takeWhile #-}
3299takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
3300takeWhile f = takeWhileM (return . f)
3302{-# INLINE_NORMAL drop #-}
3303drop :: Monad m => Int -> Stream m a -> Stream m a
3304drop n (Stream step state) = Stream step' (state, Just n)
3306 {-# INLINE_LATE step' #-}
3307 step' gst (st, Just i)
3312 Yield _ s -> Skip (s, Just (i - 1))
3313 Skip s -> Skip (s, Just i)
3315 | otherwise = return $ Skip (st, Nothing)
3317 step' gst (st, Nothing) = do
3321 Yield x s -> Yield x (s, Nothing)
3322 Skip s -> Skip (s, Nothing)
3325data DropWhileState s a
3327 | DropWhileYield a s
3330{-# INLINE_NORMAL dropWhileM #-}
3331dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
3332dropWhileM f (Stream step state) = Stream step' (DropWhileDrop state)
3334 {-# INLINE_LATE step' #-}
3335 step' gst (DropWhileDrop st) = do
3341 then return $ Skip (DropWhileDrop s)
3342 else return $ Skip (DropWhileYield x s)
3343 Skip s -> return $ Skip (DropWhileDrop s)
3346 step' gst (DropWhileNext st) = do
3349 Yield x s -> return $ Skip (DropWhileYield x s)
3350 Skip s -> return $ Skip (DropWhileNext s)
3353 step' _ (DropWhileYield x st) = return $ Yield x (DropWhileNext st)
3355{-# INLINE dropWhile #-}
3356dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
3357dropWhile f = dropWhileM (return . f)
3359{-# INLINE_NORMAL filterM #-}
3360filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
3361filterM f (Stream step state) = Stream step' state
3363 {-# INLINE_LATE step' #-}
3372 Skip s -> return $ Skip s
3375{-# INLINE filter #-}
3376filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
3377filter f = filterM (return . f)
3379{-# INLINE_NORMAL uniq #-}
3380uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
3381uniq (Stream step state) = Stream step' (Nothing, state)
3383 {-# INLINE_LATE step' #-}
3384 step' gst (Nothing, st) = do
3387 Yield x s -> return $ Yield x (Just x, s)
3388 Skip s -> return $ Skip (Nothing, s)
3390 step' gst (Just x, st) = do
3393 Yield y s | x == y -> return $ Skip (Just x, s)
3394 | otherwise -> return $ Yield y (Just y, s)
3395 Skip s -> return $ Skip (Just x, s)
3398------------------------------------------------------------------------------
3399-- Transformation by Mapping
3400------------------------------------------------------------------------------
3402{-# INLINE_NORMAL sequence #-}
3403sequence :: Monad m => Stream m (m a) -> Stream m a
3404sequence (Stream step state) = Stream step' state
3406 {-# INLINE_LATE step' #-}
3408 r <- step (adaptState gst) st
3410 Yield x s -> x >>= \a -> return (Yield a s)
3411 Skip s -> return $ Skip s
3414------------------------------------------------------------------------------
3416------------------------------------------------------------------------------
3418data LoopState x s = FirstYield s
3419 | InterspersingYield s
3422{-# INLINE_NORMAL intersperseM #-}
3423intersperseM :: Monad m => m a -> Stream m a -> Stream m a
3424intersperseM m (Stream step state) = Stream step' (FirstYield state)
3426 {-# INLINE_LATE step' #-}
3427 step' gst (FirstYield st) = do
3431 Yield x s -> Skip (YieldAndCarry x s)
3432 Skip s -> Skip (FirstYield s)
3435 step' gst (InterspersingYield st) = do
3440 return $ Yield a (YieldAndCarry x s)
3441 Skip s -> return $ Skip $ InterspersingYield s
3444 step' _ (YieldAndCarry x st) = return $ Yield x (InterspersingYield st)
3449 | SuffixYield a (SuffixState s a)
3451{-# INLINE_NORMAL intersperseSuffix #-}
3452intersperseSuffix :: forall m a. Monad m => m a -> Stream m a -> Stream m a
3453intersperseSuffix action (Stream step state) = Stream step' (SuffixElem state)
3455 {-# INLINE_LATE step' #-}
3456 step' gst (SuffixElem st) = do
3459 Yield x s -> Skip (SuffixYield x (SuffixSuffix s))
3460 Skip s -> Skip (SuffixElem s)
3463 step' _ (SuffixSuffix st) = do
3464 action >>= \r -> return $ Skip (SuffixYield r (SuffixElem st))
3466 step' _ (SuffixYield x next) = return $ Yield x next
3468data SuffixSpanState s a
3469 = SuffixSpanElem s Int
3470 | SuffixSpanSuffix s
3471 | SuffixSpanYield a (SuffixSpanState s a)
3475-- | intersperse after every n items
3476{-# INLINE_NORMAL intersperseSuffixBySpan #-}
3477intersperseSuffixBySpan :: forall m a. Monad m
3478 => Int -> m a -> Stream m a -> Stream m a
3479intersperseSuffixBySpan n action (Stream step state) =
3480 Stream step' (SuffixSpanElem state n)
3482 {-# INLINE_LATE step' #-}
3483 step' gst (SuffixSpanElem st i) | i > 0 = do
3486 Yield x s -> Skip (SuffixSpanYield x (SuffixSpanElem s (i - 1)))
3487 Skip s -> Skip (SuffixSpanElem s i)
3488 Stop -> if i == n then Stop else Skip SuffixSpanLast
3489 step' _ (SuffixSpanElem st _) = return $ Skip (SuffixSpanSuffix st)
3491 step' _ (SuffixSpanSuffix st) = do
3492 action >>= \r -> return $ Skip (SuffixSpanYield r (SuffixSpanElem st n))
3494 step' _ (SuffixSpanLast) = do
3495 action >>= \r -> return $ Skip (SuffixSpanYield r SuffixSpanStop)
3497 step' _ (SuffixSpanYield x next) = return $ Yield x next
3499 step' _ (SuffixSpanStop) = return Stop
3501{-# INLINE intersperse #-}
3502intersperse :: Monad m => a -> Stream m a -> Stream m a
3503intersperse a = intersperseM (return a)
3505{-# INLINE_NORMAL insertBy #-}
3506insertBy :: Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
3507insertBy cmp a (Stream step state) = Stream step' (state, False, Nothing)
3509 {-# INLINE_LATE step' #-}
3510 step' gst (st, False, _) = do
3513 Yield x s -> case cmp a x of
3514 GT -> return $ Yield x (s, False, Nothing)
3515 _ -> return $ Yield a (s, True, Just x)
3516 Skip s -> return $ Skip (s, False, Nothing)
3517 Stop -> return $ Yield a (st, True, Nothing)
3519 step' _ (_, True, Nothing) = return Stop
3521 step' gst (st, True, Just prev) = do
3524 Yield x s -> return $ Yield prev (s, True, Just x)
3525 Skip s -> return $ Skip (s, True, Just prev)
3526 Stop -> return $ Yield prev (st, True, Nothing)
3528------------------------------------------------------------------------------
3530------------------------------------------------------------------------------
3532{-# INLINE_NORMAL deleteBy #-}
3533deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
3534deleteBy eq x (Stream step state) = Stream step' (state, False)
3536 {-# INLINE_LATE step' #-}
3537 step' gst (st, False) = do
3540 Yield y s -> return $
3541 if eq x y then Skip (s, True) else Yield y (s, False)
3542 Skip s -> return $ Skip (s, False)
3545 step' gst (st, True) = do
3548 Yield y s -> return $ Yield y (s, True)
3549 Skip s -> return $ Skip (s, True)
3552------------------------------------------------------------------------------
3553-- Transformation by Map and Filter
3554------------------------------------------------------------------------------
3556-- XXX Will this always fuse properly?
3557{-# INLINE_NORMAL mapMaybe #-}
3558mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
3559mapMaybe f = fmap fromJust . filter isJust . map f
3561{-# INLINE_NORMAL mapMaybeM #-}
3562mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Stream m a -> Stream m b
3563mapMaybeM f = fmap fromJust . filter isJust . mapM f
3565------------------------------------------------------------------------------
3567------------------------------------------------------------------------------
3569{-# INLINE_NORMAL indexed #-}
3570indexed :: Monad m => Stream m a -> Stream m (Int, a)
3571indexed (Stream step state) = Stream step' (state, 0)
3573 {-# INLINE_LATE step' #-}
3574 step' gst (st, i) = i `seq` do
3575 r <- step (adaptState gst) st
3577 Yield x s -> return $ Yield (i, x) (s, i+1)
3578 Skip s -> return $ Skip (s, i)
3581{-# INLINE_NORMAL indexedR #-}
3582indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
3583indexedR m (Stream step state) = Stream step' (state, m)
3585 {-# INLINE_LATE step' #-}
3586 step' gst (st, i) = i `seq` do
3587 r <- step (adaptState gst) st
3589 Yield x s -> let i' = i - 1
3590 in return $ Yield (i, x) (s, i')
3591 Skip s -> return $ Skip (s, i)
3594{-# INLINE_NORMAL zipWithM #-}
3596 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
3597zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)
3599 {-# INLINE_LATE step #-}
3600 step gst (sa, sb, Nothing) = do
3601 r <- stepa (adaptState gst) sa
3604 Yield x sa' -> Skip (sa', sb, Just x)
3605 Skip sa' -> Skip (sa', sb, Nothing)
3608 step gst (sa, sb, Just x) = do
3609 r <- stepb (adaptState gst) sb
3613 return $ Yield z (sa, sb', Nothing)
3614 Skip sb' -> return $ Skip (sa, sb', Just x)
3617#if __GLASGOW_HASKELL__ >= 801
3618{-# RULES "zipWithM xs xs"
3619 forall f xs. zipWithM @Identity f xs xs = mapM (\x -> f x x) xs #-}
3622{-# INLINE zipWith #-}
3623zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
3624zipWith f = zipWithM (\a b -> return (f a b))
3626-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
3627-- are generated concurrently.
3629{-# INLINE_NORMAL zipAsyncWithM #-}
3630zipAsyncWithM :: MonadAsync m
3631 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
3632zipAsyncWithM f m1 m2 = Stream step Nothing
3636 {-# INLINE_LATE step #-}
3637 step gst Nothing = do
3638 ma <- mkParallel (adaptState gst) m1
3639 mb <- mkParallel (adaptState gst) m2
3640 return $ Skip (Just $ zipWithM f ma mb)
3642 step gst (Just (UnStream step1 st)) = do
3645 Yield a s -> Yield a (Just $ Stream step1 s)
3646 Skip s -> Skip (Just $ Stream step1 s)
3649-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
3650-- are generated concurrently.
3652{-# INLINE zipAsyncWith #-}
3653zipAsyncWith :: MonadAsync m
3654 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
3655zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
3657------------------------------------------------------------------------------
3659------------------------------------------------------------------------------
3661{-# INLINE_NORMAL mergeByM #-}
3664 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
3665mergeByM cmp (Stream stepa ta) (Stream stepb tb) =
3666 Stream step (Just ta, Just tb, Nothing, Nothing)
3668 {-# INLINE_LATE step #-}
3670 -- one of the values is missing, and the corresponding stream is running
3671 step gst (Just sa, sb, Nothing, b) = do
3674 Yield a sa' -> Skip (Just sa', sb, Just a, b)
3675 Skip sa' -> Skip (Just sa', sb, Nothing, b)
3676 Stop -> Skip (Nothing, sb, Nothing, b)
3678 step gst (sa, Just sb, a, Nothing) = do
3681 Yield b sb' -> Skip (sa, Just sb', a, Just b)
3682 Skip sb' -> Skip (sa, Just sb', a, Nothing)
3683 Stop -> Skip (sa, Nothing, a, Nothing)
3685 -- both the values are available
3686 step _ (sa, sb, Just a, Just b) = do
3688 return $ case res of
3689 GT -> Yield b (sa, sb, Just a, Nothing)
3690 _ -> Yield a (sa, sb, Nothing, Just b)
3692 -- one of the values is missing, corresponding stream is done
3693 step _ (Nothing, sb, Nothing, Just b) =
3694 return $ Yield b (Nothing, sb, Nothing, Nothing)
3696 step _ (sa, Nothing, Just a, Nothing) =
3697 return $ Yield a (sa, Nothing, Nothing, Nothing)
3699 step _ (Nothing, Nothing, Nothing, Nothing) = return Stop
3701{-# INLINE mergeBy #-}
3704 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
3705mergeBy cmp = mergeByM (\a b -> return $ cmp a b)
3707------------------------------------------------------------------------------
3708-- Transformation comprehensions
3709------------------------------------------------------------------------------
3711{-# INLINE_NORMAL the #-}
3712the :: (Eq a, Monad m) => Stream m a -> m (Maybe a)
3713the (Stream step state) = go state
3716 r <- step defState st
3718 Yield x s -> go' x s
3720 Stop -> return Nothing
3722 r <- step defState st
3724 Yield x s | x == n -> go' n s
3725 | otherwise -> return Nothing
3727 Stop -> return (Just n)
3729{-# INLINE runFold #-}
3730runFold :: (Monad m) => Fold m a b -> Stream m a -> m b
3731runFold (Fold step begin done) = foldlMx' step begin done
3733-------------------------------------------------------------------------------
3734-- Concurrent application and fold
3735-------------------------------------------------------------------------------
3737-- Using StreamD the worker stream producing code can fuse with the code to
3738-- queue output to the SVar giving some perf boost.
3740-- Note that StreamD can only be used in limited situations, specifically, we
3741-- cannot implement joinStreamVarPar using this.
3743-- XXX make sure that the SVar passed is a Parallel style SVar.
3745-- | Fold the supplied stream to the SVar asynchronously using Parallel
3746-- concurrency style.
3747-- {-# INLINE_NORMAL toSVarParallel #-}
3748{-# INLINE toSVarParallel #-}
3749toSVarParallel :: MonadAsync m
3750 => State t m a -> SVar t m a -> Stream m a -> m ()
3751toSVarParallel st sv xs =
3752 if svarInspectMode sv
3756 case getYieldLimit st of
3757 Nothing -> doFork (work Nothing)
3759 (handleChildException sv)
3760 Just _ -> doFork (workLim Nothing)
3762 (handleChildException sv)
3767 {-# NOINLINE work #-}
3768 work info = (runFold (FL.toParallelSVar sv info) xs)
3770 {-# NOINLINE workLim #-}
3771 workLim info = runFold (FL.toParallelSVarLimited sv info) xs
3773 {-# NOINLINE forkWithDiag #-}
3775 -- We do not use workerCount in case of ParallelVar but still there is
3776 -- no harm in maintaining it correctly.
3777 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
3779 -- This allocation matters when significant number of workers are being
3780 -- sent. We allocate it only when needed. The overhead increases by 4x.
3782 case yieldRateInfo sv of
3783 Nothing -> return Nothing
3784 Just _ -> liftIO $ do
3785 cntRef <- newIORef 0
3786 t <- getTime Monotonic
3787 lat <- newIORef (0, t)
3788 return $ Just WorkerInfo
3789 { workerYieldMax = 0
3790 , workerYieldCount = cntRef
3791 , workerLatencyStart = lat
3794 case getYieldLimit st of
3795 Nothing -> doFork (work winfo)
3797 (handleChildException sv)
3798 Just _ -> doFork (workLim winfo)
3800 (handleChildException sv)
3803{-# INLINE_NORMAL mkParallel #-}
3804mkParallel :: MonadAsync m => State t m a -> Stream m a -> m (Stream m a)
3806 sv <- newParallelVar StopNone defState
3807 toSVarParallel st sv m
3808 return $ fromSVar sv
3810{-# INLINE_NORMAL applyParallel #-}
3811applyParallel :: MonadAsync m
3812 => (Stream m a -> Stream m b) -> Stream m a -> Stream m b
3813applyParallel f m = Stream step Nothing
3816 {-# INLINE_LATE step #-}
3817 step gst Nothing = do
3818 sv <- newParallelVar StopNone (adaptState gst)
3819 toSVarParallel (adaptState gst) sv m
3820 return $ Skip $ Just $ f (fromSVar sv)
3822 step gst (Just (UnStream step1 st)) = do
3825 Yield a s -> Yield a (Just $ Stream step1 s)
3826 Skip s -> Skip (Just $ Stream step1 s)
3829{-# INLINE foldParallel #-}
3830foldParallel :: (K.IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
3831foldParallel f m = do
3832 sv <- newParallelVar StopNone defState
3833 toSVarParallel defState sv (toStreamD m)
3834 f $ fromStreamD $ fromSVar sv
3836-------------------------------------------------------------------------------
3838-------------------------------------------------------------------------------
3840-- | Create an SVar with a fold consumer that will fold any elements sent to it
3841-- using the supplied fold function.
3842{-# INLINE newFoldSVar #-}
3843newFoldSVar :: MonadAsync m => State t m a -> Fold m a b -> m (SVar t m a)
3844newFoldSVar stt f = do
3845 -- Buffer size for the SVar is derived from the current state
3846 sv <- newParallelVar StopAny (adaptState stt)
3847 -- Add the producer thread-id to the SVar.
3848 liftIO myThreadId >>= modifyThread sv
3849 void $ doFork (work sv) (svarMrun sv) (handleFoldException sv)
3854 {-# NOINLINE work #-}
3855 work sv = void $ runFold f $ fromProducer sv
3857data TapState sv st = TapInit | Tapping sv st | TapDone st
3859{-# INLINE_NORMAL tapAsync #-}
3860tapAsync :: MonadAsync m => Fold m a b -> Stream m a -> Stream m a
3861tapAsync f (Stream step1 state1) = Stream step TapInit
3865 -- In general, a Stop event would come equipped with the result
3866 -- of the fold. It is not used here but it would be useful in
3867 -- applicative and distribute.
3868 done <- fromConsumer svr
3869 when (not done) $ do
3870 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
3871 $ takeMVar (outputDoorBellFromConsumer svr)
3875 liftIO $ sendStop svr Nothing
3876 -- drain/wait until a stop event arrives from the fold.
3879 {-# INLINE_LATE step #-}
3880 step gst TapInit = do
3881 sv <- newFoldSVar gst f
3882 return $ Skip (Tapping sv state1)
3884 step gst (Tapping sv st) = do
3888 done <- pushToFold sv a
3891 -- XXX we do not need to wait synchronously here
3893 return $ Yield a (TapDone s)
3894 else return $ Yield a (Tapping sv s)
3895 Skip s -> return $ Skip (Tapping sv s)
3900 step gst (TapDone st) = do
3903 Yield a s -> Yield a (TapDone s)
3904 Skip s -> Skip (TapDone s)
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