route switched I/O through completion-owned UDP (#4814)

Summary:
Pull Request resolved: #4814

- add `RoutedUdpPacketIo` so quiche, CID routing, pacing, GSO/GRO, and io_uring waits share one endpoint owner
- route non-local CID aggregates directly between completion-owned UDP carriers while retaining a Unix datagram fallback for task-local leaves
- remove the threaded `CompletionUdpSocket` adapter and preserve route admission through actual submission
- use the roofline-validated eight-segment switched GSO frontier and wait for response FIN before benchmark teardown

Design document: https://mdoc.internalmeta.com/doc/e6918c35-b3d1-4118-ac39-6308b0cb55ba

Walkthrough document: https://mdoc.internalmeta.com/doc/ef7df54e-b047-4e9a-96b5-6055b2a411da

Reviewed By: shayne-fletcher

Differential Revision: D117242784

Author: mariusae

chrysalis-scale/src/benchmark.rs

@@ -84,6 +84,7 @@ const META_NETWORK_MAX_UDP_PAYLOAD: u16 = 1_450;
 const EXPERIMENT_POLL_INTERVAL: Duration = Duration::from_millis(250);
 const EXPERIMENT_LEASE_GRACE: Duration = Duration::from_secs(60);
 const WORKER_POLL_INTERVAL: Duration = Duration::from_millis(25);
+const STREAM_FIN_TIMEOUT: Duration = Duration::from_secs(5);
 const PERSISTENCE_SCHEMA_VERSION: u32 = 8;
 pub(crate) const DEFAULT_NODES_PER_TASK: usize = 100;
 pub(crate) const DEFAULT_IDENTITY_CONCURRENCY: usize = 8;
@@ -640,22 +641,18 @@ async fn create_local_node(
 ) -> Result<LocalNode> {
     let udp_address = socket.address();
     let identity = issue_scale_identity(rank).await?;
+    let quic_config = scale_quic_config()?;
     let transport = match unix_path {
         Some(path) => {
-            let primary: Arc<dyn DatagramSocket> = Arc::new(socket);
-            let alternatives: Vec<Arc<dyn DatagramSocket>> = vec![Arc::new(
+            let fallback: Arc<dyn DatagramSocket> = Arc::new(
                 UnixDatagramSocket::bind(path)
                     .with_context(|| format!("bind task-head Unix carrier {}", path.display()))?,
-            )];
-            let sockets = Arc::new(
-                DatagramSocketSet::new(primary, alternatives)
-                    .context("create task-head socket set")?,
             );
-            TransportConfig::new(sockets, identity)
+            TransportConfig::routed_udp(socket.into_std()?, Some(fallback), identity)?
         }
         None => TransportConfig::direct_udp(socket.into_std()?, identity)?,
     }
-    .with_quic_config(scale_quic_config()?);
+    .with_quic_config(quic_config);
     let mut config = NodeConfig::new(transport);
     let role = if is_root {
         NodeRole::Root
@@ -1753,7 +1750,16 @@ fn spawn_operation(
                 read_delivery_receipt(&mut recv, size).await?;
             }
         }
-        Ok(started.elapsed())
+        let elapsed = started.elapsed();
+        let mut trailing = [0];
+        anyhow::ensure!(
+            tokio::time::timeout(STREAM_FIN_TIMEOUT, recv.read(&mut trailing))
+                .await
+                .context("timed out waiting for benchmark response FIN")??
+                == 0,
+            "benchmark response contains trailing data"
+        );
+        Ok(elapsed)
     });
 }
 

chrysalis-scale/src/network_baseline.rs

@@ -17,7 +17,6 @@ use anyhow::Context;
 use anyhow::Result;
 use bytes::Bytes;
 use chrysalis::DatagramSocket;
-use chrysalis::DatagramSwitch;
 use chrysalis::Pid;
 use chrysalis::QuicConnectionStats;
 use chrysalis::QuicIoStats;
@@ -1127,24 +1126,23 @@ async fn receive_switched_quic(
     timeout: Duration,
     profile: NetworkBaselineProfile,
 ) -> Result<()> {
-    let physical = Arc::new(
-        UdpSocket::bind(local)
-            .await
-            .with_context(|| format!("bind switched QUIC receiver {local}"))?,
-    );
+    let quic_config = scale_quic_config()?;
     let router = Arc::new(Router::new());
     router.insert(
         peer,
         Route::permanent(UdpSocket::datagram_addr(peer_address)),
     );
-    let datagram_switch = DatagramSwitch::spawn(physical, router);
-    let binding = Arc::new(
-        datagram_switch
-            .bind_routed(identity.pid())
-            .context("bind switched QUIC receiver PID")?,
-    );
-    let transport = QuicTransport::spawn_with_config(binding, identity, scale_quic_config()?)
-        .context("start switched QUIC receiver")?;
+    let transport = QuicTransport::spawn_routed_udp_with_config(
+        UdpSocket::bind(local)
+            .await
+            .with_context(|| format!("bind switched QUIC receiver {local}"))?
+            .into_std()?,
+        None,
+        router,
+        identity,
+        quic_config,
+    )
+    .context("start switched QUIC receiver")?;
     control.write_all(&[PHASE_READY]).await?;
 
     let (warmup_source, connection_before) = receive_quic_stream(
@@ -1181,9 +1179,6 @@ async fn receive_switched_quic(
     signal_phase_complete(control).await?;
     transport.shutdown();
     transport.join().await;
-    drop(transport);
-    datagram_switch.shutdown();
-    datagram_switch.join().await;
     Ok(())
 }
 
@@ -1200,24 +1195,23 @@ async fn send_switched_quic(
     let mut ready = [0];
     control.read_exact(&mut ready).await?;
     anyhow::ensure!(ready == [PHASE_READY], "invalid switched QUIC start marker");
-    let physical = Arc::new(
-        UdpSocket::bind(local)
-            .await
-            .with_context(|| format!("bind switched QUIC sender {local}"))?,
-    );
+    let quic_config = scale_quic_config()?;
     let router = Arc::new(Router::new());
     router.insert(
         target,
         Route::permanent(UdpSocket::datagram_addr(peer_address)),
     );
-    let datagram_switch = DatagramSwitch::spawn(physical, router);
-    let binding = Arc::new(
-        datagram_switch
-            .bind_routed(identity.pid())
-            .context("bind switched QUIC sender PID")?,
-    );
-    let transport = QuicTransport::spawn_with_config(binding, identity, scale_quic_config()?)
-        .context("start switched QUIC sender")?;
+    let transport = QuicTransport::spawn_routed_udp_with_config(
+        UdpSocket::bind(local)
+            .await
+            .with_context(|| format!("bind switched QUIC sender {local}"))?
+            .into_std()?,
+        None,
+        router,
+        identity,
+        quic_config,
+    )
+    .context("start switched QUIC sender")?;
     let destination = UdpSocket::datagram_addr(peer_address);
 
     let connection_before = send_quic_stream(
@@ -1261,9 +1255,6 @@ async fn send_switched_quic(
     wait_for_phase_complete(control).await?;
     transport.shutdown();
     transport.join().await;
-    drop(transport);
-    datagram_switch.shutdown();
-    datagram_switch.join().await;
     Ok(())
 }