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
38 -- *** Semigroups and Monoids
57 -- *** Saturating Reducers
58 -- | 'product' terminates if it becomes 0. Other folds can theoretically
59 -- saturate on bounded types, and therefore terminate, however, they will
60 -- run forever on unbounded types like Integer/Double.
69 -- | Avoid using these folds in scalable or performance critical
70 -- applications, they buffer all the input in GC memory which can be
71 -- detrimental to performance if the input is large.
78 -- ** Terminating Folds
103 -- ** Transforming the Monad
107 -- ** Mapping on output
110 -- ** Mapping on Input
124 -- ** Mapping Filters
130 -- ** Scanning Filters
135 -- | Insertion adds more elements to the stream.
164 -- , splitIn -- sessionN
166 -- ** Parallel Distribution
175 -- ** Parallel Alternative
187 -- | Direct values in the input stream to different folds using an n-ary
189 , demux -- XXX rename this to demux_
191 , demuxDefault -- XXX rename this to demux
197 -- | In an input stream of key value pairs fold values for different keys
198 -- in individual output buckets using the given fold.
206 -- These two can be expressed using lmap/lmapM and unzip
226 -- * Running Partially
240import Control.Monad (void)
241import Data.Bifunctor (first)
242import Data.Functor.Identity (Identity(..))
243import Data.Int (Int64)
244import Data.Map.Strict (Map)
245import Data.Maybe (isJust, fromJust)
246#if __GLASGOW_HASKELL__ < 804
247import Data.Semigroup (Semigroup((<>)))
249import Streamly.Internal.Data.Either.Strict
250 (Either'(..), fromLeft', fromRight', isLeft', isRight')
251import Streamly.Internal.Data.Pipe.Type (Pipe (..), PipeState(..))
252import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
253import Streamly.Internal.Data.Stream.Serial (SerialT)
255import qualified Data.Map.Strict as Map
256import qualified Streamly.Internal.Data.Pipe.Type as Pipe
257import qualified Streamly.Internal.Data.Stream.IsStream.Enumeration as Stream
258import qualified Streamly.Internal.Data.Stream.StreamK as K
259import qualified Prelude
262 ( filter, foldl1, drop, dropWhile, take, takeWhile, zipWith
263 , foldl, foldr, map, mapM_, sequence, all, any, sum, product, elem
264 , notElem, maximum, minimum, head, last, tail, length, null
265 , reverse, iterate, init, and, or, lookup, (!!)
266 , scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip
267 , span, splitAt, break, mapM, zip)
268import Streamly.Internal.Data.Fold.Type
272-- >>> import Prelude hiding (break, map, span, splitAt)
273-- >>> import qualified Streamly.Prelude as Stream
274-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream (parse, foldMany)
275-- >>> import qualified Streamly.Data.Fold as Fold
276-- >>> import qualified Streamly.Internal.Data.Fold as Fold
277-- >>> import qualified Streamly.Internal.Data.Parser as Parser
278-- >>> import Data.IORef (newIORef, readIORef, writeIORef)
280------------------------------------------------------------------------------
282------------------------------------------------------------------------------
284-- | Change the underlying monad of a fold
287hoist :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b
288hoist f (Fold step initial extract) =
289 Fold (\x a -> f $ step x a) (f initial) (f . extract)
291-- | Adapt a pure fold to any monad
293-- > generally = Fold.hoist (return . runIdentity)
296generally :: Monad m => Fold Identity a b -> Fold m a b
297generally = hoist (return . runIdentity)
299------------------------------------------------------------------------------
300-- Transformations on fold inputs
301------------------------------------------------------------------------------
303-- | Flatten the monadic output of a fold to pure output.
306{-# DEPRECATED sequence "Use \"rmapM id\" instead" #-}
307{-# INLINE sequence #-}
308sequence :: Monad m => Fold m a (m b) -> Fold m a b
311-- | Map a monadic function on the output of a fold.
314{-# DEPRECATED mapM "Use rmapM instead" #-}
316mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
319-- | @mapMaybe f fold@ maps a 'Maybe' returning function @f@ on the input of
320-- the fold, filters out 'Nothing' elements, and return the values extracted
323-- >>> f x = if even x then Just x else Nothing
324-- >>> fld = Fold.mapMaybe f Fold.toList
325-- >>> Stream.fold fld (Stream.enumerateFromTo 1 10)
329{-# INLINE mapMaybe #-}
330mapMaybe :: (Monad m) => (a -> Maybe b) -> Fold m b r -> Fold m a r
331mapMaybe f = lmap f . filter isJust . lmap fromJust
333------------------------------------------------------------------------------
334-- Transformations on fold inputs
335------------------------------------------------------------------------------
339-- | Apply a transformation on a 'Fold' using a 'Pipe'.
342{-# INLINE transform #-}
343transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
344transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
345 Fold step initial extract
349 initial = first (Tuple' pinitial) <$> finitial
351 step (Tuple' ps fs) x = do
358 go acc (Pipe.Yield b (Consume ps')) = do
362 Partial s -> Partial $ Tuple' ps' s
364 go acc (Pipe.Yield b (Produce ps')) = do
369 Done b2 -> return $ Done b2
370 go acc (Pipe.Continue (Consume ps')) =
371 return $ Partial $ Tuple' ps' acc
372 go acc (Pipe.Continue (Produce ps')) = do
376 extract (Tuple' _ fs) = fextract fs
378------------------------------------------------------------------------------
380------------------------------------------------------------------------------
382------------------------------------------------------------------------------
384------------------------------------------------------------------------------
387-- > drainBy f = lmapM f drain
388-- > drainBy = Fold.foldMapM (void . f)
390-- Drain all input after passing it through a monadic function. This is the
391-- dual of mapM_ on stream producers.
393-- See also: 'Streamly.Prelude.mapM_'
396{-# INLINABLE drainBy #-}
397drainBy :: Monad m => (a -> m b) -> Fold m a ()
398drainBy f = lmapM f drain
403{-# INLINABLE drainBy2 #-}
404drainBy2 :: Monad m => (a -> m b) -> Fold2 m c a ()
405drainBy2 f = Fold2 (const (void . f)) (\_ -> return ()) return
407-- | Extract the last element of the input stream, if any.
409-- > last = fmap getLast $ Fold.foldMap (Last . Just)
412{-# INLINABLE last #-}
413last :: Monad m => Fold m a (Maybe a)
414last = foldl1' (\_ x -> x)
416------------------------------------------------------------------------------
418------------------------------------------------------------------------------
420-- | Like 'length', except with a more general 'Num' return value
422-- > genericLength = fmap getSum $ foldMap (Sum . const 1)
425{-# INLINE genericLength #-}
426genericLength :: (Monad m, Num b) => Fold m a b
427genericLength = foldl' (\n _ -> n + 1) 0
429-- | Determine the length of the input stream.
431-- > length = fmap getSum $ Fold.foldMap (Sum . const 1)
435length :: Monad m => Fold m a Int
436length = genericLength
438-- | Determine the sum of all elements of a stream of numbers. Returns additive
439-- identity (@0@) when the stream is empty. Note that this is not numerically
440-- stable for floating point numbers.
442-- > sum = fmap getSum $ Fold.foldMap Sum
446sum :: (Monad m, Num a) => Fold m a a
449-- | Determine the product of all elements of a stream of numbers. Returns
450-- multiplicative identity (@1@) when the stream is empty. The fold terminates
451-- when it encounters (@0@) in its input.
453-- Compare with @Fold.foldMap Product@.
456-- /Since 0.8.0 (Added 'Eq' constraint)/
457{-# INLINE product #-}
458product :: (Monad m, Num a, Eq a) => Fold m a a
459product = mkFold_ step (Partial 1)
468------------------------------------------------------------------------------
470------------------------------------------------------------------------------
472-- | Determine the maximum element in a stream using the supplied comparison
476{-# INLINE maximumBy #-}
477maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
478maximumBy cmp = foldl1' max'
489-- maximum = Fold.maximumBy compare
492-- Determine the maximum element in a stream.
494-- Compare with @Fold.foldMap Max@.
497{-# INLINE maximum #-}
498maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
501-- | Computes the minimum element with respect to the given comparison function
504{-# INLINE minimumBy #-}
505minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
506minimumBy cmp = foldl1' min'
515-- | Determine the minimum element in a stream using the supplied comparison
519-- minimum = 'minimumBy' compare
522-- Compare with @Fold.foldMap Min@.
525{-# INLINE minimum #-}
526minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
529------------------------------------------------------------------------------
530-- To Summary (Statistical)
531------------------------------------------------------------------------------
533-- | Compute a numerically stable arithmetic mean of all elements in the input
537{-# INLINABLE mean #-}
538mean :: (Monad m, Fractional a) => Fold m a a
539mean = fmap done $ foldl' step begin
545 step (Tuple' x n) y =
547 in Tuple' (x + (y - x) / n1) n1
549 done (Tuple' x _) = x
551-- | Compute a numerically stable (population) variance over all elements in
555{-# INLINABLE variance #-}
556variance :: (Monad m, Fractional a) => Fold m a a
557variance = fmap done $ foldl' step begin
561 begin = Tuple3' 0 0 0
563 step (Tuple3' n mean_ m2) x = Tuple3' n' mean' m2'
568 mean' = (n * mean_ + x) / (n + 1)
570 m2' = m2 + delta * delta * n / (n + 1)
572 done (Tuple3' n _ m2) = m2 / n
574-- | Compute a numerically stable (population) standard deviation over all
575-- elements in the input stream.
578{-# INLINABLE stdDev #-}
579stdDev :: (Monad m, Floating a) => Fold m a a
580stdDev = sqrt <$> variance
582-- | Compute an 'Int' sized polynomial rolling hash
584-- > H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0
586-- Where @c1@, @c2@, @cn@ are the elements in the input stream and @k@ is a
589-- This hash is often used in Rabin-Karp string search algorithm.
591-- See https://en.wikipedia.org/wiki/Rolling_hash
594{-# INLINABLE rollingHashWithSalt #-}
595rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64
596rollingHashWithSalt = foldl' step
600 k = 2891336453 :: Int64
602 step cksum a = cksum * k + fromIntegral (fromEnum a)
604-- | A default salt used in the implementation of 'rollingHash'.
605{-# INLINE defaultSalt #-}
607defaultSalt = -2578643520546668380
609-- | Compute an 'Int' sized polynomial rolling hash of a stream.
611-- > rollingHash = Fold.rollingHashWithSalt defaultSalt
614{-# INLINABLE rollingHash #-}
615rollingHash :: (Monad m, Enum a) => Fold m a Int64
616rollingHash = rollingHashWithSalt defaultSalt
618-- | Compute an 'Int' sized polynomial rolling hash of the first n elements of
621-- > rollingHashFirstN = Fold.take n Fold.rollingHash
624{-# INLINABLE rollingHashFirstN #-}
625rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64
626rollingHashFirstN n = take n rollingHash
628------------------------------------------------------------------------------
629-- Monoidal left folds
630------------------------------------------------------------------------------
632-- | Append the elements of an input stream to a provided starting value.
634-- >>> Stream.fold (Fold.sconcat 10) (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
638-- sconcat = Fold.foldl' (<>)
642{-# INLINE sconcat #-}
643sconcat :: (Monad m, Semigroup a) => a -> Fold m a a
646-- | Fold an input stream consisting of monoidal elements using 'mappend'
649-- >>> Stream.fold Fold.mconcat (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
652-- > mconcat = Fold.sconcat mempty
655{-# INLINE mconcat #-}
658#if !MIN_VERSION_base(4,11,0)
661 , Monoid a) => Fold m a a
662mconcat = sconcat mempty
665-- > foldMap f = Fold.lmap f Fold.mconcat
667-- Make a fold from a pure function that folds the output of the function
668-- using 'mappend' and 'mempty'.
670-- >>> Stream.fold (Fold.foldMap Data.Monoid.Sum) $ Stream.enumerateFromTo 1 10
674{-# INLINABLE foldMap #-}
675foldMap :: (Monad m, Monoid b
676#if !MIN_VERSION_base(4,11,0)
679 ) => (a -> b) -> Fold m a b
680foldMap f = lmap f mconcat
683-- > foldMapM f = Fold.lmapM f Fold.mconcat
685-- Make a fold from a monadic function that folds the output of the function
686-- using 'mappend' and 'mempty'.
688-- >>> Stream.fold (Fold.foldMapM (return . Data.Monoid.Sum)) $ Stream.enumerateFromTo 1 10
692{-# INLINABLE foldMapM #-}
693foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
694foldMapM act = foldlM' step (pure mempty)
700 return $! mappend m m'
702------------------------------------------------------------------------------
704------------------------------------------------------------------------------
707-- This is more efficient than 'Streamly.Internal.Data.Fold.toList'. toList is
708-- exactly the same as reversing the list after 'toListRev'.
710-- | Buffers the input stream to a list in the reverse order of the input.
712-- > toListRev = Fold.foldl' (flip (:)) []
714-- /Warning!/ working on large lists accumulated as buffers in memory could be
715-- very inefficient, consider using "Streamly.Array" instead.
719-- xn : ... : x2 : x1 : []
720{-# INLINABLE toListRev #-}
721toListRev :: Monad m => Fold m a [a]
722toListRev = foldl' (flip (:)) []
724------------------------------------------------------------------------------
726------------------------------------------------------------------------------
728-- | A fold that drains the first n elements of its input, running the effects
729-- and discarding the results.
731-- > drainN n = Fold.take n Fold.drain
734{-# INLINABLE drainN #-}
735drainN :: Monad m => Int -> Fold m a ()
736drainN n = take n drain
738------------------------------------------------------------------------------
740------------------------------------------------------------------------------
742-- | Like 'index', except with a more general 'Integral' argument
745{-# INLINABLE genericIndex #-}
746genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
747genericIndex i = mkFold step (Partial 0) (const Nothing)
756-- | Lookup the element at the given index.
758-- See also: 'Streamly.Prelude.!!'
761{-# INLINABLE index #-}
762index :: Monad m => Int -> Fold m a (Maybe a)
765-- | Extract the first element of the stream, if any.
768{-# INLINABLE head #-}
769head :: Monad m => Fold m a (Maybe a)
770head = mkFold_ (const (Done . Just)) (Partial Nothing)
772-- | Returns the first element that satisfies the given predicate.
775{-# INLINABLE find #-}
776find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
777find predicate = mkFold step (Partial ()) (const Nothing)
786-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
787-- first pair where the key equals the given value @a@.
789-- > lookup = snd <$> Fold.find ((==) . fst)
792{-# INLINABLE lookup #-}
793lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
794lookup a0 = mkFold step (Partial ()) (const Nothing)
803-- | Returns the first index that satisfies the given predicate.
806{-# INLINABLE findIndex #-}
807findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
808findIndex predicate = mkFold step (Partial 0) (const Nothing)
817-- | Returns the first index where a given value is found in the stream.
819-- > elemIndex a = Fold.findIndex (== a)
822{-# INLINABLE elemIndex #-}
823elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
824elemIndex a = findIndex (a ==)
826------------------------------------------------------------------------------
828------------------------------------------------------------------------------
830-- | Return 'True' if the input stream is empty.
832-- > null = fmap isJust Fold.head
835{-# INLINABLE null #-}
836null :: Monad m => Fold m a Bool
837null = mkFold (\() _ -> Done False) (Partial ()) (const True)
839-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
841-- >>> Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]
844-- > any p = Fold.lmap p Fold.or
848any :: Monad m => (a -> Bool) -> Fold m a Bool
849any predicate = mkFold_ step initial
853 initial = Partial False
860-- | Return 'True' if the given element is present in the stream.
862-- > elem a = Fold.any (== a)
865{-# INLINABLE elem #-}
866elem :: (Eq a, Monad m) => a -> Fold m a Bool
869-- | Returns 'True' if all elements of a stream satisfy a predicate.
871-- >>> Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]
874-- > all p = Fold.lmap p Fold.and
878all :: Monad m => (a -> Bool) -> Fold m a Bool
879all predicate = mkFold_ step initial
883 initial = Partial True
890-- | Returns 'True' if the given element is not present in the stream.
892-- > notElem a = Fold.all (/= a)
895{-# INLINABLE notElem #-}
896notElem :: (Eq a, Monad m) => a -> Fold m a Bool
897notElem a = all (a /=)
899-- | Returns 'True' if all elements are 'True', 'False' otherwise
901-- > and = Fold.all (== True)
905and :: Monad m => Fold m Bool Bool
908-- | Returns 'True' if any element is 'True', 'False' otherwise
910-- > or = Fold.any (== True)
914or :: Monad m => Fold m Bool Bool
917------------------------------------------------------------------------------
919------------------------------------------------------------------------------
921------------------------------------------------------------------------------
922-- Grouping without looking at elements
923------------------------------------------------------------------------------
925------------------------------------------------------------------------------
927------------------------------------------------------------------------------
929-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
930-- elements of its input are consumed by fold @f1@ and the rest of the stream
931-- is consumed by fold @f2@.
933-- >>> let splitAt_ n xs = Stream.fold (Fold.splitAt n Fold.toList Fold.toList) $ Stream.fromList xs
935-- >>> splitAt_ 6 "Hello World!"
936-- ("Hello ","World!")
938-- >>> splitAt_ (-1) [1,2,3]
941-- >>> splitAt_ 0 [1,2,3]
944-- >>> splitAt_ 1 [1,2,3]
947-- >>> splitAt_ 3 [1,2,3]
950-- >>> splitAt_ 4 [1,2,3]
953-- > splitAt n f1 f2 = Fold.serialWith (,) (Fold.take n f1) f2
957{-# INLINE splitAt #-}
964splitAt n fld = serialWith (,) (take n fld)
966------------------------------------------------------------------------------
968------------------------------------------------------------------------------
970------------------------------------------------------------------------------
972------------------------------------------------------------------------------
974-- Note: Keep this consistent with S.splitOn. In fact we should eliminate
975-- S.splitOn in favor of the fold.
977-- XXX Use Fold.many instead once it is fixed.
979-- | Like 'takeEndBy' but drops the element on which the predicate succeeds.
981-- >>> Stream.fold (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
984-- >>> Stream.toList $ Stream.foldMany (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
987-- > Stream.splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)
989-- See 'Streamly.Prelude.splitOnSuffix' for more details on splitting a
990-- stream using 'takeEndBy_'.
993{-# INLINE takeEndBy_ #-}
994takeEndBy_ :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
995takeEndBy_ predicate (Fold fstep finitial fextract) =
996 Fold step finitial fextract
1001 if not (predicate a)
1003 else Done <$> fextract s
1005-- | Take the input, stop when the predicate succeeds taking the succeeding
1008-- >>> Stream.fold (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1011-- >>> Stream.toList $ Stream.foldMany (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1012-- ["hello\n","there\n"]
1014-- > Stream.splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
1016-- See 'Streamly.Prelude.splitWithSuffix' for more details on splitting a
1017-- stream using 'takeEndBy'.
1020{-# INLINE takeEndBy #-}
1021takeEndBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1022takeEndBy predicate (Fold fstep finitial fextract) =
1023 Fold step finitial fextract
1029 if not (predicate a)
1033 Partial s1 -> Done <$> fextract s1
1034 Done b -> return $ Done b
1036------------------------------------------------------------------------------
1037-- Binary splitting on a separator
1038------------------------------------------------------------------------------
1041-- | Find the first occurrence of the specified sequence in the input stream
1042-- and break the input stream into two parts, the first part consisting of the
1043-- stream before the sequence and the second part consisting of the sequence
1044-- and the rest of the stream.
1046-- > let breakOn_ pat xs = S.fold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
1048-- >>> breakOn_ "dear" "Hello dear world!"
1049-- > ("Hello ","dear world!")
1051{-# INLINE breakOn #-}
1052breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
1053breakOn pat f m = undefined
1056------------------------------------------------------------------------------
1058------------------------------------------------------------------------------
1060-- | Distribute one copy of the stream to each fold and zip the results.
1063-- |-------Fold m a b--------|
1064-- ---stream m a---| |---m (b,c)
1065-- |-------Fold m a c--------|
1067-- >>> Stream.fold (Fold.tee Fold.sum Fold.length) (Stream.enumerateFromTo 1.0 100.0)
1070-- > tee = teeWith (,)
1074tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1077-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1078-- number of consumers? For polymorphic case a vector could be helpful. For
1079-- Storables we can use arrays. Will need separate APIs for those.
1081-- | Distribute one copy of the stream to each fold and collect the results in
1086-- |-------Fold m a b--------|
1087-- ---stream m a---| |---m [b]
1088-- |-------Fold m a b--------|
1093-- >>> Stream.fold (Fold.distribute [Fold.sum, Fold.length]) (Stream.enumerateFromTo 1 5)
1096-- > distribute = Prelude.foldr (Fold.teeWith (:)) (Fold.fromPure [])
1098-- This is the consumer side dual of the producer side 'sequence' operation.
1100-- Stops when all the folds stop.
1103{-# INLINE distribute #-}
1104distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1105distribute = Prelude.foldr (teeWith (:)) (fromPure [])
1107------------------------------------------------------------------------------
1109------------------------------------------------------------------------------
1111-- | Partition the input over two folds using an 'Either' partitioning
1116-- |-------Fold b x--------|
1117-- -----stream m a --> (Either b c)----| |----(x,y)
1118-- |-------Fold c y--------|
1121-- Send input to either fold randomly:
1124-- > import System.Random (randomIO)
1125-- > randomly a = randomIO >>= \\x -> return $ if x then Left a else Right a
1126-- > Stream.fold (Fold.partitionByM randomly Fold.length Fold.length) (Stream.enumerateFromTo 1 100)
1130-- Send input to the two folds in a proportion of 2:1:
1133-- proportionately m n = do
1134-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1136-- r <- readIORef ref
1137-- writeIORef ref $ tail r
1138-- return $ Prelude.head r a
1143-- f <- proportionately 2 1
1144-- r <- Stream.fold (Fold.partitionByM f Fold.length Fold.length) (Stream.enumerateFromTo (1 :: Int) 100)
1152-- This is the consumer side dual of the producer side 'mergeBy' operation.
1154-- When one fold is done, any input meant for it is ignored until the other
1155-- fold is also done.
1157-- Stops when both the folds stop.
1159-- /See also: 'partitionByFstM' and 'partitionByMinM'./
1162{-# INLINE partitionByM #-}
1163partitionByM :: Monad m
1164 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1165partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1166 Fold step begin done
1178 Partial sR -> RunBoth sL sR
1179 Done bR -> RunLeft sL bR
1182 Partial sR -> Partial $ RunRight bL sR
1183 Done bR -> Done (bL, bR)
1185 step (RunBoth sL sR) a = do
1193 Partial sres -> RunBoth sres sR
1194 Done bres -> RunRight bres sR
1200 Partial sres -> RunBoth sL sres
1201 Done bres -> RunLeft sL bres
1202 step (RunLeft sL bR) a = do
1209 Partial sres -> Partial $ RunLeft sres bR
1210 Done bres -> Done (bres, bR)
1211 Right _ -> return $ Partial $ RunLeft sL bR
1212 step (RunRight bL sR) a = do
1215 Left _ -> return $ Partial $ RunRight bL sR
1220 Partial sres -> Partial $ RunRight bL sres
1221 Done bres -> Done (bL, bres)
1223 done (RunBoth sL sR) = (,) <$> doneL sL <*> doneR sR
1224 done (RunLeft sL bR) = (,bR) <$> doneL sL
1225 done (RunRight bL sR) = (bL,) <$> doneR sR
1227-- | Similar to 'partitionByM' but terminates when the first fold terminates.
1230{-# INLINE partitionByFstM #-}
1231partitionByFstM :: Monad m
1232 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1233partitionByFstM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1234 Fold step begin done
1242 Partial sL -> return $
1244 Partial sR -> Partial $ RunFstBoth sL sR
1245 Done bR -> Partial $ RunFstLeft sL bR
1250 return $ Done (bL, bR)
1251 Done bR -> return $ Done (bL, bR)
1253 step (RunFstBoth sL sR) a = do
1259 Partial sres -> return $ Partial $ RunFstBoth sres sR
1262 return $ Done (bL, bR)
1269 Partial sres -> RunFstBoth sL sres
1270 Done bres -> RunFstLeft sL bres
1271 step (RunFstLeft sL bR) a = do
1278 Partial sres -> Partial $ RunFstLeft sres bR
1279 Done bres -> Done (bres, bR)
1280 Right _ -> return $ Partial $ RunFstLeft sL bR
1282 done (RunFstBoth sL sR) = (,) <$> doneL sL <*> doneR sR
1284 done (RunFstLeft sL bR) = (,bR) <$> doneL sL
1286-- | Similar to 'partitionByM' but terminates when any fold terminates.
1289{-# INLINE partitionByMinM #-}
1290partitionByMinM :: Monad m =>
1291 (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1292partitionByMinM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1293 Fold step begin done
1303 Partial sR -> return $ Partial $ Tuple' sL sR
1306 return $ Done (bL, bR)
1311 return $ Done (bL, bR)
1312 Done bR -> return $ Done (bL, bR)
1314 step (Tuple' sL sR) a = do
1320 Partial sres -> return $ Partial $ Tuple' sres sR
1323 return $ Done (bL, bR)
1328 Partial sres -> return $ Partial $ Tuple' sL sres
1331 return $ Done (bL, bR)
1333 done (Tuple' sL sR) = (,) <$> doneL sL <*> doneR sR
1335-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1336-- makes the signature clearer as to which case belongs to which fold.
1337-- XXX need to check the performance in both cases.
1338-- | Same as 'partitionByM' but with a pure partition function.
1340-- Count even and odd numbers in a stream:
1343-- let f = Fold.partitionBy (\n -> if even n then Left n else Right n)
1344-- (fmap (("Even " ++) . show) Fold.length)
1345-- (fmap (("Odd " ++) . show) Fold.length)
1346-- in Stream.fold f (Stream.enumerateFromTo 1 100)
1348-- ("Even 50","Odd 50")
1351{-# INLINE partitionBy #-}
1352partitionBy :: Monad m
1353 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1354partitionBy f = partitionByM (return . f)
1356-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1357-- and routes the 'Left' values to the first fold and 'Right' values to the
1360-- > partition = partitionBy id
1363{-# INLINE partition #-}
1365 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1366partition = partitionBy id
1369-- | Send one item to each fold in a round-robin fashion. This is the consumer
1370-- side dual of producer side 'mergeN' operation.
1372-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1373-- partitionN fs = Fold step begin done
1376-- TODO Demultiplex an input element into a number of typed variants. We want
1377-- to statically restrict the target values within a set of predefined types,
1378-- an enumeration of a GADT. We also want to make sure that the Map contains
1379-- only those types and the full set of those types.
1381-- TODO Instead of the input Map it should probably be a lookup-table using an
1382-- array and not in GC memory. The same applies to the output Map as well.
1383-- However, that would only be helpful if we have a very large data structure,
1384-- need to measure and see how it scales.
1386-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1389-- | Split the input stream based on a key field and fold each split using a
1390-- specific fold collecting the results in a map from the keys to the results.
1391-- Useful for cases like protocol handlers to handle different type of packets
1392-- using different handlers.
1396-- |-------Fold m a b
1397-- -----stream m a-----Map-----|
1398-- |-------Fold m a b
1403-- Any input that does not map to a fold in the input Map is silently ignored.
1405-- > demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain
1409{-# INLINE demuxWith #-}
1410demuxWith :: (Monad m, Ord k)
1411 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a (Map k b)
1412demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain
1414-- | Fold a stream of key value pairs using a map of specific folds for each
1415-- key into a map from keys to the results of fold outputs of the corresponding
1418-- >>> import qualified Data.Map
1420-- let table = Data.Map.fromList [("SUM", Fold.sum), ("PRODUCT", Fold.product)]
1421-- input = Stream.fromList [("SUM",1),("PRODUCT",2),("SUM",3),("PRODUCT",4)]
1422-- in Stream.fold (Fold.demux table) input
1424-- fromList [("PRODUCT",8),("SUM",4)]
1426-- > demux = demuxWith id
1430demux :: (Monad m, Ord k)
1431 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1434data DemuxState s b doneMap runMap =
1435 DemuxMapAndDefault !s !doneMap !runMap
1436 | DemuxOnlyMap b !doneMap !runMap
1437 | DemuxOnlyDefault s !doneMap
1439-- | Like 'demuxWith' but uses a default catchall fold to handle inputs which
1440-- do not have a specific fold in the map to handle them.
1442-- If any fold in the map stops, inputs meant for that fold are sent to the
1443-- catchall fold. If the catchall fold stops then inputs that do not match any
1446-- Stops when all the folds, including the catchall fold, stop.
1450{-# INLINE demuxDefaultWith #-}
1451demuxDefaultWith :: (Monad m, Ord k)
1453 -> Map k (Fold m a' b)
1455 -> Fold m a (Map k b, c)
1456demuxDefaultWith f kv (Fold dstep dinitial dextract) =
1457 Fold step initial extract
1462 let runInit (Fold step1 initial1 done1) = do
1466 Partial _ -> Right' (Fold step1 (return r) done1)
1469 -- initialize folds in the kv map and separate the ones that are done
1470 -- from running ones
1471 kv1 <- Prelude.mapM runInit kv
1472 let runMap = Map.map fromRight' $ Map.filter isRight' kv1
1473 doneMap = Map.map fromLeft' $ Map.filter isLeft' kv1
1475 -- Run the default fold, and decide the next state based on its result
1481 $ if Map.size runMap > 0
1482 then DemuxMapAndDefault s doneMap runMap
1483 else DemuxOnlyDefault s doneMap
1485 if Map.size runMap > 0
1486 then Partial $ DemuxOnlyMap b doneMap runMap
1487 else Done (doneMap, b)
1489 {-# INLINE runFold #-}
1490 runFold fPartial fDone doneMap runMap (Fold step1 initial1 done1) k a1 = do
1495 return $ case res of
1497 let fld = Fold step1 (return $ Partial s) done1
1498 runMap1 = Map.insert k fld runMap
1499 in Partial $ fPartial doneMap runMap1
1501 let runMap1 = Map.delete k runMap
1502 doneMap1 = Map.insert k b doneMap
1503 if Map.size runMap1 == 0
1505 else Partial $ fPartial doneMap1 runMap1
1506 Done _ -> error "Bug: demuxDefaultWith: Done fold"
1508 step (DemuxMapAndDefault dacc doneMap runMap) a = do
1510 case Map.lookup k runMap of
1512 res <- dstep dacc (k, a1)
1516 Partial s -> DemuxMapAndDefault s doneMap runMap
1517 Done b -> DemuxOnlyMap b doneMap runMap
1520 (DemuxMapAndDefault dacc)
1521 (Partial . DemuxOnlyDefault dacc)
1522 doneMap runMap fld k a1
1524 step (DemuxOnlyMap dval doneMap runMap) a = do
1526 case Map.lookup k runMap of
1527 Nothing -> return $ Partial $ DemuxOnlyMap dval doneMap runMap
1532 doneMap runMap fld k a1
1533 step (DemuxOnlyDefault dacc doneMap) a = do
1535 res <- dstep dacc (k, a1)
1538 Partial s -> Partial $ DemuxOnlyDefault s doneMap
1539 Done b -> Done (doneMap, b)
1541 runExtract (Fold _ initial1 done1) = do
1544 Partial s -> done1 s
1547 extract (DemuxMapAndDefault dacc doneMap runMap) = do
1549 runMap1 <- Prelude.mapM runExtract runMap
1550 return (doneMap `Map.union` runMap1, b)
1551 extract (DemuxOnlyMap dval doneMap runMap) = do
1552 runMap1 <- Prelude.mapM runExtract runMap
1553 return (doneMap `Map.union` runMap1, dval)
1554 extract (DemuxOnlyDefault dacc doneMap) = do
1559-- > demuxDefault = demuxDefaultWith id
1562{-# INLINE demuxDefault #-}
1563demuxDefault :: (Monad m, Ord k)
1564 => Map k (Fold m a b) -> Fold m (k, a) b -> Fold m (k, a) (Map k b, b)
1565demuxDefault = demuxDefaultWith id
1567-- TODO If the data is large we may need a map/hashmap in pinned memory instead
1568-- of a regular Map. That may require a serializable constraint though. We can
1569-- have another API for that.
1571-- | Split the input stream based on a key field and fold each split using the
1572-- given fold. Useful for map/reduce, bucketizing the input in different bins
1573-- or for generating histograms.
1576-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
1577-- in Stream.fold (Fold.classifyWith fst (Fold.map snd Fold.toList)) input
1579-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
1581-- If the classifier fold stops for a particular key any further inputs in that
1582-- bucket are ignored.
1588{-# INLINE classifyWith #-}
1589classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1590classifyWith f (Fold step1 initial1 extract1) =
1591 rmapM extract $ foldlM' step initial
1595 initial = return Map.empty
1598 case Map.lookup k kv of
1605 $ flip (Map.insert k) kv
1607 Partial s1 -> Left' s1
1609 Done b -> return $ Map.insert k (Right' b) kv
1615 $ flip (Map.insert k) kv
1617 Partial s1 -> Left' s1
1619 Right' _ -> return kv
1628 Left' s -> extract1 s
1629 Right' b -> return b)
1631-- | Given an input stream of key value pairs and a fold for values, fold all
1632-- the values belonging to each key. Useful for map/reduce, bucketizing the
1633-- input in different bins or for generating histograms.
1636-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
1637-- in Stream.fold (Fold.classify Fold.toList) input
1639-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
1643-- > classify fld = Fold.classifyWith fst (lmap snd fld)
1646{-# INLINE classify #-}
1647classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1648classify fld = classifyWith fst (lmap snd fld)
1650------------------------------------------------------------------------------
1652------------------------------------------------------------------------------
1654-- | Like 'unzipWith' but with a monadic splitter function.
1656-- @unzipWithM k f1 f2 = lmapM k (unzip f1 f2)@
1659{-# INLINE unzipWithM #-}
1660unzipWithM :: Monad m
1661 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1662unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1663 Fold step begin done
1675 Partial sR -> RunBoth sL sR
1676 Done bR -> RunLeft sL bR
1679 Partial sR -> Partial $ RunRight bL sR
1680 Done bR -> Done (bL, bR)
1682 step (RunBoth sL sR) a = do
1691 Partial sresR -> RunBoth sresL sresR
1692 Done bresR -> RunLeft sresL bresR
1696 Partial sresR -> Partial $ RunRight bresL sresR
1697 Done bresR -> Done (bresL, bresR)
1698 step (RunLeft sL bR) a = do
1703 Partial sresL -> Partial $ RunLeft sresL bR
1704 Done bresL -> Done (bresL, bR)
1705 step (RunRight bL sR) a = do
1710 Partial sresR -> Partial $ RunRight bL sresR
1711 Done bresR -> Done (bL, bresR)
1713 done (RunBoth sL sR) = (,) <$> doneL sL <*> doneR sR
1714 done (RunLeft sL bR) = (,bR) <$> doneL sL
1715 done (RunRight bL sR) = (bL,) <$> doneR sR
1717-- | Similar to 'unzipWithM' but terminates when the first fold terminates.
1720{-# INLINE unzipWithFstM #-}
1721unzipWithFstM :: Monad m =>
1722 (a -> m (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1723unzipWithFstM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1724 Fold step begin done
1732 Partial sL -> return $ Partial $
1734 Partial sR -> RunFstBoth sL sR
1735 Done bR -> RunFstLeft sL bR
1740 return $ Done (bL, bR)
1741 Done bR -> return $ Done (bL, bR)
1743 step (RunFstBoth sL sR) a = do
1748 Partial sresL -> return $ Partial $
1750 Partial sresR -> RunFstBoth sresL sresR
1751 Done bresR -> RunFstLeft sresL bresR
1756 return $ Done (bL, bR)
1757 Done bR -> return $ Done (bL, bR)
1759 step (RunFstLeft sL bR) a = do
1764 Partial sresL -> Partial $ RunFstLeft sresL bR
1765 Done bresL -> Done (bresL, bR)
1767 done (RunFstBoth sL sR) = (,) <$> doneL sL <*> doneR sR
1768 done (RunFstLeft sL bR) = (,bR) <$> doneL sL
1770-- | Similar to 'unzipWithM' but terminates when any fold terminates.
1773{-# INLINE unzipWithMinM #-}
1774unzipWithMinM :: Monad m =>
1775 (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1776unzipWithMinM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1777 Fold step begin done
1787 Partial sR -> return $ Partial $ Tuple' sL sR
1790 return $ Done (bL, bR)
1795 return $ Done (bL, bR)
1796 Done bR -> return $ Done (bL, bR)
1798 step (Tuple' sL sR) a = do
1805 Partial sresR -> return $ Partial $ Tuple' sresL sresR
1808 return $ Done (bL, bR)
1813 return $ Done (bL, bR)
1814 Done bR -> return $ Done (bL, bR)
1816 done (Tuple' sL sR) = (,) <$> doneL sL <*> doneR sR
1818-- | Split elements in the input stream into two parts using a pure splitter
1819-- function, direct each part to a different fold and zip the results.
1821-- @unzipWith f fld1 fld2 = Fold.lmap f (Fold.unzip fld1 fld2)@
1823-- This fold terminates when both the input folds terminate.
1826{-# INLINE unzipWith #-}
1828 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1829unzipWith f = unzipWithM (return . f)
1831-- | Send the elements of tuples in a stream of tuples through two different
1836-- |-------Fold m a x--------|
1837-- ---------stream of (a,b)--| |----m (x,y)
1838-- |-------Fold m b y--------|
1842-- > unzip = Fold.unzipWith id
1844-- This is the consumer side dual of the producer side 'zip' operation.
1848unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1851------------------------------------------------------------------------------
1852-- Combining streams and folds - Zipping
1853------------------------------------------------------------------------------
1855-- | Zip a stream with the input of a fold using the supplied function.
1859{-# INLINE zipWithM #-}
1860zipWithM :: -- Monad m =>
1861 (a -> b -> m c) -> t m a -> Fold m c x -> Fold m b x
1864-- | Zip a stream with the input of a fold.
1869zip :: Monad m => t m a -> Fold m (a, b) x -> Fold m b x
1870zip = zipWithM (curry return)
1872-- | Pair each element of a fold input with its index, starting from index 0.
1875{-# INLINE indexed #-}
1876indexed :: forall m a b. Monad m => Fold m (Int, a) b -> Fold m a b
1877indexed = zip (Stream.enumerateFrom 0 :: SerialT m Int)
1879-- | Change the predicate function of a Fold from @a -> b@ to accept an
1880-- additional state input @(s, a) -> b@. Convenient to filter with an
1881-- addiitonal index or time input.
1884-- filterWithIndex = with indexed filter
1885-- filterWithAbsTime = with timestamped filter
1886-- filterWithRelTime = with timeIndexed filter
1892 (Fold m (s, a) b -> Fold m a b)
1893 -> (((s, a) -> c) -> Fold m (s, a) b -> Fold m (s, a) b)
1894 -> (((s, a) -> c) -> Fold m a b -> Fold m a b)
1895with f comb g = f . comb g . lmap snd
1897-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
1898-- then every element at strides of @stride@.
1901{-# INLINE sampleFromthen #-}
1902sampleFromthen :: Monad m => Int -> Int -> Fold m a b -> Fold m a b
1903sampleFromthen offset size =
1904 with indexed filter (\(i, _) -> (i + offset) `mod` size == 0)
1906------------------------------------------------------------------------------
1908------------------------------------------------------------------------------
1910-- | @concatSequence f t@ applies folds from stream @t@ sequentially and
1911-- collects the results using the fold @f@.
1915{-# INLINE concatSequence #-}
1918 Fold m b c -> t (Fold m a b) -> Fold m a c
1919concatSequence _f _p = undefined
1921-- | Group the input stream into groups of elements between @low@ and @high@.
1922-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1923-- @high@. Each chunk is folded using the provided fold function.
1925-- This could be useful, for example, when we are folding a stream of unknown
1926-- size to a stream of arrays and we want to minimize the number of
1929-- NOTE: this would be an application of "many" using a terminating fold.
1933{-# INLINE chunksBetween #-}
1934chunksBetween :: -- Monad m =>
1935 Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1936chunksBetween _low _high _f1 _f2 = undefined
1938-- | A fold that buffers its input to a pure stream.
1940-- /Warning!/ working on large streams accumulated as buffers in memory could
1941-- be very inefficient, consider using "Streamly.Data.Array" instead.
1943-- > toStream = foldr K.cons K.nil
1946{-# INLINE toStream #-}
1947toStream :: Monad m => Fold m a (SerialT Identity a)
1948toStream = foldr K.cons K.nil
1950-- This is more efficient than 'toStream'. toStream is exactly the same as
1951-- reversing the stream after toStreamRev.
1953-- | Buffers the input stream to a pure stream in the reverse order of the
1956-- > toStreamRev = foldl' (flip K.cons) K.nil
1958-- /Warning!/ working on large streams accumulated as buffers in memory could
1959-- be very inefficient, consider using "Streamly.Data.Array" instead.
1963-- xn : ... : x2 : x1 : []
1964{-# INLINABLE toStreamRev #-}
1965toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1966toStreamRev = foldl' (flip K.cons) K.nil
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