2-- Module : Streamly.Internal.Data.Fold
3-- Copyright : (c) 2019 Composewell Technologies
4-- (c) 2013 Gabriel Gonzalez
6-- Maintainer : streamly@composewell.com
7-- Stability : experimental
10-- See "Streamly.Data.Fold" for an overview and
11-- "Streamly.Internal.Data.Fold.Types" for design notes.
13-- IMPORTANT: keep the signatures consistent with the folds in Streamly.Prelude
15module Streamly.Internal.Data.Fold
22 -- | Which constructor to use?
24 -- * @foldl*@: If the fold never terminates i.e. does not use the 'Done'
25 -- constructor otherwise use the @mkFold@ variants.
26 -- * @*M@: Use the @M@ suffix variants if any of the step, initial, or
27 -- extract function is monadic, otherwise use the pure variants.
28 -- * @*_@: Use the @_@ suffix variants if the extract function is 'id' or
40 -- | Monadic functions useful with mapM/lmapM on folds or streams.
50 -- *** Semigroups and Monoids
71 -- *** Saturating Reducers
72 -- | 'product' terminates if it becomes 0. Other folds can theoretically
73 -- saturate on bounded types, and therefore terminate, however, they will
74 -- run forever on unbounded types like Integer/Double.
83 -- | Avoid using these folds in scalable or performance critical
84 -- applications, they buffer all the input in GC memory which can be
85 -- detrimental to performance if the input is large.
97 -- | Stateful transformation of the elements. Useful in combination with
98 -- the 'scanMaybe' combinator. For scanners the result of the fold is
99 -- usually a transformation of the current element rather than an
100 -- aggregation of all elements till now.
102 -- , nthLast -- using Ring array
109 -- | Useful in combination with the 'scanMaybe' combinator.
120 -- ** Terminating Folds
121 -- Element folds. Terminate after inspecting one element. All these can be
122 -- implemented in terms of the 'maybe' fold.
128 -- Sequence folds. Terminate after inspecting a sequence of elements.
147 -- | Useful in combination with the 'scanMaybe' combinator.
162 -- ** Transforming the Monad
166 -- ** Mapping on output
169 -- ** Mapping on Input
188 -- ** Filtering Input
204 -- | Insertion adds more elements to the stream.
233 -- , splitIn -- sessionN
235 -- ** Parallel Distribution
246 -- These two can be expressed using lmap/lmapM and unzip
252 -- ** Parallel Alternative
264 -- | Direct values in the input stream to different folds using an n-ary
265 -- fold selector. 'demux' is a generalization of 'classify' (and
266 -- 'partition') where each key of the classifier can use a different fold.
273 -- TODO: These can be implemented using the above operations
274 -- , demuxWithSel -- Stop when the fold for the specified key stops
275 -- , demuxWithMin -- Stop when any of the folds stop
276 -- , demuxWithAll -- Stop when all the folds stop (run once)
279 -- | In an input stream of key value pairs fold values for different keys
280 -- in individual output buckets using the given fold. 'classify' is a
281 -- special case of 'demux' where all the branches of the demultiplexer use
284 -- Different types of maps can be used with these combinators via the IsMap
285 -- type class. Hashmap performs better when there are more collisions, trie
286 -- Map performs better otherwise. Trie has an advantage of sorting the keys
287 -- at the same time. For example if we want to store a dictionary of words
288 -- and their meanings then trie Map would be better if we also want to
289 -- display them in sorted order.
309 -- | Normally you would run a fold to completion by supplying it a stream,
310 -- e.g. using 'Stream.fold'. However, you could also run a fold partially
311 -- by using 'duplicate' on it and then running it with a stream.
312 -- Alternatively, 'initialize', 'snoc' and 'finish' can be used to run a
313 -- fold incrementally, however, that may not be the most efficient way to
330#include "ArrayMacros.h"
332import Control.Monad (void)
333import Control.Monad.IO.Class (MonadIO(..))
334import Data.Bifunctor (first)
335import Data.Bits (shiftL, shiftR, (.|.), (.&.))
336import Data.Either (isLeft, isRight, fromLeft, fromRight)
337import Data.Functor.Identity (Identity(..))
338import Data.Int (Int64)
339import Data.IORef (newIORef, readIORef, writeIORef)
340import Data.Map.Strict (Map)
341import Data.Word (Word32)
342import Foreign.Storable (peek, sizeOf)
343import Streamly.Internal.Data.IsMap (IsMap(..))
344import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
345import Streamly.Internal.Data.Pipe.Type (Pipe (..), PipeState(..))
346import Streamly.Internal.Data.Unboxed (Unboxed)
347import Streamly.Internal.Data.Unfold.Type (Unfold(..))
348import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
349import Streamly.Internal.Data.Stream.Type (Stream)
351import qualified Data.IntSet as IntSet
352import qualified Data.Set as Set
353import qualified Prelude
354import qualified Streamly.Internal.Data.Array.Unboxed.Mut.Type as MA
355import qualified Streamly.Internal.Data.Array.Unboxed.Type as Array
356import qualified Streamly.Internal.Data.Fold.Window as FoldW
357import qualified Streamly.Internal.Data.IsMap as IsMap
358import qualified Streamly.Internal.Data.Pipe.Type as Pipe
359import qualified Streamly.Internal.Data.Ring.Unboxed as Ring
360import qualified Streamly.Internal.Data.Stream.StreamD.Type as StreamD
361import qualified Streamly.Internal.Data.Stream.Type as Stream
364 ( filter, foldl1, drop, dropWhile, take, takeWhile, zipWith
365 , foldl, foldr, map, mapM_, sequence, all, any, sum, product, elem
366 , notElem, maximum, minimum, head, last, tail, length, null
367 , reverse, iterate, init, and, or, lookup, (!!)
368 , scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip
369 , span, splitAt, break, mapM, zip, maybe)
370import Streamly.Internal.Data.Fold.Type
374-- >>> :set -package streamly
375-- >>> import Prelude hiding (break, map, span, splitAt)
376-- >>> import qualified Streamly.Data.Array.Unboxed as Array
377-- >>> import qualified Streamly.Data.Fold as Fold
378-- >>> import qualified Streamly.Data.Stream as Stream
379-- >>> import qualified Streamly.Internal.Data.Array.Unboxed.Mut.Type as MA
380-- >>> import qualified Streamly.Internal.Data.Fold as Fold
381-- >>> import qualified Streamly.Internal.Data.Fold.Type as Fold
382-- >>> import qualified Streamly.Internal.Data.Fold.Window as FoldW
383-- >>> import qualified Streamly.Internal.Data.Parser as Parser
384-- >>> import qualified Streamly.Internal.Data.Stream.Type as Stream
385-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
386-- >>> import Streamly.Internal.Data.Stream.Type (Stream)
387-- >>> import Data.IORef (newIORef, readIORef, writeIORef)
389------------------------------------------------------------------------------
391------------------------------------------------------------------------------
393-- | Change the underlying monad of a fold
396hoist :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b
397hoist f (Fold step initial extract) =
398 Fold (\x a -> f $ step x a) (f initial) (f . extract)
400-- | Adapt a pure fold to any monad
402-- > generally = Fold.hoist (return . runIdentity)
405generally :: Monad m => Fold Identity a b -> Fold m a b
406generally = hoist (return . runIdentity)
408------------------------------------------------------------------------------
409-- Transformations on fold inputs
410------------------------------------------------------------------------------
412-- | Flatten the monadic output of a fold to pure output.
415{-# DEPRECATED sequence "Use \"rmapM id\" instead" #-}
416{-# INLINE sequence #-}
417sequence :: Monad m => Fold m a (m b) -> Fold m a b
420-- | Map a monadic function on the output of a fold.
423{-# DEPRECATED mapM "Use rmapM instead" #-}
425mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
429-- >>> mapMaybeM f = Fold.lmapM f . Fold.catMaybes
431{-# INLINE mapMaybeM #-}
432mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Fold m b r -> Fold m a r
433mapMaybeM f = lmapM f . catMaybes
435-- | @mapMaybe f fold@ maps a 'Maybe' returning function @f@ on the input of
436-- the fold, filters out 'Nothing' elements, and return the values extracted
439-- >>> mapMaybe f = Fold.lmap f . Fold.catMaybes
440-- >>> mapMaybe f = Fold.mapMaybeM (return . f)
442-- >>> f x = if even x then Just x else Nothing
443-- >>> fld = Fold.mapMaybe f Fold.toList
444-- >>> Stream.fold fld (Stream.enumerateFromTo 1 10)
448{-# INLINE mapMaybe #-}
449mapMaybe :: Monad m => (a -> Maybe b) -> Fold m b r -> Fold m a r
450mapMaybe f = lmap f . catMaybes
452------------------------------------------------------------------------------
453-- Transformations on fold inputs
454------------------------------------------------------------------------------
456-- | Apply a monadic function on the input and return the input.
458-- >>> Stream.fold (Fold.lmapM (Fold.tracing print) Fold.drain) $ (Stream.enumerateFromTo (1 :: Int) 2)
464{-# INLINE tracing #-}
465tracing :: Monad m => (a -> m b) -> (a -> m a)
466tracing f x = void (f x) >> return x
468-- | Apply a monadic function to each element flowing through and discard the
471-- >>> Stream.fold (Fold.trace print Fold.drain) $ (Stream.enumerateFromTo (1 :: Int) 2)
475-- >>> trace f = Fold.lmapM (Fold.tracing f)
479trace :: Monad m => (a -> m b) -> Fold m a r -> Fold m a r
480trace f = lmapM (tracing f)
484-- | Apply a transformation on a 'Fold' using a 'Pipe'.
487{-# INLINE transform #-}
488transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
489transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
490 Fold step initial extract
494 initial = first (Tuple' pinitial) <$> finitial
496 step (Tuple' ps fs) x = do
503 go acc (Pipe.Yield b (Consume ps')) = do
507 Partial s -> Partial $ Tuple' ps' s
509 go acc (Pipe.Yield b (Produce ps')) = do
514 Done b2 -> return $ Done b2
515 go acc (Pipe.Continue (Consume ps')) =
516 return $ Partial $ Tuple' ps' acc
517 go acc (Pipe.Continue (Produce ps')) = do
521 extract (Tuple' _ fs) = fextract fs
523{-# INLINE scanWith #-}
524scanWith :: Monad m => Bool -> Fold m a b -> Fold m b c -> Fold m a c
525scanWith isMany (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
526 Fold step initial extract
530 {-# INLINE runStep #-}
531 runStep actionL sR = do
539 then runStep initialL sR1
540 else Done <$> extractR sR1
541 Done bR -> return $ Done bR
547 Partial sR1 -> Partial (sL, sR1)
553 Partial sR -> runStep initialL sR
554 Done b -> return $ Done b
556 step (sL, sR) x = runStep (stepL sL x) sR
558 extract = extractR . snd
560-- | Scan the input of a 'Fold' to change it in a stateful manner using another
561-- 'Fold'. The scan stops as soon as the fold terminates.
565scan :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
568-- XXX This does not fuse beacuse of the recursive step. Need to investigate.
570-- | Scan the input of a 'Fold' to change it in a stateful manner using another
571-- 'Fold'. The scan restarts with a fresh state if the fold terminates.
574{-# INLINE scanMany #-}
575scanMany :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
576scanMany = scanWith True
578------------------------------------------------------------------------------
580------------------------------------------------------------------------------
582-- | A filtering scan that deletes the first occurrence of the element in the
583-- stream that satisfies the given equality predicate.
585-- >>> input = Stream.unfold Unfold.fromList [1,3,3,5]
586-- >>> Stream.fold Fold.toList $ Stream.scanMaybe (Fold.deleteBy (==) 3) input
589{-# INLINE_NORMAL deleteBy #-}
590deleteBy :: Monad m => (a -> a -> Bool) -> a -> Fold m a (Maybe a)
591deleteBy eq x0 = fmap extract $ foldl' step (Tuple' False Nothing)
595 step (Tuple' False _) x =
597 then Tuple' True Nothing
598 else Tuple' False (Just x)
599 step (Tuple' True _) x = Tuple' True (Just x)
601 extract (Tuple' _ x) = x
603-- | Provide a sliding window of length 2 elements.
605-- See "Streamly.Internal.Data.Fold.Window".
608slide2 :: Monad m => Fold m (a, Maybe a) b -> Fold m a b
609slide2 (Fold step1 initial1 extract1) = Fold step initial extract
614 first (Tuple' Nothing) <$> initial1
616 step (Tuple' prev s) cur =
617 first (Tuple' (Just cur)) <$> step1 s (cur, prev)
619 extract (Tuple' _ s) = extract1 s
621-- | Drop repeated elements that are adjacent to each other using the supplied
622-- comparison function.
624-- To strip duplicate path separators:
626-- >>> input = Stream.unfold Unfold.fromList "//a//b"
627-- >>> f x y = x == '/' && y == '/'
628-- >>> Stream.fold Fold.toList $ Stream.scanMaybe (Fold.uniqBy f) input
638uniqBy :: Monad m => (a -> a -> Bool) -> Fold m a (Maybe a)
639uniqBy eq = rollingMap f
646 Just x -> if x `eq` curr then Nothing else Just curr
648-- | Drop repeated elements that are adjacent to each other.
650-- >>> uniq = Fold.uniqBy (==)
653uniq :: (Monad m, Eq a) => Fold m a (Maybe a)
656-- | Strip all leading and trailing occurrences of an element passing a
657-- predicate and make all other consecutive occurrences uniq.
659-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
662-- > Stream.prune isSpace (Stream.fromList " hello world! ")
672 -- (Monad m, Eq a) =>
673 (a -> Bool) -> Fold m a (Maybe a)
674prune = error "Not implemented yet!"
676-- | Emit only repeated elements, once.
679repeated :: -- (Monad m, Eq a) =>
681repeated = error "Not implemented yet!"
683-- | Used as a scan. Returns 'Just' for the first occurrence of an element,
684-- returns 'Nothing' for any other occurrences.
686-- >>> stream = Stream.fromList [1::Int,1,2,3,4,4,5,1,5,7]
687-- >>> Stream.fold Fold.toList $ Stream.mapMaybe id $ Stream.postscan Fold.nub stream
692nub :: (Monad m, Ord a) => Fold m a (Maybe a)
693nub = fmap (\(Tuple' _ x) -> x) $ foldl' step initial
697 initial = Tuple' Set.empty Nothing
699 step (Tuple' set _) x =
701 then Tuple' set Nothing
702 else Tuple' (Set.insert x set) (Just x)
704-- | Like 'nub' but specialized to a stream of 'Int', for better performance.
708nubInt :: Monad m => Fold m Int (Maybe Int)
709nubInt = fmap (\(Tuple' _ x) -> x) $ foldl' step initial
713 initial = Tuple' IntSet.empty Nothing
715 step (Tuple' set _) x =
716 if IntSet.member x set
717 then Tuple' set Nothing
718 else Tuple' (IntSet.insert x set) (Just x)
720------------------------------------------------------------------------------
722------------------------------------------------------------------------------
724------------------------------------------------------------------------------
726------------------------------------------------------------------------------
729-- > drainBy f = lmapM f drain
730-- > drainBy = Fold.foldMapM (void . f)
732-- Drain all input after passing it through a monadic function. This is the
733-- dual of mapM_ on stream producers.
735-- See also: 'Streamly.Prelude.mapM_'
738{-# INLINE drainBy #-}
739drainBy :: Monad m => (a -> m b) -> Fold m a ()
740drainBy f = lmapM f drain
742-- | Extract the last element of the input stream, if any.
744-- > last = fmap getLast $ Fold.foldMap (Last . Just)
748last :: Monad m => Fold m a (Maybe a)
749last = foldl1' (\_ x -> x)
751-- | If the stream consists of one or more occurences of the same element
752-- then returns that element else terminates and returns 'Nothing'.
756the :: (Monad m, Eq a) => Fold m a (Maybe a)
757the = foldt' step initial id
761 initial = Partial Nothing
763 step Nothing x = Partial (Just x)
764 step old@(Just x0) x =
769------------------------------------------------------------------------------
771------------------------------------------------------------------------------
773-- | Like 'length', except with a more general 'Num' return value
775-- > genericLength = fmap getSum $ foldMap (Sum . const 1)
778{-# INLINE genericLength #-}
779genericLength :: (Monad m, Num b) => Fold m a b
780genericLength = foldl' (\n _ -> n + 1) 0
782-- | Determine the length of the input stream.
784-- > length = fmap getSum $ Fold.foldMap (Sum . const 1)
788length :: Monad m => Fold m a Int
789length = genericLength
792-- XXX Add a countDistinct window fold
793-- XXX Add a bloom filter fold
795-- | Count non-duplicate elements in the stream.
797-- Equivalent to using 'nub' followed by 'length' on a stream.
799-- >>> countDistinct = Fold.postscan Fold.nub $ Fold.catMaybes $ Fold.length
801-- The memory used is proportional to the number of distinct elements in the
802-- stream, to guard against using too much memory use it as a scan and
803-- terminate if the count reaches more than a threshold.
805-- /Space/: \(\mathcal{O}(n)\)
809{-# INLINE countDistinct #-}
810countDistinct :: (Monad m, Ord a) => Fold m a Int
811countDistinct = postscan nub $ catMaybes length
813countDistinct = fmap (\(Tuple' _ n) -> n) $ foldl' step initial
817 initial = Tuple' Set.empty 0
819 step (Tuple' set n) x = do
825 in Tuple' (Set.insert x set) cnt
828-- | Like 'countDistinct' but specialized to a stream of 'Int', for better
831-- >>> countDistinctInt = Fold.postscan Fold.nubInt $ Fold.catMaybes $ Fold.length
834{-# INLINE countDistinctInt #-}
835countDistinctInt :: Monad m => Fold m Int Int
836countDistinctInt = postscan nubInt $ catMaybes length
838countDistinctInt = fmap (\(Tuple' _ n) -> n) $ foldl' step initial
842 initial = Tuple' IntSet.empty 0
844 step (Tuple' set n) x = do
845 if IntSet.member x set
850 in Tuple' (IntSet.insert x set) cnt
853-- | Determine the sum of all elements of a stream of numbers. Returns additive
854-- identity (@0@) when the stream is empty. Note that this is not numerically
855-- stable for floating point numbers.
857-- >>> sum = FoldW.cumulative FoldW.sum
859-- Same as following but numerically stable:
861-- >>> sum = Fold.foldl' (+) 0
862-- >>> sum = fmap Data.Monoid.getSum $ Fold.foldMap Data.Monoid.Sum
866sum :: (Monad m, Num a) => Fold m a a
867sum = FoldW.cumulative FoldW.sum
869-- | Determine the product of all elements of a stream of numbers. Returns
870-- multiplicative identity (@1@) when the stream is empty. The fold terminates
871-- when it encounters (@0@) in its input.
873-- Compare with @Fold.foldMap Product@.
876-- /Since 0.8.0 (Added 'Eq' constraint)/
877{-# INLINE product #-}
878product :: (Monad m, Num a, Eq a) => Fold m a a
879product = foldt' step (Partial 1) id
888------------------------------------------------------------------------------
890------------------------------------------------------------------------------
892-- | Determine the maximum element in a stream using the supplied comparison
896{-# INLINE maximumBy #-}
897maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
898maximumBy cmp = foldl1' max'
909-- maximum = Fold.maximumBy compare
912-- Determine the maximum element in a stream.
914-- Compare with @Fold.foldMap Max@.
917{-# INLINE maximum #-}
918maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
921-- | Computes the minimum element with respect to the given comparison function
924{-# INLINE minimumBy #-}
925minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
926minimumBy cmp = foldl1' min'
935-- | Determine the minimum element in a stream using the supplied comparison
939-- minimum = 'minimumBy' compare
942-- Compare with @Fold.foldMap Min@.
945{-# INLINE minimum #-}
946minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
949------------------------------------------------------------------------------
950-- To Summary (Statistical)
951------------------------------------------------------------------------------
953-- | Compute a numerically stable arithmetic mean of all elements in the input
958mean :: (Monad m, Fractional a) => Fold m a a
959mean = fmap done $ foldl' step begin
965 step (Tuple' x n) y =
967 in Tuple' (x + (y - x) / n1) n1
969 done (Tuple' x _) = x
971-- | Compute a numerically stable (population) variance over all elements in
975{-# DEPRECATED variance "Use the streamly-statistics package instead" #-}
976{-# INLINE variance #-}
977variance :: (Monad m, Fractional a) => Fold m a a
978variance = fmap done $ foldl' step begin
982 begin = Tuple3' 0 0 0
984 step (Tuple3' n mean_ m2) x = Tuple3' n' mean' m2'
989 mean' = (n * mean_ + x) / (n + 1)
991 m2' = m2 + delta * delta * n / (n + 1)
993 done (Tuple3' n _ m2) = m2 / n
995-- | Compute a numerically stable (population) standard deviation over all
996-- elements in the input stream.
999{-# DEPRECATED stdDev "Use the streamly-statistics package instead" #-}
1000{-# INLINE stdDev #-}
1001stdDev :: (Monad m, Floating a) => Fold m a a
1002stdDev = sqrt <$> variance
1004-- | Compute an 'Int' sized polynomial rolling hash
1006-- > H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0
1008-- Where @c1@, @c2@, @cn@ are the elements in the input stream and @k@ is a
1011-- This hash is often used in Rabin-Karp string search algorithm.
1013-- See https://en.wikipedia.org/wiki/Rolling_hash
1016{-# INLINE rollingHashWithSalt #-}
1017rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64
1018rollingHashWithSalt = foldl' step
1022 k = 2891336453 :: Int64
1024 step cksum a = cksum * k + fromIntegral (fromEnum a)
1026-- | A default salt used in the implementation of 'rollingHash'.
1027{-# INLINE defaultSalt #-}
1029defaultSalt = -2578643520546668380
1031-- | Compute an 'Int' sized polynomial rolling hash of a stream.
1033-- > rollingHash = Fold.rollingHashWithSalt defaultSalt
1036{-# INLINE rollingHash #-}
1037rollingHash :: (Monad m, Enum a) => Fold m a Int64
1038rollingHash = rollingHashWithSalt defaultSalt
1040-- | Compute an 'Int' sized polynomial rolling hash of the first n elements of
1043-- > rollingHashFirstN = Fold.take n Fold.rollingHash
1046{-# INLINE rollingHashFirstN #-}
1047rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64
1048rollingHashFirstN n = take n rollingHash
1050-- XXX Compare this with the implementation in Fold.Window, preferrably use the
1051-- latter if performance is good.
1053-- | Apply a function on every two successive elements of a stream. The first
1054-- argument of the map function is the previous element and the second argument
1055-- is the current element. When processing the very first element in the
1056-- stream, the previous element is 'Nothing'.
1060{-# INLINE rollingMapM #-}
1061rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Fold m a b
1062rollingMapM f = Fold step initial extract
1066 -- XXX We need just a postscan. We do not need an initial result here.
1067 -- Or we can supply a default initial result as an argument to rollingMapM.
1068 initial = return $ Partial (Nothing, error "Empty stream")
1070 step (prev, _) cur = do
1072 return $ Partial (Just cur, x)
1074 extract = return . snd
1076{-# INLINE rollingMap #-}
1077rollingMap :: Monad m => (Maybe a -> a -> b) -> Fold m a b
1078rollingMap f = rollingMapM (\x y -> return $ f x y)
1080------------------------------------------------------------------------------
1081-- Monoidal left folds
1082------------------------------------------------------------------------------
1084-- | Append the elements of an input stream to a provided starting value.
1086-- >>> Stream.fold (Fold.sconcat 10) (fmap Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
1090-- sconcat = Fold.foldl' (<>)
1094{-# INLINE sconcat #-}
1095sconcat :: (Monad m, Semigroup a) => a -> Fold m a a
1096sconcat = foldl' (<>)
1098-- | Fold an input stream consisting of monoidal elements using 'mappend'
1101-- >>> Stream.fold Fold.mconcat (fmap Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
1104-- > mconcat = Fold.sconcat mempty
1107{-# INLINE mconcat #-}
1110 , Monoid a) => Fold m a a
1111mconcat = sconcat mempty
1114-- > foldMap f = Fold.lmap f Fold.mconcat
1116-- Make a fold from a pure function that folds the output of the function
1117-- using 'mappend' and 'mempty'.
1119-- >>> Stream.fold (Fold.foldMap Data.Monoid.Sum) $ Stream.enumerateFromTo 1 10
1123{-# INLINE foldMap #-}
1124foldMap :: (Monad m, Monoid b
1125 ) => (a -> b) -> Fold m a b
1126foldMap f = lmap f mconcat
1129-- > foldMapM f = Fold.lmapM f Fold.mconcat
1131-- Make a fold from a monadic function that folds the output of the function
1132-- using 'mappend' and 'mempty'.
1134-- >>> Stream.fold (Fold.foldMapM (return . Data.Monoid.Sum)) $ Stream.enumerateFromTo 1 10
1138{-# INLINE foldMapM #-}
1139foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
1140foldMapM act = foldlM' step (pure mempty)
1146 return $! mappend m m'
1148------------------------------------------------------------------------------
1150------------------------------------------------------------------------------
1153-- This is more efficient than 'Streamly.Internal.Data.Fold.toList'. toList is
1154-- exactly the same as reversing the list after 'toListRev'.
1156-- | Buffers the input stream to a list in the reverse order of the input.
1158-- > toListRev = Fold.foldl' (flip (:)) []
1160-- /Warning!/ working on large lists accumulated as buffers in memory could be
1161-- very inefficient, consider using "Streamly.Array" instead.
1165-- xn : ... : x2 : x1 : []
1166{-# INLINE toListRev #-}
1167toListRev :: Monad m => Fold m a [a]
1168toListRev = foldl' (flip (:)) []
1170------------------------------------------------------------------------------
1172------------------------------------------------------------------------------
1174-- | A fold that drains the first n elements of its input, running the effects
1175-- and discarding the results.
1177-- > drainN n = Fold.take n Fold.drain
1180{-# INLINE drainN #-}
1181drainN :: Monad m => Int -> Fold m a ()
1182drainN n = take n drain
1184------------------------------------------------------------------------------
1186------------------------------------------------------------------------------
1188-- | Like 'index', except with a more general 'Integral' argument
1191{-# INLINE genericIndex #-}
1192genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
1193genericIndex i = foldt' step (Partial 0) (const Nothing)
1200 else Partial (j + 1)
1202-- | Lookup the element at the given index.
1204-- See also: 'Streamly.Prelude.!!'
1208index :: Monad m => Int -> Fold m a (Maybe a)
1211-- | Map a 'Maybe' returning function on the next element in the stream.
1216maybe :: Monad m => (a -> Maybe b) -> Fold m a (Maybe b)
1217maybe f = foldt' (const (Done . f)) (Partial Nothing) id
1219-- | Test if the next element satisfies the supplied predicate.
1222{-# INLINE satisfy #-}
1223satisfy :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1224satisfy f = Fold step (return $ Partial ()) (const (return Nothing))
1228 step () a = return $ Done $ if f a then Just a else Nothing
1232-- "head" and "next" are two alternative names for the same API. head sounds
1233-- apt in the context of lists but next sounds more apt in the context of
1234-- streams where we think in terms of generating and consuming the next element
1235-- rather than taking the head of some static/persistent structure.
1237-- We also want to keep the nomenclature consistent across folds and parsers,
1238-- "head" becomes even more unintuitive for parsers because there are two
1239-- possible variants viz. peek and next.
1241-- Also, the "head" fold creates confusion in situations like
1242-- https://github.com/composewell/streamly/issues/1404 where intuitive
1243-- expectation from head is to consume the entire stream and just give us the
1244-- head. There we want to convey the notion that we consume one element from
1245-- the stream and stop. The name "one" already being used in parsers for this
1246-- purpose sounds more apt from this perspective.
1248-- The source of confusion is perhaps due to the fact that some folds consume
1249-- the entire stream and others terminate early. It may have been clearer if we
1250-- had separate abstractions for the two use cases.
1252-- XXX We can possibly use "head" for the purposes of reducing the entire
1253-- stream to the head element i.e. take the head and drain the rest.
1255-- | Take one element from the stream and stop.
1260one :: Monad m => Fold m a (Maybe a)
1263-- | Extract the first element of the stream, if any.
1265-- >>> head = Fold.one
1268{-# DEPRECATED head "Please use \"one\" instead" #-}
1270head :: Monad m => Fold m a (Maybe a)
1273-- | Returns the first element that satisfies the given predicate.
1277findM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
1278findM predicate = Fold step (return $ Partial ()) (const $ return Nothing)
1287 in f <$> predicate a
1289-- | Returns the first element that satisfies the given predicate.
1293find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1294find p = findM (return . p)
1296-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1297-- first pair where the key equals the given value @a@.
1299-- > lookup = snd <$> Fold.find ((==) . fst)
1302{-# INLINE lookup #-}
1303lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
1304lookup a0 = foldt' step (Partial ()) (const Nothing)
1313-- | Returns the first index that satisfies the given predicate.
1316{-# INLINE findIndex #-}
1317findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
1318findIndex predicate = foldt' step (Partial 0) (const Nothing)
1325 else Partial (i + 1)
1327-- XXX If we have a Continue contructor we won't need a Maybe type for the
1328-- scan. The fold will have the ability to skip producing output.
1330-- | Returns all indices that satisfy the given predicate.
1332{-# INLINE findIndices #-}
1333findIndices :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
1334findIndices predicate =
1335 -- XXX implement by combining indexing and filtering scans
1336 fmap (either (const Nothing) Just) $ foldl' step (Left (-1))
1342 then Right (either id id i + 1)
1343 else Left (either id id i + 1)
1345-- | Find all the indices where the value of the element in the stream is equal
1346-- to the given value.
1348-- >>> elemIndices a = Fold.findIndices (== a)
1350{-# INLINE elemIndices #-}
1351elemIndices :: (Monad m, Eq a) => a -> Fold m a (Maybe Int)
1352elemIndices a = findIndices (== a)
1354-- | Returns the first index where a given value is found in the stream.
1356-- > elemIndex a = Fold.findIndex (== a)
1359{-# INLINE elemIndex #-}
1360elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
1361elemIndex a = findIndex (a ==)
1363------------------------------------------------------------------------------
1365------------------------------------------------------------------------------
1367-- | Return 'True' if the input stream is empty.
1369-- > null = fmap isJust Fold.one
1373null :: Monad m => Fold m a Bool
1374null = foldt' (\() _ -> Done False) (Partial ()) (const True)
1376-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
1378-- >>> Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]
1381-- > any p = Fold.lmap p Fold.or
1385any :: Monad m => (a -> Bool) -> Fold m a Bool
1386any predicate = foldt' step initial id
1390 initial = Partial False
1397-- | Return 'True' if the given element is present in the stream.
1399-- > elem a = Fold.any (== a)
1403elem :: (Eq a, Monad m) => a -> Fold m a Bool
1406-- | Returns 'True' if all elements of a stream satisfy a predicate.
1408-- >>> Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]
1411-- > all p = Fold.lmap p Fold.and
1415all :: Monad m => (a -> Bool) -> Fold m a Bool
1416all predicate = foldt' step initial id
1420 initial = Partial True
1427-- | Returns 'True' if the given element is not present in the stream.
1429-- > notElem a = Fold.all (/= a)
1432{-# INLINE notElem #-}
1433notElem :: (Eq a, Monad m) => a -> Fold m a Bool
1434notElem a = all (a /=)
1436-- | Returns 'True' if all elements are 'True', 'False' otherwise
1438-- > and = Fold.all (== True)
1442and :: Monad m => Fold m Bool Bool
1445-- | Returns 'True' if any element is 'True', 'False' otherwise
1447-- > or = Fold.any (== True)
1451or :: Monad m => Fold m Bool Bool
1454------------------------------------------------------------------------------
1455-- Grouping/Splitting
1456------------------------------------------------------------------------------
1458------------------------------------------------------------------------------
1459-- Grouping without looking at elements
1460------------------------------------------------------------------------------
1462------------------------------------------------------------------------------
1464------------------------------------------------------------------------------
1466-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
1467-- elements of its input are consumed by fold @f1@ and the rest of the stream
1468-- is consumed by fold @f2@.
1470-- >>> let splitAt_ n xs = Stream.fold (Fold.splitAt n Fold.toList Fold.toList) $ Stream.fromList xs
1472-- >>> splitAt_ 6 "Hello World!"
1473-- ("Hello ","World!")
1475-- >>> splitAt_ (-1) [1,2,3]
1478-- >>> splitAt_ 0 [1,2,3]
1481-- >>> splitAt_ 1 [1,2,3]
1484-- >>> splitAt_ 3 [1,2,3]
1487-- >>> splitAt_ 4 [1,2,3]
1490-- > splitAt n f1 f2 = Fold.serialWith (,) (Fold.take n f1) f2
1494{-# INLINE splitAt #-}
1501splitAt n fld = serialWith (,) (take n fld)
1503------------------------------------------------------------------------------
1504-- Element Aware APIs
1505------------------------------------------------------------------------------
1507------------------------------------------------------------------------------
1509------------------------------------------------------------------------------
1511{-# INLINE takingEndByM #-}
1512takingEndByM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
1513takingEndByM p = Fold step initial (return . toMaybe)
1517 initial = return $ Partial Nothing'
1524 else Partial $ Just' a
1528-- >>> takingEndBy p = Fold.takingEndByM (return . p)
1530{-# INLINE takingEndBy #-}
1531takingEndBy :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1532takingEndBy p = takingEndByM (return . p)
1534{-# INLINE takingEndByM_ #-}
1535takingEndByM_ :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
1536takingEndByM_ p = Fold step initial (return . toMaybe)
1540 initial = return $ Partial Nothing'
1547 else Partial $ Just' a
1551-- >>> takingEndBy_ p = Fold.takingEndByM_ (return . p)
1553{-# INLINE takingEndBy_ #-}
1554takingEndBy_ :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1555takingEndBy_ p = takingEndByM_ (return . p)
1557{-# INLINE droppingWhileM #-}
1558droppingWhileM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
1559droppingWhileM p = Fold step initial (return . toMaybe)
1563 initial = return $ Partial Nothing'
1565 step Nothing' a = do
1572 step _ a = return $ Partial $ Just' a
1574{-# INLINE droppingWhile #-}
1575droppingWhile :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1576droppingWhile p = droppingWhileM (return . p)
1578-- Note: Keep this consistent with S.splitOn. In fact we should eliminate
1579-- S.splitOn in favor of the fold.
1581-- XXX Use Fold.many instead once it is fixed.
1583-- | Like 'takeEndBy' but drops the element on which the predicate succeeds.
1585-- >>> Stream.fold (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1588-- >>> Stream.fold Fold.toList $ Stream.foldMany (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1591-- > Stream.splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)
1593-- See 'Streamly.Prelude.splitOnSuffix' for more details on splitting a
1594-- stream using 'takeEndBy_'.
1597{-# INLINE takeEndBy_ #-}
1598takeEndBy_ :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1599-- takeEndBy_ predicate = scanMaybe (takingEndBy_ predicate)
1600takeEndBy_ predicate (Fold fstep finitial fextract) =
1601 Fold step finitial fextract
1606 if not (predicate a)
1608 else Done <$> fextract s
1610-- | Take the input, stop when the predicate succeeds taking the succeeding
1613-- >>> Stream.fold (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1616-- >>> Stream.fold Fold.toList $ Stream.foldMany (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1617-- ["hello\n","there\n"]
1619-- > Stream.splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
1621-- See 'Streamly.Prelude.splitWithSuffix' for more details on splitting a
1622-- stream using 'takeEndBy'.
1625{-# INLINE takeEndBy #-}
1626takeEndBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1627-- takeEndBy predicate = scanMaybe (takingEndBy predicate)
1628takeEndBy predicate (Fold fstep finitial fextract) =
1629 Fold step finitial fextract
1635 if not (predicate a)
1639 Partial s1 -> Done <$> fextract s1
1640 Done b -> return $ Done b
1642------------------------------------------------------------------------------
1643-- Binary splitting on a separator
1644------------------------------------------------------------------------------
1646data SplitOnSeqState acc a rb rh w ck =
1647 SplitOnSeqEmpty !acc
1648 | SplitOnSeqSingle !acc !a
1649 | SplitOnSeqWord !acc !Int !w
1650 | SplitOnSeqWordLoop !acc !w
1651 | SplitOnSeqKR !acc !Int !rb !rh
1652 | SplitOnSeqKRLoop !acc !ck !rb !rh
1654-- XXX Need to add tests for takeEndBySeq, we have tests for takeEndBySeq_ .
1656-- | Continue taking the input until the input sequence matches the supplied
1657-- sequence, taking the supplied sequence as well. If the pattern is empty this
1658-- acts as an identity fold.
1660-- >>> s = Stream.fromList "hello there. How are you?"
1661-- >>> f = Fold.takeEndBySeq (Array.fromList "re") Fold.toList
1662-- >>> Stream.fold f s
1665-- >>> Stream.fold Fold.toList $ Stream.foldMany f s
1666-- ["hello there",". How are"," you?"]
1669{-# INLINE takeEndBySeq #-}
1670takeEndBySeq :: forall m a b. (MonadIO m, Unboxed a, Enum a, Eq a) =>
1674takeEndBySeq patArr (Fold fstep finitial fextract) =
1675 Fold step initial extract
1679 patLen = Array.length patArr
1686 -- XXX Should we match nothing or everything on empty
1688 -- Done <$> fextract acc
1689 return $ Partial $ SplitOnSeqEmpty acc
1691 pat <- liftIO $ Array.unsafeIndexIO 0 patArr
1692 return $ Partial $ SplitOnSeqSingle acc pat
1693 | SIZE_OF(a) * patLen <= sizeOf (undefined :: Word) ->
1694 return $ Partial $ SplitOnSeqWord acc 0 0
1696 (rb, rhead) <- liftIO $ Ring.new patLen
1697 return $ Partial $ SplitOnSeqKR acc 0 rb rhead
1698 Done b -> return $ Done b
1700 -- Word pattern related
1701 maxIndex = patLen - 1
1703 elemBits = SIZE_OF(a) * 8
1706 wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1709 wordPat = wordMask .&. Array.foldl' addToWord 0 patArr
1711 addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1713 -- For Rabin-Karp search
1714 k = 2891336453 :: Word32
1717 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1719 deltaCksum cksum old new =
1720 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1722 -- XXX shall we use a random starting hash or 1 instead of 0?
1723 -- XXX Need to keep this cached across fold calls in foldmany
1724 -- XXX We may need refold to inject the cached state instead of
1725 -- initializing the state every time.
1726 -- XXX Allocation of ring buffer should also be done once
1727 patHash = Array.foldl' addCksum 0 patArr
1729 step (SplitOnSeqEmpty s) x = do
1732 Partial s1 -> return $ Partial $ SplitOnSeqEmpty s1
1733 Done b -> return $ Done b
1734 step (SplitOnSeqSingle s pat) x = do
1738 | pat /= x -> return $ Partial $ SplitOnSeqSingle s1 pat
1739 | otherwise -> Done <$> fextract s1
1740 Done b -> return $ Done b
1741 step (SplitOnSeqWord s idx wrd) x = do
1743 let wrd1 = addToWord wrd x
1746 | idx == maxIndex -> do
1747 if wrd1 .&. wordMask == wordPat
1748 then Done <$> fextract s1
1749 else return $ Partial $ SplitOnSeqWordLoop s1 wrd1
1751 return $ Partial $ SplitOnSeqWord s1 (idx + 1) wrd1
1752 Done b -> return $ Done b
1753 step (SplitOnSeqWordLoop s wrd) x = do
1755 let wrd1 = addToWord wrd x
1758 | wrd1 .&. wordMask == wordPat ->
1759 Done <$> fextract s1
1761 return $ Partial $ SplitOnSeqWordLoop s1 wrd1
1762 Done b -> return $ Done b
1763 step (SplitOnSeqKR s idx rb rh) x = do
1767 rh1 <- liftIO $ Ring.unsafeInsert rb rh x
1770 let fld = Ring.unsafeFoldRing (Ring.ringBound rb)
1771 let !ringHash = fld addCksum 0 rb
1772 if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
1773 then Done <$> fextract s1
1774 else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
1776 return $ Partial $ SplitOnSeqKR s1 (idx + 1) rb rh1
1777 Done b -> return $ Done b
1778 step (SplitOnSeqKRLoop s cksum rb rh) x = do
1782 old <- liftIO $ peek rh
1783 rh1 <- liftIO $ Ring.unsafeInsert rb rh x
1784 let ringHash = deltaCksum cksum old x
1785 if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
1786 then Done <$> fextract s1
1787 else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
1788 Done b -> return $ Done b
1793 SplitOnSeqEmpty s -> s
1794 SplitOnSeqSingle s _ -> s
1795 SplitOnSeqWord s _ _ -> s
1796 SplitOnSeqWordLoop s _ -> s
1797 SplitOnSeqKR s _ _ _ -> s
1798 SplitOnSeqKRLoop s _ _ _ -> s
1801-- | Like 'takeEndBySeq' but discards the matched sequence.
1805{-# INLINE takeEndBySeq_ #-}
1806takeEndBySeq_ :: forall m a b. (MonadIO m, Unboxed a, Enum a, Eq a) =>
1810takeEndBySeq_ patArr (Fold fstep finitial fextract) =
1811 Fold step initial extract
1815 patLen = Array.length patArr
1822 -- XXX Should we match nothing or everything on empty
1824 -- Done <$> fextract acc
1825 return $ Partial $ SplitOnSeqEmpty acc
1827 pat <- liftIO $ Array.unsafeIndexIO 0 patArr
1828 return $ Partial $ SplitOnSeqSingle acc pat
1829 -- XXX Need to add tests for this case
1830 | SIZE_OF(a) * patLen <= sizeOf (undefined :: Word) ->
1831 return $ Partial $ SplitOnSeqWord acc 0 0
1833 (rb, rhead) <- liftIO $ Ring.new patLen
1834 return $ Partial $ SplitOnSeqKR acc 0 rb rhead
1835 Done b -> return $ Done b
1837 -- Word pattern related
1838 maxIndex = patLen - 1
1840 elemBits = SIZE_OF(a) * 8
1843 wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1846 elemMask = (1 `shiftL` elemBits) - 1
1849 wordPat = wordMask .&. Array.foldl' addToWord 0 patArr
1851 addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1853 -- For Rabin-Karp search
1854 k = 2891336453 :: Word32
1857 addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1859 deltaCksum cksum old new =
1860 addCksum cksum new - coeff * fromIntegral (fromEnum old)
1862 -- XXX shall we use a random starting hash or 1 instead of 0?
1863 -- XXX Need to keep this cached across fold calls in foldMany
1864 -- XXX We may need refold to inject the cached state instead of
1865 -- initializing the state every time.
1866 -- XXX Allocation of ring buffer should also be done once
1867 patHash = Array.foldl' addCksum 0 patArr
1869 step (SplitOnSeqEmpty s) x = do
1872 Partial s1 -> return $ Partial $ SplitOnSeqEmpty s1
1873 Done b -> return $ Done b
1874 step (SplitOnSeqSingle s pat) x = do
1879 Partial s1 -> return $ Partial $ SplitOnSeqSingle s1 pat
1880 Done b -> return $ Done b
1881 else Done <$> fextract s
1882 step (SplitOnSeqWord s idx wrd) x = do
1883 let wrd1 = addToWord wrd x
1886 if wrd1 .&. wordMask == wordPat
1887 then Done <$> fextract s
1888 else return $ Partial $ SplitOnSeqWordLoop s wrd1
1889 else return $ Partial $ SplitOnSeqWord s (idx + 1) wrd1
1890 step (SplitOnSeqWordLoop s wrd) x = do
1891 let wrd1 = addToWord wrd x
1892 old = (wordMask .&. wrd)
1893 `shiftR` (elemBits * (patLen - 1))
1894 res <- fstep s (toEnum $ fromIntegral old)
1897 | wrd1 .&. wordMask == wordPat ->
1898 Done <$> fextract s1
1900 return $ Partial $ SplitOnSeqWordLoop s1 wrd1
1901 Done b -> return $ Done b
1902 step (SplitOnSeqKR s idx rb rh) x = do
1903 rh1 <- liftIO $ Ring.unsafeInsert rb rh x
1906 let fld = Ring.unsafeFoldRing (Ring.ringBound rb)
1907 let !ringHash = fld addCksum 0 rb
1908 if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
1909 then Done <$> fextract s
1910 else return $ Partial $ SplitOnSeqKRLoop s ringHash rb rh1
1911 else return $ Partial $ SplitOnSeqKR s (idx + 1) rb rh1
1912 step (SplitOnSeqKRLoop s cksum rb rh) x = do
1913 old <- liftIO $ peek rh
1917 rh1 <- liftIO $ Ring.unsafeInsert rb rh x
1918 let ringHash = deltaCksum cksum old x
1919 if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
1920 then Done <$> fextract s1
1921 else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
1922 Done b -> return $ Done b
1924 -- XXX extract should return backtrack count as well. If the fold
1925 -- terminates early inside extract, we may still have buffered data
1926 -- remaining which will be lost if we do not communicate that to the
1929 let consumeWord s n wrd = do
1933 let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
1934 r <- fstep s (toEnum $ fromIntegral old)
1936 Partial s1 -> consumeWord s1 (n - 1) wrd
1939 let consumeRing s n rb rh =
1943 old <- liftIO $ peek rh
1944 let rh1 = Ring.advance rb rh
1947 Partial s1 -> consumeRing s1 (n - 1) rb rh1
1951 SplitOnSeqEmpty s -> fextract s
1952 SplitOnSeqSingle s _ -> fextract s
1953 SplitOnSeqWord s idx wrd -> consumeWord s idx wrd
1954 SplitOnSeqWordLoop s wrd -> consumeWord s patLen wrd
1955 SplitOnSeqKR s idx rb _ -> consumeRing s idx rb (Ring.startOf rb)
1956 SplitOnSeqKRLoop s _ rb rh -> consumeRing s patLen rb rh
1958------------------------------------------------------------------------------
1960------------------------------------------------------------------------------
1962-- | Distribute one copy of the stream to each fold and zip the results.
1965-- |-------Fold m a b--------|
1966-- ---stream m a---| |---m (b,c)
1967-- |-------Fold m a c--------|
1969-- >>> Stream.fold (Fold.tee Fold.sum Fold.length) (Stream.enumerateFromTo 1.0 100.0)
1972-- > tee = teeWith (,)
1976tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1979-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1980-- number of consumers? For polymorphic case a vector could be helpful. For
1981-- Storables we can use arrays. Will need separate APIs for those.
1983-- | Distribute one copy of the stream to each fold and collect the results in
1988-- |-------Fold m a b--------|
1989-- ---stream m a---| |---m [b]
1990-- |-------Fold m a b--------|
1995-- >>> Stream.fold (Fold.distribute [Fold.sum, Fold.length]) (Stream.enumerateFromTo 1 5)
1998-- >>> distribute = Prelude.foldr (Fold.teeWith (:)) (Fold.fromPure [])
2000-- This is the consumer side dual of the producer side 'sequence' operation.
2002-- Stops when all the folds stop.
2005{-# INLINE distribute #-}
2006distribute :: Monad m => [Fold m a b] -> Fold m a [b]
2007distribute = Prelude.foldr (teeWith (:)) (fromPure [])
2009------------------------------------------------------------------------------
2011------------------------------------------------------------------------------
2013{-# INLINE partitionByMUsing #-}
2014partitionByMUsing :: Monad m =>
2015 ( (x -> y -> (x, y))
2016 -> Fold m (Either b c) x
2017 -> Fold m (Either b c) y
2018 -> Fold m (Either b c) (x, y)
2020 -> (a -> m (Either b c))
2024partitionByMUsing t f fld1 fld2 =
2025 let l = lmap (fromLeft undefined) fld1 -- :: Fold m (Either b c) x
2026 r = lmap (fromRight undefined) fld2 -- :: Fold m (Either b c) y
2027 in lmapM f (t (,) (filter isLeft l) (filter isRight r))
2029-- | Partition the input over two folds using an 'Either' partitioning
2034-- |-------Fold b x--------|
2035-- -----stream m a --> (Either b c)----| |----(x,y)
2036-- |-------Fold c y--------|
2039-- Send input to either fold randomly:
2042-- > import System.Random (randomIO)
2043-- > randomly a = randomIO >>= \\x -> return $ if x then Left a else Right a
2044-- > Stream.fold (Fold.partitionByM randomly Fold.length Fold.length) (Stream.enumerateFromTo 1 100)
2048-- Send input to the two folds in a proportion of 2:1:
2051-- proportionately m n = do
2052-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
2054-- r <- readIORef ref
2055-- writeIORef ref $ tail r
2056-- return $ Prelude.head r a
2061-- f <- proportionately 2 1
2062-- r <- Stream.fold (Fold.partitionByM f Fold.length Fold.length) (Stream.enumerateFromTo (1 :: Int) 100)
2070-- This is the consumer side dual of the producer side 'mergeBy' operation.
2072-- When one fold is done, any input meant for it is ignored until the other
2073-- fold is also done.
2075-- Stops when both the folds stop.
2077-- /See also: 'partitionByFstM' and 'partitionByMinM'./
2080{-# INLINE partitionByM #-}
2081partitionByM :: Monad m
2082 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2083partitionByM = partitionByMUsing teeWith
2085-- | Similar to 'partitionByM' but terminates when the first fold terminates.
2088{-# INLINE partitionByFstM #-}
2089partitionByFstM :: Monad m
2090 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2091partitionByFstM = partitionByMUsing teeWithFst
2093-- | Similar to 'partitionByM' but terminates when any fold terminates.
2096{-# INLINE partitionByMinM #-}
2097partitionByMinM :: Monad m =>
2098 (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2099partitionByMinM = partitionByMUsing teeWithMin
2101-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
2102-- makes the signature clearer as to which case belongs to which fold.
2103-- XXX need to check the performance in both cases.
2104-- | Same as 'partitionByM' but with a pure partition function.
2106-- Count even and odd numbers in a stream:
2109-- let f = Fold.partitionBy (\n -> if even n then Left n else Right n)
2110-- (fmap (("Even " ++) . show) Fold.length)
2111-- (fmap (("Odd " ++) . show) Fold.length)
2112-- in Stream.fold f (Stream.enumerateFromTo 1 100)
2114-- ("Even 50","Odd 50")
2117{-# INLINE partitionBy #-}
2118partitionBy :: Monad m
2119 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2120partitionBy f = partitionByM (return . f)
2122-- | Compose two folds such that the combined fold accepts a stream of 'Either'
2123-- and routes the 'Left' values to the first fold and 'Right' values to the
2126-- > partition = partitionBy id
2129{-# INLINE partition #-}
2131 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
2132partition = partitionBy id
2135-- | Send one item to each fold in a round-robin fashion. This is the consumer
2136-- side dual of producer side 'mergeN' operation.
2138-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
2139-- partitionN fs = Fold step begin done
2142------------------------------------------------------------------------------
2143-- demux: in a key value stream fold each key sub-stream with a different fold
2144------------------------------------------------------------------------------
2146-- TODO Demultiplex an input element into a number of typed variants. We want
2147-- to statically restrict the target values within a set of predefined types,
2148-- an enumeration of a GADT.
2150-- This is the consumer side dual of the producer side 'mux' operation (XXX to
2153-- XXX If we use Refold in it, it can perhaps fuse/be more efficient. For
2154-- example we can store just the result rather than storing the whole fold in
2157-- Note: There are separate functions to determine Key and Fold from the input
2158-- because key is to be determined on each input whereas fold is to be
2159-- determined only once for a key.
2161-- | In a key value stream, fold values corresponding to each key with a key
2162-- specific fold. The fold returns the fold result as the second component of
2163-- the output tuple whenever a fold terminates. The first component of the
2164-- tuple is a container of in-progress folds. If a fold terminates, another
2165-- instance of the fold is started upon receiving an input with that key.
2167-- This can be used to scan a stream and collect the results from the scan
2172{-# INLINE demuxScanWith #-}
2173demuxScanWith :: (Monad m, IsMap f, Traversable f) =>
2175 -> (a -> m (Fold m a b))
2176 -> Fold m a (m (f b), Maybe (Key f, b))
2177demuxScanWith getKey getFold = fmap extract $ foldlM' step initial
2181 initial = return $ Tuple' IsMap.mapEmpty Nothing
2183 {-# INLINE runFold #-}
2184 runFold kv (Fold step1 initial1 extract1) (k, a) = do
2192 let fld = Fold step1 (return res1) extract1
2193 in Tuple' (IsMap.mapInsert k fld kv) Nothing
2194 Done b -> Tuple' (IsMap.mapDelete k kv) (Just (k, b))
2195 Done b -> return $ Tuple' kv (Just (k, b))
2197 step (Tuple' kv _) a = do
2199 case IsMap.mapLookup k kv of
2202 runFold kv fld (k, a)
2203 Just f -> runFold kv f (k, a)
2205 extract (Tuple' kv x) = (Prelude.mapM f kv, x)
2215-- | This is specialized version of 'demuxScanWith' that uses mutable cells as
2216-- fold accumulators for better performance.
2218-- >>> demuxScanMutWith = Fold.demuxScanWith
2220{-# INLINE demuxScanMutWith #-}
2221demuxScanMutWith :: (MonadIO m, IsMap f, Traversable f) =>
2223 -> (a -> m (Fold m a b))
2224 -> Fold m a (m (f b), Maybe (Key f, b))
2225demuxScanMutWith getKey getFold = fmap extract $ foldlM' step initial
2229 initial = return $ Tuple' IsMap.mapEmpty Nothing
2231 {-# INLINE initFold #-}
2232 initFold kv (Fold step1 initial1 extract1) (k, a) = do
2239 let fld = Fold step1 (return res1) extract1
2240 ref <- liftIO $ newIORef fld
2241 return $ Tuple' (IsMap.mapInsert k ref kv) Nothing
2242 Done b -> return $ Tuple' kv (Just (k, b))
2243 Done b -> return $ Tuple' kv (Just (k, b))
2245 {-# INLINE runFold #-}
2246 runFold kv ref (Fold step1 initial1 extract1) (k, a) = do
2253 let fld = Fold step1 (return res1) extract1
2254 liftIO $ writeIORef ref fld
2255 return $ Tuple' kv Nothing
2257 let kv1 = IsMap.mapDelete k kv
2258 in return $ Tuple' kv1 (Just (k, b))
2259 Done _ -> error "demuxScanMutWith: unreachable"
2261 step (Tuple' kv _) a = do
2263 case IsMap.mapLookup k kv of
2266 initFold kv f (k, a)
2268 f <- liftIO $ readIORef ref
2269 runFold kv ref f (k, a)
2271 extract (Tuple' kv x) = (Prelude.mapM f kv, x)
2276 (Fold _ i e) <- liftIO $ readIORef ref
2282-- | This collects all the results of 'demuxScanWith' in a container.
2284{-# INLINE demuxWith #-}
2285demuxWith :: (Monad m, IsMap f, Traversable f) =>
2286 (a -> Key f) -> (a -> m (Fold m a b)) -> Fold m a (f b)
2287demuxWith getKey getFold =
2289 classifier = demuxScanWith getKey getFold
2290 getMap Nothing = pure IsMap.mapEmpty
2291 getMap (Just action) = action
2293 teeWith IsMap.mapUnion
2294 (rmapM getMap $ lmap fst last)
2295 (lmap snd $ catMaybes toMap)
2296 in postscan classifier aggregator
2298-- | Same as 'demuxWith' but uses 'demuxScanMutWith' for better performance.
2300-- >>> demuxMutWith = Fold.demuxWith
2302{-# INLINE demuxMutWith #-}
2303demuxMutWith :: (MonadIO m, IsMap f, Traversable f) =>
2304 (a -> Key f) -> (a -> m (Fold m a b)) -> Fold m a (f b)
2305demuxMutWith getKey getFold =
2307 classifier = demuxScanMutWith getKey getFold
2308 getMap Nothing = pure IsMap.mapEmpty
2309 getMap (Just action) = action
2311 teeWith IsMap.mapUnion
2312 (rmapM getMap $ lmap fst last)
2313 (lmap snd $ catMaybes toMap)
2314 in postscan classifier aggregator
2316-- | Fold a stream of key value pairs using a function that maps keys to folds.
2318-- >>> import Data.Map (Map)
2320-- let f "SUM" = return Fold.sum
2321-- f _ = return Fold.product
2322-- input = Stream.fromList [("SUM",1),("PRODUCT",2),("SUM",3),("PRODUCT",4)]
2323-- in Stream.fold (Fold.demux f) input :: IO (Map String Int)
2325-- fromList [("PRODUCT",8),("SUM",4)]
2327-- >>> demux f = Fold.demuxWith fst (Fold.lmap snd . f)
2331demux :: (Monad m, IsMap f, Traversable f) =>
2332 (Key f -> m (Fold m a b)) -> Fold m (Key f, a) (f b)
2333demux f = demuxWith fst (\(k, _) -> fmap (lmap snd) (f k))
2335------------------------------------------------------------------------------
2336-- Classify: Like demux but uses the same fold for all keys.
2337------------------------------------------------------------------------------
2339-- XXX Change these to make the behavior similar to demux* variants. We can
2340-- implement this using classifyScanManyWith. Maintain a set of done folds in
2341-- the underlying monad, and when initial is called look it up, if the fold is
2342-- done then initial would set a flag in the state to ignore the input or
2345-- | Folds the values for each key using the supplied fold. When scanning, as
2346-- soon as the fold is complete, its result is available in the second
2347-- component of the tuple. The first component of the tuple is a snapshot of
2348-- the in-progress folds.
2350-- Once the fold for a key is done, any future values of the key are ignored.
2352-- >>> classifyScanWith f fld = Fold.demuxScanWith f (const fld)
2354{-# INLINE classifyScanWith #-}
2355classifyScanWith :: (Monad m, IsMap f, Traversable f, Ord (Key f)) =>
2356 -- Note: we need to return the Map itself to display the in-progress values
2357 -- e.g. to implement top. We could possibly create a separate abstraction
2358 -- for that use case. We return an action because we want it to be lazy so
2359 -- that the downstream consumers can choose to process or discard it.
2360 (a -> Key f) -> Fold m a b -> Fold m a (m (f b), Maybe (Key f, b))
2361classifyScanWith f (Fold step1 initial1 extract1) =
2362 fmap extract $ foldlM' step initial
2366 initial = return $ Tuple3' IsMap.mapEmpty Set.empty Nothing
2368 {-# INLINE initFold #-}
2369 initFold kv set k a = do
2377 Tuple3' (IsMap.mapInsert k s1 kv) set Nothing
2379 Tuple3' kv set (Just (k, b))
2380 Done b -> return (Tuple3' kv (Set.insert k set) (Just (k, b)))
2382 step (Tuple3' kv set _) a = do
2384 case IsMap.mapLookup k kv of
2387 then return (Tuple3' kv set Nothing)
2388 else initFold kv set k a
2394 Tuple3' (IsMap.mapInsert k s1 kv) set Nothing
2396 let kv1 = IsMap.mapDelete k kv
2397 in Tuple3' kv1 (Set.insert k set) (Just (k, b))
2399 extract (Tuple3' kv _ x) = (Prelude.mapM extract1 kv, x)
2401-- XXX we can use a Prim IORef if we can constrain the state "s" to be Prim
2403-- The code is almost the same as classifyScanWith except the IORef operations.
2405-- | Same as classifyScanWith except that it uses mutable IORef cells in the
2406-- Map providing better performance. Be aware that if this is used as a scan,
2407-- the values in the intermediate Maps would be mutable.
2409-- >>> classifyScanMutWith = classifyScanWith
2410-- >>> classifyScanMutWith f fld = demuxScanMutWith f (const fld)
2412{-# INLINE classifyScanMutWith #-}
2413classifyScanMutWith :: (MonadIO m, IsMap f, Traversable f, Ord (Key f)) =>
2414 (a -> Key f) -> Fold m a b -> Fold m a (m (f b), Maybe (Key f, b))
2415classifyScanMutWith f (Fold step1 initial1 extract1) =
2416 fmap extract $ foldlM' step initial
2420 initial = return $ Tuple3' IsMap.mapEmpty Set.empty Nothing
2422 {-# INLINE initFold #-}
2423 initFold kv set k a = do
2430 ref <- liftIO $ newIORef s1
2431 return $ Tuple3' (IsMap.mapInsert k ref kv) set Nothing
2433 return $ Tuple3' kv set (Just (k, b))
2434 Done b -> return (Tuple3' kv (Set.insert k set) (Just (k, b)))
2436 step (Tuple3' kv set _) a = do
2438 case IsMap.mapLookup k kv of
2441 then return (Tuple3' kv set Nothing)
2442 else initFold kv set k a
2444 s <- liftIO $ readIORef ref
2448 liftIO $ writeIORef ref s1
2449 return $ Tuple3' kv set Nothing
2451 let kv1 = IsMap.mapDelete k kv
2453 $ Tuple3' kv1 (Set.insert k set) (Just (k, b))
2455 extract (Tuple3' kv _ x) =
2456 (Prelude.mapM (\ref -> liftIO (readIORef ref) >>= extract1) kv, x)
2458-- | Fold a key value stream to a key-value container. If the same key appears
2459-- multiple times, only the last value is retained.
2461toMap :: (Monad m, IsMap f) => Fold m (Key f, a) (f a)
2462toMap = foldl' (\kv (k, v) -> IsMap.mapInsert k v kv) IsMap.mapEmpty
2464-- | Split the input stream based on a key field and fold each split using the
2465-- given fold. Useful for map/reduce, bucketizing the input in different bins
2466-- or for generating histograms.
2468-- >>> import Data.Map.Strict (Map)
2470-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
2471-- classify = Fold.classifyWith fst (Fold.lmap snd Fold.toList)
2472-- in Stream.fold classify input :: IO (Map String [Double])
2474-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
2476-- Once the classifier fold terminates for a particular key any further inputs
2477-- in that bucket are ignored.
2479-- Space used is proportional to the number of keys seen till now and
2480-- monotonically increases because it stores whether a key has been seen or
2487{-# INLINE classifyWith #-}
2488classifyWith :: (Monad m, IsMap f, Traversable f, Ord (Key f)) =>
2489 (a -> Key f) -> Fold m a b -> Fold m a (f b)
2492 classifier = classifyScanWith f fld
2493 getMap Nothing = pure IsMap.mapEmpty
2494 getMap (Just action) = action
2496 teeWith IsMap.mapUnion
2497 (rmapM getMap $ lmap fst last)
2498 (lmap snd $ catMaybes toMap)
2499 in postscan classifier aggregator
2501-- | Same as 'classifyWith' but maybe faster because it uses mutable cells as
2502-- fold accumulators in the Map.
2504-- >>> classifyMutWith = Fold.classifyWith
2506{-# INLINE classifyMutWith #-}
2507classifyMutWith :: (MonadIO m, IsMap f, Traversable f, Ord (Key f)) =>
2508 (a -> Key f) -> Fold m a b -> Fold m a (f b)
2509classifyMutWith f fld =
2511 classifier = classifyScanMutWith f fld
2512 getMap Nothing = pure IsMap.mapEmpty
2513 getMap (Just action) = action
2515 teeWith IsMap.mapUnion
2516 (rmapM getMap $ lmap fst last)
2517 (lmap snd $ catMaybes toMap)
2518 in postscan classifier aggregator
2520-- | Given an input stream of key value pairs and a fold for values, fold all
2521-- the values belonging to each key. Useful for map/reduce, bucketizing the
2522-- input in different bins or for generating histograms.
2525-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
2526-- in Stream.fold (Fold.classify Fold.toList) input
2528-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
2532-- >>> classify = Fold.classifyWith fst . Fold.lmap snd
2535{-# INLINE classify #-}
2536classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
2537classify = classifyWith fst . lmap snd
2539-- | Determine the frequency of each element in the stream.
2541-- You can just collect the keys of the resulting map to get the unique
2542-- elements in the stream.
2544{-# INLINE frequency #-}
2545frequency :: (Monad m, Ord a) => Fold m a (Map a Int)
2546frequency = classifyWith id length
2548------------------------------------------------------------------------------
2550------------------------------------------------------------------------------
2552{-# INLINE unzipWithMUsing #-}
2553unzipWithMUsing :: Monad m =>
2554 ( (x -> y -> (x, y))
2557 -> Fold m (b, c) (x, y)
2563unzipWithMUsing t f fld1 fld2 =
2564 let f1 = lmap fst fld1 -- :: Fold m (b, c) b
2565 f2 = lmap snd fld2 -- :: Fold m (b, c) c
2566 in lmapM f (t (,) f1 f2)
2568-- | Like 'unzipWith' but with a monadic splitter function.
2570-- @unzipWithM k f1 f2 = lmapM k (unzip f1 f2)@
2573{-# INLINE unzipWithM #-}
2574unzipWithM :: Monad m
2575 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2576unzipWithM = unzipWithMUsing teeWith
2578-- | Similar to 'unzipWithM' but terminates when the first fold terminates.
2581{-# INLINE unzipWithFstM #-}
2582unzipWithFstM :: Monad m =>
2583 (a -> m (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2584unzipWithFstM = unzipWithMUsing teeWithFst
2586-- | Similar to 'unzipWithM' but terminates when any fold terminates.
2589{-# INLINE unzipWithMinM #-}
2590unzipWithMinM :: Monad m =>
2591 (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2592unzipWithMinM = unzipWithMUsing teeWithMin
2594-- | Split elements in the input stream into two parts using a pure splitter
2595-- function, direct each part to a different fold and zip the results.
2597-- @unzipWith f fld1 fld2 = Fold.lmap f (Fold.unzip fld1 fld2)@
2599-- This fold terminates when both the input folds terminate.
2602{-# INLINE unzipWith #-}
2604 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2605unzipWith f = unzipWithM (return . f)
2607-- | Send the elements of tuples in a stream of tuples through two different
2612-- |-------Fold m a x--------|
2613-- ---------stream of (a,b)--| |----m (x,y)
2614-- |-------Fold m b y--------|
2618-- > unzip = Fold.unzipWith id
2620-- This is the consumer side dual of the producer side 'zip' operation.
2624unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
2627------------------------------------------------------------------------------
2628-- Combining streams and folds - Zipping
2629------------------------------------------------------------------------------
2631-- XXX These can be implemented using the fold scan, using the stream as a
2633-- XXX Stream Skip state cannot be efficiently handled in folds but can be
2634-- handled in parsers using the Continue facility. See zipWithM in the Parser
2637-- | Zip a stream with the input of a fold using the supplied function.
2641{-# INLINE zipWithM #-}
2642zipWithM :: -- Monad m =>
2643 (a -> b -> m c) -> t m a -> Fold m c x -> Fold m b x
2646-- | Zip a stream with the input of a fold.
2651zip :: Monad m => t m a -> Fold m (a, b) x -> Fold m b x
2652zip = zipWithM (curry return)
2654-- | Pair each element of a fold input with its index, starting from index 0.
2656{-# INLINE indexingWith #-}
2657indexingWith :: Monad m => Int -> (Int -> Int) -> Fold m a (Maybe (Int, a))
2658indexingWith i f = fmap toMaybe $ foldl' step initial
2664 step Nothing' a = Just' (i, a)
2665 step (Just' (n, _)) a = Just' (f n, a)
2667{-# INLINE indexing #-}
2668indexing :: Monad m => Fold m a (Maybe (Int, a))
2669indexing = indexingWith 0 (+ 1)
2671{-# INLINE indexingRev #-}
2672indexingRev :: Monad m => Int -> Fold m a (Maybe (Int, a))
2673indexingRev n = indexingWith n (subtract 1)
2675-- | Pair each element of a fold input with its index, starting from index 0.
2677{-# INLINE indexed #-}
2678indexed :: Monad m => Fold m (Int, a) b -> Fold m a b
2679indexed = scanMaybe indexing
2681-- | Change the predicate function of a Fold from @a -> b@ to accept an
2682-- additional state input @(s, a) -> b@. Convenient to filter with an
2683-- addiitonal index or time input.
2686-- filterWithIndex = with indexed filter
2687-- filterWithAbsTime = with timestamped filter
2688-- filterWithRelTime = with timeIndexed filter
2694 (Fold m (s, a) b -> Fold m a b)
2695 -> (((s, a) -> c) -> Fold m (s, a) b -> Fold m (s, a) b)
2696 -> (((s, a) -> c) -> Fold m a b -> Fold m a b)
2697with f comb g = f . comb g . lmap snd
2699-- XXX Implement as a filter
2700-- sampleFromthen :: Monad m => Int -> Int -> Fold m a (Maybe a)
2702-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
2703-- then every element at strides of @stride@.
2705{-# INLINE sampleFromthen #-}
2706sampleFromthen :: Monad m => Int -> Int -> Fold m a b -> Fold m a b
2707sampleFromthen offset size =
2708 with indexed filter (\(i, _) -> (i + offset) `mod` size == 0)
2710------------------------------------------------------------------------------
2712------------------------------------------------------------------------------
2714-- | @concatSequence f t@ applies folds from stream @t@ sequentially and
2715-- collects the results using the fold @f@.
2719{-# INLINE concatSequence #-}
2722 Fold m b c -> t (Fold m a b) -> Fold m a c
2723concatSequence _f _p = undefined
2725-- | Group the input stream into groups of elements between @low@ and @high@.
2726-- Collection starts in chunks of @low@ and then keeps doubling until we reach
2727-- @high@. Each chunk is folded using the provided fold function.
2729-- This could be useful, for example, when we are folding a stream of unknown
2730-- size to a stream of arrays and we want to minimize the number of
2733-- NOTE: this would be an application of "many" using a terminating fold.
2737{-# INLINE chunksBetween #-}
2738chunksBetween :: -- Monad m =>
2739 Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
2740chunksBetween _low _high _f1 _f2 = undefined
2742-- | A fold that buffers its input to a pure stream.
2744-- /Warning!/ working on large streams accumulated as buffers in memory could
2745-- be very inefficient, consider using "Streamly.Data.Array" instead.
2747-- >>> toStream = fmap Stream.fromStreamK Fold.toStreamK
2750{-# INLINE toStream #-}
2751toStream :: Monad m => Fold m a (Stream n a)
2752toStream = fmap Stream.fromStreamK toStreamK
2754-- This is more efficient than 'toStream'. toStream is exactly the same as
2755-- reversing the stream after toStreamRev.
2757-- | Buffers the input stream to a pure stream in the reverse order of the
2760-- >>> toStreamRev = fmap Stream.fromStreamK Fold.toStreamKRev
2762-- /Warning!/ working on large streams accumulated as buffers in memory could
2763-- be very inefficient, consider using "Streamly.Data.Array" instead.
2767-- xn : ... : x2 : x1 : []
2768{-# INLINE toStreamRev #-}
2769toStreamRev :: Monad m => Fold m a (Stream n a)
2770toStreamRev = fmap Stream.fromStreamK toStreamKRev
2772-- XXX This does not fuse. It contains a recursive step function. We will need
2773-- a Skip input constructor in the fold type to make it fuse.
2775-- | Unfold and flatten the input stream of a fold.
2778-- Stream.fold (unfoldMany u f) = Stream.fold f . Stream.unfoldMany u
2782{-# INLINE unfoldMany #-}
2783unfoldMany :: Monad m => Unfold m a b -> Fold m b c -> Fold m a c
2784unfoldMany (Unfold ustep inject) (Fold fstep initial extract) =
2785 Fold consume initial extract
2789 {-# INLINE produce #-}
2793 StreamD.Yield b us1 -> do
2796 Partial fs1 -> produce fs1 us1
2797 -- XXX What to do with the remaining stream?
2798 Done c -> return $ Done c
2799 StreamD.Skip us1 -> produce fs us1
2800 StreamD.Stop -> return $ Partial fs
2802 {-# INLINE_LATE consume #-}
2803 consume s a = inject a >>= produce s
2806topBy :: (MonadIO m, Unboxed a) =>
2807 (a -> a -> Ordering)
2809 -> Fold m a (MA.Array a)
2810topBy cmp n = Fold step initial extract
2815 arr <- MA.newPinned n
2817 then return $ Done arr
2818 else return $ Partial (arr, 0)
2823 arr' <- MA.snoc arr x
2825 return $ Partial (arr', i + 1)
2827 x1 <- MA.getIndexUnsafe (i - 1) arr
2830 MA.putIndexUnsafe (i - 1) x arr
2832 return $ Partial (arr, i)
2833 _ -> return $ Partial (arr, i)
2835 extract = return . fst
2837-- | Fold the input stream to top n elements.
2839-- >>> stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]
2840-- >>> Stream.fold (Fold.top 3) stream >>= MA.toList
2845top :: (MonadIO m, Unboxed a, Ord a) => Int -> Fold m a (MA.Array a)
2846top = topBy $ flip compare
2848-- | Fold the input stream to bottom n elements.
2850-- >>> stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]
2851-- >>> Stream.fold (Fold.bottom 3) stream >>= MA.toList
2855{-# INLINE bottom #-}
2856bottom :: (MonadIO m, Unboxed a, Ord a) => Int -> Fold m a (MA.Array a)
2857bottom = topBy compare
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