ruda_runtime/runtime/tune/stack/
cache.rs

1//! Bounded persistent records. Files contain the FULL key; the filename hash is only an index.
2use std::{fs::{self, OpenOptions}, io::{self, Read, Write}, path::{Path, PathBuf},
3    string::{String, ToString}, sync::atomic::{AtomicU64, Ordering}, time::{SystemTime, UNIX_EPOCH}, vec::Vec};
4
5const MAGIC: &[u8; 8] = b"RUDATN01";
6const MAX_FILE: usize = 512 * 1024;
7const MAX_KEY: usize = 384 * 1024;
8const MAX_NAME: usize = 4096;
9static NEXT_FILE: AtomicU64 = AtomicU64::new(0);
10
11/// Internal versioned disk payload; not exported as the public stack cache API.
12#[derive(Debug, Clone)]
13pub struct Record {
14    /// Operator/graph/pipeline discriminator, encoded as 0/1/2.
15    pub scope: u8,
16    /// Complete canonical key; hash-only lookup is insufficient.
17    pub key: String,
18    /// Stable candidate name, not a process-local candidate index.
19    pub winner: String,
20    /// Creation time in Unix seconds, used by controller freshness checks.
21    pub created: u64,
22    /// Median paired reference duration in nanoseconds, or the single reference baseline.
23    pub reference_ns: u64,
24    /// Median selected duration in nanoseconds, or the single reference baseline.
25    pub winner_ns: u64,
26    /// Positive finite selected/reference score.
27    pub ratio: f64,
28    /// Whether a supported validator passed before the record was created.
29    pub verified: bool,
30}
31/// This is NOT a cryptographic authenticity check. Caches must be in a trusted user directory.
32pub fn digest(bytes: &[u8]) -> String {
33    let mut a = 0xcbf29ce484222325u64;
34    let mut b = 0x84222325cbf29ce4u64;
35    for &v in bytes {
36        a = (a ^ v as u64).wrapping_mul(0x100000001b3);
37        b = (b ^ (v as u64).wrapping_add(1)).wrapping_mul(0x100000001b3);
38    }
39    std::format!("{a:016x}{b:016x}")
40}
41/// Unix seconds; a clock before the epoch yields zero rather than panicking.
42pub fn now_seconds() -> u64 { SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() }
43impl Record {
44    /// Accept only non-future records younger than ttl seconds; the endpoint expires.
45    pub fn is_fresh(&self, now: u64, ttl: u64) -> bool {
46        self.created <= now && now - self.created < ttl
47    }
48    fn encode(&self) -> io::Result<Vec<u8>> {
49        if self.scope > 2 || self.key.len() > MAX_KEY || self.winner.len() > MAX_NAME || self.winner.is_empty()
50            || self.reference_ns == 0 || self.winner_ns == 0 || !self.ratio.is_finite() || self.ratio <= 0.0 {
51            return Err(invalid("invalid tuning record"));
52        }
53        let mut bytes = MAGIC.to_vec();
54        for text in [&self.key, &self.winner] {
55            bytes.extend_from_slice(&(text.len() as u32).to_le_bytes());
56            bytes.extend_from_slice(text.as_bytes());
57        }
58        for value in [self.created, self.reference_ns, self.winner_ns, self.ratio.to_bits()] {
59            bytes.extend_from_slice(&value.to_le_bytes());
60        }
61        bytes.push(u8::from(self.verified));
62        bytes.push(self.scope);
63        let checksum = digest(&bytes);
64        bytes.extend_from_slice(checksum.as_bytes());
65        Ok(bytes)
66    }
67    fn decode(bytes: &[u8]) -> io::Result<Self> {
68        if bytes.len() > MAX_FILE || bytes.len() < 83 || !bytes.starts_with(MAGIC) { return Err(invalid("bad tuning cache header/length")); }
69        let end = bytes.len() - 32;
70        if digest(&bytes[..end]).as_bytes() != &bytes[end..] { return Err(invalid("tuning cache checksum mismatch")); }
71        let mut p = 8;
72        fn take<'a>(bytes: &'a [u8], p: &mut usize, count: usize) -> io::Result<&'a [u8]> {
73            let stop = p.checked_add(count).ok_or_else(|| invalid("cache length overflow"))?;
74            let value = bytes.get(*p..stop).ok_or_else(|| invalid("truncated tuning cache"))?;
75            *p = stop; Ok(value)
76        }
77        let mut text = |limit: usize| -> io::Result<String> {
78            let length = u32::from_le_bytes(take(&bytes[..end], &mut p, 4)?.try_into().map_err(|_| invalid("bad length"))?) as usize;
79            if length > limit { return Err(invalid("cache field too large")); }
80            String::from_utf8(take(&bytes[..end], &mut p, length)?.to_vec()).map_err(|_| invalid("invalid cache UTF-8"))
81        };
82        let key = text(MAX_KEY)?;
83        let winner = text(MAX_NAME)?;
84        let mut number = || -> io::Result<u64> {
85            Ok(u64::from_le_bytes(take(&bytes[..end], &mut p, 8)?.try_into().map_err(|_| invalid("bad u64"))?))
86        };
87        let created = number()?;
88        let reference_ns = number()?;
89        let winner_ns = number()?;
90        let ratio = f64::from_bits(number()?);
91        let verified = match take(&bytes[..end], &mut p, 1)?[0] { 0 => false, 1 => true, _ => return Err(invalid("invalid validation flag")) };
92        let scope = take(&bytes[..end], &mut p, 1)?[0];
93        if scope > 2 || p != end || winner.is_empty() || reference_ns == 0 || winner_ns == 0 || !ratio.is_finite() || ratio <= 0.0 { return Err(invalid("invalid cache record")); }
94        Ok(Self { scope, key, winner, created, reference_ns, winner_ns, ratio, verified })
95    }
96}
97fn invalid(message: &str) -> io::Error { io::Error::new(io::ErrorKind::InvalidData, message.to_string()) }
98
99/// Internal bounded disk store. Controller APIs supply policy/freshness checks;
100/// decoding a record does not itself validate numerical output or driver identity.
101#[derive(Debug)]
102pub struct DiskCache { root: PathBuf, capacity: usize }
103impl DiskCache {
104    /// A dedicated directory is used, never the legacy autotune cache directory itself.
105    pub fn new(root: PathBuf, capacity: usize) -> Self { Self { root: root.join("stack-autotune-v1"), capacity } }
106    fn path(&self, key: &str) -> PathBuf { self.root.join(std::format!("stack-v1-{}.rtune", digest(key.as_bytes()))) }
107    /// Read at most 512 KiB, verify framing/checksum and compare the complete key.
108    /// Absent files/key mismatch are None; corruption and other I/O failures are errors.
109    /// TTL and current-process numerical validation are handled by the controller.
110    pub fn load(&self, key: &str) -> io::Result<Option<Record>> {
111        let path = self.path(key);
112        let mut file = match fs::File::open(path) { Ok(f) => f, Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None), Err(e) => return Err(e) };
113        if file.metadata()?.len() > MAX_FILE as u64 { return Err(invalid("tuning cache file too large")); }
114        let mut bytes = Vec::new();
115        (&mut file).take((MAX_FILE + 1) as u64).read_to_end(&mut bytes)?;
116        let record = Record::decode(&bytes)?;
117        // Hash collision or copied cache files are misses, not alternate executable choices.
118        if record.key == key { Ok(Some(record)) } else { Ok(None) }
119    }
120    /// Remove only this key's cache file; an absent file is already successful.
121    pub fn remove(&self, key: &str) -> io::Result<()> {
122        match fs::remove_file(self.path(key)) { Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()), r => r }
123    }
124    /// Encode, fsync a newly created temporary file, rename, then prune old records.
125    /// Failed replacement never deletes the previous destination first. A pruning
126    /// failure can be reported after the new record has already been installed.
127    pub fn save(&self, record: &Record) -> io::Result<()> {
128        let bytes = record.encode()?;
129        fs::create_dir_all(&self.root)?;
130        let temp = self.root.join(std::format!(".tmp-{}-{}", std::process::id(), NEXT_FILE.fetch_add(1, Ordering::Relaxed)));
131        // create_new avoids silently following an existing temporary path or symlink.
132        let result = (|| {
133            let mut file = OpenOptions::new().write(true).create_new(true).open(&temp)?;
134            file.write_all(&bytes)?; file.sync_all()?; drop(file);
135            // Unix atomically replaces. Platforms rejecting replacement report a cache warning;
136            // they never delete a known good record first to force a rename.
137            fs::rename(&temp, self.path(&record.key))?;
138            self.prune()
139        })();
140        if result.is_err() { let _ = fs::remove_file(&temp); }
141        result
142    }
143    fn prune(&self) -> io::Result<()> {
144        let mut files = Vec::new();
145        for entry in fs::read_dir(&self.root)? {
146            let entry = entry?; let name = entry.file_name(); let name = name.to_string_lossy();
147            if name.starts_with("stack-v1-") && name.ends_with(".rtune") && entry.file_type()?.is_file() {
148                files.push((entry.metadata()?.modified()?, entry.path()));
149            }
150        }
151        files.sort_by(|a,b| a.0.cmp(&b.0).then(a.1.cmp(&b.1)));
152        let extra = files.len().saturating_sub(self.capacity);
153        for (_, path) in files.into_iter().take(extra) {
154            if let Err(e) = fs::remove_file(path) { if e.kind() != io::ErrorKind::NotFound { return Err(e); } }
155        }
156        Ok(())
157    }
158    /// Dedicated stack-autotune-v1 directory, not the legacy LocalTuner cache root.
159    pub fn directory(&self) -> &Path { &self.root }
160}
161
162#[cfg(test)]
163mod tests {
164    use super::*;
165    fn record() -> Record { Record { scope: 0, key: "test/测试".into(), winner: "safe".into(), created: 50, reference_ns: 100, winner_ns: 75, ratio: 0.75, verified: true } }
166    #[test] fn python_generated_golden_fixture() {
167        let raw = include_bytes!("testdata/cache_v1.fixture");
168        let record = Record::decode(raw).unwrap();
169        assert_eq!(record.key, "known/测试"); assert_eq!(record.winner, "reference-plan");
170        assert_eq!(record.scope, 1); assert_eq!(record.ratio, 0.8); assert_eq!(record.encode().unwrap(), raw.to_vec());
171    }
172    #[test] fn binary_round_trip() { let r = record(); let d = Record::decode(&r.encode().unwrap()).unwrap(); assert_eq!(d.key, r.key); assert_eq!(d.winner, r.winner); assert!(d.verified); }
173    #[test] fn all_truncations_rejected() { let b = record().encode().unwrap(); for i in 0..b.len() { assert!(Record::decode(&b[..i]).is_err()); } }
174    #[test] fn corruption_rejected() { let mut b = record().encode().unwrap(); b[14] ^= 1; assert!(Record::decode(&b).is_err()); }
175    #[test] fn oversized_field_rejected() { let mut r = record(); r.key = "x".repeat(MAX_KEY+1); assert!(r.encode().is_err()); }
176    #[test] fn nan_ratio_rejected() { let mut r = record(); r.ratio = f64::NAN; assert!(r.encode().is_err()); }
177    #[test] fn zero_duration_rejected() { let mut r = record(); r.winner_ns = 0; assert!(r.encode().is_err()); }
178    #[test] fn future_and_expired_records_miss() { let r = record(); assert!(!r.is_fresh(49, 100)); assert!(!r.is_fresh(150, 100)); assert!(r.is_fresh(149, 100)); }
179    #[test] fn file_round_trip_and_prune() {
180        let dir = std::env::temp_dir().join(std::format!("ruda-stack-test-{}-{}", std::process::id(), NEXT_FILE.fetch_add(1, Ordering::Relaxed)));
181        let c = DiskCache::new(dir.clone(), 1); let r = record();
182        c.save(&r).unwrap(); assert!(c.load(&r.key).unwrap().is_some());
183        let mut r2 = r; r2.key = "another".into(); c.save(&r2).unwrap();
184        assert_eq!(fs::read_dir(c.directory()).unwrap().count(), 1);
185        fs::remove_dir_all(dir).unwrap();
186    }
187}
ruda_runtime/runtime/tune/stack/cache.rs

ruda_runtime/runtime/tune/stack/
cache.rs
