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-- Also see the "Streamly.Internal.Data.Sink" module that provides specialized left folds
11-- that discard the outputs.
13-- IMPORTANT: keep the signatures consistent with the folds in Streamly.Prelude
15module Streamly.Internal.Data.Fold
26 -- * Fold Creation Utilities
56 -- ** Full Folds (Monoidal)
61 -- ** Full Folds (To Containers)
64 , toListRevF -- experimental
82 -- XXX these are slower than right folds even when full input is used
90 -- ** Covariant Operations
107 -- ** Mapping Filters
111 -- ** Scanning Filters
116 -- | Insertion adds more elements to the stream.
144 -- , splitIn -- sessionN
148 , break -- breakBefore
157 -- , breakOnStream -- on a stream
187 -- These can be expressed using lmap/lmapM and unzip
202 , toParallelSVarLimited
206import Control.Monad (join, void)
207import Control.Monad.IO.Class (MonadIO(..))
208import Data.Functor.Identity (Identity(..))
209import Data.Int (Int64)
210import Data.Map.Strict (Map)
211#if __GLASGOW_HASKELL__ < 804
212import Data.Semigroup (Semigroup((<>)))
214import Streamly.Internal.Data.Pipe.Types (Pipe (..), PipeState(..))
215import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
216import Streamly.Internal.Data.Either.Strict (Either'(..))
217import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
219import qualified Streamly.Internal.Data.Pipe.Types as Pipe
220import qualified Data.Map.Strict as Map
221import qualified Prelude
224 ( filter, drop, dropWhile, take, takeWhile, zipWith
225 , foldl, map, mapM_, sequence, all, any, sum, product, elem
226 , notElem, maximum, minimum, head, last, tail, length, null
227 , reverse, iterate, init, and, or, lookup, (!!)
228 , scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip
229 , span, splitAt, break, mapM)
230import Streamly.Internal.Data.SVar
231import Streamly.Internal.Data.Fold.Types
233------------------------------------------------------------------------------
235------------------------------------------------------------------------------
237-- | Make a fold using a pure step function, a pure initial state and
238-- a pure state extraction function.
243mkPure :: Monad m => (s -> a -> s) -> s -> (s -> b) -> Fold m a b
244mkPure step initial extract =
245 Fold (\s a -> return $ step s a) (return initial) (return . extract)
247-- | Make a fold using a pure step function and a pure initial state. The
248-- final state extracted is identical to the intermediate state.
252{-# INLINE mkPureId #-}
253mkPureId :: Monad m => (b -> a -> b) -> b -> Fold m a b
254mkPureId step initial = mkPure step initial id
256-- | Make a fold with an effectful step function and initial state, and a state
257-- extraction function.
261-- We can just use 'Fold' but it is provided for completeness.
266mkFold :: (s -> a -> m s) -> m s -> (s -> m b) -> Fold m a b
269-- | Make a fold with an effectful step function and initial state. The final
270-- state extracted is identical to the intermediate state.
274{-# INLINE mkFoldId #-}
275mkFoldId :: Monad m => (b -> a -> m b) -> m b -> Fold m a b
276mkFoldId step initial = Fold step initial return
278------------------------------------------------------------------------------
280------------------------------------------------------------------------------
282-- | Change the underlying monad of a fold
285hoist :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b
286hoist f (Fold step initial extract) =
287 Fold (\x a -> f $ step x a) (f initial) (f . extract)
289-- | Adapt a pure fold to any monad
291-- > generally = hoist (return . runIdentity)
294generally :: Monad m => Fold Identity a b -> Fold m a b
295generally = hoist (return . runIdentity)
297------------------------------------------------------------------------------
298-- Transformations on fold inputs
299------------------------------------------------------------------------------
301-- | Flatten the monadic output of a fold to pure output.
304{-# INLINE sequence #-}
305sequence :: Monad m => Fold m a (m b) -> Fold m a b
306sequence (Fold step initial extract) = Fold step initial extract'
308 extract' x = join (extract x)
310-- | Map a monadic function on the output of a fold.
314mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
315mapM f = sequence . fmap f
317------------------------------------------------------------------------------
318-- Transformations on fold inputs
319------------------------------------------------------------------------------
323-- | Apply a transformation on a 'Fold' using a 'Pipe'.
326{-# INLINE transform #-}
327transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
328transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
329 Fold step initial extract
333 initial = Tuple' pinitial <$> finitial
334 step (Tuple' ps fs) x = do
340 go acc (Pipe.Yield b (Consume ps')) = do
342 return (Tuple' ps' acc')
344 go acc (Pipe.Yield b (Produce ps')) = do
349 go acc (Pipe.Continue (Consume ps')) = return (Tuple' ps' acc)
351 go acc (Pipe.Continue (Produce ps')) = do
355 extract (Tuple' _ fs) = fextract fs
357------------------------------------------------------------------------------
359------------------------------------------------------------------------------
361-- | @_Fold1 step@ returns a new 'Fold' using just a step function that has the
362-- same type for the accumulator and the element. The result type is the
363-- accumulator type wrapped in 'Maybe'. The initial accumulator is retrieved
364-- from the 'Foldable', the result is 'None' for empty containers.
365{-# INLINABLE _Fold1 #-}
366_Fold1 :: Monad m => (a -> a -> a) -> Fold m a (Maybe a)
367_Fold1 step = Fold step_ (return Nothing') (return . toMaybe)
369 step_ mx a = return $ Just' $
374------------------------------------------------------------------------------
376------------------------------------------------------------------------------
378------------------------------------------------------------------------------
380------------------------------------------------------------------------------
382-- | A fold that drains all its input, running the effects and discarding the
385-- > drain = drainBy (const (return ()))
388{-# INLINABLE drain #-}
389drain :: Monad m => Fold m a ()
390drain = Fold step begin done
397-- > drainBy f = lmapM f drain
398-- > drainBy = FL.foldMapM (void . f)
400-- Drain all input after passing it through a monadic function. This is the
401-- dual of mapM_ on stream producers.
404{-# INLINABLE drainBy #-}
405drainBy :: Monad m => (a -> m b) -> Fold m a ()
406drainBy f = Fold (const (void . f)) (return ()) return
408{-# INLINABLE drainBy2 #-}
409drainBy2 :: Monad m => (a -> m b) -> Fold2 m c a ()
410drainBy2 f = Fold2 (const (void . f)) (\_ -> return ()) return
412-- | Extract the last element of the input stream, if any.
414-- > last = fmap getLast $ FL.foldMap (Last . Just)
417{-# INLINABLE last #-}
418last :: Monad m => Fold m a (Maybe a)
419last = _Fold1 (flip const)
421------------------------------------------------------------------------------
423------------------------------------------------------------------------------
425-- | Like 'length', except with a more general 'Num' return value
427-- > genericLength = fmap getSum $ foldMap (Sum . const 1)
430{-# INLINABLE genericLength #-}
431genericLength :: (Monad m, Num b) => Fold m a b
432genericLength = Fold (\n _ -> return $ n + 1) (return 0) return
434-- | Determine the length of the input stream.
436-- > length = fmap getSum $ foldMap (Sum . const 1)
439{-# INLINABLE length #-}
440length :: Monad m => Fold m a Int
441length = genericLength
443-- | Determine the sum of all elements of a stream of numbers. Returns additive
444-- identity (@0@) when the stream is empty. Note that this is not numerically
445-- stable for floating point numbers.
447-- > sum = fmap getSum $ FL.foldMap Sum
451sum :: (Monad m, Num a) => Fold m a a
452sum = Fold (\x a -> return $ x + a) (return 0) return
454-- | Determine the product of all elements of a stream of numbers. Returns
455-- multiplicative identity (@1@) when the stream is empty.
457-- > product = fmap getProduct $ FL.foldMap Product
460{-# INLINABLE product #-}
461product :: (Monad m, Num a) => Fold m a a
462product = Fold (\x a -> return $ x * a) (return 1) return
464------------------------------------------------------------------------------
466------------------------------------------------------------------------------
468-- | Determine the maximum element in a stream using the supplied comparison
472{-# INLINABLE maximumBy #-}
473maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
474maximumBy cmp = _Fold1 max'
476 max' x y = case cmp x y of
482-- maximum = 'maximumBy' compare
485-- Determine the maximum element in a stream.
487-- Compare with @FL.foldMap Max@.
490{-# INLINABLE maximum #-}
491maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
494-- | Computes the minimum element with respect to the given comparison function
497{-# INLINABLE minimumBy #-}
498minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
499minimumBy cmp = _Fold1 min'
501 min' x y = case cmp x y of
505-- | Determine the minimum element in a stream using the supplied comparison
509-- minimum = 'minimumBy' compare
512-- Compare with @FL.foldMap Min@.
515{-# INLINABLE minimum #-}
516minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
519------------------------------------------------------------------------------
520-- To Summary (Statistical)
521------------------------------------------------------------------------------
523-- | Compute a numerically stable arithmetic mean of all elements in the input
527{-# INLINABLE mean #-}
528mean :: (Monad m, Fractional a) => Fold m a a
529mean = Fold step (return begin) (return . done)
532 step (Tuple' x n) y = return $
534 in Tuple' (x + (y - x) / n') n'
535 done (Tuple' x _) = x
537-- | Compute a numerically stable (population) variance over all elements in
541{-# INLINABLE variance #-}
542variance :: (Monad m, Fractional a) => Fold m a a
543variance = Fold step (return begin) (return . done)
545 begin = Tuple3' 0 0 0
547 step (Tuple3' n mean_ m2) x = return $ Tuple3' n' mean' m2'
550 mean' = (n * mean_ + x) / (n + 1)
552 m2' = m2 + delta * delta * n / (n + 1)
554 done (Tuple3' n _ m2) = m2 / n
556-- | Compute a numerically stable (population) standard deviation over all
557-- elements in the input stream.
560{-# INLINABLE stdDev #-}
561stdDev :: (Monad m, Floating a) => Fold m a a
562stdDev = sqrt variance
564-- | Compute an 'Int' sized polynomial rolling hash
566-- > H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0
568-- Where @c1@, @c2@, @cn@ are the elements in the input stream and @k@ is a
571-- This hash is often used in Rabin-Karp string search algorithm.
573-- See https://en.wikipedia.org/wiki/Rolling_hash
576{-# INLINABLE rollingHashWithSalt #-}
577rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64
578rollingHashWithSalt salt = Fold step initial extract
580 k = 2891336453 :: Int64
581 initial = return salt
582 step cksum a = return $ cksum * k + fromIntegral (fromEnum a)
585-- | A default salt used in the implementation of 'rollingHash'.
586{-# INLINE defaultSalt #-}
588defaultSalt = -2578643520546668380
590-- | Compute an 'Int' sized polynomial rolling hash of a stream.
592-- > rollingHash = rollingHashWithSalt defaultSalt
595{-# INLINABLE rollingHash #-}
596rollingHash :: (Monad m, Enum a) => Fold m a Int64
597rollingHash = rollingHashWithSalt defaultSalt
599-- | Compute an 'Int' sized polynomial rolling hash of the first n elements of
602-- > rollingHashFirstN = ltake n rollingHash
603{-# INLINABLE rollingHashFirstN #-}
604rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64
605rollingHashFirstN n = ltake n rollingHash
607------------------------------------------------------------------------------
608-- Monoidal left folds
609------------------------------------------------------------------------------
611-- | Append the elements of an input stream to a provided starting value.
613-- > S.fold (FL.sconcat 10) (S.map Sum $ S.enumerateFromTo 1 10)
617{-# INLINE sconcat #-}
618sconcat :: (Monad m, Semigroup a) => a -> Fold m a a
619sconcat i = Fold (\x a -> return $ x <> a) (return i) return
621-- | Fold an input stream consisting of monoidal elements using 'mappend'
624-- > S.fold FL.mconcat (S.map Sum $ S.enumerateFromTo 1 10)
627{-# INLINE mconcat #-}
628mconcat :: (Monad m, Monoid a
629#if __GLASGOW_HASKELL__ < 804
633mconcat = sconcat mempty
636-- > foldMap f = lmap f mconcat
638-- Make a fold from a pure function that folds the output of the function
639-- using 'mappend' and 'mempty'.
641-- > S.fold (FL.foldMap Sum) $ S.enumerateFromTo 1 10
644{-# INLINABLE foldMap #-}
645foldMap :: (Monad m, Monoid b
646#if __GLASGOW_HASKELL__ < 804
649 ) => (a -> b) -> Fold m a b
650foldMap f = lmap f mconcat
653-- > foldMapM f = lmapM f mconcat
655-- Make a fold from a monadic function that folds the output of the function
656-- using 'mappend' and 'mempty'.
658-- > S.fold (FL.foldMapM (return . Sum)) $ S.enumerateFromTo 1 10
661{-# INLINABLE foldMapM #-}
662foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
663foldMapM act = Fold step begin done
666 begin = return mempty
669 return $! mappend m m'
671------------------------------------------------------------------------------
673------------------------------------------------------------------------------
675-- | Folds the input stream to a list.
677-- /Warning!/ working on large lists accumulated as buffers in memory could be
678-- very inefficient, consider using "Streamly.Memory.Array" instead.
682-- id . (x1 :) . (x2 :) . (x3 :) . ... . (xn :) $ []
683{-# INLINABLE toList #-}
684toList :: Monad m => Fold m a [a]
685toList = Fold (\f x -> return $ f . (x :))
689------------------------------------------------------------------------------
691------------------------------------------------------------------------------
693-- | A fold that drains the first n elements of its input, running the effects
694-- and discarding the results.
695{-# INLINABLE drainN #-}
696drainN :: Monad m => Int -> Fold m a ()
697drainN n = ltake n drain
699-- | A fold that drains elements of its input as long as the predicate succeeds,
700-- running the effects and discarding the results.
701{-# INLINABLE drainWhile #-}
702drainWhile :: Monad m => (a -> Bool) -> Fold m a ()
703drainWhile p = ltakeWhile p drain
705------------------------------------------------------------------------------
707------------------------------------------------------------------------------
709-- | Like 'index', except with a more general 'Integral' argument
712{-# INLINABLE genericIndex #-}
713genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
714genericIndex i = Fold step (return $ Left' 0) done
727-- | Lookup the element at the given index.
730{-# INLINABLE index #-}
731index :: Monad m => Int -> Fold m a (Maybe a)
734-- | Extract the first element of the stream, if any.
736-- > head = fmap getFirst $ FL.foldMap (First . Just)
739{-# INLINABLE head #-}
740head :: Monad m => Fold m a (Maybe a)
743-- | Returns the first element that satisfies the given predicate.
746-- find p = fmap getFirst $
747-- FL.foldMap (\x -> First (if p x then Just x else Nothing))
751{-# INLINABLE find #-}
752find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
753find predicate = Fold step (return Nothing') (return . toMaybe)
757 Nothing' -> if predicate a
762-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
763-- first pair where the key equals the given value @a@.
765-- > lookup = snd <$> find ((==) . fst)
768{-# INLINABLE lookup #-}
769lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
770lookup a0 = Fold step (return Nothing') (return . toMaybe)
772 step x (a,b) = return $
774 Nothing' -> if a == a0
779-- | Convert strict 'Either'' to lazy 'Maybe'
780{-# INLINABLE hush #-}
781hush :: Either' a b -> Maybe b
782hush (Left' _) = Nothing
783hush (Right' b) = Just b
785-- | Returns the first index that satisfies the given predicate.
788{-# INLINABLE findIndex #-}
789findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
790findIndex predicate = Fold step (return $ Left' 0) (return . hush)
800-- | Returns the first index where a given value is found in the stream.
802-- > elemIndex a = findIndex (== a)
805{-# INLINABLE elemIndex #-}
806elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
807elemIndex a = findIndex (a ==)
809------------------------------------------------------------------------------
811------------------------------------------------------------------------------
813-- | Return 'True' if the input stream is empty.
815-- > null = fmap isJust head
818{-# INLINABLE null #-}
819null :: Monad m => Fold m a Bool
820null = Fold (\_ _ -> return False) (return True) return
823-- > any p = lmap p or
824-- > any p = fmap getAny . FL.foldMap (Any . p)
826-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
830any :: Monad m => (a -> Bool) -> Fold m a Bool
831any predicate = Fold (\x a -> return $ x || predicate a) (return False) return
833-- | Return 'True' if the given element is present in the stream.
835-- > elem a = any (== a)
838{-# INLINABLE elem #-}
839elem :: (Eq a, Monad m) => a -> Fold m a Bool
843-- > all p = lmap p and
844-- > all p = fmap getAll . FL.foldMap (All . p)
846-- | Returns 'True' if all elements of a stream satisfy a predicate.
850all :: Monad m => (a -> Bool) -> Fold m a Bool
851all predicate = Fold (\x a -> return $ x && predicate a) (return True) return
853-- | Returns 'True' if the given element is not present in the stream.
855-- > notElem a = all (/= a)
858{-# INLINABLE notElem #-}
859notElem :: (Eq a, Monad m) => a -> Fold m a Bool
860notElem a = all (a /=)
862-- | Returns 'True' if all elements are 'True', 'False' otherwise
864-- > and = all (== True)
865-- > and = fmap getAll . FL.foldMap All
869and :: Monad m => Fold m Bool Bool
870and = Fold (\x a -> return $ x && a) (return True) return
872-- | Returns 'True' if any element is 'True', 'False' otherwise
874-- > or = any (== True)
875-- > or = fmap getAny . FL.foldMap Any
879or :: Monad m => Fold m Bool Bool
880or = Fold (\x a -> return $ x || a) (return False) return
882------------------------------------------------------------------------------
884------------------------------------------------------------------------------
886------------------------------------------------------------------------------
887-- Grouping without looking at elements
888------------------------------------------------------------------------------
890------------------------------------------------------------------------------
892------------------------------------------------------------------------------
894-- XXX These would just be applicative compositions of terminating folds.
896-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
897-- elements of its input are consumed by fold @f1@ and the rest of the stream
898-- is consumed by fold @f2@.
900-- > let splitAt_ n xs = S.fold (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
902-- >>> splitAt_ 6 "Hello World!"
903-- > ("Hello ","World!")
905-- >>> splitAt_ (-1) [1,2,3]
908-- >>> splitAt_ 0 [1,2,3]
911-- >>> splitAt_ 1 [1,2,3]
914-- >>> splitAt_ 3 [1,2,3]
917-- >>> splitAt_ 4 [1,2,3]
922-- This can be considered as a two-fold version of 'ltake' where we take both
923-- the segments instead of discarding the leftover.
925{-# INLINE splitAt #-}
932splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
933 Fold step initial extract
935 initial = Tuple3' n <$> initialL <*> initialR
937 step (Tuple3' i xL xR) input =
939 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
940 else stepR xR input >>= (return . Tuple3' i xL)
942 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
944------------------------------------------------------------------------------
946------------------------------------------------------------------------------
948------------------------------------------------------------------------------
950------------------------------------------------------------------------------
952-- | Break the input stream into two groups, the first group takes the input as
953-- long as the predicate applied to the first element of the stream and next
954-- input element holds 'True', the second group takes the rest of the input.
964spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
965 Fold step initial extract
968 initial = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
970 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
971 if cmp frst input && isFirstG
973 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
975 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
977 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
980 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
982 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
984 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
986-- | @span p f1 f2@ composes folds @f1@ and @f2@ such that @f1@ consumes the
987-- input as long as the predicate @p@ is 'True'. @f2@ consumes the rest of the
990-- > let span_ p xs = S.fold (S.span p FL.toList FL.toList) $ S.fromList xs
992-- >>> span_ (< 1) [1,2,3]
995-- >>> span_ (< 2) [1,2,3]
998-- >>> span_ (< 4) [1,2,3]
1003-- This can be considered as a two-fold version of 'ltakeWhile' where we take
1004-- both the segments instead of discarding the leftover.
1012span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1013 Fold step initial extract
1017 initial = Tuple3' <$> initialL <*> initialR <*> return True
1019 step (Tuple3' a b isFirstG) input =
1020 if isFirstG && p input
1021 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
1022 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
1024 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1027-- > break p = span (not . p)
1029-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
1030-- folds @f1@ and @f2@ such that @f1@ stops consuming input as soon as the
1031-- predicate @p@ becomes 'True'. The rest of the input is consumed @f2@.
1033-- This is the binary version of 'splitBy'.
1035-- > let break_ p xs = S.fold (S.break p FL.toList FL.toList) $ S.fromList xs
1037-- >>> break_ (< 1) [3,2,1]
1040-- >>> break_ (< 2) [3,2,1]
1043-- >>> break_ (< 4) [3,2,1]
1054break p = span (not . p)
1056-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
1057-- predicate is applied to the previous and the next input elements.
1060{-# INLINE spanByRolling #-}
1067spanByRolling cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1068 Fold step initial extract
1071 initial = Tuple3' <$> initialL <*> initialR <*> return Nothing
1073 step (Tuple3' a b (Just frst)) input =
1075 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
1076 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
1078 step (Tuple3' a b Nothing) input =
1079 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
1081 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1083------------------------------------------------------------------------------
1084-- Binary splitting on a separator
1085------------------------------------------------------------------------------
1088-- | Find the first occurrence of the specified sequence in the input stream
1089-- and break the input stream into two parts, the first part consisting of the
1090-- stream before the sequence and the second part consisting of the sequence
1091-- and the rest of the stream.
1093-- > let breakOn_ pat xs = S.fold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
1095-- >>> breakOn_ "dear" "Hello dear world!"
1096-- > ("Hello ","dear world!")
1098{-# INLINE breakOn #-}
1099breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
1100breakOn pat f m = undefined
1103------------------------------------------------------------------------------
1105------------------------------------------------------------------------------
1107-- | Distribute one copy of the stream to each fold and zip the results.
1110-- |-------Fold m a b--------|
1111-- ---stream m a---| |---m (b,c)
1112-- |-------Fold m a c--------|
1114-- >>> S.fold (FL.tee FL.sum FL.length) (S.enumerateFromTo 1.0 100.0)
1119tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1120tee f1 f2 = (,) <$> f1 <*> f2
1122{-# INLINE foldNil #-}
1123foldNil :: Monad m => Fold m a [b]
1124foldNil = Fold step begin done where
1126 step _ _ = return []
1129{-# INLINE foldCons #-}
1130foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]
1131foldCons (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1132 Fold step begin done
1136 begin = Tuple' <$> beginL <*> beginR
1137 step (Tuple' xL xR) a = Tuple' <$> stepL xL a <*> stepR xR a
1138 done (Tuple' xL xR) = (:) <$> doneL xL <*> doneR xR
1140-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1141-- number of consumers? For polymorphic case a vector could be helpful. For
1142-- Storables we can use arrays. Will need separate APIs for those.
1144-- | Distribute one copy of the stream to each fold and collect the results in
1149-- |-------Fold m a b--------|
1150-- ---stream m a---| |---m [b]
1151-- |-------Fold m a b--------|
1156-- >>> S.fold (FL.distribute [FL.sum, FL.length]) (S.enumerateFromTo 1 5)
1159-- This is the consumer side dual of the producer side 'sequence' operation.
1162{-# INLINE distribute #-}
1163distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1164distribute = foldr foldCons foldNil
1166-- | Like 'distribute' but for folds that return (), this can be more efficient
1167-- than 'distribute' as it does not need to maintain state.
1169{-# INLINE distribute_ #-}
1170distribute_ :: Monad m => [Fold m a ()] -> Fold m a ()
1171distribute_ fs = Fold step initial extract
1173 initial = Prelude.mapM (\(Fold s i e) ->
1174 i >>= \r -> return (Fold s (return r) e)) fs
1176 Prelude.mapM_ (\(Fold s i _) -> i >>= \r -> void (s r a)) ss
1179 Prelude.mapM_ (\(Fold _ i e) -> i >>= \r -> e r)
1181------------------------------------------------------------------------------
1183------------------------------------------------------------------------------
1185-- | Partition the input over two folds using an 'Either' partitioning
1190-- |-------Fold b x--------|
1191-- -----stream m a --> (Either b c)----| |----(x,y)
1192-- |-------Fold c y--------|
1195-- Send input to either fold randomly:
1197-- >>> import System.Random (randomIO)
1198-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
1199-- >>> S.fold (FL.partitionByM randomly FL.length FL.length) (S.enumerateFromTo 1 100)
1202-- Send input to the two folds in a proportion of 2:1:
1205-- import Data.IORef (newIORef, readIORef, writeIORef)
1206-- proportionately m n = do
1207-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1208-- return $ \\a -> do
1209-- r <- readIORef ref
1210-- writeIORef ref $ tail r
1214-- f <- proportionately 2 1
1215-- r <- S.fold (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1222-- This is the consumer side dual of the producer side 'mergeBy' operation.
1225{-# INLINE partitionByM #-}
1226partitionByM :: Monad m
1227 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1228partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1230 Fold step begin done
1234 begin = Tuple' <$> beginL <*> beginR
1235 step (Tuple' xL xR) a = do
1238 Left b -> Tuple' <$> stepL xL b <*> return xR
1239 Right c -> Tuple' xL <$> stepR xR c
1240 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1242-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1243-- makes the signature clearer as to which case belongs to which fold.
1244-- XXX need to check the performance in both cases.
1246-- | Same as 'partitionByM' but with a pure partition function.
1248-- Count even and odd numbers in a stream:
1251-- >>> let f = FL.partitionBy (\\n -> if even n then Left n else Right n)
1252-- (fmap (("Even " ++) . show) FL.length)
1253-- (fmap (("Odd " ++) . show) FL.length)
1254-- in S.fold f (S.enumerateFromTo 1 100)
1255-- ("Even 50","Odd 50")
1259{-# INLINE partitionBy #-}
1260partitionBy :: Monad m
1261 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1262partitionBy f = partitionByM (return . f)
1264-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1265-- and routes the 'Left' values to the first fold and 'Right' values to the
1268-- > partition = partitionBy id
1271{-# INLINE partition #-}
1273 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1274partition = partitionBy id
1277-- | Send one item to each fold in a round-robin fashion. This is the consumer
1278-- side dual of producer side 'mergeN' operation.
1280-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1281-- partitionN fs = Fold step begin done
1284-- TODO Demultiplex an input element into a number of typed variants. We want
1285-- to statically restrict the target values within a set of predefined types,
1286-- an enumeration of a GADT. We also want to make sure that the Map contains
1287-- only those types and the full set of those types.
1289-- TODO Instead of the input Map it should probably be a lookup-table using an
1290-- array and not in GC memory. The same applies to the output Map as well.
1291-- However, that would only be helpful if we have a very large data structure,
1292-- need to measure and see how it scales.
1294-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1297-- | Split the input stream based on a key field and fold each split using a
1298-- specific fold collecting the results in a map from the keys to the results.
1299-- Useful for cases like protocol handlers to handle different type of packets
1300-- using different handlers.
1304-- |-------Fold m a b
1305-- -----stream m a-----Map-----|
1306-- |-------Fold m a b
1312{-# INLINE demuxWith #-}
1313demuxWith :: (Monad m, Ord k)
1314 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a (Map k b)
1315demuxWith f kv = Fold step initial extract
1320-- alterF is available only since containers version 0.5.8.2
1321#if MIN_VERSION_containers(0,5,8)
1322 step mp a = case f a of
1323 (k, a') -> Map.alterF twiddle k mp
1324 -- XXX should we raise an exception in Nothing case?
1325 -- Ideally we should enforce that it is a total map over k so that look
1327 -- XXX we could use a monadic update function for a single lookup and
1328 -- update in the map.
1330 twiddle Nothing = pure Nothing
1331 twiddle (Just (Fold step' acc extract')) = do
1332 !r <- acc >>= \x -> step' x a'
1333 pure . Just $ Fold step' (return r) extract'
1337 in case Map.lookup k mp of
1338 Nothing -> return mp
1339 Just (Fold step' acc extract') -> do
1340 !r <- acc >>= \x -> step' x a'
1341 return $ Map.insert k (Fold step' (return r) extract') mp
1343 extract = Prelude.mapM (\(Fold _ acc e) -> acc >>= e)
1345-- | Fold a stream of key value pairs using a map of specific folds for each
1346-- key into a map from keys to the results of fold outputs of the corresponding
1350-- > let table = Data.Map.fromList [(\"SUM", FL.sum), (\"PRODUCT", FL.product)]
1351-- input = S.fromList [(\"SUM",1),(\"PRODUCT",2),(\"SUM",3),(\"PRODUCT",4)]
1352-- in S.fold (FL.demux table) input
1353-- fromList [("PRODUCT",8),("SUM",4)]
1358demux :: (Monad m, Ord k)
1359 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1362{-# INLINE demuxWithDefault_ #-}
1363demuxWithDefault_ :: (Monad m, Ord k)
1364 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m (k, a') b -> Fold m a ()
1365demuxWithDefault_ f kv (Fold dstep dinitial dextract) =
1366 Fold step initial extract
1370 initFold (Fold s i e) = i >>= \r -> return (Fold s (return r) e)
1372 mp <- Prelude.mapM initFold kv
1373 Tuple' mp <$> dinitial
1375 step (Tuple' mp dacc) a
1377 = case Map.lookup k mp of
1379 acc <- dstep dacc (k, a')
1380 return (Tuple' mp acc)
1381 Just (Fold step' acc _) -> do
1382 _ <- acc >>= \x -> step' x a'
1383 return (Tuple' mp dacc)
1384 extract (Tuple' mp dacc) = do
1385 void $ dextract dacc
1386 Prelude.mapM_ (\(Fold _ acc e) -> acc >>= e) mp
1388-- | Split the input stream based on a key field and fold each split using a
1389-- specific fold without collecting the results. Useful for cases like protocol
1390-- handlers to handle different type of packets.
1394-- |-------Fold m a ()
1395-- -----stream m a-----Map-----|
1396-- |-------Fold m a ()
1404-- demuxWith_ can be slightly faster than demuxWith because we do not need to
1405-- update the Map in this case. This may be significant only if the map is
1407{-# INLINE demuxWith_ #-}
1408demuxWith_ :: (Monad m, Ord k)
1409 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a ()
1410demuxWith_ f kv = Fold step initial extract
1415 Prelude.mapM (\(Fold s i e) ->
1416 i >>= \r -> return (Fold s (return r) e)) kv
1418 -- XXX should we raise an exception in Nothing case?
1419 -- Ideally we should enforce that it is a total map over k so that look
1422 = case Map.lookup k mp of
1423 Nothing -> return mp
1424 Just (Fold step' acc _) -> do
1425 _ <- acc >>= \x -> step' x a'
1427 extract = Prelude.mapM_ (\(Fold _ acc e) -> acc >>= e)
1429-- | Given a stream of key value pairs and a map from keys to folds, fold the
1430-- values for each key using the corresponding folds, discarding the outputs.
1433-- > let prn = FL.drainBy print
1434-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]
1435-- input = S.fromList [(\"ONE",1),(\"TWO",2)]
1436-- in S.fold (FL.demux_ table) input
1442{-# INLINE demux_ #-}
1443demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()
1444demux_ = demuxWith_ id
1446{-# INLINE demuxDefault_ #-}
1447demuxDefault_ :: (Monad m, Ord k)
1448 => Map k (Fold m a ()) -> Fold m (k, a) () -> Fold m (k, a) ()
1449demuxDefault_ = demuxWithDefault_ id
1451-- TODO If the data is large we may need a map/hashmap in pinned memory instead
1452-- of a regular Map. That may require a serializable constraint though. We can
1453-- have another API for that.
1455-- | Split the input stream based on a key field and fold each split using the
1456-- given fold. Useful for map/reduce, bucketizing the input in different bins
1457-- or for generating histograms.
1460-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1461-- in S.fold (FL.classify FL.toList) input
1462-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1466{-# INLINE classifyWith #-}
1467classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1468classifyWith f (Fold step initial extract) = Fold step' initial' extract'
1472 initial' = return Map.empty
1475 in case Map.lookup k kv of
1479 return $ Map.insert k r kv
1482 return $ Map.insert k r kv
1483 extract' = Prelude.mapM extract
1485-- | Given an input stream of key value pairs and a fold for values, fold all
1486-- the values belonging to each key. Useful for map/reduce, bucketizing the
1487-- input in different bins or for generating histograms.
1490-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1491-- in S.fold (FL.classify FL.toList) input
1492-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1499-- > classify fld = classifyWith fst (lmap snd fld)
1501{-# INLINE classify #-}
1502classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1503classify fld = classifyWith fst (lmap snd fld)
1505------------------------------------------------------------------------------
1507------------------------------------------------------------------------------
1509-- | Like 'unzipWith' but with a monadic splitter function.
1512{-# INLINE unzipWithM #-}
1513unzipWithM :: Monad m
1514 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1515unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1516 Fold step begin done
1520 step (Tuple' xL xR) a = do
1522 Tuple' <$> stepL xL b <*> stepR xR c
1523 begin = Tuple' <$> beginL <*> beginR
1524 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1526-- | Split elements in the input stream into two parts using a pure splitter
1527-- function, direct each part to a different fold and zip the results.
1530{-# INLINE unzipWith #-}
1532 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1533unzipWith f = unzipWithM (return . f)
1535-- | Send the elements of tuples in a stream of tuples through two different
1540-- |-------Fold m a x--------|
1541-- ---------stream of (a,b)--| |----m (x,y)
1542-- |-------Fold m b y--------|
1546-- This is the consumer side dual of the producer side 'zip' operation.
1550unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1553------------------------------------------------------------------------------
1555------------------------------------------------------------------------------
1558-- All the stream flattening transformations can also be applied to a fold
1561-- | This can be used to apply all the stream generation operations on folds.
1562lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
1565lconcatMap s f1 f2 = undefined
1568-- All the grouping transformation that we apply to a stream can also be
1569-- applied to a fold input stream. groupBy et al can be written as terminating
1570-- folds and then we can apply foldChunks to use those repeatedly on a stream.
1572-- | Apply a terminating fold repeatedly to the input of another fold.
1574-- Compare with: Streamly.Prelude.concatMap, Streamly.Prelude.foldChunks
1578{-# INLINABLE foldChunks #-}
1581 Fold m a b -> Fold m b c -> Fold m a c
1582foldChunks = undefined
1585-- XXX this would be an application of foldChunks using a terminating fold.
1587-- | Group the input stream into groups of elements between @low@ and @high@.
1588-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1589-- @high@. Each chunk is folded using the provided fold function.
1591-- This could be useful, for example, when we are folding a stream of unknown
1592-- size to a stream of arrays and we want to minimize the number of
1597-- XXX we should be able to implement it with parsers/terminating folds.
1599{-# INLINE lchunksInRange #-}
1600lchunksInRange :: Monad m
1601 => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1602lchunksInRange low high (Fold step1 initial1 extract1)
1603 (Fold step2 initial2 extract2) = undefined
1606------------------------------------------------------------------------------
1607-- Fold to a Parallel SVar
1608------------------------------------------------------------------------------
1610{-# INLINE toParallelSVar #-}
1611toParallelSVar :: MonadIO m => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
1612toParallelSVar svar winfo = Fold step initial extract
1617 step () x = liftIO $ do
1618 -- XXX we can have a separate fold for unlimited buffer case to avoid a
1619 -- branch in the step here.
1620 decrementBufferLimit svar
1621 void $ send svar (ChildYield x)
1623 extract () = liftIO $
1626{-# INLINE toParallelSVarLimited #-}
1627toParallelSVarLimited :: MonadIO m
1628 => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
1629toParallelSVarLimited svar winfo = Fold step initial extract
1632 initial = return True
1634 step True x = liftIO $ do
1635 yieldLimitOk <- decrementYieldLimit svar
1638 decrementBufferLimit svar
1639 void $ send svar (ChildYield x)
1642 cleanupSVarFromWorker svar
1645 step False _ = return False
1647 extract True = liftIO $ sendStop svar winfo
1648 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