Function backpressure

pub fn backpressure<T, G, O>(
    scope: &mut G,
    name: &str,
    data: &Stream<G, O>,
    flow_control: FlowControl<G>,
    chosen_worker: usize,
    probe: Option<UnboundedSender<(Antichain<(T, Subtime)>, usize, usize)>>,
) -> (Stream<G, O>, PressOnDropButton)
where
    T: TimelyTimestamp + Lattice + Codec64 + TotalOrder,
    G: Scope<Timestamp = (T, Subtime)>,
    O: Backpressureable + Debug,

Apply flow control to the data input, based on the given FlowControl.

The FlowControl should have a progress_stream that is the pristine, unaltered frontier of the downstream operator we want to backpressure from, a max_inflight_bytes, and a summary. Note that the data input expects all the second part of the tuple timestamp to be 0, and all data to be on the chosen_worker worker.

The summary represents the minimum range of timestamps that needs to be emitted before reasoning about max_inflight_bytes. In practice this means that we may overshoot max_inflight_bytes.

The implementation of this operator is very subtle. Many inline comments have been added.