mirror of
https://github.com/jmcorgan/fips.git
synced 2026-07-30 11:36:15 +00:00
Add a feature-gated profiler for the rx-loop maintenance tick
The tick arm runs twenty-six housekeeping steps in sequence on the one runtime thread, and is polled last, so anything slow in it holds up inbound packets, TUN traffic and control commands behind it. Field evidence says that happens for over a second at a time, but the attribution behind that is two months old and predates the connect-on-send gate, the control read isolation, and the peer lifecycle rework. This measures it rather than continuing to reason about it. Per step it records exact count, max and total into fixed static counters; a dedicated writer thread drains them every ten seconds to a TSV under /var/log/fips, one file per capture, capped at 32 MB. Nothing accumulates: the counters are swapped to zero each interval and the thread holds no history. Arming is `fipsctl profile tick on`, served in the control accept task so the toggle cannot queue behind the very behaviour it measures, and it does not survive a restart. The whole thing is behind a Cargo feature that is off by default, because the risk worth eliminating is the twenty-six edited call sites in the hot loop. With the feature off the macro expands to the bare expression, which makes the default build's neutrality something you read off the generated code rather than something a benchmark fails to disprove. The measurement that matters is how late each tick is against the deadline it was scheduled for, since the arm is polled last and that lateness is the delay. Two earlier designs derived it from the interval between entries and both under-reported: the schedule is fixed, so a steady delay leaves every gap exactly one period and any gap-derived figure reads zero under precisely the sustained overload this is meant to find. The interval hands back its own deadline, so the delay is now a subtraction with no model behind it, and a test drives three late ticks at a constant gap to keep it that way. CI gains a default-features clippy and a feature-on build and test on both runners, closing the gap left by clippy already running with all features.
This commit is contained in:
Johnathan Corgan
@@ -110,6 +110,17 @@ jobs:
|
||||
restore-keys: |
|
||||
${{ runner.os }}-cargo-
|
||||
- run: cargo clippy --all-targets --all-features -- -D warnings
|
||||
# An optional feature means two source trees, and --all-features lints
|
||||
# only one of them. The default build is what ships, so lint it
|
||||
# explicitly: without this stage, code that compiles only with
|
||||
# `profiling` enabled would pass CI while breaking every release build.
|
||||
# Mirrored in testing/ci-local.sh — check-ci-parity.sh compares
|
||||
# integration suites only and will not catch a stage added to one runner
|
||||
# and not the other.
|
||||
- name: Clippy (default features)
|
||||
run: cargo clippy --all-targets -- -D warnings
|
||||
- name: Build with the tick-body profiler enabled
|
||||
run: cargo build --workspace --features profiling
|
||||
|
||||
# ───────────────────────────────────────────────────────────────────────────
|
||||
# Android cross-check
|
||||
@@ -293,6 +304,12 @@ jobs:
|
||||
check_name: Unit Tests Summary
|
||||
fail_on_failure: false
|
||||
|
||||
# The `profiling` feature adds a module, a recorder and a writer thread
|
||||
# that the default-feature run above never compiles, so its own tests do
|
||||
# not execute there. Mirrored in testing/ci-local.sh.
|
||||
- name: Run library tests with the tick-body profiler enabled
|
||||
run: cargo test --lib --features profiling
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Job 2b – Unit tests (macOS)
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -9,6 +9,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
### Added
|
||||
|
||||
- An optional tick-body profiler behind the new `profiling` Cargo feature,
|
||||
**off by default**. When enabled, `fipsctl profile tick on [--dir PATH]` /
|
||||
`off` / `status` starts and stops a capture at runtime with no restart. Each
|
||||
capture writes one tab-separated file (default `/var/log/fips`, capped at
|
||||
32 MB) carrying, per ten-second interval, the exact count, max and total for
|
||||
every step of the rx-loop tick arm, the whole-tick span, and gauges for ticks,
|
||||
peer count, the gap between successive tick-arm entries and the resulting
|
||||
arm-starvation delay. With the feature off the instrumentation macro is a pure
|
||||
pass-through, so a default build contains no timing code on the tick path.
|
||||
`LogsDirectory=fips` was added to the packaged systemd units so the capture
|
||||
directory is created and cleaned up declaratively.
|
||||
|
||||
### Changed
|
||||
|
||||
- The Ethernet transport's per-interface `discovery` flag was renamed to
|
||||
|
||||
@@ -11,6 +11,14 @@ readme = "README.md"
|
||||
keywords = ["mesh", "p2p", "decentralized", "overlay-network", "nostr"]
|
||||
categories = ["network-programming", "command-line-utilities", "cryptography"]
|
||||
|
||||
[features]
|
||||
default = []
|
||||
# Tick-body profiler (`src/instr`). Off by default: enabling it edits 26 call
|
||||
# sites in the rx loop's hot tick arm, and only a compile-time gate makes the
|
||||
# default build's neutrality a property of the generated code rather than of a
|
||||
# runtime check. Build with `--features profiling` for a measurement run.
|
||||
profiling = []
|
||||
|
||||
[dependencies]
|
||||
ratatui = "0.30"
|
||||
secp256k1 = { version = "0.30", features = ["rand", "global-context"] }
|
||||
|
||||
@@ -115,6 +115,58 @@ Tell the daemon to drop a peer link.
|
||||
| -------- | ----------- |
|
||||
| `peer` | npub (bech32) or hostname from `/etc/fips/hosts`. |
|
||||
|
||||
### `profile tick <on|off|status>`
|
||||
|
||||
> **Reading the output.** Step durations are wall clock measured across `await`
|
||||
> points, not CPU time: a step that waits on I/O accrues that wait, and other
|
||||
> tasks may run inside the span. That is the intended measure for head-of-line
|
||||
> delay, and it means a large step is not necessarily an expensive one.
|
||||
> `arm_starvation` is measured directly as the entry time minus the deadline
|
||||
> the interval scheduled that tick for. It is not derived from
|
||||
> `tick_entry_gap`, which carries no starvation signal on its own: under a
|
||||
> steady delay every gap is exactly one tick period.
|
||||
|
||||
Start, stop and inspect a capture of the rx-loop tick body. **Present
|
||||
only when both `fipsctl` and the daemon are built with
|
||||
`--features profiling`**; the feature is off by default, so a stock
|
||||
package does not carry this subcommand and a stock daemon reports
|
||||
`profile_tick_*` as an unknown command.
|
||||
|
||||
| Subcommand | Control-socket command | Description |
|
||||
| ---------- | ---------------------- | ----------- |
|
||||
| `profile tick on` | `profile_tick_on` | Create the capture file and start recording. Fails if a capture is already running (naming the active file) or if the directory cannot be written. |
|
||||
| `profile tick off` | `profile_tick_off` | Stop the capture. The writer is woken immediately, drains once more and is joined, so the command returns promptly. Succeeds, reporting nothing active, when no capture is running. |
|
||||
| `profile tick status` | `profile_tick_status` | Report `idle`, `running`, `stopped_by_cap` or `stopped_by_error`, plus the active path, bytes written, flush interval and byte cap. |
|
||||
|
||||
`profile tick on` options:
|
||||
|
||||
| Flag | Argument | Default | Description |
|
||||
| ---- | -------- | ------- | ----------- |
|
||||
| `--dir` | directory path | `/var/log/fips` | Where to write the capture. Created if absent. Use it to profile a non-root `cargo run`, or on a platform whose log root differs. |
|
||||
|
||||
One file is written per capture, named `profile-<UTC timestamp>.tsv`.
|
||||
It opens with a `#`-prefixed header block (node npub, build version,
|
||||
platform, configured tick period, flush interval, byte cap, start
|
||||
time), then a tab-separated column header, then one row per measured
|
||||
step per flush interval:
|
||||
|
||||
```text
|
||||
ts_unix kind domain name count max total unit
|
||||
```
|
||||
|
||||
`kind` is `step` for a timed span and `gauge` for a sampled scalar, so
|
||||
a gauge value never lands under a duration column; `unit` names the
|
||||
unit of `max` and `total` for that row. Every step present in the build
|
||||
gets a row every interval, including zero-count rows. Gauges cover
|
||||
ticks per interval, peer count, the wall gap between successive
|
||||
tick-arm entries, and the arm-starvation delay, which is measured
|
||||
against the deadline the tick was scheduled for rather than derived
|
||||
from the gap.
|
||||
|
||||
A capture stops itself on reaching 32 MB, appending a `#` line saying
|
||||
so; `profile tick status` then reports `stopped_by_cap` until the next
|
||||
`on` or `off` clears it.
|
||||
|
||||
## Exit Codes
|
||||
|
||||
| Code | Meaning |
|
||||
|
||||
@@ -159,6 +159,21 @@ not reproduced here to avoid duplicating the source.
|
||||
Both commands run on the daemon's main task and may block briefly
|
||||
while the node mutates its state.
|
||||
|
||||
#### Profiler toggle (`--features profiling` builds only)
|
||||
|
||||
| Command | Params | Behaviour |
|
||||
| ------- | ------ | --------- |
|
||||
| `profile_tick_on` | `dir` (optional directory path; default `/var/log/fips`) | Creates the capture file, publishes its path, and starts the writer thread. `data`: `state`, `path`, `interval_secs`, `byte_cap`. Errors if a capture is already running (naming the active file) or the directory is unwritable. |
|
||||
| `profile_tick_off` | — | Stops the capture, drains once more, joins the writer. `data`: `state`, `stopped`, `stopped_by_cap`, `stopped_by_error`, `path`, `bytes`. |
|
||||
| `profile_tick_status` | — | `data`: `state` (`idle` / `running` / `stopped_by_cap` / `stopped_by_error`), `path`, `bytes`, `byte_cap`, `interval_secs`. |
|
||||
|
||||
Unlike `connect` and `disconnect`, these three are served in the
|
||||
control accept task rather than on the daemon's main task. All of their
|
||||
state is process statics and none of them needs `&mut Node`, so
|
||||
routing them through the main loop would only make the toggle queue
|
||||
behind the tick body it exists to measure. They are absent from a
|
||||
default build, where the daemon answers them as unknown commands.
|
||||
|
||||
## Gateway Command Catalog
|
||||
|
||||
`fips-gateway` exposes a separate control socket with its own command
|
||||
|
||||
@@ -19,6 +19,11 @@ RestartSec=5
|
||||
RuntimeDirectory=fips
|
||||
RuntimeDirectoryMode=0750
|
||||
|
||||
# Log directory (/var/log/fips/), where the built-in tick-body profiler writes
|
||||
# its capture files. Declared so systemd creates it on start and removes it on
|
||||
# purge; the daemon runs as root and already has access without it.
|
||||
LogsDirectory=fips
|
||||
|
||||
# Security hardening (daemon runs as root for TUN and raw sockets)
|
||||
ProtectHome=yes
|
||||
PrivateTmp=yes
|
||||
|
||||
@@ -14,6 +14,11 @@ RestartSec=5
|
||||
RuntimeDirectory=fips
|
||||
RuntimeDirectoryMode=0750
|
||||
|
||||
# Log directory (/var/log/fips/), where the built-in tick-body profiler writes
|
||||
# its capture files. Declared so systemd creates it on start and removes it on
|
||||
# purge; the daemon runs as root and already has access without it.
|
||||
LogsDirectory=fips
|
||||
|
||||
# Security hardening (daemon runs as root for TUN and raw sockets)
|
||||
ProtectHome=yes
|
||||
PrivateTmp=yes
|
||||
|
||||
@@ -76,6 +76,37 @@ enum Commands {
|
||||
#[command(subcommand)]
|
||||
what: StatsCommands,
|
||||
},
|
||||
/// Control the built-in profiler (requires a `--features profiling` build)
|
||||
#[cfg(feature = "profiling")]
|
||||
Profile {
|
||||
#[command(subcommand)]
|
||||
what: ProfileCommands,
|
||||
},
|
||||
}
|
||||
|
||||
#[cfg(feature = "profiling")]
|
||||
#[derive(Subcommand, Debug)]
|
||||
enum ProfileCommands {
|
||||
/// Profile the rx-loop tick body
|
||||
Tick {
|
||||
#[command(subcommand)]
|
||||
action: ProfileTickAction,
|
||||
},
|
||||
}
|
||||
|
||||
#[cfg(feature = "profiling")]
|
||||
#[derive(Subcommand, Debug)]
|
||||
enum ProfileTickAction {
|
||||
/// Start a capture
|
||||
On {
|
||||
/// Directory for the capture file (default /var/log/fips)
|
||||
#[arg(long)]
|
||||
dir: Option<PathBuf>,
|
||||
},
|
||||
/// Stop the running capture
|
||||
Off,
|
||||
/// Report capture state
|
||||
Status,
|
||||
}
|
||||
|
||||
#[derive(Subcommand, Debug)]
|
||||
@@ -471,6 +502,20 @@ fn main() {
|
||||
build_command("show_stats_history", params)
|
||||
}
|
||||
},
|
||||
#[cfg(feature = "profiling")]
|
||||
Commands::Profile { what } => match what {
|
||||
ProfileCommands::Tick { action } => match action {
|
||||
ProfileTickAction::On { dir } => match dir {
|
||||
Some(dir) => build_command(
|
||||
"profile_tick_on",
|
||||
serde_json::json!({"dir": dir.display().to_string()}),
|
||||
),
|
||||
None => build_query("profile_tick_on"),
|
||||
},
|
||||
ProfileTickAction::Off => build_query("profile_tick_off"),
|
||||
ProfileTickAction::Status => build_query("profile_tick_status"),
|
||||
},
|
||||
},
|
||||
Commands::Keygen { .. } => unreachable!(),
|
||||
};
|
||||
|
||||
|
||||
@@ -118,10 +118,41 @@ impl ControlReadHandle {
|
||||
/// Cutover queries (R1) read only `NodeContext` / `MetricsRegistry` (the state
|
||||
/// the read handle already bundles) plus host-OS facts (`/proc`, nftables), so
|
||||
/// they render entirely in the control task without touching `Node`.
|
||||
///
|
||||
/// **It now also carries mutating commands**, namely the `profile_tick_*`
|
||||
/// family under the `profiling` feature. They are served here rather than on
|
||||
/// the rx_loop deliberately: all of their state is process statics, they need
|
||||
/// no `&mut Node`, and routing them through the loop would make the toggle
|
||||
/// queue behind the very behavior it exists to measure.
|
||||
pub(crate) fn snapshot_dispatch(request: &Request, handle: &ControlReadHandle) -> Option<Response> {
|
||||
use crate::control::queries;
|
||||
|
||||
match request.command.as_str() {
|
||||
// Tick-body profiler toggle. Present only in a `--features profiling`
|
||||
// build; otherwise these fall through to the rx_loop dispatch, which
|
||||
// reports them as unknown commands.
|
||||
#[cfg(feature = "profiling")]
|
||||
"profile_tick_on" => {
|
||||
let dir = request
|
||||
.params
|
||||
.as_ref()
|
||||
.and_then(|p| p.get("dir"))
|
||||
.and_then(|v| v.as_str());
|
||||
let context = handle.context();
|
||||
let npub = context.identity.npub();
|
||||
let period = context.config.node.tick_interval_secs;
|
||||
Some(match crate::instr::capture::start(dir, &npub, period) {
|
||||
Ok(value) => Response::ok(value),
|
||||
Err(e) => Response::error(e),
|
||||
})
|
||||
}
|
||||
#[cfg(feature = "profiling")]
|
||||
"profile_tick_off" => Some(match crate::instr::capture::stop() {
|
||||
Ok(value) => Response::ok(value),
|
||||
Err(e) => Response::error(e),
|
||||
}),
|
||||
#[cfg(feature = "profiling")]
|
||||
"profile_tick_status" => Some(Response::ok(crate::instr::capture::status())),
|
||||
"show_listening_sockets" => Some(Response::ok(
|
||||
queries::show_listening_sockets_from_handle(handle),
|
||||
)),
|
||||
|
||||
@@ -0,0 +1,418 @@
|
||||
//! Capture lifecycle: the arm/disarm state machine, the sink file, and the
|
||||
//! `fipsctl`-facing operations.
|
||||
//!
|
||||
//! The toggle — not the writer — creates and opens the sink and publishes its
|
||||
//! path, so an unwritable directory fails the `on` command loudly instead of
|
||||
//! being discovered later by a background thread with nobody to report to.
|
||||
//!
|
||||
//! Capture state is a single atomic state machine (`Idle`, `Running`,
|
||||
//! `StoppedByCap`) transitioned by `compare_exchange`. Every accepted control
|
||||
//! connection is served by its own spawned task, so two simultaneous `on`
|
||||
//! requests are genuinely concurrent and must not both create a writer.
|
||||
|
||||
use std::fs::File;
|
||||
use std::io::Write;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use super::recorder;
|
||||
use super::writer;
|
||||
|
||||
/// Default sink directory. Overridable per capture with `--dir`.
|
||||
pub(crate) const DEFAULT_DIR: &str = "/var/log/fips";
|
||||
|
||||
/// Writer flush interval.
|
||||
pub(crate) const INTERVAL: Duration = Duration::from_secs(10);
|
||||
|
||||
/// Size at which a capture stops itself. Reaching it stops the capture rather
|
||||
/// than rotating: the point of a capture is a bounded, self-describing window.
|
||||
pub(crate) const BYTE_CAP: u64 = 32 * 1024 * 1024;
|
||||
|
||||
pub(crate) const IDLE: u8 = 0;
|
||||
pub(crate) const RUNNING: u8 = 1;
|
||||
pub(crate) const STOPPED_BY_CAP: u8 = 2;
|
||||
/// The writer could not write and stopped itself. Distinct from a cap stop:
|
||||
/// a capture that died on a full disk produced a truncated window, and calling
|
||||
/// that "stopped_by_cap" tells the operator it ran to its limit when it did
|
||||
/// not. The trailer line explaining it goes to the same failing file, so the
|
||||
/// state is the only signal that survives.
|
||||
pub(crate) const STOPPED_BY_ERROR: u8 = 3;
|
||||
|
||||
static STATE: AtomicU8 = AtomicU8::new(IDLE);
|
||||
static GATE: AtomicBool = AtomicBool::new(false);
|
||||
static BYTES: AtomicU64 = AtomicU64::new(0);
|
||||
static ACTIVE_PATH: Mutex<Option<PathBuf>> = Mutex::new(None);
|
||||
static WRITER: Mutex<Option<writer::Handle>> = Mutex::new(None);
|
||||
|
||||
/// The per-tick gate. One relaxed load per tick when the feature is compiled in
|
||||
/// and no capture is running.
|
||||
#[inline]
|
||||
pub(crate) fn gate() -> bool {
|
||||
GATE.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
pub(crate) fn bytes_written() -> u64 {
|
||||
BYTES.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
pub(crate) fn add_bytes(n: u64) -> u64 {
|
||||
BYTES.fetch_add(n, Ordering::Relaxed) + n
|
||||
}
|
||||
|
||||
fn active_path() -> Option<PathBuf> {
|
||||
ACTIVE_PATH
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner())
|
||||
.clone()
|
||||
}
|
||||
|
||||
fn path_display() -> String {
|
||||
active_path()
|
||||
.map(|p| p.display().to_string())
|
||||
.unwrap_or_else(|| "<none>".to_string())
|
||||
}
|
||||
|
||||
fn state_name(state: u8) -> &'static str {
|
||||
match state {
|
||||
RUNNING => "running",
|
||||
STOPPED_BY_CAP => "stopped_by_cap",
|
||||
STOPPED_BY_ERROR => "stopped_by_error",
|
||||
_ => "idle",
|
||||
}
|
||||
}
|
||||
|
||||
/// Called by the writer when it stops itself. `terminal` is `STOPPED_BY_CAP`
|
||||
/// or `STOPPED_BY_ERROR`. Returns true if this call is the one that stopped it.
|
||||
pub(crate) fn mark_stopped(terminal: u8) -> bool {
|
||||
debug_assert!(terminal == STOPPED_BY_CAP || terminal == STOPPED_BY_ERROR);
|
||||
GATE.store(false, Ordering::Relaxed);
|
||||
STATE
|
||||
.compare_exchange(RUNNING, terminal, Ordering::AcqRel, Ordering::Acquire)
|
||||
.is_ok()
|
||||
}
|
||||
|
||||
/// Join the writer thread, if one exists. Never called while holding another
|
||||
/// lock the writer might want.
|
||||
fn reap() {
|
||||
let handle = WRITER.lock().unwrap_or_else(|e| e.into_inner()).take();
|
||||
if let Some(handle) = handle {
|
||||
handle.stop_and_join();
|
||||
}
|
||||
}
|
||||
|
||||
/// Arm a capture.
|
||||
///
|
||||
/// Opens the sink first and only then starts the writer, so a bad `--dir` is
|
||||
/// reported to the caller rather than logged into the void.
|
||||
pub(crate) fn start(
|
||||
dir: Option<&str>,
|
||||
node_npub: &str,
|
||||
tick_period_secs: u64,
|
||||
) -> Result<serde_json::Value, String> {
|
||||
claim()?;
|
||||
|
||||
match open_sink(dir, node_npub, tick_period_secs) {
|
||||
Ok((file, path, header_len)) => {
|
||||
recorder::reset();
|
||||
BYTES.store(header_len, Ordering::Relaxed);
|
||||
match writer::spawn(file) {
|
||||
Ok(handle) => {
|
||||
*WRITER.lock().unwrap_or_else(|e| e.into_inner()) = Some(handle);
|
||||
*ACTIVE_PATH.lock().unwrap_or_else(|e| e.into_inner()) = Some(path.clone());
|
||||
GATE.store(true, Ordering::Release);
|
||||
Ok(serde_json::json!({
|
||||
"state": "running",
|
||||
"path": path.display().to_string(),
|
||||
"interval_secs": INTERVAL.as_secs(),
|
||||
"byte_cap": BYTE_CAP,
|
||||
}))
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = std::fs::remove_file(&path);
|
||||
BYTES.store(0, Ordering::Relaxed);
|
||||
STATE.store(IDLE, Ordering::Release);
|
||||
Err(format!("cannot start profile writer thread: {e}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
BYTES.store(0, Ordering::Relaxed);
|
||||
STATE.store(IDLE, Ordering::Release);
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Take the capture slot, reaping a cap-stopped predecessor if that is what is
|
||||
/// in the way.
|
||||
fn claim() -> Result<(), String> {
|
||||
match STATE.compare_exchange(IDLE, RUNNING, Ordering::AcqRel, Ordering::Acquire) {
|
||||
Ok(_) => Ok(()),
|
||||
Err(RUNNING) => Err(format!("capture already running: {}", path_display())),
|
||||
Err(stopped @ (STOPPED_BY_CAP | STOPPED_BY_ERROR)) => {
|
||||
reap();
|
||||
*ACTIVE_PATH.lock().unwrap_or_else(|e| e.into_inner()) = None;
|
||||
STATE
|
||||
.compare_exchange(stopped, RUNNING, Ordering::AcqRel, Ordering::Acquire)
|
||||
.map(|_| ())
|
||||
.map_err(|_| "capture state changed concurrently; retry".to_string())
|
||||
}
|
||||
Err(_) => Err("capture in an unexpected state".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Disarm the capture. Succeeds when nothing is running, reporting so.
|
||||
pub(crate) fn stop() -> Result<serde_json::Value, String> {
|
||||
let previous = STATE.load(Ordering::Acquire);
|
||||
if previous == IDLE {
|
||||
return Ok(serde_json::json!({"state": "idle", "stopped": false}));
|
||||
}
|
||||
GATE.store(false, Ordering::Release);
|
||||
// The writer wakes on the stop message rather than after the interval, so
|
||||
// this join returns promptly instead of parking the caller for up to one
|
||||
// flush interval.
|
||||
reap();
|
||||
let path = path_display();
|
||||
*ACTIVE_PATH.lock().unwrap_or_else(|e| e.into_inner()) = None;
|
||||
let bytes = bytes_written();
|
||||
// Clear the counter with the slot: a later `status` while idle must not
|
||||
// report the previous capture's byte total as though a capture were live.
|
||||
BYTES.store(0, Ordering::Relaxed);
|
||||
STATE.store(IDLE, Ordering::Release);
|
||||
Ok(serde_json::json!({
|
||||
"state": "idle",
|
||||
"stopped": true,
|
||||
"stopped_by_cap": previous == STOPPED_BY_CAP,
|
||||
"stopped_by_error": previous == STOPPED_BY_ERROR,
|
||||
"path": path,
|
||||
"bytes": bytes,
|
||||
}))
|
||||
}
|
||||
|
||||
/// Report capture state. Distinguishes all three states.
|
||||
pub(crate) fn status() -> serde_json::Value {
|
||||
let state = STATE.load(Ordering::Acquire);
|
||||
serde_json::json!({
|
||||
"state": state_name(state),
|
||||
"path": active_path().map(|p| p.display().to_string()),
|
||||
"bytes": bytes_written(),
|
||||
"byte_cap": BYTE_CAP,
|
||||
"interval_secs": INTERVAL.as_secs(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Stop and reap at daemon teardown. Idempotent.
|
||||
pub(crate) fn shutdown() {
|
||||
if STATE.load(Ordering::Acquire) != IDLE {
|
||||
let _ = stop();
|
||||
}
|
||||
}
|
||||
|
||||
/// Create the sink file and write its header block. Returns the open file, its
|
||||
/// path, and the number of header bytes written.
|
||||
fn open_sink(
|
||||
dir: Option<&str>,
|
||||
node_npub: &str,
|
||||
tick_period_secs: u64,
|
||||
) -> Result<(File, PathBuf, u64), String> {
|
||||
let dir = PathBuf::from(dir.unwrap_or(DEFAULT_DIR));
|
||||
std::fs::create_dir_all(&dir)
|
||||
.map_err(|e| format!("cannot use profile directory {}: {e}", dir.display()))?;
|
||||
|
||||
let start_unix = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0);
|
||||
let path = dir.join(format!("profile-{}.tsv", compact_utc(start_unix)));
|
||||
|
||||
let mut file = File::create(&path)
|
||||
.map_err(|e| format!("cannot create profile file {}: {e}", path.display()))?;
|
||||
|
||||
let header = format!(
|
||||
"# fips tick profile\n\
|
||||
# node\t{node}\n\
|
||||
# build\t{build}\n\
|
||||
# platform\t{platform}\n\
|
||||
# tick_period_secs\t{period}\n\
|
||||
# interval_secs\t{interval}\n\
|
||||
# byte_cap\t{cap}\n\
|
||||
# start_utc\t{start_utc}\n\
|
||||
# start_unix\t{start_unix}\n\
|
||||
# NOTE\tstep durations are WALL CLOCK across await points, not CPU time:\n\
|
||||
# NOTE\ta step that awaits I/O accrues the wait, and other tasks may run\n\
|
||||
# NOTE\tinside that span. That is the intended measure for head-of-line\n\
|
||||
# NOTE\tdelay; do not read a large step as CPU cost.\n\
|
||||
# NOTE\tarm_starvation is measured directly as (entry time - the deadline\n\
|
||||
# NOTE\tthe interval scheduled the tick for). It is NOT derived from\n\
|
||||
# NOTE\ttick_entry_gap, which carries no starvation signal by itself:\n\
|
||||
# NOTE\tunder a steady delay every gap is exactly one tick period.\n\
|
||||
ts_unix\tkind\tdomain\tname\tcount\tmax\ttotal\tunit\n",
|
||||
node = node_npub,
|
||||
build = crate::version::short_version(),
|
||||
platform = std::env::consts::OS,
|
||||
period = tick_period_secs,
|
||||
interval = INTERVAL.as_secs(),
|
||||
cap = BYTE_CAP,
|
||||
start_utc = iso_utc(start_unix),
|
||||
start_unix = start_unix,
|
||||
);
|
||||
file.write_all(header.as_bytes())
|
||||
.map_err(|e| format!("cannot write profile header to {}: {e}", path.display()))?;
|
||||
|
||||
Ok((file, path, header.len() as u64))
|
||||
}
|
||||
|
||||
/// Break a Unix timestamp into UTC `(year, month, day, hour, minute, second)`.
|
||||
///
|
||||
/// Hinnant's `civil_from_days`, era-based. No date crate is in the dependency
|
||||
/// set and one filename stamp does not justify adding one.
|
||||
fn utc_parts(unix: u64) -> (i64, u32, u32, u32, u32, u32) {
|
||||
let days = (unix / 86_400) as i64;
|
||||
let secs = unix % 86_400;
|
||||
let z = days + 719_468;
|
||||
let era = z.div_euclid(146_097);
|
||||
let doe = z.rem_euclid(146_097);
|
||||
let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
|
||||
let y = yoe + era * 400;
|
||||
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
|
||||
let mp = (5 * doy + 2) / 153;
|
||||
let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
|
||||
let m = (if mp < 10 { mp + 3 } else { mp - 9 }) as u32;
|
||||
let y = if m <= 2 { y + 1 } else { y };
|
||||
(
|
||||
y,
|
||||
m,
|
||||
d,
|
||||
(secs / 3_600) as u32,
|
||||
((secs % 3_600) / 60) as u32,
|
||||
(secs % 60) as u32,
|
||||
)
|
||||
}
|
||||
|
||||
/// `20260727T191500Z` — filename-safe.
|
||||
fn compact_utc(unix: u64) -> String {
|
||||
let (y, mo, d, h, mi, s) = utc_parts(unix);
|
||||
format!("{y:04}{mo:02}{d:02}T{h:02}{mi:02}{s:02}Z")
|
||||
}
|
||||
|
||||
/// `2026-07-27T19:15:00Z` — for the header block.
|
||||
fn iso_utc(unix: u64) -> String {
|
||||
let (y, mo, d, h, mi, s) = utc_parts(unix);
|
||||
format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z")
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn utc_parts_matches_known_instants() {
|
||||
assert_eq!(utc_parts(0), (1970, 1, 1, 0, 0, 0));
|
||||
assert_eq!(utc_parts(946_684_800), (2000, 1, 1, 0, 0, 0));
|
||||
// 2026-07-27T19:15:00Z
|
||||
assert_eq!(utc_parts(1_785_179_700), (2026, 7, 27, 19, 15, 0));
|
||||
// Leap day.
|
||||
assert_eq!(utc_parts(1_709_164_800), (2024, 2, 29, 0, 0, 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stamps_render_expected_shapes() {
|
||||
assert_eq!(compact_utc(1_785_179_700), "20260727T191500Z");
|
||||
assert_eq!(iso_utc(1_785_179_700), "2026-07-27T19:15:00Z");
|
||||
}
|
||||
|
||||
// The lock these tests take is shared with the recorder tests, which
|
||||
// mutate the same statics. See `crate::instr::test_serial`.
|
||||
|
||||
#[test]
|
||||
fn capture_round_trip_writes_header_and_rows() {
|
||||
let _guard = crate::instr::test_serial();
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let dir_str = dir.path().to_str().unwrap().to_string();
|
||||
|
||||
let started = start(Some(&dir_str), "npub1test", 1).expect("start");
|
||||
assert_eq!(started["state"], "running");
|
||||
assert!(gate(), "gate must be armed while running");
|
||||
let path = PathBuf::from(started["path"].as_str().unwrap());
|
||||
|
||||
// A second `on` is refused while one is running, and names the file.
|
||||
let refused = start(Some(&dir_str), "npub1test", 1).unwrap_err();
|
||||
assert!(refused.contains(&path.display().to_string()), "{refused}");
|
||||
|
||||
// Feed one observation so the drained rows are not all zero.
|
||||
recorder::record(
|
||||
recorder::Domain::Tick,
|
||||
recorder::Step::WholeTick,
|
||||
Duration::from_millis(7),
|
||||
);
|
||||
|
||||
// Stopping wakes the writer immediately; it drains once more and joins.
|
||||
let stopped = stop().expect("stop");
|
||||
assert_eq!(stopped["stopped"], true);
|
||||
assert_eq!(stopped["stopped_by_cap"], false);
|
||||
assert!(!gate(), "gate must be clear after stop");
|
||||
|
||||
let text = std::fs::read_to_string(&path).expect("read capture");
|
||||
assert!(text.starts_with("# fips tick profile\n"), "{text}");
|
||||
assert!(text.contains("# node\tnpub1test\n"), "{text}");
|
||||
assert!(
|
||||
text.contains("ts_unix\tkind\tdomain\tname\tcount\tmax\ttotal\tunit\n"),
|
||||
"{text}"
|
||||
);
|
||||
// The final drain emitted one row per emitted step, plus the gauges.
|
||||
let rows: Vec<&str> = text
|
||||
.lines()
|
||||
.filter(|l| l.starts_with(|c: char| c.is_ascii_digit()))
|
||||
.collect();
|
||||
let expected_steps = recorder::STEPS.iter().filter(|s| s.emitted()).count();
|
||||
assert_eq!(rows.len(), expected_steps + recorder::N_GAUGES);
|
||||
// The 7 ms observation above is in the whole-tick row, converted to
|
||||
// microseconds. Bounds rather than equality: the gate is process-wide,
|
||||
// so a node under test elsewhere in this binary may have ticked into
|
||||
// the same capture window.
|
||||
let whole_tick = rows
|
||||
.iter()
|
||||
.find(|r| r.contains("\tstep\ttick\twhole_tick\t"))
|
||||
.expect("whole_tick row");
|
||||
let fields: Vec<&str> = whole_tick.split('\t').collect();
|
||||
assert_eq!(fields.last(), Some(&"us"), "{whole_tick}");
|
||||
assert!(
|
||||
fields[4].parse::<u64>().unwrap() >= 1,
|
||||
"count: {whole_tick}"
|
||||
);
|
||||
assert!(
|
||||
fields[5].parse::<u64>().unwrap() >= 7_000,
|
||||
"max: {whole_tick}"
|
||||
);
|
||||
assert!(
|
||||
rows.iter()
|
||||
.any(|r| r.contains("\tgauge\ttick\tarm_starvation\t")),
|
||||
"{text}"
|
||||
);
|
||||
|
||||
// A stop with nothing running is not an error.
|
||||
let again = stop().expect("second stop");
|
||||
assert_eq!(again["stopped"], false);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn start_fails_loudly_on_an_unwritable_directory() {
|
||||
let _guard = crate::instr::test_serial();
|
||||
let err = start(Some("/proc/fips-profile-should-not-exist"), "npub1test", 1)
|
||||
.expect_err("must fail");
|
||||
assert!(err.contains("profile directory"), "{err}");
|
||||
// The failed attempt must leave the slot free for the next try.
|
||||
assert_eq!(STATE.load(Ordering::Acquire), IDLE);
|
||||
assert!(!gate());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn status_reports_the_bounds_it_is_enforcing() {
|
||||
let _guard = crate::instr::test_serial();
|
||||
let value = status();
|
||||
assert_eq!(value["byte_cap"], BYTE_CAP);
|
||||
assert_eq!(value["interval_secs"], INTERVAL.as_secs());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
//! Tick-body instrumentation.
|
||||
//!
|
||||
//! A purpose-built, feature-gated profiler for the rx-loop tick arm. It exists
|
||||
//! to answer one question with field data: which subsystem step dominates the
|
||||
//! tick body, and how long does the tick arm wait behind the other `select!`
|
||||
//! arms before it runs at all.
|
||||
//!
|
||||
//! # Shape
|
||||
//!
|
||||
//! - Everything that costs anything at runtime is behind the `profiling` Cargo
|
||||
//! feature, which is **off by default**. The default build's neutrality is a
|
||||
//! property of the generated code, not of a runtime check.
|
||||
//! - The instrumentation macro is defined twice, once per feature state. The
|
||||
//! feature-off definition is a pure pass-through: it expands to the measured
|
||||
//! expression and nothing else, so no timing code exists in a default build.
|
||||
//! - The always-present surface — [`gate`], [`tick_entry`], [`tick_gauges`],
|
||||
//! [`shutdown`] — exists in both feature states because the call sites in
|
||||
//! `rx_loop.rs` and the lifecycle teardown must compile either way. Their
|
||||
//! feature-off forms are empty (and [`gate`] is a `const fn` returning
|
||||
//! `false`), so they cost nothing.
|
||||
//! - The module is named `instr` rather than `profiling` so that it sorts
|
||||
//! before `node` in `lib.rs`'s alphabetical module list: a `#[macro_use]`
|
||||
//! module must be declared before the modules that use its macros.
|
||||
//!
|
||||
//! # Data model
|
||||
//!
|
||||
//! Domain above step: [`Domain`] carries exactly one variant today
|
||||
//! (`Domain::Tick`). The primitive, the recorder, the writer and the `fipsctl`
|
||||
//! surface all take a domain, so adding a data-path domain later is additive.
|
||||
//! No second domain is declared until something records into it.
|
||||
//!
|
||||
//! Per (domain, step) the recorder keeps an exact count, max and total in fixed
|
||||
//! static `AtomicU64` arrays — no histogram, no accumulation, a fixed footprint
|
||||
//! regardless of run length. Gauges (ticks per interval, peer count, and the
|
||||
//! arm-starvation figures) live in a parallel array and are emitted with an
|
||||
//! explicit row kind so a gauge value never lands under a duration column.
|
||||
|
||||
#[cfg(feature = "profiling")]
|
||||
pub(crate) mod capture;
|
||||
#[cfg(feature = "profiling")]
|
||||
mod recorder;
|
||||
#[cfg(feature = "profiling")]
|
||||
mod writer;
|
||||
|
||||
#[cfg(feature = "profiling")]
|
||||
pub(crate) use recorder::{Domain, Step, now, record};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The macro pair.
|
||||
//
|
||||
// Every path in the body is `$crate::`-qualified. `macro_rules!` bodies are not
|
||||
// path-hygienic: an unqualified `Instant::now()` or `record(..)` would resolve
|
||||
// at the *call site* (`rx_loop.rs`), where neither name is in scope. Importing
|
||||
// them there is worse still, because the imports would be unused in the
|
||||
// feature-off build and red it under `-D warnings`.
|
||||
//
|
||||
// `$e` is evaluated exactly once in both forms, which is what makes nesting the
|
||||
// whole-tick span around the per-step spans safe.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Time `$e` as one step of `$domain`, when `$on` is true.
|
||||
///
|
||||
/// `$on` is the per-tick gate hoist: the enable flag is read once at the top of
|
||||
/// the tick arm into a local, and that local is passed explicitly to every
|
||||
/// invocation, because macro hygiene makes a call-site local invisible inside
|
||||
/// the macro body.
|
||||
#[cfg(feature = "profiling")]
|
||||
macro_rules! instr_step {
|
||||
($on:expr, $domain:expr, $step:expr, $e:expr) => {{
|
||||
let t0 = if $on {
|
||||
Some($crate::instr::now())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let r = $e;
|
||||
if let Some(t) = t0 {
|
||||
$crate::instr::record($domain, $step, t.elapsed());
|
||||
}
|
||||
r
|
||||
}};
|
||||
}
|
||||
|
||||
/// Feature-off form: a pure pass-through. The expansion contains no clock read,
|
||||
/// no counter update and no reference to the recorder — only the measured
|
||||
/// expression, plus a discard of the gate local so it is not unused.
|
||||
#[cfg(not(feature = "profiling"))]
|
||||
macro_rules! instr_step {
|
||||
($on:expr, $domain:expr, $step:expr, $e:expr) => {{
|
||||
let _ = &$on;
|
||||
$e
|
||||
}};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Always-present surface.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Whether a capture is armed. Read **once per tick** into a local that is then
|
||||
/// passed to each `instr_step!` invocation, so the feature-on-but-idle cost of
|
||||
/// the whole tick arm is a single relaxed load.
|
||||
#[cfg(feature = "profiling")]
|
||||
#[inline]
|
||||
pub(crate) fn gate() -> bool {
|
||||
capture::gate()
|
||||
}
|
||||
|
||||
/// Feature-off gate: a `const fn` returning `false`, so the whole tick arm
|
||||
/// folds to the uninstrumented sequence at compile time.
|
||||
#[cfg(not(feature = "profiling"))]
|
||||
#[inline]
|
||||
pub(crate) const fn gate() -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// Record how late this tick-arm entry is against its scheduled deadline.
|
||||
///
|
||||
/// The arm is polled **last** under `biased;`, so its lateness is the time it
|
||||
/// spent waiting behind the packet, TUN and control arms. `tokio::time::
|
||||
/// interval::tick` returns the deadline it was scheduled for, so this is a
|
||||
/// direct subtraction rather than a model. Two earlier designs derived it from
|
||||
/// the inter-entry gap instead and both under-reported: one by the previous
|
||||
/// body, the other by reporting only the first difference of the delay, so a
|
||||
/// sustained stall read as zero. The inter-entry gap is still recorded as its
|
||||
/// own gauge, but it carries no starvation signal on its own.
|
||||
#[cfg(feature = "profiling")]
|
||||
#[inline]
|
||||
pub(crate) fn tick_entry(on: bool, deadline: std::time::Instant, now: std::time::Instant) {
|
||||
recorder::tick_entry(on, deadline, now);
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "profiling"))]
|
||||
#[inline]
|
||||
pub(crate) fn tick_entry(_on: bool, _deadline: std::time::Instant, _now: std::time::Instant) {}
|
||||
|
||||
/// Sample the per-tick gauges taken from node state.
|
||||
#[cfg(feature = "profiling")]
|
||||
#[inline]
|
||||
pub(crate) fn tick_gauges(on: bool, peers: u64) {
|
||||
recorder::tick_gauges(on, peers);
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "profiling"))]
|
||||
#[inline]
|
||||
pub(crate) fn tick_gauges(_on: bool, _peers: u64) {}
|
||||
|
||||
/// Stop and reap any running capture at daemon teardown. Idempotent.
|
||||
#[cfg(feature = "profiling")]
|
||||
pub(crate) fn shutdown() {
|
||||
capture::shutdown();
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "profiling"))]
|
||||
pub(crate) fn shutdown() {}
|
||||
|
||||
/// One serialization lock for every test in this module tree.
|
||||
///
|
||||
/// The recorder counters and the capture state machine are the *same* process
|
||||
/// statics: `capture::start` calls `recorder::reset`, and `capture::stop`
|
||||
/// drains every slot. Two suites with their own locks therefore do not
|
||||
/// serialize against each other, and the feature-on stage runs tests as
|
||||
/// threads in one process, so a capture round-trip can zero the counters a
|
||||
/// recorder test is mid-way through asserting on. One lock for both.
|
||||
#[cfg(all(test, feature = "profiling"))]
|
||||
pub(crate) static TEST_SERIAL: std::sync::Mutex<()> = std::sync::Mutex::new(());
|
||||
|
||||
/// Take the shared test lock, recovering from a poisoned mutex so one failing
|
||||
/// test does not cascade into every other one.
|
||||
#[cfg(all(test, feature = "profiling"))]
|
||||
pub(crate) fn test_serial() -> std::sync::MutexGuard<'static, ()> {
|
||||
TEST_SERIAL.lock().unwrap_or_else(|e| e.into_inner())
|
||||
}
|
||||
@@ -0,0 +1,478 @@
|
||||
//! Fixed-footprint recorder: exact count / max / total per (domain, step).
|
||||
//!
|
||||
//! All state is process statics, not `Node` state, because the `fipsctl`
|
||||
//! handler that arms and disarms a capture runs in the control accept task and
|
||||
//! has no `&Node` — that is the whole point of serving it off-loop, so it
|
||||
//! cannot queue behind the behavior it is measuring.
|
||||
//!
|
||||
//! The writer thread is the only reader. It takes each interval's figures with
|
||||
//! `swap(0)`, so there are no "previous value" arrays to carry and the counters
|
||||
//! are per-interval by construction.
|
||||
|
||||
use std::sync::LazyLock;
|
||||
use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// Measurement domain. Structural only: one variant today.
|
||||
///
|
||||
/// A data-path domain is deliberately **not** declared until something records
|
||||
/// into it. What generalizes here is the enum, the counter table and the
|
||||
/// writer; the per-tick gate hoist does not, so a data-path domain will need
|
||||
/// its own gate strategy.
|
||||
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
|
||||
#[repr(usize)]
|
||||
pub(crate) enum Domain {
|
||||
Tick = 0,
|
||||
}
|
||||
|
||||
pub(crate) const N_DOMAINS: usize = 1;
|
||||
pub(crate) const DOMAINS: [Domain; N_DOMAINS] = [Domain::Tick];
|
||||
|
||||
impl Domain {
|
||||
pub(crate) const fn name(self) -> &'static str {
|
||||
match self {
|
||||
Domain::Tick => "tick",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One measured step of the rx-loop tick arm, in call order, plus the
|
||||
/// whole-body span.
|
||||
///
|
||||
/// `as usize` indexes the counter arrays, so the discriminants are dense and
|
||||
/// `WholeTick` is last (it defines `N_STEPS`). Variants are declared
|
||||
/// unconditionally — see [`Step::emitted`] for how the two platform- and
|
||||
/// profile-conditional steps are kept out of the emitted table.
|
||||
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
|
||||
#[repr(usize)]
|
||||
pub(crate) enum Step {
|
||||
CheckTimeouts = 0,
|
||||
ReloadPeerAcl,
|
||||
ReloadHostMap,
|
||||
PollPendingConnects,
|
||||
PollNostrRendezvous,
|
||||
PollLanRendezvous,
|
||||
DrivePeerTimers,
|
||||
ResendPendingRekeys,
|
||||
ResendPendingSessionHandshakes,
|
||||
ResendPendingSessionMsg3,
|
||||
PurgeIdleSessions,
|
||||
ProcessPendingRetries,
|
||||
CheckTreeState,
|
||||
CheckBloomState,
|
||||
ComputeMeshSize,
|
||||
RecordStatsHistory,
|
||||
CheckMmpReports,
|
||||
CheckSessionMmpReports,
|
||||
CheckLinkHeartbeats,
|
||||
CheckRekey,
|
||||
CheckSessionRekey,
|
||||
CheckPendingLookups,
|
||||
PollTransportDiscovery,
|
||||
SampleTransportCongestion,
|
||||
ActivateConnectedUdpSessions,
|
||||
DebugAssertPeerMapsCoherent,
|
||||
/// The whole tick-arm body, from before `check_timeouts` to after the last
|
||||
/// step. Composes safely with the per-step spans because the macro
|
||||
/// evaluates its measured expression exactly once.
|
||||
WholeTick,
|
||||
}
|
||||
|
||||
pub(crate) const N_STEPS: usize = Step::WholeTick as usize + 1;
|
||||
|
||||
/// Every step, in emission order. Index `i` of this table is `STEPS[i] as
|
||||
/// usize`; `steps_table_is_dense` asserts it.
|
||||
pub(crate) const STEPS: [Step; N_STEPS] = [
|
||||
Step::CheckTimeouts,
|
||||
Step::ReloadPeerAcl,
|
||||
Step::ReloadHostMap,
|
||||
Step::PollPendingConnects,
|
||||
Step::PollNostrRendezvous,
|
||||
Step::PollLanRendezvous,
|
||||
Step::DrivePeerTimers,
|
||||
Step::ResendPendingRekeys,
|
||||
Step::ResendPendingSessionHandshakes,
|
||||
Step::ResendPendingSessionMsg3,
|
||||
Step::PurgeIdleSessions,
|
||||
Step::ProcessPendingRetries,
|
||||
Step::CheckTreeState,
|
||||
Step::CheckBloomState,
|
||||
Step::ComputeMeshSize,
|
||||
Step::RecordStatsHistory,
|
||||
Step::CheckMmpReports,
|
||||
Step::CheckSessionMmpReports,
|
||||
Step::CheckLinkHeartbeats,
|
||||
Step::CheckRekey,
|
||||
Step::CheckSessionRekey,
|
||||
Step::CheckPendingLookups,
|
||||
Step::PollTransportDiscovery,
|
||||
Step::SampleTransportCongestion,
|
||||
Step::ActivateConnectedUdpSessions,
|
||||
Step::DebugAssertPeerMapsCoherent,
|
||||
Step::WholeTick,
|
||||
];
|
||||
|
||||
impl Step {
|
||||
pub(crate) const fn name(self) -> &'static str {
|
||||
match self {
|
||||
Step::CheckTimeouts => "check_timeouts",
|
||||
Step::ReloadPeerAcl => "reload_peer_acl",
|
||||
Step::ReloadHostMap => "reload_host_map",
|
||||
Step::PollPendingConnects => "poll_pending_connects",
|
||||
Step::PollNostrRendezvous => "poll_nostr_rendezvous",
|
||||
Step::PollLanRendezvous => "poll_lan_rendezvous",
|
||||
Step::DrivePeerTimers => "drive_peer_timers",
|
||||
Step::ResendPendingRekeys => "resend_pending_rekeys",
|
||||
Step::ResendPendingSessionHandshakes => "resend_pending_session_handshakes",
|
||||
Step::ResendPendingSessionMsg3 => "resend_pending_session_msg3",
|
||||
Step::PurgeIdleSessions => "purge_idle_sessions",
|
||||
Step::ProcessPendingRetries => "process_pending_retries",
|
||||
Step::CheckTreeState => "check_tree_state",
|
||||
Step::CheckBloomState => "check_bloom_state",
|
||||
Step::ComputeMeshSize => "compute_mesh_size",
|
||||
Step::RecordStatsHistory => "record_stats_history",
|
||||
Step::CheckMmpReports => "check_mmp_reports",
|
||||
Step::CheckSessionMmpReports => "check_session_mmp_reports",
|
||||
Step::CheckLinkHeartbeats => "check_link_heartbeats",
|
||||
Step::CheckRekey => "check_rekey",
|
||||
Step::CheckSessionRekey => "check_session_rekey",
|
||||
Step::CheckPendingLookups => "check_pending_lookups",
|
||||
Step::PollTransportDiscovery => "poll_transport_discovery",
|
||||
Step::SampleTransportCongestion => "sample_transport_congestion",
|
||||
Step::ActivateConnectedUdpSessions => "activate_connected_udp_sessions",
|
||||
Step::DebugAssertPeerMapsCoherent => "debug_assert_peer_maps_coherent",
|
||||
Step::WholeTick => "whole_tick",
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether this step gets a row in this build.
|
||||
///
|
||||
/// Two steps are conditionally compiled at their call sites. Emitting a row
|
||||
/// for them in a build where the call site does not exist would publish a
|
||||
/// count that is structurally zero forever, which reads as "this step never
|
||||
/// runs" rather than "this step is not in this build". The predicates below
|
||||
/// are the same `cfg` expressions that gate the call sites in
|
||||
/// `node::dataplane::rx_loop`; keep them in step.
|
||||
pub(crate) const fn emitted(self) -> bool {
|
||||
match self {
|
||||
Step::ActivateConnectedUdpSessions => {
|
||||
cfg!(any(target_os = "linux", target_os = "macos"))
|
||||
}
|
||||
Step::DebugAssertPeerMapsCoherent => cfg!(debug_assertions),
|
||||
_ => true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A scalar sampled once per tick, as opposed to a duration.
|
||||
///
|
||||
/// Gauges carry their own row kind and their own unit in the output so a gauge
|
||||
/// value can never be read as a duration.
|
||||
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
|
||||
#[repr(usize)]
|
||||
pub(crate) enum Gauge {
|
||||
Ticks = 0,
|
||||
Peers,
|
||||
TickGap,
|
||||
ArmStarvation,
|
||||
}
|
||||
|
||||
pub(crate) const N_GAUGES: usize = Gauge::ArmStarvation as usize + 1;
|
||||
|
||||
pub(crate) const GAUGES: [Gauge; N_GAUGES] = [
|
||||
Gauge::Ticks,
|
||||
Gauge::Peers,
|
||||
Gauge::TickGap,
|
||||
Gauge::ArmStarvation,
|
||||
];
|
||||
|
||||
impl Gauge {
|
||||
pub(crate) const fn name(self) -> &'static str {
|
||||
match self {
|
||||
Gauge::Ticks => "ticks",
|
||||
Gauge::Peers => "peers",
|
||||
Gauge::TickGap => "tick_entry_gap",
|
||||
Gauge::ArmStarvation => "arm_starvation",
|
||||
}
|
||||
}
|
||||
|
||||
/// Unit of the `max` and `total` columns for this gauge.
|
||||
pub(crate) const fn unit(self) -> &'static str {
|
||||
match self {
|
||||
Gauge::Ticks => "ticks",
|
||||
Gauge::Peers => "peers",
|
||||
Gauge::TickGap | Gauge::ArmStarvation => "us",
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether the gauge's stored values are nanosecond durations that the
|
||||
/// writer converts to microseconds.
|
||||
pub(crate) const fn is_duration(self) -> bool {
|
||||
matches!(self, Gauge::TickGap | Gauge::ArmStarvation)
|
||||
}
|
||||
}
|
||||
|
||||
const N_SLOTS: usize = N_DOMAINS * N_STEPS;
|
||||
|
||||
static COUNT: [AtomicU64; N_SLOTS] = [const { AtomicU64::new(0) }; N_SLOTS];
|
||||
static MAX_NS: [AtomicU64; N_SLOTS] = [const { AtomicU64::new(0) }; N_SLOTS];
|
||||
static TOTAL_NS: [AtomicU64; N_SLOTS] = [const { AtomicU64::new(0) }; N_SLOTS];
|
||||
|
||||
static G_COUNT: [AtomicU64; N_GAUGES] = [const { AtomicU64::new(0) }; N_GAUGES];
|
||||
static G_MAX: [AtomicU64; N_GAUGES] = [const { AtomicU64::new(0) }; N_GAUGES];
|
||||
static G_TOTAL: [AtomicU64; N_GAUGES] = [const { AtomicU64::new(0) }; N_GAUGES];
|
||||
|
||||
/// Monotonic baseline so tick-arm entry times fit in an atomic. Offset by one
|
||||
/// on store so that zero can mean "no previous entry".
|
||||
static BASE: LazyLock<Instant> = LazyLock::new(Instant::now);
|
||||
static PREV_ENTRY_NS: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
#[inline]
|
||||
const fn slot(domain: Domain, step: Step) -> usize {
|
||||
(domain as usize * N_STEPS) + step as usize
|
||||
}
|
||||
|
||||
/// Read the clock for a step span.
|
||||
#[inline]
|
||||
pub(crate) fn now() -> Instant {
|
||||
Instant::now()
|
||||
}
|
||||
|
||||
/// Record one observation of `step`.
|
||||
#[inline]
|
||||
pub(crate) fn record(domain: Domain, step: Step, elapsed: Duration) {
|
||||
let ns = elapsed.as_nanos() as u64;
|
||||
let idx = slot(domain, step);
|
||||
COUNT[idx].fetch_add(1, Relaxed);
|
||||
TOTAL_NS[idx].fetch_add(ns, Relaxed);
|
||||
MAX_NS[idx].fetch_max(ns, Relaxed);
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn record_gauge(gauge: Gauge, value: u64) {
|
||||
let idx = gauge as usize;
|
||||
G_COUNT[idx].fetch_add(1, Relaxed);
|
||||
G_TOTAL[idx].fetch_add(value, Relaxed);
|
||||
G_MAX[idx].fetch_max(value, Relaxed);
|
||||
}
|
||||
|
||||
/// Sample the inter-entry gap and the measured arm-starvation delay.
|
||||
pub(crate) fn tick_entry(on: bool, deadline: Instant, now: Instant) {
|
||||
if !on {
|
||||
return;
|
||||
}
|
||||
// Offset by one so that a stored zero unambiguously means "no previous
|
||||
// entry", even for an entry that lands on the baseline instant.
|
||||
let stamp = BASE.elapsed().as_nanos() as u64 + 1;
|
||||
let late = now.saturating_duration_since(deadline).as_nanos() as u64;
|
||||
tick_entry_at(stamp, late);
|
||||
}
|
||||
|
||||
/// The clock-free half of [`tick_entry`]: both times are inputs, so the
|
||||
/// arithmetic can be driven with synthetic stamps in a test.
|
||||
///
|
||||
/// `late_ns` is how far past its scheduled deadline this entry was, measured
|
||||
/// directly rather than derived. Two earlier designs derived it from the
|
||||
/// inter-entry gap and were both wrong: subtracting the previous body
|
||||
/// understated it by exactly the body, and subtracting `max(period, body)`
|
||||
/// reported the *first difference* of the delay, so a sustained stall — the
|
||||
/// overload regime this measurement exists to characterize — read as zero
|
||||
/// forever. `tokio::time::interval::tick` hands back the deadline it was
|
||||
/// scheduled for, so the delay is a subtraction with no model behind it.
|
||||
fn tick_entry_at(stamp: u64, late_ns: u64) {
|
||||
let prev = PREV_ENTRY_NS.swap(stamp, Relaxed);
|
||||
record_gauge(Gauge::Ticks, 1);
|
||||
record_gauge(Gauge::ArmStarvation, late_ns);
|
||||
if prev == 0 {
|
||||
// First entry of this capture: there is no previous entry to measure a
|
||||
// gap against, and the idle interval before arming is not a gap. The
|
||||
// lateness above does not depend on a previous entry, so it still counts.
|
||||
return;
|
||||
}
|
||||
record_gauge(Gauge::TickGap, stamp.saturating_sub(prev));
|
||||
}
|
||||
|
||||
/// Sample the gauges that come from node state.
|
||||
pub(crate) fn tick_gauges(on: bool, peers: u64) {
|
||||
if !on {
|
||||
return;
|
||||
}
|
||||
record_gauge(Gauge::Peers, peers);
|
||||
}
|
||||
|
||||
/// Take (and clear) this interval's figures for one step: count, max ns, total
|
||||
/// ns. Called only by the writer thread.
|
||||
pub(crate) fn take_step(domain: Domain, step: Step) -> (u64, u64, u64) {
|
||||
let idx = slot(domain, step);
|
||||
(
|
||||
COUNT[idx].swap(0, Relaxed),
|
||||
MAX_NS[idx].swap(0, Relaxed),
|
||||
TOTAL_NS[idx].swap(0, Relaxed),
|
||||
)
|
||||
}
|
||||
|
||||
/// Take (and clear) this interval's figures for one gauge.
|
||||
pub(crate) fn take_gauge(gauge: Gauge) -> (u64, u64, u64) {
|
||||
let idx = gauge as usize;
|
||||
(
|
||||
G_COUNT[idx].swap(0, Relaxed),
|
||||
G_MAX[idx].swap(0, Relaxed),
|
||||
G_TOTAL[idx].swap(0, Relaxed),
|
||||
)
|
||||
}
|
||||
|
||||
/// Zero every counter so a capture starts from a clean slate.
|
||||
pub(crate) fn reset() {
|
||||
for i in 0..N_SLOTS {
|
||||
COUNT[i].store(0, Relaxed);
|
||||
MAX_NS[i].store(0, Relaxed);
|
||||
TOTAL_NS[i].store(0, Relaxed);
|
||||
}
|
||||
for i in 0..N_GAUGES {
|
||||
G_COUNT[i].store(0, Relaxed);
|
||||
G_MAX[i].store(0, Relaxed);
|
||||
G_TOTAL[i].store(0, Relaxed);
|
||||
}
|
||||
PREV_ENTRY_NS.store(0, Relaxed);
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::MutexGuard;
|
||||
|
||||
/// Shared with the capture tests: they mutate the same statics. See
|
||||
/// `crate::instr::test_serial`.
|
||||
fn serial() -> MutexGuard<'static, ()> {
|
||||
crate::instr::test_serial()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn steps_table_is_dense() {
|
||||
for (i, step) in STEPS.iter().enumerate() {
|
||||
assert_eq!(*step as usize, i, "step {} is out of order", step.name());
|
||||
}
|
||||
assert_eq!(STEPS.len(), N_STEPS);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gauges_table_is_dense() {
|
||||
for (i, gauge) in GAUGES.iter().enumerate() {
|
||||
assert_eq!(*gauge as usize, i, "gauge {} is out of order", gauge.name());
|
||||
}
|
||||
assert_eq!(GAUGES.len(), N_GAUGES);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn step_names_are_unique() {
|
||||
let mut names: Vec<&str> = STEPS.iter().map(|s| s.name()).collect();
|
||||
names.sort_unstable();
|
||||
let before = names.len();
|
||||
names.dedup();
|
||||
assert_eq!(before, names.len(), "duplicate step name");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn emitted_row_count_matches_build() {
|
||||
let emitted = STEPS.iter().filter(|s| s.emitted()).count();
|
||||
// 24 unconditional subsystem steps + the whole-tick span, plus the two
|
||||
// conditionally-compiled steps where this build has them.
|
||||
let mut expected = 25;
|
||||
if cfg!(any(target_os = "linux", target_os = "macos")) {
|
||||
expected += 1;
|
||||
}
|
||||
if cfg!(debug_assertions) {
|
||||
expected += 1;
|
||||
}
|
||||
assert_eq!(emitted, expected);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn record_accumulates_count_max_and_total() {
|
||||
let _guard = serial();
|
||||
reset();
|
||||
record(Domain::Tick, Step::CheckRekey, Duration::from_nanos(10));
|
||||
record(Domain::Tick, Step::CheckRekey, Duration::from_nanos(30));
|
||||
let (count, max, total) = take_step(Domain::Tick, Step::CheckRekey);
|
||||
assert_eq!((count, max, total), (2, 30, 40));
|
||||
// Taking clears the slot.
|
||||
assert_eq!(take_step(Domain::Tick, Step::CheckRekey), (0, 0, 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn starvation_is_the_measured_lateness_of_the_entry() {
|
||||
let _guard = serial();
|
||||
reset();
|
||||
// The interval hands back the deadline it was scheduled for, so the
|
||||
// delay is `now - deadline` and nothing is derived from the period, the
|
||||
// previous entry, or the previous body.
|
||||
PREV_ENTRY_NS.store(1_000_000_000, Relaxed);
|
||||
tick_entry_at(1_100_000_000, 50_000_000);
|
||||
assert_eq!(take_gauge(Gauge::TickGap), (1, 100_000_000, 100_000_000));
|
||||
assert_eq!(
|
||||
take_gauge(Gauge::ArmStarvation),
|
||||
(1, 50_000_000, 50_000_000)
|
||||
);
|
||||
assert_eq!(take_gauge(Gauge::Ticks).0, 1);
|
||||
reset();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sustained_lateness_is_reported_on_every_tick() {
|
||||
let _guard = serial();
|
||||
reset();
|
||||
// The regime the two earlier designs both hid. Three consecutive entries
|
||||
// each 50 ms past their deadline, one period apart, i.e. the arm waiting
|
||||
// a constant amount behind the other select arms every round. The gaps
|
||||
// are all exactly one period, so any formula derived from the
|
||||
// inter-entry gap reports zero here; measured lateness reports 50 ms
|
||||
// three times, which is the truth.
|
||||
PREV_ENTRY_NS.store(1_000_000_000, Relaxed);
|
||||
tick_entry_at(1_050_000_000, 50_000_000);
|
||||
tick_entry_at(1_100_000_000, 50_000_000);
|
||||
tick_entry_at(1_150_000_000, 50_000_000);
|
||||
let (count, max, total) = take_gauge(Gauge::ArmStarvation);
|
||||
assert_eq!(count, 3);
|
||||
assert_eq!(max, 50_000_000);
|
||||
assert_eq!(total, 150_000_000);
|
||||
// ...and the gap alone carries no signal about it: every gap is one
|
||||
// period, exactly as it would be on a perfectly healthy node.
|
||||
assert_eq!(take_gauge(Gauge::TickGap), (3, 50_000_000, 150_000_000));
|
||||
reset();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn first_entry_of_a_capture_records_no_gap_but_still_records_lateness() {
|
||||
let _guard = serial();
|
||||
reset();
|
||||
tick_entry_at(500, 7_000_000);
|
||||
assert_eq!(take_gauge(Gauge::Ticks).0, 1);
|
||||
assert_eq!(take_gauge(Gauge::TickGap), (0, 0, 0));
|
||||
// Lateness does not depend on a previous entry, so the first tick of a
|
||||
// capture still contributes one.
|
||||
assert_eq!(take_gauge(Gauge::ArmStarvation), (1, 7_000_000, 7_000_000));
|
||||
reset();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_on_schedule_entry_reports_no_starvation() {
|
||||
let _guard = serial();
|
||||
reset();
|
||||
PREV_ENTRY_NS.store(1_000_000_000, Relaxed);
|
||||
tick_entry_at(1_050_000_000, 0);
|
||||
assert_eq!(take_gauge(Gauge::ArmStarvation), (1, 0, 0));
|
||||
assert_eq!(take_gauge(Gauge::TickGap), (1, 50_000_000, 50_000_000));
|
||||
reset();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gate_off_records_nothing() {
|
||||
let _guard = serial();
|
||||
reset();
|
||||
let t = Instant::now();
|
||||
tick_entry(false, t, t);
|
||||
tick_gauges(false, 42);
|
||||
assert_eq!(take_gauge(Gauge::Ticks), (0, 0, 0));
|
||||
assert_eq!(take_gauge(Gauge::Peers), (0, 0, 0));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
//! The capture writer: a dedicated, named OS thread with an explicit
|
||||
//! lifecycle.
|
||||
//!
|
||||
//! There is no worker-thread lifecycle in this codebase to copy — the crypto
|
||||
//! worker pools are never torn down and drop their join handles at spawn — so
|
||||
//! this one is designed here.
|
||||
//!
|
||||
//! Two properties matter:
|
||||
//!
|
||||
//! - **It is an OS thread, not a spawned task.** The runtime is
|
||||
//! `current_thread`, so file I/O on a task would run on the rx loop's own
|
||||
//! thread and stall every `select!` arm, including the one being measured.
|
||||
//! - **It waits on a channel with a timeout, not on a sleep.** `recv_timeout`
|
||||
//! returns immediately when the toggle sends stop, so `off` performs a final
|
||||
//! drain and joins promptly instead of parking the caller for up to a full
|
||||
//! flush interval.
|
||||
//!
|
||||
//! The thread is created lazily when a capture starts, so a node that never
|
||||
//! arms one never has the thread.
|
||||
|
||||
use std::fs::File;
|
||||
use std::io::Write;
|
||||
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
|
||||
use std::thread::{self, JoinHandle};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use super::capture::{self, BYTE_CAP, INTERVAL};
|
||||
use super::recorder::{self, DOMAINS, GAUGES, STEPS};
|
||||
|
||||
/// Owner-side handle to the writer thread.
|
||||
pub(crate) struct Handle {
|
||||
stop_tx: Sender<()>,
|
||||
join: JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl Handle {
|
||||
/// Wake the writer, let it drain once more, and join it.
|
||||
pub(crate) fn stop_and_join(self) {
|
||||
// A send error means the thread already exited (cap stop); joining is
|
||||
// still correct and returns at once.
|
||||
let _ = self.stop_tx.send(());
|
||||
let _ = self.join.join();
|
||||
}
|
||||
}
|
||||
|
||||
/// Start the writer thread on an already-open sink.
|
||||
pub(crate) fn spawn(file: File) -> std::io::Result<Handle> {
|
||||
let (stop_tx, stop_rx) = mpsc::channel();
|
||||
let join = thread::Builder::new()
|
||||
.name("fips-profile".to_string())
|
||||
.spawn(move || run(file, stop_rx))?;
|
||||
Ok(Handle { stop_tx, join })
|
||||
}
|
||||
|
||||
fn run(mut file: File, stop_rx: Receiver<()>) {
|
||||
loop {
|
||||
match stop_rx.recv_timeout(INTERVAL) {
|
||||
// Stop requested, or the owner went away: final drain, then exit.
|
||||
Ok(()) | Err(RecvTimeoutError::Disconnected) => {
|
||||
let _ = flush(&mut file);
|
||||
let _ = file.flush();
|
||||
return;
|
||||
}
|
||||
Err(RecvTimeoutError::Timeout) => {
|
||||
if flush(&mut file).is_err() {
|
||||
// The sink is gone or full; stop the capture rather than
|
||||
// spinning on a broken file for the rest of the run.
|
||||
let _ = note(&mut file, "capture stopped: write error");
|
||||
// The trailer above goes to the same failing file, so it is
|
||||
// not a signal that survives. `stop` and a subsequent `on`
|
||||
// both clear the state without surfacing it, so an operator
|
||||
// would otherwise never learn the window was truncated.
|
||||
tracing::warn!(
|
||||
target: "fips::instr",
|
||||
"profile capture stopped: write error on the sink"
|
||||
);
|
||||
capture::mark_stopped(capture::STOPPED_BY_ERROR);
|
||||
return;
|
||||
}
|
||||
if capture::bytes_written() >= BYTE_CAP {
|
||||
let _ = note(
|
||||
&mut file,
|
||||
&format!("capture stopped: byte cap {BYTE_CAP} reached"),
|
||||
);
|
||||
let _ = file.flush();
|
||||
capture::mark_stopped(capture::STOPPED_BY_CAP);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Append a `#`-prefixed trailer line.
|
||||
fn note(file: &mut File, text: &str) -> std::io::Result<()> {
|
||||
let line = format!("# {text}\n");
|
||||
file.write_all(line.as_bytes())?;
|
||||
capture::add_bytes(line.len() as u64);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Emit one interval: every step of every domain, then the gauges.
|
||||
///
|
||||
/// Every emitted step gets a row every interval, including zero-count rows, so
|
||||
/// "this step did not run" is visible rather than absent. The two steps whose
|
||||
/// call sites are conditionally compiled are excluded in builds that do not
|
||||
/// have them, so no row is structurally zero forever.
|
||||
fn flush(file: &mut File) -> std::io::Result<()> {
|
||||
let ts = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0);
|
||||
|
||||
let mut out = String::with_capacity(4096);
|
||||
for domain in DOMAINS {
|
||||
for step in STEPS {
|
||||
if !step.emitted() {
|
||||
continue;
|
||||
}
|
||||
let (count, max_ns, total_ns) = recorder::take_step(domain, step);
|
||||
out.push_str(&format!(
|
||||
"{ts}\tstep\t{domain}\t{name}\t{count}\t{max}\t{total}\tus\n",
|
||||
domain = domain.name(),
|
||||
name = step.name(),
|
||||
max = max_ns / 1_000,
|
||||
total = total_ns / 1_000,
|
||||
));
|
||||
}
|
||||
}
|
||||
// Gauges carry the tick domain today; the row kind and the unit column keep
|
||||
// them distinguishable from the duration rows above.
|
||||
for gauge in GAUGES {
|
||||
let (count, mut max, mut total) = recorder::take_gauge(gauge);
|
||||
if gauge.is_duration() {
|
||||
max /= 1_000;
|
||||
total /= 1_000;
|
||||
}
|
||||
out.push_str(&format!(
|
||||
"{ts}\tgauge\t{domain}\t{name}\t{count}\t{max}\t{total}\t{unit}\n",
|
||||
domain = recorder::Domain::Tick.name(),
|
||||
name = gauge.name(),
|
||||
unit = gauge.unit(),
|
||||
));
|
||||
}
|
||||
|
||||
file.write_all(out.as_bytes())?;
|
||||
capture::add_bytes(out.len() as u64);
|
||||
Ok(())
|
||||
}
|
||||
@@ -15,6 +15,11 @@ pub mod control;
|
||||
#[cfg(target_os = "linux")]
|
||||
pub mod gateway;
|
||||
pub mod identity;
|
||||
// Declared before `node` (and named to sort there) because it carries
|
||||
// `#[macro_use]`: the tick instrumentation macro must be in scope for the
|
||||
// modules that follow.
|
||||
#[macro_use]
|
||||
pub(crate) mod instr;
|
||||
pub mod mdns;
|
||||
pub mod node;
|
||||
pub mod noise;
|
||||
|
||||
@@ -107,8 +107,8 @@ impl Node {
|
||||
}
|
||||
};
|
||||
|
||||
let mut tick =
|
||||
tokio::time::interval(Duration::from_secs(self.config().node.tick_interval_secs));
|
||||
let tick_period = Duration::from_secs(self.config().node.tick_interval_secs);
|
||||
let mut tick = tokio::time::interval(tick_period);
|
||||
|
||||
// Set up control socket channel
|
||||
let (control_tx, mut control_rx) =
|
||||
@@ -319,46 +319,93 @@ impl Node {
|
||||
).await;
|
||||
let _ = response_tx.send(response);
|
||||
}
|
||||
_ = tick.tick() => {
|
||||
self.check_timeouts();
|
||||
let now_ms = Self::now_ms();
|
||||
self.reload_peer_acl().await;
|
||||
// The host map hot-reloads on the same tick as the ACL. It
|
||||
// is polled separately from `reload_peer_acl` because the
|
||||
// ACL's embedded alias reloader and this snapshot are
|
||||
// distinct resources; the `path_mtu_lookup` cache and the
|
||||
// `nostr_rendezvous` subsystem are deliberately excluded
|
||||
// from `Reloadable` since neither reloads from a backing
|
||||
// file (see `node::reloadable`).
|
||||
self.reload_host_map().await;
|
||||
self.poll_pending_connects().await;
|
||||
self.poll_nostr_rendezvous().await;
|
||||
self.poll_lan_rendezvous().await;
|
||||
self.drive_peer_timers(now_ms).await;
|
||||
self.resend_pending_rekeys(now_ms).await;
|
||||
self.resend_pending_session_handshakes(now_ms).await;
|
||||
self.resend_pending_session_msg3(now_ms).await;
|
||||
self.purge_idle_sessions(now_ms);
|
||||
self.process_pending_retries(now_ms).await;
|
||||
self.check_tree_state().await;
|
||||
self.check_bloom_state().await;
|
||||
self.compute_mesh_size();
|
||||
self.record_stats_history();
|
||||
self.check_mmp_reports().await;
|
||||
self.check_session_mmp_reports().await;
|
||||
self.check_link_heartbeats().await;
|
||||
self.check_rekey().await;
|
||||
self.check_session_rekey().await;
|
||||
self.check_pending_lookups(now_ms).await;
|
||||
self.poll_transport_discovery().await;
|
||||
self.sample_transport_congestion();
|
||||
#[cfg(any(target_os = "linux", target_os = "macos"))]
|
||||
self.activate_connected_udp_sessions().await;
|
||||
// Debug-build sweep of the peer-lifecycle map invariant
|
||||
// (leaked machines / machine-less legs); two map scans,
|
||||
// compiled out of release builds.
|
||||
#[cfg(debug_assertions)]
|
||||
self.debug_assert_peer_maps_coherent();
|
||||
deadline = tick.tick() => {
|
||||
// Tick-body instrumentation. The gate is read ONCE per tick
|
||||
// into `instr_on`, which is then passed explicitly to every
|
||||
// `instr_step!` invocation — macro hygiene makes a call-site
|
||||
// local invisible inside the macro body. With the
|
||||
// `profiling` feature off, `gate()` is a `const fn`
|
||||
// returning false and the macro is a pure pass-through, so
|
||||
// the whole arm compiles to the uninstrumented sequence.
|
||||
//
|
||||
// `tick_entry` records how late this entry is against the
|
||||
// deadline the interval scheduled it for. That is the
|
||||
// measurement this instrumentation exists for: the arm is
|
||||
// polled LAST under `biased;`, so the
|
||||
// lateness IS the time it spent waiting behind the packet,
|
||||
// TUN and control arms. `tick()` hands back its scheduled
|
||||
// deadline, so this is a subtraction rather than a model.
|
||||
// The whole-tick span below measures the body alone.
|
||||
let instr_on = crate::instr::gate();
|
||||
crate::instr::tick_entry(instr_on, deadline.into_std(), std::time::Instant::now());
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::WholeTick, {
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckTimeouts,
|
||||
self.check_timeouts());
|
||||
let now_ms = Self::now_ms();
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ReloadPeerAcl,
|
||||
self.reload_peer_acl().await);
|
||||
// The host map hot-reloads on the same tick as the ACL. It
|
||||
// is polled separately from `reload_peer_acl` because the
|
||||
// ACL's embedded alias reloader and this snapshot are
|
||||
// distinct resources; the `path_mtu_lookup` cache and the
|
||||
// `nostr_rendezvous` subsystem are deliberately excluded
|
||||
// from `Reloadable` since neither reloads from a backing
|
||||
// file (see `node::reloadable`).
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ReloadHostMap,
|
||||
self.reload_host_map().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollPendingConnects,
|
||||
self.poll_pending_connects().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollNostrRendezvous,
|
||||
self.poll_nostr_rendezvous().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollLanRendezvous,
|
||||
self.poll_lan_rendezvous().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::DrivePeerTimers,
|
||||
self.drive_peer_timers(now_ms).await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingRekeys,
|
||||
self.resend_pending_rekeys(now_ms).await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingSessionHandshakes,
|
||||
self.resend_pending_session_handshakes(now_ms).await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ResendPendingSessionMsg3,
|
||||
self.resend_pending_session_msg3(now_ms).await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PurgeIdleSessions,
|
||||
self.purge_idle_sessions(now_ms));
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ProcessPendingRetries,
|
||||
self.process_pending_retries(now_ms).await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckTreeState,
|
||||
self.check_tree_state().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckBloomState,
|
||||
self.check_bloom_state().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ComputeMeshSize,
|
||||
self.compute_mesh_size());
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::RecordStatsHistory,
|
||||
self.record_stats_history());
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckMmpReports,
|
||||
self.check_mmp_reports().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckSessionMmpReports,
|
||||
self.check_session_mmp_reports().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckLinkHeartbeats,
|
||||
self.check_link_heartbeats().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckRekey,
|
||||
self.check_rekey().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckSessionRekey,
|
||||
self.check_session_rekey().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::CheckPendingLookups,
|
||||
self.check_pending_lookups(now_ms).await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::PollTransportDiscovery,
|
||||
self.poll_transport_discovery().await);
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::SampleTransportCongestion,
|
||||
self.sample_transport_congestion());
|
||||
#[cfg(any(target_os = "linux", target_os = "macos"))]
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::ActivateConnectedUdpSessions,
|
||||
self.activate_connected_udp_sessions().await);
|
||||
// Debug-build sweep of the peer-lifecycle map invariant
|
||||
// (leaked machines / machine-less legs); two map scans,
|
||||
// compiled out of release builds.
|
||||
#[cfg(debug_assertions)]
|
||||
instr_step!(instr_on, crate::instr::Domain::Tick, crate::instr::Step::DebugAssertPeerMapsCoherent,
|
||||
self.debug_assert_peer_maps_coherent());
|
||||
});
|
||||
crate::instr::tick_gauges(instr_on, self.peers.len() as u64);
|
||||
}
|
||||
// Shutdown signal → enter the bounded drain in place, ONCE.
|
||||
// Gated on `is_none()` so it only fires while serving; after
|
||||
|
||||
@@ -1985,6 +1985,10 @@ impl Node {
|
||||
// which the helper emits at the Dns→rest seam.
|
||||
self.execute_teardown(actions, true).await;
|
||||
|
||||
// Stop and join the profile writer thread if a capture is armed.
|
||||
// Idempotent, and compiled away in a default build.
|
||||
crate::instr::shutdown();
|
||||
|
||||
self.supervisor.state = NodeState::Stopped;
|
||||
info!(state = %self.supervisor.state, "Node stopped");
|
||||
Ok(())
|
||||
@@ -2239,6 +2243,7 @@ impl Node {
|
||||
info!(state = %self.supervisor.state, "Node stopping (drain complete)");
|
||||
let stop_actions = self.supervisor.fsm.step(Event::DrainDeadlineElapsed);
|
||||
self.execute_teardown(stop_actions, false).await;
|
||||
crate::instr::shutdown();
|
||||
self.supervisor.state = NodeState::Stopped;
|
||||
info!(state = %self.supervisor.state, "Node stopped");
|
||||
} else if let Err(e) = self.stop().await {
|
||||
|
||||
@@ -521,6 +521,30 @@ run_build() {
|
||||
return 1
|
||||
fi
|
||||
|
||||
# An optional feature means two source trees, and --all-features lints only
|
||||
# one of them. The default build is what ships, so lint it explicitly:
|
||||
# without this stage, code that compiles only with `profiling` enabled would
|
||||
# pass CI while breaking every release build.
|
||||
info "cargo clippy --all-targets -- -D warnings (default features)"
|
||||
if cargo clippy --all-targets -- -D warnings 2>&1; then
|
||||
record "clippy-default-features" 0
|
||||
else
|
||||
record "clippy-default-features" 1
|
||||
return 1
|
||||
fi
|
||||
|
||||
# Tick-body profiler: the feature-on tree must build too. Added alongside
|
||||
# the default-feature stages rather than replacing them. Mirrored in
|
||||
# .github/workflows/ci.yml — check-ci-parity.sh compares integration suites
|
||||
# only and will not catch a stage added to one runner and not the other.
|
||||
info "cargo build --workspace --features profiling"
|
||||
if cargo build --workspace --features profiling 2>&1; then
|
||||
record "build-profiling" 0
|
||||
else
|
||||
record "build-profiling" 1
|
||||
return 1
|
||||
fi
|
||||
|
||||
# Guard: the effectively-immutable state lives solely in NodeContext. The
|
||||
# Node struct must not re-declare a bundled field (config/identity/
|
||||
# startup_epoch/started_at/is_leaf_only/node_profile/max_*) — a shadow field
|
||||
@@ -560,6 +584,17 @@ run_tests() {
|
||||
record "unit-tests" 1
|
||||
fi
|
||||
fi
|
||||
|
||||
# The `profiling` feature adds a module, a recorder and a writer thread that
|
||||
# the default-feature run above never compiles. Run the library tests once
|
||||
# more with it on so its own tests execute at all. Mirrored in
|
||||
# .github/workflows/ci.yml.
|
||||
info "cargo test --lib --features profiling"
|
||||
if cargo test --lib --features profiling 2>&1; then
|
||||
record "unit-tests-profiling" 0
|
||||
else
|
||||
record "unit-tests-profiling" 1
|
||||
fi
|
||||
}
|
||||
|
||||
# ── Stage 3: Integration Tests ─────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user