Implement a host memory monitor

This monitor checks the ratio of memory available on the host and, if
needed, contacts the VM (through muvm-server) to adjust
/proc/sys/vm/watermark_scale_factor and, if the situation is critical,
to request it to drop its caches.

Signed-off-by: Sergio Lopez <[email protected]>

Author: slp

Cargo.lock

@@ -16,6 +16,7 @@ env_logger = { version = "0.11.3", default-features = false, features = ["auto-c
 krun-sys = { path = "../krun-sys", version = "1.9.1", default-features = false, features = [] }
 log = { version = "0.4.21", default-features = false, features = ["kv"] }
 nix = { version = "0.28.0", default-features = false, features = ["user"] }
+procfs = { version = "0.17.0", default-features = false, features = [] }
 rustix = { version = "0.38.34", default-features = false, features = ["fs", "mount", "process", "std", "stdio", "system", "use-libc-auxv"] }
 serde = { version = "1.0.203", default-features = false, features = ["derive"] }
 serde_json = { version = "1.0.117", default-features = false, features = ["std"] }

crates/muvm/Cargo.toml

@@ -16,6 +16,7 @@ use muvm::cli_options::options;
 use muvm::cpu::{get_fallback_cores, get_performance_cores};
 use muvm::env::{find_muvm_exec, prepare_env_vars};
 use muvm::launch::{launch_or_lock, LaunchResult};
+use muvm::monitor::spawn_monitor;
 use muvm::net::{connect_to_passt, start_passt};
 use muvm::types::MiB;
 use nix::sys::sysinfo::sysinfo;
@@ -435,6 +436,8 @@ fn main() -> Result<()> {
         }
     }
 
+    spawn_monitor(options.root_server_port, cookie);
+
     {
         // Start and enter the microVM. Unless there is some error while creating the
         // microVM this function never returns.

crates/muvm/src/bin/muvm.rs

@@ -148,7 +148,7 @@ fn lock_file() -> Result<(Option<File>, Uuid)> {
     Ok((Some(lock_file), cookie))
 }
 
-fn request_launch(
+pub fn request_launch(
     server_port: u32,
     cookie: Uuid,
     command: PathBuf,

crates/muvm/src/launch.rs

@@ -2,6 +2,7 @@ pub mod cli_options;
 pub mod cpu;
 pub mod env;
 pub mod launch;
+pub mod monitor;
 pub mod net;
 pub mod types;
 

crates/muvm/src/lib.rs

@@ -0,0 +1,96 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::thread;
+use std::time;
+
+use anyhow::Result;
+use log::debug;
+use procfs::{Current, Meminfo};
+use uuid::Uuid;
+
+use crate::launch::request_launch;
+
+#[derive(Clone, Debug, PartialEq)]
+pub enum GuestPressure {
+    None,
+    Low,
+    Medium,
+    High,
+    Critical,
+}
+
+impl From<GuestPressure> for u32 {
+    fn from(pressure: GuestPressure) -> u32 {
+        match pressure {
+            GuestPressure::None => 10,
+            GuestPressure::Low => 1000,
+            GuestPressure::Medium => 2000,
+            GuestPressure::High => 3000,
+            // Same waterlevel as High, but also explicitly requesting
+            // the guest to drop its page cache.
+            GuestPressure::Critical => 3000,
+        }
+    }
+}
+
+pub fn spawn_monitor(server_port: u32, cookie: Uuid) {
+    thread::spawn(move || run(server_port, cookie));
+}
+
+fn set_guest_pressure(server_port: u32, cookie: Uuid, pressure: GuestPressure) -> Result<()> {
+    if pressure == GuestPressure::Critical {
+        debug!("requesting the guest to drop its caches");
+        // This is a fake command that tells muvm-server to write to "/proc/sys/vm/drop_caches"
+        let command = PathBuf::from("/muvmdropcaches");
+        let command_args = vec![];
+        let env = HashMap::new();
+        request_launch(server_port, cookie, command, command_args, env)?;
+    }
+
+    let wsf: u32 = pressure.into();
+    debug!("setting watermark_scale_factor to {wsf}");
+
+    let command = PathBuf::from("/sbin/sysctl");
+    let command_args = vec![format!("vm.watermark_scale_factor={}", wsf)];
+    let env = HashMap::new();
+    request_launch(server_port, cookie, command, command_args, env)
+}
+
+fn run(server_port: u32, cookie: Uuid) {
+    let mut guest_pressure = GuestPressure::None;
+    loop {
+        let meminfo = Meminfo::current().ok();
+        if let Some(meminfo) = meminfo {
+            if let Some(available) = meminfo.mem_available {
+                let avail_ratio = (available * 100) / meminfo.mem_total;
+                debug!(
+                    "avail_ratio={avail_ratio}, avail={available}, total={}",
+                    meminfo.mem_total
+                );
+                let new_pressure = if avail_ratio <= 10 {
+                    GuestPressure::Critical
+                } else if avail_ratio <= 15 {
+                    GuestPressure::High
+                } else if avail_ratio <= 20 {
+                    GuestPressure::Medium
+                } else if avail_ratio <= 25 {
+                    GuestPressure::Low
+                } else {
+                    GuestPressure::None
+                };
+
+                debug!("Pressure at {:?}", new_pressure);
+
+                if new_pressure != guest_pressure {
+                    if let Err(err) = set_guest_pressure(server_port, cookie, new_pressure.clone())
+                    {
+                        println!("Failed to set the new pressure in the guest: {err}");
+                    } else {
+                        guest_pressure = new_pressure;
+                    }
+                }
+            }
+        }
+        thread::sleep(time::Duration::from_millis(500));
+    }
+}

crates/muvm/src/monitor.rs

@@ -1,6 +1,8 @@
 use std::collections::HashMap;
+use std::fs::File;
+use std::io::Write;
 use std::os::unix::process::ExitStatusExt as _;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::process::{ExitStatus, Stdio};
 use std::{env, io};
 
@@ -18,6 +20,11 @@ use uuid::Uuid;
 use crate::utils::launch::Launch;
 use crate::utils::stdio::make_stdout_stderr;
 
+pub enum ConnRequest {
+    DropCaches,
+    ExecuteCommand { command: PathBuf, child: Child },
+}
+
 #[derive(Debug)]
 pub struct Worker {
     cookie: Uuid,
@@ -60,9 +67,12 @@ impl Worker {
                     let stream = BufStream::new(stream);
 
                     match handle_connection(self.cookie, stream).await {
-                        Ok((command, mut child)) => {
-                            self.child_set.spawn(async move { (command, child.wait().await) });
-                            self.set_child_processes(self.child_set.len());
+                        Ok(request) => match request {
+                            ConnRequest::DropCaches => {},
+                            ConnRequest::ExecuteCommand {command, mut child } => {
+                                self.child_set.spawn(async move { (command, child.wait().await) });
+                                self.set_child_processes(self.child_set.len());
+                            }
                         },
                         Err(err) => {
                             eprintln!("Failed to process client request: {err:?}");
@@ -164,7 +174,7 @@ async fn read_request(stream: &mut BufStream<TcpStream>) -> Result<Launch> {
 async fn handle_connection(
     server_cookie: Uuid,
     mut stream: BufStream<TcpStream>,
-) -> Result<(PathBuf, Child)> {
+) -> Result<ConnRequest> {
     let mut envs: HashMap<String, String> = env::vars().collect();
 
     let Launch {
@@ -182,6 +192,21 @@ async fn handle_connection(
         return Err(anyhow!(msg));
     }
 
+    if command == Path::new("/muvmdropcaches") {
+        let mut file = File::options()
+            .write(true)
+            .open("/proc/sys/vm/drop_caches")
+            .context("Failed to open /proc/sys/vm/drop_caches for writing")?;
+
+        {
+            file.write_all(b"1")
+                .context("Failed to write to /proc/sys/vm/drop_caches")?;
+        }
+        stream.write_all(b"OK").await.ok();
+        stream.flush().await.ok();
+        return Ok(ConnRequest::DropCaches);
+    }
+
     envs.extend(env);
 
     let (stdout, stderr) = make_stdout_stderr(&command, &envs)?;
@@ -202,5 +227,5 @@ async fn handle_connection(
     }
     stream.flush().await.ok();
 
-    res.map(|child| (command, child))
+    res.map(|child| ConnRequest::ExecuteCommand { command, child })
 }