Skip to content

Pass immutable data to ScalingEngines #109

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lforchini wants to merge 1 commit into from
Open

Pass immutable data to ScalingEngines #109

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions src/backends/fake/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,6 @@ impl FakeConfig {
struct VmFile {
thread_count: u32,
vcpu_count: u32,
per_thread_util: f64,
/// First-snapshot ownership. `true` lets the controller scale the VM.
managed: bool,
fail_snapshot: bool,
Expand All @@ -101,7 +100,6 @@ impl Default for VmFile {
Self {
thread_count: 1,
vcpu_count: 1,
per_thread_util: 0.0,
managed: true,
fail_snapshot: false,
read_io_count: 0,
Expand Down Expand Up @@ -162,6 +160,7 @@ fn discover_vms(state_dir: &Path) -> Vec<Arc<Instance>> {
let Some(vm) = read_vm(&path, id) else {
continue;
};
let pid = fake_pid(id);
let client = FakeClient {
vm_path: path.clone(),
calls_path: state_dir.join("calls.json"),
Expand All @@ -170,7 +169,7 @@ fn discover_vms(state_dir: &Path) -> Vec<Arc<Instance>> {
instances.push(Arc::new(Instance::new(
id.to_string(),
Path::new(id),
0,
pid,
client,
)));
}
Expand All @@ -191,6 +190,13 @@ struct FakeClient {
managed: bool,
}

fn fake_pid(id: &str) -> i32 {
let hash = id.bytes().fold(0u32, |hash, byte| {
hash.wrapping_mul(31).wrapping_add(u32::from(byte))
});
10_000 + (hash % 1_000_000) as i32
}

/// `<vm-id>.json` shares `<vm-id>.lock` with the pytest harness.
fn lock_path_for(vm_path: &std::path::Path) -> std::path::PathBuf {
vm_path.with_extension("lock")
Expand Down Expand Up @@ -267,7 +273,6 @@ impl InstanceClient for FakeClient {
)));
}
Ok(ThreadPoolSnapshot {
thread_count: vm.thread_count,
vcpu_count: vm.vcpu_count,
perf: Some(InstancePerfSample {
read_io_count: vm.read_io_count,
Expand All @@ -277,7 +282,6 @@ impl InstanceClient for FakeClient {
write_bytes_total: 0,
..Default::default()
}),
per_thread_util: Some(vm.per_thread_util),
})
}

Expand Down
34 changes: 3 additions & 31 deletions src/backends/qemu/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,7 @@
//! * the [`super::libvirt::LibvirtQmp`] transport (QMP multiplexed through
//! libvirt's `virDomainQemuMonitorCommand`);
//! * the per-VM topology cache (iothread ids + TIDs, virtio-scsi device path,
//! virtqueue count);
//! * the per-tick CPU sample used to compute per-IOThread utilisation.
//! virtqueue count).
//!
//! Everything the controller used to keep in `QemuInstance` /
//! `QemuState` now lives here, behind the generic per-VM interface.
Expand All @@ -29,9 +28,7 @@ use crate::{
backends::{
BackendClientError, IoThreadProperties, VqMapping,
qemu::{
helpers::{
CpuSample, QemuError, QemuTopology, next_qemu_iothread_id, round_robin_vq_mapping,
},
helpers::{QemuError, QemuTopology, next_qemu_iothread_id, round_robin_vq_mapping},
libvirt::LibvirtQmp,
},
},
Expand Down Expand Up @@ -69,8 +66,6 @@ struct Inner {
/// Number of command virtqueues exposed by the device.
vq_count: u16,

/// Previous per-task CPU sample used to calculate utilisation.
last_sample: Option<CpuSample>,
/// Whether we've already warned that prometheus metrics are unsupported.
warned_unsupported_metrics: bool,
}
Expand Down Expand Up @@ -180,10 +175,7 @@ impl QemuInstanceClient {
};

Ok(ThreadPoolSnapshot {
thread_count,
perf,
// Cannot derive per-thread util without fresh topology/TID data.
per_thread_util: None,
vcpu_count: self.vcpu_count,
})
}
Expand Down Expand Up @@ -271,31 +263,13 @@ impl InstanceClient for QemuInstanceClient {
"invalid pid {pid} from tid {first_tid}"
)));
}
let sample = CpuSample::new(pid, &topo.iothread_tids)?;
if sample.jiffies.is_empty() {
return Err(BackendClientError::InvalidState(
"qemu per-thread sample empty".into(),
));
}

let mut s = self.inner.lock().await;
s.pid = pid;
s.iothreads = topo.iothreads.clone();
s.iothread_tids = topo.iothread_tids.clone();
s.device_path = topo.device_path.clone();
s.vq_count = topo.vq_count;

let per_thread_util = if let Some(prev) = s.last_sample.take() {
let util = &sample - &prev;
s.last_sample = Some(sample);
Some(util)
} else {
// Warm-up tick: report None so the engine leaves
// per_thread_util at 0.0. Next tick has a `prev`
// sample and returns a real delta.
s.last_sample = Some(sample);
None
};
// Feed block statistics to the engine so
// revert-on-drop validation compares apples to apples
// across backends. A best-effort call: an error here
Expand All @@ -320,11 +294,9 @@ impl InstanceClient for QemuInstanceClient {
{
perf.per_vq_depth = Some(topo.vq_inflight.clone());
}
let thread_count = topo.iothreads.len() as u32;

Ok(ThreadPoolSnapshot {
thread_count,
perf,
per_thread_util,
vcpu_count: self.vcpu_count,
})
}
Expand Down
96 changes: 3 additions & 93 deletions src/backends/qemu/helpers.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//! Shared QEMU helpers.
//!
//! Everything here is transport-neutral: prometheus-body topology
//! parsing, /proc jiffies sampling, VQ round-robin, iothread id
//! allocation, and the [`QemuError`] shape used by the QMP
//! parsing, VQ round-robin, iothread id allocation, and the [`QemuError`]
//! shape used by the QMP
//! transport in [`crate::backends::qemu::libvirt`]. The libvirt-
//! backed transport lives in `backends/qemu/libvirt.rs`; the
//! per-VM state that used to live in this module now belongs to
Expand All @@ -16,23 +16,16 @@
use std::{
collections::{BTreeMap, HashMap, HashSet},
io,
ops::Sub,
sync::{
LazyLock,
atomic::{AtomicU64, Ordering},
},
time::Instant,
};

use procfs::process::Process;
use regex::Regex;
use rustix::param;
use thiserror::Error;

use crate::{
backends::{BackendClientError, VqMapping},
util::Path,
};
use crate::backends::{BackendClientError, VqMapping};

/// Errors produced while querying or reconfiguring a QEMU vm.
#[derive(Debug, Error)]
Expand Down Expand Up @@ -198,89 +191,6 @@ impl QemuTopology {
}
}

// ---------------------------------------------------------------------
// /proc helpers
// ---------------------------------------------------------------------

/// Per-IOThread cumulative CPU counters sampled at one wall-clock instant.
#[derive(Debug, Clone)]
pub struct CpuSample {
/// Cumulative user-plus-system jiffies keyed by host task ID.
pub jiffies: HashMap<i32, u64>,
/// Wall-clock time at which the counters were sampled.
// FIXME rename to sampled_at
pub wall: Instant,
}

impl Default for CpuSample {
fn default() -> Self {
Self {
jiffies: HashMap::new(),
wall: Instant::now(),
}
}
}

impl CpuSample {
/// Sample cumulative CPU jiffies for every discovered IOThread task.
pub fn new(
pid: i32,
iothread_tids: &BTreeMap<String, i32>,
) -> Result<Self, BackendClientError> {
let mut out = Self {
jiffies: HashMap::new(),
wall: Instant::now(),
};

let p = Process::new_with_root(Path::proc_pid(pid))?;
for &tid in iothread_tids.values() {
let stat = p.task_from_tid(tid)?.stat()?;
out.jiffies.insert(tid, stat.utime + stat.stime);
}
Ok(out)
}

/// Compute average per-IOThread CPU utilisation between two samples.
// TODO add unit test for this function and the call site, and that it can be
// folded into the Sub.
pub fn diff_jiffies_sample(prev: &Self, cur: &Self) -> f64 {
let Some(elapsed) = cur.wall.checked_duration_since(prev.wall) else {
return 0.0;
};
let elapsed = elapsed.as_secs_f64();
if elapsed <= 0.0 {
return 0.0;
}
let hz = param::clock_ticks_per_second() as f64;
assert!(hz > 0.0);
let mut total = 0.0;
let mut counted = 0u32;
for (&tid, &after) in &cur.jiffies {
let Some(&before) = prev.jiffies.get(&tid) else {
continue;
};
if after < before {
continue;
}
let delta = (after - before) as f64;
let frac = ((delta / hz) / elapsed).clamp(0.0, 1.0);
total += frac;
counted += 1;
}
if counted == 0 {
return 0.0;
}
total / counted as f64
}
}

impl Sub<&CpuSample> for &CpuSample {
type Output = f64;
fn sub(self, rhs: &CpuSample) -> Self::Output {
CpuSample::diff_jiffies_sample(rhs, self)
}
}

// ---------------------------------------------------------------------
// VQ mapping helpers
// ---------------------------------------------------------------------
Expand Down
19 changes: 8 additions & 11 deletions src/backends/qemu/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ use serde::{Deserialize, Serialize};
use crate::{
backends::{Backend, BackendClientError, BackendRegistration},
config::{self, ConfigError},
instance::{Instance, InstanceClient},
instance::{Instance, InstanceClient, ThreadNameFilter},
util::Path,
};

Expand Down Expand Up @@ -153,10 +153,7 @@ impl Backend for QemuBackend {
/// in a generic [`Instance`] whose wire client is a
/// [`client::QemuInstanceClient`]. We seed
/// [`Instance::pid`] with libvirt's per-VM pidfile so the
/// engine's cgroup / procfs sampling kicks in immediately;
/// the QEMU backend still supplies its own per-thread util on
/// every snapshot so the /proc walker's thread-name filter is
/// never consulted for this transport.
/// controller's cgroup and procfs sampling kicks in immediately.
async fn discover(&self) -> Vec<Arc<Instance>> {
let conn = match self.get_or_open_conn(&self.cfg.libvirt_uri).await {
Some(c) => c,
Expand Down Expand Up @@ -208,12 +205,12 @@ impl Backend for QemuBackend {
// it is purely informational (log context), so we
// stuff the libvirt URI in there.
let sock_path = Path::new(&self.cfg.libvirt_uri);
out.push(Arc::new(Instance::new(
uuid.to_string(),
sock_path,
pid,
client,
)));
let filter = ThreadNameFilter::new("^IO iothread", &[])
.expect("static QEMU IOThread regex must compile");
out.push(Arc::new(
Instance::new(uuid.to_string(), sock_path, pid, client)
.with_thread_name_filter(filter),
));
}
out
}
Expand Down
47 changes: 47 additions & 0 deletions src/bin/mockfsd.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,42 @@ impl Inner {
self.callbacks.insert(parent, callback);
}

/// Attempt to unregister a file, returns true if successful.
fn unregister(&mut self, name: &Path) -> bool {
assert!(name.is_absolute());
let mut parent = INodeNo::ROOT;
let mut target = None;
for part in name.components() {
let Component::Normal(part) = part else {
continue;
};
let Some(inode) = self.children.get(&(parent, part.to_owned())).copied() else {
return false;
};
target = Some((parent, part.to_owned(), inode));
parent = inode;
}
let Some((target_parent, target_name, target_inode)) = target else {
return false;
};
let mut removed = vec![target_inode];
let mut index = 0;
while index < removed.len() {
let inode = removed[index];
removed.extend(
self.children
.iter()
.filter_map(|((parent, _), child)| (*parent == inode).then_some(*child)),
);
index += 1;
}
self.children
.retain(|(parent, _), child| !removed.contains(parent) && !removed.contains(child));
self.children.remove(&(target_parent, target_name));
self.callbacks.retain(|inode, _| !removed.contains(inode));
true
}

/// Get child [`INodeNo`] given its name and parents [`INodeNo`].
fn lookup(&self, parent: INodeNo, name: &OsStr) -> Option<INodeNo> {
self.children.get(&(parent, name.to_owned())).copied()
Expand Down Expand Up @@ -312,6 +348,11 @@ impl MockFsHandle {
self.register_iterator(name, values);
Ok(())
}

fn unregister(&self, name: &Path) -> Result<bool, RegisterError> {
validate_registration_path(name)?;
Ok(self.inner.lock().expect("mockfs lock").unregister(name))
}
}

/// Ensure a given [`Path`] is well-formed:
Expand Down Expand Up @@ -648,6 +689,12 @@ impl Control {
.map_err(invalid_args)
}

async fn unregister(&self, mount_path: String, path: String) -> zbus::fdo::Result<bool> {
self.mounted_handle(&mount_path)?
.unregister(Path::new(&path))
.map_err(invalid_args)
}

// TODO register callback, how to implement?

async fn ping(&self) -> String {
Expand Down
Loading
Loading