pub fn trim<G, N, R>(graph: &Collection<G, (N,N), R>) -> Collection<G, (N,N), R>
where
    G: Scope,
    G::Timestamp: Lattice+Ord,
    N: ExchangeData+Hash,
    R: ExchangeData + Abelian,
    R: Multiply<R, Output=R>,
    R: From<i8>,
{
    graph.iterate(|edges| {
        // keep edges from active edge destinations.
        let active =
        edges.map(|(_src,dst)| dst)
             .threshold(|_,c| if c.is_zero() { R::from(0_i8) } else { R::from(1_i8) });

        graph.enter(&edges.scope())
             .semijoin(&active)
    })
}
Iteratively removes nodes with no in-edges.
differential-dataflow/src/algorithms/graphs/scc.rs