ruda_runtime/runtime/tune/stack/
runtime_adapter.rs

1use super::*;
2use super::policy::CandidateView;
3use crate::runtime::{backend::Runtime, client::ComputeClient, tune::{AutotuneKey, AutotuneOutput, TunableSet, TuneInputs, AutotuneError}};
4use std::{any::type_name, path::PathBuf, string::{String, ToString}, sync::{Arc, OnceLock}, time::{Duration, Instant, SystemTime, UNIX_EPOCH}, vec::Vec};
5
6static GLOBAL: OnceLock<StackTuner> = OnceLock::new();
7static SESSION: OnceLock<String> = OnceLock::new();
8/// Install once, before worker threads/model loading. No implicit environment mutation or I/O.
9pub fn enable_stack_autotune(policy: StackPolicy, cache_directory: Option<PathBuf>) -> Result<&'static StackTuner, TuneFailure> {
10    let tuner = StackTuner::new(policy, cache_directory)?;
11    GLOBAL.set(tuner).map_err(|_| TuneFailure::invalid("stack autotuning was already configured"))?;
12    Ok(GLOBAL.get().expect("controller was initialized"))
13}
14/// Return the installed controller, or None to preserve the legacy tuner route.
15/// Merely querying it does not create a controller, inspect files or run trials.
16pub fn stack_autotuner() -> Option<&'static StackTuner> { GLOBAL.get() }
17
18/// Supply extra tags for a deployment's compiler flags and load/topology/power regime.
19/// Set these BEFORE constructing devices/starting inference; changing them live is unsupported.
20#[derive(Debug, Clone)]
21pub struct RuntimeEnvironment {
22    /// Canonical backend/device/driver/build/runtime-options signature.
23    pub fingerprint: String,
24    /// Whether a driver identity and compiled build identity permit disk reuse.
25    pub persistent: bool,
26    /// Deployment-supplied regime, defaulting to isolated-single-device.
27    /// The default does not establish actual isolation or topology.
28    pub execution_context: String
29}
30/// Probe and memoize immutable environment identity for this runtime/client.
31/// Deployment tags must be set before first use; live mutation is not supported.
32/// Missing identity yields session-only caching rather than a fabricated driver version.
33pub fn runtime_environment<R: Runtime>(client: &ComputeClient<R>) -> RuntimeEnvironment {
34    use std::{any::TypeId, collections::BTreeMap, sync::RwLock};
35    type Key = (TypeId, u16, u16, u64);
36    static ENVIRONMENTS: OnceLock<RwLock<BTreeMap<Key, RuntimeEnvironment>>> = OnceLock::new();
37    let id = client.device_id();
38    let key = (TypeId::of::<R>(), id.type_id, id.index_id, client.properties_fingerprint());
39    let cache = ENVIRONMENTS.get_or_init(|| RwLock::new(BTreeMap::new()));
40    if let Some(env) = cache.read().unwrap_or_else(|p| p.into_inner()).get(&key).cloned() { return env; }
41    let env = probe_environment(client);
42    let mut cache = cache.write().unwrap_or_else(|p| p.into_inner());
43    // Device/runtime metadata is immutable after initialization. Keep this memo bounded.
44    if cache.len() < 256 { cache.insert(key, env.clone()); }
45    env
46}
47fn probe_environment<R: Runtime>(client: &ComputeClient<R>) -> RuntimeEnvironment {
48    let driver_override = std::env::var("RUDA_AUTOTUNE_DRIVER_TAG").ok().filter(|s| !s.is_empty());
49    let driver = match (R::autotune_driver_fingerprint(client), driver_override) {
50        (Some(probed), Some(extra)) => Some(fields(&[&probed, &extra])),
51        (Some(probed), None) => Some(probed),
52        (None, Some(asserted)) => Some(asserted),
53        (None, None) => None,
54    };
55    let persistent = driver.is_some() && env!("RUDA_STACK_BUILD_ID") != "unavailable";
56    let driver = driver.unwrap_or_else(|| SESSION.get_or_init(|| std::format!("session-only:{}:{}", std::process::id(), SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos())).clone());
57    let build_extra = std::env::var("RUDA_AUTOTUNE_BUILD_TAG").unwrap_or_default();
58    let execution_context = std::env::var("RUDA_AUTOTUNE_CONTEXT_TAG").unwrap_or_else(|_| "isolated-single-device".into());
59    use crate::runtime::config::{RudaRuntimeConfig, RuntimeConfig};
60    let config = RudaRuntimeConfig::get();
61    let runtime_options = std::format!("compilation={:?};streaming={:?};memory={:?}", config.compilation, config.streaming, config.memory);
62    let fingerprint = fields(&[R::name(client), &std::format!("{:?}", client.device_id()), &std::format!("{:016x}", client.properties_fingerprint()),
63        &driver, type_name::<R::Compiler>(), env!("RUDA_STACK_BUILD_ID"), &build_extra,
64        &std::format!("{:?}", client.info()), &runtime_options]);
65    RuntimeEnvironment { fingerprint, persistent, execution_context }
66}
67
68fn complete<R: Runtime>(client: &ComputeClient<R>) -> Result<(), TuneFailure> {
69    let submitted = client.flush();
70    // Completion errors are never demoted to a failed candidate followed by another launch.
71    ruda_core::future::block_on(client.sync()).map_err(|e| TuneFailure::device(std::format!("autotune synchronization failed: {e:?}")))?;
72    submitted.map_err(|e| TuneFailure::rejected(std::format!("candidate submission failed after a successful completion fence: {e:?}")))
73}
74fn run_output<R: Runtime, K: AutotuneKey, I: TuneInputs, O: AutotuneOutput>(
75    client: &ComputeClient<R>, set: &TunableSet<K,I,O>, index: usize, input: I::At<'_>,
76) -> Result<O, TuneFailure> {
77    let result = set.fastest(index).execute(input);
78    complete(client)?;
79    result.map_err(|e| TuneFailure::rejected(std::format!("candidate rejected: {e:?}")))
80}
81struct RuntimeTrials<'set, 'inp, R: Runtime, K: AutotuneKey, I: TuneInputs, O: AutotuneOutput> {
82    client: ComputeClient<R>, set: &'set TunableSet<K,I,O>, key: &'set K,
83    input: &'set I::At<'inp>, indices: &'set [usize], timing: Timing,
84}
85impl<R: Runtime, K: AutotuneKey, I: TuneInputs, O: AutotuneOutput> TrialRunner for RuntimeTrials<'_, '_, R,K,I,O> {
86    fn validate(&mut self, reference: usize, candidate: usize, tolerance: Tolerance) -> Result<Validation, TuneFailure> {
87        let a = self.set.generate_inputs(self.key, self.input);
88        let b = self.set.generate_inputs(self.key, self.input);
89        complete(&self.client)?;
90        let client = self.client.clone(); let set = self.set;
91        let ri = self.indices[reference]; let ci = self.indices[candidate];
92        self.client.exclusive(move || {
93            let _depth = super::engine::DepthGuard::enter();
94            let expected = run_output(&client, set, ri, a)?;
95            let actual = run_output(&client, set, ci, b)?;
96            let result = expected.validate_for_tuning(&actual, tolerance.absolute, tolerance.relative, tolerance.max_bytes);
97            // Readback/contiguous conversion can also launch kernels. Fence before accepting a
98            // validation mismatch or freeing trial state.
99            complete(&client)?;
100            match result {
101                Ok(true) => Ok(Validation::Passed), Ok(false) => Ok(Validation::Unsupported),
102                Err(e) => Err(TuneFailure::rejected(e)),
103            }
104        }).map_err(|e| TuneFailure::device(std::format!("exclusive validation failed: {e:?}")))?
105    }
106    fn measure(&mut self, candidate: usize) -> Result<Duration, TuneFailure> {
107        let input = self.set.generate_inputs(self.key, self.input);
108        complete(&self.client)?;
109        let client = self.client.clone(); let set = self.set;
110        let index = self.indices[candidate]; let timing = self.timing;
111        self.client.exclusive(move || {
112            let _depth = super::engine::DepthGuard::enter();
113            // Preparation/sandbox copies are outside the measured region. All allocations and
114            // relayouts made INSIDE the actual candidate operation are part of EndToEnd timing.
115            complete(&client)?;
116            match timing {
117                Timing::EndToEnd => {
118                    let start = Instant::now();
119                    let out = run_output(&client, set, index, input)?;
120                    let elapsed = start.elapsed();
121                    std::hint::black_box(&out); drop(out);
122                    Ok(elapsed)
123                }
124                Timing::Device => {
125                    let profiled = client.profile(move || set.fastest(index).execute(input), "stack-autotune");
126                    complete(&client)?;
127                    let (output, profile) = profiled.map_err(|e| TuneFailure::rejected(std::format!("device profile failed: {e:?}")))?;
128                    let output = output.map_err(|e| TuneFailure::rejected(std::format!("candidate rejected: {e:?}")))?;
129                    let elapsed = ruda_core::future::block_on(profile.resolve()).duration();
130                    std::hint::black_box(&output); drop(output);
131                    Ok(elapsed)
132                }
133            }
134        }).map_err(|e| TuneFailure::device(std::format!("exclusive measurement failed: {e:?}")))?
135    }
136}
137
138/// Fallible full-stack path. Actual request execution is performed ONCE after selection; an
139/// execution error invalidates future choices but is never silently replayed on another kernel.
140/// Requires an installed controller, an in-range explicit reference and an exact workload
141/// signature on the set. Missing metadata returns AutotuneError before live execution.
142pub fn try_execute_stack<'a, R: Runtime, K: AutotuneKey, I: TuneInputs, O: AutotuneOutput>(
143    name: &str, device_id: &str, client: &ComputeClient<R>, set: Arc<TunableSet<K,I,O>>, input: I::At<'a>,
144) -> Result<O, AutotuneError> {
145    let tuner = stack_autotuner().ok_or_else(|| autotune_error(name, "stack autotuning is not enabled"))?;
146    let reference = set.stack_reference().ok_or_else(|| autotune_error(name, "no explicit reference/workload signature was registered"))?;
147    if reference >= set.len() { return Err(autotune_error(name, "reference index out of range")); }
148    let key = set.generate_key(&input);
149    let env = runtime_environment(client);
150    let problem = Problem { scope: if name.contains("fusion") { Scope::Graph } else { Scope::Operator }, operation: fields(&[name, device_id, set.stack_checksum_ref()]), environment: env.fingerprint,
151        workload: set.stack_workload(&input).ok_or_else(|| autotune_error(name, "no exact workload signature"))?,
152        execution_context: env.execution_context, persistent: env.persistent };
153    // Enumerate EVERY eligible priority group. Group order remains a useful bounded-search
154    // heuristic but a first valid group no longer terminates the search.
155    let mut indices = std::vec![reference]; let mut plan = set.plan(&key);
156    let mut included = std::vec![false; set.len()];
157    included[reference] = true;
158    loop { let batch = plan.next(None); if batch.is_empty() { break; }
159        for index in batch { if !included[index] { indices.push(index); included[index] = true; } }
160    }
161    let candidates: Vec<_> = indices.iter().map(|&i| CandidateView::new(&set.fastest(i).name)).collect();
162    // Cache hits must retain normal asynchronous execution. TrialRunner fences before every
163    // validation/measurement; do NOT synchronize ordinary requests just to read this cache.
164    let mut trials = RuntimeTrials { client: client.clone(), set: &set, key: &key, input: &input, indices: &indices, timing: tuner.policy().timing };
165    let decision = tuner.select_candidates(&problem, &candidates, 0, &mut trials).map_err(|e| autotune_error(name, &e.to_string()))?;
166    let output = set.fastest(indices[decision.index]).execute(input);
167    if output.is_err() { tuner.invalidate(&decision, true); }
168    output
169}
170fn autotune_error(name: &str, message: &str) -> AutotuneError {
171    AutotuneError::Unknown { name: name.to_string(), err: message.to_string() }
172}
ruda_runtime/runtime/tune/stack/runtime_adapter.rs

ruda_runtime/runtime/tune/stack/
runtime_adapter.rs
