vmm: add upper limit for amount of parallel connections during migration

Check that the amount of parallel connections does not exceed 128 and
update documentation.

On-behalf-of: SAP sebastian.eydam@sap.com
Signed-off-by: Sebastian Eydam <sebastian.eydam@cyberus-technology.de>

docs/live_migration.md

@@ -207,3 +207,7 @@ migration process. Via the API or `ch-remote`, you may specify:
   Cancel will abort the migration and keep the VM running on the source.
   Ignore will proceed with the migration regardless of the downtime requirement.
   Defaults to `cancel`.
+- `connections <amount>`: \
+  The number of parallel TCP connections to use for migration.
+  Must be between `1` and `128`. Defaults to `1`.
+  Multiple connections are not supported with local UNIX-socket migration.

vmm/src/api/mod.rs

@@ -53,6 +53,7 @@ pub use self::http::{start_http_fd_thread, start_http_path_thread};
 use crate::Error as VmmError;
 use crate::config::RestoreConfig;
 use crate::device_tree::DeviceTree;
+use crate::migration_transport::MAX_MIGRATION_CONNECTIONS;
 use crate::vm::{Error as VmError, VmState};
 use crate::vm_config::{
     DeviceConfig, DiskConfig, FsConfig, GenericVhostUserConfig, NetConfig, PmemConfig,
@@ -324,7 +325,9 @@ pub struct VmSendMigrationData {
     #[serde(default)]
     pub timeout_strategy: TimeoutStrategy,
 
-    /// The number of parallel connections for migration.
+    /// The number of parallel TCP connections for migration.
+    ///
+    /// Must be between 1 and `MAX_MIGRATION_CONNECTIONS` inclusive.
     #[serde(default = "VmSendMigrationData::default_connections")]
     pub connections: NonZeroU32,
 }
@@ -459,6 +462,12 @@ impl VmSendMigrationData {
             }
         }
 
+        if self.connections.get() > MAX_MIGRATION_CONNECTIONS {
+            return Err(VmSendMigrationConfigError::ValidationError(format!(
+                "connections must not exceed {MAX_MIGRATION_CONNECTIONS}."
+            )));
+        }
+
         if self.local {
             if !self.destination_url.starts_with("unix:") {
                 return Err(VmSendMigrationConfigError::ValidationError(
@@ -1785,8 +1794,14 @@ mod unit_tests {
             .expect_err("zero timeout_s should be rejected");
 
         // Zero connections is rejected
-        let _data = VmSendMigrationData::parse("destination_url=unix:/tmp/sock,connections=0")
-            .expect_err("zero connections should be rejected");
+        let _data =
+            VmSendMigrationData::parse("destination_url=tcp:192.168.1.1:8080,connections=0")
+                .expect_err("zero connections should be rejected");
+
+        // Excessive numbers of parallel connections are rejected
+        let _data =
+            VmSendMigrationData::parse("destination_url=tcp:192.168.1.1:8080,connections=129")
+                .expect_err("too many connections should be rejected");
 
         // Unknown option is an error
         VmSendMigrationData::parse("destination_url=unix:/tmp/sock,unknown_field=foo").unwrap_err();

vmm/src/api/openapi/cloud-hypervisor.yaml

@@ -1410,6 +1410,11 @@ components:
           format: int64
           default: 1
           minimum: 1
+          maximum: 128
+          description: >
+            The number of parallel TCP connections to use for migration.
+            Must be between 1 and 128. Multiple connections are not supported
+            with local UNIX-socket migration.
 
     VmAddUserDevice:
       required:

vmm/src/migration_transport.rs

@@ -32,6 +32,10 @@ use vmm_sys_util::eventfd::EventFd;
 use crate::sync_utils::Gate;
 use crate::{GuestMemoryMmap, VmMigrationConfig};
 
+/// Hard upper bound for migration worker connections on both the sender and
+/// receiver side.
+pub(crate) const MAX_MIGRATION_CONNECTIONS: u32 = 128;
+
 /// Transport-agnostic listener used to receive connections.
 #[derive(Debug)]
 pub(crate) enum ReceiveListener {
@@ -288,12 +292,30 @@ impl ReceiveAdditionalConnections {
         guest_memory: &GuestMemoryAtomic<GuestMemoryMmap>,
     ) -> Result<(), MigratableError> {
         let mut threads: Vec<thread::JoinHandle<Result<(), MigratableError>>> = Vec::new();
-        while let Some(mut socket) = listener.abortable_accept(terminate_fd)? {
+        let mut first_err = loop {
+            let socket = match listener.abortable_accept(terminate_fd) {
+                Ok(socket) => socket,
+                Err(e) => break Err(e),
+            };
+            let Some(mut socket) = socket else {
+                break Ok(());
+            };
+
+            if threads.len() >= MAX_MIGRATION_CONNECTIONS as usize {
+                break Err(MigratableError::MigrateReceive(anyhow!(
+                    "Received more than {MAX_MIGRATION_CONNECTIONS} additional migration connections."
+                )));
+            }
+
             let guest_memory = guest_memory.clone();
-            let terminate_fd = terminate_fd
+            let terminate_fd = match terminate_fd
                 .try_clone()
                 .context("Error cloning terminate fd")
-                .map_err(MigratableError::MigrateReceive)?;
+                .map_err(MigratableError::MigrateReceive)
+            {
+                Ok(terminate_fd) => terminate_fd,
+                Err(e) => break Err(e),
+            };
 
             match thread::Builder::new()
                 .name(format!("migrate-receive-memory-{}", threads.len()).to_owned())
@@ -303,15 +325,21 @@ impl ReceiveAdditionalConnections {
                 Ok(t) => threads.push(t),
                 Err(e) => {
                     error!("Error spawning receive-memory thread: {e}");
-                    break;
+                    break Err(MigratableError::MigrateReceive(
+                        anyhow!(e).context("Error spawning receive-memory thread"),
+                    ));
                 }
             }
+        };
+
+        if first_err.is_err() {
+            warn!("Signaling termination due to an error while accepting connections.");
+            let _ = terminate_fd.write(1);
         }
 
         info!("Stopped accepting additional connections. Cleaning up threads.");
 
         // We only return the first error we encounter here.
-        let mut first_err = Ok(());
         for thread in threads {
             let err = match thread.join() {
                 Ok(Ok(())) => None,