[Linux] Support wakeup watermark for perf ring buffers

[lalitb]

On Linux, one-collect currently polls the perf ring buffer on a fixed sleep interval when no data is available. The perf subsystem supports event-driven wakeup via a configurable byte watermark - the kernel signals the perf fd only when enough data has accumulated, eliminating unnecessary polling.

Supporting this would allow callers to tune the latency/throughput tradeoff:

• Low watermark -> wake on every event, minimum latency
• High watermark -> amortize wakeup cost, better throughput at scale

[beaubelgrave]

Lets do this as an option during the builder pattern(s) we have.

[lalitb]

Sounds good. I’ll wire it up as an opt-in builder option, leaving existing polling behavior unchanged by default.

[utpilla]

There are two levels of polling to address here.

Level 1: Kernel level

Wire up wakeup_watermark so the kernel signals the perf fd when data arrives. This is a prerequisite for level 2.

Level 2: Consumer-level readiness

Even with watermarks wired up, consumers have no way to receive the kernel's readiness signal. one_collect's API is pull-based. You call parse_for_duration, it reads whatever is in the ring and returns. There's no "notify me when data is available" mechanism.

For consumers that run on async runtimes (e.g., a single-threaded tokio current_thread runtime) neither blocking nor polling is viable. They need a fd they can register with their own event loop.

Proposed API: expose per-CPU perf fds so consumers can decide how to compose them, single-core consumers use one fd directly, multi-core consumers create their own epoll:

Option A lookup by CPU:

impl PerfSession {
    pub fn perf_fd_for_cpu(&self, cpu: u32) -> Option<BorrowedFd<'_>>;
}

Option B iterate all:

impl PerfSession {
    pub fn perf_fds(&self) -> impl Iterator<Item = (u32, BorrowedFd<'_>)>; // could return a slice as well?
}

Option C both, for convenience.

In all cases, BorrowedFd<'_> ties the fd lifetime to the session, preventing use-after-close. one_collect retains ownership and stays runtime-agnostic (no dependency on any particular async runtime implementation). Consumers build their own async wrappers (e.g., AsyncFd on top.

Both levels are needed together: watermark makes the fd meaningful as a readiness signal; the exposed fd lets consumers receive that signal. Without watermark, poll()/epoll() on the fd would never wake. Without the fd, the watermark signal stays trapped inside one_collect.

[beaubelgrave]

I like having option B, so callers can get all CPUs based on the builder that created the session (some scenarios we mask out certain CPUs). Having a caller do a loop with incrementing CPUs (IE: Option A) might bail out on the first None returned or something, even though there are more CPUs.

I do wonder though: How were you thinking the BorrowedFd relates to a PerfSession for per-CPU scenarios? Would they call parse_all / friends? They'd want the PerfSession to process normally, but only for that CPU. Right now, if you asyncly invoked parse_all on the PerfSession, it would process all ring buffers within a source. You could have a PerfSession per-CPU to get around this, with some overhead involved.

[utpilla]

In our case, each pipeline thread creates its own PerfSession targeting a single CPU (cpu_ids: vec![core_id]), so parse_all naturally scopes to that one ring. The fd maps 1:1 to the session, so no per-CPU parsing is needed on our side.

But for the general case (multi-CPU session where a specific CPU's fd fires), there are a couple of new API directions worth considering:

Option 1: parse_for_cpu(cpu: u32) (push model)

A new method on PerfSession, same as parse_all but scoped to one CPU's ring. one_collect still drives the loop and fires registered callbacks. This is the smallest change since existing event registration, callbacks, and field refs all work as-is. The main internal cost is that begin_reading / read / end_reading and the oldest_cpu time-ordering logic currently assume all CPUs; scoping to one would need some refactoring there.

Consumer usage with AsyncFd:

let (cpu, fd) = session.perf_fds().next().unwrap();
let async_fd = AsyncFd::new(fd)?;

loop {
    let guard = async_fd.readable().await?;
    session.parse_for_cpu(cpu);  // fires registered callbacks for this CPU only
    guard.clear_ready();
}

Option 2: drain_cpu(cpu: u32) -> impl Iterator<Item = PerfData> (pull model)

A new method on PerfSession where instead of one_collect driving the loop and firing callbacks, the consumer pulls parsed events directly from a specific CPU's ring. one_collect still handles perf record parsing (header decoding, field population, LOST records etc.), but the consumer controls the iteration.

Consumer usage with AsyncFd:

let (cpu, fd) = session.perf_fds().next().unwrap();
let async_fd = AsyncFd::new(fd)?;

loop {
    let guard = async_fd.readable().await?;
    for event in session.drain_cpu(cpu) {
        // consumer processes PerfData directly
    }
    guard.clear_ready();
}

The tradeoff between them:

	parse_for_cpu (push)	drain_cpu iterator (pull)
Consumer complexity	Low, same callback model as today	Medium, consumer drives the loop
Backpressure control	Via parse_until's existing should_stop callback	Consumer decides how many events to pull per turn
API change in one_collect	Medium, scope existing parse loop to one CPU	Medium, return parsed events instead of dispatching them

For this one, I think Option 1 would be better since it's closer to one_collect's existing callback model and doesn't require consumers to handle PerfData directly.

[beaubelgrave]

I prefer option 1 as well, I do not have a problem if you want to start staging these changes. I'd be fine with it just working with parse_all first (to unblock you all) and then later coming back and doing the per-CPU pattern discussed here.

[beaubelgrave]

@utpilla can we close this out?

[utpilla]

@beaubelgrave Still need to implement Option 1 (parse_for_cpu(cpu: u32) from #254 (comment)