Struct PollDiagnostics 

pub struct PollDiagnostics { /* private fields */ }

Samples the time every subscription’s handler spends in poll, keeps a moving average and a histogram of it per subscription, and warns once the average says a handler computes on the app’s runtime.

With the poll-diagnostics feature on, every subscription is sampled: one poll in sample_every (64 by default) is timed with two reads of the CPU’s cycle counter. When a subscription’s average crosses the threshold (100 microseconds by default), the runtime logs one warning naming the subscription, its average and p99 poll time, and the fix: threads(n), which moves the handler onto threads of its own. The average is exponential: each sample moves it by a sixteenth of the difference, so a recent sample weighs most and an old one fades without dropping out at once. It warns only after 16 samples.

A handle is cheap to clone, and every clone reads the same reports. Hand one to RustStream::poll_diagnostics to set the threshold or the sampling interval, and keep another to read reports or to export them through the metrics or otel feature.

Examples

use std::time::Duration;

use ruststream::nonzero;
use ruststream::runtime::{App, AppInfo, PollDiagnostics, RustStream};

/// The app gets the settings, and the caller keeps a clone that reads the same reports.
fn app(diagnostics: &PollDiagnostics) -> impl App {
    let diagnostics = diagnostics
        .clone()
        .threshold(Duration::from_micros(250))
        .sample_every(nonzero!(16u32));
    RustStream::new(AppInfo::new("svc", "0.1.0")).poll_diagnostics(diagnostics)
}

/// A periodic task the service runs next to the app: the handlers that belong on `threads(n)`.
fn log_slow(diagnostics: &PollDiagnostics) {
    for report in diagnostics.reports() {
        if report.average() > Duration::from_micros(250) {
            let subscription = report.subscription();
            tracing::warn!(subscription, average = ?report.average(), "slow handler");
        }
    }
}

Implementations

impl PollDiagnostics

pub fn new() -> Self

Diagnostics with the default threshold (100 microseconds) and sampling interval (one poll in 64).

Examples

use ruststream::runtime::{App, AppInfo, PollDiagnostics, RustStream};

/// The app under the defaults (a 100 microsecond threshold, one poll in 64 timed), and
/// the handle the caller keeps to read the reports.
fn app() -> (impl App, PollDiagnostics) {
    let diagnostics = PollDiagnostics::new();
    let app =
        RustStream::new(AppInfo::new("svc", "0.1.0")).poll_diagnostics(diagnostics.clone());
    (app, diagnostics)
}

pub const fn threshold(self, threshold: Duration) -> Self

The average poll time over which a subscription is reported.

Examples

use std::time::Duration;

use ruststream::runtime::{App, AppInfo, PollDiagnostics, RustStream};

fn app() -> impl App {
    // A handler that holds a poll for a millisecond belongs on threads of its own.
    let diagnostics = PollDiagnostics::new().threshold(Duration::from_millis(1));
    RustStream::new(AppInfo::new("svc", "0.1.0")).poll_diagnostics(diagnostics)
}

pub const fn sample_every(self, every: NonZeroU32) -> Self

Times one poll in every. One times every poll.

Examples

use ruststream::nonzero;
use ruststream::runtime::{App, AppInfo, PollDiagnostics, RustStream};

fn app() -> impl App {
    // A hot subscription times one poll in eight for a finer average.
    let diagnostics = PollDiagnostics::new().sample_every(nonzero!(8u32));
    RustStream::new(AppInfo::new("svc", "0.1.0")).poll_diagnostics(diagnostics)
}

pub fn report(&self, subscription: &str) -> Option<PollReport>

What has been measured on subscription so far, or None for a subscription that has not opened.

Examples

use ruststream::runtime::PollDiagnostics;

/// The `/debug/poll` route: the moving average of one subscription, once it was sampled.
fn orders_average(diagnostics: &PollDiagnostics) -> Option<String> {
    let report = diagnostics.report("orders")?;
    Some(format!(
        "{:?} over {} polls",
        report.average(),
        report.samples()
    ))
}

pub fn reports(&self) -> Vec<PollReport>

What has been measured on every subscription that has opened, in the order they opened.

Examples

use ruststream::runtime::PollDiagnostics;

/// Logged at shutdown: every subscription the app sampled, in the order they opened.
fn log_reports(diagnostics: &PollDiagnostics) {
    for report in diagnostics.reports() {
        let subscription = report.subscription();
        tracing::info!(subscription, average = ?report.average(), "poll time");
    }
}

Trait Implementations

impl Clone for PollDiagnostics

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for PollDiagnostics

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for PollDiagnostics

fn default() -> Self

Returns the “default value” for a type.