strat9_kernel/async_io/
ring.rs

1//! Ring buffer for async I/O submission and completion.
2//!
3//! Each ring consists of two pages mapped into the owning process's address
4//! space: a Submission Queue (SQ) page and a Completion Queue (CQ) page.
5//! The kernel tracks metadata (head/tail pointers, ownership) internally;
6//! only the raw SQE/CQE arrays are visible to userspace.
7
8use crate::{
9    memory::{
10        self,
11        address_space::{VmaFlags, VmaPageSize, VmaType},
12        phys_to_virt, PhysFrame,
13    },
14    sync::SpinLock,
15};
16use alloc::{sync::Arc, vec::Vec};
17use core::sync::atomic::{AtomicU32, AtomicU64, Ordering};
18
19use super::ops::{AsyncCqe, AsyncSqe, DEFAULT_RING_ENTRIES};
20
21// =============================================================================
22// Ring
23// =============================================================================
24
25/// A shared submission/completion ring for a single process.
26pub struct Ring {
27    /// Opaque handle returned to userspace.
28    pub id: u64,
29    /// Process that owns this ring (by PID).
30    pub owner_pid: u32,
31    /// Stable mapping capability for the SQ pages.
32    pub sq_mapping_cap_id: crate::capability::CapId,
33    /// Physical frame backing the SQ page.
34    pub sq_frame: PhysFrame,
35    /// Kernel virtual address of the SQ metadata header.
36    pub sq_virt: u64,
37    /// Stable mapping capability for the CQ pages.
38    pub cq_mapping_cap_id: crate::capability::CapId,
39    /// Physical frame backing the CQ page.
40    pub cq_frame: PhysFrame,
41    /// Kernel virtual address of the CQ metadata header.
42    pub cq_virt: u64,
43    /// Number of entries (power of two, min 2, max MAX_IN_FLIGHT).
44    pub entries: u32,
45    /// Number of in-flight operations (not yet completed).
46    pub in_flight: AtomicU32,
47    /// Whether this ring has been destroyed (no new ops accepted).
48    pub destroyed: AtomicU32,
49    /// Lock for pushing completions.
50    pub cq_lock: SpinLock<()>,
51    /// Completions retained temporarily when the visible CQ is full.
52    pub completion_backlog: SpinLock<Vec<AsyncCqe>>,
53    /// Order used for SQ allocation.
54    pub sq_order: u8,
55    /// Order used for CQ allocation.
56    pub cq_order: u8,
57    /// Wait queue for tasks blocked on this ring.
58    pub wq: crate::sync::WaitQueue,
59}
60
61impl Ring {
62    /// Create a new ring and map it into the calling process's address space.
63    pub fn create(pid: u32, entries: u32) -> Result<u64, RingError> {
64        if RING_REGISTRY.lock().len() >= MAX_RINGS {
65            return Err(RingError::TooManyRings);
66        }
67
68        let entries = entries.max(2).min(DEFAULT_RING_ENTRIES).next_power_of_two();
69
70        let sq_bytes = RING_META_SIZE + (entries as usize * core::mem::size_of::<AsyncSqe>());
71        let cq_bytes = RING_META_SIZE + (entries as usize * core::mem::size_of::<AsyncCqe>());
72
73        let sq_order = ((sq_bytes + 4095) / 4096)
74            .next_power_of_two()
75            .trailing_zeros() as u8;
76        let cq_order = ((cq_bytes + 4095) / 4096)
77            .next_power_of_two()
78            .trailing_zeros() as u8;
79
80        // Allocate SQ frames
81        let sq_frame =
82            crate::sync::with_irqs_disabled(|t| memory::allocate_phys_contiguous(t, sq_order))
83                .map_err(|_| RingError::Alloc)?;
84        let sq_virt = phys_to_virt(sq_frame.start_address.as_u64());
85
86        // Allocate CQ frames
87        let cq_frame = match crate::sync::with_irqs_disabled(|t| {
88            memory::allocate_phys_contiguous(t, cq_order)
89        }) {
90            Ok(f) => f,
91            Err(_) => {
92                crate::sync::with_irqs_disabled(|t| {
93                    memory::free_phys_contiguous(t, sq_frame, sq_order);
94                });
95                return Err(RingError::Alloc);
96            }
97        };
98        let cq_virt = phys_to_virt(cq_frame.start_address.as_u64());
99
100        // Initialize metadata
101        unsafe {
102            let sq = sq_virt as *mut RingMeta;
103            (*sq).entries.store(entries, Ordering::Release);
104            (*sq).mask.store(entries - 1, Ordering::Release);
105            (*sq).head.store(0, Ordering::Release);
106            (*sq).tail.store(0, Ordering::Release);
107            (*sq).flags.store(0, Ordering::Release);
108
109            let cq = cq_virt as *mut RingMeta;
110            (*cq).entries.store(entries, Ordering::Release);
111            (*cq).mask.store(entries - 1, Ordering::Release);
112            (*cq).head.store(0, Ordering::Release);
113            (*cq).tail.store(0, Ordering::Release);
114            (*cq).flags.store(0, Ordering::Release);
115        }
116
117        let ring = Arc::new(Ring {
118            id: next_ring_id(),
119            owner_pid: pid,
120            sq_mapping_cap_id: memory::allocate_mapping_cap_id(),
121            sq_frame,
122            sq_virt,
123            cq_mapping_cap_id: memory::allocate_mapping_cap_id(),
124            cq_frame,
125            cq_virt,
126            entries,
127            in_flight: AtomicU32::new(0),
128            destroyed: AtomicU32::new(0),
129            cq_lock: SpinLock::new(()),
130            completion_backlog: SpinLock::new(Vec::with_capacity(16)),
131            sq_order,
132            cq_order,
133            wq: crate::sync::WaitQueue::new(),
134        });
135
136        let id = ring.id;
137        let mut registry = RING_REGISTRY.lock();
138        if registry.len() >= MAX_RINGS {
139            return Err(RingError::TooManyRings);
140        }
141        registry.push(ring);
142        Ok(id)
143    }
144
145    /// Map the SQ and CQ buffers into the provided address space.
146    pub fn map_into_process(
147        &self,
148        addr_space: &crate::memory::AddressSpace,
149    ) -> Result<super::syscall::AsyncRingMapping, &'static str> {
150        let sq_page_count = 1usize << self.sq_order;
151        let cq_page_count = 1usize << self.cq_order;
152        let sq_phys_addrs = self.contiguous_phys_addrs(self.sq_frame, sq_page_count);
153        let cq_phys_addrs = self.contiguous_phys_addrs(self.cq_frame, cq_page_count);
154        let sq_mapping_cap_ids = alloc::vec![self.sq_mapping_cap_id; sq_page_count];
155        let cq_mapping_cap_ids = alloc::vec![self.cq_mapping_cap_id; cq_page_count];
156
157        let sq_base = addr_space
158            .find_free_vma_range(
159                crate::syscall::mmap::MMAP_BASE,
160                sq_page_count,
161                VmaPageSize::Small,
162            )
163            .ok_or("async ring: no free range for SQ")?;
164        addr_space.map_shared_frames_with_cap_ids(
165            sq_base,
166            &sq_phys_addrs,
167            Some(&sq_mapping_cap_ids),
168            VmaFlags {
169                readable: true,
170                writable: true,
171                executable: false,
172                user_accessible: true,
173            },
174            VmaType::Anonymous,
175        )?;
176
177        let cq_base = match addr_space.find_free_vma_range(
178            crate::syscall::mmap::MMAP_BASE,
179            cq_page_count,
180            VmaPageSize::Small,
181        ) {
182            Some(base) => base,
183            None => {
184                let _ = addr_space.unmap_region(sq_base, sq_page_count, VmaPageSize::Small);
185                return Err("async ring: no free range for CQ");
186            }
187        };
188
189        if let Err(err) = addr_space.map_shared_frames_with_cap_ids(
190            cq_base,
191            &cq_phys_addrs,
192            Some(&cq_mapping_cap_ids),
193            VmaFlags {
194                readable: true,
195                writable: true,
196                executable: false,
197                user_accessible: true,
198            },
199            VmaType::Anonymous,
200        ) {
201            let _ = addr_space.unmap_region(sq_base, sq_page_count, VmaPageSize::Small);
202            return Err(err);
203        }
204
205        Ok(super::syscall::AsyncRingMapping {
206            sq_base,
207            cq_base,
208            sq_size: (sq_page_count * 4096) as u64,
209            cq_size: (cq_page_count * 4096) as u64,
210            entries: self.entries,
211        })
212    }
213
214    fn contiguous_phys_addrs(&self, frame: PhysFrame, page_count: usize) -> Vec<u64> {
215        let base = frame.start_address.as_u64();
216        (0..page_count)
217            .map(|index| base + (index as u64) * 4096)
218            .collect()
219    }
220
221    /// Read the SQ tail (written by userspace, read by kernel).
222    #[inline]
223    pub fn sq_tail(&self) -> u32 {
224        self.sq_meta().tail.load(Ordering::Acquire)
225    }
226
227    /// Read the SQ head (written by kernel, read by userspace).
228    #[inline]
229    pub fn sq_head(&self) -> u32 {
230        self.sq_meta().head.load(Ordering::Acquire)
231    }
232
233    /// Advance the SQ head after processing submissions.
234    #[inline]
235    pub fn sq_advance_head(&self, count: u32) {
236        self.sq_meta().head.fetch_add(count, Ordering::Release);
237    }
238
239    /// Advance the CQ tail after pushing completions.
240    #[inline]
241    pub fn cq_advance_tail(&self, count: u32) {
242        self.cq_meta().tail.fetch_add(count, Ordering::Release);
243    }
244
245    /// Read the CQ head (written by userspace after consuming).
246    #[inline]
247    pub fn cq_head(&self) -> u32 {
248        self.cq_meta().head.load(Ordering::Acquire)
249    }
250
251    /// Get a pointer to the SQE at the given index.
252    #[inline]
253    pub unsafe fn sqe_at(&self, index: u32) -> *const AsyncSqe {
254        let base = (self.sq_virt + RING_META_SIZE as u64) as *const AsyncSqe;
255        base.add(index as usize)
256    }
257
258    /// Get a mutable pointer to the CQE at the given index.
259    #[inline]
260    pub unsafe fn cqe_at(&self, index: u32) -> *mut AsyncCqe {
261        let base = (self.cq_virt + RING_META_SIZE as u64) as *mut AsyncCqe;
262        base.add(index as usize)
263    }
264
265    /// Mark the ring as destroyed (no new submissions accepted).
266    #[inline]
267    pub fn destroy(&self) {
268        self.destroyed.store(1, Ordering::Release);
269    }
270
271    pub(crate) fn sq_meta(&self) -> &RingMeta {
272        unsafe { &*(self.sq_virt as *const RingMeta) }
273    }
274    pub(crate) fn cq_meta(&self) -> &RingMeta {
275        unsafe { &*(self.cq_virt as *const RingMeta) }
276    }
277}
278
279impl Drop for Ring {
280    fn drop(&mut self) {
281        let _ = memory::revoke_mapping_cap_id(self.sq_mapping_cap_id);
282        let _ = memory::revoke_mapping_cap_id(self.cq_mapping_cap_id);
283        crate::sync::with_irqs_disabled(|t| {
284            memory::free_phys_contiguous(t, self.sq_frame, self.sq_order);
285            memory::free_phys_contiguous(t, self.cq_frame, self.cq_order);
286        });
287    }
288}
289
290// =============================================================================
291// RingMeta : header at the start of each ring page
292// =============================================================================
293
294/// Metadata header embedded at offset 0 of each ring page.
295/// The SQE/CQE array starts at `RING_META_SIZE` bytes from the page base.
296#[repr(C)]
297pub(crate) struct RingMeta {
298    pub(crate) head: AtomicU32,    // Producer offset
299    pub(crate) tail: AtomicU32,    // Consumer offset
300    pub(crate) mask: AtomicU32,    // entries - 1 (for index wrapping)
301    pub(crate) entries: AtomicU32, // total number of slots
302    pub(crate) flags: AtomicU32,   // ring flags (NEED_WAKEUP, etc.)
303}
304
305const RING_META_SIZE: usize = 64; // leave room for future flags
306
307// =============================================================================
308// RingRegistry
309// =============================================================================
310
311const MAX_RINGS: usize = 128;
312
313static RING_REGISTRY: SpinLock<Vec<Arc<Ring>>> = SpinLock::new(Vec::new());
314
315static NEXT_RING_ID: AtomicU64 = AtomicU64::new(1);
316
317fn next_ring_id() -> u64 {
318    NEXT_RING_ID.fetch_add(1, Ordering::Relaxed)
319}
320
321/// Find a ring by its opaque id.
322pub fn find_ring(id: u64) -> Option<Arc<Ring>> {
323    let guard = RING_REGISTRY.lock();
324    guard.iter().find(|r| r.id == id).cloned()
325}
326
327/// Remove a ring from the registry and free its pages.
328pub fn destroy_ring(id: u64) -> Result<(), RingError> {
329    let mut guard = RING_REGISTRY.lock();
330    if let Some(pos) = guard.iter().position(|r| r.id == id) {
331        let ring = guard.remove(pos);
332        ring.destroy();
333        ring.wq.wake_all();
334        crate::hardware::storage::ahci::discard_deferred_async_read_completions(id);
335        Ok(())
336    } else {
337        Err(RingError::NotFound)
338    }
339}
340
341// =============================================================================
342// RingError
343// =============================================================================
344
345#[derive(Debug, Clone, Copy, PartialEq, Eq)]
346pub enum RingError {
347    Alloc,
348    NotFound,
349    TooManyRings,
350}
strat9_kernel/async_io/ring.rs

strat9_kernel/async_io/
ring.rs
