diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 65f9001a4..199d9d070 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -95,7 +95,13 @@ jobs: - name: Install Whitaker uses: leynos/shared-actions/.github/actions/install-whitaker@8e6a325765ba382691d1b1c65fcccf1fc1d1b00c - name: Lint - run: make lint + # Dylint 6.0.1 expects a local master branch in its rust-clippy clone. + # Limit the compatibility wrapper to this step; all other Git calls + # pass through to the runner's normal Git executable. + run: | + export DYLINT_REAL_GIT="$(command -v git)" + export PATH="$GITHUB_WORKSPACE/scripts/dylint-git:$PATH" + make lint - name: Setup uv uses: astral-sh/setup-uv@18d369b23da25a81e9458561248e1aafff6a70fa with: @@ -134,6 +140,8 @@ jobs: # about whether the gate ran. - name: Feature-gate compile checks run: make test-feature-gate + - name: Pool without metrics compile check + run: make test-pool-no-metrics # Coverage runs on pull requests only and drives the ratchet gate. # Main-branch coverage — the CodeScene upload plus the authoritative # ratchet baseline every pull request compares against — is produced diff --git a/Makefile b/Makefile index 6302483bf..7ccf86267 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: help all clean test test-doc test-ui test-feature-gate bless-ui test-workflow-contracts test-loom doctest-benchmark +.PHONY: help all clean test test-doc test-ui test-feature-gate test-pool-no-metrics bless-ui test-workflow-contracts test-loom doctest-benchmark .PHONY: bench-codec bench-runtime-ownership bench build dev-build dev-test release lint fmt check-fmt markdownlint nixie typecheck .PHONY: spelling .PHONY: install-kani check-kani-version install-verus run-verus test-verification kani \ @@ -123,6 +123,9 @@ test-ui: ## Run trybuild compile-time API tests test-feature-gate: ## Verify feature-gated APIs are actually absent when disabled ./scripts/check-feature-gate.sh +test-pool-no-metrics: ## Check the pool builds without metrics + RUSTFLAGS="$(DEV_WARNING_FLAGS)" $(CARGO) $(DEV_FAST) check --lib --no-default-features --features pool,serializer-bincode $(BUILD_JOBS) + # Developer-only: review every changed snapshot; this target must not run in CI. bless-ui: ## Refresh trybuild snapshots for deliberate fixture changes TRYBUILD=overwrite $(CARGO) test --test compile_error diff --git a/benches/runtime_ownership_baselines/scenarios.rs b/benches/runtime_ownership_baselines/scenarios.rs index dd2ce2b15..86d9dac66 100644 --- a/benches/runtime_ownership_baselines/scenarios.rs +++ b/benches/runtime_ownership_baselines/scenarios.rs @@ -20,6 +20,7 @@ use wireframe_testing::runtime_ownership_benchmarks::{ PushWorkload, StartupPath, StartupWorkload, + contended_fifo_pool_workloads, }; /// Criterion sample count for every runtime-ownership group. @@ -245,21 +246,17 @@ pub(crate) fn uncontended_pool_workloads() -> Vec { .collect() } -/// The contended pool workloads reported as bounded observations, not timings. +/// Contended workloads reported as bounded timing observations. /// -/// Both fairness policies run over one pool size: the issue asks for the -/// policies to be compared, not for every policy to be repeated at every size. +/// FIFO runs cover all baseline pool sizes. Round-robin remains the existing +/// size-four comparison so its label and workload stay directly comparable. pub(crate) fn contended_pool_workloads() -> Vec { - [ - AcquisitionMode::ContendedFifo, - AcquisitionMode::ContendedRoundRobin, - ] - .into_iter() - .map(|mode| PoolWorkload { + let mut workloads = contended_fifo_pool_workloads().to_vec(); + workloads.push(PoolWorkload { pool_size: CONTENDED_POOL_SIZE, - mode, + mode: AcquisitionMode::ContendedRoundRobin, contenders: CONTENDED_POOL_CONTENDERS, ..PoolWorkload::default() - }) - .collect() + }); + workloads } diff --git a/benches/runtime_ownership_baselines_alloc/main.rs b/benches/runtime_ownership_baselines_alloc/main.rs index c91cc1a61..d3d2bc638 100644 --- a/benches/runtime_ownership_baselines_alloc/main.rs +++ b/benches/runtime_ownership_baselines_alloc/main.rs @@ -39,14 +39,14 @@ //! //! # Contended pool acquisition //! -//! Contended pool acquisition is counted once, after `Criterion::final_summary`, -//! and reported as a `tracing` event rather than timed. Sustained contention -//! intermittently fails or stalls the pool acquire path — a pre-existing defect -//! recorded in `docs/runtime-ownership-baselines.md` and belonging to ADR 013 / -//! issue #647. Criterion's timing pass runs far more iterations than an -//! allocation window, so it reaches that interleaving almost every run; that is -//! how this binary failed on its first execution. The count itself is unaffected -//! because it comes from a single counted window. +//! FIFO contended pool acquisition is counted once per baseline pool size, after +//! `Criterion::final_summary`, and reported as a `tracing` event rather than +//! timed. Sustained contention intermittently fails or stalls the pool acquire +//! path — a pre-existing defect recorded in `docs/runtime-ownership-baselines.md` +//! and belonging to ADR 013 / issue #647. Criterion's timing pass runs far more +//! iterations than an allocation window, so it reaches that interleaving almost +//! every run; that is how this binary failed on its first execution. Each count +//! comes from one bracketed window. //! //! # Layout //! diff --git a/benches/runtime_ownership_baselines_alloc/pool_bracket.rs b/benches/runtime_ownership_baselines_alloc/pool_bracket.rs index b15fceb1e..80035a27d 100644 --- a/benches/runtime_ownership_baselines_alloc/pool_bracket.rs +++ b/benches/runtime_ownership_baselines_alloc/pool_bracket.rs @@ -13,10 +13,7 @@ use wireframe_testing::{ runtime_ownership_benchmarks::{PoolHarness, PoolWorkload}, }; -use crate::{ - probes::expect_bench, - scopes::{ALLOCATION_ITERATIONS, AllocationWorkloads}, -}; +use crate::{probes::expect_bench, scopes::ALLOCATION_ITERATIONS}; /// Run the pool scope the way the timing pass should replay it. /// @@ -103,7 +100,7 @@ fn pool_bracket_readings( } /// Build the workload the bracket checks measure, at their own parameters. -fn bracket_workload() -> PoolWorkload { AllocationWorkloads::new().pool } +fn bracket_workload() -> PoolWorkload { PoolWorkload::default() } /// Prove the pool scope's bracketed window really excludes pool construction. /// diff --git a/benches/runtime_ownership_baselines_alloc/reporting.rs b/benches/runtime_ownership_baselines_alloc/reporting.rs index fa1fb116b..946a27736 100644 --- a/benches/runtime_ownership_baselines_alloc/reporting.rs +++ b/benches/runtime_ownership_baselines_alloc/reporting.rs @@ -2,15 +2,14 @@ //! //! A count is recorded before its timing pass and emitted under a label whose //! identity does not depend on it, so Criterion's saved results stay comparable -//! between runs. The contended pool scope is published as an event alone: it -//! cannot survive a Criterion timing pass, but the issue still asks for its -//! number. +//! between runs. FIFO contended pool counts are published as events alone: they +//! cannot survive a Criterion timing pass, but the baseline needs their counts. use std::{hint::black_box, time::Duration}; use criterion::{BenchmarkId, Criterion}; use tokio::runtime::Runtime; -use wireframe_testing::runtime_ownership_benchmarks::{AcquisitionMode, PoolWorkload}; +use wireframe_testing::runtime_ownership_benchmarks::contended_fifo_pool_workloads; use crate::{ pool_bracket::count_pool_acquisition_window, @@ -58,26 +57,6 @@ where total } -/// Contended pool workload, recorded as an observation rather than a timing. -/// -/// Contenders must outnumber slots for the pool to be exhausted, and the -/// default `contenders` of 16 clears the default pool size of 4 comfortably. -/// -/// This workload is deliberately absent from `allocation_scopes`: sustained -/// contention intermittently stalls the pool acquire path or fails it with a -/// spurious `disconnected`, and Criterion's timing pass runs far more -/// iterations than the four in an allocation window, so it reaches that -/// interleaving almost every run. Measuring it here once, under a counted -/// window, records the number the issue asks for without handing the scenario -/// to a pass that cannot survive it. See `docs/runtime-ownership-baselines.md`. -fn contended_pool_workload() -> PoolWorkload { - PoolWorkload { - pool_size: 4, - mode: AcquisitionMode::ContendedFifo, - ..PoolWorkload::default() - } -} - /// Emit a successful contended allocation count as an event. /// /// Kept as its own function because a `tracing` macro expands to an @@ -111,7 +90,7 @@ fn report_contended_allocations(label: &str, outcome: Result) { } } -/// Record the contended pool allocation count as a bounded observation. +/// Record FIFO contended allocation counts across the baseline pool sizes. /// /// Run once after Criterion has finished, so a failure is reported rather than /// aborting the benchmark run. The count comes from the same bracketed window the @@ -119,11 +98,11 @@ fn report_contended_allocations(label: &str, outcome: Result) { /// and the timing are dropped, because repetition is what the acquire-path defect /// makes unreliable. See `docs/runtime-ownership-baselines.md`. pub(crate) fn observe_contended_pool_allocations(runtime: &Runtime) { - let workload = contended_pool_workload(); - let label = workload.label(); - let outcome = count_pool_acquisition_window(runtime, workload, ALLOCATION_ITERATIONS); - - report_contended_allocations(&label, outcome); + for workload in contended_fifo_pool_workloads() { + let label = workload.label(); + let outcome = count_pool_acquisition_window(runtime, workload, ALLOCATION_ITERATIONS); + report_contended_allocations(&label, outcome); + } } /// Install a subscriber for the allocation binary's own events only. diff --git a/benches/runtime_ownership_baselines_alloc/scopes.rs b/benches/runtime_ownership_baselines_alloc/scopes.rs index 94d1ace1c..768039486 100644 --- a/benches/runtime_ownership_baselines_alloc/scopes.rs +++ b/benches/runtime_ownership_baselines_alloc/scopes.rs @@ -17,6 +17,7 @@ use wireframe_testing::{ AcquisitionMode, ActorWorkload, CancellationMode, + POOL_SIZES, PoolWorkload, PushScenario, PushWorkload, @@ -128,7 +129,7 @@ impl<'a> AllocationScope<'a> { /// measurement's own counters are dropped: a scope closure's only job is to /// perform the work the counting window observes. Adapting every scope through /// one helper is what keeps [`allocation_scopes`] short enough to read and keeps -/// the six scopes visibly identical in shape. +/// the scopes visibly identical in shape. fn run_measurement( runtime: &Runtime, measurement: impl std::future::Future>, @@ -141,11 +142,10 @@ fn run_measurement( /// The workload parameters this binary records allocations for, by scope. /// -/// Split from [`allocation_scopes`] because the six literal constructions alone -/// push that function past the workspace's line limit. Keeping the parameters in -/// one block also makes them easy to compare at a glance, which is their point: -/// each differs from the timing binary's defaults deliberately, to keep the -/// allocation windows short enough to compare as small integers. +/// Split from [`allocation_scopes`] so its closures stay below the workspace's +/// function-size limit. Keeping the parameters in one block also makes them +/// easy to compare at a glance, which is their point: each differs from the +/// timing binary's defaults deliberately, to keep allocation windows short. pub(crate) struct AllocationWorkloads { /// Actor drain, with cancellation disabled so the window covers the loop. actor: ActorWorkload, @@ -157,12 +157,12 @@ pub(crate) struct AllocationWorkloads { startup_prepared: StartupWorkload, /// The per-connection factory path, which prepares once per connection. startup_per_connection: StartupWorkload, - /// Uncontended acquire/release, the only pool mode safe to time. - pub(crate) pool: PoolWorkload, + /// Uncontended acquire/release at the named baseline pool sizes. + pool_workloads: [PoolWorkload; POOL_SIZES.len()], } impl AllocationWorkloads { - /// Build the six scopes at their allocation-window parameters. + /// Build the scopes at their allocation-window parameters. pub(crate) fn new() -> Self { Self { actor: ActorWorkload { @@ -191,11 +191,11 @@ impl AllocationWorkloads { path: StartupPath::PerConnectionFactory, ..StartupWorkload::default() }, - pool: PoolWorkload { - pool_size: 4, + pool_workloads: POOL_SIZES.map(|pool_size| PoolWorkload { + pool_size, mode: AcquisitionMode::Uncontended, ..PoolWorkload::default() - }, + }), } } } @@ -209,7 +209,7 @@ pub(crate) fn allocation_scopes(runtime: &Runtime) -> Vec> { let workloads = AllocationWorkloads::new(); let iterations = ALLOCATION_ITERATIONS; - vec![ + let mut scopes = vec![ AllocationScope::windowed(workloads.actor.label(), move || { run_measurement(runtime, measure_actor_drain(workloads.actor, iterations)) }), @@ -231,12 +231,18 @@ pub(crate) fn allocation_scopes(runtime: &Runtime) -> Vec> { measure_startup(workloads.startup_per_connection, iterations), ) }), - AllocationScope { - label: workloads.pool.label(), - count: Box::new(move || { - count_pool_acquisition_window(runtime, workloads.pool, iterations) + ]; + scopes.extend( + workloads + .pool_workloads + .into_iter() + .map(|workload| AllocationScope { + label: workload.label(), + count: Box::new(move || { + count_pool_acquisition_window(runtime, workload, iterations) + }), + replay: Box::new(move || run_pool_acquisition(runtime, workload, iterations)), }), - replay: Box::new(move || run_pool_acquisition(runtime, workloads.pool, iterations)), - }, - ] + ); + scopes } diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index 4c8a8aa0e..db68aae16 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -541,12 +541,44 @@ are internal scaffolding removed before this ADR flips to Accepted. ### Phase 4: use scoped permit races -Remove the per-slot `Arc` clones and `'static` boxed slot futures -from the current permit race -([#646](https://github.com/leynos/wireframe/issues/646)). The current code -clones the core once for each slot future. The scoped-race rework deliberately -follows the actor phase so the races are written once against the actor's grant -loop rather than twice. +Use the fixed slot storage and index-based leases introduced by #645 to remove +the per-slot `Arc` clones and `'static` boxed futures from permit +acquisition ([#646](https://github.com/leynos/wireframe/issues/646)). Advance +the round-robin cursor once, probe the rotated slots without allocation, then +use `FuturesUnordered` only when every slot is blocked. Each blocked future +borrows its stable `PoolSlot` and registers a FIFO semaphore waiter when +polled. Dropping the stream cancels losing acquisitions and releases any +permits they acquired before the winner was selected. + +The immediate probe preserves deterministic start-slot behaviour because +`FuturesUnordered` does not promise poll order. It allocates no storage +proportional to pool size when a probe succeeds. A fully blocked race allocates +one `FuturesUnordered` task node per slot. The serial service worker permits at +most one active race per pool scheduler, and the scheduler keeps its existing +fairness policy. + +The service worker retains its `Arc` across `tokio::select!`, so the +acquisition can borrow the core while shutdown and closed-sender branches +remain active. This avoids a per-race or per-slot core clone. + +The race reuses the `rotation_start` and `rotation_indices` helpers from +[#536](https://github.com/leynos/wireframe/issues/536). The iterator exposes +`Clone` so the probe and blocked race share one rotated order after exactly one +cursor advance, without materializing an index collection. The bounded +acquisition metric wrappers introduced alongside those helpers remain around +both immediate and asynchronous selection. + +The scheduler and metrics entry points use their respective `mod.rs` files, +following the module layout established by #798. The metrics-disabled pool +build uses the existing `tokio-util` `io` feature. Validation follows the +repository-pinned Rust toolchain and CI action versions; the historical +baseline captures retain the toolchain metadata from their runs. + +Blocked races record their starts, terminal outcomes, and durations. A guard +records cancellation when the scheduler drops an unfinished acquisition; the +immediate probe does not create a blocked-race observation. The guard borrows a +monotonic clock: production uses `Instant`, and recorder-backed tests inject a +deterministic clock to verify the duration value. ### Phase 5: integrate existing safety work @@ -575,11 +607,18 @@ Complete or coordinate the following, plus any remaining useful scope from rather than passing. - Dropping all public handles without `close` eventually releases the pool core, including with the `JoinHandle` held by the non-shared close owner. -- Loom or deterministic state-machine tests cover enqueue, capacity, - cancellation, deregistration, and shutdown races, naming at minimum: the - pending-flag RAII clear when a permit race is dropped mid-`select!`, and a - permit won exactly as its waiter's receiver drops. No loom coverage exists - for the pool today; the only loom suite exercises the push queues. +- Deterministic current-thread tests manually poll pool permit races to cover + registration, simultaneous readiness, winner and loser permit ownership, + outer cancellation, and close. A saturated-core test verifies shutdown + rejects an acquisition before it queues behind slot waiters. This uses + Tokio's actual semaphore and scheduler primitives, whose queue registration + and drop behaviour a downstream Loom build cannot schedule. The Loom suites + cover push queues and connection-gauge transitions; they do not model the + client pool. +- Bounded tests enumerate every non-empty readiness set and release order for + pools of two, three, and four slots. They check winner and loser capacity, + cancellation after release, shutdown after registration, and cancellation + after shutdown, using Tokio's actual semaphores. - A panic injected into the scheduler task surfaces as `SchedulerFailed` with the required log event and metric, not as a silent hang or a clean `PoolClosed`. diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 9a301d64b..2cfcfa9ec 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -331,10 +331,10 @@ boundaries for Epic 635: phased client-pool ownership change. Phase 2 stores fixed slots by value in `PoolCore` and keeps scheduler state in `PoolSchedulerControl`; pools and handles retain both roots, while leases retain the core, a checked slot - index, a permit, and an optional control handle. The spawn-on-demand - scheduler and per-slot `Arc` permit-race clones remain in this - intermediate phase. The persistent owner task and scoped permit race belong - to later phases. + index, a permit, and an optional control handle. Phase 4 probes the rotated + slots and races borrowed acquisitions when all slots are blocked. The + spawn-on-demand scheduler remains; a persistent owner task is planned for + [#647](https://github.com/leynos/wireframe/issues/647). The first implementation slice is now in place: consuming `WireframeApp::prepare().await` returns an immutable `PreparedApp` or a typed @@ -521,14 +521,32 @@ of two: a wrapped `usize` differs from its true value by a multiple of reducing after the addition would repeat one slot and skip another once the cursor overflowed. An empty pool consumes no cursor value. The synchronous path borrows each slot in rotation order, copies the winning slot's index, and moves -its permit into `PooledClientLease::new`; the asynchronous path boxes one -permit future per index, each of which clones the shared `Arc` rather -than a slot. `src/client/pool/rotation_tests.rs` pins the ordering, the -overflow cases, and the one-increment-per-attempt cursor contract. Every -acquisition attempt also increments `wireframe_pool_acquisitions_total` with -bounded `mode` (`immediate` or `permit_race`) and `outcome` (`granted`, -`exhausted`, `shutdown`, or `error`) labels, so saturation and shutdown are -observable without adding per-slot or caller identifiers. +its permit into `PooledClientLease::new`. The asynchronous path borrows the +core and probes the same rotation without allocation proportional to pool size. +When all slots are blocked, `FuturesUnordered` races borrowed slot futures; +dropping the stream cancels the losers and releases their permits. Each blocked +race allocates a task node per slot. `src/client/pool/rotation_tests.rs` pins +the ordering, the overflow cases, and the one-increment-per-attempt cursor +contract. Synchronous acquisition attempts and completed asynchronous attempts +increment `wireframe_pool_acquisitions_total` with bounded `mode` (`immediate` +or `permit_race`) and `outcome` (`granted`, `exhausted`, `shutdown`, or +`error`) labels. A blocked asynchronous acquisition cancelled before completion +does not increment this counter. Saturation and shutdown remain observable +without adding per-slot or caller identifiers. + +`wireframe_pool_permit_races_started_total` counts blocked race starts. +`wireframe_pool_permit_races_total` counts their `granted`, `shutdown`, +`error`, or `cancelled` terminal outcomes. +`wireframe_pool_permit_race_duration_seconds` records their duration with the +same terminal labels. Compare starts with terminal outcomes to inspect stuck +races, and graph duration and cancellation rates to investigate contention. The +metrics use no slot or caller identifiers; the immediate probe does not start a +blocked-race observation. + +The `tokio-util` dependency enables `io` directly because request streaming uses +`StreamReader`. This also keeps the pool build usable with the `metrics` +feature disabled, without relying on feature unification from the metrics +exporter. Each `SlotConnection` checkout starts armed. `dispatch_on_connection` disarms the guard only after the operation future returns; dropping an armed guard @@ -636,11 +654,20 @@ Use the Makefile targets as the contributor entrypoint for routine validation: `whitaker --all -- --all-targets --all-features`. - `make test` runs the main automated test suite with warnings treated as errors. +- `make test-pool-no-metrics` checks the `pool,serializer-bincode` feature set + with default features disabled and warnings denied. The required `build-test` + CI job runs it in the `Pool without metrics compile check` step. Install Whitaker through the standalone installer described in the [Whitaker user's guide](whitaker-users-guide.md) so local linting matches continuous integration (CI). +The CI lint step uses a scoped Git wrapper for Dylint 6.0.1: its driver build +expects a local `master` branch in its `rust-clippy` clone, whose default +branch is now `main`. The wrapper creates the expected local `master` reference +after that clone and forwards other Git commands unchanged. Remove it when the +Dylint dependency no longer assumes the old default branch. + ### CodeScene coverage baseline The `Coverage (main)` workflow in `.github/workflows/coverage-main.yml` runs on diff --git a/docs/runtime-ownership-baselines.md b/docs/runtime-ownership-baselines.md index 14172c106..834513193 100644 --- a/docs/runtime-ownership-baselines.md +++ b/docs/runtime-ownership-baselines.md @@ -107,8 +107,8 @@ contended burst carries a timeout. Every figure below is a single run on the named machine at the sample budget the binaries pin (20 samples for timings, 10 for allocations). Criterion -reports an interval, not a point; the figure quoted is the middle estimate, -with the interval beside it so the run-to-run spread is visible. +reports a point estimate with confidence-interval bounds; those bounds do not +represent the observed spread between runs. A baseline without its revision is not reproducible, so each capture is named with the commit that produced it. Two captures are recorded here, and a table @@ -647,6 +647,71 @@ eightfold to sixteenfold. Both figures move between captures by more than the correction accounts for, which is the same caution the allocation table states: at this scope, prefer the spawn count. +## Issue #646 permit-race comparison + +Both captures use `make bench-runtime-ownership` on the named machine and the +same pool workloads: 256 uncontended acquire/release operations per benchmark +iteration, 20 Criterion samples, 20 contended bursts, and four iterations for +allocation observations. The pre-change run used the Phase 1 benchmark worktree +based on `ab630169`, with the benchmark scenarios already registered; the +production race still used boxed `select_all` futures. The after-change run used +`d29c51d`. The captures were taken on 2026-10-22. + +The earlier #639 allocation row of 2052 for four slots predates the #645 +`PoolCore` work. This comparison uses post-#645 worktrees on both sides, so its +1407 and 1124 counts describe different revisions from that earlier row and +should not be compared with its same-binary jitter range. + +The uncontended scenarios use the pool's immediate `try_acquire_permit` path, +not the blocked race. Each timing cell gives Criterion's lower confidence +bound, point estimate, and upper confidence bound for the estimated +per-iteration time in microseconds; these are not per-function measurements or +sample extrema. + +| Slots | Before latency (lower / estimate / upper, µs) | After latency (lower / estimate / upper, µs) | Before allocations | After allocations | +| ----- | --------------------------------------------- | -------------------------------------------- | ------------------ | ----------------- | +| 1 | 158.50 / 164.95 / 171.08 | 143.70 / 148.48 / 154.49 | 1222 | 1233 | +| 4 | 138.28 / 146.01 / 152.66 | 120.83 / 124.68 / 128.41 | 1407 | 1124 | +| 16 | 147.24 / 151.21 / 155.15 | 115.07 / 120.78 / 127.88 | 1616 | 1103 | + +Allocation counts cover four iterations of each burst. They include runtime and +benchmark-harness work and do not attribute allocations to +`acquire_slot_permit`. Permit selection allocates nothing when its probe +succeeds; metrics-enabled acquisition recording may allocate a fixed label key. +The captures predate the later acquisition instrumentation, so they do not +measure that cost. When every slot is blocked, `FuturesUnordered` allocates one +task node for each slot future. + +The contended distribution records successful bursts only; `outcomes` reports +successful bursts out of 20. Each distribution cell gives median, p90, and +maximum in microseconds, so the spread remains visible even when failures +remove bursts from the sample. + +| Slots | Contenders | Before outcomes | Before median / p90 / max (µs) | After outcomes | After median / p90 / max (µs) | +| ----- | ---------- | --------------- | ------------------------------ | -------------- | ----------------------------- | +| 1 | 16 | 20/20 | 269.404 / 536.208 / 1371.982 | 20/20 | 400.096 / 432.627 / 496.668 | +| 4 | 16 | 20/20 | 305.135 / 408.506 / 914.004 | 18/20 | 318.435 / 460.577 / 708.211 | +| 16 | 17 | 4/20 | 374.066 / 850.513 / 850.513 | 3/20 | 354.006 / 601.319 / 601.319 | + +The contended allocation observations cover four iterations and include +scheduler and harness allocations. The allocation window runs separately from +the timing bursts; a row has no allocation count only when its own +four-iteration allocation window fails. + +| Slots | Contenders | Before allocations | After allocations | +| ----- | ---------- | ------------------ | ----------------- | +| 1 | 16 | 1839 | 2074 | +| 4 | 16 | 2177 | 1440 | +| 16 | 17 | unavailable | unavailable | + +The 16-slot contended run failed with `connection closed by peer` before and +after the change, so its latency cells describe only a small successful subset +and its allocation cell is unavailable. The same error appeared in two of 20 +size-4 bursts after the change. This slice therefore does not establish a +change in the spurious-disconnect rate described below. Treat the contended +latencies and process-global allocation counts as workload observations, not +per-function attribution or proof of a throughput change. + ## Known defects in contended pool acquisition Capturing the pool baseline surfaced two pre-existing defects in the acquire diff --git a/docs/users-guide.md b/docs/users-guide.md index 4c1bf9222..5006bc74f 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -1934,12 +1934,14 @@ acquisition waits for a permit rather than reusing a slot that is already at its #### Client pool metrics -When the `metrics` feature is enabled, every acquisition attempt increments -`wireframe_pool_acquisitions_total`. The metric has two bounded labels, so pool -activity cannot create new time series: +When the `metrics` feature is enabled, synchronous attempts and completed +asynchronous attempts increment `wireframe_pool_acquisitions_total`. An +asynchronous acquisition cancelled while waiting is excluded. The metric has +two bounded labels, so pool activity cannot create new time series: - `mode` records which path made the attempt: `immediate` for the synchronous - fast path, or `permit_race` for the asynchronous race across slots. + fast path, or `permit_race` for asynchronous acquisition, including its + immediate probe and any subsequent race across slots. - `outcome` records the result: `granted`, `exhausted`, `shutdown`, or `error`. `exhausted` only applies to `immediate`; the asynchronous path either gains a @@ -1952,6 +1954,20 @@ rising `outcome="exhausted"` rate means acquisitions are arriving faster than the configured permits can serve them, and a rising `outcome="error"` rate on the `permit_race` mode means slots are failing instead of granting capacity. +When the asynchronous probe finds every slot busy, +`wireframe_pool_permit_races_started_total` counts the start of a blocked race. +`wireframe_pool_permit_races_total` records one terminal outcome: `granted`, +`shutdown`, `error`, or `cancelled` when the acquisition is dropped before +completion. `wireframe_pool_permit_race_duration_seconds` is a histogram of +blocked-race waits in seconds with those terminal outcome labels. Immediate +probes that find capacity do not start a blocked-race observation. + +Compare the rates of starts and terminal outcomes to investigate races that +remain active, and graph duration and cancellation rates to understand +contention. A `shutdown` outcome indicates pool closure; `cancelled` indicates +an acquisition dropped while the pool was still open. These metrics use fixed +labels and do not identify slots or callers. + ### Client preamble exchange The client builder supports an optional preamble exchange before framing diff --git a/docs/wireframe-client-design.md b/docs/wireframe-client-design.md index 9ad3cea18..c401f88b2 100644 --- a/docs/wireframe-client-design.md +++ b/docs/wireframe-client-design.md @@ -685,6 +685,38 @@ sequenceDiagram _Figure 2: Bounded waiter admission and saturation rejection._ +For screen readers: the sequence shows the scheduler asking `PoolCore` to +acquire a slot permit. The core obtains the rotated slot order and probes each +slot. If a permit is available, it returns immediately. If every slot is +blocked, it races futures that borrow the stable slots; the first completed +acquisition returns its slot index and permit, then dropping the stream cancels +the losing acquisitions. The core returns the slot and permit or a disconnected +error. + +```mermaid +sequenceDiagram + participant Scheduler + participant PoolCore + participant PoolSlot + participant FuturesUnordered + + Scheduler->>PoolCore: acquire_slot_permit() + PoolCore->>PoolCore: ordered_slot_indices() + PoolCore->>PoolSlot: try_acquire_permit() for rotated slots + alt permit available + PoolSlot-->>PoolCore: OwnedSemaphorePermit + else every slot blocked + PoolCore->>FuturesUnordered: push borrowed slot acquisitions + FuturesUnordered->>PoolSlot: acquire_permit() for each slot + PoolSlot-->>FuturesUnordered: first available permit + FuturesUnordered-->>PoolCore: slot_index and permit + PoolCore->>FuturesUnordered: drop(pending) + end + PoolCore-->>Scheduler: SlotPermit or disconnected +``` + +_Figure 3: Rotated slot-permit probing and the borrowed blocked-slot race._ + ## Decision record for 17.4.2 - Decision: document only the currently reachable troubleshooting surface for diff --git a/scripts/dylint-git/git b/scripts/dylint-git/git new file mode 100755 index 000000000..763610086 --- /dev/null +++ b/scripts/dylint-git/git @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +# Dylint 6.0.1 expects a local master branch after cloning rust-clippy. +# Its default branch is now main, so create the expected local ref after clone. +set -eu + +if [[ "${1-}" == clone && "${2-}" == https://github.com/rust-lang/rust-clippy ]]; then + destination="${3:?clone destination is required}" + "${DYLINT_REAL_GIT:?}" "$@" + exec "${DYLINT_REAL_GIT:?}" -C "$destination" branch master +fi + +exec "${DYLINT_REAL_GIT:?}" "$@" diff --git a/src/client/pool/client_pool_tests.rs b/src/client/pool/client_pool_tests.rs index 692887441..8c89f1733 100644 --- a/src/client/pool/client_pool_tests.rs +++ b/src/client/pool/client_pool_tests.rs @@ -1,6 +1,8 @@ //! Tests for pool, handle, lease, and shared-core lifetimes. use std::{ + error::Error, + io, sync::{ Arc, atomic::{AtomicUsize, Ordering}, @@ -11,6 +13,7 @@ use std::{ use crate::{ client::pool::{ PoolFairnessPolicy, + PoolTaskChurnObserver, client_pool::{PoolSchedulerControl, WireframeClientPool}, core::PoolCore, lease::PooledClientLease, @@ -47,6 +50,30 @@ fn test_lease(pool: &TestPool) -> PooledClientLease { .expect("the selected slot should resolve in the pool core") } +fn has_unexpected_eof(error: &(dyn Error + 'static)) -> bool { + let mut source = Some(error); + while let Some(error) = source { + if error + .downcast_ref::() + .is_some_and(|io_error| io_error.kind() == io::ErrorKind::UnexpectedEof) + { + return true; + } + source = error.source(); + } + false +} + +async fn wait_for_capacity_wait(observer: &PoolTaskChurnObserver, message: &str) { + tokio::time::timeout(Duration::from_secs(1), async { + while observer.snapshot().capacity_waits == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect(message); +} + #[tokio::test] async fn pool_and_handle_drop_keep_lease_slot_alive_until_final_lease_drop() { let drops = Arc::new(AtomicUsize::new(0)); @@ -79,30 +106,81 @@ async fn close_rejects_new_and_queued_acquisitions_while_an_existing_lease_keeps let mut queued_handle = pool.handle(); let queued_acquisition = tokio::spawn(async move { queued_handle.acquire().await }); - tokio::time::timeout(Duration::from_secs(1), async { - while observer.snapshot().capacity_waits == 0 { - tokio::task::yield_now().await; - } - }) - .await - .expect("the second acquisition should enter the wait path before close"); + wait_for_capacity_wait( + &observer, + "the second acquisition should enter the wait path before close", + ) + .await; pool.close().await; + let new_result = handle.acquire().await; + let Err(new_error) = new_result else { + panic!("a closed pool must reject new leases"); + }; assert!( - handle.acquire().await.is_err(), - "a closed pool must reject new leases" + has_unexpected_eof(&new_error), + "a closed pool should report the disconnected UnexpectedEof error: {new_error}" ); let queued_result = tokio::time::timeout(Duration::from_secs(1), queued_acquisition) .await .expect("pool close should complete the queued acquisition") .expect("the queued acquisition task should complete"); + let Err(queued_error) = queued_result else { + panic!("a queued acquisition must not receive a lease after close"); + }; assert!( - queued_result.is_err(), - "a queued acquisition must not receive a lease after close" + has_unexpected_eof(&queued_error), + "a queued acquisition should report the disconnected UnexpectedEof error: {queued_error}" ); assert!( lease.slot().is_some(), "an existing lease must retain its selected slot through close" ); } + +#[tokio::test] +async fn cancelled_queued_acquisition_releases_capacity_and_core_ownership() { + let drops = Arc::new(AtomicUsize::new(0)); + let pool = test_pool(Some(Arc::clone(&drops))); + let mut owner_handle = pool.handle(); + let lease = test_lease(&pool); + let observer = pool.task_churn_observer(); + let mut queued_handle = pool.handle(); + let queued_acquisition = tokio::spawn(async move { queued_handle.acquire().await }); + + wait_for_capacity_wait( + &observer, + "the acquisition should enter the capacity wait before cancellation", + ) + .await; + + queued_acquisition.abort(); + let cancelled = tokio::time::timeout(Duration::from_secs(1), queued_acquisition) + .await + .expect("the cancelled acquisition task should stop promptly"); + let Err(join_error) = cancelled else { + panic!("aborting the acquisition should cancel its task"); + }; + assert!( + join_error.is_cancelled(), + "the queued acquisition task should have been cancelled" + ); + + drop(lease); + let later_lease = tokio::time::timeout(Duration::from_secs(1), owner_handle.acquire()) + .await + .expect("capacity released after cancellation should wake a fresh acquisition") + .expect("cancellation must not consume the released slot permit"); + drop(later_lease); + drop(owner_handle); + drop(pool); + + tokio::time::timeout(Duration::from_secs(1), async { + while drops.load(Ordering::Acquire) != 2 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the cancelled race must not retain the pool core or its slot"); +} diff --git a/src/client/pool/core.rs b/src/client/pool/core.rs index cba95e071..6ca5d3918 100644 --- a/src/client/pool/core.rs +++ b/src/client/pool/core.rs @@ -2,29 +2,22 @@ use std::{ future::Future, - pin::Pin, - sync::{ - Arc, - atomic::{AtomicBool, AtomicUsize, Ordering}, - }, + sync::atomic::{AtomicBool, AtomicUsize, Ordering}, }; -use futures::future::select_all; +use futures::stream::{FuturesUnordered, StreamExt}; use tokio::sync::{Notify, OwnedSemaphorePermit}; use super::slot::PoolSlot; use crate::{ client::ClientError, - metrics::{PoolAcquisitionMode, PoolAcquisitionOutcome}, + metrics::{PoolAcquisitionMode, PoolAcquisitionOutcome, PoolPermitRaceGuard}, serializer::Serializer, }; /// Result of selecting a slot while retaining its independent admission token. pub(crate) type SlotPermit = (usize, OwnedSemaphorePermit); -/// Owned future used by the existing permit race. -type SlotPermitFuture = Pin> + Send>>; - /// One shared root for the pool's fixed physical slots and shutdown state. pub(crate) struct PoolCore where @@ -138,9 +131,16 @@ where } } - /// Race slot permits and return the first available capacity or error. - pub(crate) async fn acquire_slot_permit(self: &Arc) -> Result { - let result = self.acquire_slot_permit_inner().await; + /// Probe rotated slots before racing borrowed futures over blocked slots. + /// + /// A free slot at the rotated starting point wins before any waiters are + /// registered. If every slot is full, the race futures borrow the stable + /// slot storage; when polled, each future registers one FIFO waiter on its + /// semaphore. The serial service worker runs at most one race at a time for + /// the pool scheduler. + pub(crate) async fn acquire_slot_permit(&self) -> Result { + let mut race = None; + let result = self.acquire_slot_permit_inner(&mut race).await; let outcome = match &result { Ok(_) => PoolAcquisitionOutcome::Granted, Err(_) if self.is_shutdown() || self.slots.is_empty() => { @@ -149,33 +149,35 @@ where Err(_) => PoolAcquisitionOutcome::Error, }; crate::metrics::inc_pool_acquisition(PoolAcquisitionMode::PermitRace, outcome); + if let Some(race) = race { + race.finish(outcome); + } result } - /// Race one owned permit future per slot, keeping the core alive for each. - async fn acquire_slot_permit_inner(self: &Arc) -> Result { + /// Try rotated slots before registering blocked semaphore waiters. + async fn acquire_slot_permit_inner<'a>( + &'a self, + race: &mut Option>, + ) -> Result { if self.is_shutdown() || self.slots.is_empty() { return Err(ClientError::disconnected()); } - let waiters = self - .ordered_slot_indices() - .map(|slot_index| { - // Keep the existing owned-future race in this slice. The - // futures retain the core root instead of individual slots; - // the scoped-race follow-up can remove these per-slot clones. - let core = Arc::clone(self); - Box::pin(async move { - let slot = core - .slot(slot_index) - .ok_or_else(ClientError::disconnected)?; - let permit = slot.acquire_permit().await?; - Ok((slot_index, permit)) - }) as SlotPermitFuture - }) - .collect::>(); - let (result, ..) = select_all(waiters).await; - let (slot_index, permit) = result?; + let ordered_indices = self.ordered_slot_indices(); + let selected = ordered_indices.clone().find_map(|slot_index| { + self.slot(slot_index)? + .try_acquire_permit() + .map(|permit| (slot_index, permit)) + }); + + let (slot_index, permit) = if let Some(selected) = selected { + selected + } else { + *race = Some(PoolPermitRaceGuard::start(&self.shutdown)); + self.acquire_from_blocked_slots(ordered_indices).await? + }; + if self.is_shutdown() { drop(permit); Err(ClientError::disconnected()) @@ -184,7 +186,34 @@ where } } - /// Rotate slot indices without allocating or cloning slot owners. + /// Race borrowed FIFO acquisitions after every immediate probe fails. + async fn acquire_from_blocked_slots( + &self, + ordered_indices: impl Iterator, + ) -> Result { + // FuturesUnordered does not promise poll order, so the caller probes + // first to make a free rotated starting slot deterministic. Each + // future borrows the core's stable slots. Dropping the stream cancels + // losing semaphore acquisitions and releases any assigned permits. + let mut pending = FuturesUnordered::new(); + for slot_index in ordered_indices { + let slot = self + .slot(slot_index) + .ok_or_else(ClientError::disconnected)?; + pending.push(async move { + let permit = slot.acquire_permit().await?; + Ok((slot_index, permit)) + }); + } + let result = pending + .next() + .await + .unwrap_or_else(|| Err(ClientError::disconnected())); + drop(pending); + result + } + + /// Rotate slot indices once for the immediate probe and blocked race. /// /// The cursor is read once per attempt, and only for a non-empty pool, so /// startup rotation stays cheap and an empty pool consumes no cursor value. @@ -193,7 +222,10 @@ where /// `len` only for power-of-two pool sizes. Reducing after the addition /// would therefore repeat one slot and skip another in a three- or /// five-slot pool once the cursor overflowed. - fn ordered_slot_indices(&self) -> impl Iterator + '_ { + /// + /// Each call advances the starting cursor exactly once, even when a slot + /// is selected by the immediate probe. + fn ordered_slot_indices(&self) -> impl Iterator + Clone + '_ { let len = self.slots.len(); let start = if len == 0 { 0 @@ -222,7 +254,7 @@ fn rotation_start(len: usize, cursor: usize) -> usize { /// `start` must be below `len`, which [`rotation_start`] guarantees, so the /// additions cannot overflow; `wrapping_rem` is the lint-permitted spelling of /// the remainder rather than an overflow workaround here. -fn rotation_indices(len: usize, start: usize) -> impl Iterator { +fn rotation_indices(len: usize, start: usize) -> impl Iterator + Clone { (0..len).map(move |offset| start.wrapping_add(offset).wrapping_rem(len.max(1))) } diff --git a/src/client/pool/core_tests/metric_tests.rs b/src/client/pool/core_tests/metric_tests.rs index aa36b50d0..2bbdf8fcb 100644 --- a/src/client/pool/core_tests/metric_tests.rs +++ b/src/client/pool/core_tests/metric_tests.rs @@ -168,3 +168,84 @@ fn expect_pool_acquisition(observability: &mut ObservabilityHandle, mode: &str, ) .expect("the acquisition wrapper should record the expected label pair"); } + +/// A free rotated probe must not start a blocked race. +#[cfg(feature = "metrics")] +#[test] +fn free_async_probe_skips_blocked_race() { + let mut observability = ObservabilityHandle::new(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build a current-thread runtime"); + metrics::with_local_recorder(observability.recorder(), || { + runtime.block_on(async { + let core = test_core(4); + for expected in 0..4 { + let (index, permit) = core.acquire_slot_permit().await.expect("free slot grants"); + assert_eq!(index, expected); + drop(permit); + } + }); + }); + observability.snapshot(); + observability + .assert_counter(crate::metrics::POOL_PERMIT_RACES_STARTED, [], 0) + .expect("free probes must not start a blocked race"); +} + +/// Exercise blocked-race metrics through the real acquisition future. +#[cfg(feature = "metrics")] +#[rstest] +#[case::granted("granted", false)] +#[case::shutdown("shutdown", false)] +#[case::shutdown_drop("shutdown", true)] +#[case::cancelled("cancelled", true)] +#[test] +fn blocked_race_records_terminal_outcome_and_duration( + #[case] outcome: &str, + #[case] drop_pending: bool, +) { + let mut observability = ObservabilityHandle::new(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build a current-thread runtime"); + + metrics::with_local_recorder(observability.recorder(), || { + runtime.block_on(async { + let core = test_core(1); + let mut held = core + .slot(0) + .expect("the only slot should exist") + .try_acquire_permit(); + assert!(held.is_some()); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + if outcome == "shutdown" { + core.shutdown(); + } + if !drop_pending { + drop(held.take()); + let result = acquisition.as_mut().await; + assert_eq!(result.is_ok(), outcome == "granted"); + } + } + drop(held); + }); + }); + observability.snapshot(); + observability + .assert_counter(crate::metrics::POOL_PERMIT_RACES_STARTED, [], 1) + .expect("the blocked race should record its start"); + observability + .assert_counter(crate::metrics::POOL_PERMIT_RACES, [("outcome", outcome)], 1) + .expect("the blocked race should record its terminal outcome"); + observability + .assert_histogram_recorded( + crate::metrics::POOL_PERMIT_RACE_DURATION, + [("outcome", outcome)], + ) + .expect("the blocked race should record its duration"); +} diff --git a/src/client/pool/core_tests/mod.rs b/src/client/pool/core_tests/mod.rs index 538407834..9f63a412a 100644 --- a/src/client/pool/core_tests/mod.rs +++ b/src/client/pool/core_tests/mod.rs @@ -2,9 +2,12 @@ //! //! The suite is split by concern so each file stays reviewable: //! `support` holds the shared cores and fixtures, `sequence_tests` covers slot -//! identity and lifecycle, and `metric_tests` covers the labels the real -//! acquisition wrappers record. +//! identity and lifecycle, `permit_race` covers asynchronous selection and +//! cancellation, and `metric_tests` covers the labels the real acquisition +//! wrappers record. mod metric_tests; +mod permit_race; mod sequence_tests; mod support; +mod transition_model; diff --git a/src/client/pool/core_tests/permit_race.rs b/src/client/pool/core_tests/permit_race.rs new file mode 100644 index 000000000..c29ca7479 --- /dev/null +++ b/src/client/pool/core_tests/permit_race.rs @@ -0,0 +1,314 @@ +//! Rotation, cancellation, and resource-drop coverage for permit selection. + +use std::{sync::Arc, time::Duration}; + +use rstest::rstest; +use tokio::sync::OwnedSemaphorePermit; + +use super::support::{TestCore, four_slot_core, test_core}; + +pub(super) fn hold_all_permits(core: &TestCore) -> Vec { + core.slots + .iter() + .map(|slot| { + slot.try_acquire_permit() + .expect("each slot should have one available permit") + }) + .collect() +} + +fn hold_every_permit_except(core: &TestCore, free_index: usize) -> Vec { + core.slots + .iter() + .enumerate() + .filter(|(slot_index, _)| *slot_index != free_index) + .map(|(_, slot)| { + slot.try_acquire_permit() + .expect("every slot other than the free slot should be held") + }) + .collect() +} + +pub(super) fn assert_all_slots_are_available(core: &TestCore) { + let permits = hold_all_permits(core); + drop(permits); +} + +#[rstest] +#[case::one_slot(1, &[0, 0, 0])] +#[case::two_slots(2, &[0, 1, 0])] +#[case::four_slots(4, &[0, 1, 2, 3, 0])] +#[tokio::test] +async fn immediate_probe_rotates_across_slots( + #[case] slot_count: usize, + #[case] expected: &[usize], +) { + let core = test_core(slot_count); + let selected = expected + .iter() + .map(|_| { + let (slot_index, permit) = core + .try_acquire_permit() + .expect("a free slot should grant an immediate permit"); + drop(permit); + slot_index + }) + .collect::>(); + + assert_eq!(selected.as_slice(), expected); +} + +#[rstest] +#[tokio::test(flavor = "current_thread")] +async fn failed_and_cancelled_acquisitions_each_advance_rotation_once() { + let core = test_core(2); + let held = hold_all_permits(&core); + + assert!(core.try_acquire_permit().is_none()); + drop(held); + + let (after_failed_try, permit) = core + .try_acquire_permit() + .expect("the next immediate probe should find free capacity"); + assert_eq!(after_failed_try, 1); + drop(permit); + + let held = hold_all_permits(&core); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + } + drop(held); + + let (after_cancelled_race, permit) = core + .try_acquire_permit() + .expect("the next immediate probe should find free capacity"); + assert_eq!(after_cancelled_race, 1); + drop(permit); +} + +#[rstest] +#[tokio::test] +async fn empty_core_rejects_both_immediate_and_async_acquisition() { + let core = test_core(0); + + assert!(core.try_acquire_permit().is_none()); + assert!(core.acquire_slot_permit().await.is_err()); +} + +#[rstest] +#[tokio::test(flavor = "current_thread")] +async fn shutdown_rejects_acquisition_before_waiting_for_slot_capacity() { + let core = test_core(2); + let held = hold_all_permits(&core); + core.shutdown(); + + let result = tokio::time::timeout(Duration::from_millis(100), core.acquire_slot_permit()).await; + + assert!(matches!(result, Ok(Err(_)))); + drop(held); + assert_all_slots_are_available(&core); +} + +#[rstest] +#[tokio::test] +async fn async_probe_rotates_across_four_slots(four_slot_core: Arc) { + let mut selected = Vec::new(); + for _ in 0..5 { + let (slot_index, permit) = four_slot_core + .acquire_slot_permit() + .await + .expect("a free slot should grant an immediate async permit"); + drop(permit); + selected.push(slot_index); + } + + assert_eq!(selected, [0, 1, 2, 3, 0]); +} + +#[rstest] +#[tokio::test] +async fn sync_and_async_probes_share_one_rotation_cursor(four_slot_core: Arc) { + let (first, permit) = four_slot_core + .try_acquire_permit() + .expect("the first immediate probe should succeed"); + drop(permit); + let (second, permit) = four_slot_core + .acquire_slot_permit() + .await + .expect("the second immediate probe should succeed"); + drop(permit); + let (third, permit) = four_slot_core + .try_acquire_permit() + .expect("the third immediate probe should succeed"); + drop(permit); + let (fourth, permit) = four_slot_core + .acquire_slot_permit() + .await + .expect("the fourth immediate probe should succeed"); + + assert_eq!([first, second, third, fourth], [0, 1, 2, 3]); + drop(permit); +} + +#[rstest] +#[case::two_first(2, 0)] +#[case::two_second(2, 1)] +#[case::four_first(4, 0)] +#[case::four_second(4, 1)] +#[case::four_third(4, 2)] +#[case::four_fourth(4, 3)] +#[tokio::test(flavor = "current_thread")] +async fn immediate_probe_finds_each_free_slot( + #[case] slot_count: usize, + #[case] free_index: usize, +) { + let core = test_core(slot_count); + let _held = hold_every_permit_except(&core, free_index); + + let (selected, permit) = core + .acquire_slot_permit() + .await + .expect("the only free slot should grant its permit"); + + assert_eq!(selected, free_index); + drop(permit); +} + +#[rstest] +#[case::two_first(2, 0)] +#[case::two_second(2, 1)] +#[case::four_first(4, 0)] +#[case::four_second(4, 1)] +#[case::four_third(4, 2)] +#[case::four_fourth(4, 3)] +#[tokio::test(flavor = "current_thread")] +async fn blocked_race_returns_the_single_slot_released_after_registration( + #[case] slot_count: usize, + #[case] free_index: usize, +) { + let core = test_core(slot_count); + let mut held = hold_all_permits(&core); + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + + drop(held.swap_remove(free_index)); + let (selected, permit) = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .expect("released capacity should wake the blocked race") + .expect("the released slot should grant exactly one permit"); + + assert_eq!(selected, free_index); + assert!( + core.slot(selected) + .expect("the selected slot should exist") + .try_acquire_permit() + .is_none(), + "the winner's permit must remain owned by the acquisition" + ); + drop(permit); + drop(held); + assert_all_slots_are_available(&core); +} + +#[rstest] +#[tokio::test(flavor = "current_thread")] +async fn simultaneous_readiness_returns_one_permit_and_drops_the_loser() { + let core = test_core(2); + let held = hold_all_permits(&core); + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + + drop(held); + let (winner_index, winner_permit) = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .expect("both released slots should wake the race") + .expect("one of the ready slots should win"); + let loser_index = 1_usize.saturating_sub(winner_index); + + assert!( + core.slot(winner_index) + .expect("the winner should be a valid slot") + .try_acquire_permit() + .is_none(), + "the returned permit must reserve exactly the winning slot" + ); + let loser_permit = core + .slot(loser_index) + .expect("the other slot should exist") + .try_acquire_permit(); + assert!( + loser_permit.is_some(), + "a losing ready future must not retain its acquired permit" + ); + + drop(loser_permit); + drop(winner_permit); + assert_all_slots_are_available(&core); +} + +#[rstest] +#[tokio::test(flavor = "current_thread")] +async fn cancelling_a_blocked_race_leaves_every_slot_usable() { + let core = test_core(4); + let held = hold_all_permits(&core); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + } + + drop(held); + assert_all_slots_are_available(&core); + let (slot_index, permit) = + tokio::time::timeout(Duration::from_secs(1), core.acquire_slot_permit()) + .await + .expect("a fresh acquisition should complete after cancellation") + .expect("cancellation must not lose any slot permit"); + + assert!(slot_index < core.slots.len()); + drop(permit); +} + +/// Exhaust all non-empty readiness sets for small blocked races. +#[rstest] +#[case::two_slots(2)] +#[case::three_slots(3)] +#[case::four_slots(4)] +#[tokio::test(flavor = "current_thread")] +async fn blocked_race_preserves_capacity_for_every_ready_set(#[case] slot_count: usize) { + for ready in 1..(1_usize << slot_count) { + let core = test_core(slot_count); + let mut held = hold_all_permits(&core) + .into_iter() + .map(Some) + .collect::>(); + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + + for (index, permit) in held.iter_mut().enumerate() { + if ready & (1 << index) != 0 { + drop(permit.take()); + } + } + let (winner, permit) = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .expect("a ready slot must wake the race") + .expect("a ready slot must grant a permit"); + assert_ne!(ready & (1 << winner), 0); + + for index in 0..slot_count { + let available = core + .slot(index) + .expect("the model slot should exist") + .try_acquire_permit(); + assert_eq!( + available.is_some(), + index != winner && ready & (1 << index) != 0, + "unexpected capacity for slot {index}, ready set {ready:#b}" + ); + } + drop(permit); + drop(held); + assert_all_slots_are_available(&core); + } +} diff --git a/src/client/pool/core_tests/transition_model.rs b/src/client/pool/core_tests/transition_model.rs new file mode 100644 index 000000000..d81fc4bca --- /dev/null +++ b/src/client/pool/core_tests/transition_model.rs @@ -0,0 +1,298 @@ +//! Bounded release, cancellation, and shutdown schedules for blocked races. + +use std::{future::Future, io::ErrorKind, pin::Pin, task::Poll, time::Duration}; + +use rstest::rstest; + +use super::{ + permit_race::hold_all_permits, + support::{TestCore, test_core}, +}; +use crate::{ + WireframeError, + client::{ClientError, pool::core::SlotPermit}, +}; + +/// Enumerate each ordering of a ready subset without relying on poll order. +fn release_orders(remaining: &[usize], prefix: &mut Vec, result: &mut Vec>) { + if remaining.is_empty() { + result.push(prefix.clone()); + return; + } + for (position, &slot) in remaining.iter().enumerate() { + let rest = remaining + .iter() + .enumerate() + .filter_map(|(index, &value)| (index != position).then_some(value)) + .collect::>(); + prefix.push(slot); + release_orders(&rest, prefix, result); + let _ = prefix.pop(); + } +} + +/// Enumerate every non-empty ready set and each release order for that set. +fn readiness_orders(slot_count: usize) -> Vec<(usize, Vec)> { + let mut schedules = Vec::new(); + for ready in 1..(1_usize << slot_count) { + let slots = (0..slot_count) + .filter(|index| ready & (1 << *index) != 0) + .collect::>(); + let mut orders = Vec::new(); + release_orders(&slots, &mut Vec::new(), &mut orders); + schedules.extend(orders.into_iter().map(|order| (ready, order))); + } + schedules +} + +/// Terminal action after a set of slots has become ready. +#[derive(Clone, Copy, Debug)] +enum Terminal { + /// Let one ready acquisition complete. + Grant, + /// Drop the outer acquisition before it is polled again. + Cancel, + /// Publish shutdown before polling the ready acquisitions. + Shutdown, + /// Publish shutdown, then drop the still-pending acquisition. + ShutdownCancel, +} + +impl Terminal { + /// Report whether the scenario should return a winning permit. + const fn is_grant(self) -> bool { matches!(self, Self::Grant) } + + /// Report whether cancellation should leave the core open. + const fn is_cancel(self) -> bool { matches!(self, Self::Cancel) } + + /// Report whether shutdown is published before the race is dropped. + const fn is_shutdown(self) -> bool { matches!(self, Self::Shutdown | Self::ShutdownCancel) } + + /// Report whether the shutdown race is polled after capacity is released. + const fn is_inflight_shutdown(self) -> bool { matches!(self, Self::Shutdown) } +} + +/// One bounded readiness and terminal-action scenario. +struct Schedule { + /// Number of stable slots in the core. + slot_count: usize, + /// Bit set of slots released after registration. + ready: usize, + /// Order in which the selected slot permits are released. + order: Vec, + /// Action taken after the selected slots are ready. + terminal: Terminal, +} + +/// Recognize the typed disconnected error used by pool shutdown. +fn check_disconnected(error: ClientError) -> Result<(), String> { + match error { + ClientError::Wireframe(WireframeError::Io(source)) + if source.kind() == ErrorKind::UnexpectedEof => + { + Ok(()) + } + other => Err(format!("expected disconnected error, got {other}")), + } +} + +/// Check every raw slot can grant a permit at the same time. +fn check_all_slots_available(core: &TestCore) -> Result<(), String> { + let permits = core + .slots + .iter() + .map(|slot| { + slot.try_acquire_permit() + .ok_or_else(|| "a slot lost capacity".to_owned()) + }) + .collect::, _>>()?; + drop(permits); + Ok(()) +} + +/// Verify the winner is ready and no losing future retains released capacity. +async fn check_grant( + core: &TestCore, + schedule: &Schedule, + acquisition: Pin<&mut F>, +) -> Result<(), String> +where + F: Future>, +{ + let (winner, permit) = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .map_err(|error| format!("released capacity did not wake the race: {error}"))? + .map_err(|error| format!("a ready slot did not grant a permit: {error}"))?; + if schedule.ready & (1 << winner) == 0 { + return Err(format!( + "winner {winner} was not ready for {:?}", + schedule.order + )); + } + for index in 0..schedule.slot_count { + let available = core + .slot(index) + .ok_or("slot index must exist")? + .try_acquire_permit(); + let expected = index != winner && schedule.ready & (1 << index) != 0; + if available.is_some() != expected { + return Err(format!( + "unexpected capacity for {:?}, order {:?}", + schedule.terminal, schedule.order + )); + } + } + drop(permit); + Ok(()) +} + +/// Verify a ready in-flight race rejects its permit after shutdown. +async fn check_inflight_shutdown(acquisition: Pin<&mut F>) -> Result<(), String> +where + F: Future>, +{ + let result = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .map_err(|error| format!("ready capacity did not wake the shutdown race: {error}"))?; + match result { + Err(error) => check_disconnected(error), + Ok(_) => Err("shutdown granted a permit".to_owned()), + } +} + +/// Apply the terminal action while the pinned acquisition remains in scope. +async fn apply_terminal( + core: &TestCore, + schedule: &Schedule, + acquisition: Pin<&mut F>, +) -> Result<(), String> +where + F: Future>, +{ + if schedule.terminal.is_shutdown() { + core.shutdown(); + } + if schedule.terminal.is_grant() { + return check_grant(core, schedule, acquisition).await; + } + if schedule.terminal.is_inflight_shutdown() { + return check_inflight_shutdown(acquisition).await; + } + Ok(()) +} + +/// Check a later acquisition after cancellation or shutdown. +async fn check_later_acquisition(core: &TestCore, terminal: Terminal) -> Result<(), String> { + if terminal.is_grant() { + return Ok(()); + } + let result = tokio::time::timeout(Duration::from_secs(1), core.acquire_slot_permit()) + .await + .map_err(|error| format!("a later race remained queued: {error}"))?; + if terminal.is_cancel() { + let (_, permit) = + result.map_err(|error| format!("a later race did not acquire capacity: {error}"))?; + drop(permit); + return check_all_slots_available(core); + } + match result { + Err(error) => check_disconnected(error), + Ok(_) => Err("a shut-down core granted a later permit".to_owned()), + } +} + +/// Register one race, release its selected slots, and check the terminal state. +async fn run_schedule(schedule: &Schedule) -> Result<(), String> { + let core = test_core(schedule.slot_count); + let mut held = hold_all_permits(&core) + .into_iter() + .map(Some) + .collect::>(); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + if !matches!(futures::poll!(acquisition.as_mut()), Poll::Pending) { + return Err("a saturated race must register and remain pending".to_owned()); + } + for index in &schedule.order { + drop(held.get_mut(*index).ok_or("slot index must exist")?.take()); + } + apply_terminal(&core, schedule, acquisition.as_mut()).await?; + } + drop(held); + check_all_slots_available(&core)?; + check_later_acquisition(&core, schedule.terminal).await +} + +/// Exercise all non-empty readiness subsets and their release permutations. +#[rstest] +#[case::grant_two(2, Terminal::Grant)] +#[case::grant_three(3, Terminal::Grant)] +#[case::grant_four(4, Terminal::Grant)] +#[case::cancel_two(2, Terminal::Cancel)] +#[case::cancel_three(3, Terminal::Cancel)] +#[case::cancel_four(4, Terminal::Cancel)] +#[case::shutdown_two(2, Terminal::Shutdown)] +#[case::shutdown_three(3, Terminal::Shutdown)] +#[case::shutdown_four(4, Terminal::Shutdown)] +#[case::shutdown_cancel_two(2, Terminal::ShutdownCancel)] +#[case::shutdown_cancel_three(3, Terminal::ShutdownCancel)] +#[case::shutdown_cancel_four(4, Terminal::ShutdownCancel)] +#[tokio::test(flavor = "current_thread")] +async fn blocked_race_transition_schedules_preserve_capacity( + #[case] slot_count: usize, + #[case] terminal: Terminal, +) { + for (ready, order) in readiness_orders(slot_count) { + let schedule = Schedule { + slot_count, + ready, + order, + terminal, + }; + run_schedule(&schedule) + .await + .expect("the bounded race schedule must preserve capacity"); + } +} + +/// Poll after one release while the remaining slot permits stay held. +#[rstest] +#[case(2)] +#[case(4)] +#[tokio::test(flavor = "current_thread")] +async fn first_released_slot_wins_before_later_releases(#[case] slot_count: usize) { + for first in 0..slot_count { + let core = test_core(slot_count); + let mut held = hold_all_permits(&core) + .into_iter() + .map(Some) + .collect::>(); + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + drop(held.get_mut(first).expect("slot permit exists").take()); + let (winner, permit) = tokio::time::timeout(Duration::from_secs(1), acquisition.as_mut()) + .await + .expect("released slot must wake the race") + .expect("released slot grants"); + assert_eq!(winner, first); + assert!( + core.slot(first) + .expect("winner exists") + .try_acquire_permit() + .is_none() + ); + drop(held); + for index in 0..slot_count { + if index != winner { + assert!( + core.slot(index) + .expect("slot exists") + .try_acquire_permit() + .is_some() + ); + } + } + drop(permit); + check_all_slots_available(&core).expect("every slot must retain capacity"); + } +} diff --git a/src/client/pool/scheduler/service.rs b/src/client/pool/scheduler/service.rs index 02bc024c3..6968c79bd 100644 --- a/src/client/pool/scheduler/service.rs +++ b/src/client/pool/scheduler/service.rs @@ -41,9 +41,8 @@ where core: Arc>, control: Arc>, ) { - let permit_race_core = Arc::clone(&core); let result = tokio::select! { - permit_result = permit_race_core.acquire_slot_permit() => { + permit_result = core.acquire_slot_permit() => { permit_result.and_then(|(slot_index, permit)| { if core.is_shutdown() { drop(permit); diff --git a/src/metrics/client_pool.rs b/src/metrics/client_pool.rs index 10e0ccf46..29670920d 100644 --- a/src/metrics/client_pool.rs +++ b/src/metrics/client_pool.rs @@ -2,8 +2,15 @@ // The import is used only by the real (non-loom) increment functions; a // `loom` build compiles the tracing-only test matrix and would see it unused. +#[cfg(all(feature = "pool", not(loom)))] +use std::sync::atomic::AtomicBool; +#[cfg(all(feature = "metrics", feature = "pool", not(loom)))] +use std::sync::atomic::Ordering; +#[cfg(all(feature = "metrics", feature = "pool", not(loom)))] +use std::time::Instant; + #[cfg(all(feature = "metrics", feature = "pool", not(loom)))] -use metrics::counter; +use metrics::{counter, histogram}; /// Name of the counter tracking pooled sockets discarded before safe reuse. /// @@ -64,6 +71,105 @@ pub(crate) fn inc_pool_pending_response_discard() {} /// ``` pub const POOL_ACQUISITIONS: &str = "wireframe_pool_acquisitions_total"; +/// Counter of blocked permit races that have started. +pub const POOL_PERMIT_RACES_STARTED: &str = "wireframe_pool_permit_races_started_total"; + +/// Counter of completed blocked permit races, labelled by `granted`, +/// `shutdown`, `error`, or `cancelled` outcome. +pub const POOL_PERMIT_RACES: &str = "wireframe_pool_permit_races_total"; + +/// Duration of blocked permit races in seconds, labelled by terminal outcome. +pub const POOL_PERMIT_RACE_DURATION: &str = "wireframe_pool_permit_race_duration_seconds"; + +/// Production clock used to measure blocked races. +#[cfg(all(feature = "metrics", feature = "pool", not(loom)))] +static POOL_PERMIT_RACE_CLOCK: fn() -> Instant = Instant::now; + +/// Records cancellation when a blocked race is dropped before completion. +#[cfg(all(feature = "pool", not(loom)))] +pub(crate) struct PoolPermitRaceGuard<'a> { + /// Classify a dropped race as shutdown when pool closure caused it. + shutdown: &'a AtomicBool, + /// Measure the time spent waiting after the probe found no capacity. + #[cfg(feature = "metrics")] + started: Instant, + /// Monotonic time source, borrowed for the duration of the race. + #[cfg(feature = "metrics")] + clock: &'a (dyn Fn() -> Instant + Send + Sync), + /// Prevent the drop handler from reporting a completed race twice. + #[cfg(feature = "metrics")] + finished: bool, +} + +#[cfg(all(feature = "pool", not(loom)))] +impl<'a> PoolPermitRaceGuard<'a> { + /// Start observing a race only after every immediate probe fails. + pub(crate) fn start(shutdown: &'a AtomicBool) -> Self { + #[cfg(feature = "metrics")] + { + Self::start_with_clock(shutdown, &POOL_PERMIT_RACE_CLOCK) + } + #[cfg(not(feature = "metrics"))] + { + Self { shutdown } + } + } + + /// Begin a blocked race using a supplied monotonic clock. + #[cfg(feature = "metrics")] + fn start_with_clock( + shutdown: &'a AtomicBool, + clock: &'a (dyn Fn() -> Instant + Send + Sync), + ) -> Self { + counter!(POOL_PERMIT_RACES_STARTED).increment(1); + Self { + shutdown, + started: clock(), + clock, + finished: false, + } + } + + /// Record a completed race after its shutdown recheck. + #[cfg(feature = "metrics")] + pub(crate) fn finish(mut self, outcome: PoolAcquisitionOutcome) { + self.record(outcome.as_str()); + self.finished = true; + } + + /// Finish the no-op observer when metrics are disabled. + #[cfg(not(feature = "metrics"))] + pub(crate) fn finish(self, _outcome: PoolAcquisitionOutcome) {} + + /// Record one terminal outcome and its elapsed wait. + #[cfg(feature = "metrics")] + fn record(&self, outcome: &'static str) { + counter!(POOL_PERMIT_RACES, "outcome" => outcome).increment(1); + histogram!(POOL_PERMIT_RACE_DURATION, "outcome" => outcome).record( + (self.clock)() + .saturating_duration_since(self.started) + .as_secs_f64(), + ); + } +} + +#[cfg(all(feature = "pool", not(loom)))] +impl Drop for PoolPermitRaceGuard<'_> { + fn drop(&mut self) { + #[cfg(feature = "metrics")] + if !self.finished { + let outcome = if self.shutdown.load(Ordering::Acquire) { + "shutdown" + } else { + "cancelled" + }; + self.record(outcome); + } + #[cfg(not(feature = "metrics"))] + let _ = self.shutdown; + } +} + /// Bounded acquisition paths that decide how a pooled slot is selected. /// /// Gated like [`inc_pool_acquisition`]: under `cfg(loom)` the client module @@ -73,7 +179,9 @@ pub const POOL_ACQUISITIONS: &str = "wireframe_pool_acquisitions_total"; pub(crate) enum PoolAcquisitionMode { /// The synchronous fast path tries each slot once without waiting. Immediate, - /// The asynchronous path races one permit future per slot. + /// The asynchronous path probes rotated slots, then races borrowed permit + /// futures only when all slots are blocked. [`POOL_PERMIT_RACES`] counts + /// the blocked races separately. PermitRace, } @@ -142,14 +250,21 @@ pub(crate) fn inc_pool_acquisition(_mode: PoolAcquisitionMode, _outcome: PoolAcq mod tests { //! Tests for bounded client-pool discard and acquisition labels. + use std::{ + sync::atomic::{AtomicBool, AtomicU64, Ordering}, + time::{Duration, Instant}, + }; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use rstest::rstest; use super::{ POOL_ACQUISITIONS, POOL_CONNECTION_DISCARDS, + POOL_PERMIT_RACE_DURATION, PoolAcquisitionMode, PoolAcquisitionOutcome, + PoolPermitRaceGuard, inc_pool_acquisition, inc_pool_incomplete_operation_discard, inc_pool_pending_response_discard, @@ -226,4 +341,43 @@ mod tests { "acquisition metric for {mode:?}/{outcome:?} was not recorded" ); } + + #[rstest] + #[case::granted(Some(PoolAcquisitionOutcome::Granted), "granted")] + #[case::cancelled(None, "cancelled")] + fn blocked_race_records_duration_from_injected_clock( + #[case] outcome: Option, + #[case] expected_label: &str, + ) { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let shutdown = AtomicBool::new(false); + let elapsed = AtomicU64::new(0); + let origin = Instant::now(); + let clock = || origin + Duration::from_secs(elapsed.load(Ordering::Relaxed)); + + ::metrics::with_local_recorder(&recorder, || { + let guard = PoolPermitRaceGuard::start_with_clock(&shutdown, &clock); + elapsed.store(7, Ordering::Relaxed); + if let Some(outcome) = outcome { + guard.finish(outcome); + } else { + drop(guard); + } + }); + + let found = snapshotter.snapshot().into_vec().iter().any(|(key, _, _, value)| { + key.key().name() == POOL_PERMIT_RACE_DURATION + && key.key().labels().any(|label| { + label.key() == "outcome" && label.value() == expected_label + }) + && matches!(value, DebugValue::Histogram(samples) + if samples.len() == 1 + && samples.first().is_some_and(|sample| sample.0.to_bits() == 7.0_f64.to_bits())) + }); + assert!( + found, + "the histogram must record the injected seven-second duration" + ); + } } diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index f54199d1c..074bf3b5a 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -22,12 +22,23 @@ use metrics::{counter, gauge, histogram}; mod client_pool; #[cfg(all(feature = "pool", not(loom)))] -pub(crate) use client_pool::{PoolAcquisitionMode, PoolAcquisitionOutcome, inc_pool_acquisition}; +pub(crate) use client_pool::{ + PoolAcquisitionMode, + PoolAcquisitionOutcome, + PoolPermitRaceGuard, + inc_pool_acquisition, +}; mod decode_rejections; mod server_startup; mod server_supervisor; -pub use client_pool::{POOL_ACQUISITIONS, POOL_CONNECTION_DISCARDS}; +pub use client_pool::{ + POOL_ACQUISITIONS, + POOL_CONNECTION_DISCARDS, + POOL_PERMIT_RACE_DURATION, + POOL_PERMIT_RACES, + POOL_PERMIT_RACES_STARTED, +}; #[cfg(all(feature = "pool", not(loom)))] pub(crate) use client_pool::{ inc_pool_incomplete_operation_discard, @@ -152,6 +163,7 @@ pub(crate) enum PreparationOutcome { Failure, } +#[cfg(feature = "metrics")] impl PreparationOutcome { /// Return the stable metric label value for this preparation outcome. pub(crate) const fn as_str(self) -> &'static str { @@ -171,6 +183,7 @@ pub enum Direction { Outbound, } +#[cfg(feature = "metrics")] impl Direction { /// Supplies the bounded metric label value, avoiding call-site strings /// that could make inbound and outbound observations incomparable. diff --git a/src/metrics/server_startup.rs b/src/metrics/server_startup.rs index cd6ef4f48..aadcc131f 100644 --- a/src/metrics/server_startup.rs +++ b/src/metrics/server_startup.rs @@ -52,7 +52,7 @@ pub(crate) enum ServerStartupOutcome { Cancelled, } -#[cfg(not(loom))] +#[cfg(all(feature = "metrics", not(loom)))] impl ServerStartupOutcome { /// Return the stable metric label value for this outcome. pub(crate) const fn as_str(self) -> &'static str { diff --git a/tests/workflow_contracts/ci_workflow_test.py b/tests/workflow_contracts/ci_workflow_test.py index c1384a46d..75abd68c2 100644 --- a/tests/workflow_contracts/ci_workflow_test.py +++ b/tests/workflow_contracts/ci_workflow_test.py @@ -256,8 +256,14 @@ def _assert_installed_before_lint(install: dict[str, object]) -> None: workflow = yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8")) steps = cast("list[dict[str, object]]", workflow["jobs"]["build-test"]["steps"]) install_index = next((i for i, step in enumerate(steps) if step is install or step == install), None) + lint_command = ( + 'export DYLINT_REAL_GIT="$(command -v git)"\n' + 'export PATH="$GITHUB_WORKSPACE/scripts/dylint-git:$PATH"\n' + "make lint" + ) lint_index = next( - (i for i, step in enumerate(steps) if step.get("run") == "make lint"), None + (i for i, step in enumerate(steps) if str(step.get("run", "")).strip() == lint_command), + None, ) assert install_index is not None, "install-whitaker must be a build-test step" assert lint_index is not None, "build-test must run `make lint`" diff --git a/tests/workflow_contracts/dylint_git_test.py b/tests/workflow_contracts/dylint_git_test.py new file mode 100644 index 000000000..783c84b12 --- /dev/null +++ b/tests/workflow_contracts/dylint_git_test.py @@ -0,0 +1,87 @@ +"""Verify the CI Git wrapper only changes Dylint's rust-clippy clone.""" + +import json +import os +import subprocess +from pathlib import Path + +import pytest + +WRAPPER = Path(__file__).resolve().parents[2] / "scripts/dylint-git/git" +CLIPPY = "https://github.com/rust-lang/rust-clippy" + + +@pytest.mark.parametrize( + ("arguments", "expected"), + [ + ( + ["clone", CLIPPY, "/tmp/destination", "--quiet"], + [ + ["clone", CLIPPY, "/tmp/destination", "--quiet"], + ["-C", "/tmp/destination", "branch", "master"], + ], + ), + ( + ["clone", "https://github.com/example/other", "/tmp/destination"], + [["clone", "https://github.com/example/other", "/tmp/destination"]], + ), + (["--version"], [["--version"]]), + ], +) +def test_git_wrapper_forwards_commands( + tmp_path: Path, arguments: list[str], expected: list[list[str]] +) -> None: + """Create the legacy ref only for Dylint's exact clone command.""" + fake_git = tmp_path / "git" + fake_git.write_text( + "#!/usr/bin/env python3\nimport json, sys\nprint(json.dumps(sys.argv[1:]))\n", + encoding="utf-8", + ) + fake_git.chmod(0o755) + env = os.environ | {"DYLINT_REAL_GIT": str(fake_git)} + result = subprocess.run( + [str(WRAPPER), *arguments], env=env, check=True, capture_output=True, text=True + ) + assert [json.loads(line) for line in result.stdout.splitlines()] == expected + + +def test_clippy_clone_creates_master_when_default_is_main(tmp_path: Path) -> None: + """A real clone exposes the local branch expected by Dylint.""" + source = tmp_path / "source" + subprocess.run(["git", "init", "-q", "-b", "main", str(source)], check=True) + subprocess.run( + [ + "git", + "-C", + str(source), + "-c", + "user.name=Test", + "-c", + "user.email=test@example.invalid", + "commit", + "--allow-empty", + "-qm", + "initial", + ], + check=True, + ) + fake_git = tmp_path / "git" + fake_git.write_text( + "#!/usr/bin/env python3\n" + "import os, sys\n" + f"args = [({str(source)!r} if arg == {CLIPPY!r} else arg) for arg in sys.argv[1:]]\n" + "os.execvp('git', ['git', *args])\n", + encoding="utf-8", + ) + fake_git.chmod(0o755) + target = tmp_path / "clone" + env = os.environ | {"DYLINT_REAL_GIT": str(fake_git)} + subprocess.run( + [str(WRAPPER), "clone", CLIPPY, str(target), "--quiet"], env=env, check=True + ) + refs = subprocess.check_output( + ["git", "-C", str(target), "show-ref", "--heads"], text=True + ) + assert "refs/heads/main" in refs + assert "refs/heads/master" in refs + subprocess.run(["git", "-C", str(target), "checkout", "-q", "master"], check=True) diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/mod.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/mod.rs index 9294de731..daba018df 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/mod.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/mod.rs @@ -131,4 +131,5 @@ pub use workload::{ POOL_SIZES, PoolMeasurement, PoolWorkload, + contended_fifo_pool_workloads, }; diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs index 77e942f75..8a84bba4b 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs @@ -4,6 +4,7 @@ use crate::runtime_ownership_benchmarks::client_pool_benchmark_support::{ AcquisitionMode, PoolMeasurement, PoolWorkload, + contended_fifo_pool_workloads, measure_pool_acquisition, }; @@ -20,6 +21,33 @@ use crate::runtime_ownership_benchmarks::client_pool_benchmark_support::{ /// serious defect, and is reported as such. const CONTENDED_BURST_ATTEMPTS: usize = 8; +#[test] +fn fifo_baseline_workloads_cover_each_pool_size_and_can_contend() { + let workloads = contended_fifo_pool_workloads(); + + assert_eq!(workloads.map(|workload| workload.pool_size), [1, 4, 16]); + assert_eq!(workloads.map(|workload| workload.contenders), [16, 16, 17]); + assert_eq!( + workloads.map(|workload| workload.mode), + [AcquisitionMode::ContendedFifo; 3] + ); + for workload in workloads { + assert!( + workload.validate().is_ok(), + "{} must exceed its slot count to enter the contended path", + workload.label() + ); + } + assert_eq!( + workloads.map(PoolWorkload::label), + [ + "contended_fifo_pool_1_contenders_16_rounds_8", + "contended_fifo_pool_4_contenders_16_rounds_8", + "contended_fifo_pool_16_contenders_17_rounds_8", + ] + ); +} + /// Run one contended burst, retrying the acquire path's known spurious failures. /// /// Both recorded failure modes — the spurious `disconnected`, and the stall @@ -66,37 +94,38 @@ async fn contended_mode_needs_more_contenders_than_slots() { ); } -/// A contended burst must be *observed* to contend, not assumed to. -/// -/// The burst empties the pool and holds it empty while every contender asks for -/// capacity, so each contender must park: a contended workload with `c` -/// contenders records at least `c` waits before any capacity is returned. That -/// is a floor, not an equality — the contended rounds that follow routinely push -/// it higher — but it is enough to falsify a burst that never actually -/// contended, which is what a start barrier alone fails to rule out. -/// -/// The spawn count is kept beside it as task-churn evidence. It is not evidence -/// of contention: every release notifies the scheduler whether or not anyone is -/// waiting. +/// A successful contended burst must also deliver leases and exercise the scheduler. #[rstest::rstest] -#[case(AcquisitionMode::ContendedFifo)] -#[case(AcquisitionMode::ContendedRoundRobin)] +#[case::one_slot_fifo(1, 16, AcquisitionMode::ContendedFifo)] +#[case::four_slots_fifo(4, 16, AcquisitionMode::ContendedFifo)] +#[case::four_slots_round_robin(4, 16, AcquisitionMode::ContendedRoundRobin)] #[tokio::test] -async fn contended_acquisition_observes_capacity_waits(#[case] mode: AcquisitionMode) { - const CONTENDERS: usize = 16; +async fn contended_acquisition_grants_leases( + #[case] pool_size: usize, + #[case] contenders: usize, + #[case] mode: AcquisitionMode, +) -> Result<(), String> { + assert_contended_grants(pool_size, contenders, mode).await +} + +/// Check successful bursts share the same capacity and progress contract. +async fn assert_contended_grants( + pool_size: usize, + contenders: usize, + mode: AcquisitionMode, +) -> Result<(), String> { let measurement = contended_burst_with_retry(PoolWorkload { - pool_size: 4, + pool_size, mode, acquisitions: 8, - contenders: CONTENDERS, + contenders, }) - .await - .expect("contended measurement failed in every retry attempt"); + .await?; assert!( - measurement.capacity_waits >= u64::try_from(CONTENDERS).unwrap_or(u64::MAX), - "a burst that empties a 4-slot pool and holds it empty must park every one of \ - {CONTENDERS} contenders, but {} waits were recorded", + measurement.capacity_waits >= u64::try_from(contenders).unwrap_or(u64::MAX), + "a burst that empties a {pool_size}-slot pool and holds it empty must park every one of \ + {contenders} contenders, but {} waits were recorded", measurement.capacity_waits ); assert!( @@ -111,4 +140,13 @@ async fn contended_acquisition_observes_capacity_waits(#[case] mode: Acquisition measurement.acquisitions >= 1, "contended contenders must still be granted leases" ); + Ok(()) +} + +/// Keep the size-16 workload available to benchmarks, but not in the test suite +/// until its tracked spurious disconnect is resolved. +#[tokio::test] +#[ignore = "FIXME: flaky 16-slot disconnect: https://github.com/leynos/wireframe/issues/647"] +async fn sixteen_slots_fifo_grants_leases() -> Result<(), String> { + assert_contended_grants(16, 17, AcquisitionMode::ContendedFifo).await } diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs index c670624e2..52e8e1a56 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs @@ -213,6 +213,31 @@ impl PoolWorkload { } } +/// FIFO contended workloads for every pool size used by runtime baselines. +/// +/// The default of sixteen contenders exceeds pool sizes one and four. A +/// 16-slot pool needs one additional contender for the harness to observe a +/// capacity wait, so its label records seventeen. +/// +/// # Examples +/// +/// ``` +/// use wireframe_testing::runtime_ownership_benchmarks::contended_fifo_pool_workloads; +/// +/// let workloads = contended_fifo_pool_workloads(); +/// assert_eq!(workloads[0].pool_size, 1); +/// assert_eq!(workloads[2].contenders, 17); +/// ``` +#[must_use] +pub fn contended_fifo_pool_workloads() -> [PoolWorkload; POOL_SIZES.len()] { + POOL_SIZES.map(|pool_size| PoolWorkload { + pool_size, + mode: AcquisitionMode::ContendedFifo, + contenders: DEFAULT_CONTENDERS.max(pool_size.saturating_add(1)), + ..PoolWorkload::default() + }) +} + /// Result of measuring one pool workload. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct PoolMeasurement { diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/mod.rs b/wireframe_testing/src/runtime_ownership_benchmarks/mod.rs index e58c4b947..ec335562d 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/mod.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/mod.rs @@ -58,6 +58,7 @@ pub use client_pool_benchmark_support::{ PoolHarness, PoolMeasurement, PoolWorkload, + contended_fifo_pool_workloads, measure_contended_latency, measure_pool_acquisition, };