Phase 5 tranche 1: ring reader protocol

crates/luminal-vgd-core/src/ring.rs

@@ -135,6 +135,26 @@ impl RingPolicy {
         self.slots[index] = Slot::Free;
     }
 
+    /// Absorb the host's shared-slot-state transitions (the reader marks
+    /// slots READING while it holds them and FREE when done — the shared
+    /// `SlotMetadata.state` is the only channel the host has). Call for
+    /// each slot before writer decisions so:
+    /// - a host-held slot (`READING`) is never chosen for overwrite, and
+    /// - consumed slots (`FREE`) are reused without counting a drop.
+    ///
+    /// Driver-owned transitions (Writing) and unknown values are ignored.
+    pub fn reconcile_shared(&mut self, index: usize, shared_state: u32) {
+        use luminal_driver_proto::slot_state as ss;
+        self.slots[index] = match (self.slots[index], shared_state) {
+            // Host checked the slot out for reading.
+            (Slot::Published(_), s) if s == ss::READING => Slot::Reading,
+            // Host consumed (or abandoned) the slot and released it.
+            (Slot::Published(_), s) if s == ss::FREE => Slot::Free,
+            (Slot::Reading, s) if s == ss::FREE => Slot::Free,
+            (cur, _) => cur,
+        };
+    }
+
     /// TDR/device-reset rebuild (§3.3 rule 2): all slots reset, generation
     /// bumps (host re-opens textures by name — the generation is baked into
     /// the name), but sequences CONTINUE so the host's gap detection spans
@@ -233,6 +253,50 @@ mod tests {
         assert_eq!(r.frames_dropped, 1);
     }
 
+    #[test]
+    fn reconcile_reading_protects_slot_and_free_reclaims_it() {
+        use luminal_driver_proto::slot_state as ss;
+        let mut r = RingPolicy::new(2);
+        // Publish into both slots.
+        for _ in 0..2 {
+            let w = r.writer_acquire().unwrap();
+            r.publish(w.index);
+        }
+        // Host checks out slot of seq 2 (the newest) via shared state.
+        let newest = 1; // second publish landed in slot 1
+        r.reconcile_shared(newest, ss::READING);
+        assert_eq!(r.slot(newest), Slot::Reading);
+
+        // Writer must overwrite the OTHER slot, never the host-held one.
+        let w = r.writer_acquire().unwrap();
+        assert_ne!(w.index, newest);
+        r.publish(w.index);
+
+        // Host finishes: slot becomes Free and is reused WITHOUT counting
+        // a drop (the frame was consumed, not lost).
+        let drops_before = r.frames_dropped;
+        r.reconcile_shared(newest, ss::FREE);
+        assert_eq!(r.slot(newest), Slot::Free);
+        let w = r.writer_acquire().unwrap();
+        assert_eq!(w.index, newest);
+        assert_eq!(w.overwrote, None);
+        assert_eq!(r.frames_dropped, drops_before);
+    }
+
+    #[test]
+    fn reconcile_ignores_driver_owned_and_bogus_states() {
+        use luminal_driver_proto::slot_state as ss;
+        let mut r = RingPolicy::new(2);
+        let w = r.writer_acquire().unwrap();
+        // Mid-write: shared state says WRITING (we wrote it) — no change.
+        r.reconcile_shared(w.index, ss::WRITING);
+        assert_eq!(r.slot(w.index), Slot::Writing);
+        r.publish(w.index);
+        // Bogus shared value: ignored.
+        r.reconcile_shared(w.index, 0xDEAD);
+        assert!(matches!(r.slot(w.index), Slot::Published(_)));
+    }
+
     #[test]
     fn writer_abort_reverts_and_counts_without_a_sequence() {
         let mut r = RingPolicy::new(2);

crates/luminal-vgd-driver/src/shell/ring.rs

@@ -226,6 +226,13 @@ impl RingSection {
         }
     }
 
+    /// Acquire-load one slot's shared state (the host writes READING/FREE
+    /// transitions here — feed into `RingPolicy::reconcile_shared`).
+    pub fn slot_state(&self, index: usize) -> u32 {
+        let slot = self.slot_ptr(index);
+        unsafe { AtomicU32::from_ptr(&mut (*slot).state).load(Ordering::Acquire) }
+    }
+
     /// Reset one slot to FREE (aborted write).
     pub fn reset_slot_free(&self, index: usize) {
         let slot = self.slot_ptr(index);

crates/luminal-vgd-driver/src/shell/swapchain.rs

@@ -415,6 +415,15 @@ fn publish_frame(
         );
     }
 
+    // Absorb the host's reader transitions (READING claims, FREE
+    // releases) so writer decisions respect checked-out slots and reuse
+    // consumed ones without counting drops.
+    if let Some(s) = &ring.section {
+        for index in 0..ring.policy.slot_count() {
+            ring.policy.reconcile_shared(index, s.slot_state(index));
+        }
+    }
+
     let Some(writer) = ring.policy.writer_acquire() else {
         // Reader holds everything (pathological): drop, never block.
         if let Some(s) = &ring.section {

crates/luminal-vgd-host/src/device.rs

@@ -29,7 +29,7 @@ use windows_sys::Win32::Storage::FileSystem::{
     CreateFileW, FILE_SHARE_READ, FILE_SHARE_WRITE, OPEN_EXISTING,
 };
 use windows_sys::Win32::System::Memory::{
-    MapViewOfFile, OpenFileMappingW, UnmapViewOfFile, FILE_MAP_READ,
+    MapViewOfFile, OpenFileMappingW, UnmapViewOfFile, FILE_MAP_READ, FILE_MAP_WRITE,
     MEMORY_MAPPED_VIEW_ADDRESS,
 };
 use windows_sys::Win32::System::IO::DeviceIoControl;
@@ -186,10 +186,13 @@ impl VgdDevice {
     }
 }
 
-/// Read-only mapping of a monitor's shared ring section.
+/// Mapping of a monitor's shared ring section. Reads are volatile
+/// snapshots; the ONLY writes the host ever performs are the reader-side
+/// slot-state transitions (`PUBLISHED→READING→FREE`, atomic CAS) — the
+/// rest of the section belongs to the driver.
 pub struct RingView {
     mapping: HANDLE,
-    view: *const u8,
+    view: *mut u8,
     slot_count: u32,
 }
 
@@ -199,45 +202,122 @@ impl Drop for RingView {
     fn drop(&mut self) {
         unsafe {
             UnmapViewOfFile(MEMORY_MAPPED_VIEW_ADDRESS {
-                Value: self.view.cast_mut().cast(),
+                Value: self.view.cast(),
             });
             CloseHandle(self.mapping);
         }
     }
 }
 
+/// A ring slot the host has checked out (shared state `READING`): the
+/// driver will not touch its texture until [`RingView::release`]. Copy of
+/// the claim-time metadata; the texture itself is opened by name on the
+/// consumer's D3D device (`names::slot_texture_name` with `generation`).
+#[derive(Clone, Copy, Debug)]
+pub struct ClaimedFrame {
+    pub index: u32,
+    pub sequence: u64,
+    pub present_qpc: u64,
+    /// Ring generation at claim time — bake into the texture name; if the
+    /// header generation no longer matches at use time, release and
+    /// re-claim (the driver rebuilt the ring).
+    pub generation: u32,
+}
+
 impl RingView {
     /// Map the ring section for `session_id` (name derived through the
     /// shared proto helper — never hand-composed).
     pub fn open(session_id: u64, slot_count: u32) -> io::Result<Self> {
         let mut name = [0u16; 64];
         names::ring_section_name(session_id, &mut name);
-        let mapping = unsafe { OpenFileMappingW(FILE_MAP_READ, 0, name.as_ptr()) };
+        let access = FILE_MAP_READ | FILE_MAP_WRITE;
+        let mapping = unsafe { OpenFileMappingW(access, 0, name.as_ptr()) };
         if mapping.is_null() {
             return Err(io::Error::last_os_error());
         }
-        let view = unsafe {
-            MapViewOfFile(mapping, FILE_MAP_READ, 0, 0, ring_section_size(slot_count))
-        };
+        let view =
+            unsafe { MapViewOfFile(mapping, access, 0, 0, ring_section_size(slot_count)) };
         if view.Value.is_null() {
             let e = io::Error::last_os_error();
             unsafe { CloseHandle(mapping) };
             return Err(e);
         }
-        Ok(Self { mapping, view: view.Value.cast::<u8>().cast_const(), slot_count })
+        Ok(Self { mapping, view: view.Value.cast::<u8>(), slot_count })
     }
 
     /// Volatile snapshot of the header (driver writes concurrently).
     pub fn header(&self) -> RingHeader {
-        unsafe { read_volatile(self.view.cast::<RingHeader>()) }
+        unsafe { read_volatile(self.view.cast::<RingHeader>().cast_const()) }
+    }
+
+    fn slot_ptr(&self, index: u32) -> *mut SlotMetadata {
+        debug_assert!(index < self.slot_count);
+        unsafe {
+            self.view
+                .add(RING_SLOTS_OFFSET + index as usize * size_of::<SlotMetadata>())
+                .cast::<SlotMetadata>()
+        }
+    }
+
+    fn slot_state_atomic(&self, index: u32) -> &core::sync::atomic::AtomicU32 {
+        unsafe { core::sync::atomic::AtomicU32::from_ptr(&mut (*self.slot_ptr(index)).state) }
     }
 
     /// Volatile snapshot of one slot's metadata.
     pub fn slot(&self, index: u32) -> Option<SlotMetadata> {
         if index >= self.slot_count {
             return None;
         }
-        let offset = RING_SLOTS_OFFSET + index as usize * size_of::<SlotMetadata>();
-        Some(unsafe { read_volatile(self.view.add(offset).cast::<SlotMetadata>()) })
+        Some(unsafe { read_volatile(self.slot_ptr(index).cast_const()) })
+    }
+
+    /// Check out the freshest published frame: pick the PUBLISHED slot
+    /// with the highest sequence and CAS it `PUBLISHED→READING`. Returns
+    /// None when nothing is published (or the driver won every race).
+    /// Streams want freshness — older published frames stay eligible for
+    /// the driver's drop-oldest overwrite.
+    pub fn claim_latest(&self) -> Option<ClaimedFrame> {
+        use core::sync::atomic::Ordering;
+        use luminal_driver_proto::slot_state as ss;
+        // Bounded retries: a failed CAS means the driver took the slot
+        // mid-claim; rescan at most once per slot.
+        for _ in 0..=self.slot_count {
+            let mut newest: Option<ClaimedFrame> = None;
+            for index in 0..self.slot_count {
+                let meta = self.slot(index)?;
+                if meta.state == ss::PUBLISHED
+                    && newest.is_none_or(|n| meta.sequence > n.sequence)
+                {
+                    newest = Some(ClaimedFrame {
+                        index,
+                        sequence: meta.sequence,
+                        present_qpc: meta.present_qpc,
+                        generation: self.header().ring_generation,
+                    });
+                }
+            }
+            let candidate = newest?;
+            if self
+                .slot_state_atomic(candidate.index)
+                .compare_exchange(ss::PUBLISHED, ss::READING, Ordering::AcqRel, Ordering::Acquire)
+                .is_ok()
+            {
+                return Some(candidate);
+            }
+        }
+        None
+    }
+
+    /// Release a claimed slot back to the driver (`READING→FREE`). Safe
+    /// to call exactly once per successful claim.
+    pub fn release(&self, index: u32) {
+        use core::sync::atomic::Ordering;
+        use luminal_driver_proto::slot_state as ss;
+        let _ = self.slot_state_atomic(index).compare_exchange(
+            ss::READING,
+            ss::FREE,
+            Ordering::AcqRel,
+            Ordering::Acquire,
+        );
     }
 }

crates/vgd-probe/src/main.rs

@@ -60,6 +60,10 @@ struct Args {
     /// (no remembered topology/disconnect state) — useful when the stable
     /// probe identity has accumulated unwanted display-settings memory.
     ephemeral: bool,
+    /// Act as a ring consumer during the hold: claim/release published
+    /// slots continuously (~5 ms cadence). Exercises the reader protocol
+    /// end to end; with a consumer draining, driver drops should stay ≈0.
+    consume: bool,
 }
 
 /// Default mode list when none is given: 4K120 preferred (LG-OLED-class
@@ -78,12 +82,14 @@ fn parse_args() -> Result<Args, String> {
         explicit_mode: None,
         hold_secs: 15,
         ephemeral: false,
+        consume: false,
     };
     let mut it = std::env::args().skip(1);
     while let Some(a) = it.next() {
         match a.as_str() {
             "status" => args.status_only = true,
             "--ephemeral" => args.ephemeral = true,
+            "--consume" => args.consume = true,
             "--hold" => {
                 let v = it.next().ok_or("--hold needs a value")?;
                 args.hold_secs = v.parse().map_err(|_| format!("bad --hold value: {v}"))?;
@@ -247,8 +253,24 @@ fn main() -> ExitCode {
     let mut first_seq = None;
     let mut prev_seq = 0u64;
     let mut prev_heartbeat = 0u64;
+    let mut consumed_total = 0u64;
     for tick in 0..args.hold_secs {
-        std::thread::sleep(Duration::from_secs(1));
+        if args.consume {
+            // Drain published slots at ~5 ms cadence for the whole tick:
+            // claim newest → (a real consumer would encode here) → release.
+            let tick_end = std::time::Instant::now() + Duration::from_secs(1);
+            while std::time::Instant::now() < tick_end {
+                if let Some(view) = &ring {
+                    while let Some(frame) = view.claim_latest() {
+                        view.release(frame.index);
+                        consumed_total += 1;
+                    }
+                }
+                std::thread::sleep(Duration::from_millis(5));
+            }
+        } else {
+            std::thread::sleep(Duration::from_secs(1));
+        }
         match dev.ping(session_id) {
             Ok(r) if r == err::OK => {}
             Ok(r) => {
@@ -270,15 +292,21 @@ fn main() -> ExitCode {
             };
             let fps = h.latest_sequence.saturating_sub(prev_seq);
             let beating = if h.driver_heartbeat_qpc != prev_heartbeat { "beating" } else { "STALE" };
+            let consumed = if args.consume {
+                format!(" consumed {consumed_total}")
+            } else {
+                String::new()
+            };
             println!(
-                "  [{:>2}s] gen {} {} seq {} (+{}/s) published {} dropped {} heartbeat {}",
+                "  [{:>2}s] gen {} {} seq {} (+{}/s) published {} dropped {}{} heartbeat {}",
                 tick + 1,
                 h.ring_generation,
                 state,
                 h.latest_sequence,
                 fps,
                 h.frames_published,
                 h.frames_dropped,
+                consumed,
                 beating
             );
             if first_seq.is_none() {