fn on_frame_ack()

in dc/s2n-quic-dc/src/stream/send/state.rs [430:552]

• 105 lines of code
• 10 McCabe index (conditional complexity)

    fn on_frame_ack<Ack, Clk, const IS_STREAM: bool>(
        &mut self,
        credentials: &Credentials,
        ack: &frame::Ack<Ack>,
        random: &mut dyn random::Generator,
        clock: &Clk,
        newly_acked: &mut bool,
        max_acked_stream: &mut Option<VarInt>,
        max_acked_recovery: &mut Option<VarInt>,
        segment_alloc: &buffer::Allocator,
    ) -> Result<(), Error>
    where
        Ack: frame::ack::AckRanges,
        Clk: Clock,
    {
        let mut cca_args = None;
        let mut bytes_acked = 0;

        macro_rules! impl_ack_processing {
            ($space:ident, $sent_packets:ident, $on_packet_number:expr) => {
                for range in ack.ack_ranges() {
                    let pmin = PacketNumberSpace::Initial.new_packet_number(*range.start());
                    let pmax = PacketNumberSpace::Initial.new_packet_number(*range.end());
                    let range = s2n_quic_core::packet::number::PacketNumberRange::new(pmin, pmax);
                    for (num, packet) in self.$sent_packets.remove_range(range) {
                        let num_varint = unsafe { VarInt::new_unchecked(num.as_u64()) };

                        #[allow(clippy::redundant_closure_call)]
                        ($on_packet_number)(num_varint, &packet);

                        let _ = self.unacked_ranges.remove(packet.info.tracking_range());

                        self.ecn
                            .on_packet_ack(packet.info.time_sent, packet.info.ecn);
                        bytes_acked += packet.info.cca_len() as usize;

                        // record the most recent packet
                        if cca_args
                            .as_ref()
                            .map_or(true, |prev: &(Timestamp, _)| prev.0 < packet.info.time_sent)
                        {
                            cca_args = Some((packet.info.time_sent, packet.cc_info));
                        }

                        // free the retransmission segment
                        if let Some(segment) = packet.info.retransmission {
                            if let Some(segment) = self.stream_packet_buffers.remove(segment) {
                                // push the segment so the application can reuse it
                                if segment.capacity() >= self.max_sent_segment_size as usize {
                                    segment_alloc.free(segment);
                                }
                            }
                        }

                        probes::on_packet_ack(
                            credentials.id,
                            credentials.key_id,
                            stream::PacketSpace::$space,
                            num.as_u64(),
                            packet.info.packet_len,
                            packet.info.stream_offset,
                            packet.info.payload_len,
                            clock
                                .get_time()
                                .saturating_duration_since(packet.info.time_sent),
                        );

                        *newly_acked = true;
                    }
                }
            };
        }

        if IS_STREAM {
            impl_ack_processing!(
                Stream,
                sent_stream_packets,
                |packet_number: VarInt, _packet: &SentStreamPacket| {
                    *max_acked_stream = (*max_acked_stream).max(Some(packet_number));
                }
            );
        } else {
            impl_ack_processing!(
                Recovery,
                sent_recovery_packets,
                |packet_number: VarInt, sent_packet: &SentRecoveryPacket| {
                    *max_acked_recovery = (*max_acked_recovery).max(Some(packet_number));
                    *max_acked_stream =
                        (*max_acked_stream).max(Some(sent_packet.max_stream_packet_number));

                    // increase the max stream packet if this was a probe
                    if sent_packet.info.retransmission.is_none() {
                        self.max_stream_packet_number = self
                            .max_stream_packet_number
                            .max(sent_packet.max_stream_packet_number + 1);
                    }
                }
            );
        };

        if let Some((time_sent, cc_info)) = cca_args {
            let rtt_sample = clock.get_time().saturating_duration_since(time_sent);

            self.rtt_estimator.update_rtt(
                ack.ack_delay(),
                rtt_sample,
                clock.get_time(),
                true,
                PacketNumberSpace::ApplicationData,
            );

            self.cca.on_packet_ack(
                cc_info.first_sent_time,
                bytes_acked,
                cc_info,
                &self.rtt_estimator,
                random,
                clock.get_time(),
            );
        }

        Ok(())
    }