vmm: fix CpuManager::resume(): gracefully wait for run vCPU loop ACK

Fix a race condition that happens in resume()-pause() cycles.

It is odd that for pause(), the CpuManager waited via `state.paused`
for the vCPU thread to ACK the state change but not for `resume()`.
In the `resume()` case, oddly CpuManager "owned" the state change
in `state.paused`. This commit changes this so that the vCPU ACKs
its state change itself in `state.paused` when it transitions from
pause->run. Further, `CpuManager::resume()` now gracefully waits for
the vCPU to be resumed.

More technical:

This change ensures proper synchronization and prevents situations
in that park() follows right after unpark(), causing deadlocks and
other weird behavior due to race conditions.

Calling resume() now takes slightly longer, very similar to pause().
This is, however, even for 254 vCPUs in the range of less than 10ms,
and ultimately we now have correct behaviour.

## Reproducer

Since [0] is merged, the underlying problem can be tested without this
commit by modifying the pause() API call to run `CpuManager::pause()`
and `CpuManager::resume()` in a loop a thousand times.

`ch-remote --api-socket ... pause`

```patch
diff --git a/vmm/src/vm.rs b/vmm/src/vm.rs
index d7bba25cc..35557d58f 100644
--- a/vmm/src/vm.rs
+++ b/vmm/src/vm.rs
@@ -2687,6 +2687,10 @@ impl Pausable for Vm {
             MigratableError::Pause(anyhow!("Error activating pending virtio devices: {:?}", e))
         })?;

+        for _ in 0..1000 {
+            self.cpu_manager.lock().unwrap().pause()?;
+            self.cpu_manager.lock().unwrap().resume()?;
+        }
         self.cpu_manager.lock().unwrap().pause()?;
         self.device_manager.lock().unwrap().pause()?;
```

## Outlook

Decades of experience in VMM development showed us that using
many AtomicBools is a footgun. They are not synchronized with each
other at all. On the long term, we might want to refactor things to
have a single shared AtomicU64 with different bits having different
meanings.

[0] https://github.com/cloud-hypervisor/cloud-hypervisor/pull/7290

Signed-off-by: Philipp Schuster <philipp.schuster@cyberus-technology.de>
On-behalf-of: SAP philipp.schuster@sap.com

vmm/src/cpu.rs

@@ -684,6 +684,7 @@ struct VcpuState {
     handle: Option<thread::JoinHandle<()>>,
     kill: Arc<AtomicBool>,
     vcpu_run_interrupted: Arc<AtomicBool>,
+    /// Used to ACK state changes from the run vCPU loop to the CPU Manager.
     paused: Arc<AtomicBool>,
 }
 
@@ -1179,6 +1180,7 @@ impl CpuManager {
                                 while vcpus_pause_signalled.load(Ordering::SeqCst) {
                                     thread::park();
                                 }
+                                vcpu_paused.store(false, Ordering::SeqCst);
                                 vcpu_run_interrupted.store(false, Ordering::SeqCst);
                             }
 
@@ -2454,6 +2456,7 @@ impl Pausable for CpuManager {
         // activated vCPU change their state to ensure they have parked.
         for state in self.vcpu_states.iter() {
             if state.active() {
+                // wait for vCPU to update state
                 while !state.paused.load(Ordering::SeqCst) {
                     // To avoid a priority inversion with the vCPU thread
                     thread::sleep(std::time::Duration::from_millis(1));
@@ -2465,16 +2468,26 @@ impl Pausable for CpuManager {
     }
 
     fn resume(&mut self) -> std::result::Result<(), MigratableError> {
-        // Toggle the vCPUs pause boolean
+        // Ensure that vCPUs keep running after being unpark() in
+        // their run vCPU loop.
         self.vcpus_pause_signalled.store(false, Ordering::SeqCst);
 
-        // Unpark all the VCPU threads.
-        // Once unparked, the next thing they will do is checking for the pause
-        // boolean. Since it'll be set to false, they will exit their pause loop
-        // and go back to vmx root.
-        for state in self.vcpu_states.iter() {
-            state.paused.store(false, Ordering::SeqCst);
-            state.unpark_thread();
+        // Unpark all the vCPU threads.
+        // Step 1/2: signal each thread
+        {
+            for state in self.vcpu_states.iter() {
+                state.unpark_thread();
+            }
+        }
+        // Step 2/2: wait for state ACK
+        {
+            for state in self.vcpu_states.iter() {
+                // wait for vCPU to update state
+                while state.paused.load(Ordering::SeqCst) {
+                    // To avoid a priority inversion with the vCPU thread
+                    thread::sleep(std::time::Duration::from_millis(1));
+                }
+            }
         }
         Ok(())
     }