fn report_frontiers(&mut self) {
        fn add_progress(
            id: GlobalId,
            new_frontier: &Antichain<Timestamp>,
            prev_frontier: &Antichain<Timestamp>,
            progress: &mut Vec<(GlobalId, ChangeBatch<Timestamp>)>,
        ) {
            let mut changes = ChangeBatch::new();
            for time in prev_frontier.elements().iter() {
                changes.update(time.clone(), -1);
            }
            for time in new_frontier.elements().iter() {
                changes.update(time.clone(), 1);
            }
            changes.compact();
            if !changes.is_empty() {
                progress.push((id, changes));
            }
        }

        let mut new_frontier = Antichain::new();
        let mut progress = Vec::new();
        for (id, traces) in self.render_state.traces.traces.iter_mut() {
            // Read the upper frontier and compare to what we've reported.
            traces.oks_mut().read_upper(&mut new_frontier);
            let prev_frontier = self
                .reported_frontiers
                .get_mut(&id)
                .expect("Index frontier missing!");
            if prev_frontier != &new_frontier {
                add_progress(*id, &new_frontier, &prev_frontier, &mut progress);
                prev_frontier.clone_from(&new_frontier);
            }
        }

        // Log index frontier changes
        if let Some(logger) = self.materialized_logger.as_mut() {
            for (id, changes) in &mut progress {
                for (time, diff) in changes.iter() {
                    logger.log(MaterializedEvent::Frontier(*id, *time, *diff));
                }
            }
        }

        for (id, history) in self.render_state.ts_histories.iter() {
            // Read the upper frontier and compare to what we've reported.
            history.read_upper(&mut new_frontier);
            let prev_frontier = self
                .reported_frontiers
                .get_mut(&id)
                .expect("Source frontier missing!");
            assert!(<_ as PartialOrder>::less_equal(
                prev_frontier,
                &new_frontier
            ));
            if prev_frontier != &new_frontier {
                add_progress(*id, &new_frontier, &prev_frontier, &mut progress);
                prev_frontier.clone_from(&new_frontier);
            }
        }

        for (id, frontier) in self.render_state.sink_write_frontiers.iter() {
            new_frontier.clone_from(&frontier.borrow());
            let prev_frontier = self
                .reported_frontiers
                .get_mut(&id)
                .expect("Sink frontier missing!");
            assert!(<_ as PartialOrder>::less_equal(
                prev_frontier,
                &new_frontier
            ));
            if prev_frontier != &new_frontier {
                add_progress(*id, &new_frontier, &prev_frontier, &mut progress);
                prev_frontier.clone_from(&new_frontier);
            }
        }

        if !progress.is_empty() {
            self.send_response(Response::FrontierUppers(progress));
        }
    }
/// Send progress information to the coordinator.
Send progress information to the coordinator.
Send progress information to the coordinator .
fn report_frontiers(&mut self) {
        fn add_progress(
            id: GlobalId,
            new_frontier: &Antichain<Timestamp>,
            prev_frontier: &Antichain<Timestamp>,
            progress: &mut Vec<(GlobalId, ChangeBatch<Timestamp>)>,
        ) {
            let mut changes = ChangeBatch::new();
            for time in prev_frontier.elements().iter() {
                changes.update(time.clone(), -1);
            }
            for time in new_frontier.elements().iter() {
                changes.update(time.clone(), 1);
            }
            changes.compact();
            if !changes.is_empty() {
                progress.push((id, changes));
            }
        }
        let mut new_frontier = Antichain::new();
        let mut progress = Vec::new();
        for (id, traces) in self.render_state.traces.traces.iter_mut() {
            // Read the upper frontier and compare to what we've reported.
            traces.oks_mut().read_upper(&mut new_frontier);
            let prev_frontier = self
                .reported_frontiers
                .get_mut(&id)
                .expect("Index frontier missing!");
            if prev_frontier != &new_frontier {
                add_progress(*id, &new_frontier, &prev_frontier, &mut progress);
                prev_frontier.clone_from(&new_frontier);
            }
        }
        // Log index frontier changes
        if let Some(logger) = self.materialized_logger.as_mut() {
            for (id, changes) in &mut progress {
                for (time, diff) in changes.iter() {
                    logger.log(MaterializedEvent::Frontier(*id, *time, *diff));
                }
            }
        }
        for (id, history) in self.render_state.ts_histories.iter() {
            // Read the upper frontier and compare to what we've reported.
            history.read_upper(&mut new_frontier);
            let prev_frontier = self
                .reported_frontiers
                .get_mut(&id)
                .expect("Source frontier missing!");
            assert!(<_ as PartialOrder>::less_equal(
                prev_frontier,
                &new_frontier
            ));
            if prev_frontier != &new_frontier {
                add_progress(*id, &new_frontier, &prev_frontier, &mut progress);
                prev_frontier.clone_from(&new_frontier);
            }
        }
        for (id, frontier) in self.render_state.sink_write_frontiers.iter() {
            new_frontier.clone_from(&frontier.borrow());
            let prev_frontier = self
                .reported_frontiers
                .get_mut(&id)
                .expect("Sink frontier missing!");
            assert!(<_ as PartialOrder>::less_equal(
                prev_frontier,
                &new_frontier
            ));
            if prev_frontier != &new_frontier {
                add_progress(*id, &new_frontier, &prev_frontier, &mut progress);
                prev_frontier.clone_from(&new_frontier);
            }
        }
        if !progress.is_empty() {
            self.send_response(Response::FrontierUppers(progress));
        }
    }
fn report_frontiers ( & mut self ) { fn add_progress ( id : GlobalId , new_frontier : & Antichain < Timestamp > , prev_frontier : & Antichain < Timestamp > , progress : & mut Vec < ( GlobalId , ChangeBatch < Timestamp > ) > , ) { let mut changes = ChangeBatch :: new ( ) ; for time in prev_frontier . elements ( ) . iter ( ) { changes . update ( time . clone ( ) , - 1 ) ; } for time in new_frontier . elements ( ) . iter ( ) { changes . update ( time . clone ( ) , 1 ) ; } changes . compact ( ) ; if ! changes . is_empty ( ) { progress . push ( ( id , changes ) ) ; } } let mut new_frontier = Antichain :: new ( ) ; let mut progress = Vec :: new ( ) ; for ( id , traces ) in self . render_state . traces . traces . iter_mut ( ) { // Read the upper frontier and compare to what we've reported. traces . oks_mut ( ) . read_upper ( & mut new_frontier ) ; let prev_frontier = self . reported_frontiers . get_mut ( & id ) . expect ( " " ) ; if prev_frontier != & new_frontier { add_progress ( * id , & new_frontier , & prev_frontier , & mut progress ) ; prev_frontier . clone_from ( & new_frontier ) ; } } // Log index frontier changes if let Some ( logger ) = self . materialized_logger . as_mut ( ) { for ( id , changes ) in & mut progress { for ( time , diff ) in changes . iter ( ) { logger . log ( MaterializedEvent :: Frontier ( * id , * time , * diff ) ) ; } } } for ( id , history ) in self . render_state . ts_histories . iter ( ) { // Read the upper frontier and compare to what we've reported. history . read_upper ( & mut new_frontier ) ; let prev_frontier = self . reported_frontiers . get_mut ( & id ) . expect ( " " ) ; assert ! ( as PartialOrder less_equal ( prev_frontier new_frontier ) ) ; if prev_frontier != & new_frontier { add_progress ( * id , & new_frontier , & prev_frontier , & mut progress ) ; prev_frontier . clone_from ( & new_frontier ) ; } } for ( id , frontier ) in self . render_state . sink_write_frontiers . iter ( ) { new_frontier . clone_from ( & frontier . borrow ( ) ) ; let prev_frontier = self . reported_frontiers . get_mut ( & id ) . expect ( " " ) ; assert ! ( as PartialOrder less_equal ( prev_frontier new_frontier ) ) ; if prev_frontier != & new_frontier { add_progress ( * id , & new_frontier , & prev_frontier , & mut progress ) ; prev_frontier . clone_from ( & new_frontier ) ; } } if ! progress . is_empty ( ) { self . send_response ( Response :: FrontierUppers ( progress ) ) ; } }
Send progress information to the coordinator.
Send progress information to the coordinator .
&mut self
&mut self