From 9be6c198dba47fb959e6ab9f267beba716f3d2aa Mon Sep 17 00:00:00 2001 From: Jinrong Liang Date: Mon, 14 Apr 2025 21:21:23 +0800 Subject: [PATCH 1/7] vm-migration: Add support for downtime limits Add handling of migration timeout failures to provide more flexible live migration options. Implement downtime limiting logic to minimize service disruptions. Support for setting downtime thresholds and migration timeouts. Signed-off-by: Jinrong Liang --- src/bin/ch-remote.rs | 40 +++++++- vmm/src/api/mod.rs | 6 ++ vmm/src/lib.rs | 229 ++++++++++++++++++++++++++++++++++++------- 3 files changed, 240 insertions(+), 35 deletions(-) diff --git a/src/bin/ch-remote.rs b/src/bin/ch-remote.rs index ec363342de..44bb9caf8c 100644 --- a/src/bin/ch-remote.rs +++ b/src/bin/ch-remote.rs @@ -478,6 +478,16 @@ fn rest_api_do_command(matches: &ArgMatches, socket: &mut UnixStream) -> ApiResu .subcommand_matches("send-migration") .unwrap() .get_flag("send_migration_local"), + *matches + .subcommand_matches("send-migration") + .unwrap() + .get_one::("downtime") + .unwrap_or(&500), + *matches + .subcommand_matches("send-migration") + .unwrap() + .get_one::("migration_timeout") + .unwrap_or(&3600), ); simple_api_command(socket, "PUT", "send-migration", Some(&send_migration_data)) .map_err(Error::HttpApiClient) @@ -692,6 +702,16 @@ fn dbus_api_do_command(matches: &ArgMatches, proxy: &DBusApi1ProxyBlocking<'_>) .subcommand_matches("send-migration") .unwrap() .get_flag("send_migration_local"), + *matches + .subcommand_matches("send-migration") + .unwrap() + .get_one::("downtime") + .unwrap_or(&500), + *matches + .subcommand_matches("send-migration") + .unwrap() + .get_one::("migration_timeout") + .unwrap_or(&3600), ); proxy.api_vm_send_migration(&send_migration_data) } @@ -885,10 +905,12 @@ fn receive_migration_data(url: &str) -> String { serde_json::to_string(&receive_migration_data).unwrap() } -fn send_migration_data(url: &str, local: bool) -> String { +fn send_migration_data(url: &str, local: bool, downtime: u64, migration_timeout: u64) -> String { let send_migration_data = vmm::api::VmSendMigrationData { destination_url: url.to_owned(), local, + downtime, + migration_timeout, }; serde_json::to_string(&send_migration_data).unwrap() @@ -1050,6 +1072,22 @@ fn get_cli_commands_sorted() -> Box<[Command]> { Command::new("resume").about("Resume the VM"), Command::new("send-migration") .about("Initiate a VM migration") + .arg( + Arg::new("downtime") + .long("downtime") + .help("Set the expected maximum downtime in milliseconds") + .num_args(1) + .value_parser(clap::value_parser!(u64).range(1..)) + .default_value("500"), + ) + .arg( + Arg::new("migration_timeout") + .long("migration-timeout") + .help("Set the maximum allowed migration time in seconds") + .num_args(1) + .value_parser(clap::value_parser!(u64)) + .default_value("3600"), + ) .arg( Arg::new("send_migration_config") .index(1) diff --git a/vmm/src/api/mod.rs b/vmm/src/api/mod.rs index d83f46c75d..9c46172c17 100644 --- a/vmm/src/api/mod.rs +++ b/vmm/src/api/mod.rs @@ -260,6 +260,12 @@ pub struct VmSendMigrationData { /// Send memory across socket without copying #[serde(default)] pub local: bool, + /// Downtime + #[serde(default)] + pub downtime: u64, + /// Timeout for migration + #[serde(default)] + pub migration_timeout: u64, } pub enum ApiResponsePayload { diff --git a/vmm/src/lib.rs b/vmm/src/lib.rs index 03c8ee773e..fc17580653 100644 --- a/vmm/src/lib.rs +++ b/vmm/src/lib.rs @@ -20,13 +20,14 @@ use std::rc::Rc; use std::sync::mpsc::{Receiver, RecvError, SendError, Sender}; use std::sync::{Arc, Mutex}; #[cfg(not(target_arch = "riscv64"))] -use std::time::Instant; +use std::time::{Duration, Instant}; use std::{io, result, thread}; use anyhow::anyhow; #[cfg(feature = "dbus_api")] use api::dbus::{DBusApiOptions, DBusApiShutdownChannels}; use api::http::HttpApiHandle; +use arch::PAGE_SIZE; use console_devices::{pre_create_console_devices, ConsoleInfo}; use landlock::LandlockError; use libc::{tcsetattr, termios, EFD_NONBLOCK, SIGINT, SIGTERM, TCSANOW}; @@ -640,6 +641,45 @@ impl VmmVersionInfo { } } +#[derive(Debug, Clone)] +struct MigrationState { + current_dirty_pages: u64, + downtime: Duration, + downtime_start: Instant, + iteration: u64, + iteration_cost_time: Duration, + iteration_start_time: Instant, + mb_per_sec: f64, + pages_per_second: u64, + pending_size: u64, + start_time: Instant, + threshold_size: u64, + total_time: Duration, + total_transferred_bytes: u64, + total_transferred_dirty_pages: u64, +} + +impl MigrationState { + pub fn new() -> Self { + Self { + current_dirty_pages: 0, + downtime: Duration::default(), + downtime_start: Instant::now(), + iteration: 0, + iteration_cost_time: Duration::default(), + iteration_start_time: Instant::now(), + mb_per_sec: 0.0, + pages_per_second: 0, + pending_size: 0, + start_time: Instant::now(), + threshold_size: 0, + total_time: Duration::default(), + total_transferred_bytes: 0, + total_transferred_dirty_pages: 0, + } + } +} + pub struct VmmThreadHandle { pub thread_handle: thread::JoinHandle>, #[cfg(feature = "dbus_api")] @@ -1076,10 +1116,8 @@ impl Vmm { fn vm_maybe_send_dirty_pages( vm: &mut Vm, socket: &mut SocketStream, + table: MemoryRangeTable, ) -> result::Result { - // Send (dirty) memory table - let table = vm.dirty_log()?; - // But if there are no regions go straight to pause if table.regions().is_empty() { return Ok(false); @@ -1097,6 +1135,150 @@ impl Vmm { Ok(true) } + fn memory_copy_iterations( + vm: &mut Vm, + socket: &mut SocketStream, + s: &mut MigrationState, + migration_timeout: Duration, + migrate_downtime_limit: Duration, + ) -> result::Result { + let mut bandwidth = 0.0; + let mut iteration_table; + + loop { + // Update the start time of the iteration + s.iteration_start_time = Instant::now(); + + // Increment iteration counter + s.iteration += 1; + + // Check if migration has timed out + // migration_timeout > 0 means enabling the timeout check, 0 means disabling the timeout check + if !migration_timeout.is_zero() && s.start_time.elapsed() > migration_timeout { + warn!("Migration timed out after {:?}", migration_timeout); + Request::abandon().write_to(socket)?; + Response::read_from(socket)?.ok_or_abandon( + socket, + MigratableError::MigrateSend(anyhow!("Migration timed out")), + )?; + } + + // Get the dirty page table + iteration_table = vm.dirty_log()?; + + // Update the pending size (amount of data to transfer) + s.pending_size = iteration_table + .regions() + .iter() + .map(|range| range.length) + .sum(); + + // Update thresholds + if bandwidth > 0.0 { + s.threshold_size = bandwidth as u64 * migrate_downtime_limit.as_millis() as u64; + } + + // Enter the final stage of migration when the suspension conditions are met + if s.iteration > 1 && s.pending_size < s.threshold_size { + break; + } + + // Update the number of dirty pages + s.total_transferred_bytes += s.pending_size; + s.current_dirty_pages = s.pending_size.div_ceil(PAGE_SIZE as u64); + s.total_transferred_dirty_pages += s.current_dirty_pages; + + // Send the current dirty pages + let transfer_start = Instant::now(); + Self::vm_maybe_send_dirty_pages(vm, socket, iteration_table.clone())?; + let transfer_time = transfer_start.elapsed().as_millis() as f64; + + // Update bandwidth + if transfer_time > 0.0 && s.pending_size > 0 { + bandwidth = s.pending_size as f64 / transfer_time; + // Convert bandwidth to MB/s + s.mb_per_sec = (bandwidth * 1000.0) / (1024.0 * 1024.0); + } + + // Update iteration cost time + s.iteration_cost_time = s.iteration_start_time.elapsed(); + if s.iteration_cost_time.as_millis() > 0 { + s.pages_per_second = + s.current_dirty_pages * 1000 / s.iteration_cost_time.as_millis() as u64; + } + } + + Ok(iteration_table) + } + + fn do_memory_migration( + vm: &mut Vm, + socket: &mut SocketStream, + s: &mut MigrationState, + send_data_migration: &VmSendMigrationData, + ) -> result::Result<(), MigratableError> { + // Start logging dirty pages + vm.start_dirty_log()?; + + // Send memory table + let table = vm.memory_range_table()?; + Request::memory(table.length()).write_to(socket).unwrap(); + table.write_to(socket)?; + // And then the memory itself + vm.send_memory_regions(&table, socket)?; + Response::read_from(socket)?.ok_or_abandon( + socket, + MigratableError::MigrateSend(anyhow!("Error during dirty memory migration")), + )?; + + // Define the maximum allowed downtime 2000 seconds(2000000 milliseconds) + const MAX_MIGRATE_DOWNTIME: u64 = 2000000; + + // Verify that downtime must be between 1 and MAX_MIGRATE_DOWNTIME + if send_data_migration.downtime == 0 || send_data_migration.downtime > MAX_MIGRATE_DOWNTIME + { + return Err(MigratableError::MigrateSend(anyhow!( + "downtime_limit must be an integer in the range of 1 to {} ms", + MAX_MIGRATE_DOWNTIME + ))); + } + + let migration_timeout = Duration::from_secs(send_data_migration.migration_timeout); + let migrate_downtime_limit = Duration::from_millis(send_data_migration.downtime); + + // Verify that downtime must be less than the migration timeout + if !migration_timeout.is_zero() && migrate_downtime_limit >= migration_timeout { + return Err(MigratableError::MigrateSend(anyhow!( + "downtime_limit {}ms must be less than migration_timeout {}ms", + send_data_migration.downtime, + send_data_migration.migration_timeout * 1000 + ))); + } + + let iteration_table = + Self::memory_copy_iterations(vm, socket, s, migration_timeout, migrate_downtime_limit)?; + + info!("Entering downtime phase"); + s.downtime_start = Instant::now(); + vm.pause()?; + + // Send last batch of dirty pages + let mut final_table = vm.dirty_log()?; + final_table.extend(iteration_table.clone()); + Self::vm_maybe_send_dirty_pages(vm, socket, final_table.clone())?; + + // Update statistics + s.pending_size = final_table.regions().iter().map(|range| range.length).sum(); + s.total_transferred_bytes += s.pending_size; + s.current_dirty_pages = s.pending_size.div_ceil(PAGE_SIZE as u64); + s.total_transferred_dirty_pages += s.current_dirty_pages; + + // Stop logging dirty pages + vm.stop_dirty_log()?; + + Ok(()) + } + fn send_migration( vm: &mut Vm, #[cfg(all(feature = "kvm", target_arch = "x86_64"))] hypervisor: Arc< @@ -1104,6 +1286,8 @@ impl Vmm { >, send_data_migration: VmSendMigrationData, ) -> result::Result<(), MigratableError> { + let mut s = MigrationState::new(); + // Set up the socket connection let mut socket = Self::send_migration_socket(&send_data_migration.destination_url)?; @@ -1181,36 +1365,7 @@ impl Vmm { // Now pause VM vm.pause()?; } else { - // Start logging dirty pages - vm.start_dirty_log()?; - - // Send memory table - let table = vm.memory_range_table()?; - Request::memory(table.length()) - .write_to(&mut socket) - .unwrap(); - table.write_to(&mut socket)?; - // And then the memory itself - vm.send_memory_regions(&table, &mut socket)?; - Response::read_from(&mut socket)?.ok_or_abandon( - &mut socket, - MigratableError::MigrateSend(anyhow!("Error during dirty memory migration")), - )?; - - // Try at most 5 passes of dirty memory sending - const MAX_DIRTY_MIGRATIONS: usize = 5; - for i in 0..MAX_DIRTY_MIGRATIONS { - info!("Dirty memory migration {} of {}", i, MAX_DIRTY_MIGRATIONS); - if !Self::vm_maybe_send_dirty_pages(vm, &mut socket)? { - break; - } - } - - // Now pause VM - vm.pause()?; - - // Send last batch of dirty pages - Self::vm_maybe_send_dirty_pages(vm, &mut socket)?; + Self::do_memory_migration(vm, &mut socket, &mut s, &send_data_migration)?; } // We release the locks early to enable locking them on the destination host. @@ -1237,11 +1392,17 @@ impl Vmm { MigratableError::MigrateSend(anyhow!("Error completing migration")), )?; + // Record downtime + s.downtime = s.downtime_start.elapsed(); + // Stop logging dirty pages if !send_data_migration.local { vm.stop_dirty_log()?; } + // Record total migration time + s.total_time = s.start_time.elapsed(); + info!("Migration complete"); // Let every Migratable object know about the migration being complete From c51cd0e4573024e7139198d3a94dac56c908bc0a Mon Sep 17 00:00:00 2001 From: Jinrong Liang Date: Wed, 4 Jun 2025 17:03:49 +0800 Subject: [PATCH 2/7] docs: Add migration parameters to live migration document Updated live migration documentation to include migration timeout controls and downtime limits. Signed-off-by: Jinrong Liang --- docs/live_migration.md | 29 ++++++++++++++++++++++- vmm/src/api/openapi/cloud-hypervisor.yaml | 10 ++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/docs/live_migration.md b/docs/live_migration.md index 94c9afc236..5c77d2625f 100644 --- a/docs/live_migration.md +++ b/docs/live_migration.md @@ -171,7 +171,13 @@ After a few seconds the VM should be up and you can interact with it. Initiate the Migration over TCP: ```console -src $ ch-remote --api-socket=/tmp/api send-migration tcp:{dst}:{port} +src $ ch-remote --api-socket=/tmp/api send-migration tcp:{dst}:{port} +``` + +With migration parameters: + +```console +src $ ch-remote --api-socket=/tmp/api send-migration tcp:{dst}:{port} --migration-timeout 60 --downtime 5000 ``` > Replace {dst}:{port} with the actual IP address and port of your destination host. @@ -180,3 +186,24 @@ After completing the above commands, the source VM will be migrated to the destination host and continue running there. The source VM instance will terminate normally. All ongoing processes and connections within the VM should remain intact after the migration. + +#### Migration Parameters + +Cloud Hypervisor supports additional parameters to control the +migration process: + +- `migration-timeout ` +Sets the maximum time (in seconds) allowed for the migration process. +If the migration takes longer than this timeout, it will be aborted. A +value of 0 means no timeout limit. +- `downtime ` +Sets the maximum acceptable downtime (in milliseconds) during the +migration. This parameter helps control the trade-off between migration +time and VM downtime. + +> The downtime limit is related to the cost of serialization +(deserialization) of vCPU and device state. Therefore, the expected +downtime is always shorter than the actual downtime. + +These parameters can be used with the `send-migration` command to +fine-tune the migration behavior according to your requirements. \ No newline at end of file diff --git a/vmm/src/api/openapi/cloud-hypervisor.yaml b/vmm/src/api/openapi/cloud-hypervisor.yaml index 80a4fa2572..7e667a7410 100644 --- a/vmm/src/api/openapi/cloud-hypervisor.yaml +++ b/vmm/src/api/openapi/cloud-hypervisor.yaml @@ -1272,6 +1272,16 @@ components: type: string local: type: boolean + downtime: + type: integer + format: int64 + description: Maximum downtime in milliseconds during migration + default: 500 + migration_timeout: + type: integer + format: int64 + description: Total timeout for migration in milliseconds (0 = no limit) + default: 0 VmAddUserDevice: required: From e8a2a6b76c319b20bfda25e11ac364fd57259b49 Mon Sep 17 00:00:00 2001 From: Jinrong Liang Date: Wed, 4 Jun 2025 18:51:27 +0800 Subject: [PATCH 3/7] tests: Add downtime and migration timeout tests Signed-off-by: Jinrong Liang --- tests/integration.rs | 97 +++++++++++++++++++++++++++++++++++++++----- 1 file changed, 87 insertions(+), 10 deletions(-) diff --git a/tests/integration.rs b/tests/integration.rs index 0536864fd2..995ba87558 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -9208,6 +9208,8 @@ mod live_migration { src_api_socket: &str, dest_api_socket: &str, local: bool, + downtime: u64, + timeout: u64, ) -> bool { // Start to receive migration from the destination VM let mut receive_migration = Command::new(clh_command("ch-remote")) @@ -9228,6 +9230,10 @@ mod live_migration { format!("--api-socket={}", &src_api_socket), "send-migration".to_string(), format! {"unix:{migration_socket}"}, + "--downtime".to_string(), + format!("{downtime}"), + "--migration-timeout".to_string(), + format!("{timeout}"), ] .to_vec(); @@ -9244,7 +9250,7 @@ mod live_migration { // The 'send-migration' command should be executed successfully within the given timeout let send_success = if let Some(status) = send_migration - .wait_timeout(std::time::Duration::from_secs(30)) + .wait_timeout(std::time::Duration::from_secs(500)) .unwrap() { status.success() @@ -9266,7 +9272,7 @@ mod live_migration { // The 'receive-migration' command should be executed successfully within the given timeout let receive_success = if let Some(status) = receive_migration - .wait_timeout(std::time::Duration::from_secs(30)) + .wait_timeout(std::time::Duration::from_secs(500)) .unwrap() { status.success() @@ -9447,8 +9453,18 @@ mod live_migration { .unwrap(), ); + let downtime = 100000; + let migration_timeout = 1000; + assert!( - start_live_migration(&migration_socket, &src_api_socket, &dest_api_socket, local), + start_live_migration( + &migration_socket, + &src_api_socket, + &dest_api_socket, + local, + downtime, + migration_timeout + ), "Unsuccessful command: 'send-migration' or 'receive-migration'." ); }); @@ -9621,8 +9637,18 @@ mod live_migration { .unwrap(), ); + let downtime = 100000; + let migration_timeout = 1000; + assert!( - start_live_migration(&migration_socket, &src_api_socket, &dest_api_socket, local), + start_live_migration( + &migration_socket, + &src_api_socket, + &dest_api_socket, + local, + downtime, + migration_timeout + ), "Unsuccessful command: 'send-migration' or 'receive-migration'." ); }); @@ -9839,8 +9865,18 @@ mod live_migration { .unwrap(), ); + let downtime = 100000; + let migration_timeout = 1000; + assert!( - start_live_migration(&migration_socket, &src_api_socket, &dest_api_socket, local), + start_live_migration( + &migration_socket, + &src_api_socket, + &dest_api_socket, + local, + downtime, + migration_timeout + ), "Unsuccessful command: 'send-migration' or 'receive-migration'." ); }); @@ -10055,8 +10091,18 @@ mod live_migration { .unwrap(), ); + let downtime = 100000; + let migration_timeout = 1000; + assert!( - start_live_migration(&migration_socket, &src_api_socket, &dest_api_socket, local), + start_live_migration( + &migration_socket, + &src_api_socket, + &dest_api_socket, + local, + downtime, + migration_timeout + ), "Unsuccessful command: 'send-migration' or 'receive-migration'." ); }); @@ -10165,8 +10211,18 @@ mod live_migration { .unwrap(), ); + let downtime = 100000; + let migration_timeout = 1000; + assert!( - start_live_migration(&migration_socket, &src_api_socket, &dest_api_socket, local), + start_live_migration( + &migration_socket, + &src_api_socket, + &dest_api_socket, + local, + downtime, + migration_timeout + ), "Unsuccessful command: 'send-migration' or 'receive-migration'." ); }); @@ -10312,8 +10368,18 @@ mod live_migration { .unwrap(), ); + let downtime = 100000; + let migration_timeout = 1000; + assert!( - start_live_migration(&migration_socket, &src_api_socket, &dest_api_socket, true), + start_live_migration( + &migration_socket, + &src_api_socket, + &dest_api_socket, + true, + downtime, + migration_timeout + ), "Unsuccessful command: 'send-migration' or 'receive-migration'." ); }); @@ -10368,7 +10434,12 @@ mod live_migration { .port() } - fn start_live_migration_tcp(src_api_socket: &str, dest_api_socket: &str) -> bool { + fn start_live_migration_tcp( + src_api_socket: &str, + dest_api_socket: &str, + downtime: u64, + timeout: u64, + ) -> bool { // Get an available TCP port let migration_port = get_available_port(); let host_ip = "127.0.0.1"; @@ -10395,6 +10466,10 @@ mod live_migration { &format!("--api-socket={src_api_socket}"), "send-migration", &format!("tcp:{host_ip}:{migration_port}"), + "--downtime", + &format!("{downtime}"), + "--migration-timeout", + &format!("{timeout}"), ]) .stdin(Stdio::null()) .stderr(Stdio::piped()) @@ -10465,6 +10540,8 @@ mod live_migration { .output() .expect("Expect creating disk image to succeed"); let pmem_path = String::from("/dev/pmem0"); + let downtime = 100000; + let timeout = 1000; // Start the source VM let src_vm_path = clh_command("cloud-hypervisor"); @@ -10527,7 +10604,7 @@ mod live_migration { } // Start TCP live migration assert!( - start_live_migration_tcp(&src_api_socket, &dest_api_socket), + start_live_migration_tcp(&src_api_socket, &dest_api_socket, downtime, timeout), "Unsuccessful command: 'send-migration' or 'receive-migration'." ); }); From e29fa51d23bbe0e36ddb013fc9c8bafa110d132c Mon Sep 17 00:00:00 2001 From: Jinrong Liang Date: Fri, 18 Apr 2025 16:40:26 +0800 Subject: [PATCH 4/7] vmm: Modify dirty page table log level from info to debug Live migration generates a large number of logs with GPA and size information for memory regions. Adjusting the level to debug avoids excessive logging while retaining the information needed for debugging. Signed-off-by: Jinrong Liang --- vmm/src/memory_manager.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/vmm/src/memory_manager.rs b/vmm/src/memory_manager.rs index a5ab297182..4b6a93751e 100644 --- a/vmm/src/memory_manager.rs +++ b/vmm/src/memory_manager.rs @@ -2819,11 +2819,11 @@ impl Migratable for MemoryManager { let sub_table = MemoryRangeTable::from_bitmap(dirty_bitmap, r.gpa, 4096); if sub_table.regions().is_empty() { - info!("Dirty Memory Range Table is empty"); + debug!("Dirty Memory Range Table is empty"); } else { - info!("Dirty Memory Range Table:"); + debug!("Dirty Memory Range Table:"); for range in sub_table.regions() { - info!("GPA: {:x} size: {} (KiB)", range.gpa, range.length / 1024); + debug!("GPA: {:x} size: {} (KiB)", range.gpa, range.length / 1024); } } From c244dae5c83b995db67523f58be2e9d5c7ea9d81 Mon Sep 17 00:00:00 2001 From: Songqian Li Date: Wed, 6 Aug 2025 11:07:29 +0800 Subject: [PATCH 5/7] vmm: Fix the end condition of sending dirty page loop This patch fixes an issue that the loop didn't exit in time when threshold wasn't set. Signed-off-by: Songqian Li --- vmm/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vmm/src/lib.rs b/vmm/src/lib.rs index fc17580653..09c25e8bd0 100644 --- a/vmm/src/lib.rs +++ b/vmm/src/lib.rs @@ -1179,7 +1179,7 @@ impl Vmm { } // Enter the final stage of migration when the suspension conditions are met - if s.iteration > 1 && s.pending_size < s.threshold_size { + if s.iteration > 1 && s.pending_size <= s.threshold_size { break; } From 15468e245e0ca5e2350c4721857df8e9be220784 Mon Sep 17 00:00:00 2001 From: Jinrong Liang Date: Mon, 14 Apr 2025 21:21:23 +0800 Subject: [PATCH 6/7] vm-migration: Add support for downtime limits Add handling of migration timeout failures to provide more flexible live migration options. Implement downtime limiting logic to minimize service disruptions. Support for setting downtime thresholds and migration timeouts. Signed-off-by: Jinrong Liang --- src/bin/ch-remote.rs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/bin/ch-remote.rs b/src/bin/ch-remote.rs index 44bb9caf8c..319b64f32f 100644 --- a/src/bin/ch-remote.rs +++ b/src/bin/ch-remote.rs @@ -1098,6 +1098,22 @@ fn get_cli_commands_sorted() -> Box<[Command]> { .long("local") .num_args(0) .action(ArgAction::SetTrue), + ) + .arg( + Arg::new("downtime") + .long("downtime") + .help("Set the expected maximum downtime in milliseconds") + .num_args(1) + .value_parser(clap::value_parser!(u64).range(1..)) + .default_value("500"), + ) + .arg( + Arg::new("migration_timeout") + .long("migration-timeout") + .help("Set the maximum allowed migration time in seconds") + .num_args(1) + .value_parser(clap::value_parser!(u64)) + .default_value("3600"), ), Command::new("shutdown").about("Shutdown the VM"), Command::new("shutdown-vmm").about("Shutdown the VMM"), From a26ef711c002b70eeeb67f8dfe9ff31074684aeb Mon Sep 17 00:00:00 2001 From: Philipp Schuster Date: Tue, 10 Jun 2025 11:38:41 +0200 Subject: [PATCH 7/7] migration: add vCPU throttling (auto-converge) for pre-copy auto-converge (vCPU throttling) is a crucial technique to migrate VMs with a high dirty rate (high working set with intense usage). It is an alternative to postcopy migration, which is not yet implemented in Cloud Hypervisor. vCPU throttling was implemented with a different thread and a manager for that thread. It is possible to abort vCPU throttling in case one aborts a live-migration, for example. The rather complex thread state management is covered in a unit test ensuring liveliness throughout all possible scenarios. The throttling itself is implemented by utilizing the CpuManager and its pause() and resume() functions. Signed-off-by: Philipp Schuster On-behalf-of: SAP philipp.schuster@sap.com --- tests/integration.rs | 3 +- vmm/src/config.rs | 2 +- vmm/src/lib.rs | 36 ++++ vmm/src/memory_manager.rs | 6 +- vmm/src/vcpu_throttling.rs | 410 +++++++++++++++++++++++++++++++++++++ vmm/src/vm.rs | 32 +++ 6 files changed, 483 insertions(+), 6 deletions(-) create mode 100644 vmm/src/vcpu_throttling.rs diff --git a/tests/integration.rs b/tests/integration.rs index 995ba87558..3d7a99897a 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -520,8 +520,7 @@ fn temp_snapshot_dir_path(tmp_dir: &TempDir) -> String { } fn temp_vmcore_file_path(tmp_dir: &TempDir) -> String { - let vmcore_file = String::from(tmp_dir.as_path().join("vmcore").to_str().unwrap()); - vmcore_file + String::from(tmp_dir.as_path().join("vmcore").to_str().unwrap()) } // Creates the path for direct kernel boot and return the path. diff --git a/vmm/src/config.rs b/vmm/src/config.rs index eefacd2b6a..84a3a07205 100644 --- a/vmm/src/config.rs +++ b/vmm/src/config.rs @@ -592,7 +592,7 @@ impl CpusConfig { // ref type in the match. // The issue will go away once kvm_hyperv is moved under the features // list as it will always be checked for. - #[allow(unused_mut)] + #[allow(unused_mut, clippy::never_loop)] let mut features = CpuFeatures::default(); for s in features_list.0 { match >::as_ref(&s) { diff --git a/vmm/src/lib.rs b/vmm/src/lib.rs index 09c25e8bd0..d986da6573 100644 --- a/vmm/src/lib.rs +++ b/vmm/src/lib.rs @@ -8,6 +8,15 @@ extern crate event_monitor; #[macro_use] extern crate log; +/// Amount of iterations before auto-converging starts. +const AUTO_CONVERGE_ITERATION_DELAY: u64 = 10; +/// Step size in percent to increase the vCPU throttling. +const AUTO_CONVERGE_STEP_SIZE: u8 = 10; +/// Amount of iterations after that we increase vCPU throttling. +const AUTO_CONVERGE_ITERATION_INCREASE: u64 = 2; +/// Maximum vCPU throttling value. +const AUTO_CONVERGE_MAX: u8 = 99; + use std::collections::HashMap; use std::fs::File; use std::io::{stdout, Read, Write}; @@ -89,6 +98,7 @@ mod pci_segment; pub mod seccomp_filters; mod serial_manager; mod sigwinch_listener; +mod vcpu_throttling; pub mod vm; pub mod vm_config; @@ -1135,6 +1145,11 @@ impl Vmm { Ok(true) } + fn can_increase_autoconverge_step(s: &MigrationState) -> bool { + let iteration = s.iteration.saturating_sub(AUTO_CONVERGE_ITERATION_DELAY); + iteration % AUTO_CONVERGE_ITERATION_INCREASE == 0 + } + fn memory_copy_iterations( vm: &mut Vm, socket: &mut SocketStream, @@ -1146,6 +1161,15 @@ impl Vmm { let mut iteration_table; loop { + // todo: check if auto-converge is enabled at all? + if Self::can_increase_autoconverge_step(s) { + let current_throttle = vm.throttle_percent(); + let new_throttle = current_throttle + AUTO_CONVERGE_STEP_SIZE; + let new_throttle = std::cmp::min(new_throttle, AUTO_CONVERGE_MAX); + log::info!("Increasing auto-converge step: percentage={new_throttle}%"); + vm.set_throttle_percent(new_throttle); + } + // Update the start time of the iteration s.iteration_start_time = Instant::now(); @@ -1206,6 +1230,12 @@ impl Vmm { s.pages_per_second = s.current_dirty_pages * 1000 / s.iteration_cost_time.as_millis() as u64; } + debug!( + "iteration {}: cost={}ms, throttle={}%", + s.iteration, + s.iteration_cost_time.as_millis(), + vm.throttle_percent() + ); } Ok(iteration_table) @@ -1260,7 +1290,13 @@ impl Vmm { info!("Entering downtime phase"); s.downtime_start = Instant::now(); + // End throttle thread + info!("stopping vcpu thread"); + vm.stop_vcpu_throttling(); + info!("stopped vcpu thread"); + info!("pausing VM"); vm.pause()?; + info!("paused VM"); // Send last batch of dirty pages let mut final_table = vm.dirty_log()?; diff --git a/vmm/src/memory_manager.rs b/vmm/src/memory_manager.rs index 4b6a93751e..a51d24f4b8 100644 --- a/vmm/src/memory_manager.rs +++ b/vmm/src/memory_manager.rs @@ -2819,11 +2819,11 @@ impl Migratable for MemoryManager { let sub_table = MemoryRangeTable::from_bitmap(dirty_bitmap, r.gpa, 4096); if sub_table.regions().is_empty() { - debug!("Dirty Memory Range Table is empty"); + // debug!("Dirty Memory Range Table is empty"); } else { - debug!("Dirty Memory Range Table:"); + // debug!("Dirty Memory Range Table:"); for range in sub_table.regions() { - debug!("GPA: {:x} size: {} (KiB)", range.gpa, range.length / 1024); + // debug!("GPA: {:x} size: {} (KiB)", range.gpa, range.length / 1024); } } diff --git a/vmm/src/vcpu_throttling.rs b/vmm/src/vcpu_throttling.rs new file mode 100644 index 0000000000..127227c5de --- /dev/null +++ b/vmm/src/vcpu_throttling.rs @@ -0,0 +1,410 @@ +// Copyright © 2025 Cyberus Technology GmbH +// +// SPDX-License-Identifier: Apache-2.0 + +use std::ops::Deref; +use std::sync::{Arc, Condvar, Mutex, MutexGuard}; +use std::thread; +use std::thread::{sleep, JoinHandle}; +use std::time::{Duration, Instant}; + +use vm_migration::Pausable; + +use crate::cpu::CpuManager; + +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +enum ThreadState { + /// Waiting for next event. + Waiting, + /// Ongoing vCPU throttling. + /// + /// The inner value shows the current throttling percentage in range `1..=99`. + Throttling(u8 /* `1..=99` */), + /// Thread is shutting down gracefully. + Exiting, +} + +impl ThreadState { + /// Returns the currently throttling level in percent. + fn throttle_percent(&self) -> u8 { + match self { + ThreadState::Throttling(val) => *val, + _ => 0, + } + } +} + +/// Type to indicate state change acknowledgments between the handler and the +/// thread. +/// +/// See: +/// - [`VcpuThrottleThreadHandle::CHANGE_ACK`] +/// - [`VcpuThrottleThreadHandle::CHANGE_NACK`] +type StateChangeAckT = bool; + +type SynchronizedStateInner = (StateChangeAckT, ThreadState); +type SynchronizedState = (Mutex, Condvar); + +/// Handler for controlling the vCPU throttle thread. +/// +/// vCPU throttling is needed for live-migration of memory-intensive workloads. +/// The current design assumes that all vCPUs are throttled equally. +/// +/// # States and Transitions +/// - `Waiting` -> `Throttle(x %)`, `Exit` +/// - `Throttle(x %)` -> `Waiting`, `Throttle(y %)` +/// - `Exit` +pub struct VcpuThrottleThreadHandle { + /// Thread state wrapped by synchronization primitives. + shared_state: Arc, + /// The underlying thread handle. + // Option so that we can .take() the value later + thread_handle: Option>, +} + +impl VcpuThrottleThreadHandle { + /// The timeslice for a throttling cycle (vCPU pause & resume). + const TIMESLICE_MS: u64 = 500; // QEMU uses here 5000ms + + const THREAD_NAME: &'static str = "vcpu-throttle"; + + /// State change acknowledge by thread. + const CHANGE_ACK: StateChangeAckT = true; + /// State change not yet acknowledge by thread. + const CHANGE_NACK: StateChangeAckT = false; + + /// Waits for the thread to ACK the state. + fn wait_for_thread_ack(&self, guard: MutexGuard) { + debug!("Waiting for thread to ACK waiting state"); + let _guard = self + .shared_state + .1 + .wait_while(guard, |(ack, _)| *ack == Self::CHANGE_ACK); + debug!("Thread has ACKed waiting state"); + } + + /// Updates the state of the thread, and - depending on the state change - + /// waits for the state to become effective gracefully. + /// + /// This function behaves differently depending on the new [`ThreadState`]: + /// - [`ThreadState::Waiting`]: Waits until thread is in wait state + /// - [`ThreadState::Throttling`]: Returns right away and let the thread + /// start its work. In case of an ongoing throttling cycle + /// (vCPU pause & resume), the new state will be applied when the next + /// cycle starts. + /// - [`ThreadState::Exiting`]: Returns right away as the thread exits + /// gracefully. + fn set_state(&self, new_state: ThreadState) { + let mut lock = self.shared_state.0.lock().unwrap(); + let (ack, current_thread_state) = &mut *lock; + + // Panic to catch bugs in the higher management layers. + assert_ne!(new_state, *current_thread_state); + + *ack = Self::CHANGE_NACK; + let old_thread_state = *current_thread_state; + *current_thread_state = new_state; + debug!("Setting new thread state: {new_state:?}"); + + // This is the only state where the thread is actively waiting for new + // commands. Unblock it. + if old_thread_state == ThreadState::Waiting { + self.shared_state.1.notify_one(); + } + + // The only state transition where we care for a synchronization. + if new_state == ThreadState::Waiting { + self.wait_for_thread_ack(lock); + } + } + + /// Helper that executes a callback and then sleeps. Ensures that the time + /// to execute the callback is taken into the account and subtracted from + /// the sleep time. In other words: The function takes as long as + /// `target_duration`. + fn run_min_execution_time_or_sleep(cb: &impl Fn(), target_duration: Duration) { + let begin = Instant::now(); + cb(); + let cb_duration = begin.elapsed(); + debug!("cb_duration: {}ms", cb_duration.as_millis()); + let sleep_duration = target_duration.saturating_sub(cb_duration); + if sleep_duration.as_millis() > 0 { + sleep(sleep_duration); + } + } + + /// Helper for [`Self::build_thread_fn`] returning the function that performs + /// the actual throttling. + fn thread_fn_inner( + shared_state: &SynchronizedState, + throttle_callback: &impl Fn(), + unthrottle_callback: &impl Fn(), + ) { + loop { + let shared_state_guard = shared_state.0.lock().unwrap(); + let (_, thread_state) = &*shared_state_guard; + if !matches!(*thread_state, ThreadState::Throttling(_)) { + // Return to the control loop in case of a state change. + break; + } + + let throttle_percentage = thread_state.throttle_percent() as u64; + + // Dropping early, so that calls to `set_state` don't block. + drop(shared_state_guard); + + let wait_ms_vcpus = Self::TIMESLICE_MS * throttle_percentage / 100; + let wait_ms_thread = Self::TIMESLICE_MS - wait_ms_vcpus; + + // pause vCPUs + Self::run_min_execution_time_or_sleep( + throttle_callback, + Duration::from_millis(wait_ms_vcpus), + ); + // let vCPUs run + Self::run_min_execution_time_or_sleep( + unthrottle_callback, + Duration::from_millis(wait_ms_thread), + ); + } + } + + /// Returns the threads main function. + /// + /// This wraps the actual throttling with the necessary thread state and + /// lifecycle management. + fn build_thread_fn( + shared_state: Arc, + throttle_callback: impl Fn() + Send + 'static, + unthrottle_callback: impl Fn() + Send + 'static, + ) -> impl Fn() { + move || { + // In the outer loop, we gracefully wait for commands. + 'control: loop { + let mut shared_state_guard = shared_state.0.lock().unwrap(); + + // Handle special waiting case: go to sleep and replace state_guard + let shared_state_guard = if shared_state_guard.deref().1 == ThreadState::Waiting { + // Notify handler we are waiting. + shared_state_guard.0 = Self::CHANGE_ACK; + shared_state.1.notify_one(); + + // Wait for a state change. + shared_state + .1 + .wait_while(shared_state_guard, |(ack, _)| *ack == Self::CHANGE_NACK) + .unwrap() + } else { + shared_state_guard + }; + + let (_ack, thread_state) = &*shared_state_guard; + match *thread_state { + ThreadState::Exiting => { + break 'control; + } + ThreadState::Waiting => { + continue 'control; + } + ThreadState::Throttling(_) => {} + } + + // Release lock early to enable handler to change the thread's + // state while vCPU throttling is ongoing. + drop(shared_state_guard); + Self::thread_fn_inner(&shared_state, &throttle_callback, &unthrottle_callback); + } + debug!("thread exited gracefully"); + } + } + + /// Spawns a new thread and returning a handle to it. + /// + /// This function returns when the thread gracefully arrived in + /// [`ThreadState::Waiting`]. + /// + /// # Parameters + /// - `cpu_manager`: CPU manager to pause and resume vCPUs + pub fn new_from_cpu_manager(cpu_manager: &Arc>) -> Self { + let throttle_callback = { + let cpu_manager = cpu_manager.clone(); + Box::new(move || cpu_manager.lock().unwrap().pause().unwrap()) + }; + + let unthrottle_callback = { + let cpu_manager = cpu_manager.clone(); + Box::new(move || cpu_manager.lock().unwrap().resume().unwrap()) + }; + + Self::new(throttle_callback, unthrottle_callback) + } + + /// Spawns a new thread and returning a handle to it. + /// + /// This function returns when the thread gracefully arrived in + /// [`ThreadState::Waiting`]. + /// + /// # Parameters + /// - `throttle_callback`: Function putting all vCPUs into pause state. The + /// function must not perform any artificial delay itself. + /// - `unthrottle_callback`: Function putting all vCPUs back into running + /// state. The function must not perform any artificial delay itself. + fn new( + throttle_callback: Box, + unthrottle_callback: Box, + ) -> Self { + let initial_state = (Self::CHANGE_NACK, ThreadState::Waiting); + let shared_state = Arc::new((Mutex::new(initial_state), Condvar::new())); + + let handle = { + let thread_fn = + Self::build_thread_fn(shared_state.clone(), throttle_callback, unthrottle_callback); + thread::Builder::new() + .name(String::from(Self::THREAD_NAME)) + .spawn(thread_fn) + .expect("should spawn thread") + }; + + let this = Self { + shared_state, + thread_handle: Some(handle), + }; + + let guard = this.shared_state.0.lock().unwrap(); + this.wait_for_thread_ack(guard); + + this + } + + /// Set's the throttle percentage to a value in range `0..=99` and updates + /// the thread's state. + /// + /// Setting the value back to `0` brings the thread back into a waiting + /// state. This is a convenient wrapper around [`Self::set_state`]. + /// + /// In case of an ongoing throttling cycle (vCPU pause & resume), the new + /// will be applied when the cycle ends. + /// + /// # Panic + /// Panics, if `percent_new` is not in range `0..=99`. + pub fn set_throttle_percent(&self, percent_new: u8) { + assert!( + percent_new <= 100, + "setting a percentage of 100 or above is not allowed: {percent_new}%" + ); + + // We have no problematic race condition here as in normal operation + // there is exactly one thread calling these functions. + let percent_old = self.throttle_percent(); + + // Return early, no action needed. + if percent_old == percent_new { + return; + } + + if percent_new == 0 { + self.set_state(ThreadState::Waiting); + } else { + self.set_state(ThreadState::Throttling(percent_new)); + } + } + + /// Get the current throttle percentage in range `0..=99`. + /// + /// Please note that the value is not synchronized. + pub fn throttle_percent(&self) -> u8 { + let shared_state_guard = self.shared_state.0.lock().unwrap(); + let state = shared_state_guard.deref().1; + state.throttle_percent() + } + + /// Stops and terminates the thread gracefully. + /// + /// Waits for the thread to finish. + pub fn stop(&mut self) { + if let Some(handle) = self.thread_handle.take() { + self.set_state(ThreadState::Exiting); + handle.join().expect("thread should have succeeded"); + } + } +} + +impl Drop for VcpuThrottleThreadHandle { + fn drop(&mut self) { + // Idempotent; in case this wasn't called. + self.stop(); + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicBool, Ordering}; + + use super::*; + + /*fn setup_stderr_logger() { + use log::{LevelFilter, Metadata, Record};; + struct Logger; + + impl log::Log for Logger { + fn enabled(&self, _metadata: &Metadata) -> bool { + true + } + + fn log(&self, record: &Record) { + eprintln!("{}: {}", record.level(), record.args()); + } + + fn flush(&self) {} + } + + static LOGGER: Logger = Logger; + log::set_logger(&LOGGER).unwrap(); + log::set_max_level(LevelFilter::max()); + }*/ + + #[test] + fn test_thread_lifecycle() { + //setup_stderr_logger(); + + // Dummy CpuManager + let cpus_throttled = Arc::new(AtomicBool::new(false)); + let throttle_callback = { + let cpus_running = cpus_throttled.clone(); + Box::new(move || { + let old = cpus_running.swap(true, Ordering::SeqCst); + assert_eq!(old, false); + }) + }; + let unthrottle_callback = { + let cpus_running = cpus_throttled.clone(); + Box::new(move || { + let old = cpus_running.swap(false, Ordering::SeqCst); + assert_eq!(old, true); + }) + }; + + let mut handler = VcpuThrottleThreadHandle::new(throttle_callback, unthrottle_callback); + handler.set_state(ThreadState::Throttling(5)); + sleep(Duration::from_millis( + VcpuThrottleThreadHandle::TIMESLICE_MS, + )); + handler.set_state(ThreadState::Throttling(10)); + sleep(Duration::from_millis( + VcpuThrottleThreadHandle::TIMESLICE_MS, + )); + + // Assume we aborted vCPU throttling (or the live-migration at all). + handler.set_state(ThreadState::Waiting); + handler.set_state(ThreadState::Throttling(5)); + sleep(Duration::from_millis( + VcpuThrottleThreadHandle::TIMESLICE_MS, + )); + handler.set_state(ThreadState::Throttling(10)); + sleep(Duration::from_millis( + VcpuThrottleThreadHandle::TIMESLICE_MS, + )); + + handler.stop(); + } +} diff --git a/vmm/src/vm.rs b/vmm/src/vm.rs index d7bba25cc0..d062f16b28 100644 --- a/vmm/src/vm.rs +++ b/vmm/src/vm.rs @@ -91,6 +91,7 @@ use crate::migration::get_vm_snapshot; #[cfg(all(target_arch = "x86_64", feature = "guest_debug"))] use crate::migration::url_to_file; use crate::migration::{url_to_path, SNAPSHOT_CONFIG_FILE, SNAPSHOT_STATE_FILE}; +use crate::vcpu_throttling::VcpuThrottleThreadHandle; use crate::vm_config::{ DeviceConfig, DiskConfig, FsConfig, HotplugMethod, NetConfig, NumaConfig, PayloadConfig, PmemConfig, UserDeviceConfig, VdpaConfig, VmConfig, VsockConfig, @@ -482,6 +483,7 @@ pub struct Vm { hypervisor: Arc, stop_on_boot: bool, load_payload_handle: Option>>, + vcpu_throttle_thread_handle: VcpuThrottleThreadHandle, } impl Vm { @@ -751,6 +753,9 @@ impl Vm { VmState::Created }; + let vcpu_throttle_thread_handle = + VcpuThrottleThreadHandle::new_from_cpu_manager(&cpu_manager); + Ok(Vm { #[cfg(feature = "tdx")] kernel, @@ -770,6 +775,7 @@ impl Vm { hypervisor, stop_on_boot, load_payload_handle, + vcpu_throttle_thread_handle, }) } @@ -857,6 +863,32 @@ impl Vm { Ok(numa_nodes) } + /// Set's the throttle percentage to a value in range `0..=99`. + /// + /// Setting the value back to `0` brings the thread back into a waiting + /// state. + /// + /// # Panic + /// Panics, if `percent_new` is not in range `0..=99`. + pub fn set_throttle_percent(&self, percent: u8 /* 1..=99 */) { + self.vcpu_throttle_thread_handle + .set_throttle_percent(percent); + } + + /// Get the current throttle percentage in range `0..=99`. + /// + /// Please note that the value is not synchronized. + pub fn throttle_percent(&self) -> u8 { + self.vcpu_throttle_thread_handle.throttle_percent() + } + + /// Stops and terminates the thread gracefully. + /// + /// Waits for the thread to finish. + pub fn stop_vcpu_throttling(&mut self) { + self.vcpu_throttle_thread_handle.stop(); + } + #[allow(clippy::too_many_arguments)] pub fn new( vm_config: Arc>,