Split chunks by elements evaluated to true by f and reduce each chunk by reducing function rf.
Note
This is an internal implementation detail of SplitBy for now.
The reducing function rf receives either a Bulk(x) if !f(x) or a End(x) if f(x) returns true. Just just after rf is called with End(x), its accumulator is finalized by complete and then passed to the downstream transducer/reducing function.
Split chunks by elements evaluated to true by f and reduce each chunk by reducing function rf.
Note
This is an internal implementation detail of SplitBy for now.
The reducing function rf receives either a Bulk(x) if !f(x) or a End(x) if f(x) returns true. Just just after rf is called with End(x), its accumulator is finalized by complete and then passed to the downstream transducer/reducing function.
Examples
julia> using Transducers
using Transducers: ReduceSplitBy, Bulk, End
julia> 1:5 |> ReduceSplitBy(isodd, Map(getindex)'(string), "") |> collect
@@ -67,13 +67,13 @@
3-element Vector{String}:
""
"2"
- "4"
Set .state field of the PrivateState of the first "unbalanced" Joiner. A Joiner matched with preceding Splitter would be treated as a regular reducing function node. Thus, private state ps must have one more Joiner than Splitter.
By default, this function does nothing, but it can be overloaded to convert an input into another type before reducing over it. This allows one to implement a foldable in terms of transducers over an existing type. For instance,
Set .state field of the PrivateState of the first "unbalanced" Joiner. A Joiner matched with preceding Splitter would be treated as a regular reducing function node. Thus, private state ps must have one more Joiner than Splitter.
By default, this function does nothing, but it can be overloaded to convert an input into another type before reducing over it. This allows one to implement a foldable in terms of transducers over an existing type. For instance,
struct VectorOfVectors{T}
v::Vector{Vector{T}}
end
Transducers.asfoldable(vov::VectorOfVectors{T}) = vov.v |> Cat()
This is an optional interface for a transducer. If transducer X is stateful (i.e., wrap is used in start), it has to be able to combine the private states to support fold functions that require an associative reducing function such as foldxt. Typical implementation takes the following form:
This is an optional interface for a transducer. If transducer X is stateful (i.e., wrap is used in start), it has to be able to combine the private states to support fold functions that require an associative reducing function such as foldxt. Typical implementation takes the following form:
function combine(rf::R_{X}, a, b)
# ,---- `ua` and `ub` are the private state of the transducer `X`
# / ,-- `ira` and `irb` are the states of inner reducing functions
# / /
@@ -82,7 +82,7 @@
irc = combine(inner(rf), ira, irb)
uc = # somehow combine private states `ua` and `ub`
return wrap(rf, uc, irc)
-end
Process basecase result state before merged by combine.
For example, on GPU, this function can be used to translate mutable states to immutable values for exchanging them through (un-GC-managed) memory. See whencompletebasecase.
Note
This function is an internal experimental interface for FoldsCUDA.
Process basecase result state before merged by combine.
For example, on GPU, this function can be used to translate mutable states to immutable values for exchanging them through (un-GC-managed) memory. See whencompletebasecase.
Note
This function is an internal experimental interface for FoldsCUDA.
Return an initial value for op. Throw an error if initializer (e.g., Init) creates unknown initial value.
Examples
julia> using Transducers
using Transducers: initialize
@@ -109,8 +109,8 @@
julia> initialize(Init, unknown_op)
ERROR: IdentityNotDefinedError: `init = Init` is specified but the identity element `InitialValue(op)` is not defined for
op = unknown_op
-[...]
Check if reducible collection is considered small compared to basesize (an integer). Fold functions such as foldxt switches to sequential __foldl__ when issmall returns true.
Default implementation is amount(reducible) <= basesize.
Check if reducible collection is considered small compared to basesize (an integer). Fold functions such as foldxt switches to sequential __foldl__ when issmall returns true.
Default implementation is amount(reducible) <= basesize.
Extract transformations in rf and itr and use the appropriate adjoint for better performance.
Note that the reducing function extracted from a comprehension such as (f(x) for x ∈ itr) may not be == to f because of the way generator comprehensions work in Julia. Use Iterators.map to specify an exact mapping function.
Examples
julia> using Transducers
+ Transducers.right)))))))
Extract transformations in rf and itr and use the appropriate adjoint for better performance.
Note that the reducing function extracted from a comprehension such as (f(x) for x ∈ itr) may not be == to f because of the way generator comprehensions work in Julia. Use Iterators.map to specify an exact mapping function.
When defining a transducer type X, it is often required to dispatch on type rf::R_{X} (Reducing Function) which bundles the current transducer xform(rf)::X and the inner reducing function inner(rf)::R_.
Side notes: There is no related API in Clojure's Transducers. Transducers.jl uses it to implement stateful transducers using "pure" functions. The idea is based on a slightly different approach taken in C++ Transducer library atria.
When defining a transducer type X, it is often required to dispatch on type rf::R_{X} (Reducing Function) which bundles the current transducer xform(rf)::X and the inner reducing function inner(rf)::R_.
Side notes: There is no related API in Clojure's Transducers. Transducers.jl uses it to implement stateful transducers using "pure" functions. The idea is based on a slightly different approach taken in C++ Transducer library atria.
If start(rf::R_{X}, state) is defined, completemust unwarp state before returning state to the outer reducing function.
Transducers.jl 0.3
In Transducers.jl 0.2, complete had a fallback implementation to automatically call unwrap when wrap is called in start. Relying on this fallback implementation is now deprecated.
This is an optional interface for a transducer. If transducer X is stateful (i.e., wrap is used in start), it has to be able to combine the private states to support fold functions that require an associative reducing function such as foldxt. Typical implementation takes the following form:
function combine(rf::R_{X}, a, b)
+result
This is usually the best way to call next as checking for Reduced is required to support early termination.
If start(rf::R_{X}, state) is defined, completemust unwarp state before returning state to the outer reducing function.
Transducers.jl 0.3
In Transducers.jl 0.2, complete had a fallback implementation to automatically call unwrap when wrap is called in start. Relying on this fallback implementation is now deprecated.
This is an optional interface for a transducer. If transducer X is stateful (i.e., wrap is used in start), it has to be able to combine the private states to support fold functions that require an associative reducing function such as foldxt. Typical implementation takes the following form:
function combine(rf::R_{X}, a, b)
# ,---- `ua` and `ub` are the private state of the transducer `X`
# / ,-- `ira` and `irb` are the states of inner reducing functions
# / /
@@ -11,7 +11,7 @@
irc = combine(inner(rf), ira, irb)
uc = # somehow combine private states `ua` and `ub`
return wrap(rf, uc, irc)
-end
Pack private state for reducing function rf (or rather the transducer X) with the result iresult returned from the inner reducing function inner(rf). This packed result is typically passed to the outer reducing function.
This is intended to be used only in start. Inside next, use wrapping.
Implementation detail
If iresult is a Reduced, wrap actually unwraps all internal state iresult recursively. However, this is an implementation detail that should not matter when writing transducers.
Consider a reducing step constructed as
rf = opcompose(xf₁, xf₂, xf₃)'(f)
where each xfₙ is a stateful transducer and hence needs a private state stateₙ and this stateₙ is constructed in each start(::R_{typeof(xfₙ)}, result). Then, calling start(rf, result)) is equivalent to
Pack private state for reducing function rf (or rather the transducer X) with the result iresult returned from the inner reducing function inner(rf). This packed result is typically passed to the outer reducing function.
This is intended to be used only in start. Inside next, use wrapping.
Implementation detail
If iresult is a Reduced, wrap actually unwraps all internal state iresult recursively. However, this is an implementation detail that should not matter when writing transducers.
Consider a reducing step constructed as
rf = opcompose(xf₁, xf₂, xf₃)'(f)
where each xfₙ is a stateful transducer and hence needs a private state stateₙ and this stateₙ is constructed in each start(::R_{typeof(xfₙ)}, result). Then, calling start(rf, result)) is equivalent to
wrap(rf,
state₁, # private state for xf₁
wrap(inner(rf),
state₂, # private state for xf₂
@@ -20,16 +20,16 @@
result)))
The inner most step function receives the original result as the first argument while transducible processes such as foldl only sees the outer-most "tree" result₀ during the reduction.
Function f must take two argument state and iresult, and return a tuple (state, iresult). This is intended to be used only in next, possibly with a do block.
The inner most step function receives the original result as the first argument while transducible processes such as foldl only sees the outer-most "tree" result₀ during the reduction.
Function f must take two argument state and iresult, and return a tuple (state, iresult). This is intended to be used only in next, possibly with a do block.
Left fold a reducible with reducing function rf and initial value init. This is primary an API for overloading when the reducible "container" or "context" (e.g., I/O stream) of type T can provide a better reduction mechanism than the default iterator-based one.
For a simple iterable type MyType, a valid implementation is:
Left fold a reducible with reducing function rf and initial value init. This is primary an API for overloading when the reducible "container" or "context" (e.g., I/O stream) of type T can provide a better reduction mechanism than the default iterator-based one.
For a simple iterable type MyType, a valid implementation is:
function __foldl__(rf, val, itr::MyType)
for x in itr
val = @next(rf, val, x)
end
return complete(rf, val)
-end
although in this case default __foldl__ can handle MyType and thus there is no need for defining it. In general, defining __foldl__ is useful only when there is a better way to go over items in reducible than Base.iterate.
although in this case default __foldl__ can handle MyType and thus there is no need for defining it. In general, defining __foldl__ is useful only when there is a better way to go over items in reducible than Base.iterate.
In v0.2, the calling convention was @return_if_reduced complete(rf, val) and it was transformed to val isa Reduced && return reduced(complete(rf, unreduced(val))). For the rationale behind the change, see this commit message.
Examples
julia> using Transducers: @return_if_reduced
@@ -42,4 +42,4 @@
#= ... =#
#158#val
end
-end
Feed the results of xf processing items in reducible into a unary function eff. This is useful when the primary computation at the bottom is the side-effect. It is also equivalent to foreach(eff, eduction(xf, coll)). Note that
foreach(eduction(xf, coll)) do x
...
end
can be more efficient than
for x in eduction(xf, coll)
@@ -81,7 +81,7 @@
true
julia> simpler_has2([1, missing])
-false
foldxt(step, xf, reducible; [init, simd, basesize, stoppable, nestlevel]) :: T
eXtended threaded fold (reduce). This is a multi-threaded reduce based on extended fold protocol defined in Transducers.jl.
The "bottom" reduction function step(::T, ::T) :: T must be associative and init must be its identity element.
Transducers composing xf must be stateless (e.g., Map, Filter, Cat, etc.) except for ScanEmit. Note that Scan is not supported (although possible in theory). Early termination requires Julia ≥ 1.3.
Use tcollect or tcopy to collect results into a container.
basesize::Integer = amount(reducible) ÷ nthreads(): A size of chunk in reducible that is processed by each worker. A smaller size may be required when:
computation time for processing each item fluctuates a lot
computation can be terminated by reduced or transducers using it, such as ReduceIf
stoppable::Bool: [This option usually does not have to be set manually.] The threaded fold executed in the "stoppable" mode used for optimizing reduction with reduced has a slight overhead if reduced is not used. This mode can be disabled by passing stoppable = false. It is usually automatically detected and set appropriately. Note that this option is purely for optimization and does not affect the result value.
nestlevel::Union{Integer,Val}: Specify how many inner Cat (flatten) transducers to be multi-threaded (using TCat). It must be a positive integer, Val of positive integer, or Val(:inf). Val(:inf) means to use multi-threading for all Cat transducers. Note that Cat transducer should be statically known. That is to say, the fold implementation sees two Cats in ... |> Map(f) |> Cat() |> Cat() but only one Cat in ... |> Map(x -> f(x) |> Cat()) |> Cat() even though they are semantically identical.
foldxt(step, xf, reducible; [init, simd, basesize, stoppable, nestlevel]) :: T
eXtended threaded fold (reduce). This is a multi-threaded reduce based on extended fold protocol defined in Transducers.jl.
The "bottom" reduction function step(::T, ::T) :: T must be associative and init must be its identity element.
Transducers composing xf must be stateless (e.g., Map, Filter, Cat, etc.) except for ScanEmit. Note that Scan is not supported (although possible in theory). Early termination requires Julia ≥ 1.3.
Use tcollect or tcopy to collect results into a container.
basesize::Integer = amount(reducible) ÷ nthreads(): A size of chunk in reducible that is processed by each worker. A smaller size may be required when:
computation time for processing each item fluctuates a lot
computation can be terminated by reduced or transducers using it, such as ReduceIf
stoppable::Bool: [This option usually does not have to be set manually.] The threaded fold executed in the "stoppable" mode used for optimizing reduction with reduced has a slight overhead if reduced is not used. This mode can be disabled by passing stoppable = false. It is usually automatically detected and set appropriately. Note that this option is purely for optimization and does not affect the result value.
nestlevel::Union{Integer,Val}: Specify how many inner Cat (flatten) transducers to be multi-threaded (using TCat). It must be a positive integer, Val of positive integer, or Val(:inf). Val(:inf) means to use multi-threading for all Cat transducers. Note that Cat transducer should be statically known. That is to say, the fold implementation sees two Cats in ... |> Map(f) |> Cat() |> Cat() but only one Cat in ... |> Map(x -> f(x) |> Cat()) |> Cat() even though they are semantically identical.
pool::AbstractWorkerPool: Passed to Distributed.remotecall.
basesize::Integer = amount(array) ÷ nworkers(): A size of chunk in array that is processed by each worker. A smaller size may be required when computation time for processing each item can fluctuate a lot.
threads_basesize::Integer = basesize ÷ nthreads(): A size of chunk in array that is processed by each task in each worker process. The default setting assumes that the number of threads used in all workers are the same. For heterogeneous setup where each worker process has different number of threads, it may be required to use smaller threads_basesizeandbasesize to get a good performance.
pool::AbstractWorkerPool: Passed to Distributed.remotecall.
basesize::Integer = amount(array) ÷ nworkers(): A size of chunk in array that is processed by each worker. A smaller size may be required when computation time for processing each item can fluctuate a lot.
threads_basesize::Integer = basesize ÷ nthreads(): A size of chunk in array that is processed by each task in each worker process. The default setting assumes that the number of threads used in all workers are the same. For heterogeneous setup where each worker process has different number of threads, it may be required to use smaller threads_basesizeandbasesize to get a good performance.
Even though eduction returns an iterable, it is highly recommended to use the foldl-based method provided by Transducers.jl when the performance is important.
Examples
julia> using Transducers
@@ -123,7 +123,7 @@
end;
x = 1
x = 3
-x = 5
Feed src to transducer xf, storing the result in dest. Collections dest and src must have the same shape. Transducer xf may contain filtering transducers. If some entries src are skipped, the corresponding entries in dest will be unchanged. Transducer xf must not contain any expansive transducers such as MapCat.
Feed src to transducer xf, storing the result in dest. Collections dest and src must have the same shape. Transducer xf may contain filtering transducers. If some entries src are skipped, the corresponding entries in dest will be unchanged. Transducer xf must not contain any expansive transducers such as MapCat.
Feed src to transducer xf, storing the result in dest. Collections dest and src may have the same shape. Source src must be iterable. Destination dest must implement empty! and push!.
Feed src to transducer xf, storing the result in dest. Collections dest and src may have the same shape. Source src must be iterable. Destination dest must implement empty! and push!.
Process foldable with a transducer xf and then create a container of type T filled with the result. Return BangBang.Empty{T} if the transducer does not produce anything. (This is because there is no consistent interface to create an empty container given its type and not all containers support creating an empty container.)
Thread-based parallel version of collect. This is just a short-hand notation of tcopy(xf, Vector, reducible). Use tcopy to get a container other than a Vector.
Distributed.jl-based parallel version of collect. This is just a short-hand notation of dcopy(xf, Vector, reducible). Use dcopy to get a container other than a Vector.
Distributed.jl-based parallel version of collect. This is just a short-hand notation of dcopy(xf, Vector, reducible). Use dcopy to get a container other than a Vector.
Pipe items from an iterable itr processed by the transducer xf through a channel. Channel(xf, itr) and Channel(eduction(xf, itr)) are equivalent. Note that itr itself can be a Channel.
Keyword arguments are passed to Channel(function; kwargs...).
Provide elements in input processed by a transducer xf through a Channel.
Unary method channel_unordered(itr) produces a Channel that provides elements in the input iterator itr with possibly different order. Iterator comprehensions and eductions can be passed as the input itr.
Process input elements through a transducer xf and then push! them into output in undefined order.
Binary method append_unordered!(output, itr) is like append!(output, itr) but without order guarantee. Iterator comprehensions and eductions can be passed as the input itr.
output (typically a Channel) must implement thread-safe push!(output, x) method.
Transducers.jl 0.4.39 or later is required for composing transducers with ∘ and other operators and functions derived from it.
Transducers written as f |> g |> h in previous versions of Transducers.jl can now be written as f ⨟ g ⨟ h (in Julia 1.5 or later) or opcompose(f, g, h).
Note
"op" in opcompose does not stand for operator; it stands for opposite.
xf'(rf₁) is a shortcut for calling reducingfunction(xf, rf₁).
More precisely, adjoint xf′ of a transducer xf is a reducing function transformrf₁ -> rf₂. That is to say, xf' a function that maps a reducing function rf₁ to another reducing function rf₂.
Examples
julia> using Transducers
+compose(g, f)
Composition of transducers.
Transducers.jl 0.4.39
Transducers.jl 0.4.39 or later is required for composing transducers with ∘ and other operators and functions derived from it.
Transducers written as f |> g |> h in previous versions of Transducers.jl can now be written as f ⨟ g ⨟ h (in Julia 1.5 or later) or opcompose(f, g, h).
Note
"op" in opcompose does not stand for operator; it stands for opposite.
xf'(rf₁) is a shortcut for calling reducingfunction(xf, rf₁).
More precisely, adjoint xf′ of a transducer xf is a reducing function transformrf₁ -> rf₂. That is to say, xf' a function that maps a reducing function rf₁ to another reducing function rf₂.
Examples
julia> using Transducers
julia> y = Map(inv)'(+)(10, 2)
10.5
julia> y == +(10, inv(2))
-true
Broadcast inner reducing function over elements in the input. Roughly speaking, it transforms the inner reducing function op to op′(a, b) = op.(a, b). However, it has a better memory usage and better initial value handling.
If the input is an array, the array created at the first iteration is reused if it can hold the element types of subsequent iterations. Otherwise, the array type is widen as needed.
If init passed to the fold function is a lazy "initializer" object such as OnInit, it is initialized independently for each item in the first input array. This makes using Broadcasting for (possibly) in-place functions safe.
Transducers.jl 0.4.32
New in version 0.4.32.
Note
Broadcasting transducer is not supported in Julia 1.0.
Broadcast inner reducing function over elements in the input. Roughly speaking, it transforms the inner reducing function op to op′(a, b) = op.(a, b). However, it has a better memory usage and better initial value handling.
If the input is an array, the array created at the first iteration is reused if it can hold the element types of subsequent iterations. Otherwise, the array type is widen as needed.
If init passed to the fold function is a lazy "initializer" object such as OnInit, it is initialized independently for each item in the first input array. This makes using Broadcasting for (possibly) in-place functions safe.
Transducers.jl 0.4.32
New in version 0.4.32.
Note
Broadcasting transducer is not supported in Julia 1.0.
Sliding window of width size and interval step. Yield tuples.
This transducer is like Partition but feeds tuples to the downstream transducers instead of vectors.
If step == 1, this transducer supports parallel reduction for any collections; i.e., Consecutive(size, 1)'(op) is associative if op is associative.
Warning
Currently, in parallel folds, Consecutive(size, 1) cannot be used with reducing functions that can produce a Reduced.
If step > 1, this transducer can, in principle, support parallel reduction if the input collection allows random access (e.g., arrays). However, this feature is not implemented yet.
Group the input stream by a function key and then fan-out each group of key-value pairs to the eduction xf'(step). This is similar to the groupby relational database operation.
returns a result equivalent to Dict("1"=>[1,1], "2"=>[2,2], "3"=>[3,3]).
Alternatively, one can provide a reducing function directly, though this is disfavored since it prevents results from being combined with Transducers.combine and therefore cannot be used with foldxt or foldxd. For example, if GroupBy is used as in:
Reduce partitions determined by isequal on the output value of f with an associative reducing function rf. Partitions are reduced on-the-fly and no intermediate arrays are allocated.
Examples
Consider the input 1:6 "keyed" by a function x -> x ÷ 3:
Reduce partitions determined by isequal on the output value of f with an associative reducing function rf. Partitions are reduced on-the-fly and no intermediate arrays are allocated.
Examples
Consider the input 1:6 "keyed" by a function x -> x ÷ 3:
Replace each input with the value in the associative container assoc (e.g., a dictionary, array, string) if it matches with a key/index. Otherwise output the input as-is.
Replace each input with the value in the associative container assoc (e.g., a dictionary, array, string) if it matches with a key/index. Otherwise output the input as-is.
To use this transducer, all the downstream (inner) transducers must be stateless (or of type ScanEmit) and the reducing function must be associative. See also: Parallel processing tutorial.
Note that the upstream (outer) transducers need not to be stateless as long as it is called with non-parallel reduction such as foldl and collect.
To use this transducer, all the downstream (inner) transducers must be stateless (or of type ScanEmit) and the reducing function must be associative. See also: Parallel processing tutorial.
Note that the upstream (outer) transducers need not to be stateless as long as it is called with non-parallel reduction such as foldl and collect.
Zip outputs of transducers xforms in a tuple and pass it to the inner reduction step.
Warning
Head transducers drive tail transducers. Be careful when using it with transducers other than Map, especially the contractive ones like PartitionBy and the expansive ones like MapCat.
Zip outputs of transducers xforms in a tuple and pass it to the inner reduction step.
Warning
Head transducers drive tail transducers. Be careful when using it with transducers other than Map, especially the contractive ones like PartitionBy and the expansive ones like MapCat.
Combine N reducing functions into a new reducing function that work on N-tuple. The i-th reducing function receives the i-th element of the input tuple.
Roughly speaking, ProductRF(op₁, op₂, ..., opₙ) is equivalent to
Like TeeRF, ProductRF can be used to drive multiple reducing functions. ProductRF is more "low-level" in the sense that TeeRF can be defined in terms of ProductRF (other direction is much harder):
The type signaling transducible processes to abort.
Note
Call reduced function for aborting the transducible process since reduced makes sure x is not doubly wrapped. Reduced is meant to be used as x isa Reduced for checking if the result from transduce is due to early termination.
The type signaling transducible processes to abort.
Note
Call reduced function for aborting the transducible process since reduced makes sure x is not doubly wrapped. Reduced is meant to be used as x isa Reduced for checking if the result from transduce is due to early termination.
simd: If true or :ivdep, enable SIMD using Base.@simd. If :ivdep, use @simd ivdep for ... end variant. Read Julia manual of Base.@simd to understand when it is appropriate to use this option. For example, simd = :ivdepmust not be used with stateful transducer like Scan. If false (default), Base.@simd is not used.
simd: If true or :ivdep, enable SIMD using Base.@simd. If :ivdep, use @simd ivdep for ... end variant. Read Julia manual of Base.@simd to understand when it is appropriate to use this option. For example, simd = :ivdepmust not be used with stateful transducer like Scan. If false (default), Base.@simd is not used.
Examples
julia> using Folds
julia> Folds.sum(1:3, SequentialEx())
-6
basesize::Integer = amount(reducible) ÷ nthreads(): A size of chunk in reducible that is processed by each worker. A smaller size may be required when:
computation time for processing each item fluctuates a lot
computation can be terminated by reduced or transducers using it, such as ReduceIf
stoppable::Bool: [This option usually does not have to be set manually.] The threaded fold executed in the "stoppable" mode used for optimizing reduction with reduced has a slight overhead if reduced is not used. This mode can be disabled by passing stoppable = false. It is usually automatically detected and set appropriately. Note that this option is purely for optimization and does not affect the result value.
nestlevel::Union{Integer,Val}: Specify how many inner Cat (flatten) transducers to be multi-threaded (using TCat). It must be a positive integer, Val of positive integer, or Val(:inf). Val(:inf) means to use multi-threading for all Cat transducers. Note that Cat transducer should be statically known. That is to say, the fold implementation sees two Cats in ... |> Map(f) |> Cat() |> Cat() but only one Cat in ... |> Map(x -> f(x) |> Cat()) |> Cat() even though they are semantically identical.
simd: If true or :ivdep, enable SIMD using Base.@simd. If :ivdep, use @simd ivdep for ... end variant. Read Julia manual of Base.@simd to understand when it is appropriate to use this option. For example, simd = :ivdepmust not be used with stateful transducer like Scan. If false (default), Base.@simd is not used.
basesize::Integer = amount(reducible) ÷ nthreads(): A size of chunk in reducible that is processed by each worker. A smaller size may be required when:
computation time for processing each item fluctuates a lot
computation can be terminated by reduced or transducers using it, such as ReduceIf
stoppable::Bool: [This option usually does not have to be set manually.] The threaded fold executed in the "stoppable" mode used for optimizing reduction with reduced has a slight overhead if reduced is not used. This mode can be disabled by passing stoppable = false. It is usually automatically detected and set appropriately. Note that this option is purely for optimization and does not affect the result value.
nestlevel::Union{Integer,Val}: Specify how many inner Cat (flatten) transducers to be multi-threaded (using TCat). It must be a positive integer, Val of positive integer, or Val(:inf). Val(:inf) means to use multi-threading for all Cat transducers. Note that Cat transducer should be statically known. That is to say, the fold implementation sees two Cats in ... |> Map(f) |> Cat() |> Cat() but only one Cat in ... |> Map(x -> f(x) |> Cat()) |> Cat() even though they are semantically identical.
simd: If true or :ivdep, enable SIMD using Base.@simd. If :ivdep, use @simd ivdep for ... end variant. Read Julia manual of Base.@simd to understand when it is appropriate to use this option. For example, simd = :ivdepmust not be used with stateful transducer like Scan. If false (default), Base.@simd is not used.
Examples
julia> using Folds
julia> Folds.sum(1:3, ThreadedEx(basesize = 1))
-6
pool::AbstractWorkerPool: Passed to Distributed.remotecall.
basesize::Integer = amount(array) ÷ nworkers(): A size of chunk in array that is processed by each worker. A smaller size may be required when computation time for processing each item can fluctuate a lot.
threads_basesize::Integer = basesize ÷ nthreads(): A size of chunk in array that is processed by each task in each worker process. The default setting assumes that the number of threads used in all workers are the same. For heterogeneous setup where each worker process has different number of threads, it may be required to use smaller threads_basesizeandbasesize to get a good performance.
simd: If true or :ivdep, enable SIMD using Base.@simd. If :ivdep, use @simd ivdep for ... end variant. Read Julia manual of Base.@simd to understand when it is appropriate to use this option. For example, simd = :ivdepmust not be used with stateful transducer like Scan. If false (default), Base.@simd is not used.
pool::AbstractWorkerPool: Passed to Distributed.remotecall.
basesize::Integer = amount(array) ÷ nworkers(): A size of chunk in array that is processed by each worker. A smaller size may be required when computation time for processing each item can fluctuate a lot.
threads_basesize::Integer = basesize ÷ nthreads(): A size of chunk in array that is processed by each task in each worker process. The default setting assumes that the number of threads used in all workers are the same. For heterogeneous setup where each worker process has different number of threads, it may be required to use smaller threads_basesizeandbasesize to get a good performance.
simd: If true or :ivdep, enable SIMD using Base.@simd. If :ivdep, use @simd ivdep for ... end variant. Read Julia manual of Base.@simd to understand when it is appropriate to use this option. For example, simd = :ivdepmust not be used with stateful transducer like Scan. If false (default), Base.@simd is not used.
Examples
julia> using Folds
julia> Folds.sum(1:3, DistributedEx())
-6
A "placeholder" executor that indicates preference to parallel execution.
This lets the input data collection decide preferred execution strategy (e.g., CUDAEx for CuArray when FoldsCUDA.jl is available), assuming that the reducing function is associative. The default executor is ThreadedEx. As an optional feature, some input data collections support (e.g., AbstractChannel) automatically demoting the execution strategy to SequentialEx. An error is thrown if the automatic detection fails,
Split input collection into chunks delimited by the elements on which f returns true. This can be used to implement parallel and lazy versions of functions like eachline and split.
If keepend is true (or Val(true)), include the "delimiter"/end element at the end of each chunk. If keepempty is true (or Val(true)), include empty chunks. When keepend is true, the value of keepempty is irrelevant since the chunks cannot be empty (i.e., it at least contains the end).
The input collection (xs in SplitBy(...)(xs)) has to support eachindex and view or SubString.
Extended Help
Examples
For demonstration, consider the following input stream and SplitBy(iszero; ...) used with the following options:
A "placeholder" executor that indicates preference to parallel execution.
This lets the input data collection decide preferred execution strategy (e.g., CUDAEx for CuArray when FoldsCUDA.jl is available), assuming that the reducing function is associative. The default executor is ThreadedEx. As an optional feature, some input data collections support (e.g., AbstractChannel) automatically demoting the execution strategy to SequentialEx. An error is thrown if the automatic detection fails,
Split input collection into chunks delimited by the elements on which f returns true. This can be used to implement parallel and lazy versions of functions like eachline and split.
If keepend is true (or Val(true)), include the "delimiter"/end element at the end of each chunk. If keepempty is true (or Val(true)), include empty chunks. When keepend is true, the value of keepempty is irrelevant since the chunks cannot be empty (i.e., it at least contains the end).
The input collection (xs in SplitBy(...)(xs)) has to support eachindex and view or SubString.
Extended Help
Examples
For demonstration, consider the following input stream and SplitBy(iszero; ...) used with the following options:
Extract "processing" part of an iterator as a Transducer. The "data source" iterator (i.e., xs in (f(x) for x in xs)) is ignored and nothing must be used as a place holder (i.e., (f(x) for x in nothing)).
Apply transducer xf to the reducing function step to create a new reducing function.
Transducers.jl 0.3
New in version 0.3.
Warning
Be careful using reducingfunction with stateful transducers like Scan with mutable init (e.g., Scan(push!, [])). See more in Examples below.
Arguments
xf::Transducer: A transducer.
step: A callable which accepts 1 and 2 arguments. If it only accepts 2 arguments, wrap it with Completing to "add" 1-argument form (i.e., complete protocol).
The canonical initializer; i.e., a singleton placeholder usable for init argument of foldl for binary functions with known initial values.
When init = Init is passed to foldl etc., Init(op) is called for the bottom reducing function op during the start phase. Init(op) returns InitialValue(op) which acts as the canonical initial value of op.
Previously, setinput combined with eduction was a recommended way to use transducers in a type stable manner. As of v0.3, all the foldl-like functions and eduction are type stable for many cases. This workaround is no more necessary.
Previously, setinput combined with eduction was a recommended way to use transducers in a type stable manner. As of v0.3, all the foldl-like functions and eduction are type stable for many cases. This workaround is no more necessary.
Examples
julia> using Transducers
julia> ed = eduction(Map(x -> 2x), Float64[]);
julia> xs = ones(2, 3);
julia> foldl(+, setinput(ed, xs))
-12.0
Wrap a foldable so that progress is shown in logging-based progress meter (e.g., Juno) during foldl, foldxt, foldxd, etc.
For parallel reduction such as foldxt and foldxd, reasonably small basesize and threads_basesize (for foldxd) must be used to ensure that progress information is updated frequently. However, it may slow down the computation if basesize is too small.
Keyword Arguments
interval::Real: Minimum interval (in seconds) for how often progress is logged.
Wrap a foldable so that progress is shown in logging-based progress meter (e.g., Juno) during foldl, foldxt, foldxd, etc.
For parallel reduction such as foldxt and foldxd, reasonably small basesize and threads_basesize (for foldxd) must be used to ensure that progress information is updated frequently. However, it may slow down the computation if basesize is too small.
Keyword Arguments
interval::Real: Minimum interval (in seconds) for how often progress is logged.
Examples
julia> using Transducers
julia> xf = Map() do x
sleep(0.01)
@@ -1288,7 +1288,7 @@
);
julia> foldxt(+, xf, withprogress(1:10; interval=1e-3); basesize=1)
-220
mapfoldl(::Transducer, rf, itr) is deprecated. Use foldl(rf, ::Transducer, itr) if you do not need to call single-argument rf on complete. Use foldl(whencomplete(rf, rf), ::Transducer, itr) to call the single-argument method of rf on complete.
Like foldl but step is not automatically wrapped by Completing.
mapfoldl(::Transducer, rf, itr) is deprecated. Use foldl(rf, ::Transducer, itr) if you do not need to call single-argument rf on complete. Use foldl(whencomplete(rf, rf), ::Transducer, itr) to call the single-argument method of rf on complete.
Like foldl but step is not automatically wrapped by Completing.
Examples
julia> using Transducers
julia> function step_demo(state, input)
@show state, input
@@ -1304,4 +1304,4 @@
(state, input) = (0.0, 1)
(state, input) = (1.0, 3)
Finishing with state = 4.0
-4.0
mapreduce(::Transducer, rf, itr) is deprecated. Use foldxt(rf, ::Transducer, itr) if you do not need to call single-argument rf on complete. Use foldxt(whencomplete(rf, rf), ::Transducer, itr) to call the single-argument method of rf on complete.
Like foldxt but step is not automatically wrapped by Completing.
mapreduce(::Transducer, rf, itr) is deprecated. Use foldxt(rf, ::Transducer, itr) if you do not need to call single-argument rf on complete. Use foldxt(whencomplete(rf, rf), ::Transducer, itr) to call the single-argument method of rf on complete.
Like foldxt but step is not automatically wrapped by Completing.