vyre_driver/pipeline/
cache.rs

1//! Shared persistent cache for backend compiled-pipeline blobs.
2
3use super::hashing::{
4    dispatch_policy_cache_digest, dispatch_policy_cache_string, hex_encode,
5    normalized_program_cache_digest, try_normalized_program_cache_digest,
6    PipelineDeviceFingerprint,
7};
8use super::CURRENT_PIPELINE_CACHE_KEY_VERSION;
9use crate::backend::DispatchConfig;
10use std::sync::{Arc, MutexGuard};
11use vyre_foundation::ir::Program;
12use vyre_spec::BackendId;
13
14/// Maximum persistent pipeline blob read into memory.
15pub const MAX_DISK_PIPELINE_BLOB_BYTES: u64 = 64 * 1024 * 1024;
16
17/// Disk cache for compiled pipeline blobs keyed by program and device.
18pub struct DiskPipelineCache {
19    root: std::path::PathBuf,
20    pending_flushes: std::sync::Mutex<Vec<std::path::PathBuf>>,
21}
22
23impl DiskPipelineCache {
24    fn try_lock_pending_flushes(&self) -> std::io::Result<MutexGuard<'_, Vec<std::path::PathBuf>>> {
25        self.pending_flushes.lock().map_err(|error| {
26            std::io::Error::other(format!(
27                "Vyre disk pipeline cache pending-flush lock was poisoned: {error}. Fix: discard this cache instance after a panic; continuing could lose or duplicate compiled-pipeline fsync work."
28            ))
29        })
30    }
31
32    /// Open a cache rooted at `root`.
33    ///
34    /// # Errors
35    ///
36    /// Returns when the root directory cannot be created.
37    pub fn open(root: impl Into<std::path::PathBuf>) -> std::io::Result<Self> {
38        let root = root.into();
39        std::fs::create_dir_all(&root)?;
40        Ok(Self {
41            root,
42            pending_flushes: std::sync::Mutex::new(Vec::new()),
43        })
44    }
45
46    /// Default cache directory.
47    #[must_use]
48    pub fn default_root() -> std::path::PathBuf {
49        if let Some(xdg) = std::env::var_os("XDG_CACHE_HOME") {
50            return std::path::PathBuf::from(xdg).join("vyre").join("pipelines");
51        }
52        if let Some(home) = std::env::var_os("HOME") {
53            #[cfg(target_os = "macos")]
54            {
55                return std::path::PathBuf::from(home)
56                    .join("Library")
57                    .join("Caches")
58                    .join("vyre")
59                    .join("pipelines");
60            }
61            #[cfg(not(target_os = "macos"))]
62            {
63                return std::path::PathBuf::from(home)
64                    .join(".cache")
65                    .join("vyre")
66                    .join("pipelines");
67            }
68        }
69        if let Some(appdata) = std::env::var_os("LOCALAPPDATA") {
70            return std::path::PathBuf::from(appdata)
71                .join("vyre")
72                .join("pipelines");
73        }
74        std::path::PathBuf::from("./vyre-cache/pipelines")
75    }
76
77    /// Derive the cache path for a program digest and device fingerprint.
78    #[must_use]
79    pub fn path_for(
80        &self,
81        program_digest: [u8; 32],
82        fingerprint: PipelineDeviceFingerprint,
83    ) -> std::path::PathBuf {
84        let key = fingerprint.cache_key(program_digest);
85        let mut file_name = hex_encode(&key);
86        let mut path = self.root.join(&file_name[..2]);
87        file_name.push_str(".bin");
88        path.push(file_name);
89        path
90    }
91
92    /// Read a cached blob. Returns `None` on a miss.
93    ///
94    /// # Errors
95    ///
96    /// Returns when an existing entry cannot be read.
97    pub fn read(
98        &self,
99        program_digest: [u8; 32],
100        fingerprint: PipelineDeviceFingerprint,
101    ) -> std::io::Result<Option<Vec<u8>>> {
102        let path = self.path_for(program_digest, fingerprint);
103        match read_bounded(&path, MAX_DISK_PIPELINE_BLOB_BYTES) {
104            Ok(bytes) => Ok(Some(bytes)),
105            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
106            Err(error) => Err(error),
107        }
108    }
109
110    /// Write a cache blob with atomic install.
111    ///
112    /// # Errors
113    ///
114    /// Returns when the entry is oversized or cannot be written.
115    pub fn write(
116        &self,
117        program_digest: [u8; 32],
118        fingerprint: PipelineDeviceFingerprint,
119        bytes: &[u8],
120    ) -> std::io::Result<()> {
121        if bytes.len() as u64 > MAX_DISK_PIPELINE_BLOB_BYTES {
122            return Err(std::io::Error::new(
123                std::io::ErrorKind::InvalidInput,
124                format!("pipeline cache blob exceeds {MAX_DISK_PIPELINE_BLOB_BYTES} byte limit"),
125            ));
126        }
127        let path = self.path_for(program_digest, fingerprint);
128        if let Some(parent) = path.parent() {
129            std::fs::create_dir_all(parent)?;
130        }
131        let tmp = self.tmp_path_for(&path);
132        let write_result = (|| -> std::io::Result<()> {
133            let mut file = std::fs::File::create(&tmp)?;
134            use std::io::Write as _;
135            file.write_all(bytes)?;
136            drop(file);
137            std::fs::rename(&tmp, &path)
138        })();
139        if write_result.is_err() {
140            remove_failed_atomic_write(&tmp)?;
141        }
142        write_result?;
143        self.try_lock_pending_flushes()?.push(path);
144        Ok(())
145    }
146
147    /// Durably flush entries written by [`Self::write`].
148    ///
149    /// # Errors
150    ///
151    /// Returns when a pending path cannot be flushed.
152    pub fn flush(&self) -> std::io::Result<()> {
153        let paths = {
154            let mut pending = self.try_lock_pending_flushes()?;
155            pending.sort();
156            pending.dedup();
157            std::mem::take(&mut *pending)
158        };
159        if let Err(error) = flush_paths(&paths) {
160            self.try_lock_pending_flushes()?.extend(paths);
161            return Err(error);
162        }
163        Ok(())
164    }
165
166    /// Remove entries selected by an impact mask.
167    ///
168    /// # Errors
169    ///
170    /// Returns when an impacted entry exists but cannot be removed.
171    pub fn invalidate_impacted(
172        &self,
173        impact_mask: &[u32],
174        program_digests: &[[u8; 32]],
175        fingerprint: PipelineDeviceFingerprint,
176    ) -> std::io::Result<()> {
177        for (index, &is_impacted) in impact_mask.iter().enumerate() {
178            if is_impacted != 0 {
179                if let Some(&digest) = program_digests.get(index) {
180                    let path = self.path_for(digest, fingerprint);
181                    if path.exists() {
182                        std::fs::remove_file(path)?;
183                    }
184                }
185            }
186        }
187        Ok(())
188    }
189
190    /// Root directory used by this cache.
191    #[must_use]
192    pub fn root(&self) -> &std::path::Path {
193        &self.root
194    }
195
196    fn tmp_path_for(&self, path: &std::path::Path) -> std::path::PathBuf {
197        static TMP_COUNTER: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
198        let tmp_id = TMP_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
199        path.with_extension(format!("bin.tmp.{}.{}", std::process::id(), tmp_id))
200    }
201}
202
203fn remove_failed_atomic_write(path: &std::path::Path) -> std::io::Result<()> {
204    match std::fs::remove_file(path) {
205        Ok(()) => Ok(()),
206        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
207        Err(error) => Err(error),
208    }
209}
210
211fn read_bounded(path: &std::path::Path, max_bytes: u64) -> std::io::Result<Vec<u8>> {
212    use std::io::Read as _;
213
214    let mut file = std::fs::File::open(path)?;
215    let metadata = file.metadata()?;
216    if metadata.len() > max_bytes {
217        return Err(std::io::Error::new(
218            std::io::ErrorKind::InvalidData,
219            format!("pipeline cache blob exceeds {max_bytes} byte limit"),
220        ));
221    }
222    let byte_capacity = usize::try_from(metadata.len()).map_err(|error| {
223        std::io::Error::new(
224            std::io::ErrorKind::InvalidData,
225            format!(
226                "pipeline cache blob length {} does not fit usize: {error}",
227                metadata.len()
228            ),
229        )
230    })?;
231    let mut bytes = Vec::new();
232    crate::allocation::try_reserve_vec_to_capacity(&mut bytes, byte_capacity).map_err(|error| {
233        std::io::Error::new(
234            std::io::ErrorKind::OutOfMemory,
235            format!(
236                "pipeline cache bounded read could not reserve {byte_capacity} byte(s): {error}. Fix: lower the pipeline cache blob limit or evict oversized entries."
237            ),
238        )
239    })?;
240    file.by_ref().take(max_bytes + 1).read_to_end(&mut bytes)?;
241    if bytes.len() as u64 > max_bytes {
242        return Err(std::io::Error::new(
243            std::io::ErrorKind::InvalidData,
244            format!("pipeline cache blob exceeded {max_bytes} byte bounded read limit"),
245        ));
246    }
247    Ok(bytes)
248}
249
250fn flush_paths(paths: &[std::path::PathBuf]) -> std::io::Result<()> {
251    let mut parents = Vec::new();
252    crate::allocation::try_reserve_vec_to_capacity(&mut parents, paths.len()).map_err(|error| {
253        std::io::Error::new(
254            std::io::ErrorKind::OutOfMemory,
255            format!(
256                "pipeline cache flush could not reserve {} parent path slot(s): {error}. Fix: flush fewer cache paths per batch.",
257                paths.len()
258            ),
259        )
260    })?;
261    sync_files_bounded(
262        paths,
263        std::fs::File::sync_data,
264        "disk cache file sync worker panicked",
265    )?;
266    for path in paths {
267        if let Some(parent) = path.parent() {
268            parents.push(parent.to_path_buf());
269        }
270    }
271    parents.sort();
272    parents.dedup();
273    sync_parent_dirs(&parents)?;
274    Ok(())
275}
276
277#[cfg(unix)]
278fn sync_parent_dirs(parents: &[std::path::PathBuf]) -> std::io::Result<()> {
279    sync_files_bounded(
280        parents,
281        std::fs::File::sync_all,
282        "disk cache dir sync worker panicked",
283    )
284}
285
286#[cfg(not(unix))]
287fn sync_parent_dirs(_parents: &[std::path::PathBuf]) -> std::io::Result<()> {
288    Ok(())
289}
290
291fn sync_files_bounded(
292    paths: &[std::path::PathBuf],
293    sync: fn(&std::fs::File) -> std::io::Result<()>,
294    panic_message: &'static str,
295) -> std::io::Result<()> {
296    if paths.is_empty() {
297        return Ok(());
298    }
299    let workers = std::thread::available_parallelism()
300        .map(usize::from)
301        .unwrap_or(1)
302        .clamp(1, 16);
303    for chunk in paths.chunks(workers) {
304        std::thread::scope(|scope| {
305            let mut handles = Vec::new();
306            crate::allocation::try_reserve_vec_to_capacity(&mut handles, chunk.len()).map_err(|error| {
307                std::io::Error::new(
308                    std::io::ErrorKind::OutOfMemory,
309                    format!(
310                        "pipeline cache sync could not reserve {} worker handle(s): {error}. Fix: lower pipeline cache sync fan-out.",
311                        chunk.len()
312                    ),
313                )
314            })?;
315            for path in chunk {
316                handles.push(scope.spawn(move || {
317                    let file = std::fs::File::open(path)?;
318                    sync(&file)
319                }));
320            }
321            for handle in handles {
322                handle
323                    .join()
324                    .map_err(|_| std::io::Error::other(panic_message))??;
325            }
326            Ok::<(), std::io::Error>(())
327        })?;
328    }
329    Ok(())
330}
331
332/// Capability bits that participate in pipeline-cache identity.
333///
334/// Two otherwise-identical pipelines compiled with different
335/// `PipelineFeatureFlags` produce different cache keys  -  a pipeline
336/// that assumed subgroup-op support cannot be reused on an adapter
337/// that does not expose subgroup ops even if the shader bytes match.
338///
339/// Encoded as a bitfield so the wire form is compact and trivially
340/// hashable. Bits `0x01..0x80` are allocated here; higher bits are
341/// reserved for additive backend capability flags.
342#[derive(
343    Copy, Clone, Debug, Default, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize,
344)]
345pub struct PipelineFeatureFlags(pub u32);
346
347impl PipelineFeatureFlags {
348    /// Pipeline was compiled against a lowering that emits subgroup /
349    /// wave intrinsics.
350    pub const SUBGROUP_OPS: Self = Self(1 << 0);
351    /// Pipeline was compiled with native `f16` support.
352    pub const F16: Self = Self(1 << 1);
353    /// Pipeline was compiled with native `bf16` support.
354    pub const BF16: Self = Self(1 << 2);
355    /// Pipeline was compiled with tensor-core / matrix-engine
356    /// intrinsics enabled.
357    pub const TENSOR_CORES: Self = Self(1 << 3);
358    /// Pipeline expects an async-compute queue at dispatch time.
359    pub const ASYNC_COMPUTE: Self = Self(1 << 4);
360    /// Pipeline expects push-constant support at dispatch time.
361    pub const PUSH_CONSTANTS: Self = Self(1 << 5);
362    /// Pipeline emits indirect-dispatch commands.
363    pub const INDIRECT_DISPATCH: Self = Self(1 << 6);
364    /// Pipeline was compiled for speculative (fused prefilter+confirmer)
365    /// dispatch.
366    pub const SPECULATIVE: Self = Self(1 << 7);
367    /// Pipeline was compiled for persistent-thread (device-side work queue)
368    /// dispatch.
369    pub const PERSISTENT_THREAD: Self = Self(1 << 8);
370
371    /// Empty flag set.
372    #[must_use]
373    pub const fn empty() -> Self {
374        Self(0)
375    }
376
377    /// Contains at least every bit of `other`.
378    #[must_use]
379    pub const fn contains(self, other: Self) -> bool {
380        (self.0 & other.0) == other.0
381    }
382
383    /// Union of two flag sets.
384    #[must_use]
385    pub const fn union(self, other: Self) -> Self {
386        Self(self.0 | other.0)
387    }
388
389    /// Raw bit representation.
390    #[must_use]
391    pub const fn bits(self) -> u32 {
392        self.0
393    }
394}
395
396/// Versioned pipeline-cache key shared by every backend.
397///
398/// Replaces the pre-0.6 pattern of using a raw blake3 hash as the key.
399/// A raw hash is not robust: two pipelines that should miss (different
400/// bind-group layout, different push-constant size, different
401/// workgroup-size selection) hashed identically because the hash
402/// covered the shader source only. Silent cache hits against a
403/// non-equivalent pipeline are a correctness hazard (wrong bind-group
404/// layout binds undefined data; wrong workgroup-size launches beyond
405/// guarantees).
406///
407/// `#[non_exhaustive]` is enforced at the type level via the private
408/// `__phantom` field: external callers construct keys through
409/// [`PipelineCacheKey::new`] and cannot match exhaustively, so additive
410/// key fields do not break downstream matches.
411#[derive(Clone, Debug, Eq, PartialEq, Hash)]
412
413pub struct PipelineCacheKey {
414    /// Key format version. Bumped to invalidate every cache entry
415    /// without an API break.
416    pub version: u32,
417    /// blake3 hash of the canonical backend pipeline-source bytes.
418    pub shader_hash: [u8; 32],
419    /// Structural hash of the bind-group layout descriptors. Not the
420    /// backend handle; the bytes that describe slot count, types,
421    /// visibility, and access modes per bind group.
422    pub bind_group_layout_hash: [u8; 32],
423    /// Push-constant range in bytes. Included so a pipeline compiled
424    /// for 16 B push constants never reuses against a layout that
425    /// expects 32 B.
426    pub push_constant_size: u32,
427    /// Workgroup-size `[x, y, z]` the pipeline was specialized for.
428    pub workgroup_size: [u32; 3],
429    /// Feature-flag bits the pipeline assumes at dispatch time.
430    pub feature_flags: PipelineFeatureFlags,
431    /// Backend identity. Prevents pipelines from different backends from
432    /// colliding when they happen to produce identical shader hashes.
433    pub backend_id: BackendId,
434    /// Reserved private field so `PipelineCacheKey` cannot be
435    /// constructed by structural literal (forward-compatibility lever).
436    #[allow(dead_code)]
437    __phantom: core::marker::PhantomData<()>,
438}
439
440impl PipelineCacheKey {
441    /// Construct a key at the current version.
442    #[must_use]
443    #[allow(clippy::too_many_arguments)]
444    pub fn new(
445        shader_hash: [u8; 32],
446        bind_group_layout_hash: [u8; 32],
447        push_constant_size: u32,
448        workgroup_size: [u32; 3],
449        feature_flags: PipelineFeatureFlags,
450        backend_id: BackendId,
451    ) -> Self {
452        Self {
453            version: CURRENT_PIPELINE_CACHE_KEY_VERSION,
454            shader_hash,
455            bind_group_layout_hash,
456            push_constant_size,
457            workgroup_size,
458            feature_flags,
459            backend_id,
460            __phantom: core::marker::PhantomData,
461        }
462    }
463}
464
465/// Shared in-memory identity for a backend-compiled pipeline.
466///
467/// Backends may keep their own cache maps and compiled handles, but this object
468/// keeps the identity facts single-sourced: normalized Program digest, dispatch
469/// policy digest, device/runtime fingerprint, and the final digest used as the
470/// lookup key.
471#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
472pub struct PipelineCacheIdentity {
473    /// Final tuple-boundary-preserving lookup digest.
474    pub digest: [u8; 32],
475    /// Normalized Program digest used by backend pipeline caches.
476    pub program_digest: [u8; 32],
477    /// Dispatch policy fields that alter generated backend code.
478    pub policy_digest: [u8; 32],
479    /// Backend/device/runtime identity participating in the final digest.
480    pub device_fingerprint: PipelineDeviceFingerprint,
481}
482
483impl PipelineCacheIdentity {
484    /// Build identity from already-computed shared components.
485    #[must_use]
486    pub fn from_parts(
487        program_digest: [u8; 32],
488        policy_digest: [u8; 32],
489        device_fingerprint: PipelineDeviceFingerprint,
490    ) -> Self {
491        let mut hasher = blake3::Hasher::new();
492        hasher.update(b"vyre-pipeline-cache-identity-v1\0program\0");
493        hasher.update(&program_digest);
494        hasher.update(b"\0policy\0");
495        hasher.update(&policy_digest);
496        hasher.update(b"\0vendor\0");
497        hasher.update(&device_fingerprint.vendor.to_le_bytes());
498        hasher.update(b"\0device\0");
499        hasher.update(&device_fingerprint.device.to_le_bytes());
500        hasher.update(b"\0driver\0");
501        hasher.update(&device_fingerprint.driver_digest);
502        Self {
503            digest: *hasher.finalize().as_bytes(),
504            program_digest,
505            policy_digest,
506            device_fingerprint,
507        }
508    }
509
510    /// Build identity from a public Program and dispatch config.
511    ///
512    /// # Errors
513    ///
514    /// Returns when the Program cannot be serialized into stable cache
515    /// identity. Callers should surface the error through their backend error
516    /// type instead of hashing invalid IR lossy.
517    pub fn try_from_program(
518        program: &Program,
519        config: &DispatchConfig,
520        device_fingerprint: PipelineDeviceFingerprint,
521    ) -> Result<Self, String> {
522        let program_digest = try_normalized_program_cache_digest(program)?;
523        let policy_digest = dispatch_policy_cache_digest(config);
524        Ok(Self::from_parts(
525            program_digest,
526            policy_digest,
527            device_fingerprint,
528        ))
529    }
530}
531
532/// Evidence a backend can provide when a pipeline-cache lookup misses.
533///
534/// The classifier is intentionally backend-neutral: a concrete driver keeps its
535/// own fast cache key, but records enough adjacent identity facts to explain why
536/// a miss happened without duplicating per-backend reason logic.
537#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
538pub struct PipelineCacheMissEvidence {
539    /// Number of compiled pipeline entries present at lookup time.
540    pub total_entries: usize,
541    /// Entries whose normalized Program digest matched the requested program.
542    pub same_program_entries: usize,
543    /// Entries whose Program digest and dispatch policy digest both matched.
544    pub same_program_and_policy_entries: usize,
545    /// Entries whose Program digest, dispatch policy digest, and
546    /// device/runtime fingerprint all matched.
547    pub same_program_policy_and_device_entries: usize,
548}
549
550impl PipelineCacheMissEvidence {
551    /// Build miss evidence from cached identities adjacent to the requested key.
552    #[must_use]
553    pub fn from_identities<'a>(
554        cached: impl Iterator<Item = &'a PipelineCacheIdentity>,
555        requested: &PipelineCacheIdentity,
556    ) -> Self {
557        let mut evidence = Self::default();
558        for identity in cached {
559            evidence.total_entries += 1;
560            if identity.program_digest == requested.program_digest {
561                evidence.same_program_entries += 1;
562                if identity.policy_digest == requested.policy_digest {
563                    evidence.same_program_and_policy_entries += 1;
564                    if identity.device_fingerprint == requested.device_fingerprint {
565                        evidence.same_program_policy_and_device_entries += 1;
566                    }
567                }
568            }
569        }
570        evidence
571    }
572}
573
574/// Backend-neutral cache-miss reason for operator telemetry.
575#[derive(Copy, Clone, Debug, Eq, PartialEq)]
576pub enum PipelineCacheMissReason {
577    /// The backend had no compiled entries at lookup time.
578    EmptyCache,
579    /// Existing entries were for different normalized programs.
580    ProgramChanged,
581    /// The same program existed, but dispatch policy changed the generated
582    /// pipeline identity.
583    DispatchPolicyChanged,
584    /// Program and dispatch policy matched, but device/runtime identity changed.
585    DeviceOrRuntimeChanged,
586    /// The supplied evidence says adjacent identity matched but the final key
587    /// still missed; this catches future key fields and malformed bookkeeping.
588    KeyAbsent,
589}
590
591impl PipelineCacheMissReason {
592    /// Classify a miss from backend-supplied adjacent identity evidence.
593    #[must_use]
594    pub const fn classify(evidence: PipelineCacheMissEvidence) -> Self {
595        if evidence.total_entries == 0 {
596            Self::EmptyCache
597        } else if evidence.same_program_entries == 0 {
598            Self::ProgramChanged
599        } else if evidence.same_program_and_policy_entries == 0 {
600            Self::DispatchPolicyChanged
601        } else if evidence.same_program_policy_and_device_entries == 0 {
602            Self::DeviceOrRuntimeChanged
603        } else {
604            Self::KeyAbsent
605        }
606    }
607
608    /// Classify a miss directly from cached pipeline identities.
609    #[must_use]
610    pub fn classify_identities<'a>(
611        cached: impl Iterator<Item = &'a PipelineCacheIdentity>,
612        requested: &PipelineCacheIdentity,
613    ) -> Self {
614        Self::classify(PipelineCacheMissEvidence::from_identities(
615            cached, requested,
616        ))
617    }
618
619    /// Stable metric suffix for backend metric snapshots.
620    #[must_use]
621    pub const fn metric_suffix(self) -> &'static str {
622        match self {
623            Self::EmptyCache => "empty_cache",
624            Self::ProgramChanged => "program_changed",
625            Self::DispatchPolicyChanged => "dispatch_policy_changed",
626            Self::DeviceOrRuntimeChanged => "device_or_runtime_changed",
627            Self::KeyAbsent => "key_absent",
628        }
629    }
630}
631
632#[cfg(test)]
633mod pipeline_cache_key_tests {
634    use super::*;
635
636    fn hash32(byte: u8) -> [u8; 32] {
637        [byte; 32]
638    }
639
640    #[test]
641    fn different_workgroup_size_differs() {
642        let a = PipelineCacheKey::new(
643            hash32(1),
644            hash32(2),
645            0,
646            [64, 1, 1],
647            PipelineFeatureFlags::empty(),
648            BackendId::from("backend-a"),
649        );
650        let b = PipelineCacheKey::new(
651            hash32(1),
652            hash32(2),
653            0,
654            [128, 1, 1],
655            PipelineFeatureFlags::empty(),
656            BackendId::from("backend-a"),
657        );
658        assert_ne!(a, b);
659    }
660
661    #[test]
662    fn different_feature_flags_differ() {
663        let a = PipelineCacheKey::new(
664            hash32(1),
665            hash32(2),
666            0,
667            [1, 1, 1],
668            PipelineFeatureFlags::empty(),
669            BackendId::from("backend-a"),
670        );
671        let b = PipelineCacheKey::new(
672            hash32(1),
673            hash32(2),
674            0,
675            [1, 1, 1],
676            PipelineFeatureFlags::SUBGROUP_OPS,
677            BackendId::from("backend-a"),
678        );
679        assert_ne!(a, b);
680    }
681
682    #[test]
683    fn different_backend_id_differs() {
684        let a = PipelineCacheKey::new(
685            hash32(1),
686            hash32(2),
687            0,
688            [1, 1, 1],
689            PipelineFeatureFlags::empty(),
690            BackendId::from("backend-a"),
691        );
692        let b = PipelineCacheKey::new(
693            hash32(1),
694            hash32(2),
695            0,
696            [1, 1, 1],
697            PipelineFeatureFlags::empty(),
698            BackendId::from("backend-b"),
699        );
700        assert_ne!(a, b);
701    }
702
703    #[test]
704    fn flag_containment_is_correct() {
705        let a = PipelineFeatureFlags::SUBGROUP_OPS.union(PipelineFeatureFlags::F16);
706        assert!(a.contains(PipelineFeatureFlags::SUBGROUP_OPS));
707        assert!(a.contains(PipelineFeatureFlags::F16));
708        assert!(!a.contains(PipelineFeatureFlags::TENSOR_CORES));
709    }
710
711    #[test]
712    fn version_is_current() {
713        let k = PipelineCacheKey::new(
714            hash32(1),
715            hash32(2),
716            0,
717            [1, 1, 1],
718            PipelineFeatureFlags::empty(),
719            BackendId::from("backend-a"),
720        );
721        assert_eq!(k.version, CURRENT_PIPELINE_CACHE_KEY_VERSION);
722    }
723
724    #[test]
725    fn shared_cache_identity_separates_program_policy_and_device_facts() {
726        let program_a = hash32(1);
727        let program_b = hash32(2);
728        let policy_a = hash32(3);
729        let policy_b = hash32(4);
730        let device_a = PipelineDeviceFingerprint::from_parts(1, 2, "driver-a", "runtime-a");
731        let device_b = PipelineDeviceFingerprint::from_parts(1, 2, "driver-a", "runtime-b");
732
733        let base = PipelineCacheIdentity::from_parts(program_a, policy_a, device_a);
734
735        assert_eq!(base.program_digest, program_a);
736        assert_eq!(base.policy_digest, policy_a);
737        assert_eq!(base.device_fingerprint, device_a);
738        assert_ne!(
739            base.digest,
740            PipelineCacheIdentity::from_parts(program_b, policy_a, device_a).digest,
741            "Fix: shared pipeline cache identity must include the normalized Program digest."
742        );
743        assert_ne!(
744            base.digest,
745            PipelineCacheIdentity::from_parts(program_a, policy_b, device_a).digest,
746            "Fix: shared pipeline cache identity must include dispatch policy as its own tuple field."
747        );
748        assert_ne!(
749            base.digest,
750            PipelineCacheIdentity::from_parts(program_a, policy_a, device_b).digest,
751            "Fix: shared pipeline cache identity must include device/runtime fingerprint facts."
752        );
753    }
754
755    #[test]
756    fn miss_reason_classifies_adjacent_identity_evidence() {
757        assert_eq!(
758            PipelineCacheMissReason::classify(PipelineCacheMissEvidence {
759                total_entries: 0,
760                same_program_entries: 0,
761                same_program_and_policy_entries: 0,
762                same_program_policy_and_device_entries: 0,
763            }),
764            PipelineCacheMissReason::EmptyCache
765        );
766        assert_eq!(
767            PipelineCacheMissReason::classify(PipelineCacheMissEvidence {
768                total_entries: 3,
769                same_program_entries: 0,
770                same_program_and_policy_entries: 0,
771                same_program_policy_and_device_entries: 0,
772            }),
773            PipelineCacheMissReason::ProgramChanged
774        );
775        assert_eq!(
776            PipelineCacheMissReason::classify(PipelineCacheMissEvidence {
777                total_entries: 3,
778                same_program_entries: 2,
779                same_program_and_policy_entries: 0,
780                same_program_policy_and_device_entries: 0,
781            }),
782            PipelineCacheMissReason::DispatchPolicyChanged
783        );
784        assert_eq!(
785            PipelineCacheMissReason::classify(PipelineCacheMissEvidence {
786                total_entries: 3,
787                same_program_entries: 2,
788                same_program_and_policy_entries: 1,
789                same_program_policy_and_device_entries: 0,
790            }),
791            PipelineCacheMissReason::DeviceOrRuntimeChanged
792        );
793        assert_eq!(
794            PipelineCacheMissReason::classify(PipelineCacheMissEvidence {
795                total_entries: 3,
796                same_program_entries: 2,
797                same_program_and_policy_entries: 1,
798                same_program_policy_and_device_entries: 1,
799            }),
800            PipelineCacheMissReason::KeyAbsent
801        );
802    }
803
804    #[test]
805    fn miss_reason_metric_suffixes_are_stable_snake_case() {
806        assert_eq!(
807            PipelineCacheMissReason::EmptyCache.metric_suffix(),
808            "empty_cache"
809        );
810        assert_eq!(
811            PipelineCacheMissReason::ProgramChanged.metric_suffix(),
812            "program_changed"
813        );
814        assert_eq!(
815            PipelineCacheMissReason::DispatchPolicyChanged.metric_suffix(),
816            "dispatch_policy_changed"
817        );
818        assert_eq!(
819            PipelineCacheMissReason::DeviceOrRuntimeChanged.metric_suffix(),
820            "device_or_runtime_changed"
821        );
822        assert_eq!(
823            PipelineCacheMissReason::KeyAbsent.metric_suffix(),
824            "key_absent"
825        );
826    }
827
828    #[test]
829    fn miss_reason_classifies_cached_shared_identities() {
830        let program = hash32(1);
831        let other_program = hash32(2);
832        let policy = hash32(3);
833        let other_policy = hash32(4);
834        let device = PipelineDeviceFingerprint::from_parts(1, 2, "driver-a", "runtime-a");
835        let other_device = PipelineDeviceFingerprint::from_parts(1, 2, "driver-a", "runtime-b");
836        let requested = PipelineCacheIdentity::from_parts(program, policy, device);
837
838        assert_eq!(
839            PipelineCacheMissReason::classify_identities([].iter(), &requested),
840            PipelineCacheMissReason::EmptyCache
841        );
842        assert_eq!(
843            PipelineCacheMissReason::classify_identities(
844                [PipelineCacheIdentity::from_parts(
845                    other_program,
846                    policy,
847                    device
848                )]
849                .iter(),
850                &requested,
851            ),
852            PipelineCacheMissReason::ProgramChanged
853        );
854        assert_eq!(
855            PipelineCacheMissReason::classify_identities(
856                [PipelineCacheIdentity::from_parts(
857                    program,
858                    other_policy,
859                    device
860                )]
861                .iter(),
862                &requested,
863            ),
864            PipelineCacheMissReason::DispatchPolicyChanged
865        );
866        assert_eq!(
867            PipelineCacheMissReason::classify_identities(
868                [PipelineCacheIdentity::from_parts(
869                    program,
870                    policy,
871                    other_device
872                )]
873                .iter(),
874                &requested,
875            ),
876            PipelineCacheMissReason::DeviceOrRuntimeChanged
877        );
878        assert_eq!(
879            PipelineCacheMissReason::classify_identities(
880                [PipelineCacheIdentity::from_parts(program, policy, device)].iter(),
881                &requested,
882            ),
883            PipelineCacheMissReason::KeyAbsent
884        );
885    }
886
887    #[test]
888    fn poisoned_pending_flush_lock_returns_structured_error() {
889        let cache = Arc::new(DiskPipelineCache {
890            root: std::env::temp_dir(),
891            pending_flushes: std::sync::Mutex::new(Vec::new()),
892        });
893        let poisoned = Arc::clone(&cache);
894        let _ = std::thread::spawn(move || {
895            let _guard = poisoned
896                .try_lock_pending_flushes()
897                .expect("Fix: first pending-flush lock acquisition should succeed");
898            panic!("poison disk pipeline cache pending flushes");
899        })
900        .join();
901
902        let error = cache
903            .flush()
904            .expect_err("poisoned disk pipeline cache must return an io error");
905        let message = error.to_string();
906        assert!(
907            message.contains("pending-flush lock was poisoned"),
908            "{message}"
909        );
910    }
911}
vyre_driver/pipeline/cache.rs

vyre_driver/pipeline/
cache.rs
