bench(rwlock): add upgradable lock stage timings

.github/workflows/ci.yml

@@ -65,6 +65,8 @@ jobs:
         run: cargo x check
       - name: Smoke benchmarks
         run: cargo x bench -- --test
+      - name: Smoke upgradable-lock stage benchmark
+        run: cargo x bench --bench upgradable_stages -- --test
 
   test:
     name: Run tests

benchmarks/Cargo.toml

@@ -71,5 +71,10 @@ harness = false
 name = "ecosystem"
 path = "benches/ecosystem/main.rs"
 
+[[bench]]
+harness = false
+name = "upgradable_stages"
+path = "benches/upgradable_stages.rs"
+
 [lints]
 workspace = true

benchmarks/README.md

@@ -28,6 +28,8 @@ cargo x bench --no-run
 cargo x bench -- --test
 ```
 
+The `upgradable_stages` target is selected explicitly so it does not interfere with the default benchmark listing. CI smoke-tests it in a separate command.
+
 ## Compare implementations
 
 ```shell
@@ -60,3 +62,7 @@ cargo x bench --bench primitives -- rwlock::upgradable --repeat 5
 ```
 
 The `primitives` target measures Asyncband-only operation cycles for focused before/after comparisons, including ready paths, queued handoffs, wake registration, cancellation, fan-out, and contended tables. Hand-polled cases use a reference-counted waker and exercise bookkeeping without executor scheduling. For `threads=N` cases the workers share a fixture, start at a barrier, and divide a fixed total number of rounds, including the remainder. Smoke mode runs one outer round per worker, preserving inner message batches, topology and correctness checks without making timing claims.
+
+For an upgradable-lock breakdown, run `cargo x bench --bench upgradable_stages`. It reports uncontended acquisition and promotion separately on one reused lock, then measures acquisition while an ordinary reader is held. It also measures promotion with 1, 4, and 16 readers held, promotion while a writer waits behind the reservation, and a later upgradable acquisition queued behind a writer.
+
+For the promotion-with-readers cases, the benchmark queues the upgrade before dropping the readers; those times include releasing the guards and the resulting wake/handoff. The reader vector is allocated before the timed loop and reused; clearing it drops the guards without freeing its backing allocation in the timed interval. The promotion-with-writer case checks that an upgrade holding the reservation completes before the queued writer. The later acquisition includes the writer's handoff. This target manually polls futures with the benchmark's non-scheduling waker, so its timings exclude executor scheduling. Each run reports an empty timing interval using the same clock reads; measured values remain unadjusted and should be treated as diagnostic. Compare runs made with the same toolchain and machine. `cargo x bench --bench upgradable_stages -- --test` runs a single smoke pass without reporting measurements, as in CI. Use `upgradable_read_then_upgrade` in the `primitives` target for the uninstrumented uncontended whole-cycle measurement.

benchmarks/benches/upgradable_stages.rs

@@ -0,0 +1,242 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::hint::black_box;
+use std::pin::pin;
+use std::time::Duration;
+use std::time::Instant;
+
+use asyncband::rwlock::RwLockUpgradableReadGuard;
+use asyncband::rwlock::UpgradableRwLock;
+use benchmarks::support::bench_context;
+use benchmarks::support::poll_pending;
+use benchmarks::support::poll_pinned_ready;
+use benchmarks::support::poll_ready;
+use clap::Parser;
+
+#[derive(Parser)]
+#[command(about = "Time upgradable-lock acquisition and promotion separately.")]
+struct Options {
+    #[arg(long, default_value_t = 2_000_000)]
+    rounds: usize,
+    #[arg(long, default_value_t = 10)]
+    repeat: usize,
+    #[arg(long)]
+    test: bool,
+    #[arg(long = "bench", hide = true)]
+    _bench: bool,
+}
+
+const READER_COUNTS: &[usize] = &[1, 4, 16];
+
+struct Results {
+    empty_interval: Vec<Duration>,
+    acquisition: Vec<Duration>,
+    promotion: Vec<Duration>,
+    acquisition_with_reader: Vec<Duration>,
+    promotion_with_readers: Vec<Vec<Duration>>,
+    promotion_with_waiting_writer: Vec<Duration>,
+    acquisition_after_waiting_writer: Vec<Duration>,
+}
+
+fn measure(rounds: usize, repeat: usize) -> Results {
+    let lock = UpgradableRwLock::new(0usize);
+    let mut context = bench_context();
+    let mut results = Results {
+        empty_interval: Vec::with_capacity(repeat),
+        acquisition: Vec::with_capacity(repeat),
+        promotion: Vec::with_capacity(repeat),
+        acquisition_with_reader: Vec::with_capacity(repeat),
+        promotion_with_readers: READER_COUNTS
+            .iter()
+            .map(|_| Vec::with_capacity(repeat))
+            .collect(),
+        promotion_with_waiting_writer: Vec::with_capacity(repeat),
+        acquisition_after_waiting_writer: Vec::with_capacity(repeat),
+    };
+    let mut readers = Vec::with_capacity(*READER_COUNTS.last().unwrap());
+
+    for _ in 0..repeat {
+        let mut empty_interval = Duration::ZERO;
+        let mut acquisition = Duration::ZERO;
+        let mut promotion = Duration::ZERO;
+        let mut acquisition_with_reader = Duration::ZERO;
+        let mut promotion_with_readers = vec![Duration::ZERO; READER_COUNTS.len()];
+        let mut promotion_with_waiting_writer = Duration::ZERO;
+        let mut acquisition_after_waiting_writer = Duration::ZERO;
+
+        for _ in 0..rounds {
+            let start = Instant::now();
+            empty_interval += start.elapsed();
+
+            let start = Instant::now();
+            let guard = poll_ready(lock.upgradable_read(), &mut context);
+            acquisition += start.elapsed();
+
+            let start = Instant::now();
+            let writer = poll_ready(RwLockUpgradableReadGuard::upgrade(guard), &mut context);
+            promotion += start.elapsed();
+
+            black_box(*writer);
+            drop(writer);
+
+            let reader = poll_ready(lock.read(), &mut context);
+            let start = Instant::now();
+            let guard = poll_ready(lock.upgradable_read(), &mut context);
+            acquisition_with_reader += start.elapsed();
+            black_box(*guard);
+            drop(guard);
+            drop(reader);
+
+            for (index, &reader_count) in READER_COUNTS.iter().enumerate() {
+                let guard = poll_ready(lock.upgradable_read(), &mut context);
+                for _ in 0..reader_count {
+                    readers.push(poll_ready(lock.read(), &mut context));
+                }
+                let mut upgrade = pin!(RwLockUpgradableReadGuard::upgrade(guard));
+                let start = Instant::now();
+                poll_pending(upgrade.as_mut(), &mut context);
+                readers.clear();
+                let writer = poll_pinned_ready(upgrade.as_mut(), &mut context);
+                promotion_with_readers[index] += start.elapsed();
+                black_box(*writer);
+                drop(writer);
+            }
+
+            let guard = poll_ready(lock.upgradable_read(), &mut context);
+            let mut queued_writer = pin!(lock.write());
+            poll_pending(queued_writer.as_mut(), &mut context);
+            let start = Instant::now();
+            let writer = poll_ready(RwLockUpgradableReadGuard::upgrade(guard), &mut context);
+            promotion_with_waiting_writer += start.elapsed();
+            black_box(*writer);
+            drop(writer);
+            drop(poll_pinned_ready(queued_writer.as_mut(), &mut context));
+
+            let first_guard = poll_ready(lock.upgradable_read(), &mut context);
+            let mut queued_writer = pin!(lock.write());
+            poll_pending(queued_writer.as_mut(), &mut context);
+            let mut queued_upgradable = pin!(lock.upgradable_read());
+            poll_pending(queued_upgradable.as_mut(), &mut context);
+            let start = Instant::now();
+            drop(first_guard);
+            drop(poll_pinned_ready(queued_writer.as_mut(), &mut context));
+            let guard = poll_pinned_ready(queued_upgradable.as_mut(), &mut context);
+            acquisition_after_waiting_writer += start.elapsed();
+            black_box(*guard);
+            drop(guard);
+        }
+
+        results.empty_interval.push(empty_interval);
+        results.acquisition.push(acquisition);
+        results.promotion.push(promotion);
+        results
+            .acquisition_with_reader
+            .push(acquisition_with_reader);
+        for (samples, elapsed) in results
+            .promotion_with_readers
+            .iter_mut()
+            .zip(promotion_with_readers)
+        {
+            samples.push(elapsed);
+        }
+        results
+            .promotion_with_waiting_writer
+            .push(promotion_with_waiting_writer);
+        results
+            .acquisition_after_waiting_writer
+            .push(acquisition_after_waiting_writer);
+    }
+
+    results
+}
+
+fn report(name: &str, rounds: usize, mut samples: Vec<Duration>) {
+    samples.sort_unstable();
+    let middle = samples.len() / 2;
+    let median = if samples.len() % 2 == 0 {
+        let lower = samples[middle - 1];
+        lower + (samples[middle] - lower) / 2
+    } else {
+        samples[middle]
+    };
+    let minimum = samples.first().unwrap();
+    let maximum = samples.last().unwrap();
+    let ns_per_round = median.as_nanos() as f64 / rounds as f64;
+    let runs = samples
+        .iter()
+        .map(|sample| format!("{:.3}", sample.as_secs_f64() * 1_000.0))
+        .collect::<Vec<_>>()
+        .join(", ");
+
+    println!(
+        "{name:<24} median {:>8.2} ns/round  ({:.2}..{:.2} ms; runs (ms): [{runs}])",
+        ns_per_round,
+        minimum.as_secs_f64() * 1_000.0,
+        maximum.as_secs_f64() * 1_000.0,
+    );
+}
+
+fn main() {
+    let options = Options::parse();
+    if options.test {
+        let _ = measure(1, 1);
+        println!("ok  upgradable-lock stage benchmark smoke checks");
+        return;
+    }
+    assert!(options.rounds > 0, "rounds must be positive");
+    assert!(options.repeat > 0, "repeat must be positive");
+
+    let results = measure(options.rounds, options.repeat);
+    println!(
+        "UpgradableRwLock, {} rounds per run, {} runs",
+        options.rounds, options.repeat
+    );
+    println!("The clock reads add measurable overhead; treat these numbers as diagnostic.");
+    println!(
+        "The empty interval uses the same clock reads and is reported without subtracting it."
+    );
+    report(
+        "empty timing interval",
+        options.rounds,
+        results.empty_interval,
+    );
+    report("acquisition", options.rounds, results.acquisition);
+    report("promotion", options.rounds, results.promotion);
+    report(
+        "acquisition + reader held",
+        options.rounds,
+        results.acquisition_with_reader,
+    );
+    for (&reader_count, samples) in READER_COUNTS.iter().zip(results.promotion_with_readers) {
+        report(
+            &format!("promotion + {reader_count} readers"),
+            options.rounds,
+            samples,
+        );
+    }
+    report(
+        "promotion + writer queued",
+        options.rounds,
+        results.promotion_with_waiting_writer,
+    );
+    report(
+        "acquire behind queued writer",
+        options.rounds,
+        results.acquisition_after_waiting_writer,
+    );
+}

xtask/src/main.rs

@@ -74,7 +74,7 @@ struct CommandBench {
 
     #[arg(
         long,
-        help = "Select a target (ecosystem or primitives; default: all)."
+        help = "Select a target (ecosystem, primitives, or upgradable_stages; default: ecosystem and primitives)."
     )]
     bench: Option<String>,
 
@@ -88,11 +88,9 @@ struct CommandBench {
 
 impl CommandBench {
     fn run(self) {
-        run_command(make_bench_cmd(
-            self.no_run,
-            self.bench.as_deref(),
-            &self.args,
-        ));
+        for cmd in make_bench_cmds(self.no_run, self.bench.as_deref(), &self.args) {
+            run_command(cmd);
+        }
     }
 }
 
@@ -394,22 +392,23 @@ fn classify_release_type(baseline: &Version, release: &Version) -> SemverRelease
     }
 }
 
-fn make_bench_cmd(no_run: bool, bench: Option<&str>, args: &[String]) -> StdCommand {
-    let mut cmd = find_command("cargo");
-    cmd.args([
-        "bench",
-        "--workspace",
-        "--all-features",
-        "--bench",
-        bench.unwrap_or("*"),
-    ]);
-    if no_run {
-        cmd.arg("--no-run");
-    }
-    if !args.is_empty() {
-        cmd.arg("--").args(args);
-    }
-    cmd
+fn make_bench_cmds(no_run: bool, bench: Option<&str>, args: &[String]) -> Vec<StdCommand> {
+    let benches = bench.map_or_else(|| vec!["ecosystem", "primitives"], |bench| vec![bench]);
+
+    benches
+        .into_iter()
+        .map(|bench| {
+            let mut cmd = find_command("cargo");
+            cmd.args(["bench", "--workspace", "--all-features", "--bench", bench]);
+            if no_run {
+                cmd.arg("--no-run");
+            }
+            if !args.is_empty() {
+                cmd.arg("--").args(args);
+            }
+            cmd
+        })
+        .collect()
 }
 
 fn make_build_cmd(locked: bool) -> StdCommand {
@@ -598,4 +597,23 @@ mod tests {
             SemverReleaseType::Patch
         );
     }
+
+    #[test]
+    fn default_benchmarks_leave_the_stage_target_opt_in() {
+        let commands = make_bench_cmds(false, None, &[]);
+        let selected = commands
+            .iter()
+            .map(|cmd| cmd.get_args().last().unwrap().to_str().unwrap())
+            .collect::<Vec<_>>();
+
+        assert_eq!(selected, ["ecosystem", "primitives"]);
+
+        let stage_command = make_bench_cmds(false, Some("upgradable_stages"), &[]);
+        assert_eq!(stage_command.len(), 1);
+        assert!(
+            stage_command[0]
+                .get_args()
+                .any(|arg| arg == "upgradable_stages")
+        );
+    }
 }