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
125 -- ** Mapping Filters
131 -- ** Scanning Filters
136 -- | 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 -- | Normally you would run a fold to completion by supplying it a stream,
227 -- e.g. using 'Stream.fold'. However, you could also run a fold partially
228 -- by using 'duplicate' on it and then running it with a stream.
229 -- Alternatively, 'initialize', 'snoc' and 'finish' can be used to run a
230 -- fold incrementally, however, that may not be the most efficient way to
243import Data.Bifunctor (first)
244import Data.Either (isLeft, isRight)
245import Data.Functor.Identity (Identity(..))
246import Data.Int (Int64)
247import Data.Map.Strict (Map)
248import Data.Maybe (isJust, fromJust)
249#if __GLASGOW_HASKELL__ < 804
250import Data.Semigroup (Semigroup((<>)))
252import Streamly.Internal.BaseCompat (fromLeft, fromRight)
253import Streamly.Internal.Data.Either.Strict
254 (Either'(..), fromLeft', fromRight', isLeft', isRight')
255import Streamly.Internal.Data.Pipe.Type (Pipe (..), PipeState(..))
256import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
257import Streamly.Internal.Data.Stream.Serial (SerialT(..))
259import qualified Data.Map.Strict as Map
260import qualified Streamly.Internal.Data.Pipe.Type as Pipe
261-- import qualified Streamly.Internal.Data.Stream.IsStream.Enumeration as Stream
262import qualified Prelude
265 ( filter, foldl1, drop, dropWhile, take, takeWhile, zipWith
266 , foldl, foldr, map, mapM_, sequence, all, any, sum, product, elem
267 , notElem, maximum, minimum, head, last, tail, length, null
268 , reverse, iterate, init, and, or, lookup, (!!)
269 , scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip
270 , span, splitAt, break, mapM, zip)
271import Streamly.Internal.Data.Fold.Type
275-- >>> import Prelude hiding (break, map, span, splitAt)
276-- >>> import qualified Streamly.Prelude as Stream
277-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream (parse, foldMany)
278-- >>> import qualified Streamly.Data.Fold as Fold
279-- >>> import qualified Streamly.Internal.Data.Fold as Fold
280-- >>> import qualified Streamly.Internal.Data.Fold.Type as Fold
281-- >>> import qualified Streamly.Internal.Data.Parser as Parser
282-- >>> import Streamly.Internal.Data.Stream.Serial (SerialT(..))
283-- >>> import Data.IORef (newIORef, readIORef, writeIORef)
285------------------------------------------------------------------------------
287------------------------------------------------------------------------------
289-- | Change the underlying monad of a fold
292hoist :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b
293hoist f (Fold step initial extract) =
294 Fold (\x a -> f $ step x a) (f initial) (f . extract)
296-- | Adapt a pure fold to any monad
298-- > generally = Fold.hoist (return . runIdentity)
301generally :: Monad m => Fold Identity a b -> Fold m a b
302generally = hoist (return . runIdentity)
304------------------------------------------------------------------------------
305-- Transformations on fold inputs
306------------------------------------------------------------------------------
308-- | Flatten the monadic output of a fold to pure output.
311{-# DEPRECATED sequence "Use \"rmapM id\" instead" #-}
312{-# INLINE sequence #-}
313sequence :: Monad m => Fold m a (m b) -> Fold m a b
316-- | Map a monadic function on the output of a fold.
319{-# DEPRECATED mapM "Use rmapM instead" #-}
321mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
324-- | @mapMaybe f fold@ maps a 'Maybe' returning function @f@ on the input of
325-- the fold, filters out 'Nothing' elements, and return the values extracted
328-- >>> f x = if even x then Just x else Nothing
329-- >>> fld = Fold.mapMaybe f Fold.toList
330-- >>> Stream.fold fld (Stream.enumerateFromTo 1 10)
334{-# INLINE mapMaybe #-}
335mapMaybe :: (Monad m) => (a -> Maybe b) -> Fold m b r -> Fold m a r
336mapMaybe f = lmap f . filter isJust . lmap fromJust
338------------------------------------------------------------------------------
339-- Transformations on fold inputs
340------------------------------------------------------------------------------
344-- | Apply a transformation on a 'Fold' using a 'Pipe'.
347{-# INLINE transform #-}
348transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
349transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
350 Fold step initial extract
354 initial = first (Tuple' pinitial) <$> finitial
356 step (Tuple' ps fs) x = do
363 go acc (Pipe.Yield b (Consume ps')) = do
367 Partial s -> Partial $ Tuple' ps' s
369 go acc (Pipe.Yield b (Produce ps')) = do
374 Done b2 -> return $ Done b2
375 go acc (Pipe.Continue (Consume ps')) =
376 return $ Partial $ Tuple' ps' acc
377 go acc (Pipe.Continue (Produce ps')) = do
381 extract (Tuple' _ fs) = fextract fs
383-- | Scan the input of a 'Fold' to change it in a stateful manner using another
387scan :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
388scan (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
389 Fold step initial extract
393 runStep actionL sR = do
399 Partial sR1 -> Done <$> extractR sR1
400 Done bR -> return $ Done bR
406 Partial sR1 -> Partial (sL, sR1)
412 Partial sR -> runStep initialL sR
413 Done b -> return $ Done b
415 step (sL, sR) x = runStep (stepL sL x) sR
417 extract = extractR . snd
419------------------------------------------------------------------------------
421------------------------------------------------------------------------------
423------------------------------------------------------------------------------
425------------------------------------------------------------------------------
428-- > drainBy f = lmapM f drain
429-- > drainBy = Fold.foldMapM (void . f)
431-- Drain all input after passing it through a monadic function. This is the
432-- dual of mapM_ on stream producers.
434-- See also: 'Streamly.Prelude.mapM_'
437{-# INLINABLE drainBy #-}
438drainBy :: Monad m => (a -> m b) -> Fold m a ()
439drainBy f = lmapM f drain
441-- | Extract the last element of the input stream, if any.
443-- > last = fmap getLast $ Fold.foldMap (Last . Just)
446{-# INLINABLE last #-}
447last :: Monad m => Fold m a (Maybe a)
448last = foldl1' (\_ x -> x)
450------------------------------------------------------------------------------
452------------------------------------------------------------------------------
454-- | Like 'length', except with a more general 'Num' return value
456-- > genericLength = fmap getSum $ foldMap (Sum . const 1)
459{-# INLINE genericLength #-}
460genericLength :: (Monad m, Num b) => Fold m a b
461genericLength = foldl' (\n _ -> n + 1) 0
463-- | Determine the length of the input stream.
465-- > length = fmap getSum $ Fold.foldMap (Sum . const 1)
469length :: Monad m => Fold m a Int
470length = genericLength
472-- | Determine the sum of all elements of a stream of numbers. Returns additive
473-- identity (@0@) when the stream is empty. Note that this is not numerically
474-- stable for floating point numbers.
476-- > sum = fmap getSum $ Fold.foldMap Sum
480sum :: (Monad m, Num a) => Fold m a a
483-- | Determine the product of all elements of a stream of numbers. Returns
484-- multiplicative identity (@1@) when the stream is empty. The fold terminates
485-- when it encounters (@0@) in its input.
487-- Compare with @Fold.foldMap Product@.
490-- /Since 0.8.0 (Added 'Eq' constraint)/
491{-# INLINE product #-}
492product :: (Monad m, Num a, Eq a) => Fold m a a
493product = mkFold_ step (Partial 1)
502------------------------------------------------------------------------------
504------------------------------------------------------------------------------
506-- | Determine the maximum element in a stream using the supplied comparison
510{-# INLINE maximumBy #-}
511maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
512maximumBy cmp = foldl1' max'
523-- maximum = Fold.maximumBy compare
526-- Determine the maximum element in a stream.
528-- Compare with @Fold.foldMap Max@.
531{-# INLINE maximum #-}
532maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
535-- | Computes the minimum element with respect to the given comparison function
538{-# INLINE minimumBy #-}
539minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
540minimumBy cmp = foldl1' min'
549-- | Determine the minimum element in a stream using the supplied comparison
553-- minimum = 'minimumBy' compare
556-- Compare with @Fold.foldMap Min@.
559{-# INLINE minimum #-}
560minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
563------------------------------------------------------------------------------
564-- To Summary (Statistical)
565------------------------------------------------------------------------------
567-- | Compute a numerically stable arithmetic mean of all elements in the input
571{-# INLINABLE mean #-}
572mean :: (Monad m, Fractional a) => Fold m a a
573mean = fmap done $ foldl' step begin
579 step (Tuple' x n) y =
581 in Tuple' (x + (y - x) / n1) n1
583 done (Tuple' x _) = x
585-- | Compute a numerically stable (population) variance over all elements in
589{-# INLINABLE variance #-}
590variance :: (Monad m, Fractional a) => Fold m a a
591variance = fmap done $ foldl' step begin
595 begin = Tuple3' 0 0 0
597 step (Tuple3' n mean_ m2) x = Tuple3' n' mean' m2'
602 mean' = (n * mean_ + x) / (n + 1)
604 m2' = m2 + delta * delta * n / (n + 1)
606 done (Tuple3' n _ m2) = m2 / n
608-- | Compute a numerically stable (population) standard deviation over all
609-- elements in the input stream.
612{-# INLINABLE stdDev #-}
613stdDev :: (Monad m, Floating a) => Fold m a a
614stdDev = sqrt <$> variance
616-- | Compute an 'Int' sized polynomial rolling hash
618-- > H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0
620-- Where @c1@, @c2@, @cn@ are the elements in the input stream and @k@ is a
623-- This hash is often used in Rabin-Karp string search algorithm.
625-- See https://en.wikipedia.org/wiki/Rolling_hash
628{-# INLINABLE rollingHashWithSalt #-}
629rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64
630rollingHashWithSalt = foldl' step
634 k = 2891336453 :: Int64
636 step cksum a = cksum * k + fromIntegral (fromEnum a)
638-- | A default salt used in the implementation of 'rollingHash'.
639{-# INLINE defaultSalt #-}
641defaultSalt = -2578643520546668380
643-- | Compute an 'Int' sized polynomial rolling hash of a stream.
645-- > rollingHash = Fold.rollingHashWithSalt defaultSalt
648{-# INLINABLE rollingHash #-}
649rollingHash :: (Monad m, Enum a) => Fold m a Int64
650rollingHash = rollingHashWithSalt defaultSalt
652-- | Compute an 'Int' sized polynomial rolling hash of the first n elements of
655-- > rollingHashFirstN = Fold.take n Fold.rollingHash
658{-# INLINABLE rollingHashFirstN #-}
659rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64
660rollingHashFirstN n = take n rollingHash
662------------------------------------------------------------------------------
663-- Monoidal left folds
664------------------------------------------------------------------------------
666-- | Append the elements of an input stream to a provided starting value.
668-- >>> Stream.fold (Fold.sconcat 10) (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
672-- sconcat = Fold.foldl' (<>)
676{-# INLINE sconcat #-}
677sconcat :: (Monad m, Semigroup a) => a -> Fold m a a
680-- | Fold an input stream consisting of monoidal elements using 'mappend'
683-- >>> Stream.fold Fold.mconcat (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
686-- > mconcat = Fold.sconcat mempty
689{-# INLINE mconcat #-}
692#if !MIN_VERSION_base(4,11,0)
695 , Monoid a) => Fold m a a
696mconcat = sconcat mempty
699-- > foldMap f = Fold.lmap f Fold.mconcat
701-- Make a fold from a pure function that folds the output of the function
702-- using 'mappend' and 'mempty'.
704-- >>> Stream.fold (Fold.foldMap Data.Monoid.Sum) $ Stream.enumerateFromTo 1 10
708{-# INLINABLE foldMap #-}
709foldMap :: (Monad m, Monoid b
710#if !MIN_VERSION_base(4,11,0)
713 ) => (a -> b) -> Fold m a b
714foldMap f = lmap f mconcat
717-- > foldMapM f = Fold.lmapM f Fold.mconcat
719-- Make a fold from a monadic function that folds the output of the function
720-- using 'mappend' and 'mempty'.
722-- >>> Stream.fold (Fold.foldMapM (return . Data.Monoid.Sum)) $ Stream.enumerateFromTo 1 10
726{-# INLINABLE foldMapM #-}
727foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
728foldMapM act = foldlM' step (pure mempty)
734 return $! mappend m m'
736------------------------------------------------------------------------------
738------------------------------------------------------------------------------
741-- This is more efficient than 'Streamly.Internal.Data.Fold.toList'. toList is
742-- exactly the same as reversing the list after 'toListRev'.
744-- | Buffers the input stream to a list in the reverse order of the input.
746-- > toListRev = Fold.foldl' (flip (:)) []
748-- /Warning!/ working on large lists accumulated as buffers in memory could be
749-- very inefficient, consider using "Streamly.Array" instead.
753-- xn : ... : x2 : x1 : []
754{-# INLINABLE toListRev #-}
755toListRev :: Monad m => Fold m a [a]
756toListRev = foldl' (flip (:)) []
758------------------------------------------------------------------------------
760------------------------------------------------------------------------------
762-- | A fold that drains the first n elements of its input, running the effects
763-- and discarding the results.
765-- > drainN n = Fold.take n Fold.drain
768{-# INLINABLE drainN #-}
769drainN :: Monad m => Int -> Fold m a ()
770drainN n = take n drain
772------------------------------------------------------------------------------
774------------------------------------------------------------------------------
776-- | Like 'index', except with a more general 'Integral' argument
779{-# INLINABLE genericIndex #-}
780genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
781genericIndex i = mkFold step (Partial 0) (const Nothing)
790-- | Lookup the element at the given index.
792-- See also: 'Streamly.Prelude.!!'
795{-# INLINABLE index #-}
796index :: Monad m => Int -> Fold m a (Maybe a)
799-- | Extract the first element of the stream, if any.
802{-# INLINABLE head #-}
803head :: Monad m => Fold m a (Maybe a)
804head = mkFold_ (const (Done . Just)) (Partial Nothing)
806-- | Returns the first element that satisfies the given predicate.
809{-# INLINABLE find #-}
810find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
811find predicate = mkFold step (Partial ()) (const Nothing)
820-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
821-- first pair where the key equals the given value @a@.
823-- > lookup = snd <$> Fold.find ((==) . fst)
826{-# INLINABLE lookup #-}
827lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
828lookup a0 = mkFold step (Partial ()) (const Nothing)
837-- | Returns the first index that satisfies the given predicate.
840{-# INLINABLE findIndex #-}
841findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
842findIndex predicate = mkFold step (Partial 0) (const Nothing)
851-- | Returns the first index where a given value is found in the stream.
853-- > elemIndex a = Fold.findIndex (== a)
856{-# INLINABLE elemIndex #-}
857elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
858elemIndex a = findIndex (a ==)
860------------------------------------------------------------------------------
862------------------------------------------------------------------------------
864-- | Return 'True' if the input stream is empty.
866-- > null = fmap isJust Fold.head
869{-# INLINABLE null #-}
870null :: Monad m => Fold m a Bool
871null = mkFold (\() _ -> Done False) (Partial ()) (const True)
873-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
875-- >>> Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]
878-- > any p = Fold.lmap p Fold.or
882any :: Monad m => (a -> Bool) -> Fold m a Bool
883any predicate = mkFold_ step initial
887 initial = Partial False
894-- | Return 'True' if the given element is present in the stream.
896-- > elem a = Fold.any (== a)
899{-# INLINABLE elem #-}
900elem :: (Eq a, Monad m) => a -> Fold m a Bool
903-- | Returns 'True' if all elements of a stream satisfy a predicate.
905-- >>> Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]
908-- > all p = Fold.lmap p Fold.and
912all :: Monad m => (a -> Bool) -> Fold m a Bool
913all predicate = mkFold_ step initial
917 initial = Partial True
924-- | Returns 'True' if the given element is not present in the stream.
926-- > notElem a = Fold.all (/= a)
929{-# INLINABLE notElem #-}
930notElem :: (Eq a, Monad m) => a -> Fold m a Bool
931notElem a = all (a /=)
933-- | Returns 'True' if all elements are 'True', 'False' otherwise
935-- > and = Fold.all (== True)
939and :: Monad m => Fold m Bool Bool
942-- | Returns 'True' if any element is 'True', 'False' otherwise
944-- > or = Fold.any (== True)
948or :: Monad m => Fold m Bool Bool
951------------------------------------------------------------------------------
953------------------------------------------------------------------------------
955------------------------------------------------------------------------------
956-- Grouping without looking at elements
957------------------------------------------------------------------------------
959------------------------------------------------------------------------------
961------------------------------------------------------------------------------
963-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
964-- elements of its input are consumed by fold @f1@ and the rest of the stream
965-- is consumed by fold @f2@.
967-- >>> let splitAt_ n xs = Stream.fold (Fold.splitAt n Fold.toList Fold.toList) $ Stream.fromList xs
969-- >>> splitAt_ 6 "Hello World!"
970-- ("Hello ","World!")
972-- >>> splitAt_ (-1) [1,2,3]
975-- >>> splitAt_ 0 [1,2,3]
978-- >>> splitAt_ 1 [1,2,3]
981-- >>> splitAt_ 3 [1,2,3]
984-- >>> splitAt_ 4 [1,2,3]
987-- > splitAt n f1 f2 = Fold.serialWith (,) (Fold.take n f1) f2
991{-# INLINE splitAt #-}
998splitAt n fld = serialWith (,) (take n fld)
1000------------------------------------------------------------------------------
1001-- Element Aware APIs
1002------------------------------------------------------------------------------
1004------------------------------------------------------------------------------
1006------------------------------------------------------------------------------
1008-- Note: Keep this consistent with S.splitOn. In fact we should eliminate
1009-- S.splitOn in favor of the fold.
1011-- XXX Use Fold.many instead once it is fixed.
1013-- | Like 'takeEndBy' but drops the element on which the predicate succeeds.
1015-- >>> Stream.fold (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1018-- >>> Stream.toList $ Stream.foldMany (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1021-- > Stream.splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)
1023-- See 'Streamly.Prelude.splitOnSuffix' for more details on splitting a
1024-- stream using 'takeEndBy_'.
1027{-# INLINE takeEndBy_ #-}
1028takeEndBy_ :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1029takeEndBy_ predicate (Fold fstep finitial fextract) =
1030 Fold step finitial fextract
1035 if not (predicate a)
1037 else Done <$> fextract s
1039-- | Take the input, stop when the predicate succeeds taking the succeeding
1042-- >>> Stream.fold (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1045-- >>> Stream.toList $ Stream.foldMany (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1046-- ["hello\n","there\n"]
1048-- > Stream.splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
1050-- See 'Streamly.Prelude.splitWithSuffix' for more details on splitting a
1051-- stream using 'takeEndBy'.
1054{-# INLINE takeEndBy #-}
1055takeEndBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1056takeEndBy predicate (Fold fstep finitial fextract) =
1057 Fold step finitial fextract
1063 if not (predicate a)
1067 Partial s1 -> Done <$> fextract s1
1068 Done b -> return $ Done b
1070------------------------------------------------------------------------------
1071-- Binary splitting on a separator
1072------------------------------------------------------------------------------
1075-- | Find the first occurrence of the specified sequence in the input stream
1076-- and break the input stream into two parts, the first part consisting of the
1077-- stream before the sequence and the second part consisting of the sequence
1078-- and the rest of the stream.
1080-- > let breakOn_ pat xs = S.fold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
1082-- >>> breakOn_ "dear" "Hello dear world!"
1083-- > ("Hello ","dear world!")
1085{-# INLINE breakOn #-}
1086breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
1087breakOn pat f m = undefined
1090------------------------------------------------------------------------------
1092------------------------------------------------------------------------------
1094-- | Distribute one copy of the stream to each fold and zip the results.
1097-- |-------Fold m a b--------|
1098-- ---stream m a---| |---m (b,c)
1099-- |-------Fold m a c--------|
1101-- >>> Stream.fold (Fold.tee Fold.sum Fold.length) (Stream.enumerateFromTo 1.0 100.0)
1104-- > tee = teeWith (,)
1108tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1111-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1112-- number of consumers? For polymorphic case a vector could be helpful. For
1113-- Storables we can use arrays. Will need separate APIs for those.
1115-- | Distribute one copy of the stream to each fold and collect the results in
1120-- |-------Fold m a b--------|
1121-- ---stream m a---| |---m [b]
1122-- |-------Fold m a b--------|
1127-- >>> Stream.fold (Fold.distribute [Fold.sum, Fold.length]) (Stream.enumerateFromTo 1 5)
1130-- > distribute = Prelude.foldr (Fold.teeWith (:)) (Fold.fromPure [])
1132-- This is the consumer side dual of the producer side 'sequence' operation.
1134-- Stops when all the folds stop.
1137{-# INLINE distribute #-}
1138distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1139distribute = Prelude.foldr (teeWith (:)) (fromPure [])
1141------------------------------------------------------------------------------
1143------------------------------------------------------------------------------
1145{-# INLINE partitionByMUsing #-}
1146partitionByMUsing :: Monad m =>
1147 ( (x -> y -> (x, y))
1148 -> Fold m (Either b c) x
1149 -> Fold m (Either b c) y
1150 -> Fold m (Either b c) (x, y)
1152 -> (a -> m (Either b c))
1156partitionByMUsing t f fld1 fld2 =
1157 let l = lmap (fromLeft undefined) fld1 -- :: Fold m (Either b c) x
1158 r = lmap (fromRight undefined) fld2 -- :: Fold m (Either b c) y
1159 in lmapM f (t (,) (filter isLeft l) (filter isRight r))
1161-- | Partition the input over two folds using an 'Either' partitioning
1166-- |-------Fold b x--------|
1167-- -----stream m a --> (Either b c)----| |----(x,y)
1168-- |-------Fold c y--------|
1171-- Send input to either fold randomly:
1174-- > import System.Random (randomIO)
1175-- > randomly a = randomIO >>= \\x -> return $ if x then Left a else Right a
1176-- > Stream.fold (Fold.partitionByM randomly Fold.length Fold.length) (Stream.enumerateFromTo 1 100)
1180-- Send input to the two folds in a proportion of 2:1:
1183-- proportionately m n = do
1184-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1186-- r <- readIORef ref
1187-- writeIORef ref $ tail r
1188-- return $ Prelude.head r a
1193-- f <- proportionately 2 1
1194-- r <- Stream.fold (Fold.partitionByM f Fold.length Fold.length) (Stream.enumerateFromTo (1 :: Int) 100)
1202-- This is the consumer side dual of the producer side 'mergeBy' operation.
1204-- When one fold is done, any input meant for it is ignored until the other
1205-- fold is also done.
1207-- Stops when both the folds stop.
1209-- /See also: 'partitionByFstM' and 'partitionByMinM'./
1212{-# INLINE partitionByM #-}
1213partitionByM :: Monad m
1214 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1215partitionByM = partitionByMUsing teeWith
1217-- | Similar to 'partitionByM' but terminates when the first fold terminates.
1220{-# INLINE partitionByFstM #-}
1221partitionByFstM :: Monad m
1222 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1223partitionByFstM = partitionByMUsing teeWithFst
1225-- | Similar to 'partitionByM' but terminates when any fold terminates.
1228{-# INLINE partitionByMinM #-}
1229partitionByMinM :: Monad m =>
1230 (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1231partitionByMinM = partitionByMUsing teeWithMin
1233-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1234-- makes the signature clearer as to which case belongs to which fold.
1235-- XXX need to check the performance in both cases.
1236-- | Same as 'partitionByM' but with a pure partition function.
1238-- Count even and odd numbers in a stream:
1241-- let f = Fold.partitionBy (\n -> if even n then Left n else Right n)
1242-- (fmap (("Even " ++) . show) Fold.length)
1243-- (fmap (("Odd " ++) . show) Fold.length)
1244-- in Stream.fold f (Stream.enumerateFromTo 1 100)
1246-- ("Even 50","Odd 50")
1249{-# INLINE partitionBy #-}
1250partitionBy :: Monad m
1251 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1252partitionBy f = partitionByM (return . f)
1254-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1255-- and routes the 'Left' values to the first fold and 'Right' values to the
1258-- > partition = partitionBy id
1261{-# INLINE partition #-}
1263 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1264partition = partitionBy id
1267-- | Send one item to each fold in a round-robin fashion. This is the consumer
1268-- side dual of producer side 'mergeN' operation.
1270-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1271-- partitionN fs = Fold step begin done
1274-- TODO Demultiplex an input element into a number of typed variants. We want
1275-- to statically restrict the target values within a set of predefined types,
1276-- an enumeration of a GADT. We also want to make sure that the Map contains
1277-- only those types and the full set of those types.
1279-- TODO Instead of the input Map it should probably be a lookup-table using an
1280-- array and not in GC memory. The same applies to the output Map as well.
1281-- However, that would only be helpful if we have a very large data structure,
1282-- need to measure and see how it scales.
1284-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1287-- | Split the input stream based on a key field and fold each split using a
1288-- specific fold collecting the results in a map from the keys to the results.
1289-- Useful for cases like protocol handlers to handle different type of packets
1290-- using different handlers.
1294-- |-------Fold m a b
1295-- -----stream m a-----Map-----|
1296-- |-------Fold m a b
1301-- Any input that does not map to a fold in the input Map is silently ignored.
1303-- > demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain
1307{-# INLINE demuxWith #-}
1308demuxWith :: (Monad m, Ord k)
1309 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a (Map k b)
1310demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain
1312-- | Fold a stream of key value pairs using a map of specific folds for each
1313-- key into a map from keys to the results of fold outputs of the corresponding
1316-- >>> import qualified Data.Map
1318-- let table = Data.Map.fromList [("SUM", Fold.sum), ("PRODUCT", Fold.product)]
1319-- input = Stream.fromList [("SUM",1),("PRODUCT",2),("SUM",3),("PRODUCT",4)]
1320-- in Stream.fold (Fold.demux table) input
1322-- fromList [("PRODUCT",8),("SUM",4)]
1324-- > demux = demuxWith id
1328demux :: (Monad m, Ord k)
1329 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1332data DemuxState s b doneMap runMap =
1333 DemuxMapAndDefault !s !doneMap !runMap
1334 | DemuxOnlyMap b !doneMap !runMap
1335 | DemuxOnlyDefault s !doneMap
1337-- | Like 'demuxWith' but uses a default catchall fold to handle inputs which
1338-- do not have a specific fold in the map to handle them.
1340-- If any fold in the map stops, inputs meant for that fold are sent to the
1341-- catchall fold. If the catchall fold stops then inputs that do not match any
1344-- Stops when all the folds, including the catchall fold, stop.
1348{-# INLINE demuxDefaultWith #-}
1349demuxDefaultWith :: (Monad m, Ord k)
1351 -> Map k (Fold m a' b)
1353 -> Fold m a (Map k b, c)
1354demuxDefaultWith f kv (Fold dstep dinitial dextract) =
1355 Fold step initial extract
1360 let runInit (Fold step1 initial1 done1) = do
1364 Partial _ -> Right' (Fold step1 (return r) done1)
1367 -- initialize folds in the kv map and separate the ones that are done
1368 -- from running ones
1369 kv1 <- Prelude.mapM runInit kv
1370 let runMap = Map.map fromRight' $ Map.filter isRight' kv1
1371 doneMap = Map.map fromLeft' $ Map.filter isLeft' kv1
1373 -- Run the default fold, and decide the next state based on its result
1379 $ if Map.size runMap > 0
1380 then DemuxMapAndDefault s doneMap runMap
1381 else DemuxOnlyDefault s doneMap
1383 if Map.size runMap > 0
1384 then Partial $ DemuxOnlyMap b doneMap runMap
1385 else Done (doneMap, b)
1387 {-# INLINE runFold #-}
1388 runFold fPartial fDone doneMap runMap (Fold step1 initial1 done1) k a1 = do
1393 return $ case res of
1395 let fld = Fold step1 (return $ Partial s) done1
1396 runMap1 = Map.insert k fld runMap
1397 in Partial $ fPartial doneMap runMap1
1399 let runMap1 = Map.delete k runMap
1400 doneMap1 = Map.insert k b doneMap
1401 if Map.size runMap1 == 0
1403 else Partial $ fPartial doneMap1 runMap1
1404 Done _ -> error "Bug: demuxDefaultWith: Done fold"
1406 step (DemuxMapAndDefault dacc doneMap runMap) a = do
1408 case Map.lookup k runMap of
1410 res <- dstep dacc (k, a1)
1414 Partial s -> DemuxMapAndDefault s doneMap runMap
1415 Done b -> DemuxOnlyMap b doneMap runMap
1418 (DemuxMapAndDefault dacc)
1419 (Partial . DemuxOnlyDefault dacc)
1420 doneMap runMap fld k a1
1422 step (DemuxOnlyMap dval doneMap runMap) a = do
1424 case Map.lookup k runMap of
1425 Nothing -> return $ Partial $ DemuxOnlyMap dval doneMap runMap
1430 doneMap runMap fld k a1
1431 step (DemuxOnlyDefault dacc doneMap) a = do
1433 res <- dstep dacc (k, a1)
1436 Partial s -> Partial $ DemuxOnlyDefault s doneMap
1437 Done b -> Done (doneMap, b)
1439 runExtract (Fold _ initial1 done1) = do
1442 Partial s -> done1 s
1445 extract (DemuxMapAndDefault dacc doneMap runMap) = do
1447 runMap1 <- Prelude.mapM runExtract runMap
1448 return (doneMap `Map.union` runMap1, b)
1449 extract (DemuxOnlyMap dval doneMap runMap) = do
1450 runMap1 <- Prelude.mapM runExtract runMap
1451 return (doneMap `Map.union` runMap1, dval)
1452 extract (DemuxOnlyDefault dacc doneMap) = do
1457-- > demuxDefault = demuxDefaultWith id
1460{-# INLINE demuxDefault #-}
1461demuxDefault :: (Monad m, Ord k)
1462 => Map k (Fold m a b) -> Fold m (k, a) b -> Fold m (k, a) (Map k b, b)
1463demuxDefault = demuxDefaultWith id
1465-- TODO If the data is large we may need a map/hashmap in pinned memory instead
1466-- of a regular Map. That may require a serializable constraint though. We can
1467-- have another API for that.
1469-- | Split the input stream based on a key field and fold each split using the
1470-- given fold. Useful for map/reduce, bucketizing the input in different bins
1471-- or for generating histograms.
1474-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
1475-- in Stream.fold (Fold.classifyWith fst (Fold.map snd Fold.toList)) input
1477-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
1479-- If the classifier fold stops for a particular key any further inputs in that
1480-- bucket are ignored.
1486{-# INLINE classifyWith #-}
1487classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1488classifyWith f (Fold step1 initial1 extract1) =
1489 rmapM extract $ foldlM' step initial
1493 initial = return Map.empty
1496 case Map.lookup k kv of
1503 $ flip (Map.insert k) kv
1505 Partial s1 -> Left' s1
1507 Done b -> return $ Map.insert k (Right' b) kv
1513 $ flip (Map.insert k) kv
1515 Partial s1 -> Left' s1
1517 Right' _ -> return kv
1526 Left' s -> extract1 s
1527 Right' b -> return b)
1529-- | Given an input stream of key value pairs and a fold for values, fold all
1530-- the values belonging to each key. Useful for map/reduce, bucketizing the
1531-- input in different bins or for generating histograms.
1534-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
1535-- in Stream.fold (Fold.classify Fold.toList) input
1537-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
1541-- > classify fld = Fold.classifyWith fst (lmap snd fld)
1544{-# INLINE classify #-}
1545classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1546classify fld = classifyWith fst (lmap snd fld)
1548------------------------------------------------------------------------------
1550------------------------------------------------------------------------------
1552{-# INLINE unzipWithMUsing #-}
1553unzipWithMUsing :: Monad m =>
1554 ( (x -> y -> (x, y))
1557 -> Fold m (b, c) (x, y)
1563unzipWithMUsing t f fld1 fld2 =
1564 let f1 = lmap fst fld1 -- :: Fold m (b, c) b
1565 f2 = lmap snd fld2 -- :: Fold m (b, c) c
1566 in lmapM f (t (,) f1 f2)
1568-- | Like 'unzipWith' but with a monadic splitter function.
1570-- @unzipWithM k f1 f2 = lmapM k (unzip f1 f2)@
1573{-# INLINE unzipWithM #-}
1574unzipWithM :: Monad m
1575 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1576unzipWithM = unzipWithMUsing teeWith
1578-- | Similar to 'unzipWithM' but terminates when the first fold terminates.
1581{-# INLINE unzipWithFstM #-}
1582unzipWithFstM :: Monad m =>
1583 (a -> m (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1584unzipWithFstM = unzipWithMUsing teeWithFst
1586-- | Similar to 'unzipWithM' but terminates when any fold terminates.
1589{-# INLINE unzipWithMinM #-}
1590unzipWithMinM :: Monad m =>
1591 (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1592unzipWithMinM = unzipWithMUsing teeWithMin
1594-- | Split elements in the input stream into two parts using a pure splitter
1595-- function, direct each part to a different fold and zip the results.
1597-- @unzipWith f fld1 fld2 = Fold.lmap f (Fold.unzip fld1 fld2)@
1599-- This fold terminates when both the input folds terminate.
1602{-# INLINE unzipWith #-}
1604 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1605unzipWith f = unzipWithM (return . f)
1607-- | Send the elements of tuples in a stream of tuples through two different
1612-- |-------Fold m a x--------|
1613-- ---------stream of (a,b)--| |----m (x,y)
1614-- |-------Fold m b y--------|
1618-- > unzip = Fold.unzipWith id
1620-- This is the consumer side dual of the producer side 'zip' operation.
1624unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1627------------------------------------------------------------------------------
1628-- Combining streams and folds - Zipping
1629------------------------------------------------------------------------------
1631-- | Zip a stream with the input of a fold using the supplied function.
1635{-# INLINE zipWithM #-}
1636zipWithM :: -- Monad m =>
1637 (a -> b -> m c) -> t m a -> Fold m c x -> Fold m b x
1640-- | Zip a stream with the input of a fold.
1645zip :: Monad m => t m a -> Fold m (a, b) x -> Fold m b x
1646zip = zipWithM (curry return)
1648-- | Pair each element of a fold input with its index, starting from index 0.
1651{-# INLINE indexed #-}
1652indexed :: -- forall m a b. Monad m =>
1653 Fold m (Int, a) b -> Fold m a b
1654indexed = undefined -- zip (Stream.enumerateFrom 0 :: SerialT m Int)
1656-- | Change the predicate function of a Fold from @a -> b@ to accept an
1657-- additional state input @(s, a) -> b@. Convenient to filter with an
1658-- addiitonal index or time input.
1661-- filterWithIndex = with indexed filter
1662-- filterWithAbsTime = with timestamped filter
1663-- filterWithRelTime = with timeIndexed filter
1669 (Fold m (s, a) b -> Fold m a b)
1670 -> (((s, a) -> c) -> Fold m (s, a) b -> Fold m (s, a) b)
1671 -> (((s, a) -> c) -> Fold m a b -> Fold m a b)
1672with f comb g = f . comb g . lmap snd
1674-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
1675-- then every element at strides of @stride@.
1678{-# INLINE sampleFromthen #-}
1679sampleFromthen :: Monad m => Int -> Int -> Fold m a b -> Fold m a b
1680sampleFromthen offset size =
1681 with indexed filter (\(i, _) -> (i + offset) `mod` size == 0)
1683------------------------------------------------------------------------------
1685------------------------------------------------------------------------------
1687-- | @concatSequence f t@ applies folds from stream @t@ sequentially and
1688-- collects the results using the fold @f@.
1692{-# INLINE concatSequence #-}
1695 Fold m b c -> t (Fold m a b) -> Fold m a c
1696concatSequence _f _p = undefined
1698-- | Group the input stream into groups of elements between @low@ and @high@.
1699-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1700-- @high@. Each chunk is folded using the provided fold function.
1702-- This could be useful, for example, when we are folding a stream of unknown
1703-- size to a stream of arrays and we want to minimize the number of
1706-- NOTE: this would be an application of "many" using a terminating fold.
1710{-# INLINE chunksBetween #-}
1711chunksBetween :: -- Monad m =>
1712 Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1713chunksBetween _low _high _f1 _f2 = undefined
1715-- | A fold that buffers its input to a pure stream.
1717-- /Warning!/ working on large streams accumulated as buffers in memory could
1718-- be very inefficient, consider using "Streamly.Data.Array" instead.
1720-- >>> toStream = fmap SerialT Fold.toStreamK
1723{-# INLINE toStream #-}
1724toStream :: Monad m => Fold m a (SerialT n a)
1725toStream = fmap SerialT toStreamK
1727-- This is more efficient than 'toStream'. toStream is exactly the same as
1728-- reversing the stream after toStreamRev.
1730-- | Buffers the input stream to a pure stream in the reverse order of the
1733-- >>> toStreamRev = fmap SerialT Fold.toStreamKRev
1735-- /Warning!/ working on large streams accumulated as buffers in memory could
1736-- be very inefficient, consider using "Streamly.Data.Array" instead.
1740-- xn : ... : x2 : x1 : []
1741{-# INLINABLE toStreamRev #-}
1742toStreamRev :: Monad m => Fold m a (SerialT n a)
1743toStreamRev = fmap SerialT toStreamKRev
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