fn garbage_collect_local_state(
        &self,
        mut state: ReplicatedState,
        stream_slices: &BTreeMap<SubnetId, StreamSlice>,
    ) -> ReplicatedState {
        let mut streams = state.take_streams();
        for (remote_subnet, stream_slice) in stream_slices {
            match streams.get_mut(remote_subnet) {
                Some(stream) => {
                    self.garbage_collect_messages(stream, *remote_subnet, stream_slice);
                }
                None => {
                    // New stream.
                    assert_eq!(
                        stream_slice.header().signals_end,
                        StreamIndex::from(0),
                        "Cannot garbage collect a stream for subnet {} that does not exist",
                        remote_subnet
                    );
                    assert_eq!(
                        stream_slice.header().begin, StreamIndex::from(0),
                        "Signals from subnet {} do not start from 0 in the first communication attempt",
                        remote_subnet
                    );
                }
            }
            let backlog = if let Some(messages) = stream_slice.messages() {
                stream_slice.header().end - messages.end()
            } else {
                stream_slice.header().end - stream_slice.header().begin
            };
            self.metrics
                .xnet_message_backlog
                .with_label_values(&[&remote_subnet.to_string()])
                .set(backlog.get() as i64);
        }
        state.put_streams(streams);
        state
    }
/// Garbage collects outgoing `Streams` based on the signals present in
/// incoming `stream_slices`. As we do not have explicit signals (only a
/// `signals_end` index) there is no need to garbage collect signals.
Garbage collects outgoing `Streams` based on the signals present in
incoming `stream_slices`. As we do not have explicit signals (only a
`signals_end` index) there is no need to garbage collect signals.
Garbage collects outgoing ` Streams ` based on the signals present in incoming ` stream_slices ` . As we do not have explicit signals ( only a ` signals_end ` index ) there is no need to garbage collect signals .
fn garbage_collect_local_state(
        &self,
        mut state: ReplicatedState,
        stream_slices: &BTreeMap<SubnetId, StreamSlice>,
    ) -> ReplicatedState {
        let mut streams = state.take_streams();
        for (remote_subnet, stream_slice) in stream_slices {
            match streams.get_mut(remote_subnet) {
                Some(stream) => {
                    self.garbage_collect_messages(stream, *remote_subnet, stream_slice);
                }
                None => {
                    // New stream.
                    assert_eq!(
                        stream_slice.header().signals_end,
                        StreamIndex::from(0),
                        "Cannot garbage collect a stream for subnet {} that does not exist",
                        remote_subnet
                    );
                    assert_eq!(
                        stream_slice.header().begin, StreamIndex::from(0),
                        "Signals from subnet {} do not start from 0 in the first communication attempt",
                        remote_subnet
                    );
                }
            }
            let backlog = if let Some(messages) = stream_slice.messages() {
                stream_slice.header().end - messages.end()
            } else {
                stream_slice.header().end - stream_slice.header().begin
            };
            self.metrics
                .xnet_message_backlog
                .with_label_values(&[&remote_subnet.to_string()])
                .set(backlog.get() as i64);
        }
        state.put_streams(streams);
        state
    }
fn garbage_collect_local_state ( & self , mut state : ReplicatedState , stream_slices : & BTreeMap < SubnetId , StreamSlice > , ) -> ReplicatedState { let mut streams = state . take_streams ( ) ; for ( remote_subnet , stream_slice ) in stream_slices { match streams . get_mut ( remote_subnet ) { Some ( stream ) => { self . garbage_collect_messages ( stream , * remote_subnet , stream_slice ) ; } None => { // New stream. assert_eq ! ( stream_slice header ( ) signals_end StreamIndex from ( 0 ) " " remote_subnet ) ; assert_eq ! ( stream_slice header ( ) begin StreamIndex from ( 0 ) " " remote_subnet ) ; } } let backlog = if let Some ( messages ) = stream_slice . messages ( ) { stream_slice . header ( ) . end - messages . end ( ) } else { stream_slice . header ( ) . end - stream_slice . header ( ) . begin } ; self . metrics . xnet_message_backlog . with_label_values ( & [ & remote_subnet . to_string ( ) ] ) . set ( backlog . get ( ) as i64 ) ; } state . put_streams ( streams ) ; state }
Garbage collects outgoing `Streams` based on the signals present in
incoming `stream_slices`.
Garbage collects outgoing ` Streams ` based on the signals present in incoming ` stream_slices ` .
&self
state
ReplicatedState
stream_slices
&BTreeMap<SubnetId, StreamSlice>
&self
state
ReplicatedState
stream_slices
&BTreeMap<SubnetId, StreamSlice>