pub fn absorb_response(
            &mut self,
            shard_id: usize,
            message: Response,
        ) -> Box<dyn Iterator<Item = Response> + Send> {
            match message {
                Response::Compute(ComputeResponse::FrontierUppers(mut list), instance) => {
                    assert_eq!(instance, super::DEFAULT_COMPUTE_INSTANCE_ID);
                    for (id, changes) in list.iter_mut() {
                        if let Some(frontier) = self.uppers.get_mut(id) {
                            let iter = frontier.update_iter(changes.drain());
                            changes.extend(iter);
                        } else {
                            changes.clear();
                        }
                    }

                    // The following block implements a `list.retain()` of non-empty change batches.
                    // This is more verbose than `list.retain()` because that method cannot mutate
                    // its argument, and `is_empty()` may need to do this (as it is lazily compacted).
                    let mut cursor = 0;
                    while let Some((_id, changes)) = list.get_mut(cursor) {
                        if changes.is_empty() {
                            list.swap_remove(cursor);
                        } else {
                            cursor += 1;
                        }
                    }

                    Box::new(
                        Some(Response::Compute(
                            ComputeResponse::FrontierUppers(list),
                            instance,
                        ))
                        .into_iter(),
                    )
                }
                // Avoid multiple retractions of minimum time, to present as updates from one worker.
                Response::Storage(StorageResponse::TimestampBindings(mut feedback)) => {
                    for (id, changes) in feedback.changes.iter_mut() {
                        if let Some(frontier) = self.uppers.get_mut(id) {
                            let iter = frontier.update_iter(changes.drain());
                            changes.extend(iter);
                        } else {
                            changes.clear();
                        }
                    }
                    Box::new(
                        Some(Response::Storage(StorageResponse::TimestampBindings(
                            feedback,
                        )))
                        .into_iter(),
                    )
                }
                Response::Compute(
                    ComputeResponse::PeekResponse(connection, response),
                    instance,
                ) => {
                    // Incorporate new peek responses; awaiting all responses.
                    let entry = self
                        .peek_responses
                        .entry(connection)
                        .or_insert_with(Default::default);
                    let novel = entry.insert(shard_id, response);
                    assert!(novel.is_none(), "Duplicate peek response");
                    // We may be ready to respond.
                    if entry.len() == self.parts {
                        let mut response = PeekResponse::Rows(Vec::new());
                        for (_part, r) in std::mem::take(entry).into_iter() {
                            response = match (response, r) {
                                (_, PeekResponse::Canceled) => PeekResponse::Canceled,
                                (PeekResponse::Canceled, _) => PeekResponse::Canceled,
                                (_, PeekResponse::Error(e)) => PeekResponse::Error(e),
                                (PeekResponse::Error(e), _) => PeekResponse::Error(e),
                                (PeekResponse::Rows(mut rows), PeekResponse::Rows(r)) => {
                                    rows.extend(r.into_iter());
                                    PeekResponse::Rows(rows)
                                }
                            };
                        }
                        self.peek_responses.remove(&connection);
                        Box::new(
                            Some(Response::Compute(
                                ComputeResponse::PeekResponse(connection, response),
                                instance,
                            ))
                            .into_iter(),
                        )
                    } else {
                        Box::new(None.into_iter())
                    }
                }
                Response::Compute(ComputeResponse::TailResponse(id, response), instance) => {
                    let maybe_entry = self.pending_tails.entry(id).or_insert_with(|| {
                        Some(PendingTail {
                            frontiers: HashMap::new(),
                            change_batch: Default::default(),
                            buffer: Vec::new(),
                            parts: self.parts,
                            reported_frontier: Antichain::from_elem(0),
                        })
                    });

                    let entry = match maybe_entry {
                        None => {
                            // This tail has been dropped;
                            // we should permanently block
                            // any messages from it
                            return Box::new(None.into_iter());
                        }
                        Some(entry) => entry,
                    };

                    let responses: Box<dyn Iterator<Item = Response> + Send> = match response {
                        TailResponse::Progress(frontier) => {
                            if let Some(new_frontier) = entry.record_progress(shard_id, frontier) {
                                let data = entry.consolidate_up_to(new_frontier.clone());
                                let progress_response = TailResponse::Progress(new_frontier);
                                if data.is_empty() {
                                    Box::new(
                                        Some(Response::Compute(
                                            ComputeResponse::TailResponse(id, progress_response),
                                            instance,
                                        ))
                                        .into_iter(),
                                    )
                                } else {
                                    // Return the data first, then the progress message.
                                    Box::new(
                                        [
                                            Response::Compute(
                                                ComputeResponse::TailResponse(
                                                    id,
                                                    TailResponse::Rows(data),
                                                ),
                                                instance,
                                            ),
                                            Response::Compute(
                                                ComputeResponse::TailResponse(
                                                    id,
                                                    progress_response,
                                                ),
                                                instance,
                                            ),
                                        ]
                                        .into_iter(),
                                    )
                                }
                            } else {
                                Box::new(None.into_iter())
                            }
                        }
                        TailResponse::Rows(rows) => {
                            entry.push_data(rows);
                            Box::new(None.into_iter())
                        }
                        TailResponse::Dropped => {
                            *maybe_entry = None;
                            Box::new(
                                Some(Response::Compute(
                                    ComputeResponse::TailResponse(id, TailResponse::Dropped),
                                    instance,
                                ))
                                .into_iter(),
                            )
                        }
                    };
                    responses
                }
            }
        }
/// Absorbs a response, and produces response that should be emitted.
Absorbs a response, and produces response that should be emitted.
Absorbs a response and produces response that should be emitted .
pub fn absorb_response(
            &mut self,
            shard_id: usize,
            message: Response,
        ) -> Box<dyn Iterator<Item = Response> + Send> {
            match message {
                Response::Compute(ComputeResponse::FrontierUppers(mut list), instance) => {
                    assert_eq!(instance, super::DEFAULT_COMPUTE_INSTANCE_ID);
                    for (id, changes) in list.iter_mut() {
                        if let Some(frontier) = self.uppers.get_mut(id) {
                            let iter = frontier.update_iter(changes.drain());
                            changes.extend(iter);
                        } else {
                            changes.clear();
                        }
                    }
                    // The following block implements a `list.retain()` of non-empty change batches.
                    // This is more verbose than `list.retain()` because that method cannot mutate
                    // its argument, and `is_empty()` may need to do this (as it is lazily compacted).
                    let mut cursor = 0;
                    while let Some((_id, changes)) = list.get_mut(cursor) {
                        if changes.is_empty() {
                            list.swap_remove(cursor);
                        } else {
                            cursor += 1;
                        }
                    }
                    Box::new(
                        Some(Response::Compute(
                            ComputeResponse::FrontierUppers(list),
                            instance,
                        ))
                        .into_iter(),
                    )
                }
                // Avoid multiple retractions of minimum time, to present as updates from one worker.
                Response::Storage(StorageResponse::TimestampBindings(mut feedback)) => {
                    for (id, changes) in feedback.changes.iter_mut() {
                        if let Some(frontier) = self.uppers.get_mut(id) {
                            let iter = frontier.update_iter(changes.drain());
                            changes.extend(iter);
                        } else {
                            changes.clear();
                        }
                    }
                    Box::new(
                        Some(Response::Storage(StorageResponse::TimestampBindings(
                            feedback,
                        )))
                        .into_iter(),
                    )
                }
                Response::Compute(
                    ComputeResponse::PeekResponse(connection, response),
                    instance,
                ) => {
                    // Incorporate new peek responses; awaiting all responses.
                    let entry = self
                        .peek_responses
                        .entry(connection)
                        .or_insert_with(Default::default);
                    let novel = entry.insert(shard_id, response);
                    assert!(novel.is_none(), "Duplicate peek response");
                    // We may be ready to respond.
                    if entry.len() == self.parts {
                        let mut response = PeekResponse::Rows(Vec::new());
                        for (_part, r) in std::mem::take(entry).into_iter() {
                            response = match (response, r) {
                                (_, PeekResponse::Canceled) => PeekResponse::Canceled,
                                (PeekResponse::Canceled, _) => PeekResponse::Canceled,
                                (_, PeekResponse::Error(e)) => PeekResponse::Error(e),
                                (PeekResponse::Error(e), _) => PeekResponse::Error(e),
                                (PeekResponse::Rows(mut rows), PeekResponse::Rows(r)) => {
                                    rows.extend(r.into_iter());
                                    PeekResponse::Rows(rows)
                                }
                            };
                        }
                        self.peek_responses.remove(&connection);
                        Box::new(
                            Some(Response::Compute(
                                ComputeResponse::PeekResponse(connection, response),
                                instance,
                            ))
                            .into_iter(),
                        )
                    } else {
                        Box::new(None.into_iter())
                    }
                }
                Response::Compute(ComputeResponse::TailResponse(id, response), instance) => {
                    let maybe_entry = self.pending_tails.entry(id).or_insert_with(|| {
                        Some(PendingTail {
                            frontiers: HashMap::new(),
                            change_batch: Default::default(),
                            buffer: Vec::new(),
                            parts: self.parts,
                            reported_frontier: Antichain::from_elem(0),
                        })
                    });
                    let entry = match maybe_entry {
                        None => {
                            // This tail has been dropped;
                            // we should permanently block
                            // any messages from it
                            return Box::new(None.into_iter());
                        }
                        Some(entry) => entry,
                    };
                    let responses: Box<dyn Iterator<Item = Response> + Send> = match response {
                        TailResponse::Progress(frontier) => {
                            if let Some(new_frontier) = entry.record_progress(shard_id, frontier) {
                                let data = entry.consolidate_up_to(new_frontier.clone());
                                let progress_response = TailResponse::Progress(new_frontier);
                                if data.is_empty() {
                                    Box::new(
                                        Some(Response::Compute(
                                            ComputeResponse::TailResponse(id, progress_response),
                                            instance,
                                        ))
                                        .into_iter(),
                                    )
                                } else {
                                    // Return the data first, then the progress message.
                                    Box::new(
                                        [
                                            Response::Compute(
                                                ComputeResponse::TailResponse(
                                                    id,
                                                    TailResponse::Rows(data),
                                                ),
                                                instance,
                                            ),
                                            Response::Compute(
                                                ComputeResponse::TailResponse(
                                                    id,
                                                    progress_response,
                                                ),
                                                instance,
                                            ),
                                        ]
                                        .into_iter(),
                                    )
                                }
                            } else {
                                Box::new(None.into_iter())
                            }
                        }
                        TailResponse::Rows(rows) => {
                            entry.push_data(rows);
                            Box::new(None.into_iter())
                        }
                        TailResponse::Dropped => {
                            *maybe_entry = None;
                            Box::new(
                                Some(Response::Compute(
                                    ComputeResponse::TailResponse(id, TailResponse::Dropped),
                                    instance,
                                ))
                                .into_iter(),
                            )
                        }
                    };
                    responses
                }
            }
        }
pub fn absorb_response ( & mut self , shard_id : usize , message : Response , ) -> Box < dyn Iterator < Item = Response > + Send > { match message { Response :: Compute ( ComputeResponse :: FrontierUppers ( mut list ) , instance ) => { assert_eq ! ( instance super DEFAULT_COMPUTE_INSTANCE_ID ) ; for ( id , changes ) in list . iter_mut ( ) { if let Some ( frontier ) = self . uppers . get_mut ( id ) { let iter = frontier . update_iter ( changes . drain ( ) ) ; changes . extend ( iter ) ; } else { changes . clear ( ) ; } } // The following block implements a `list.retain()` of non-empty change batches. // This is more verbose than `list.retain()` because that method cannot mutate // its argument, and `is_empty()` may need to do this (as it is lazily compacted). let mut cursor = 0 ; while let Some ( ( _id , changes ) ) = list . get_mut ( cursor ) { if changes . is_empty ( ) { list . swap_remove ( cursor ) ; } else { cursor += 1 ; } } Box :: new ( Some ( Response :: Compute ( ComputeResponse :: FrontierUppers ( list ) , instance , ) ) . into_iter ( ) , ) } // Avoid multiple retractions of minimum time, to present as updates from one worker. Response :: Storage ( StorageResponse :: TimestampBindings ( mut feedback ) ) => { for ( id , changes ) in feedback . changes . iter_mut ( ) { if let Some ( frontier ) = self . uppers . get_mut ( id ) { let iter = frontier . update_iter ( changes . drain ( ) ) ; changes . extend ( iter ) ; } else { changes . clear ( ) ; } } Box :: new ( Some ( Response :: Storage ( StorageResponse :: TimestampBindings ( feedback , ) ) ) . into_iter ( ) , ) } Response :: Compute ( ComputeResponse :: PeekResponse ( connection , response ) , instance , ) => { // Incorporate new peek responses; awaiting all responses. let entry = self . peek_responses . entry ( connection ) . or_insert_with ( Default :: default ) ; let novel = entry . insert ( shard_id , response ) ; assert ! ( novel is_none ( ) " " ) ; // We may be ready to respond. if entry . len ( ) == self . parts { let mut response = PeekResponse :: Rows ( Vec :: new ( ) ) ; for ( _part , r ) in std :: mem :: take ( entry ) . into_iter ( ) { response = match ( response , r ) { ( _ , PeekResponse :: Canceled ) => PeekResponse :: Canceled , ( PeekResponse :: Canceled , _ ) => PeekResponse :: Canceled , ( _ , PeekResponse :: Error ( e ) ) => PeekResponse :: Error ( e ) , ( PeekResponse :: Error ( e ) , _ ) => PeekResponse :: Error ( e ) , ( PeekResponse :: Rows ( mut rows ) , PeekResponse :: Rows ( r ) ) => { rows . extend ( r . into_iter ( ) ) ; PeekResponse :: Rows ( rows ) } } ; } self . peek_responses . remove ( & connection ) ; Box :: new ( Some ( Response :: Compute ( ComputeResponse :: PeekResponse ( connection , response ) , instance , ) ) . into_iter ( ) , ) } else { Box :: new ( None . into_iter ( ) ) } } Response :: Compute ( ComputeResponse :: TailResponse ( id , response ) , instance ) => { let maybe_entry = self . pending_tails . entry ( id ) . or_insert_with ( | | { Some ( PendingTail { frontiers : HashMap :: new ( ) , change_batch : Default :: default ( ) , buffer : Vec :: new ( ) , parts : self . parts , reported_frontier : Antichain :: from_elem ( 0 ) , } ) } ) ; let entry = match maybe_entry { None => { // This tail has been dropped; // we should permanently block // any messages from it return Box :: new ( None . into_iter ( ) ) ; } Some ( entry ) => entry , } ; let responses : Box < dyn Iterator < Item = Response > + Send > = match response { TailResponse :: Progress ( frontier ) => { if let Some ( new_frontier ) = entry . record_progress ( shard_id , frontier ) { let data = entry . consolidate_up_to ( new_frontier . clone ( ) ) ; let progress_response = TailResponse :: Progress ( new_frontier ) ; if data . is_empty ( ) { Box :: new ( Some ( Response :: Compute ( ComputeResponse :: TailResponse ( id , progress_response ) , instance , ) ) . into_iter ( ) , ) } else { // Return the data first, then the progress message. Box :: new ( [ Response :: Compute ( ComputeResponse :: TailResponse ( id , TailResponse :: Rows ( data ) , ) , instance , ) , Response :: Compute ( ComputeResponse :: TailResponse ( id , progress_response , ) , instance , ) , ] . into_iter ( ) , ) } } else { Box :: new ( None . into_iter ( ) ) } } TailResponse :: Rows ( rows ) => { entry . push_data ( rows ) ; Box :: new ( None . into_iter ( ) ) } TailResponse :: Dropped => { * maybe_entry = None ; Box :: new ( Some ( Response :: Compute ( ComputeResponse :: TailResponse ( id , TailResponse :: Dropped ) , instance , ) ) . into_iter ( ) , ) } } ; responses } } }
Absorbs a response, and produces response that should be emitted.
Absorbs a response and produces response that should be emitted .
&mut self
shard_id
usize
message
Response