fn pp_broadcast<G, D, B, F>(
    ranges: Collection<G, ((usize, usize), D)>, 
    seed: B,
    zero: D,
    combine: F) -> Collection<G, (usize, B)>
where
    G: Scope,
    G::Timestamp: Lattice+Ord+::std::fmt::Debug,
    D: Data,
    B: Data+::std::hash::Hash,
    F: Fn(&B, &D) -> B + 'static,
{
    // Each range proposes an empty first child, to provide for its second child if it has no sibling.
    // This is important if we want to reconstruct 
    let zero_ranges =
        ranges
            .map(move |((pos, log),_)| ((pos, if log > 0 { log - 1 } else { 0 }), zero.clone()))
            .antijoin(&ranges.map(|((pos, log),_)| (pos, log)));

    let aggregates = ranges.concat(&zero_ranges);

    let init_state = 
    Some(((0, seed), Default::default(), 1))
        .to_stream(&mut aggregates.scope())
        .as_collection();

    init_state
        .iterate(|state| {
            aggregates
                .filter(|&((_, log),_)| log < 64)    // the log = 64 interval doesn't help us here (overflows).
                .enter(&state.scope())
                .map(|((pos, log), data)| (pos, (log, data)))
                .join_map(state, move |&pos, &(log, ref data), state| (pos + (1 << log), combine(state, data)))
                .concat(&init_state.enter(&state.scope()))
                .distinct()
        })
        .consolidate()
}
Produces the accumulated values at each of the `usize` locations in `aggregates` (and others).
advent_of_code_2017/src/bin/day_09.rs