launchbound_bench/
run.rs

1//! Plan execution: resumable, checkpointed after every candidate (the box
2//! dies — idle guard, dead-man switch, spot reclaim — so the harness is
3//! resumable or it is broken, docs/ARCHITECTURE.md), with a CPU heartbeat so a
4//! GPU-bound sweep never looks idle to the 30-minute CPU alarm.
5
6use crate::cuda::Device;
7use crate::plan::{ArgSpec, BenchPlan, Candidate};
8use crate::stats::{Summary, summarize};
9use serde::{Deserialize, Serialize};
10use std::path::Path;
11use std::sync::atomic::{AtomicBool, Ordering};
12use std::time::Instant;
13
14/// What measuring one candidate produced — including the ways it can fail.
15#[derive(Debug, Clone, Serialize, Deserialize)]
16pub struct CandidateResult {
17    /// Its canonical `config.v1` ID, joining this to the plan and verdicts.
18    pub id: String,
19    /// Its dimension assignments.
20    pub config: String,
21    /// `ok`, `error`, or `timeout` — the last meaning a watchdog fired on
22    /// a candidate the gate had refused, which is the refusal being right.
23    pub status: String,
24    /// Why it failed, when it did.
25    #[serde(default, skip_serializing_if = "Option::is_none")]
26    pub error: Option<String>,
27    /// Untimed launches performed before measuring.
28    pub warmup: u32,
29    /// Timed launches requested.
30    pub repeats: u32,
31    /// Every raw sample, in launch order. Kept so a summary can be
32    /// recomputed without re-running the GPU.
33    #[serde(default)]
34    pub times_ms: Vec<f64>,
35    /// The reduction of `times_ms`. `None` when nothing was measured.
36    #[serde(default, skip_serializing_if = "Option::is_none")]
37    pub summary: Option<Summary>,
38    /// Wall-clock seconds this candidate consumed on the GPU host.
39    pub gpu_seconds: f64,
40}
41
42/// A `results.v1` document: every measurement, and the machine that took it.
43///
44/// Checkpointed after each candidate, so an interrupted sweep resumes from
45/// what it already measured rather than starting over.
46#[derive(Debug, Clone, Serialize, Deserialize)]
47pub struct Results {
48    /// Schema tag; always `results.v1`.
49    pub schema: String,
50    /// The kernel measured.
51    pub kernel: String,
52    /// The entry point launched.
53    pub entry: String,
54    /// The capability the plan was gated at.
55    pub plan_cc: String,
56    /// Product name of the device, as the driver reports it.
57    pub device_name: String,
58    /// The device's actual capability, which need not equal `plan_cc`.
59    pub device_cc: String,
60    /// Driver version. Part of what makes a timing reproducible, and part
61    /// of why results do not port (`docs/LIMITATIONS.md`).
62    pub driver_version: String,
63    /// One entry per candidate visited, in visiting order.
64    pub candidates: Vec<CandidateResult>,
65    /// GPU seconds the whole sweep consumed.
66    pub total_gpu_seconds: f64,
67    /// Strategy that produced the visiting order (`exhaustive` | `random:<seed>`).
68    #[serde(default, skip_serializing_if = "Option::is_none")]
69    pub strategy: Option<String>,
70    /// True when the sweep stopped because the wall budget ran out.
71    #[serde(default)]
72    pub budget_exhausted: bool,
73}
74
75/// How to visit a plan's candidates, and when to stop.
76pub struct RunOptions {
77    /// Visiting order over plan.candidates indices (a permutation or
78    /// prefix); defaults to plan order.
79    pub order: Vec<usize>,
80    /// Wall-clock budget; the sweep stops (resumably) when it is spent.
81    pub budget_secs: Option<f64>,
82    /// Recorded in results for provenance.
83    pub strategy: Option<String>,
84}
85
86impl RunOptions {
87    /// Visit every candidate in plan order, with no budget.
88    pub fn exhaustive(plan: &BenchPlan) -> Self {
89        RunOptions {
90            order: (0..plan.candidates.len()).collect(),
91            budget_secs: None,
92            strategy: Some("exhaustive".into()),
93        }
94    }
95}
96
97/// Execute `plan`, appending to `results_path` (resume: candidates already
98/// present are skipped). Writes the results file after every candidate.
99pub fn run_plan(
100    plan: &BenchPlan,
101    plan_dir: &Path,
102    results_path: &Path,
103    options: &RunOptions,
104    progress: &mut dyn FnMut(&str),
105) -> Result<Results, String> {
106    let device = Device::open()?;
107    progress(&format!(
108        "device: {} (cc {}, driver {})",
109        device.name, device.cc, device.driver_version
110    ));
111    if device.cc != plan.cc {
112        progress(&format!(
113            "WARNING: plan was gated at cc {}, device is cc {} — verdicts do not transfer \
114             across parts (docs/SAFETY.md); results will be labelled with the device cc",
115            plan.cc, device.cc
116        ));
117    }
118
119    // A checkpoint that cannot be read is not "start over": resuming would
120    // silently discard measurements that cost GPU time, and the file is the
121    // only record of them. Say what is wrong and stop.
122    let existing = Results::load(results_path)?;
123    let mut results = match existing {
124        Some(existing) => {
125            progress(&format!(
126                "resuming: {} candidates already measured",
127                existing.candidates.len()
128            ));
129            existing
130        }
131        None => Results {
132            schema: "results.v1".into(),
133            kernel: plan.kernel.clone(),
134            entry: plan.entry.clone(),
135            plan_cc: plan.cc.clone(),
136            device_name: device.name.clone(),
137            device_cc: device.cc.clone(),
138            driver_version: device.driver_version.clone(),
139            candidates: Vec::new(),
140            total_gpu_seconds: 0.0,
141            strategy: options.strategy.clone(),
142            budget_exhausted: false,
143        },
144    };
145    results.budget_exhausted = false;
146
147    let heartbeat_stop = start_heartbeat();
148    let sweep_started = Instant::now();
149
150    for &index in &options.order {
151        let Some(candidate) = plan.candidates.get(index) else {
152            return Err(format!("order index {index} out of range"));
153        };
154        if results.candidates.iter().any(|c| c.id == candidate.id) {
155            continue;
156        }
157        if let Some(budget) = options.budget_secs
158            && sweep_started.elapsed().as_secs_f64() >= budget
159        {
160            results.budget_exhausted = true;
161            progress(&format!(
162                "budget exhausted after {:.1}s: {} of {} candidates measured (resumable)",
163                sweep_started.elapsed().as_secs_f64(),
164                results.candidates.len(),
165                plan.candidates.len()
166            ));
167            break;
168        }
169        // A gate-refused candidate may genuinely hang (that is why it was
170        // refused). Checkpoint a `timeout` record BEFORE launching, so a
171        // watchdog abort (or a wedged GPU context) leaves a resumable
172        // truth on disk; overwrite it with the real outcome if we survive.
173        let watchdog = if candidate.unsafe_candidate {
174            results.candidates.push(CandidateResult {
175                id: candidate.id.clone(),
176                config: candidate.config.clone(),
177                status: "timeout".into(),
178                error: Some(format!(
179                    "unsafe candidate did not complete within {UNSAFE_TIMEOUT_SECS}s;                      presumed hung (this is the failure mode the gate predicts)"
180                )),
181                warmup: candidate.warmup,
182                repeats: candidate.repeats,
183                times_ms: Vec::new(),
184                summary: None,
185                gpu_seconds: unsafe_timeout_secs() as f64,
186            });
187            results.checkpoint(results_path)?;
188            progress(&format!(
189                "{} UNSAFE candidate: watchdog armed at {}s",
190                candidate.id,
191                unsafe_timeout_secs()
192            ));
193            Some(arm_watchdog(unsafe_timeout_secs()))
194        } else {
195            None
196        };
197        let started = Instant::now();
198        let outcome = run_candidate(&device, plan, plan_dir, candidate);
199        if let Some(armed) = watchdog {
200            armed.store(true, Ordering::Relaxed); // disarm
201            // Replace the pre-checkpointed timeout record with the truth.
202            results.candidates.retain(|c| c.id != candidate.id);
203        }
204        let gpu_seconds = started.elapsed().as_secs_f64();
205        let result = match outcome {
206            Ok(times_ms) => {
207                let summary = summarize(&times_ms);
208                progress(&format!(
209                    "{} {}: median {} over {} repeats ({:.1}s)",
210                    candidate.id,
211                    candidate.config,
212                    summary
213                        .as_ref()
214                        .map(|s| format!(
215                            "{:.4} ms [{:.4}, {:.4}]",
216                            s.median_ms, s.ci95_lo_ms, s.ci95_hi_ms
217                        ))
218                        .unwrap_or_else(|| "n/a".into()),
219                    candidate.repeats,
220                    gpu_seconds,
221                ));
222                CandidateResult {
223                    id: candidate.id.clone(),
224                    config: candidate.config.clone(),
225                    status: "ok".into(),
226                    error: None,
227                    warmup: candidate.warmup,
228                    repeats: candidate.repeats,
229                    times_ms,
230                    summary,
231                    gpu_seconds,
232                }
233            }
234            Err(e) => {
235                progress(&format!("{} ERROR: {e}", candidate.id));
236                CandidateResult {
237                    id: candidate.id.clone(),
238                    config: candidate.config.clone(),
239                    status: "error".into(),
240                    error: Some(e),
241                    warmup: candidate.warmup,
242                    repeats: candidate.repeats,
243                    times_ms: Vec::new(),
244                    summary: None,
245                    gpu_seconds,
246                }
247            }
248        };
249        results.candidates.push(result);
250        results.total_gpu_seconds = sweep_started.elapsed().as_secs_f64();
251        results.checkpoint(results_path)?;
252    }
253
254    heartbeat_stop.store(true, Ordering::Relaxed);
255    results.total_gpu_seconds = sweep_started.elapsed().as_secs_f64();
256    results.checkpoint(results_path)?;
257    Ok(results)
258}
259
260fn run_candidate(
261    device: &Device,
262    plan: &BenchPlan,
263    plan_dir: &Path,
264    candidate: &Candidate,
265) -> Result<Vec<f64>, String> {
266    let ptx_path = plan_dir.join(&candidate.ptx);
267    let ptx = std::fs::read_to_string(&ptx_path)
268        .map_err(|e| format!("reading {}: {e}", ptx_path.display()))?;
269    let module = device.load_module(&ptx, &plan.entry)?;
270
271    // Materialize buffers and the param pointer table, in ArgSpec order.
272    // Each ArgSpec is exactly one .param slot.
273    let mut buffers = Vec::new(); // (arg index, Buffer)
274    for (i, arg) in candidate.args.iter().enumerate() {
275        match arg {
276            ArgSpec::InF32 { len } => {
277                let host: Vec<f32> = deterministic_f32(*len);
278                let buf = device.alloc(host.len() * 4)?;
279                device.copy_in(&buf, cast_bytes(&host))?;
280                buffers.push((i, buf));
281            }
282            ArgSpec::InU32 { len, modulo } => {
283                let host: Vec<u32> = deterministic_u32(*len, *modulo);
284                let buf = device.alloc(host.len() * 4)?;
285                device.copy_in(&buf, cast_bytes(&host))?;
286                buffers.push((i, buf));
287            }
288            ArgSpec::OutF32 { len } | ArgSpec::OutU32 { len } => {
289                let zero = vec![0u8; (*len as usize) * 4];
290                let buf = device.alloc(zero.len())?;
291                device.copy_in(&buf, &zero)?;
292                buffers.push((i, buf));
293            }
294            ArgSpec::LenOf { .. } | ArgSpec::U32 { .. } | ArgSpec::U64 { .. } => {}
295        }
296    }
297
298    // Scalar storage must outlive the launch; the params table points into
299    // these vectors and the buffers' device pointers.
300    let mut ptr_slots: Vec<u64> = Vec::new();
301    let mut u32_slots: Vec<u32> = Vec::new();
302    let mut u64_slots: Vec<u64> = Vec::new();
303    #[derive(Clone, Copy)]
304    enum Slot {
305        Ptr(usize),
306        U32(usize),
307        U64(usize),
308    }
309    let mut slots = Vec::with_capacity(candidate.args.len());
310    for (i, arg) in candidate.args.iter().enumerate() {
311        match arg {
312            ArgSpec::InF32 { .. }
313            | ArgSpec::InU32 { .. }
314            | ArgSpec::OutF32 { .. }
315            | ArgSpec::OutU32 { .. } => {
316                // The loop above materializes a buffer for exactly these
317                // four ArgSpec kinds, so the lookup succeeds — an invariant
318                // held one loop away from the code that needs it. This
319                // function returns a `Result`; a wrong answer here is a bug
320                // report, not a crash mid-benchmark.
321                let buf = &buffers
322                    .iter()
323                    .find(|(idx, _)| *idx == i)
324                    .ok_or_else(|| format!("internal: argument {i} has no materialized buffer"))?
325                    .1;
326                ptr_slots.push(buf.ptr);
327                slots.push(Slot::Ptr(ptr_slots.len() - 1));
328            }
329            ArgSpec::LenOf { of } => {
330                let len = match candidate.args.get(*of) {
331                    Some(ArgSpec::InF32 { len })
332                    | Some(ArgSpec::OutF32 { len })
333                    | Some(ArgSpec::OutU32 { len })
334                    | Some(ArgSpec::InU32 { len, .. }) => *len,
335                    other => return Err(format!("len_of {of} points at {other:?}")),
336                };
337                u64_slots.push(len);
338                slots.push(Slot::U64(u64_slots.len() - 1));
339            }
340            ArgSpec::U32 { value } => {
341                u32_slots.push(*value as u32);
342                slots.push(Slot::U32(u32_slots.len() - 1));
343            }
344            ArgSpec::U64 { value } => {
345                u64_slots.push(*value);
346                slots.push(Slot::U64(u64_slots.len() - 1));
347            }
348        }
349    }
350    let mut params: Vec<*mut std::ffi::c_void> = slots
351        .iter()
352        .map(|slot| match slot {
353            Slot::Ptr(k) => std::ptr::from_mut(&mut ptr_slots[*k]).cast(),
354            Slot::U32(k) => std::ptr::from_mut(&mut u32_slots[*k]).cast(),
355            Slot::U64(k) => std::ptr::from_mut(&mut u64_slots[*k]).cast(),
356        })
357        .collect();
358
359    for _ in 0..candidate.warmup {
360        device.timed_launch(&module, candidate.grid, candidate.block, &mut params)?;
361    }
362    device.synchronize()?;
363
364    let mut times = Vec::with_capacity(candidate.repeats as usize);
365    for _ in 0..candidate.repeats {
366        times.push(device.timed_launch(&module, candidate.grid, candidate.block, &mut params)?);
367    }
368    device.synchronize()?;
369    Ok(times)
370}
371
372fn cast_bytes<T>(data: &[T]) -> &[u8] {
373    unsafe { std::slice::from_raw_parts(data.as_ptr().cast(), std::mem::size_of_val(data)) }
374}
375
376/// Deterministic xorshift-seeded data: reproducible across runs and hosts.
377fn deterministic_f32(len: u64) -> Vec<f32> {
378    let mut state = 0x9e3779b97f4a7c15u64;
379    (0..len)
380        .map(|_| {
381            state ^= state << 13;
382            state ^= state >> 7;
383            state ^= state << 17;
384            ((state >> 40) as f32) / ((1u64 << 24) as f32)
385        })
386        .collect()
387}
388
389fn deterministic_u32(len: u64, modulo: u64) -> Vec<u32> {
390    let modulo = modulo.max(1);
391    let mut state = 0x2545f4914f6cdd1du64;
392    (0..len)
393        .map(|_| {
394            state ^= state << 13;
395            state ^= state >> 7;
396            state ^= state << 17;
397            (state % modulo) as u32
398        })
399        .collect()
400}
401
402impl Results {
403    /// Read `results.v1` from a run directory.
404    ///
405    /// `Ok(None)` means one thing only: **the file is not there**, so the
406    /// measurement box has not run yet. Everything else is an error.
407    ///
408    /// This used to be `read_to_string(path).ok()?` then `from_str().ok()`,
409    /// so a truncated file, an empty one, `null`, `[]`, a `results.v2` from
410    /// a newer runner and a *directory* named `results.json` all collapsed
411    /// into that same `None` — and `report` rendered "nothing measured yet",
412    /// exit 0, nothing on stderr, with a JSON report that validated. The run
413    /// directory is the hand-off between two machines, and the two
414    /// conditions call for opposite actions: wait, or go and look. Nothing
415    /// told them apart.
416    ///
417    /// `verdicts.v1` two lines away in the report builder already checked
418    /// its schema tag by name; this is the same check, so a future
419    /// `results.v2` is refused by name rather than read as nothing.
420    ///
421    /// # Errors
422    ///
423    /// Any I/O error that is not `NotFound`, any parse failure, and any
424    /// document whose `schema` is not `results.v1` — each naming the path.
425    pub fn load(path: &Path) -> Result<Option<Self>, String> {
426        let text = match std::fs::read_to_string(path) {
427            Ok(text) => text,
428            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
429            Err(e) => return Err(format!("{}: {e}", path.display())),
430        };
431        // The tag first, so a wrong-schema document is named as one rather
432        // than as whatever field happens to be missing from it.
433        let value: serde_json::Value = serde_json::from_str(&text)
434            .map_err(|e| format!("{}: not JSON: {e}", path.display()))?;
435        let declared = value.get("schema").and_then(|s| s.as_str());
436        match declared {
437            Some("results.v1") => {}
438            Some(other) => {
439                return Err(format!(
440                    "{}: unsupported results schema `{other}` (expected `results.v1`)",
441                    path.display()
442                ));
443            }
444            None => {
445                return Err(format!(
446                    "{}: not a results.v1 document (no `schema` field)",
447                    path.display()
448                ));
449            }
450        }
451        serde_json::from_str(&text)
452            .map(Some)
453            .map_err(|e| format!("{}: not a results.v1 document: {e}", path.display()))
454    }
455
456    /// Atomic checkpoint: write to a temp file, then rename.
457    pub fn checkpoint(&self, path: &Path) -> Result<(), String> {
458        let tmp = path.with_extension("json.tmp");
459        let json =
460            serde_json::to_string_pretty(self).map_err(|e| format!("serializing results: {e}"))?;
461        std::fs::write(&tmp, json).map_err(|e| e.to_string())?;
462        std::fs::rename(&tmp, path).map_err(|e| e.to_string())
463    }
464}
465
466const UNSAFE_TIMEOUT_SECS: u64 = 10;
467
468fn unsafe_timeout_secs() -> u64 {
469    std::env::var("LAUNCHBOUND_UNSAFE_TIMEOUT_SECS")
470        .ok()
471        .and_then(|v| v.parse().ok())
472        .unwrap_or(UNSAFE_TIMEOUT_SECS)
473}
474
475/// Watchdog for unsafe candidates: if not disarmed within the deadline the
476/// process exits (a hung kernel cannot be cancelled from user code). The
477/// pre-checkpointed `timeout` record makes the rerun skip it.
478fn arm_watchdog(deadline_secs: u64) -> std::sync::Arc<AtomicBool> {
479    let disarmed = std::sync::Arc::new(AtomicBool::new(false));
480    let flag = disarmed.clone();
481    std::thread::spawn(move || {
482        let start = Instant::now();
483        while start.elapsed().as_secs() < deadline_secs {
484            if flag.load(Ordering::Relaxed) {
485                return;
486            }
487            std::thread::sleep(std::time::Duration::from_millis(100));
488        }
489        if !flag.load(Ordering::Relaxed) {
490            eprintln!(
491                "watchdog: unsafe candidate exceeded {deadline_secs}s; exiting so the                  checkpointed timeout record stands (exit 3, rerun to continue)"
492            );
493            std::process::exit(3);
494        }
495    });
496    disarmed
497}
498
499/// A CPU heartbeat: the box's idle alarm terminates on CPU <5% for 30 min,
500/// and a GPU-bound loop can look idle. Burn a configurable duty cycle on
501/// one core (LAUNCHBOUND_HEARTBEAT_PCT, default 40) until stopped.
502fn start_heartbeat() -> &'static AtomicBool {
503    static STOP: AtomicBool = AtomicBool::new(false);
504    STOP.store(false, Ordering::Relaxed);
505    let duty: u64 = std::env::var("LAUNCHBOUND_HEARTBEAT_PCT")
506        .ok()
507        .and_then(|v| v.parse().ok())
508        .unwrap_or(40)
509        .clamp(1, 100);
510    std::thread::spawn(move || {
511        let mut sink = 0u64;
512        while !STOP.load(Ordering::Relaxed) {
513            let spin = Instant::now();
514            while spin.elapsed().as_millis() < duty as u128 {
515                sink = sink.wrapping_mul(6364136223846793005).wrapping_add(1);
516            }
517            std::hint::black_box(sink);
518            std::thread::sleep(std::time::Duration::from_millis(100 - duty.min(99)));
519        }
520    });
521    &STOP
522}
launchbound_bench/run.rs

launchbound_bench/
run.rs
