perf(transport/recovery): optimize SentPackets::take_ranges (#2245)

* bench: add SentPackets::take_ranges benchmark

* perf(transport/recovery): optimize SentPackets::take_range

`SentPackets` keep track of all inflight packets. On receiving an ACK, the acked
packet ranges are removed via `SentPackets:take_ranges`.

In normal network scenarios one can assume that the amount of packets before a
range is small (e.g. due to reordering) or zero, while the amount of packets
after a range is large, i.e. all remaining in-flight packets.

The previous implementation assumed the opposite, leading to an overhead linear
to the amount of packets after the range. This commit changes the algorithm such
that it is linear to the amount of packets before the range, which again, is
assumed to be smaller than the amount of packets after the acked range.

* Update neqo-transport/benches/sent_packets.rs

Co-authored-by: Martin Thomson <[email protected]>
Signed-off-by: Max Inden <[email protected]>

* Update neqo-transport/src/recovery/sent.rs

Co-authored-by: Martin Thomson <[email protected]>
Signed-off-by: Max Inden <[email protected]>

* debug_assert for descending ack ranges

* single debug_assert line

* Document large `packets` in front of range scenario

Co-authored-by: Martin Thomson <[email protected]>

* Trigger benchmarks

---------

Signed-off-by: Max Inden <[email protected]>
Co-authored-by: Martin Thomson <[email protected]>

neqo-transport/src/recovery/sent.rs

@@ -204,7 +204,7 @@ impl SentPackets {
         self.packets.values_mut()
     }
 
-    /// Take values from a specified ranges of packet numbers.
+    /// Take values from specified ranges of packet numbers.
     /// The values returned will be reversed, so that the most recent packet appears first.
     /// This is because ACK frames arrive with ranges starting from the largest acknowledged
     /// and we want to match that.
@@ -214,19 +214,58 @@ impl SentPackets {
         R::IntoIter: ExactSizeIterator,
     {
         let mut result = Vec::new();
-        // Remove all packets. We will add them back as we don't need them.
+
+        // Start with all packets. We will add unacknowledged packets back.
+        //  [---------------------------packets----------------------------]
         let mut packets = std::mem::take(&mut self.packets);
+
+        let mut previous_range_start: Option<PacketNumber> = None;
+
         for range in acked_ranges {
-            // For each acked range, split off the acknowledged part,
-            // then split off the part that hasn't been acknowledged.
-            // This order works better when processing ranges that
-            // have already been processed, which is common.
-            let mut acked = packets.split_off(range.start());
-            let keep = acked.split_off(&(*range.end() + 1));
-            self.packets.extend(keep);
-            result.extend(acked.into_values().rev());
+            // Split off at the end of the acked range.
+            //
+            //  [---------packets--------][----------after_acked_range---------]
+            let after_acked_range = packets.split_off(&(*range.end() + 1));
+
+            // Split off at the start of the acked range.
+            //
+            //  [-packets-][-acked_range-][----------after_acked_range---------]
+            let acked_range = packets.split_off(range.start());
+
+            // According to RFC 9000 19.3.1 ACK ranges are in descending order:
+            //
+            // > Each ACK Range consists of alternating Gap and ACK Range Length
+            // > values in **descending packet number order**.
+            //
+            // <https://www.rfc-editor.org/rfc/rfc9000.html#section-19.3.1>
+            debug_assert!(previous_range_start.map_or(true, |s| s > *range.end()));
+            previous_range_start = Some(*range.start());
+
+            // Thus none of the following ACK ranges will acknowledge packets in
+            // `after_acked_range`. Let's put those back early.
+            //
+            //  [-packets-][-acked_range-][------------self.packets------------]
+            if self.packets.is_empty() {
+                // Don't re-insert un-acked packets into empty collection, but
+                // instead replace the empty one entirely.
+                self.packets = after_acked_range;
+            } else {
+                // Need to extend existing one. Not the first iteration, thus
+                // `after_acked_range` should be small.
+                self.packets.extend(after_acked_range);
+            }
+
+            // Take the acked packets.
+            result.extend(acked_range.into_values().rev());
         }
+
+        // Put remaining non-acked packets back.
+        //
+        // This is inefficient if the acknowledged packets include the last sent
+        // packet AND there is a large unacknowledged span of packets. That's
+        // rare enough that we won't do anything special for that case.
         self.packets.extend(packets);
+
         result
     }