1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ExistentialQuantification #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE ScopedTypeVariables #-}
10-- Module : Streamly.Internal.Data.Fold
11-- Copyright : (c) 2019 Composewell Technologies
12-- (c) 2013 Gabriel Gonzalez
14-- Maintainer : streamly@composewell.com
15-- Stability : experimental
18-- Also see the "Streamly.Internal.Data.Sink" module that provides specialized left folds
19-- that discard the outputs.
21-- IMPORTANT: keep the signatures consistent with the folds in Streamly.Prelude
23module Streamly.Internal.Data.Fold
34 -- * Fold Creation Utilities
61 -- ** Full Folds (Monoidal)
66 -- ** Full Folds (To Containers)
69 , toListRevF -- experimental
87 -- XXX these are slower than right folds even when full input is used
95 -- ** Covariant Operations
112 -- ** Mapping Filters
116 -- ** Scanning Filters
121 -- | Insertion adds more elements to the stream.
150 -- , splitIn -- sessionN
154 , break -- breakBefore
163 -- , breakOnStream -- on a stream
193 -- These can be expressed using lmap/lmapM and unzip
208 , toParallelSVarLimited
212import Control.Monad (void)
213import Control.Monad.IO.Class (MonadIO(..))
214import Data.Functor.Identity (Identity(..))
215import Data.Int (Int64)
216import Data.Map.Strict (Map)
219 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
220 foldl, map, mapM_, sequence, all, any, sum, product, elem,
221 notElem, maximum, minimum, head, last, tail, length, null,
222 reverse, iterate, init, and, or, lookup, foldr1, (!!),
223 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
224 span, splitAt, break, mapM)
226import qualified Data.Map.Strict as Map
227import qualified Prelude
229import Streamly.Internal.Data.Pipe.Types (Pipe (..), PipeState(..))
230import Streamly.Internal.Data.Fold.Types
231import Streamly.Internal.Data.Parse.Types
232import Streamly.Internal.Data.Strict
233import Streamly.Internal.Data.SVar
235import qualified Streamly.Internal.Data.Pipe.Types as Pipe
237------------------------------------------------------------------------------
239------------------------------------------------------------------------------
241-- | Make a fold using a pure step function, a pure initial state and
242-- a pure state extraction function.
247mkPure :: Monad m => (s -> a -> s) -> s -> (s -> b) -> Fold m a b
248mkPure step initial extract =
249 Fold (\s a -> return $ step s a) (return initial) (return . extract)
251-- | Make a fold using a pure step function and a pure initial state. The
252-- final state extracted is identical to the intermediate state.
256{-# INLINE mkPureId #-}
257mkPureId :: Monad m => (b -> a -> b) -> b -> Fold m a b
258mkPureId step initial = mkPure step initial id
260-- | Make a fold with an effectful step function and initial state, and a state
261-- extraction function.
265-- We can just use 'Fold' but it is provided for completeness.
270mkFold :: (s -> a -> m s) -> m s -> (s -> m b) -> Fold m a b
273-- | Make a fold with an effectful step function and initial state. The final
274-- state extracted is identical to the intermediate state.
278{-# INLINE mkFoldId #-}
279mkFoldId :: Monad m => (b -> a -> m b) -> m b -> Fold m a b
280mkFoldId step initial = Fold step initial return
282------------------------------------------------------------------------------
284------------------------------------------------------------------------------
286-- | Change the underlying monad of a fold
289hoist :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b
290hoist f (Fold step initial extract) =
291 Fold (\x a -> f $ step x a) (f initial) (f . extract)
293-- | Adapt a pure fold to any monad
295-- > generally = hoist (return . runIdentity)
298generally :: Monad m => Fold Identity a b -> Fold m a b
299generally = hoist (return . runIdentity)
301------------------------------------------------------------------------------
302-- Transformations on fold inputs
303------------------------------------------------------------------------------
305-- | Flatten the monadic output of a fold to pure output.
308{-# INLINE sequence #-}
309sequence :: Monad m => Fold m a (m b) -> Fold m a b
310sequence (Fold step initial extract) = Fold step initial extract'
316-- | Map a monadic function on the output of a fold.
320mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
321mapM f = sequence . fmap f
323------------------------------------------------------------------------------
324-- Transformations on fold inputs
325------------------------------------------------------------------------------
329-- | Apply a transformation on a 'Fold' using a 'Pipe'.
332{-# INLINE transform #-}
333transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
334transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
335 Fold step initial extract
339 initial = Tuple' <$> return pinitial <*> finitial
340 step (Tuple' ps fs) x = do
346 go acc (Pipe.Yield b (Consume ps')) = do
348 return (Tuple' ps' acc')
350 go acc (Pipe.Yield b (Produce ps')) = do
355 go acc (Pipe.Continue (Consume ps')) = return (Tuple' ps' acc)
357 go acc (Pipe.Continue (Produce ps')) = do
361 extract (Tuple' _ fs) = fextract fs
363------------------------------------------------------------------------------
365------------------------------------------------------------------------------
367-- | @_Fold1 step@ returns a new 'Fold' using just a step function that has the
368-- same type for the accumulator and the element. The result type is the
369-- accumulator type wrapped in 'Maybe'. The initial accumulator is retrieved
370-- from the 'Foldable', the result is 'None' for empty containers.
371{-# INLINABLE _Fold1 #-}
372_Fold1 :: Monad m => (a -> a -> a) -> Fold m a (Maybe a)
373_Fold1 step = Fold step_ (return Nothing') (return . toMaybe)
375 step_ mx a = return $ Just' $
380------------------------------------------------------------------------------
382------------------------------------------------------------------------------
384------------------------------------------------------------------------------
386------------------------------------------------------------------------------
388-- | A fold that drains all its input, running the effects and discarding the
392{-# INLINABLE drain #-}
393drain :: Monad m => Fold m a ()
394drain = Fold step begin done
401-- > drainBy f = lmapM f drain
403-- Drain all input after passing it through a monadic function. This is the
404-- dual of mapM_ on stream producers.
407{-# INLINABLE drainBy #-}
408drainBy :: Monad m => (a -> m b) -> Fold m a ()
409drainBy f = Fold (const (void . f)) (return ()) return
411{-# INLINABLE drainBy2 #-}
412drainBy2 :: Monad m => (a -> m b) -> Fold2 m c a ()
413drainBy2 f = Fold2 (const (void . f)) (\_ -> return ()) return
415-- | Extract the last element of the input stream, if any.
418{-# INLINABLE last #-}
419last :: Monad m => Fold m a (Maybe a)
420last = _Fold1 (flip const)
422------------------------------------------------------------------------------
424------------------------------------------------------------------------------
426-- | Like 'length', except with a more general 'Num' return value
429{-# INLINABLE genericLength #-}
430genericLength :: (Monad m, Num b) => Fold m a b
431genericLength = Fold (\n _ -> return $ n + 1) (return 0) return
433-- | Determine the length of the input stream.
436{-# INLINABLE length #-}
437length :: Monad m => Fold m a Int
438length = genericLength
440-- | Determine the sum of all elements of a stream of numbers. Returns additive
441-- identity (@0@) when the stream is empty. Note that this is not numerically
442-- stable for floating point numbers.
446sum :: (Monad m, Num a) => Fold m a a
447sum = Fold (\x a -> return $ x + a) (return 0) return
449-- | Determine the product of all elements of a stream of numbers. Returns
450-- multiplicative identity (@1@) when the stream is empty.
453{-# INLINABLE product #-}
454product :: (Monad m, Num a) => Fold m a a
455product = Fold (\x a -> return $ x * a) (return 1) return
457------------------------------------------------------------------------------
459------------------------------------------------------------------------------
461-- | Determine the maximum element in a stream using the supplied comparison
465{-# INLINABLE maximumBy #-}
466maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
467maximumBy cmp = _Fold1 max'
469 max' x y = case cmp x y of
475-- maximum = 'maximumBy' compare
478-- Determine the maximum element in a stream.
481{-# INLINABLE maximum #-}
482maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
485-- | Computes the minimum element with respect to the given comparison function
488{-# INLINABLE minimumBy #-}
489minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
490minimumBy cmp = _Fold1 min'
492 min' x y = case cmp x y of
496-- | Determine the minimum element in a stream using the supplied comparison
500{-# INLINABLE minimum #-}
501minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
504------------------------------------------------------------------------------
505-- To Summary (Statistical)
506------------------------------------------------------------------------------
508-- | Compute a numerically stable arithmetic mean of all elements in the input
512{-# INLINABLE mean #-}
513mean :: (Monad m, Fractional a) => Fold m a a
514mean = Fold step (return begin) (return . done)
517 step (Tuple' x n) y = return $
519 in Tuple' (x + (y - x) / n') n'
520 done (Tuple' x _) = x
522-- | Compute a numerically stable (population) variance over all elements in
526{-# INLINABLE variance #-}
527variance :: (Monad m, Fractional a) => Fold m a a
528variance = Fold step (return begin) (return . done)
530 begin = Tuple3' 0 0 0
532 step (Tuple3' n mean_ m2) x = return $ Tuple3' n' mean' m2'
535 mean' = (n * mean_ + x) / (n + 1)
537 m2' = m2 + delta * delta * n / (n + 1)
539 done (Tuple3' n _ m2) = m2 / n
541-- | Compute a numerically stable (population) standard deviation over all
542-- elements in the input stream.
545{-# INLINABLE stdDev #-}
546stdDev :: (Monad m, Floating a) => Fold m a a
547stdDev = sqrt variance
549-- | Compute an 'Int' sized polynomial rolling hash
551-- > H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0
553-- Where @c1@, @c2@, @cn@ are the elements in the input stream and @k@ is a
556-- This hash is often used in Rabin-Karp string search algorithm.
558-- See https://en.wikipedia.org/wiki/Rolling_hash
561{-# INLINABLE rollingHashWithSalt #-}
562rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64
563rollingHashWithSalt salt = Fold step initial extract
565 k = 2891336453 :: Int64
566 initial = return salt
567 step cksum a = return $ cksum * k + fromIntegral (fromEnum a)
570-- | A default salt used in the implementation of 'rollingHash'.
571{-# INLINE defaultSalt #-}
573defaultSalt = -2578643520546668380
575-- | Compute an 'Int' sized polynomial rolling hash of a stream.
577-- > rollingHash = rollingHashWithSalt defaultSalt
580{-# INLINABLE rollingHash #-}
581rollingHash :: (Monad m, Enum a) => Fold m a Int64
582rollingHash = rollingHashWithSalt defaultSalt
584-- | Compute an 'Int' sized polynomial rolling hash of the first n elements of
587-- > rollingHashFirstN = ltake n rollingHash
588{-# INLINABLE rollingHashFirstN #-}
589rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64
590rollingHashFirstN n = ltake n rollingHash
592------------------------------------------------------------------------------
593-- Monoidal left folds
594------------------------------------------------------------------------------
596-- | Fold an input stream consisting of monoidal elements using 'mappend'
599-- > S.fold FL.mconcat (S.map Sum $ S.enumerateFromTo 1 10)
602{-# INLINABLE mconcat #-}
603mconcat :: (Monad m, Monoid a) => Fold m a a
604mconcat = Fold (\x a -> return $ mappend x a) (return mempty) return
607-- > foldMap f = map f mconcat
609-- Make a fold from a pure function that folds the output of the function
610-- using 'mappend' and 'mempty'.
612-- > S.fold (FL.foldMap Sum) $ S.enumerateFromTo 1 10
615{-# INLINABLE foldMap #-}
616foldMap :: (Monad m, Monoid b) => (a -> b) -> Fold m a b
617foldMap f = lmap f mconcat
620-- > foldMapM f = mapM f mconcat
622-- Make a fold from a monadic function that folds the output of the function
623-- using 'mappend' and 'mempty'.
625-- > S.fold (FL.foldMapM (return . Sum)) $ S.enumerateFromTo 1 10
628{-# INLINABLE foldMapM #-}
629foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
630foldMapM act = Fold step begin done
633 begin = return mempty
636 return $! mappend m m'
638------------------------------------------------------------------------------
640------------------------------------------------------------------------------
642-- | Folds the input stream to a list.
644-- /Warning!/ working on large lists accumulated as buffers in memory could be
645-- very inefficient, consider using "Streamly.Memory.Array" instead.
649-- id . (x1 :) . (x2 :) . (x3 :) . ... . (xn :) $ []
650{-# INLINABLE toList #-}
651toList :: Monad m => Fold m a [a]
652toList = Fold (\f x -> return $ f . (x :))
656------------------------------------------------------------------------------
658------------------------------------------------------------------------------
660-- | A fold that drains the first n elements of its input, running the effects
661-- and discarding the results.
662{-# INLINABLE drainN #-}
663drainN :: Monad m => Int -> Fold m a ()
664drainN n = ltake n drain
666-- | A fold that drains elements of its input as long as the predicate succeeds,
667-- running the effects and discarding the results.
668{-# INLINABLE drainWhile #-}
669drainWhile :: Monad m => (a -> Bool) -> Fold m a ()
670drainWhile p = ltakeWhile p drain
672------------------------------------------------------------------------------
674------------------------------------------------------------------------------
676-- | Like 'index', except with a more general 'Integral' argument
679{-# INLINABLE genericIndex #-}
680genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
681genericIndex i = Fold step (return $ Left' 0) done
694-- | Lookup the element at the given index.
697{-# INLINABLE index #-}
698index :: Monad m => Int -> Fold m a (Maybe a)
701-- | Extract the first element of the stream, if any.
704{-# INLINABLE head #-}
705head :: Monad m => Fold m a (Maybe a)
708-- | Returns the first element that satisfies the given predicate.
711{-# INLINABLE find #-}
712find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
713find predicate = Fold step (return Nothing') (return . toMaybe)
717 Nothing' -> if predicate a
722-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
723-- first pair where the key equals the given value @a@.
726{-# INLINABLE lookup #-}
727lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
728lookup a0 = Fold step (return Nothing') (return . toMaybe)
730 step x (a,b) = return $
732 Nothing' -> if a == a0
737-- | Convert strict 'Either'' to lazy 'Maybe'
738{-# INLINABLE hush #-}
739hush :: Either' a b -> Maybe b
740hush (Left' _) = Nothing
741hush (Right' b) = Just b
743-- | Returns the first index that satisfies the given predicate.
746{-# INLINABLE findIndex #-}
747findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
748findIndex predicate = Fold step (return $ Left' 0) (return . hush)
758-- | Returns the first index where a given value is found in the stream.
761{-# INLINABLE elemIndex #-}
762elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
763elemIndex a = findIndex (a ==)
765------------------------------------------------------------------------------
767------------------------------------------------------------------------------
769-- | Return 'True' if the input stream is empty.
772{-# INLINABLE null #-}
773null :: Monad m => Fold m a Bool
774null = Fold (\_ _ -> return False) (return True) return
777-- > any p = lmap p or
779-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
783any :: Monad m => (a -> Bool) -> Fold m a Bool
784any predicate = Fold (\x a -> return $ x || predicate a) (return False) return
786-- | Return 'True' if the given element is present in the stream.
789{-# INLINABLE elem #-}
790elem :: (Eq a, Monad m) => a -> Fold m a Bool
794-- > all p = lmap p and
796-- | Returns 'True' if all elements of a stream satisfy a predicate.
800all :: Monad m => (a -> Bool) -> Fold m a Bool
801all predicate = Fold (\x a -> return $ x && predicate a) (return True) return
803-- | Returns 'True' if the given element is not present in the stream.
806{-# INLINABLE notElem #-}
807notElem :: (Eq a, Monad m) => a -> Fold m a Bool
808notElem a = all (a /=)
810-- | Returns 'True' if all elements are 'True', 'False' otherwise
814and :: Monad m => Fold m Bool Bool
815and = Fold (\x a -> return $ x && a) (return True) return
817-- | Returns 'True' if any element is 'True', 'False' otherwise
821or :: Monad m => Fold m Bool Bool
822or = Fold (\x a -> return $ x || a) (return False) return
824------------------------------------------------------------------------------
826------------------------------------------------------------------------------
828------------------------------------------------------------------------------
829-- Grouping without looking at elements
830------------------------------------------------------------------------------
832-- XXX Once we have terminating folds, this Parse should get replaced by Fold.
833-- Alternatively, we can name it "chunkOf" and the corresponding time domain
834-- combinator as "intervalOf" or even "chunk" and "interval".
836-- | Take at most @n@ input elements and fold them using the supplied fold.
838-- Stops after @n@ elements.
844take :: Monad m => Int -> Fold m a b -> Parse m a b
845take n (Fold fstep finitial fextract) = Parse step initial extract
849 initial = (Tuple' 0) <$> finitial
851 step (Tuple' i r) a = do
856 then return $ Keep 0 s1
857 else return $ Halt 0 s1
859 extract (Tuple' _ r) = fmap Right (fextract r)
861------------------------------------------------------------------------------
863------------------------------------------------------------------------------
865-- XXX These would just be applicative compositions of terminating folds.
867-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
868-- elements of its input are consumed by fold @f1@ and the rest of the stream
869-- is consumed by fold @f2@.
871-- > let splitAt_ n xs = S.fold (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
873-- >>> splitAt_ 6 "Hello World!"
874-- > ("Hello ","World!")
876-- >>> splitAt_ (-1) [1,2,3]
879-- >>> splitAt_ 0 [1,2,3]
882-- >>> splitAt_ 1 [1,2,3]
885-- >>> splitAt_ 3 [1,2,3]
888-- >>> splitAt_ 4 [1,2,3]
893-- This can be considered as a two-fold version of 'ltake' where we take both
894-- the segments instead of discarding the leftover.
896{-# INLINE splitAt #-}
903splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
904 Fold step initial extract
906 initial = Tuple3' <$> return n <*> initialL <*> initialR
908 step (Tuple3' i xL xR) input =
910 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
911 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
913 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
915------------------------------------------------------------------------------
917------------------------------------------------------------------------------
919------------------------------------------------------------------------------
921------------------------------------------------------------------------------
923-- | Break the input stream into two groups, the first group takes the input as
924-- long as the predicate applied to the first element of the stream and next
925-- input element holds 'True', the second group takes the rest of the input.
935spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
936 Fold step initial extract
939 initial = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
941 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
942 if cmp frst input && isFirstG
944 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
946 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
948 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
951 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
953 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
955 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
957-- | @span p f1 f2@ composes folds @f1@ and @f2@ such that @f1@ consumes the
958-- input as long as the predicate @p@ is 'True'. @f2@ consumes the rest of the
961-- > let span_ p xs = S.fold (S.span p FL.toList FL.toList) $ S.fromList xs
963-- >>> span_ (< 1) [1,2,3]
966-- >>> span_ (< 2) [1,2,3]
969-- >>> span_ (< 4) [1,2,3]
974-- This can be considered as a two-fold version of 'ltakeWhile' where we take
975-- both the segments instead of discarding the leftover.
983span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
984 Fold step initial extract
988 initial = Tuple3' <$> initialL <*> initialR <*> return True
990 step (Tuple3' a b isFirstG) input =
991 if isFirstG && p input
992 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
993 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
995 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
998-- > break p = span (not . p)
1000-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
1001-- folds @f1@ and @f2@ such that @f1@ stops consuming input as soon as the
1002-- predicate @p@ becomes 'True'. The rest of the input is consumed @f2@.
1004-- This is the binary version of 'splitBy'.
1006-- > let break_ p xs = S.fold (S.break p FL.toList FL.toList) $ S.fromList xs
1008-- >>> break_ (< 1) [3,2,1]
1011-- >>> break_ (< 2) [3,2,1]
1014-- >>> break_ (< 4) [3,2,1]
1025break p = span (not . p)
1027-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
1028-- predicate is applied to the previous and the next input elements.
1031{-# INLINE spanByRolling #-}
1038spanByRolling cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1039 Fold step initial extract
1042 initial = Tuple3' <$> initialL <*> initialR <*> return Nothing
1044 step (Tuple3' a b (Just frst)) input =
1046 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
1047 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
1049 step (Tuple3' a b Nothing) input =
1050 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
1052 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1054------------------------------------------------------------------------------
1055-- Binary splitting on a separator
1056------------------------------------------------------------------------------
1059-- | Find the first occurrence of the specified sequence in the input stream
1060-- and break the input stream into two parts, the first part consisting of the
1061-- stream before the sequence and the second part consisting of the sequence
1062-- and the rest of the stream.
1064-- > let breakOn_ pat xs = S.fold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
1066-- >>> breakOn_ "dear" "Hello dear world!"
1067-- > ("Hello ","dear world!")
1069{-# INLINE breakOn #-}
1070breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
1071breakOn pat f m = undefined
1074------------------------------------------------------------------------------
1076------------------------------------------------------------------------------
1078-- | Distribute one copy of the stream to each fold and zip the results.
1081-- |-------Fold m a b--------|
1082-- ---stream m a---| |---m (b,c)
1083-- |-------Fold m a c--------|
1085-- >>> S.fold (FL.tee FL.sum FL.length) (S.enumerateFromTo 1.0 100.0)
1090tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1091tee f1 f2 = (,) <$> f1 <*> f2
1093{-# INLINE foldNil #-}
1094foldNil :: Monad m => Fold m a [b]
1095foldNil = Fold step begin done where
1097 step _ _ = return []
1100{-# INLINE foldCons #-}
1101foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]
1102foldCons (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1103 Fold step begin done
1107 begin = Tuple' <$> beginL <*> beginR
1108 step (Tuple' xL xR) a = Tuple' <$> stepL xL a <*> stepR xR a
1109 done (Tuple' xL xR) = (:) <$> (doneL xL) <*> (doneR xR)
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-- >>> S.fold (FL.distribute [FL.sum, FL.length]) (S.enumerateFromTo 1 5)
1130-- This is the consumer side dual of the producer side 'sequence' operation.
1133{-# INLINE distribute #-}
1134distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1135distribute [] = foldNil
1136distribute (x:xs) = foldCons x (distribute xs)
1138-- | Like 'distribute' but for folds that return (), this can be more efficient
1139-- than 'distribute' as it does not need to maintain state.
1141{-# INLINE distribute_ #-}
1142distribute_ :: Monad m => [Fold m a ()] -> Fold m a ()
1143distribute_ fs = Fold step initial extract
1145 initial = Prelude.mapM (\(Fold s i e) ->
1146 i >>= \r -> return (Fold s (return r) e)) fs
1148 Prelude.mapM_ (\(Fold s i _) -> i >>= \r -> s r a >> return ()) ss
1151 Prelude.mapM_ (\(Fold _ i e) -> i >>= \r -> e r) ss
1154------------------------------------------------------------------------------
1156------------------------------------------------------------------------------
1158-- | Partition the input over two folds using an 'Either' partitioning
1163-- |-------Fold b x--------|
1164-- -----stream m a --> (Either b c)----| |----(x,y)
1165-- |-------Fold c y--------|
1168-- Send input to either fold randomly:
1170-- >>> import System.Random (randomIO)
1171-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
1172-- >>> S.fold (FL.partitionByM randomly FL.length FL.length) (S.enumerateFromTo 1 100)
1175-- Send input to the two folds in a proportion of 2:1:
1178-- import Data.IORef (newIORef, readIORef, writeIORef)
1179-- proportionately m n = do
1180-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1181-- return $ \\a -> do
1182-- r <- readIORef ref
1183-- writeIORef ref $ tail r
1187-- f <- proportionately 2 1
1188-- r <- S.fold (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1195-- This is the consumer side dual of the producer side 'mergeBy' operation.
1198{-# INLINE partitionByM #-}
1199partitionByM :: Monad m
1200 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1201partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1203 Fold step begin done
1207 begin = Tuple' <$> beginL <*> beginR
1208 step (Tuple' xL xR) a = do
1211 Left b -> Tuple' <$> stepL xL b <*> return xR
1212 Right c -> Tuple' <$> return xL <*> stepR xR c
1213 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1215-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1216-- makes the signature clearer as to which case belongs to which fold.
1217-- XXX need to check the performance in both cases.
1219-- | Same as 'partitionByM' but with a pure partition function.
1221-- Count even and odd numbers in a stream:
1224-- >>> let f = FL.partitionBy (\\n -> if even n then Left n else Right n)
1225-- (fmap (("Even " ++) . show) FL.length)
1226-- (fmap (("Odd " ++) . show) FL.length)
1227-- in S.fold f (S.enumerateFromTo 1 100)
1228-- ("Even 50","Odd 50")
1232{-# INLINE partitionBy #-}
1233partitionBy :: Monad m
1234 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1235partitionBy f = partitionByM (return . f)
1237-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1238-- and routes the 'Left' values to the first fold and 'Right' values to the
1241-- > partition = partitionBy id
1244{-# INLINE partition #-}
1246 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1247partition = partitionBy id
1250-- | Send one item to each fold in a round-robin fashion. This is the consumer
1251-- side dual of producer side 'mergeN' operation.
1253-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1254-- partitionN fs = Fold step begin done
1257-- TODO Demultiplex an input element into a number of typed variants. We want
1258-- to statically restrict the target values within a set of predefined types,
1259-- an enumeration of a GADT. We also want to make sure that the Map contains
1260-- only those types and the full set of those types.
1262-- TODO Instead of the input Map it should probably be a lookup-table using an
1263-- array and not in GC memory. The same applies to the output Map as well.
1264-- However, that would only be helpful if we have a very large data structure,
1265-- need to measure and see how it scales.
1267-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1270-- | Split the input stream based on a key field and fold each split using a
1271-- specific fold collecting the results in a map from the keys to the results.
1272-- Useful for cases like protocol handlers to handle different type of packets
1273-- using different handlers.
1277-- |-------Fold m a b
1278-- -----stream m a-----Map-----|
1279-- |-------Fold m a b
1285{-# INLINE demuxWith #-}
1286demuxWith :: (Monad m, Ord k)
1287 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a (Map k b)
1288demuxWith f kv = Fold step initial extract
1293-- alterF is available only since containers version 0.5.8.2
1294#if MIN_VERSION_containers(0,5,8)
1295 step mp a = case f a of
1296 (k, a') -> Map.alterF twiddle k mp
1297 -- XXX should we raise an exception in Nothing case?
1298 -- Ideally we should enforce that it is a total map over k so that look
1300 -- XXX we could use a monadic update function for a single lookup and
1301 -- update in the map.
1303 twiddle Nothing = pure Nothing
1304 twiddle (Just (Fold step' acc extract')) = do
1305 !r <- acc >>= \x -> step' x a'
1306 pure . Just $ Fold step' (return r) extract'
1310 in case Map.lookup k mp of
1311 Nothing -> return mp
1312 Just (Fold step' acc extract') -> do
1313 !r <- acc >>= \x -> step' x a'
1314 return $ Map.insert k (Fold step' (return r) extract') mp
1316 extract = Prelude.mapM (\(Fold _ acc e) -> acc >>= e)
1318-- | Fold a stream of key value pairs using a map of specific folds for each
1319-- key into a map from keys to the results of fold outputs of the corresponding
1323-- > let table = Data.Map.fromList [(\"SUM", FL.sum), (\"PRODUCT", FL.product)]
1324-- input = S.fromList [(\"SUM",1),(\"PRODUCT",2),(\"SUM",3),(\"PRODUCT",4)]
1325-- in S.fold (FL.demux table) input
1326-- fromList [("PRODUCT",8),("SUM",4)]
1331demux :: (Monad m, Ord k)
1332 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1335{-# INLINE demuxWithDefault_ #-}
1336demuxWithDefault_ :: (Monad m, Ord k)
1337 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m (k, a') b -> Fold m a ()
1338demuxWithDefault_ f kv (Fold dstep dinitial dextract) =
1339 Fold step initial extract
1343 initFold (Fold s i e) = i >>= \r -> return (Fold s (return r) e)
1345 mp <- Prelude.mapM initFold kv
1347 return (Tuple' mp dacc)
1348 step (Tuple' mp dacc) a
1350 = case Map.lookup k mp of
1352 acc <- dstep dacc (k, a')
1353 return (Tuple' mp acc)
1354 Just (Fold step' acc _) -> do
1355 _ <- acc >>= \x -> step' x a'
1356 return (Tuple' mp dacc)
1357 extract (Tuple' mp dacc) = do
1358 void $ dextract dacc
1359 Prelude.mapM_ (\(Fold _ acc e) -> acc >>= e) mp
1361-- | Split the input stream based on a key field and fold each split using a
1362-- specific fold without collecting the results. Useful for cases like protocol
1363-- handlers to handle different type of packets.
1367-- |-------Fold m a ()
1368-- -----stream m a-----Map-----|
1369-- |-------Fold m a ()
1377-- demuxWith_ can be slightly faster than demuxWith because we do not need to
1378-- update the Map in this case. This may be significant only if the map is
1380{-# INLINE demuxWith_ #-}
1381demuxWith_ :: (Monad m, Ord k)
1382 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a ()
1383demuxWith_ f kv = Fold step initial extract
1388 Prelude.mapM (\(Fold s i e) ->
1389 i >>= \r -> return (Fold s (return r) e)) kv
1391 -- XXX should we raise an exception in Nothing case?
1392 -- Ideally we should enforce that it is a total map over k so that look
1395 = case Map.lookup k mp of
1396 Nothing -> return mp
1397 Just (Fold step' acc _) -> do
1398 _ <- acc >>= \x -> step' x a'
1400 extract mp = Prelude.mapM_ (\(Fold _ acc e) -> acc >>= e) mp
1402-- | Given a stream of key value pairs and a map from keys to folds, fold the
1403-- values for each key using the corresponding folds, discarding the outputs.
1406-- > let prn = FL.drainBy print
1407-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]
1408-- input = S.fromList [(\"ONE",1),(\"TWO",2)]
1409-- in S.fold (FL.demux_ table) input
1415{-# INLINE demux_ #-}
1416demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()
1417demux_ = demuxWith_ id
1419{-# INLINE demuxDefault_ #-}
1420demuxDefault_ :: (Monad m, Ord k)
1421 => Map k (Fold m a ()) -> Fold m (k, a) () -> Fold m (k, a) ()
1422demuxDefault_ = demuxWithDefault_ id
1424-- TODO If the data is large we may need a map/hashmap in pinned memory instead
1425-- of a regular Map. That may require a serializable constraint though. We can
1426-- have another API for that.
1428-- | Split the input stream based on a key field and fold each split using the
1429-- given fold. Useful for map/reduce, bucketizing the input in different bins
1430-- or for generating histograms.
1433-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1434-- in S.fold (FL.classify FL.toList) input
1435-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1439{-# INLINE classifyWith #-}
1440classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1441classifyWith f (Fold step initial extract) = Fold step' initial' extract'
1445 initial' = return Map.empty
1448 in case Map.lookup k kv of
1452 return $ Map.insert k r kv
1455 return $ Map.insert k r kv
1456 extract' = Prelude.mapM extract
1458-- | Given an input stream of key value pairs and a fold for values, fold all
1459-- the values belonging to each key. Useful for map/reduce, bucketizing the
1460-- input in different bins or for generating histograms.
1463-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1464-- in S.fold (FL.classify FL.toList) input
1465-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1472-- > classify fld = classifyWith fst (lmap snd fld)
1474{-# INLINE classify #-}
1475classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1476classify fld = classifyWith fst (lmap snd fld)
1478------------------------------------------------------------------------------
1480------------------------------------------------------------------------------
1482-- | Like 'unzipWith' but with a monadic splitter function.
1485{-# INLINE unzipWithM #-}
1486unzipWithM :: Monad m
1487 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1488unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1489 Fold step begin done
1493 step (Tuple' xL xR) a = do
1495 Tuple' <$> stepL xL b <*> stepR xR c
1496 begin = Tuple' <$> beginL <*> beginR
1497 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1499-- | Split elements in the input stream into two parts using a pure splitter
1500-- function, direct each part to a different fold and zip the results.
1503{-# INLINE unzipWith #-}
1505 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1506unzipWith f = unzipWithM (return . f)
1508-- | Send the elements of tuples in a stream of tuples through two different
1513-- |-------Fold m a x--------|
1514-- ---------stream of (a,b)--| |----m (x,y)
1515-- |-------Fold m b y--------|
1519-- This is the consumer side dual of the producer side 'zip' operation.
1523unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1526------------------------------------------------------------------------------
1528------------------------------------------------------------------------------
1531-- All the stream flattening transformations can also be applied to a fold
1534-- | This can be used to apply all the stream generation operations on folds.
1535lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
1538lconcatMap s f1 f2 = undefined
1541-- All the grouping transformation that we apply to a stream can also be
1542-- applied to a fold input stream. groupBy et al can be written as terminating
1543-- folds and then we can apply foldChunks to use those repeatedly on a stream.
1545-- | Apply a terminating fold repeatedly to the input of another fold.
1547-- Compare with: Streamly.Prelude.concatMap, Streamly.Prelude.foldChunks
1551{-# INLINABLE foldChunks #-}
1554 Fold m a b -> Fold m b c -> Fold m a c
1555foldChunks = undefined
1558-- XXX this would be an application of foldChunks using a terminating fold.
1560-- | Group the input stream into groups of elements between @low@ and @high@.
1561-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1562-- @high@. Each chunk is folded using the provided fold function.
1564-- This could be useful, for example, when we are folding a stream of unknown
1565-- size to a stream of arrays and we want to minimize the number of
1570-- XXX we should be able to implement it with parsers/terminating folds.
1572{-# INLINE lchunksInRange #-}
1573lchunksInRange :: Monad m
1574 => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1575lchunksInRange low high (Fold step1 initial1 extract1)
1576 (Fold step2 initial2 extract2) = undefined
1579------------------------------------------------------------------------------
1580-- Fold to a Parallel SVar
1581------------------------------------------------------------------------------
1583{-# INLINE toParallelSVar #-}
1584toParallelSVar :: MonadIO m => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
1585toParallelSVar svar winfo = Fold step initial extract
1590 step () x = liftIO $ do
1591 -- XXX we can have a separate fold for unlimited buffer case to avoid a
1592 -- branch in the step here.
1593 decrementBufferLimit svar
1594 void $ send svar (ChildYield x)
1596 extract () = liftIO $ do
1599{-# INLINE toParallelSVarLimited #-}
1600toParallelSVarLimited :: MonadIO m
1601 => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
1602toParallelSVarLimited svar winfo = Fold step initial extract
1605 initial = return True
1607 step True x = liftIO $ do
1608 yieldLimitOk <- decrementYieldLimit svar
1611 decrementBufferLimit svar
1612 void $ send svar (ChildYield x)
1615 cleanupSVarFromWorker svar
1618 step False _ = return False
1620 extract True = liftIO $ sendStop svar winfo
1621 extract False = return ()
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