feat(vmm): run virtio-net on vhost-net with configurable queue pairs

Author: kvinwang

dstack/vmm/rpc/proto/vmm_rpc.proto

@@ -52,6 +52,10 @@ message NetworkInterfaceStatus {
   optional string bridge_name = 4;
   // QEMU netdev id, e.g. "net0".
   optional string netdev_id = 5;
+  // Effective vhost-net data plane state for this interface.
+  bool vhost = 6;
+  // Effective virtio-net queue pairs.
+  uint32 queues = 7;
 }
 
 // Structured log or lifecycle event emitted by the guest or runtime.
@@ -135,6 +139,12 @@ message NetworkingConfig {
   // Effective macvtap forwarding mode in responses. Deployment requests must
   // leave this empty because the mode is controlled by node configuration.
   string macvtap_mode = 4;
+  // Move packet processing into the host kernel vhost-net data plane. Unset
+  // inherits the node default. User mode has no vhost backend and ignores it.
+  optional bool vhost = 5;
+  // virtio-net queue pairs. Unset inherits the node default. Bounded by the
+  // node's cvm.max_net_queues.
+  optional uint32 queues = 6;
 }
 
 // Requested GPU layout for a CVM.
@@ -301,6 +311,8 @@ message NetworkingCapabilities {
   reserved "forward_service_enabled";
   // Default bridge configured in vmm.toml [cvm.networking].bridge.
   string default_bridge = 4;
+  // Largest virtio-net queue pair count a deployment request may ask for.
+  uint32 max_queues = 5;
 }
 
 // Aggregated metadata exposed through GetMeta.

dstack/vmm/src/app.rs

@@ -42,7 +42,8 @@ use tracing::{debug, error, info, warn};
 
 pub use image::{Image, ImageInfo};
 pub(crate) use network::{
-    resolve_networking, resolved_networks, validate_resolved_network, validate_resolved_networks,
+    clamp_queues_without_netd, needs_netd_interface, netd_available, resolve_networking,
+    resolved_networks, validate_resolved_network, validate_resolved_networks,
 };
 pub use qemu::VmConfig;
 pub use workdir::VmWorkDir;
@@ -355,7 +356,7 @@ impl App {
         let vm_id = manifest.id.clone();
         let mut runtime_networks = vm_work_dir.runtime_networks();
         if runtime_networks.is_empty() && cids_assigned.contains_key(&vm_id) {
-            runtime_networks = resolved_networks(&manifest, &self.config.cvm);
+            runtime_networks = self.runtime_networks(&manifest);
             if let Err(err) = vm_work_dir.set_runtime_networks(&runtime_networks) {
                 warn!(id = %vm_id, "failed to persist inferred runtime networks: {err}");
             }
@@ -454,7 +455,7 @@ impl App {
                 append_boot_separator(&path);
             }
 
-            let mut runtime_networks = resolved_networks(&vm_config.manifest, &self.config.cvm);
+            let mut runtime_networks = self.runtime_networks(&vm_config.manifest);
             let devices = self.try_allocate_gpus(&vm_config.manifest)?;
             let gpu_host_config = self.config.cvm.gpu.clone();
             let devices_to_sanitize = devices.clone();
@@ -547,19 +548,16 @@ impl App {
         vm: &VmConfig,
         networks: &mut [Networking],
     ) -> Result<()> {
-        if self.config.cvm.network_filter.mode == NetworkFilterMode::None
-            && !networks
-                .iter()
-                .any(|network| network.mode == NetworkingMode::Macvtap)
+        if !networks
+            .iter()
+            .any(|network| needs_netd_interface(network, &self.config.cvm))
         {
             return Ok(());
         }
         let qemu_uid = Uid::effective().as_raw();
         let mut prepared = Vec::new();
         for (nic_index, network) in networks.iter_mut().enumerate() {
-            if network.mode == NetworkingMode::Bridge
-                && self.config.cvm.network_filter.mode == NetworkFilterMode::None
-            {
+            if !needs_netd_interface(network, &self.config.cvm) {
                 continue;
             }
             let identity = InterfaceIdentity {
@@ -572,21 +570,35 @@ impl App {
                 &network.mac_prefix_bytes(),
                 nic_index,
             );
+            let queues = network.queue_pairs();
+            let filtered = self.filters_bridge_traffic(network);
             let request = match network.mode {
                 NetworkingMode::Bridge => NetdRequest::PrepareBridge(PrepareBridgeRequest {
                     identity: identity.clone(),
                     bridge: network.bridge.clone(),
                     mac,
                     qemu_uid,
-                    filter: self.config.cvm.network_filter.filter.clone(),
-                    parameters: self.config.cvm.network_filter.parameters.clone(),
+                    // An unfiltered TAP is only created for multiqueue, where
+                    // the node may not run libvirt at all.
+                    filter: if filtered {
+                        self.config.cvm.network_filter.filter.clone()
+                    } else {
+                        String::new()
+                    },
+                    parameters: if filtered {
+                        self.config.cvm.network_filter.parameters.clone()
+                    } else {
+                        BTreeMap::new()
+                    },
+                    queues,
                 }),
                 NetworkingMode::Macvtap => NetdRequest::PrepareMacvtap(PrepareMacvtapRequest {
                     identity: identity.clone(),
                     parent: network.parent.clone(),
                     mac,
                     qemu_uid,
                     mode: network.macvtap_mode.clone(),
+                    queues,
                 }),
                 NetworkingMode::User | NetworkingMode::Custom => continue,
             };
@@ -600,73 +612,127 @@ impl App {
                         &self.config.netd.socket,
                         &NetdRequest::Remove {
                             identity: identity.clone(),
+                            filtered,
                         },
                     )
                     .await
                     {
                         warn!(%cleanup_error, "failed to roll back in-flight filtered network");
                     }
-                    for identity in prepared.into_iter().rev() {
-                        if let Err(cleanup_error) = netd::request(
-                            &self.config.netd.socket,
-                            &NetdRequest::Remove { identity },
-                        )
-                        .await
-                        {
-                            warn!(%cleanup_error, "failed to roll back prepared filtered network");
-                        }
-                    }
-                    return Err(error).context("failed to prepare libvirt-filtered networking");
+                    self.roll_back_prepared_networks(prepared).await;
+                    return Err(error).context("failed to prepare netd-managed networking");
                 }
             };
-            if network.mode == NetworkingMode::Macvtap {
-                network.device = response
-                    .device
-                    .context("netd response omitted macvtap device")?;
+            prepared.push((identity.clone(), filtered));
+            // Everything below runs after netd already built a host interface,
+            // so a failure has to unwind the same way a failed Prepare does.
+            let accepted = (|| {
+                if network.mode == NetworkingMode::Macvtap {
+                    network.device = response
+                        .device
+                        .clone()
+                        .context("netd response omitted macvtap device")?;
+                }
+                // QEMU refuses a TAP whose IFF_MULTI_QUEUE state disagrees with
+                // its own `queues=`, and reports it from inside the per-VM
+                // launcher. netd echoes what it built, so a netd too old to
+                // understand the request fails here, where the reason is
+                // legible.
+                if queues > 1 && response.queues != Some(queues) {
+                    bail!(
+                        "netd prepared interface {nic_index} with {} queue pairs instead of \
+                         {queues}; its version may predate multiqueue support",
+                        response.queues.map_or_else(
+                            || "an unreported number of".to_string(),
+                            |q| q.to_string()
+                        )
+                    );
+                }
+                Ok(())
+            })();
+            if let Err(error) = accepted {
+                self.roll_back_prepared_networks(prepared).await;
+                return Err(error);
             }
-            prepared.push(identity);
         }
         Ok(())
     }
 
+    /// Launch-time view of a VM's NICs: node defaults merged in, the
+    /// vCPU-scaled queue count made concrete, and multiqueue dropped when this
+    /// node has no netd to build the interface.
+    pub(crate) fn runtime_networks(&self, manifest: &Manifest) -> Vec<Networking> {
+        let requested = if manifest.networks.is_empty() {
+            vec![self.config.cvm.networking.clone()]
+        } else {
+            manifest.networks.clone()
+        };
+        let mut resolved = resolved_networks(manifest, &self.config.cvm);
+        clamp_queues_without_netd(
+            &requested,
+            &mut resolved,
+            &self.config.cvm,
+            netd_available(&self.config.netd.socket),
+        );
+        resolved
+    }
+
+    /// Removes interfaces netd already built for a launch that then failed.
+    async fn roll_back_prepared_networks(&self, prepared: Vec<(InterfaceIdentity, bool)>) {
+        for (identity, filtered) in prepared.into_iter().rev() {
+            if let Err(cleanup_error) = netd::request(
+                &self.config.netd.socket,
+                &NetdRequest::Remove { identity, filtered },
+            )
+            .await
+            {
+                warn!(%cleanup_error, "failed to roll back prepared network interface");
+            }
+        }
+    }
+
+    /// Whether this NIC's host interface carries a libvirt nwfilter binding.
+    /// Macvtap never does, and a bridge NIC only does when the node filters.
+    fn filters_bridge_traffic(&self, network: &Networking) -> bool {
+        network.mode == NetworkingMode::Bridge
+            && self.config.cvm.network_filter.mode == NetworkFilterMode::Libvirt
+    }
+
     pub(crate) async fn remove_filtered_networks(
         &self,
         vm_id: &str,
         networks: &[Networking],
     ) -> Result<()> {
-        if self.config.cvm.network_filter.mode == NetworkFilterMode::None
-            && !networks
-                .iter()
-                .any(|network| network.mode == NetworkingMode::Macvtap)
+        if !networks
+            .iter()
+            .any(|network| needs_netd_interface(network, &self.config.cvm))
         {
             return Ok(());
         }
         let mut first_error = None;
         for (nic_index, network) in networks.iter().enumerate().rev() {
-            if network.mode == NetworkingMode::Bridge
-                && self.config.cvm.network_filter.mode == NetworkFilterMode::None
-            {
-                continue;
-            }
-            if !matches!(
-                network.mode,
-                NetworkingMode::Bridge | NetworkingMode::Macvtap
-            ) {
+            if !needs_netd_interface(network, &self.config.cvm) {
                 continue;
             }
             let identity = InterfaceIdentity {
                 instance_id: self.config.cvm.instance_id.clone(),
                 vm_id: vm_id.to_string(),
                 nic_index,
             };
-            if let Err(error) =
-                netd::request(&self.config.netd.socket, &NetdRequest::Remove { identity }).await
+            if let Err(error) = netd::request(
+                &self.config.netd.socket,
+                &NetdRequest::Remove {
+                    identity,
+                    filtered: self.filters_bridge_traffic(network),
+                },
+            )
+            .await
             {
                 first_error.get_or_insert(error);
             }
         }
         if let Some(error) = first_error {
-            return Err(error).context("failed to remove libvirt-filtered networking");
+            return Err(error).context("failed to remove netd-managed networking");
         }
         Ok(())
     }
@@ -1059,7 +1125,7 @@ impl App {
         let already_running = cids_assigned.contains_key(&vm_id);
         let mut runtime_networks = vm_work_dir.runtime_networks();
         if runtime_networks.is_empty() && already_running {
-            runtime_networks = resolved_networks(&manifest, &self.config.cvm);
+            runtime_networks = self.runtime_networks(&manifest);
             if let Err(err) = vm_work_dir.set_runtime_networks(&runtime_networks) {
                 warn!(id = %vm_id, "failed to persist inferred runtime networks: {err}");
             }
@@ -2241,6 +2307,8 @@ mod tests {
             dhcp_start: String::new(),
             restrict: false,
             netdev: String::new(),
+            vhost: None,
+            queues: None,
         }];
 
         workdir.put_manifest(&manifest)?;
@@ -2506,6 +2574,8 @@ mod tests {
             dhcp_start: String::new(),
             restrict: false,
             netdev: String::new(),
+            vhost: None,
+            queues: None,
         }];
         let user_manifest = test_manifest(2048);
         let image = test_tdx_image(true);