Diagnostics

The poll-diagnostics feature times the polls of every handler and names the handlers that compute on the app's runtime.

With the feature on, every subscription is sampled. One poll of its handler in 64 is timed with two reads of the CPU's cycle counter. The time measured is the time inside one poll, between two .await points. A handler that waits on I/O or a timer is not slow here; only one that computes is. Each subscription keeps an exponential moving average and a histogram of that time. When the average crosses 100 microseconds, the runtime logs one warning naming the subscription, its average and p99, and the fix: threads. It warns after 16 samples, and once per crossing. A subscription already on threads(n) is measured the same way, and no warning names it. PollDiagnostics changes the threshold and the sample_every interval, and reads the reports. RustStream::poll_diagnostics hands it to the app; a clone the caller keeps reads the same reports. Metrics::observe_poll_diagnostics and Otel::observe_poll_diagnostics export them, labelled per subscription. The cost is a countdown per poll and, on a timed poll, two counter reads and a few atomic updates. The example runs with cargo test --example poll_diagnostics --features testing,macros,memory,json,poll-diagnostics.

Example

Macros

examples/poll_diagnostics.rs

/// The app gets the settings; the caller keeps a clone of `diagnostics` to read the reports.
fn app(diagnostics: &PollDiagnostics) -> impl App {
    let diagnostics = diagnostics
        .clone()
        .threshold(Duration::from_millis(1))
        .sample_every(nonzero!(8u32));
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .poll_diagnostics(diagnostics)
        .with_broker(MemoryBroker::new(), |b| {
            b.include(accept);
            b.include(summarize);
        })
}

Manual

examples/manual/poll_diagnostics.rs

/// The app gets the settings; the caller keeps a clone of `diagnostics` to read the reports.
fn app(diagnostics: &PollDiagnostics) -> impl App {
    let diagnostics = diagnostics
        .clone()
        .threshold(Duration::from_millis(1))
        .sample_every(nonzero!(8u32));
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .poll_diagnostics(diagnostics)
        .with_broker(MemoryBroker::new(), |b| {
            b.include(subscriber("orders", Accept).build());
            b.include(subscriber("reports", Summarize).build());
        })
}

Reading the reports

reports returns a PollReport for every subscription that has opened, and report the one of a given name. A report holds the number of samples, the average and the p99 of the time in one poll.

examples/poll_diagnostics.rs

/// A periodic task beside the app: the handlers whose average poll is over `limit`.
fn slow_handlers(diagnostics: &PollDiagnostics, limit: Duration) -> Vec<PollReport> {
    diagnostics
        .reports()
        .into_iter()
        .filter(|report| report.average() > limit)
        .collect()
}