jepsen.history.fold — io.jepsen/history 0.1.3

Provides a stateful folder for running folds (like reduce) over chunked, immutable collections in linear and concurrent passes. Intended for systems where the reduction over a chunk may involve expensive work, and not fit in memory--for instance, deserializing values from disk. Provides sophisticated optimizations for running folds in parallel, and automatically fusing together multiple folds.

To build a folder, you need a chunkable collection: see jepsen.history.core. Jepsen.history chunks vectors by default at 16384 elements per chunk, which is a bit big for a demonstration, so let's chunk explicitly:

(require '[tesser.core :as t] '[jepsen.history [core :as hc] [fold :as f]]) (def dogs [{:legs 6, :name :noodle}, {:legs 4, :name :stop-it}, {:legs 4, :name :brown-one-by-the-fish-shop}]) (def chunked-dogs (hc/chunked 2 dogs)) (pprint (hc/chunks chunked-dogs)) ; ([{:legs 6, :name :noodle} {:legs 4, :name :stop-it}] ; [{:legs 4, :name :brown-one-by-the-fish-shop}])

In real use, chunks should be big enough to take a bit (a second or so?) to reduce. We keep track of some state for each chunk, so millions is probably too many. If you have fewer chunks than processors, we won't be able to optimize as efficiently.

A folder wraps a chunked collection, like so:

(def f (f/folder chunked-dogs))

Now we can perform a reduction on the folder. This works just like Clojure reduce:

(reduce (fn [max-legs dog] (max max-legs (:legs dog))) 0 e) ; => 6

Which means transducers and into work like you'd expect:

(into #{} (map :legs) f) ; => #{4 6}

OK, great. What's the point? Imagine we had a collection where getting elements was expensive--for instance, if they required IO or expensive processing. Let's put ten million dogs on disk as JSON.

(require '[jepsen.history.fold-test :as ft]) (def dogs (ft/gen-dogs-file! 1e7)) (def f (f/folder dogs))

Reducing over ten million dogs as JSON takes about ten seconds on my machine.

(time (into #{} (map :legs) dogs)) ; 10364 msecs

But with a folder, we can do something neat:

(def leg-set {:reducer-identity (constantly #{}) :reducer (fn [legs dog] (conj legs (:legs dog))) :combiner clojure.set/union}) (time (f/fold f leg-set)) ; 1660 msecs

This went roughly six times faster because the folder reduced each chunk in parallel. Now let's run, say, ten reductions in parallel.

(time (doall (pmap (fn [_] (into #{} (map :legs) dogs)) (range 10)))) ; 28477 msecs

This 28 seconds is faster than running ten folds sequentially (which would have been roughly 87 seconds), because we've got multiple cores to do the reduction. But we're still paying a significant cost because each of those reductions has to re-parse the file as it goes.

(time (doall (pmap (fn [_] (f/fold f leg-set)) (range 10)))) ; 2261 msecs

Twelve times faster than the parallel version! And roughly 45x faster than doing the reductions naively in serial. How? Because when you ask a folder to reduce something, and it's already running another reduction, it joins your new reduction to the old one and performs most (or even all) of the work in a single pass. The folder is smart enough to do this for both linear and concurrent folds--and it does it while ensuring strict order and thread safety for mutable accumulators. Let's replace that reduction with a mutable HashSet, and convert it back to a Clojure set at the end.

(import java.util.HashSet) (defn mut-hash-set [] (HashSet.)) (def fast-leg-set {:reducer-identity mut-hash-set :reducer (fn [^HashSet s, dog] (.add s (:legs dog)) s) :combiner-identity mut-hash-set :combiner (fn [^HashSet s1, ^HashSet s2] (.addAll s1 s2) s1) :post-combiner set}) (time (doall (pmap (fn [_] (f/fold f fast-leg-set)) (range 10)))) ; 2197 msecs

In general

A fold represents a reduction over a history, which can optionally be executed over chunks concurrently. It's a map with the following fields:

; Metadata

:name The unique name of this fold. May be any object, but probably a keyword.

; How to reduce a chunk

:reducer-identity A function (f history) which generates an identity object for a reduction over a chunk.

:reducer A function (f history acc op) which takes a history, a chunk accumulator, and an operation from the history, and returns a new accumulator.

:post-reducer A function (f history acc) which takes the final accumulator from a chunk and transforms it before being passed to the combiner

; How to combine chunks together

:combiner-identity A function (f history) which generates an identity object for combining chunk results together.

:combiner A function (f history acc chunk-result) which folds the result of a chunk into the combiner's accumulator. If nil, performs a left fold linearly, and does not combine at all.

:post-combiner A function (f history acc) which takes the final acc from merging all chunks and produces the fold's return value.

; Execution hints

:associative? If true, the combine function is associative, and can be applied in any order. If false, we must combine left-to-right. Right now this does nothing; we haven't implemented associative combine.

:asap? Folders ramp up processing of concurrent folds gradually to give other folds a chance to join the pass. Setting this to true disables that optimization, which means a fold can complete more quickly--at the cost of slowing down other folds.

Folds should be pure functions of their histories, though reducers and combiners are allowed to use in-memory mutability; each is guaranteed to be single-threaded. We guarantee that reducers execute at most once per element, and at most once per chunk. When not using reduced, they are exactly-once. The final return value from a fold should be immutable so that other readers or folds can use it safely in a concurrent context.

Early Termination

For linear folds (e.g. those with no combiner), we support the usual (reduced x) early-return mechanism. This means reduce and transduce work as you'd expect. Reduced values still go through the post-reducer function.

For concurrent folds, (reduced x) in a reducer terminates reduction of that particular chunk. Other chunks are still reduced. Reduced values go through post-reduce and are passed to the combiner unwrapped. If the combiner returns a reduced value, that value passes immediately to the post-combiner, without considering any other chunks.

Like transducers, post-reducers and post-combiners act on the values inside Reduced wrappers; they cannot distinguish between reduced and non-reduced values.

ary->vec clj

(ary->vec x)

Turns arrays into vectors recursively.

cancel-task-workers clj

(cancel-task-workers state task)

Takes a state and a reduce, combine, split, or join task. Cancels not only this task, but all of its dependencies which actually perform the work for its particular chunk. Returns state'.

A task of :cancelled yields no changes.

clear-old-passes! clj

(clear-old-passes! f)

Takes a Folder and clears out any old pass state.

concurrent-reduce-task-unlock-factor clj

This factor controls how quickly concurrent reduce tasks 'unlock' reductions of later chunks. 8 means that each task unlocks roughly 8 later tasks.

empty-fold clj

(empty-fold fold)

Runs a fold over zero elements.

fold clj

(fold folder fold)

Executes a fold on the given folder synchronously. See make-fold for what a fold can be. Returns result of the fold.

This is the opposite arity from (reduce f coll): here we're sort of saying (fold coll f). I've thought long and hard about this, and I think it makes sense. With reduce, you're generally composing a collection and passing it to reduce. With fold, you're generally composing a fold and executing it on the same folder over and over. This feels like it's going to be more ergonomic.

folder clj

(folder chunkable)

Creates a new stateful Folder for folds over the given chunkable.

folder-pass clj

(folder-pass f fold)

Takes a Folder and a fold. Turns the fold into a pass, ready for execution on this folder.

fuse clj

(fuse old-fold new-fold)

Takes a fold (possibly a FusedFold) and fuses a new fold into it. Also provides a means of joining in-process reducer/combiner state together, so that you can zip together two independent folds and continue with the fused fold halfway through. Returns a map of:

:fused A new FusedFold which performs everything the original fold did plus the new fold. Its post-combined results are a pair of [old-res new-res].

:join-accs A function which joins together accumulators from the original and new fold. Works on both reduce and combine accumulators. Returns an accumulator for the new fold.

Works with both folds that have combiners and those that don't.

If this isn't fast enough, we might try doing some insane reflection and dynamically compiling a new class to hold reducer state with primitive fields.

fused? clj

(fused? fold)

Is this fold a fused fold?

join-concurrent-pass clj

(join-concurrent-pass state old-pass new-pass)

Takes a task executor State, and a pair of Passes; the first (potentially) running, the second fresh. Joins these passes into a new pass which tries to merge as much work as possible into fused reduce/combine operations. Returns [state' joined-pass].

join-linear-pass clj

(join-linear-pass state old-pass new-pass)

Takes a task executor State and a pair of passes: the first (potentially) running, the second fresh. Joins these passes into a new pass which tries to merge as much work as possible into fused reduce/combine operations. Returns [state' joined-pass].

join-pass clj

(join-pass state old-pass new-pass)

Joins two passes of the same :type together.

last-missing-index clj

(last-missing-index ary)

Takes an array and returns the index of the last missing element, or Int/MIN_VALUE if none are missing. Why not -1? Simplifies some of our comparison logic.

launch-concurrent-pass clj

(launch-concurrent-pass state {:keys [chunks fold deliver] :as pass})

Takes a task executor State and an unstarted concurrent pass. Launches all the tasks required to execute this pass. Returns [state' pass'].

launch-linear-pass clj

(launch-linear-pass state {:keys [fold chunks deliver] :as pass})

Takes a task executor State and an unstarted linear pass. Launches all the tasks required to execute this pass. Returns [state' pass'].

launch-pass clj

(launch-pass state pass)

Takes a task executor state and an unstarted pass, then launches its task, returning [state' pass']

loopf clj macro

(loopf metadata reducer-forms combiner-forms)

Macro for defining folds with for/loop-like syntax. Works like two dom-top.core/reducer's glued together, plus metadata. Takes a map with metadata (e.g. :name, :associative?, ...) and two lists: one for the reducer, ne for the combiner. Turns each of those lists into a reducer fn using dom-top.core/reducer, then turns the whole thing into a fold.

make-concurrent-combine-task clj

(make-concurrent-combine-task state
                              {:keys [combiner-identity combiner]}
                              chunks
                              i
                              reduce-task)

(make-concurrent-combine-task state
                              {:keys [combiner post-combiner]}
                              chunks
                              i
                              prev-combine-task
                              reduce-task)

Takes a task executor state, a fold, chunks, a chunk index, and either:

1. A reduce task (for the first combine)
2. A previous combine task and a reduce task (for later combines)

Returns [state' task]: a task which combines that chunk with earlier combines.

make-concurrent-reduce-task clj

(make-concurrent-reduce-task state
                             {:keys [asap? reducer-identity reducer
                                     post-reducer]}
                             chunks
                             prev-reduce-tasks
                             i)

Takes a task executor state, a fold, chunks, and a vector of previously launched reduce tasks for this fold, and an index into the chunks i. Returns [state' task]: a new task to reduce that chunk.

Unless the fold requests :asap? true, we introduce synthetic dependencies to slow down later reductions. This is less speedy for single folds, but for multiple folds we actually want to defer starting work until later--that way the later folds have a chance to join and cancel our tasks. So even though we could rush ahead and launch every reduce concurrently, we inject dependencies between reduce tasks forming a tree: the first chunk unlocks the second and third, which unlock the fourth through seventh, and so on.

make-deliver-task clj

(make-deliver-task state fold task deliver-fn)

(make-deliver-task state {:keys [post-combiner]} task deliver-fn executor)

Takes a task executor state, a final combine task, and a function which delivers results to the output of a fold. Returns [state' task], where task applies the post-combiner of the fold and calls deliver-fn with the results.

Can also take an optional Folder; we clean up old passes automatically once delivery occurs.

make-fold clj

(make-fold fn-or-map)

(make-fold reducer combiner)

Takes a fold map, or a function, or two functions, and expands them into a full fold map. Generally follows the same rules as transducers: https://clojure.org/reference/transducers#_creating_transducers. With a map:

• :name: default :fold
• :associative?: default false
• :reducer-identity: defaults to :reducer
• :reducer: must be provided
• :post-reducer: defaults to :reducer, or identity if :reducer has no unary arity.
• :combiner-identity: defaults to :combiner
• :combiner: defaults to nil
• :post-combiner defaults to :combiner, or identity if :combiner has no unary arity.

With a single function, works exactly like transducers. Constructs a non-associative fold named :fold with no combiner, and using f for all three reducer functions.

With two functions, uses the first for all three reducers, and the second for all three combiners. This is the opposite arity from clojure.core.reducers/fold, but I really do think it makes more sense, since reduce happens first.

make-join-task clj

(make-join-task state name join-accs a-task b-task)

Takes a task executor state, a name, a function that joins two accumulator, and accumulator tasks a and b. Returns [state' task], where task returns the two accumulators joined. Join tasks keep a vector of :worker tasks they depend on.

make-linear-combine-task clj

(make-linear-combine-task state
                          {:keys [reducer-identity reducer post-reducer
                                  combiner-identity combiner]}
                          chunks
                          last-reduce)

Takes a task executor state, a fold, chunks, and the final reduce task. Returns [state' task] where task is a new task which applies the post-reduce to the last reduced value, then performs a single combine. Or, if combiner is not provided, just does post-reduce.

make-linear-reduce-task clj

(make-linear-reduce-task state fold chunks i)

(make-linear-reduce-task state
                         {:keys [reducer-identity reducer]}
                         chunks
                         i
                         prev-reduce)

Takes a task executor state, a fold, chunks, an index into the chunks, and, when 0 < i, a previous reduce task. Returns [state' task], where task continues the previous task's reduction.

make-split-task clj

(make-split-task state name split-accs i acc-task)

Takes a task executor state, a name, a function that splits an accumulator, an index to extract from that accumulator, and a task producing that accumulator. Returns [state' task], where task returns the given index in the accumulator task.

maybe-get-task clj

(maybe-get-task state task-id)

Gets a task in a state, or passes through nil.

maybe-task-id clj

(maybe-task-id task)

Gets the ID of a task, passing through nil.

pass clj

(pass fold chunks)

Constructs a new linear or concurrent pass map over the given chunks, using the given fold.

pass-str clj

(pass-str state pass-or-tasks)

Generates a string visualizing a pass. In its short form, takes a state and a pass. Or takes a pass and a list of [name tasks] pairs.

pfold clj

(pfold folder fold)

Executes a fold on the given folder asynchronously. See make-fold for what a fold can be. Returns a deref-able result.

reduced-wrapper clj

(reduced-wrapper reducer)

Wraps a reducing function to wrap reduced results in another reduced wrapper, so that we can detect it and terminate early.

run-fold! clj

(run-fold! folder fold)

Takes a Folder and a fold. Runs the fold, returning a deref-able output.

run-pass! clj

(run-pass! f new-pass)

Takes a folder and a pass. Launches the pass on the folder, joining it to an existing pass if possible. Returns newly-running pass.

split-deliver-fn clj

(split-deliver-fn old-pass new-pass)

Takes an old and new pass. Constructs a function which takes an [old-res new-res] pair, or a CapturedThrowable, and delivers them to the old and new folds' deliver fns, respectively.

task-work-pending? clj

(task-work-pending? state task)

We may have a join task which unifies work from two different reducers or combiners, or one which splits a reducer or combiner. If we cancel just the join (split), the reducers (combiners) will still run. This takes a state and a task and returns true if all the actual work involved in this particular chunk is still pending. You can also pass :cancelled as a task; this is always pending.

task-workers clj

(task-workers task)

Takes a Task and returns a vector of tasks which actually do work (e.g. call a reducer or combiner) for this particular chunk. We need this to figure out if tasks are safely cancellable, or if some of their work has begun.

tasks-str clj

(tasks-str state name tasks)

Generates a string visualizing a series of tasks, prefixed with name.

tesser clj

(tesser folder tesser-fold)

Shortcut for running a Tesser uncompiled fold on a folder.

validate-fold clj

(validate-fold fold)

Throws if fold is malformed. Returns fold otherwise.