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/src/bin/ch-remote.rs b/src/bin/ch-remote.rs index ec363342de..319b64f32f 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) @@ -1060,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"), diff --git a/tests/integration.rs b/tests/integration.rs index 0536864fd2..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. @@ -9208,6 +9207,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 +9229,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 +9249,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 +9271,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 +9452,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 +9636,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 +9864,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 +10090,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 +10210,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 +10367,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 +10433,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 +10465,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 +10539,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 +10603,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'." ); }); 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/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: 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 03c8ee773e..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}; @@ -20,13 +29,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}; @@ -88,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; @@ -640,6 +651,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 +1126,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 +1145,176 @@ 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, + s: &mut MigrationState, + migration_timeout: Duration, + migrate_downtime_limit: Duration, + ) -> result::Result { + let mut bandwidth = 0.0; + 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(); + + // 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; + } + debug!( + "iteration {}: cost={}ms, throttle={}%", + s.iteration, + s.iteration_cost_time.as_millis(), + vm.throttle_percent() + ); + } + + 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(); + // 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()?; + 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 +1322,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 +1401,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 +1428,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 diff --git a/vmm/src/memory_manager.rs b/vmm/src/memory_manager.rs index a5ab297182..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() { - 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); } } 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>,