Merge branch 'main' into test-more-migration

Signed-off-by: Lars Eggert <[email protected]>

.config/nextest.toml

@@ -0,0 +1,5 @@
+[profile.ci]
+# Do not cancel the test run on the first failure.
+fail-fast = false
+# Terminate test after three slow periods of 60 seconds.
+slow-timeout = { period = "60s", terminate-after = 3 }

.github/workflows/bench.yml

@@ -87,8 +87,10 @@ jobs:
           #
           # Run all benchmarks at elevated priority.
           taskset -c 0 nice -n -20 cargo "+$TOOLCHAIN" bench --workspace --exclude neqo-bin --features bench -- --noplot | tee results.txt
-          nice -n -20 cargo "+$TOOLCHAIN" bench --package neqo-bin --features bench -- --noplot | tee -a results.txt
-
+          for MTU in 1500 65536; do
+            sudo ip link set dev lo mtu "$MTU"
+            MTU=$MTU nice -n -20 cargo "+$TOOLCHAIN" bench --package neqo-bin --features bench -- --noplot | tee -a results.txt
+          done
 
       # Compare various configurations of neqo against msquic, and gather perf data
       # during the hyperfine runs.
@@ -132,50 +134,57 @@ jobs:
             fi
           }
 
-          for server in msquic neqo; do
-            for client in msquic neqo; do
-              # msquic doesn't let us configure the congestion control or pacing.
-              if [ "$client" == "msquic" ] && [ "$server" == "msquic" ]; then
-                cc_opt=("")
-                pacing_opt=("")
-              else
-                cc_opt=("reno" "cubic")
-                pacing_opt=("on" "")
-              fi
-              for cc in "${cc_opt[@]}"; do
-                for pacing in "${pacing_opt[@]}"; do
-                  # Make a tag string for this test, for the results.
-                  TAG="$client,$server,$cc,$pacing"
-                  echo "Running benchmarks for $TAG" | tee -a comparison.txt
-                  transmogrify "${server_cmd[$server]}" "$cc" "$pacing"
-                  # shellcheck disable=SC2086
-                  taskset -c 0 nice -n -20 \
-                    perf $PERF_OPT -o "$client-$server$EXT.server.perf" $CMD &
-                  PID=$!
-                  transmogrify "${client_cmd[$client]}" "$cc" "$pacing"
-                  # shellcheck disable=SC2086
-                  taskset -c 1 nice -n -20 \
-                    perf $PERF_OPT -o "$client-$server$EXT.client.perf" \
-                      hyperfine -N --output null -w 1 -s "sleep 1" -n "$TAG" -u millisecond --export-markdown step.md "$CMD" |
-                      tee -a comparison.txt
-                  echo >> comparison.txt
-                  kill $PID
-                  cat step.md >> steps.md
-                  # Sanity check the size of the last retrieved file.
-                  [ "$(wc -c <"$SIZE")" -eq "$SIZE" ] || exit 1
+          # See https://github.com/microsoft/msquic/issues/4618#issuecomment-2422611592
+          for mtu in 1504 65536; do
+            sudo ip link set dev lo mtu "$mtu"
+            for server in msquic neqo; do
+              for client in msquic neqo; do
+                # msquic doesn't let us configure the congestion control or pacing.
+                if [ "$client" == "msquic" ] && [ "$server" == "msquic" ]; then
+                  cc_opt=("")
+                  pacing_opt=("")
+                else
+                  cc_opt=("reno" "cubic")
+                  pacing_opt=("on" "")
+                fi
+                for cc in "${cc_opt[@]}"; do
+                  for pacing in "${pacing_opt[@]}"; do
+                    # Make a tag string for this test, for the results.
+                    TAG="$client,$server,$cc,$pacing,$mtu"
+                    echo "Running benchmarks for $TAG" | tee -a comparison.txt
+                    transmogrify "${server_cmd[$server]}" "$cc" "$pacing"
+                    # shellcheck disable=SC2086
+                    taskset -c 0 nice -n -20 \
+                      perf $PERF_OPT -o "$client-$server$EXT.server.perf" $CMD &
+                    PID=$!
+                    transmogrify "${client_cmd[$client]}" "$cc" "$pacing"
+                    # shellcheck disable=SC2086
+                    taskset -c 1 nice -n -20 \
+                      perf $PERF_OPT -o "$client-$server$EXT.client.perf" \
+                        hyperfine -N --output null -w 1 -s "sleep 1" -n "$TAG" -u millisecond --export-markdown step.md "$CMD" |
+                        tee -a comparison.txt
+                    echo >> comparison.txt
+                    kill $PID
+                    cat step.md >> steps.md
+                    # Sanity check the size of the last retrieved file.
+                    [ "$(wc -c <"$SIZE")" -eq "$SIZE" ] || exit 1
+                  done
                 done
               done
             done
           done
           # Merge the results tables generated by hyperfine into a single table.
           echo "Transfer of $SIZE bytes over loopback." > comparison.md
           awk '(!/^\| Command/ || !c++) && (!/^\|:/ || !d++)' < steps.md |\
-            sed -E 's/`//g; s/^\|:/\|:---\|:---\|:---\|:/g; s/,/ \| /g; s/^\| Command/\| Client \| Server \| CC \| Pacing/g' >> comparison.md
+            sed -E 's/`//g; s/^\|:/\|:---\|:---\|:---\|:---\|:/g; s/,/ \| /g; s/^\| Command/\| Client \| Server \| CC \| Pacing \| MTU/g' >> comparison.md
           rm -r "$TMP"
 
       # Re-enable turboboost, hyperthreading and use powersave governor.
       - name: Restore machine
-        run: sudo /root/bin/unprep.sh
+        run: |
+          sudo /root/bin/unprep.sh
+          # In case the previous test failed:
+          sudo ip link set dev lo mtu 65536
         if: success() || failure() || cancelled()
 
       - name: Post-process perf data

.github/workflows/check.yml

@@ -17,7 +17,6 @@ on:
 env:
   CARGO_TERM_COLOR: always
   RUST_BACKTRACE: 1
-  DUMP_SIMULATION_SEEDS: /tmp/simulation-seeds
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref_name }}
@@ -86,11 +85,13 @@ jobs:
         env:
           RUST_LOG: trace
         run: |
+          DUMP_SIMULATION_SEEDS="$(pwd)/simulation-seeds"
+          export DUMP_SIMULATION_SEEDS
           # shellcheck disable=SC2086
           if [ "${{ matrix.rust-toolchain }}" == "stable" ]; then
-            cargo +${{ matrix.rust-toolchain }} llvm-cov nextest $BUILD_TYPE --features ci --no-fail-fast --lcov --output-path lcov.info
+            cargo +${{ matrix.rust-toolchain }} llvm-cov nextest $BUILD_TYPE --features ci --profile ci --lcov --output-path lcov.info
           else
-            cargo +${{ matrix.rust-toolchain }} nextest run $BUILD_TYPE --features ci --no-fail-fast
+            cargo +${{ matrix.rust-toolchain }} nextest run $BUILD_TYPE --features ci --profile ci
           fi
 
       - name: Run client/server transfer
@@ -121,11 +122,11 @@ jobs:
         if: matrix.type == 'debug' && matrix.rust-toolchain == 'stable'
 
       - name: Save simulation seeds artifact
-        if: env.DUMP_SIMULATION_SEEDS
+        if: always()
         uses: actions/upload-artifact@b4b15b8c7c6ac21ea08fcf65892d2ee8f75cf882 # v4.4.3
         with:
           name: simulation-seeds-${{ matrix.os }}-${{ matrix.rust-toolchain }}-${{ matrix.type }}
-          path: ${{ env.DUMP_SIMULATION_SEEDS }}
+          path: simulation-seeds
           compression-level: 9
 
   bench:

.github/workflows/sanitize.yml

@@ -76,7 +76,7 @@ jobs:
             PWD=$(pwd)
             export LSAN_OPTIONS="suppressions=$PWD/suppressions.txt"
           fi
-          cargo nextest run -Z build-std --features ci --target "$TARGET"
+          cargo nextest run -Z build-std --features ci --profile ci --target "$TARGET"
 
       - name: Save simulation seeds artifact
         if: env.DUMP_SIMULATION_SEEDS

fuzz/fuzz_targets/client_initial.rs

@@ -17,7 +17,7 @@ fuzz_target!(|data: &[u8]| {
     };
 
     let mut client = default_client();
-    let ci = client.process(None, now()).dgram().expect("a datagram");
+    let ci = client.process_output(now()).dgram().expect("a datagram");
     let Some((header, d_cid, s_cid, payload)) = decode_initial_header(&ci, Role::Client) else {
         return;
     };
@@ -60,7 +60,7 @@ fuzz_target!(|data: &[u8]| {
     let fuzzed_ci = Datagram::new(ci.source(), ci.destination(), ci.tos(), ciphertext);
 
     let mut server = default_server();
-    let _response = server.process(Some(&fuzzed_ci), now());
+    let _response = server.process(Some(fuzzed_ci), now());
 });
 
 #[cfg(any(not(fuzzing), windows))]

fuzz/fuzz_targets/server_initial.rs

@@ -17,12 +17,9 @@ fuzz_target!(|data: &[u8]| {
     };
 
     let mut client = default_client();
-    let ci = client.process(None, now()).dgram().expect("a datagram");
+    let ci = client.process_output(now()).dgram().expect("a datagram");
     let mut server = default_server();
-    let si = server
-        .process(Some(&ci), now())
-        .dgram()
-        .expect("a datagram");
+    let si = server.process(Some(ci), now()).dgram().expect("a datagram");
 
     let Some((header, d_cid, s_cid, payload)) = decode_initial_header(&si, Role::Server) else {
         return;
@@ -64,7 +61,7 @@ fuzz_target!(|data: &[u8]| {
         (header_enc.len() - 1)..header_enc.len(),
     );
     let fuzzed_si = Datagram::new(si.source(), si.destination(), si.tos(), ciphertext);
-    let _response = client.process(Some(&fuzzed_si), now());
+    let _response = client.process(Some(fuzzed_si), now());
 });
 
 #[cfg(any(not(fuzzing), windows))]

neqo-bin/benches/main.rs

@@ -4,7 +4,7 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use std::{path::PathBuf, str::FromStr};
+use std::{env, path::PathBuf, str::FromStr};
 
 use criterion::{criterion_group, criterion_main, BatchSize, Criterion, Throughput};
 use neqo_bin::{client, server};
@@ -20,18 +20,18 @@ fn transfer(c: &mut Criterion) {
     neqo_crypto::init_db(PathBuf::from_str("../test-fixture/db").unwrap()).unwrap();
 
     let done_sender = spawn_server();
-
+    let mtu = env::var("MTU").map_or_else(|_| String::new(), |mtu| format!("/mtu-{mtu}"));
     for Benchmark { name, requests } in [
         Benchmark {
-            name: "1-conn/1-100mb-resp (aka. Download)".to_string(),
+            name: format!("1-conn/1-100mb-resp{mtu} (aka. Download)"),
             requests: vec![100 * 1024 * 1024],
         },
         Benchmark {
-            name: "1-conn/10_000-parallel-1b-resp (aka. RPS)".to_string(),
+            name: format!("1-conn/10_000-parallel-1b-resp{mtu} (aka. RPS)"),
             requests: vec![1; 10_000],
         },
         Benchmark {
-            name: "1-conn/1-1b-resp (aka. HPS)".to_string(),
+            name: format!("1-conn/1-1b-resp{mtu} (aka. HPS)"),
             requests: vec![1; 1],
         },
     ] {

neqo-bin/src/client/http09.rs

@@ -26,7 +26,7 @@ use neqo_transport::{
 use url::Url;
 
 use super::{get_output_file, qlog_new, Args, CloseState, Res};
-use crate::STREAM_IO_BUFFER_SIZE;
+use crate::{client::local_addr_for, STREAM_IO_BUFFER_SIZE};
 
 pub struct Handler<'a> {
     streams: HashMap<StreamId, Option<BufWriter<File>>>,
@@ -37,6 +37,7 @@ pub struct Handler<'a> {
     token: Option<ResumptionToken>,
     needs_key_update: bool,
     read_buffer: Vec<u8>,
+    migration: Option<&'a (u16, SocketAddr)>,
 }
 
 impl Handler<'_> {
@@ -85,6 +86,26 @@ impl super::Handler for Handler<'_> {
                         self.download_urls(client);
                     }
                 }
+                ConnectionEvent::StateChange(State::Confirmed) => {
+                    if let Some((local_port, migration_addr)) = self.migration.take() {
+                        let local_addr = local_addr_for(migration_addr, *local_port);
+                        qdebug!("Migrating path to {:?} -> {:?}", local_addr, migration_addr);
+                        client
+                            .migrate(
+                                Some(local_addr),
+                                Some(*migration_addr),
+                                false,
+                                Instant::now(),
+                            )
+                            .map(|()| {
+                                qinfo!(
+                                    "Connection migrated to {:?} -> {:?}",
+                                    local_addr,
+                                    migration_addr
+                                );
+                            })?;
+                    }
+                }
                 ConnectionEvent::StateChange(
                     State::WaitInitial | State::Handshaking | State::Connected,
                 ) => {
@@ -181,10 +202,11 @@ impl super::Client for Connection {
         self.process_output(now)
     }
 
-    fn process_multiple_input<'a, I>(&mut self, dgrams: I, now: Instant)
-    where
-        I: IntoIterator<Item = &'a Datagram>,
-    {
+    fn process_multiple_input<'a>(
+        &mut self,
+        dgrams: impl IntoIterator<Item = Datagram<&'a [u8]>>,
+        now: Instant,
+    ) {
         self.process_multiple_input(dgrams, now);
     }
 
@@ -211,7 +233,11 @@ impl super::Client for Connection {
 }
 
 impl<'b> Handler<'b> {
-    pub fn new(url_queue: VecDeque<Url>, args: &'b Args) -> Self {
+    pub fn new(
+        url_queue: VecDeque<Url>,
+        args: &'b Args,
+        migration: Option<&'b (u16, SocketAddr)>,
+    ) -> Self {
         Self {
             streams: HashMap::new(),
             url_queue,
@@ -221,6 +247,7 @@ impl<'b> Handler<'b> {
             token: None,
             needs_key_update: args.key_update,
             read_buffer: vec![0; STREAM_IO_BUFFER_SIZE],
+            migration,
         }
     }
 

neqo-bin/src/client/http3.rs

@@ -132,10 +132,11 @@ impl super::Client for Http3Client {
         self.process_output(now)
     }
 
-    fn process_multiple_input<'a, I>(&mut self, dgrams: I, now: Instant)
-    where
-        I: IntoIterator<Item = &'a Datagram>,
-    {
+    fn process_multiple_input<'a>(
+        &mut self,
+        dgrams: impl IntoIterator<Item = Datagram<&'a [u8]>>,
+        now: Instant,
+    ) {
         self.process_multiple_input(dgrams, now);
     }