Deprecated. Please use groupsOf instead
Modulestreamly-0.11.1Haskell2010
Streamly.Data.Fold.Prelude
All Fold related combinators including the streamly-core Streamly.Data.Fold module, concurrency, unordered container operations.
- 3 types
- 119 values
- Packagestreamly-0.11.1
- Exports122
- LanguageHaskell2010
- LicenceBSD-3-Clause
- SourcePrelude.hs
Setup
0 declarationsTo execute the code examples provided in this module in ghci, please run the following commands first.
:m:set -XFlexibleContextsimport Control.Concurrent (threadDelay)import Data.List (sortOn)import Data.HashMap.Strict (HashMap)
import Streamly.Data.Fold (Fold)import Streamly.Data.Stream (Stream)
import qualified Data.HashMap.Strict as HMimport qualified Streamly.Data.Fold.Prelude as Foldimport qualified Streamly.Data.Stream.Prelude as Stream
For APIs that have not been released yet.
import qualified Streamly.Internal.Data.Fold as Foldimport qualified Streamly.Internal.Data.Fold.Prelude as Fold
Streamly.Data.Fold
109 declarationsAll Streamly.Data.Fold combinators are re-exported via this module. For more pre-release combinators also see Streamly.Internal.Data.Fold module.
Append a stream to a fold to build the fold accumulator incrementally. We
can repeatedly call addStream on the same fold to continue building the
fold and finally use drive to finish the fold and extract the result. Also
see the Streamly.Data.Fold.addOne operation which is a singleton version
of addStream.
Definitions:
addStream stream = Fold.drive stream . Fold.duplicateExample, build a list incrementally:
:{pure (Fold.toList :: Fold IO Int [Int]) >>= Fold.addOne 1 >>= Fold.addStream (Stream.enumerateFromTo 2 4) >>= Fold.drive Stream.nil >>= print:}[1,2,3,4]
This can be used as an O(n) list append compared to the O(n^2) ++ when
used for incrementally building a list.
Example, build a stream incrementally:
:{pure (Fold.toStream :: Fold IO Int (Stream Identity Int)) >>= Fold.addOne 1 >>= Fold.addStream (Stream.enumerateFromTo 2 4) >>= Fold.drive Stream.nil >>= print:}fromList [1,2,3,4]
This can be used as an O(n) stream append compared to the O(n^2) <> when
used for incrementally building a stream.
Example, build an array incrementally:
:{pure (Array.create :: Fold IO Int (Array Int)) >>= Fold.addOne 1 >>= Fold.addStream (Stream.enumerateFromTo 2 4) >>= Fold.drive Stream.nil >>= print:}fromList [1,2,3,4]
Example, build an array stream incrementally:
:{let f :: Fold IO Int (Stream Identity (Array Int)) f = Fold.groupsOf 2 (Array.createOf 3) Fold.toStreamin pure f >>= Fold.addOne 1 >>= Fold.addStream (Stream.enumerateFromTo 2 4) >>= Fold.drive Stream.nil >>= print:}fromList [fromList [1,2],fromList [3,4]]
Returns True if all elements of the input satisfy the predicate.
Definition:
all p = Fold.lmap p Fold.andExample:
Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]False
Returns True if any element of the input satisfies the predicate.
Definition:
any p = Fold.lmap p Fold.orExample:
Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]True
Returns the latest element omitting the first occurrence that satisfies the given equality predicate.
Example:
input = Stream.fromList [1,3,3,5]> Stream.toList $ Stream.scanMaybe (Fold.deleteBy (==) 3) input- 1,3,5
Distribute one copy of the stream to each fold and collect the results in a container.
|-------Fold m a b--------|
---stream m a---| |---m [b]
|-------Fold m a b--------|
| |
...
Stream.fold (Fold.distribute [Fold.sum, Fold.length]) (Stream.enumerateFromTo 1 5)[15,5]
distribute = Prelude.foldr (Fold.teeWith (:)) (Fold.fromPure [])This is the consumer side dual of the producer side sequence operation.
Stops when all the folds stop.
Deprecated. Please use drainMapM instead.
Definitions:
drainMapM f = Fold.lmapM f Fold.draindrainMapM f = Fold.foldMapM (void . f)
Drain all input after passing it through a monadic function. This is the dual of mapM_ on stream producers.
Drive a fold using the supplied Stream, reducing the resulting expression strictly at each step.
Definition:
drive = flip Stream.foldExample:
Fold.drive (Stream.enumerateFromTo 1 100) Fold.sum5050
Return True if the given element is present in the stream.
Definition:
elem a = Fold.any (== a)Returns the first index where a given value is found in the stream.
Definition:
elemIndex a = Fold.findIndex (== a)Returns the index of the latest element if the element matches the given value.
Definition:
elemIndices a = Fold.findIndices (== a)Returns the first element that satisfies the given predicate.
Returns the first index that satisfies the given predicate.
Returns the index of the latest element if the element satisfies the given predicate.
Returns the first element that satisfies the given predicate.
Pre-release
Deprecated. Please use "one" instead
Extract the first element of the stream, if any.
head = Fold.oneReturn the element at the given index.
Definition:
index = Fold.genericIndexIn a stream of (key-value) pairs (a, b), return the value b of the
first pair where the key equals the given value a.
Definition:
lookup x = fmap snd <$> Fold.find ((== x) . fst)Deprecated. Use rmapM instead
Map a monadic function on the output of a fold.
mapMaybe f fold maps a Maybe returning function f on the input of
the fold, filters out Nothing elements, and return the values extracted
from Just.
mapMaybe f = Fold.lmap f . Fold.catMaybesmapMaybe f = Fold.mapMaybeM (return . f)
f x = if even x then Just x else Nothingfld = Fold.mapMaybe f Fold.toListStream.fold fld (Stream.enumerateFromTo 1 10)[2,4,6,8,10]
Determine the maximum element in a stream.
Definitions:
maximum = Fold.maximumBy comparemaximum = Fold.foldl1' max
Same as the following but without a default maximum. The Max Monoid uses
the minBound as the default maximum:
maximum = fmap Data.Semigroup.getMax $ Fold.foldMap Data.Semigroup.MaxDetermine the maximum element in a stream using the supplied comparison function.
Compute a numerically stable arithmetic mean of all elements in the input stream.
Determine the minimum element in a stream using the supplied comparison function.
Definitions:
minimum = Fold.minimumBy compareminimum = Fold.foldl1' min
Same as the following but without a default minimum. The Min Monoid uses the
maxBound as the default maximum:
maximum = fmap Data.Semigroup.getMin $ Fold.foldMap Data.Semigroup.MinComputes the minimum element with respect to the given comparison function
Returns True if the given element is not present in the stream.
Definition:
notElem a = Fold.all (/= a)Take one element from the stream and stop.
Definition:
one = Fold.maybe JustThis is similar to the stream Stream.uncons operation.
Determine the product of all elements of a stream of numbers. Returns
multiplicative identity (1) when the stream is empty. The fold terminates
when it encounters (0) in its input.
Same as the following but terminates on multiplication by 0:
product = fmap Data.Monoid.getProduct $ Fold.foldMap Data.Monoid.ProductCompute an Int sized polynomial rolling hash of a stream.
rollingHash = Fold.rollingHashWithSalt Fold.defaultSaltCompute an Int sized polynomial rolling hash
H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0Where c1, c2, cn are the elements in the input stream and k is a
constant.
This hash is often used in Rabin-Karp string search algorithm.
Semigroup concat. Append the elements of an input stream to a provided starting value.
Definition:
sconcat = Fold.foldl' (<>)semigroups = fmap Data.Monoid.Sum $ Stream.enumerateFromTo 1 10Stream.fold (Fold.sconcat 10) semigroupsSum {getSum = 65}
Deprecated. Use "rmapM id" instead
Flatten the monadic output of a fold to pure output.
Deprecated. Use the streamly-statistics package instead
Compute a numerically stable (population) standard deviation over all elements in the input stream.
Determine the sum of all elements of a stream of numbers. Returns additive
identity (0) when the stream is empty. Note that this is not numerically
stable for floating point numbers.
sum = Fold.fromScanl (Scanl.cumulativeScan Scanl.incrSum)Same as following but numerically stable:
sum = Fold.foldl' (+) 0sum = fmap Data.Monoid.getSum $ Fold.foldMap Data.Monoid.Sum
Continue taking the input until the input sequence matches the supplied sequence, taking the supplied sequence as well. If the pattern is empty this acts as an identity fold.
s = Stream.fromList "Gauss---Euler---Noether"f = Fold.takeEndBySeq (Array.fromList "---") Fold.toListStream.fold f s"Gauss---"
Stream.fold Fold.toList $ Stream.foldMany f s["Gauss---","Euler---","Noether"]
Uses Rabin-Karp algorithm for substring search.
See also: Streamly.Data.Stream.splitOnSeq and
Streamly.Data.Stream.splitEndBySeq.
Pre-release
Like takeEndBySeq but discards the matched sequence.
s = Stream.fromList "Gauss---Euler---Noether"f = Fold.takeEndBySeq_ (Array.fromList "---") Fold.toListStream.fold f s"Gauss"
Stream.fold Fold.toList $ Stream.foldMany f s["Gauss","Euler","Noether"]
See also: Streamly.Data.Stream.splitOnSeq and
Streamly.Data.Stream.splitEndBySeq_.
Pre-release
Distribute one copy of the stream to each fold and zip the results.
|-------Fold m a b--------|
---stream m a---| |---m (b,c)
|-------Fold m a c--------|
Definition:
tee = Fold.teeWith (,)Example:
t = Fold.tee Fold.sum Fold.lengthStream.fold t (Stream.enumerateFromTo 1.0 100.0)(5050.0,100)
Get the top n elements using the supplied comparison function.
To get bottom n elements instead:
bottomBy cmp = Fold.topBy (flip cmp)Example:
stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]Stream.fold (Fold.topBy compare 3) stream >>= MutArray.toList[17,11,9]
Pre-release
Return the latest unique element using the supplied comparison function. Returns Nothing if the current element is same as the last element otherwise returns Just.
Example, strip duplicate path separators:
input = Stream.fromList "//a//b"f x y = x == '/' && y == '/'
> Stream.toList $ Stream.scanMaybe (Fold.uniqBy f) input"ab"
Space: O(1)
Pre-release
Send the elements of tuples in a stream of tuples through two different folds.
|-------Fold m a x--------|
---------stream of (a,b)--| |----m (x,y)
|-------Fold m b y--------|
Definition:
unzip = Fold.unzipWith idThis is the consumer side dual of the producer side zip operation.
Deprecated. Use the streamly-statistics package instead
Compute a numerically stable (population) variance over all elements in the input stream.
Deprecated. Use classifyScan instead
Folds the values for each key using the supplied fold. When scanning, as soon as the fold is complete, its result is available in the second component of the tuple. The first component of the tuple is a snapshot of the in-progress folds.
Once the fold for a key is done, any future values of the key are ignored.
Definition:
> classify f fld = Fold.demux f (const fld)Deprecated. Use classifyScanIO instead
Same as classify except that it uses mutable IORef cells in the Map providing better performance. Be aware that if this is used as a scan, the values in the intermediate Maps would be mutable.
Definitions:
> classifyIO f fld = Fold.demuxIO f (const fld)Count non-duplicate elements in the stream.
Definition:
countDistinct = fmap Set.size Fold.toSetcountDistinct = Fold.postscanl Scanl.nub $ Fold.catMaybes $ Fold.length
The memory used is proportional to the number of distinct elements in the stream, to guard against using too much memory use it as a scan and terminate if the count reaches more than a threshold.
Space: \mathcal{O}(n)
Pre-release
Like countDistinct but specialized to a stream of Int, for better performance.
Definition:
countDistinctInt = fmap IntSet.size Fold.toIntSetcountDistinctInt = Fold.postscanl Scanl.nubInt $ Fold.catMaybes $ Fold.length
Pre-release
Deprecated. Use demuxScan instead
demux getKey getFold: In a key value stream, fold values corresponding
to each key using a key specific fold. getFold is invoked to generate a
key specific fold when a key is encountered for the first time in the
stream.
The first component of the output tuple is a key-value Map of in-progress folds. The fold returns the fold result as the second component of the output tuple whenever a fold terminates.
If a fold terminates, another instance of the fold is started upon receiving
an input with that key, getFold is invoked again whenever the key is
encountered again.
This can be used to scan a stream and collect the results from the scan output.
Since the fold generator function is monadic we can add folds dynamically. For example, we can maintain a Map of keys to folds in an IORef and lookup the fold from that corresponding to a key. This Map can be changed dynamically, folds for new keys can be added or folds for old keys can be deleted or modified.
Compare with classify, the fold in classify is a static fold.
Pre-release
Deprecated. Use demuxScanIO instead
This is specialized version of demux that uses mutable IO cells as fold accumulators for better performance.
Keep in mind that the values in the returned Map may be changed by the ongoing fold if you are using those concurrently in another thread.
Deprecated. Use demuxerToMap instead
This collects all the results of demux in a Map.
Deprecated. Use demuxerToMapIO instead
Same as demuxToMap but uses demuxIO for better performance.
demuxerToMap getKey getFold: In a key value stream, fold values
corresponding to each key using a key specific fold. getFold is invoked to
generate a key specific fold when a key is encountered for the first time in
the stream.
If a fold terminates, another instance of the fold is started upon receiving
an input with that key, getFold is invoked again whenever the key is
encountered again.
This combinator can be used to scan a stream and collect the results from the scan output.
Since the fold generator function is monadic, folds for new keys can be added dynamically or folds for old keys can be deleted or modified. For example, we can maintain a Map of keys to folds in an IORef and lookup the fold from that corresponding to a key. This Map can be changed dynamically.
Note that this fold never terminates. Inputs that do not correspond to a fold in the map are dropped.
Compare with classify, the fold in classify is a static fold.
Pre-release
Same as demuxerToMap but uses mutable cells for better performance.
Determine the frequency of each element in the stream.
You can just collect the keys of the resulting map to get the unique elements in the stream.
Definition:
frequency = Fold.toMap id Fold.lengthFold the input to an int set. For integer inputs this performs better than toSet.
Definition:
toIntSet = Fold.foldl' (flip IntSet.insert) IntSet.emptySplit the input stream based on a key field and fold each split using the given fold. Useful for map/reduce, bucketizing the input in different bins or for generating histograms.
Example:
import Data.Map.Strict (Map):{ let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)] classify = Fold.toMap fst (Fold.lmap snd Fold.toList) in Stream.fold classify input :: IO (Map String [Double]):}fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
Once the classifier fold terminates for a particular key any further inputs in that bucket are ignored.
Space used is proportional to the number of keys seen till now and monotonically increases because it stores whether a key has been seen or not.
See demuxToMap for a more powerful version where you can use a different fold for each key. A simpler version of toMap retaining only the last value for a key can be written as:
toMap = Fold.foldl' (\kv (k, v) -> Map.insert k v kv) Map.emptyStops: never
Pre-release
Same as toMap but maybe faster because it uses mutable cells as fold accumulators in the Map.
Fold the input to a set.
Definition:
toSet = Fold.foldl' (flip Set.insert) Set.emptyAppend a singleton value to the fold.
See examples under addStream.
Pre-release
Remove the either wrapper and flatten both lefts and as well as rights in the output stream.
Definition:
catEithers = Fold.lmap (either id id)Pre-release
Map a Fold returning function on the result of a Fold and run the returned fold. This is akin to an n-ary version of splitWith where the next fold for splitting the input is decided dynamically using the previous result. This operation can be used to express data dependencies between fold operations.
Let's say the first element in the stream is a count of the following elements that we have to add, then:
import Data.Maybe (fromJust)count = fmap fromJust Fold.onetotal n = Fold.take n Fold.sumStream.fold (Fold.concatMap total count) $ Stream.fromList [10,9..1]45
This does not fuse completely, see refold for a fusible alternative.
Time: O(n^2) where n is the number of compositions.
See also: foldIterateM, refold
A fold that drains all its input, running the effects and discarding the results.
drain = Fold.drainMapM (const (return ()))drain = Fold.foldl' (\_ _ -> ()) ()
duplicate provides the ability to run a fold in parts. The duplicated fold consumes the input and returns the same fold as output instead of returning the final result, the returned fold can be run later to consume more input.
duplicate essentially appends a stream to the fold without finishing the fold. Compare with snoc which appends a singleton value to the fold.
Pre-release
Include only those elements that pass a predicate.
Stream.fold (Fold.filter (> 5) Fold.sum) $ Stream.fromList [1..10]40
filter p = Fold.filterM (return . p)filter p = Fold.mapMaybe (\x -> if p x then Just x else Nothing)
Like filter but with a monadic predicate.
f p x = p x >>= \r -> return $ if r then Just x else NothingfilterM p = Fold.mapMaybeM (f p)
Make a fold from a left fold style pure step function and initial value of the accumulator.
If your Fold returns only Partial (i.e. never returns a Done) then you
can use foldl'* constructors.
A fold with an extract function can be expressed using fmap:
mkfoldlx :: Monad m => (s -> a -> s) -> s -> (s -> b) -> Fold m a b
mkfoldlx step initial extract = fmap extract (foldl' step initial)
Make a strict left fold, for non-empty streams, using first element as the starting value. Returns Nothing if the stream is empty.
Pre-release
Like 'foldl1'' but with a monadic step function.
Pre-release
Make a fold from a left fold style monadic step function and initial value of the accumulator.
A fold with an extract function can be expressed using rmapM:
mkFoldlxM :: Functor m => (s -> a -> m s) -> m s -> (s -> m b) -> Fold m a b
mkFoldlxM step initial extract = rmapM extract (foldlM' step initial)
Deprecated. Please use foldl1M' instead
Like 'foldl1'' but with a monadic step function.
Pre-release
Deprecated. Please use foldr' instead.
Make a fold using a right fold style step function and a terminal value. It performs a strict right fold via a left fold using function composition. Note that a strict right fold can only be useful for constructing strict structures in memory. For reductions this will be very inefficient.
Definitions:
foldr' f z = fmap (flip appEndo z) $ Fold.foldMap (Endo . f)foldr' f z = fmap ($ z) $ Fold.foldl' (\g x -> g . f x) id
Example:
Stream.fold (Fold.foldr' (:) []) $ Stream.enumerateFromTo 1 5[1,2,3,4,5]
Make a terminating fold with an effectful step function and initial state, and a state extraction function.
foldtM' = Fold.FoldWe can just use Fold but it is provided for completeness.
Pre-release
groupsOf n split collect repeatedly applies the split fold to chunks
of n items in the input stream and supplies the result to the collect
fold.
Definition:
groupsOf n split = Fold.many (Fold.take n split)Example:
twos = Fold.groupsOf 2 Fold.toList Fold.toListStream.fold twos $ Stream.fromList [1..10][[1,2],[3,4],[5,6],[7,8],[9,10]]
Stops when collect stops.
Deprecated. Please use latest instead.
Returns the latest element of the input stream, if any.
latest = Fold.foldl1' (\_ x -> x)latest = fmap getLast $ Fold.foldMap (Last . Just)
Determine the length of the input stream.
Definition:
length = Fold.genericLengthlength = fmap getSum $ Fold.foldMap (Sum . const 1)
lmap f fold maps the function f on the input of the fold.
Definition:
lmap = Fold.lmapM returnExample:
sumSquared = Fold.lmap (\x -> x * x) Fold.sumStream.fold sumSquared (Stream.enumerateFromTo 1 100)338350
lmapM f fold maps the monadic function f on the input of the fold.
Collect zero or more applications of a fold. many first second applies
the first fold repeatedly on the input stream and accumulates it's results
using the second fold.
two = Fold.take 2 Fold.toListtwos = Fold.many two Fold.toListStream.fold twos $ Stream.fromList [1..10][[1,2],[3,4],[5,6],[7,8],[9,10]]
Stops when second fold stops.
See also: Data.Stream.concatMap, Data.Stream.foldMany
Change the underlying monad of a fold. Also known as hoist.
Pre-release
postscanl scanner collector postscans the input of the collector fold
to change it in a stateful manner using scanner.
Pre-release
Map a monadic function on the output of a fold.
Deprecated. Please use postscanlMaybe instead.
Use a Maybe returning fold as a filtering scan.
> scanMaybe p f = Fold.postscan p (Fold.catMaybes f)Pre-release
Deprecated. Please use "splitWith" instead
Sequential fold application. Apply two folds sequentially to an input stream. The input is provided to the first fold, when it is done - the remaining input is provided to the second fold. When the second fold is done or if the input stream is over, the outputs of the two folds are combined using the supplied function.
Example:
header = Fold.take 8 Fold.toListline = Fold.takeEndBy (== '\n') Fold.toListf = Fold.splitWith (,) header lineStream.fold f $ Stream.fromList "header: hello\n"("header: ","hello\n")
Note: This is dual to appending streams using Data.Stream.append.
Note: this implementation allows for stream fusion but has quadratic time complexity, because each composition adds a new branch that each subsequent fold's input element has to traverse, therefore, it cannot scale to a large number of compositions. After around 100 compositions the performance starts dipping rapidly compared to a CPS style implementation.
For larger number of compositions you can convert the fold to a parser and use ParserK.
Time: O(n^2) where n is the number of compositions.
Take at most n input elements and fold them using the supplied fold. A
negative count is treated as 0.
Stream.fold (Fold.take 2 Fold.toList) $ Stream.fromList [1..10][1,2]
Take the input, stop when the predicate succeeds taking the succeeding element as well.
Example:
input = Stream.fromList "hello\nthere\n"line = Fold.takeEndBy (== '\n') Fold.toListStream.fold line input"hello\n"
Stream.fold Fold.toList $ Stream.foldMany line input["hello\n","there\n"]
Like takeEndBy but drops the element on which the predicate succeeds.
Example:
input = Stream.fromList "hello\nthere\n"line = Fold.takeEndBy_ (== '\n') Fold.toListStream.fold line input"hello"
Stream.fold Fold.toList $ Stream.foldMany line input["hello","there"]
teeWith k f1 f2 distributes its input to both f1 and f2 until both
of them terminate and combines their output using k.
Definition:
teeWith k f1 f2 = fmap (uncurry k) (Fold.tee f1 f2)Example:
avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)Stream.fold avg $ Stream.fromList [1.0..100.0]50.5
For applicative composition using this combinator see Streamly.Data.Fold.Tee.
See also: Streamly.Data.Fold.Tee
Note that nested applications of teeWith do not fuse.
Folds the input stream to a list.
Warning! working on large lists accumulated as buffers in memory could be very inefficient, consider using Streamly.Data.Array instead.
toList = Fold.foldr' (:) []Buffers the input stream to a list in the reverse order of the input.
Definition:
toListRev = Fold.foldl' (flip (:)) []Warning! working on large lists accumulated as buffers in memory could be
very inefficient, consider using Streamly.Array instead.
Tee is a newtype wrapper over the Fold type providing distributing
Applicative, Semigroup, Monoid, Num, Floating and Fractional
instances.
The input received by the composed Tee is replicated and distributed to the constituent folds of the Tee.
For example, to compute the average of numbers in a stream without going through the stream twice:
avg = (/) <$> (Tee Fold.sum) <*> (Tee $ fmap fromIntegral Fold.length)Stream.fold (unTee avg) $ Stream.fromList [1.0..100.0]50.5
Similarly, the Semigroup and Monoid instances of Tee distribute the input to both the folds and combine the outputs using Monoid or Semigroup instances of the output types:
import Data.Monoid (Sum(..))t = Tee Fold.one <> Tee Fold.latestStream.fold (unTee t) (fmap Sum $ Stream.enumerateFromTo 1.0 100.0)Just (Sum {getSum = 101.0})
The Num, Floating, and Fractional instances work in the same way.
Instances7Functor, Applicative, Floating, Fractional, Num, Semigroup, …
Functor m => Functor (Tee m a)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.TeeMonad m => Applicative (Tee m a)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.Tee(Monad m, Floating b) => Floating (Tee m a b)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.Tee(Monad m, Fractional b) => Fractional (Tee m a b)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.TeeBinary Fractional operations distribute the input to both the argument Tees and combine their outputs using the Fractional instance of the output type.
(Monad m, Num b) => Num (Tee m a b)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.Tee(Semigroup b, Monad m) => Semigroup (Tee m a b)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.Tee(Monoid b, Monad m) => Monoid (Tee m a b)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.Tee
The type Fold m a b represents a consumer of an input stream of values
of type a and returning a final value of type b in Monad m. The
constructor of a fold is Fold step initial extract final.
The fold uses an internal state of type s. The initial value of the state
s is created by initial. This function is called once and only once
before the fold starts consuming input. Any resource allocation can be done
in this function.
The step function is called on each input, it consumes an input and
returns the next intermediate state (see Step) or the final result b if
the fold terminates.
Folds are no longer used for scanning, please see the Streamly.Data.Scanl
module for scanning. The extract operation is no longer used and will be
removed in future.
Before a fold terminates, final is called once and only once (unless the
fold terminated in initial itself). Any resources allocated by initial
can be released in final.
When implementing fold combinators, care should be taken to cleanup any
state of the argument folds held by the fold by calling the respective
final at all exit points of the fold. Also, final should not be called
more than once. Note that if a fold terminates by Done constructor, there
is no state to cleanup.
NOTE: The constructor is not yet released, smart constructors are provided to create folds.
Instances2Functor, Applicative
Functor m => Functor (Fold m a)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.TypeMaps a function on the output of the fold (the type
b).Monad m => Applicative (Fold m a)Defined in streamly-core-0.3.1 · Streamly.Internal.Data.Fold.TypeApplicative form of splitWith. Split the input serially over two folds. Note that this fuses but performance degrades quadratically with respect to the number of compositions. It should be good to use for less than 8 compositions.
Concurrent Operations
0 declarationsConfiguration
An abstract type for specifying the configuration parameters of a
Channel. Use Config -> Config modifier functions to modify the default
configuration. See the individual modifier documentation for default values.
Specify the maximum size of the buffer for storing the results from concurrent computations. If the buffer becomes full we stop spawning more concurrent tasks until there is space in the buffer. A value of 0 resets the buffer size to default, a negative value means there is no limit. The default value is 1500.
CAUTION! using an unbounded maxBuffer value (i.e. a negative value)
coupled with an unbounded maxThreads value is a recipe for disaster in
presence of infinite streams, or very large streams. Especially, it must
not be used when pure is used in ZipAsyncM streams as pure in
applicative zip streams generates an infinite stream causing unbounded
concurrent generation with no limit on the buffer or threads.
Spawn bound threads (i.e., spawn threads using forkOS instead of
forkIO). The default value is False.
Currently, this only takes effect only for concurrent folds.
Print debug information about the Channel when the stream ends. When the
stream does not end normally, the channel debug information is printed when
the channel is garbage collected. If you are expecting but not seeing the
debug info try adding a performMajorGC before the program ends.
Combinators
parBuffered introduces a concurrent stage at the input of the fold. The inputs are asynchronously queued in a buffer and evaluated concurrently with the evaluation of the source stream. On finalization, parBuffered waits for the asynchronous fold to complete before it returns.
In the following example both the stream and the fold have a 1 second delay, but the delay is not compounded because both run concurrently.
delay x = threadDelay 1000000 >> print x >> return xsrc = Stream.delay 1 (Stream.enumerateFromTo 1 3)dst = Fold.parBuffered id (Fold.lmapM delay Fold.sum)Stream.fold dst src...
Another example:
Stream.toList $ Stream.groupsOf 4 dst src...
Execute both the folds in a tee concurrently.
Definition:
parTee cfg c1 c2 = Fold.teeWith (,) (Fold.parBuffered cfg c1) (Fold.parBuffered cfg c2)Example:
delay x = threadDelay 1000000 >> print x >> return xc1 = Fold.lmapM delay Fold.sumc2 = Fold.lmapM delay Fold.lengthdst = Fold.parTee id c1 c2Stream.fold dst src...
Distribute the input to all the folds in the supplied list concurrently.
Definition:
parDistribute cfg = Fold.distribute . fmap (Fold.parBuffered cfg)Example:
delay x = threadDelay 1000000 >> print x >> return xc = Fold.lmapM delay Fold.sumdst = Fold.parDistribute id [c,c,c]Stream.fold dst src...
Evaluate a stream and send its outputs to zero or more dynamically generated folds. It checks for any new folds at each input generation step. Any new fold is added to the list of folds which are currently running. If there are no folds available, the input is discarded. If a fold completes its output is emitted in the output of the scan.
import Data.IORefref <- newIORef [Fold.take 2 Fold.sum, Fold.take 2 Fold.length :: Fold.Fold IO Int Int]gen = atomicModifyIORef ref (\xs -> ([], xs))Stream.toList $ Fold.parDistributeScan id gen (Stream.enumerateFromTo 1 10)...
Select first fold for Left input and second for Right input. Both folds run concurrently.
Definition
parPartition cfg c1 c2 = Fold.partition (Fold.parBuffered cfg c1) (Fold.parBuffered cfg c2)Example:
delay x = threadDelay 1000000 >> print x >> return xc1 = Fold.lmapM delay Fold.sumc2 = Fold.lmapM delay Fold.sumdst = Fold.parPartition id c1 c2Stream.fold dst $ (fmap (\x -> if even x then Left x else Right x)) src...
Evaluate a stream and send its outputs to the selected fold. The fold is dynamically selected using a key at the time of the first input seen for that key. Any new fold is added to the list of folds which are currently running. If there are no folds available for a given key, the input is discarded. If a fold completes its output is emitted in the output of the scan.
import qualified Data.Map.Strict as Mapimport Data.Maybe (fromJust)f1 = ("even", Fold.take 2 Fold.sum)f2 = ("odd", Fold.take 2 Fold.sum)kv = Map.fromList [f1, f2]getFold k = return (fromJust $ Map.lookup k kv)getKey x = if even x then "even" else "odd"input = Stream.enumerateFromTo 1 10Stream.toList $ Fold.parDemuxScan id getKey getFold input...
Split and distribute the output to two different folds and then zip the results. Both the consumer folds run concurrently.
Definition
parUnzip cfg c1 c2 = Fold.unzip (Fold.parBuffered cfg c1) (Fold.parBuffered cfg c2)Example:
delay x = threadDelay 1000000 >> print x >> return xc1 = Fold.lmapM delay Fold.sumc2 = Fold.lmapM delay Fold.sumdst = Fold.parUnzip id c1 c2Stream.fold dst $ (fmap (\x -> (x, x * x))) src...
Container Related
1 declarationSplit the input stream based on a hashable component of the key field and fold each split using the given fold. Useful for map/reduce, bucketizing the input in different bins or for generating histograms.
Consider a stream of key value pairs:
input = Stream.fromList [("k1",1),("k1",1.1),("k2",2), ("k2",2.2)]Classify each key to a different hash bin and fold the bins:
classify = Fold.toHashMapIO fst (Fold.lmap snd Fold.toList)sortOn fst . HM.toList <$> Stream.fold classify input :: IO [(String, [Double])][("k1",[1.0,1.1]),("k2",[2.0,2.2])]
Pre-release
Deprecated
Deprecated. Please use parBuffered instead.
parBuffered introduces a concurrent stage at the input of the fold. The inputs are asynchronously queued in a buffer and evaluated concurrently with the evaluation of the source stream. On finalization, parBuffered waits for the asynchronous fold to complete before it returns.
In the following example both the stream and the fold have a 1 second delay, but the delay is not compounded because both run concurrently.
delay x = threadDelay 1000000 >> print x >> return xsrc = Stream.delay 1 (Stream.enumerateFromTo 1 3)dst = Fold.parBuffered id (Fold.lmapM delay Fold.sum)Stream.fold dst src...
Another example:
Stream.toList $ Stream.groupsOf 4 dst src...