diff --git a/crates/memtrack/src/ebpf/c/rss.bpf.h b/crates/memtrack/src/ebpf/c/rss.bpf.h index 06e5b7e2d..0220b8ebd 100644 --- a/crates/memtrack/src/ebpf/c/rss.bpf.h +++ b/crates/memtrack/src/ebpf/c/rss.bpf.h @@ -5,14 +5,18 @@ #include "utils/event_helpers.h" #include "utils/process_tracking.h" -/* (rss_stat mm_id << 32 | member) -> {owning tgid, last in-context size}. Keyed per - * counter so an external (curr==0) update is attributed only once that mm/member was - * established in-context. An external event may only lower a counter: any size above - * the last in-context value is dropped, so neither a stale/racing reclaim read nor an - * mm_id hash collision with another task can invent a peak. LRU eviction + re-seeding - * on the owner's next in-context event covers hash reuse, so no teardown hook needed. */ +/* (rss_stat mm_id << 32 | member) -> {owning tgid, that tgid's mm at seeding time, + * last in-context size}. Keyed per counter so an external (curr==0) update is + * attributed only once that mm/member was established in-context. An external event + * may only lower a counter: any size above the last in-context value is dropped, so + * neither a stale/racing reclaim read nor an mm_id hash collision with another task + * can invent a peak. mm_id is only a hash, so the pointer is stored alongside and + * revalidated against pid_mm on the external path: an entry whose mm the owner no + * longer holds describes a freed mm_struct whose slab slot (and therefore hash) has + * been recycled by an unrelated address space. */ struct rss_owner { __u32 pid; + __u64 mm; __u64 size; }; struct { @@ -20,7 +24,7 @@ struct { __uint(max_entries, 40960); __type(key, __u64); __type(value, struct rss_owner); -} mm_to_pid SEC(".maps"); +} rss_counter_owner SEC(".maps"); /* Foreign-actor rmap attribution: rmap events run by a task other than the mm's * owner (kswapd reclaim, another process's process_madvise, khugepaged, KSM, @@ -28,16 +32,13 @@ struct { * pointer. pid_mm is the inverse, letting the exec and exit hooks remove an entry * by value. * - * Lifecycle invariant: every mm_owner entry is removed when its process execs - * (the old mm is freed mid-life) or when its thread group dies, whichever comes - * first; LRU eviction is only a backstop. A stale entry surviving mm-pointer - * reuse would misattribute another process's events, so ownership is only ever - * registered from an in-context (task->mm == mm) event. + * Lifecycle invariant: mm_owner follows pid_mm. Registration does not steal a + * live CLONE_VM sibling's ownership. Moving to another mm or reaching group + * death removes the corresponding entry. Foreign attribution validates both + * maps so stale mm pointers fail closed. * - * pid_mm is a plain hash on purpose: an LRU inverse could be evicted while its - * forward twin stays lookup-hot, leaving exec/exit unable to remove the live - * mm_owner entry. Like tracked_pids, its entries are bound to the process - * lifecycle and removed at group death. */ + * pid_mm must not use LRU eviction: losing the inverse binding would prevent + * exec and exit from removing the forward entry. */ struct { __uint(type, BPF_MAP_TYPE_LRU_HASH); __uint(max_entries, 10240); @@ -46,6 +47,48 @@ struct { } mm_owner SEC(".maps"); BPF_HASH_MAP(pid_mm, __u32, __u64, 10240); +/* Delete mm_owner[mm] only while it still names pid: the slot may have been + * re-registered by another process since. */ +static __always_inline void drop_mm_owner_entry(__u64 mm, __u32 pid) { + __u32* owner = bpf_map_lookup_elem(&mm_owner, &mm); + if (owner && *owner == pid) { + bpf_map_delete_elem(&mm_owner, &mm); + } +} + +/* Register pid as mm's owner without stealing from a live one: CLONE_VM + * siblings (vfork/posix_spawn) share the mm, and overwriting would let the + * child's exec-time cleanup delete the entry out from under the still-live + * parent. Only a vacant slot or one whose owner no longer holds the mm (a + * recycled mm_struct) is claimed. */ +static __always_inline void mm_owner_take(__u64 mm, __u32 pid) { + __u32* reg = bpf_map_lookup_elem(&mm_owner, &mm); + if (!reg) { + bpf_map_update_elem(&mm_owner, &mm, &pid, BPF_ANY); + return; + } + if (*reg == pid) { + return; + } + __u64* reg_mm = bpf_map_lookup_elem(&pid_mm, reg); + if (!reg_mm || *reg_mm != mm) { + bpf_map_update_elem(&mm_owner, &mm, &pid, BPF_ANY); + } +} + +/* Remove the previous forward binding when pid_mm changes. Exec-time faults + * may expose the new mm before the exec tracepoint runs. */ +static __always_inline void refresh_pid_mm(__u32 pid, __u64 mm) { + __u64* cur_mm = bpf_map_lookup_elem(&pid_mm, &pid); + if (cur_mm && *cur_mm == mm) { + return; + } + if (cur_mm) { + drop_mm_owner_entry(*cur_mm, pid); + } + bpf_map_update_elem(&pid_mm, &pid, &mm, BPF_ANY); +} + #define FOLIO_MAPPING_ANON 0x1UL const volatile __u32 page_shift = 12; @@ -73,10 +116,17 @@ int tracepoint_rss_stat(struct trace_event_raw_rss_stat* ctx) { return 0; } owner = cur; - struct rss_owner state = {.pid = cur, .size = size}; - bpf_map_update_elem(&mm_to_pid, &key, &state, BPF_ANY); + /* curr means current->mm is the mm the counter belongs to. Recording it + * (and keeping pid_mm current, which the rmap hooks may never do when + * only rss_stat is attached) is what lets the external path below tell a + * live mm from a recycled slab slot. */ + struct task_struct* task = bpf_get_current_task_btf(); + __u64 mm = (__u64)BPF_CORE_READ(task, mm); + struct rss_owner state = {.pid = cur, .mm = mm, .size = size}; + bpf_map_update_elem(&rss_counter_owner, &key, &state, BPF_ANY); + refresh_pid_mm(cur, mm); } else { - struct rss_owner* found = bpf_map_lookup_elem(&mm_to_pid, &key); + struct rss_owner* found = bpf_map_lookup_elem(&rss_counter_owner, &key); if (!found) { return 0; } @@ -87,6 +137,14 @@ int tracepoint_rss_stat(struct trace_event_raw_rss_stat* ctx) { if (cur == owner) { return 0; } + /* mm_id is a hash of a pointer the tracepoint never exposes, so the key + * survives the mm it was seeded from. Once the owner no longer holds that + * mm (it execed, or the entry predates a pid reuse), the hash now belongs + * to a recycled mm_struct and its counters describe another address space. */ + __u64* owner_mm = bpf_map_lookup_elem(&pid_mm, &owner); + if (!owner_mm || *owner_mm != found->mm) { + return 0; + } /* An external actor may only lower a counter. A larger value is a stale * reclaim read or an mm_id hash collision with another task; dropping it * keeps the reconstructed peak identical to the in-context timeline. */ @@ -176,19 +234,8 @@ static __always_inline int submit_rmap(struct vm_area_struct* vma, __s32 member, return 0; } - /* Register ownership so foreign actors can later attribute to this pid. - * Both maps are validated (not just written) on every in-context event: - * the lookups keep the hot path cheap AND keep both entries LRU-fresh, - * since pid_mm is otherwise never read until exec/exit and could be - * evicted independently of its still-hot mm_owner twin. */ - __u32* reg = bpf_map_lookup_elem(&mm_owner, &mm); - if (!reg || *reg != pid) { - bpf_map_update_elem(&mm_owner, &mm, &pid, BPF_ANY); - } - __u64* cur_mm = bpf_map_lookup_elem(&pid_mm, &pid); - if (!cur_mm || *cur_mm != mm) { - bpf_map_update_elem(&pid_mm, &pid, &mm, BPF_ANY); - } + mm_owner_take(mm, pid); + refresh_pid_mm(pid, mm); owner = pid; } else { /* Foreign actor (task->mm != mm, including kthreads whose task->mm is NULL): @@ -202,6 +249,12 @@ static __always_inline int submit_rmap(struct vm_area_struct* vma, __s32 member, if (!is_tracked(owner)) { return 0; } + /* An mm_struct address may be reused while a stale owner entry remains. + * Accept only the current inverse binding. */ + __u64* owner_mm = bpf_map_lookup_elem(&pid_mm, &owner); + if (!owner_mm || *owner_mm != mm) { + return 0; + } } /* header.tid is stamped from the current task; for a foreign actor it @@ -307,17 +360,12 @@ int tracepoint_task_newtask(struct trace_event_raw_task_newtask* ctx) { SUBMIT_EVENT_AS(child_pid, EVENT_TYPE_FORK, { e->data.fork.parent_pid = parent_pid; }); } -/* Remove pid's ownership registration. The mm_owner value is verified against - * pid before deleting: a stale pid_mm entry (LRU eviction skew) could otherwise - * point at an mm since re-registered by another process, and deleting that - * would silence a live owner's foreign attribution. */ +/* Verify the forward binding before deletion because the mm may have been + * re-registered by another process. */ static __always_inline void drop_mm_ownership(__u32 pid) { __u64* mm = bpf_map_lookup_elem(&pid_mm, &pid); if (mm) { - __u32* owner = bpf_map_lookup_elem(&mm_owner, mm); - if (owner && *owner == pid) { - bpf_map_delete_elem(&mm_owner, mm); - } + drop_mm_owner_entry(*mm, pid); } bpf_map_delete_elem(&pid_mm, &pid); } @@ -329,15 +377,13 @@ int tracepoint_sched_process_exec(void* ctx) { return 0; } - /* Maintain ownership before submitting (SUBMIT_EVENT_AS returns from the - * function). Exec frees the old mm long before group death, so the stale - * pointer must be dropped here or a reused mm_struct would be misattributed. */ + /* SUBMIT_EVENT_AS returns, so ownership cleanup must precede it. */ drop_mm_ownership(pid); struct task_struct* task = bpf_get_current_task_btf(); __u64 new_mm = (__u64)BPF_CORE_READ(task, mm); if (new_mm) { - bpf_map_update_elem(&mm_owner, &new_mm, &pid, BPF_ANY); + mm_owner_take(new_mm, pid); bpf_map_update_elem(&pid_mm, &pid, &new_mm, BPF_ANY); } diff --git a/crates/memtrack/src/ebpf/memtrack/maps.rs b/crates/memtrack/src/ebpf/memtrack/maps.rs index 42d2dd2e7..9d936333f 100644 --- a/crates/memtrack/src/ebpf/memtrack/maps.rs +++ b/crates/memtrack/src/ebpf/memtrack/maps.rs @@ -67,6 +67,56 @@ impl MemtrackBpf { "dropped_events", ) } + + /// Live `mm_owner` entries as `(mm_struct pointer, owning pid)`. + pub fn mm_owner_entries(&self) -> Result> { + let entries = + snapshot_entries::<8, 4>(with_skel!(self, skel => &skel.maps.mm_owner), "mm_owner")?; + Ok(entries + .into_iter() + .map(|(mm, pid)| (u64::from_le_bytes(mm), u32::from_le_bytes(pid))) + .collect()) + } + + /// Live `pid_mm` entries as `(pid, mm_struct pointer)`, the inverse binding + /// `mm_owner` is validated against. + pub fn pid_mm_entries(&self) -> Result> { + let entries = + snapshot_entries::<4, 8>(with_skel!(self, skel => &skel.maps.pid_mm), "pid_mm")?; + Ok(entries + .into_iter() + .map(|(pid, mm)| (u32::from_le_bytes(pid), u64::from_le_bytes(mm))) + .collect()) + } +} + +/// Snapshot a hash map's live entries as fixed-size key/value byte pairs. +/// +/// Iteration is `BPF_MAP_GET_NEXT_KEY` followed by a separate lookup, so it is +/// not atomic: a key deleted in between is skipped rather than reported. +fn snapshot_entries( + map: &impl MapCore, + name: &str, +) -> Result> { + let mut entries = Vec::new(); + for key in map.keys() { + let key: [u8; K] = key + .as_slice() + .try_into() + .map_err(|_| anyhow!("{name} key has unexpected size"))?; + let Some(value) = map + .lookup(&key, libbpf_rs::MapFlags::ANY) + .with_context(|| format!("Failed to read {name} entry"))? + else { + continue; + }; + let value: [u8; V] = value + .as_slice() + .try_into() + .map_err(|_| anyhow!("{name} value has unexpected size"))?; + entries.push((key, value)); + } + Ok(entries) } /// Read slot 0 of a single-entry `__u64` array map. diff --git a/crates/memtrack/src/ebpf/tracker.rs b/crates/memtrack/src/ebpf/tracker.rs index b43766260..341875e04 100644 --- a/crates/memtrack/src/ebpf/tracker.rs +++ b/crates/memtrack/src/ebpf/tracker.rs @@ -113,6 +113,18 @@ impl Tracker { self.bpf.lock().dropped_events_count() } + /// Live `mm_owner` entries (`mm_struct` pointer -> owning pid). Only + /// meaningful while the BPF object is alive; teardown frees the map. + pub fn mm_owner_entries(&self) -> Result> { + self.bpf.lock().mm_owner_entries() + } + + /// Live `pid_mm` entries (pid -> `mm_struct` pointer). Same lifetime caveat + /// as [`Tracker::mm_owner_entries`]. + pub fn pid_mm_entries(&self) -> Result> { + self.bpf.lock().pid_mm_entries() + } + /// Stop the attach worker, if any, and surface any fatal error it recorded, /// including missed exec mappings (incomplete allocator coverage). A tracker /// without an allocator watcher has no worker, so this is a no-op. diff --git a/crates/memtrack/testdata/rss/exec_churn.c b/crates/memtrack/testdata/rss/exec_churn.c new file mode 100644 index 000000000..558610a45 --- /dev/null +++ b/crates/memtrack/testdata/rss/exec_churn.c @@ -0,0 +1,56 @@ +/* Repeatedly populate and replace child address spaces. + * + * parent ── fork ──▶ child faults REGION_MIB of anonymous memory + * │ │ execv(trivial image) replaces the populated mm + * │ ▼ + * │ exits immediately + * └── waitpid + * + * After the final child exits, no ownership-map entry may name these processes. + */ +#define _GNU_SOURCE +#include +#include +#include +#include + +#define ITERATIONS 8 +#define REGION_MIB 4 +#define EXIT_MARKER "exec-churn-exit" + +/* Exec into the cheapest image available. Only dropping a populated mm matters, + * so where the pid lands is irrelevant: distributions without /bin/true (NixOS) + * fall back to re-execing this fixture with a marker that returns at once. */ +static void exec_trivial(const char* self) { + static const char* const candidates[] = {"/bin/true", "/usr/bin/true"}; + for (unsigned i = 0; i < sizeof(candidates) / sizeof(*candidates); i++) { + if (access(candidates[i], X_OK) == 0) { + char* args[] = {(char*)candidates[i], NULL}; + execv(candidates[i], args); + } + } + char* args[] = {(char*)self, (char*)EXIT_MARKER, NULL}; + execv(self, args); +} + +int main(int argc, char** argv) { + if (argc > 1 && strcmp(argv[1], EXIT_MARKER) == 0) return 0; + + size_t len = (size_t)REGION_MIB * 1024 * 1024; + for (int i = 0; i < ITERATIONS; i++) { + pid_t pid = fork(); + if (pid < 0) return 1; + if (pid == 0) { + void* region = + mmap(NULL, len, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (region == MAP_FAILED) _exit(1); + memset(region, 0x42, len); + exec_trivial(argv[0]); + _exit(1); + } + int status; + if (waitpid(pid, &status, 0) < 0) return 1; + if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) return 1; + } + return 0; +} diff --git a/crates/memtrack/testdata/rss/stale_mm_owner.c b/crates/memtrack/testdata/rss/stale_mm_owner.c new file mode 100644 index 000000000..49dcda5f0 --- /dev/null +++ b/crates/memtrack/testdata/rss/stale_mm_owner.c @@ -0,0 +1,104 @@ +/* Recycled mm_struct, stale rss_stat owner. + * + * parent ── fork ──▶ child faults one page, so the kernel emits an + * in-context rss_stat for mm0 and mm0's hash is + * registered as owned by that pid + * │ execv(self, "big") → mm0 is freed, the pid lives on + * ▼ + * big: REGION_MIB of anon RSS on mm1, then idles + * parent ── fork burst ──▶ children that exit immediately. Each fork + * allocates an mm_struct from the same slab and populates + * it from the PARENT's context, then tears it down from + * the child's; both are out-of-context updates. mm0's slot + * is the head of the per-cpu freelist, so the whole burst + * lands on it and every update hashes to mm0's id. + * + * Everything is pinned to one CPU to keep the freelist LIFO order. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include + +#include "rss_report.h" + +#define REGION_MIB 256 +#define CHURN_FORKS 256 +#define READY_FD 3 + +/* Pins to the lowest CPU the current mask allows, so the mm_struct freed at + * exec stays at the head of that CPU's slab freelist. Failing here would make + * the fixture stop exercising the reuse path, so it is fatal. */ +static int pin_one_cpu(void) { + cpu_set_t allowed; + CPU_ZERO(&allowed); + if (sched_getaffinity(0, sizeof(allowed), &allowed) != 0) return 1; + for (int cpu = 0; cpu < CPU_SETSIZE; cpu++) { + if (!CPU_ISSET(cpu, &allowed)) continue; + cpu_set_t one; + CPU_ZERO(&one); + CPU_SET(cpu, &one); + return sched_setaffinity(0, sizeof(one), &one) != 0; + } + return 1; +} + +static int run_big(const char* report_path) { + if (pin_one_cpu() != 0) return 1; + size_t len = (size_t)REGION_MIB * 1024 * 1024; + void* region = mmap(NULL, len, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (region == MAP_FAILED) return 1; + memset(region, 0x42, len); + if (write_rss_report(report_path) != 0) return 1; + char ready = 'R'; + if (write(READY_FD, &ready, 1) != 1) return 1; + /* Idle while the parent churns: no further fault means no in-context + * rss_stat event repairs a bogus sample. */ + pause(); + return 0; +} + +int main(int argc, char** argv) { + if (argc < 2) return 1; + if (argc >= 3 && strcmp(argv[2], "big") == 0) return run_big(argv[1]); + + if (pin_one_cpu() != 0) return 1; + int ready[2]; + if (pipe(ready) != 0) return 1; + + pid_t big = fork(); + if (big < 0) return 1; + if (big == 0) { + if (pin_one_cpu() != 0) _exit(1); + close(ready[0]); + if (ready[1] != READY_FD) { + dup2(ready[1], READY_FD); + close(ready[1]); + } + void* page = mmap(NULL, 4096, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (page == MAP_FAILED) _exit(1); + memset(page, 1, 4096); + char* args[] = {argv[0], argv[1], "big", NULL}; + execv("/proc/self/exe", args); + _exit(1); + } + close(ready[1]); + char got = 0; + if (read(ready[0], &got, 1) != 1 || got != 'R') return 1; + + for (int i = 0; i < CHURN_FORKS; i++) { + pid_t churn = fork(); + if (churn < 0) return 1; + if (churn == 0) _exit(0); + int churn_status; + if (waitpid(churn, &churn_status, 0) < 0) return 1; + } + + if (kill(big, SIGKILL) != 0) return 1; + int status; + return waitpid(big, &status, 0) < 0; +} diff --git a/crates/memtrack/testdata/rss/vfork_spawn.c b/crates/memtrack/testdata/rss/vfork_spawn.c new file mode 100644 index 000000000..05a32c7fb --- /dev/null +++ b/crates/memtrack/testdata/rss/vfork_spawn.c @@ -0,0 +1,101 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* Exercise ownership of an mm shared through CLONE_VM. + * + * The parent faults a private file mapping to register ownership. A CLONE_VM + * child then faults anonymous pages into the shared mm and execs. The fixture + * pauses before later parent faults can refresh the binding, then reclaims the + * parent's file pages from another process. Ownership must remain with the + * parent throughout. + * + * argv: [1] = checkpoint-ready file to create, [2] = release file to wait for. */ + +#define REGION (64UL * 1024 * 1024) +#define SCRATCH (1UL * 1024 * 1024) +#define CHILD_STACK (256UL * 1024) + +static char* scratch; + +/* Fault fresh anonymous pages into the shared address space from the child. */ +static int clone_vm_child(void* arg) { + (void)arg; + memset(scratch, 0x42, SCRATCH); + /* POSIX guarantees /bin/sh; only the exec transition matters. */ + execl("/bin/sh", "sh", "-c", "exit 0", (char*)NULL); + _exit(127); +} + +int main(int argc, char** argv) { + if (argc < 3) return 2; + sleep(1); /* let the tracker attach + enable + add root pid */ + + char path[] = "/tmp/memtrack_vfork_XXXXXX"; + int fd = mkstemp(path); + if (fd < 0) return 1; + unlink(path); + if (ftruncate(fd, REGION) != 0) return 1; + + void* mem = mmap(NULL, REGION, PROT_READ, MAP_PRIVATE, fd, 0); + if (mem == MAP_FAILED) return 1; + + /* Seed ownership from the parent's context. */ + volatile char sink = 0; + for (size_t i = 0; i < REGION; i += 4096) sink ^= ((volatile char*)mem)[i]; + (void)sink; + + /* Leave the mapping untouched so only the CLONE_VM child faults its pages. */ + scratch = mmap(NULL, SCRATCH, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (scratch == MAP_FAILED) return 1; + + char* stack = mmap(NULL, CHILD_STACK, PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS | MAP_STACK, -1, 0); + if (stack == MAP_FAILED) return 1; + + /* glibc's posix_spawn uses CLONE_VM|CLONE_VFORK. CLONE_VFORK orders the + * child's exec before the later foreign reclaim. */ + pid_t c = clone(clone_vm_child, stack + CHILD_STACK, CLONE_VM | CLONE_VFORK | SIGCHLD, NULL); + if (c < 0) return 1; + int status; + if (waitpid(c, &status, 0) < 0) return 1; + /* A failed exec would skip the ownership transition under test. */ + if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) return 1; + + /* Pause before a parent-side fault can refresh the ownership binding. Use + * only already-faulted memory until the test releases the fixture. */ + int rfd = creat(argv[1], 0644); + if (rfd < 0) return 1; + close(rfd); + /* Bounded: a test that dies before writing the release file must not leave + * an orphan spinning here while holding the 64 MiB mapping. */ + for (int i = 0; access(argv[2], F_OK) != 0; i++) { + if (i > 1500) return 3; + struct timespec ts = {.tv_sec = 0, .tv_nsec = 20 * 1000 * 1000}; + nanosleep(&ts, NULL); + } + + pid_t b = fork(); + if (b < 0) return 1; + if (b == 0) { + /* Reclaim the parent's file region from a foreign process context. */ + int pidfd = syscall(SYS_pidfd_open, getppid(), 0); + if (pidfd < 0) _exit(1); + struct iovec iov = {.iov_base = mem, .iov_len = REGION}; + syscall(SYS_process_madvise, pidfd, &iov, 1UL, MADV_PAGEOUT, 0UL); + _exit(0); + } + if (waitpid(b, &status, 0) < 0) return 1; + sleep(1); /* let the external decrements flush to the ring buffer */ + /* No munmap: an in-context decrement would mask the external-path signal. */ + return 0; +} diff --git a/crates/memtrack/tests/bzip2_tests.rs b/crates/memtrack/tests/bzip2_tests.rs new file mode 100644 index 000000000..a98f2899a --- /dev/null +++ b/crates/memtrack/tests/bzip2_tests.rs @@ -0,0 +1,247 @@ +mod shared; + +use anyhow::Result; +use itertools::Itertools; +use runner_shared::artifacts::{MemtrackEvent, MemtrackEventKind}; +use std::collections::BTreeMap; +use std::fs::File; +use std::io::{BufWriter, Write}; +use std::path::{Path, PathBuf}; +use std::process::Command; +use tempfile::TempDir; + +const MIB: u64 = 1024 * 1024; + +/// Big enough that `bzip2 -9` cycles many 900k blocks and `xz -9` fills a real +/// dictionary, small enough to generate in well under a second. +const CORPUS_BYTES: u64 = 16 * MIB; + +/// Ignore small pids where page-level noise dominates a percentage comparison. +const AGREEMENT_FLOOR: u64 = 2 * MIB; + +/// Allow sampling skew while still rejecting missing multi-MiB ancestor state. +const AGREEMENT_TOLERANCE: f64 = 0.25; + +/// Allow short exec teardown transients; larger negative totals indicate +/// foreign teardown misattribution. +const MIN_CUMULATIVE_PAGES: i64 = -1024; + +fn page_size() -> u64 { + let size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; + assert!(size > 0, "sysconf(_SC_PAGESIZE) failed"); + size as u64 +} + +/// Per-pid resident state in bytes for file, anonymous, swap, and shmem pages. +#[derive(Default)] +struct Accum { + latest: [i64; 4], + max_rss: i64, + /// Running minima remain unclamped to detect negative rmap drift. + min: [i64; 4], +} + +impl Accum { + fn set(&mut self, index: usize, bytes: i64) { + self.latest[index] = bytes; + self.update(); + } + + fn add(&mut self, index: usize, delta_bytes: i64) { + self.latest[index] += delta_bytes; + self.update(); + } + + fn seed(&mut self, latest: [i64; 4]) { + self.latest = latest; + self.update(); + } + + /// Reset on exec or exit without discarding observed extrema. + fn reset(&mut self) { + self.latest = [0; 4]; + } + + fn update(&mut self) { + for (min, latest) in self.min.iter_mut().zip(self.latest) { + *min = (*min).min(latest); + } + self.max_rss = self + .max_rss + .max(self.latest[0] + self.latest[1] + self.latest[3]); + } +} + +/// Replay absolute RSS samples and rmap deltas with parser lifecycle semantics. +fn replay(events: &[MemtrackEvent]) -> (BTreeMap, BTreeMap) { + let mut rss: BTreeMap = BTreeMap::new(); + let mut rmap: BTreeMap = BTreeMap::new(); + + for event in events.iter().sorted_by_key(|event| event.timestamp) { + match event.kind { + MemtrackEventKind::Rss { member, size } => { + let Ok(index @ 0..4) = usize::try_from(member) else { + continue; + }; + rss.entry(event.pid).or_default().set(index, size as i64); + } + MemtrackEventKind::Rmap { member, delta } => { + let Ok(index @ 0..4) = usize::try_from(member) else { + continue; + }; + rmap.entry(event.pid) + .or_default() + .add(index, delta * page_size() as i64); + } + MemtrackEventKind::Fork { parent_pid } => { + let seed = rss.get(&parent_pid).map(|p| p.latest).unwrap_or_default(); + rss.entry(event.pid).or_default().seed(seed); + let seed = rmap.get(&parent_pid).map(|p| p.latest).unwrap_or_default(); + rmap.entry(event.pid).or_default().seed(seed); + } + MemtrackEventKind::Exec | MemtrackEventKind::Exit => { + if let Some(acc) = rss.get_mut(&event.pid) { + acc.reset(); + } + if let Some(acc) = rmap.get_mut(&event.pid) { + acc.reset(); + } + } + _ => {} + } + } + (rss, rmap) +} + +fn find_in_path(binary: &str) -> Option { + let paths = std::env::var_os("PATH")?; + std::env::split_paths(&paths) + .map(|dir| dir.join(binary)) + .find(|candidate| candidate.is_file()) +} + +/// Write a deterministic pseudo-text block repeatedly until the corpus reaches +/// at least `bytes`. +fn write_corpus(path: &Path, bytes: u64) -> Result<()> { + const WORDS: [&str; 16] = [ + "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", "india", + "juliett", "kilo", "lima", "mike", "november", "oscar", "papa", + ]; + const BLOCK_BYTES: usize = 256 * 1024; + + let mut state: u64 = 0x243f_6a88_85a3_08d3; + let mut block = String::with_capacity(BLOCK_BYTES + 128); + while block.len() < BLOCK_BYTES { + for _ in 0..12 { + state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + block.push_str(WORDS[(state >> 33) as usize % WORDS.len()]); + block.push(' '); + } + block.push('\n'); + } + + let mut out = BufWriter::new(File::create(path)?); + let mut written = 0u64; + while written < bytes { + out.write_all(block.as_bytes())?; + written += block.len() as u64; + } + out.flush()?; + Ok(()) +} + +/// Run pressure and measurement phases in one tracked process tree. +#[test_with::env(GITHUB_ACTIONS)] +#[test] +fn test_bzip2_rmap_matches_rss_stat() -> Result<()> { + for tool in ["bash", "xz", "bzip2"] { + if find_in_path(tool).is_none() { + eprintln!("skipping test_bzip2_rmap_matches_rss_stat: `{tool}` is not in PATH"); + return Ok(()); + } + } + + let temp_dir = TempDir::new()?; + write_corpus(&temp_dir.path().join("corpus.tar"), CORPUS_BYTES)?; + + // Preserve ancestor state across both compressor execs. + let mut command = Command::new("bash"); + command + .arg("-c") + .arg("xz -9 -k -f -T1 corpus.tar && bzip2 -9 -k -f corpus.tar") + .current_dir(temp_dir.path()); + + let (events, thread_handle) = shared::track_command_with_rmap(command)?; + thread_handle.join().unwrap(); + + let (rss, rmap) = replay(&events); + + // Confirm both compressors ran and produced a meaningful working set. + let execs = events + .iter() + .filter(|e| matches!(e.kind, MemtrackEventKind::Exec)) + .count(); + assert!( + execs >= 2, + "expected at least the xz and bzip2 execs, saw {execs}" + ); + let largest = rss.values().map(|acc| acc.max_rss.max(0) as u64).max(); + assert!( + largest.is_some_and(|peak| peak >= 6 * MIB), + "no tracked pid reached a compressor-sized working set: largest peak {largest:?} bytes" + ); + + // Significant per-pid peaks must agree between both accounting modes. + let mut compared = 0usize; + let mut disagreeing: Vec = Vec::new(); + for (pid, rss_acc) in &rss { + let rss_peak = rss_acc.max_rss.max(0) as u64; + if rss_peak <= AGREEMENT_FLOOR { + continue; + } + compared += 1; + + let rmap_peak = rmap.get(pid).map_or(0, |acc| acc.max_rss.max(0)) as u64; + let drift = (rmap_peak as f64 - rss_peak as f64) / rss_peak as f64; + if drift.abs() > AGREEMENT_TOLERANCE { + disagreeing.push(format!( + "pid {pid}: rss_stat {rss_peak} vs rmap {rmap_peak} bytes ({:+.1}%)", + drift * 100.0 + )); + } + } + assert!( + compared >= 2, + "expected the shell and at least one compressor above {AGREEMENT_FLOOR} bytes, compared {compared} pids" + ); + assert!( + disagreeing.is_empty(), + "{} pid(s) disagree by more than {:.0}% between rss_stat and rmap: {disagreeing:?}", + disagreeing.len(), + AGREEMENT_TOLERANCE * 100.0 + ); + + // Detect negative rmap drift before parser clamping hides it. + let floor_bytes = MIN_CUMULATIVE_PAGES * page_size() as i64; + let drifted: Vec = rmap + .iter() + .filter(|(_, acc)| acc.min[0].min(acc.min[1]) < floor_bytes) + .map(|(pid, acc)| { + let pages = |bytes: i64| bytes / page_size() as i64; + format!( + "pid {pid}: file {} pages, anon {} pages", + pages(acc.min[0]), + pages(acc.min[1]) + ) + }) + .collect(); + assert!( + drifted.is_empty(), + "{} pid(s) drove cumulative rmap below {MIN_CUMULATIVE_PAGES} pages, so foreign teardown \ + was charged to them: {drifted:?}", + drifted.len() + ); + Ok(()) +} diff --git a/crates/memtrack/tests/rss_tests.rs b/crates/memtrack/tests/rss_tests.rs index 197f70377..57bb00221 100644 --- a/crates/memtrack/tests/rss_tests.rs +++ b/crates/memtrack/tests/rss_tests.rs @@ -5,7 +5,7 @@ use itertools::Itertools; use rstest::rstest; use runner_shared::artifacts::{MemtrackEvent, MemtrackEventKind}; use serde::Serialize; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::process::Command; use tempfile::TempDir; @@ -427,6 +427,82 @@ fn test_rss_external_reclaim(#[case] mode: Reclaim) -> Result<(), Box Result<(), Box> { + let (_report, events) = track_fixture( + include_str!("../testdata/rss/stale_mm_owner.c"), + "stale_mm_owner", + shared::track_command_with_rmap, + )?; + + // The fixture's only fork before the burst is the child that execs into the + // memory-holding image, so it keeps its pid across the exec. + let (_parent, big) = first_fork_pair(&events).ok_or("no fork event")?; + let exec_ts = events + .iter() + .find(|e| e.pid == big && matches!(e.kind, MemtrackEventKind::Exec)) + .map(|e| e.timestamp) + .ok_or("no exec event for the memory-holding child")?; + + // Only the post-exec address space is of interest: the pre-exec image + // faulted a single page, and that mm is the one whose slab slot gets reused. + let anon_after_exec: Vec<_> = events + .iter() + .filter_map(|e| match e.kind { + MemtrackEventKind::Rss { member: 1, size } if e.pid == big && e.timestamp > exec_ts => { + Some((e, size)) + } + _ => None, + }) + .collect(); + + let (peak_ts, peak) = anon_after_exec + .iter() + .max_by_key(|&&(_, size)| size) + .map(|&(e, size)| (e.timestamp, size)) + .ok_or("no anon rss_stat event after the exec")?; + assert!(peak >= 128 * MIB, "peak anon RSS too small: {peak}"); + + // Past the peak the region is held untouched until the fixture is killed, so + // no legitimate absolute sample may fall back near zero. Before it, samples + // are just the fault-in ramp. + let collapsed: Vec<_> = anon_after_exec + .iter() + .filter(|&&(e, size)| e.timestamp > peak_ts && size < peak / 4) + .map(|&(e, size)| (e.tid, size)) + .collect(); + assert!( + collapsed.is_empty(), + "{} absolute anon samples for pid {big} collapsed below {} bytes while it held {peak}: \ + (tid, size) = {:?}", + collapsed.len(), + peak / 4, + &collapsed[..collapsed.len().min(5)] + ); + + // The rmap-reconstructed total confirms the pages stayed resident the whole + // time, so a collapsing absolute sample could only come from a recycled mm. + let rmap_net = events + .iter() + .filter_map(|e| match e.kind { + MemtrackEventKind::Rmap { member: 1, delta } if e.pid == big => Some(delta), + _ => None, + }) + .sum::() + * page_size() as i64; + assert!( + rmap_net >= (peak / 2) as i64, + "fixture never held the region per rmap: net {rmap_net} bytes" + ); + Ok(()) +} + /// A fork issued by a worker thread must still track the child: registration /// keys on the parent's tgid (task_newtask fires in the cloning task, whose /// pid_tgid upper half is the tgid), not on the raw creator tid. @@ -464,3 +540,177 @@ fn test_rss_rmap_thread_fork_tracks_child() -> Result<(), Box Result<(), Box> +{ + const ITERATIONS: usize = 8; + const REGION_MIB: u64 = 4; + + let temp_dir = TempDir::new()?; + let binary = shared::compile_c_source( + include_str!("../testdata/rss/exec_churn.c"), + "exec_churn", + temp_dir.path(), + )?; + + let (events, maps, thread_handle) = + shared::track_command_with_rmap_maps(Command::new(&binary))?; + thread_handle.join().unwrap(); + + // Confirm the fixture populated each discarded address space. + let execs = events + .iter() + .filter(|e| matches!(e.kind, MemtrackEventKind::Exec)) + .count(); + assert!( + execs >= ITERATIONS, + "fixture exec'd {execs} times, expected at least {ITERATIONS}" + ); + let anon_added: i64 = events + .iter() + .filter_map(|e| match e.kind { + MemtrackEventKind::Rmap { member: 1, delta } if delta > 0 => Some(delta), + _ => None, + }) + .sum(); + let expected = ((ITERATIONS as u64) * (REGION_MIB - 1) * MIB) as i64; + assert!( + anon_added * page_size() as i64 >= expected, + "children never populated their mm: {} anon bytes added, expected ~{expected}", + anon_added * page_size() as i64 + ); + + let tracked: BTreeSet = events.iter().map(|e| e.pid).collect(); + let pid_mm_residue: Vec<_> = maps + .pid_mm + .iter() + .filter(|(pid, _)| tracked.contains(&(*pid as i32))) + .collect(); + let mm_owner_residue: Vec<_> = maps + .mm_owner + .iter() + .filter(|(_, owner)| tracked.contains(&(*owner as i32))) + .collect(); + assert!( + pid_mm_residue.is_empty(), + "{} pid_mm entries survived their pid: {pid_mm_residue:?}", + pid_mm_residue.len() + ); + assert!( + mm_owner_residue.is_empty(), + "{} mm_owner entries still name a dead tracked pid: {mm_owner_residue:?}", + mm_owner_residue.len() + ); + Ok(()) +} + +/// A CLONE_VM child shares its parent's mm until exec. Its faults must not +/// transfer ownership because exec cleanup would remove the parent's live +/// binding. +#[test_with::env(GITHUB_ACTIONS)] +#[test] +fn test_rmap_clone_vm_keeps_parent_ownership() -> Result<(), Box> { + let tmp = TempDir::new()?; + let binary = shared::compile_c_source( + include_str!("../testdata/rss/vfork_spawn.c"), + "vfork_spawn", + tmp.path(), + )?; + let ready = tmp.path().join("checkpoint-ready"); + let release = tmp.path().join("checkpoint-release"); + let mut command = Command::new(&binary); + command.arg(&ready).arg(&release); + + let (events, checkpoint, root_pid, thread_handle) = + shared::track_command_with_rmap_checkpoint(command, &ready, &release)?; + thread_handle.join().unwrap(); + + // A stale entry naming the parent is insufficient; check its current mm. + let parent_mm = checkpoint + .pid_mm + .iter() + .find_map(|&(pid, mm)| (pid == root_pid as u32).then_some(mm)) + .expect("parent has no pid_mm binding at the checkpoint"); + assert!( + checkpoint + .mm_owner + .iter() + .any(|&(mm, owner)| mm == parent_mm && owner == root_pid as u32), + "mm_owner lost the parent's current mm after its CLONE_VM child exec'd \ + (parent mm {parent_mm:#x}, snapshot: {:?})", + checkpoint.mm_owner + ); + + let mut forks = events + .iter() + .sorted_by_key(|e| e.timestamp) + .filter_map(|e| match e.kind { + MemtrackEventKind::Fork { parent_pid } => Some((parent_pid, e.pid)), + _ => None, + }); + let (parent, shared_child) = forks.next().ok_or("CLONE_VM child was not tracked")?; + let (reclaimer_parent, reclaimer) = forks.next().ok_or("no fork event for the reclaimer")?; + assert_eq!( + reclaimer_parent, parent, + "the reclaimer was not forked by the fixture" + ); + + // Confirm both processes produced the events needed to exercise ownership. + let owner_adds = events + .iter() + .filter_map(|e| match e.kind { + MemtrackEventKind::Rmap { member: 0, delta } if e.pid == parent && delta > 0 => { + Some(delta) + } + _ => None, + }) + .sum::() as u64 + * page_size(); + assert!( + owner_adds >= 32 * MIB, + "in-context file rmap adds too small: {owner_adds}" + ); + + let shared_mm_adds = events + .iter() + .filter_map(|e| match e.kind { + MemtrackEventKind::Rmap { member: 1, delta } if e.pid == shared_child && delta > 0 => { + Some(delta) + } + _ => None, + }) + .sum::() as u64 + * page_size(); + assert!( + shared_mm_adds >= 256 * 1024, + "CLONE_VM child faulted no anon pages into the shared mm: {shared_mm_adds} bytes" + ); + assert!( + events + .iter() + .any(|e| e.pid == shared_child && matches!(e.kind, MemtrackEventKind::Exec)), + "CLONE_VM child never exec'd, so its ownership cleanup never ran" + ); + + // process_madvise runs in the reclaimer's context, so removals require + // foreign ownership attribution. + let reclaimed = events + .iter() + .filter_map(|e| match e.kind { + MemtrackEventKind::Rmap { member: 0, delta } + if e.pid == parent && e.tid == reclaimer && delta < 0 => + { + Some(-delta) + } + _ => None, + }) + .sum::() as u64 + * page_size(); + assert!( + reclaimed >= 8 * MIB, + "foreign reclaim was not attributed to the owner after its CLONE_VM child exec'd: only {reclaimed} bytes" + ); + Ok(()) +} diff --git a/crates/memtrack/tests/shared.rs b/crates/memtrack/tests/shared.rs index 2d7f360c1..5810e6de1 100644 --- a/crates/memtrack/tests/shared.rs +++ b/crates/memtrack/tests/shared.rs @@ -208,6 +208,74 @@ pub fn track_command_with_opts(command: Command, track_allocators: bool) -> Trac track_command_with_tracker(command, Tracker::new(track_allocators)?) } +/// Snapshot of the live ownership maps. +pub struct OwnershipMaps { + pub mm_owner: Vec<(u64, u32)>, + pub pid_mm: Vec<(u32, u64)>, +} + +/// Track a command with rmap hooks and snapshot its ownership maps after the +/// tracked tree exits but before tracker teardown frees the BPF maps. +pub fn track_command_with_rmap_maps( + command: Command, +) -> anyhow::Result<(Vec, OwnershipMaps, std::thread::JoinHandle<()>)> { + let (tracker, events) = run_tracked(command, Tracker::new_without_allocators_with_rmap(true)?)?; + let maps = OwnershipMaps { + mm_owner: tracker.mm_owner_entries()?, + pid_mm: tracker.pid_mm_entries()?, + }; + Ok((events, maps, std::thread::spawn(move || drop(tracker)))) +} + +/// Track a command with rmap hooks and snapshot the ownership maps at a +/// fixture-signalled checkpoint. +/// +/// The fixture creates `ready` and waits for `release`. The root pid is returned +/// for correlating snapshot entries with events. +pub fn track_command_with_rmap_checkpoint( + command: Command, + ready: &std::path::Path, + release: &std::path::Path, +) -> anyhow::Result<(Vec, OwnershipMaps, i32, std::thread::JoinHandle<()>)> { + use anyhow::ensure; + + let tracker = Tracker::new_without_allocators_with_rmap(true)?; + tracker.enable_tracking()?; + + let mut session = tracker.spawn(&command, None)?; + let rx = session.take_events()?; + let root_pid = session.pid(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + while !ready.exists() { + ensure!( + std::time::Instant::now() < deadline, + "fixture never reached the checkpoint" + ); + std::thread::sleep(std::time::Duration::from_millis(20)); + } + let maps = OwnershipMaps { + mm_owner: tracker.mm_owner_entries()?, + pid_mm: tracker.pid_mm_entries()?, + }; + std::fs::write(release, b"")?; + + session.wait()?; + // Dropping the session does a final ring buffer drain and closes the + // channel, so collecting terminates without a silence timeout. + drop(session); + let events: Vec = rx.iter().collect(); + + tracker.finish()?; + + Ok(( + events, + maps, + root_pid, + std::thread::spawn(move || drop(tracker)), + )) +} + /// How many events of each kind-and-size a run saw. Addresses, timestamps, pids /// and event order all differ legitimately between runs of the same workload, so /// none of them can be compared across variants. @@ -276,6 +344,18 @@ pub fn for_each_variant( } fn track_command_with_tracker(command: Command, tracker: Tracker) -> TrackResult { + let (tracker, events) = run_tracked(command, tracker)?; + + // Detaching the probes blocks on RCU grace periods; let the caller decide + // when to wait for it. + let thread_handle = std::thread::spawn(move || drop(tracker)); + + Ok((events, thread_handle)) +} + +/// Run `command` to completion under `tracker` and drain its events, handing +/// back the still-live tracker so BPF state can be read before teardown. +fn run_tracked(command: Command, tracker: Tracker) -> anyhow::Result<(Tracker, Vec)> { tracker.enable_tracking()?; let mut session = tracker.spawn(&command, None)?; @@ -289,12 +369,8 @@ fn track_command_with_tracker(command: Command, tracker: Tracker) -> TrackResult tracker.finish()?; - // Detaching the probes blocks on RCU grace periods; let the caller decide - // when to wait for it. - let thread_handle = std::thread::spawn(move || drop(tracker)); - info!("Tracked {} events", events.len()); trace!("Events: {events:#?}"); - Ok((events, thread_handle)) + Ok((tracker, events)) }