Merge pull request #11510 from lifupan/sync_resize_vcpu

runtime-rs: make the resize_vcpu api support sync

src/dragonball/src/api/v1/vmm_action.rs

@@ -9,8 +9,9 @@
 use std::fs::File;
 use std::sync::{Arc, Mutex};
 
-use crossbeam_channel::{unbounded, Receiver, Sender, TryRecvError};
+use crossbeam_channel::{Receiver, Sender, TryRecvError};
 use log::{debug, error, info, warn};
+use std::sync::mpsc;
 use tracing::instrument;
 
 use crate::error::{Result, StartMicroVmError, StopMicrovmError};
@@ -284,7 +285,7 @@ pub enum VmmData {
     /// Return vfio device's slot number in guest.
     VfioDeviceData(Option<u8>),
     /// Sync Hotplug
-    SyncHotplug((Sender<Option<i32>>, Receiver<Option<i32>>)),
+    SyncHotplug((mpsc::Sender<Option<i32>>, mpsc::Receiver<Option<i32>>)),
 }
 
 /// Request data type used to communicate between the API and the VMM.
@@ -900,7 +901,7 @@ impl VmmService {
             }
         })?;
 
-        let (sender, receiver) = unbounded();
+        let (sender, receiver) = mpsc::channel();
 
         // It is safe because we don't expect poison lock.
         let vfio_manager = vm.device_manager.vfio_manager.lock().unwrap();
@@ -965,15 +966,17 @@ impl VmmService {
             ));
         }
 
+        let (sender, revceiver) = mpsc::channel();
+
         #[cfg(feature = "dbs-upcall")]
-        vm.resize_vcpu(config, None).map_err(|e| {
+        vm.resize_vcpu(config, Some(sender.clone())).map_err(|e| {
             if let VcpuResizeError::UpcallServerNotReady = e {
                 return VmmActionError::UpcallServerNotReady;
             }
             VmmActionError::ResizeVcpu(e)
         })?;
 
-        Ok(VmmData::Empty)
+        Ok(VmmData::SyncHotplug((sender, revceiver)))
     }
 
     #[cfg(feature = "virtio-mem")]

src/dragonball/src/device_manager/vfio_dev_mgr/mod.rs

@@ -16,9 +16,9 @@ use std::collections::HashMap;
 use std::ops::Deref;
 use std::os::fd::RawFd;
 use std::path::Path;
+use std::sync::mpsc::Sender;
 use std::sync::{Arc, Weak};
 
-use crossbeam_channel::Sender;
 use dbs_device::resources::Resource::LegacyIrq;
 use dbs_device::resources::{DeviceResources, Resource, ResourceConstraint};
 use dbs_device::DeviceIo;

src/dragonball/src/vcpu/vcpu_manager.rs

@@ -225,7 +225,7 @@ pub struct VcpuManager {
     vm_as: GuestAddressSpaceImpl,
     pub(crate) vm_fd: Arc<VmFd>,
 
-    action_sycn_tx: Option<Sender<bool>>,
+    action_sycn_tx: Option<Sender<Option<i32>>>,
     vcpus_in_action: (VcpuAction, Vec<u8>),
     pub(crate) reset_event_fd: Option<EventFd>,
 
@@ -756,7 +756,9 @@ impl VcpuManager {
 
     fn sync_action_finish(&mut self, got_error: bool) {
         if let Some(tx) = self.action_sycn_tx.take() {
-            if let Err(e) = tx.send(got_error) {
+            let result = if got_error { 0 } else { -1 };
+
+            if let Err(e) = tx.send(Some(result)) {
                 debug!("cpu sync action send to closed channel {}", e);
             }
         }
@@ -856,7 +858,7 @@ mod hotplug {
         pub fn resize_vcpu(
             &mut self,
             vcpu_count: u8,
-            sync_tx: Option<Sender<bool>>,
+            sync_tx: Option<Sender<Option<i32>>>,
         ) -> std::result::Result<(), VcpuResizeError> {
             if self.get_vcpus_action() != VcpuAction::None {
                 return Err(VcpuResizeError::VcpuIsHotplugging);

src/dragonball/src/vm/mod.rs

@@ -832,7 +832,7 @@ impl Vm {
     pub fn resize_vcpu(
         &mut self,
         config: VcpuResizeInfo,
-        sync_tx: Option<Sender<bool>>,
+        sync_tx: Option<Sender<Option<i32>>>,
     ) -> std::result::Result<(), VcpuResizeError> {
         if self.upcall_client().is_none() {
             Err(VcpuResizeError::UpcallClientMissing)

src/runtime-rs/crates/hypervisor/src/dragonball/inner.rs

@@ -28,7 +28,7 @@ use kata_types::{
 };
 use nix::mount::MsFlags;
 use persist::sandbox_persist::Persist;
-use std::cmp::Ordering;
+use std::{cmp::Ordering, time::Duration};
 use std::{collections::HashSet, fs::create_dir_all};
 use tokio::sync::mpsc;
 
@@ -37,6 +37,9 @@ const DRAGONBALL_INITRD: &str = "initrd";
 const DRAGONBALL_ROOT_FS: &str = "rootfs";
 const BALLOON_DEVICE_ID: &str = "balloon0";
 const MEM_DEVICE_ID: &str = "memmr0";
+/// default hotplug timeout
+const DEFAULT_HOTPLUG_TIMEOUT: u64 = 250;
+
 #[derive(Debug)]
 pub struct DragonballInner {
     /// sandbox id
@@ -391,7 +394,10 @@ impl DragonballInner {
             vcpu_count: Some(new_vcpus as u8),
         };
         self.vmm_instance
-            .resize_vcpu(&cpu_resize_info)
+            .resize_vcpu(
+                &cpu_resize_info,
+                Some(Duration::from_millis(DEFAULT_HOTPLUG_TIMEOUT)),
+            )
             .context(format!(
                 "failed to do_resize_vcpus on new_vcpus={:?}",
                 new_vcpus

src/runtime-rs/crates/hypervisor/src/dragonball/vmm_instance.rs

@@ -9,6 +9,7 @@ use std::{
     os::unix::{io::IntoRawFd, prelude::AsRawFd},
     sync::{Arc, Mutex, RwLock},
     thread,
+    time::Duration,
 };
 
 use anyhow::{anyhow, Context, Result};
@@ -292,9 +293,17 @@ impl VmmInstance {
         Ok(())
     }
 
-    pub fn resize_vcpu(&self, cfg: &VcpuResizeInfo) -> Result<()> {
-        self.handle_request(Request::Sync(VmmAction::ResizeVcpu(cfg.clone())))
+    pub fn resize_vcpu(&self, cfg: &VcpuResizeInfo, timeout: Option<Duration>) -> Result<()> {
+        let vmmdata = self
+            .handle_request(Request::Sync(VmmAction::ResizeVcpu(cfg.clone())))
             .with_context(|| format!("Failed to resize_vm(hotplug vcpu), cfg: {:?}", cfg))?;
+
+        if let Some(timeout) = timeout {
+            if let VmmData::SyncHotplug((_, receiver)) = vmmdata {
+                let _ = receiver.recv_timeout(timeout)?;
+            }
+        }
+
         Ok(())
     }