strat9_kernel/async_io/
dispatch.rs

1//! SQE dispatch : routes submitted SQEs to the appropriate backend.
2//!
3//! Called from `SYS_ASYNC_ENTER` when `to_submit > 0`.  Walks the
4//! submission queue and dispatches each SQE by opcode.
5
6use crate::{
7    ipc::message::{ipc_message_from_raw, ipc_message_to_raw, IpcMessage},
8    memory::{UserSliceRead, UserSliceWrite},
9    vfs::scheme::AsyncSubmitResult,
10};
11
12use super::{
13    complete::{inc_in_flight_for_ring, push_completion_for_ring},
14    ops::AsyncOp,
15    ring::find_ring,
16};
17
18enum DispatchOutcome {
19    CompletedInline,
20    InFlight,
21}
22
23// Errno constants (no libc dependency)
24const EBADF: i32 = -9;
25const EFAULT: i32 = -14;
26const EINVAL: i32 = -22;
27const EIO: i32 = -5;
28const ENOMSG: i32 = -42;
29const IPC_MSG_SIZE: usize = core::mem::size_of::<IpcMessage>();
30
31/// Process up to `count` pending SQEs from the given ring.
32///
33/// Returns the number of SQEs successfully processed.
34pub fn drain_submissions(ring_id: u64, count: u32) -> u32 {
35    let Some(ring) = find_ring(ring_id) else {
36        return 0;
37    };
38
39    let head = ring.sq_head();
40    let tail = ring.sq_tail();
41    let mask = ring
42        .sq_meta()
43        .mask
44        .load(core::sync::atomic::Ordering::Relaxed);
45    let available = tail.wrapping_sub(head).min(count);
46    if available == 0 {
47        return 0;
48    }
49
50    for i in 0..available {
51        let idx = (head.wrapping_add(i)) & mask;
52        let sqe = unsafe { &*ring.sqe_at(idx) };
53        match dispatch_one(sqe, &ring) {
54            Err(e) => {
55                push_completion_for_ring(&ring, sqe.user_data, e, 0);
56            }
57            Ok(DispatchOutcome::CompletedInline) => {}
58            Ok(DispatchOutcome::InFlight) => {
59                inc_in_flight_for_ring(&ring);
60            }
61        }
62    }
63
64    ring.sq_advance_head(available);
65    available
66}
67
68/// Dispatch a single SQE by opcode.
69fn dispatch_one(
70    sqe: &super::ops::AsyncSqe,
71    ring: &super::ring::Ring,
72) -> Result<DispatchOutcome, i32> {
73    let ring_id = ring.id;
74    match sqe.opcode {
75        // Nop : immediate success
76        op if op == AsyncOp::Nop as u8 => {
77            push_completion_for_ring(ring, sqe.user_data, 0, 0);
78            Ok(DispatchOutcome::CompletedInline)
79        }
80
81        // VFS reads : resolve fd and submit through AsyncScheme.
82        op if op == AsyncOp::Read as u8 => {
83            let fd = sqe.fd as u32;
84            let buf_vaddr = sqe.addr;
85            let len = sqe.len as usize;
86
87            // Resolve fd => (scheme, file_id)  (VFS pattern)
88            let task = crate::process::current_task_clone().ok_or(EIO)?;
89            let fd_table = unsafe { &*task.process.fd_table.get() };
90            let file = fd_table.get(fd).map_err(|_| EBADF)?;
91            let file_id = file.file_id();
92            let scheme_ref: &dyn crate::vfs::scheme::Scheme = &**file.scheme();
93
94            match scheme_ref.async_read(file_id, sqe.off, buf_vaddr, len, ring_id, sqe.user_data) {
95                Ok(AsyncSubmitResult::Completed(n)) => {
96                    push_completion_for_ring(ring, sqe.user_data, n, 0);
97                    Ok(DispatchOutcome::CompletedInline)
98                }
99                Ok(AsyncSubmitResult::InFlight) => Ok(DispatchOutcome::InFlight),
100                Err(err) => Err(err as i32),
101            }
102        }
103
104        // VFS writes : resolve fd and submit through AsyncScheme.
105        op if op == AsyncOp::Write as u8 => {
106            let fd = sqe.fd as u32;
107            let buf_vaddr = sqe.addr;
108            let len = sqe.len as usize;
109
110            let task = crate::process::current_task_clone().ok_or(EIO)?;
111            let fd_table = unsafe { &*task.process.fd_table.get() };
112            let file = fd_table.get(fd).map_err(|_| EBADF)?;
113            let file_id = file.file_id();
114            let scheme_ref: &dyn crate::vfs::scheme::Scheme = &**file.scheme();
115
116            match scheme_ref.async_write(file_id, sqe.off, buf_vaddr, len, ring_id, sqe.user_data) {
117                Ok(AsyncSubmitResult::Completed(n)) => {
118                    push_completion_for_ring(ring, sqe.user_data, n, 0);
119                    Ok(DispatchOutcome::CompletedInline)
120                }
121                Ok(AsyncSubmitResult::InFlight) => Ok(DispatchOutcome::InFlight),
122                Err(err) => Err(err as i32),
123            }
124        }
125
126        // IPC send : try_send message to port, push CQE with result
127        op if op == AsyncOp::IpcSend as u8 => {
128            let port_id = crate::ipc::PortId::from_u64(sqe.fd as u64);
129            let Some(port) = crate::ipc::port::get_port(port_id) else {
130                push_completion_for_ring(ring, sqe.user_data, EBADF, 0);
131                return Ok(DispatchOutcome::CompletedInline);
132            };
133            let user_msg = UserSliceRead::new(sqe.addr, core::mem::size_of::<IpcMessage>())
134                .map_err(|_| EFAULT)?;
135            let mut raw = [0u8; IPC_MSG_SIZE];
136            user_msg.copy_to(&mut raw);
137            let msg = ipc_message_from_raw(&raw);
138            let res_code = match port.try_send(msg) {
139                Ok(()) => 0,
140                Err(_) => ENOMSG,
141            };
142            push_completion_for_ring(ring, sqe.user_data, res_code, 0);
143            Ok(DispatchOutcome::CompletedInline)
144        }
145
146        // IPC recv : try_recv message from port, copy to buffer, push CQE
147        op if op == AsyncOp::IpcRecv as u8 => {
148            let port_id = crate::ipc::PortId::from_u64(sqe.fd as u64);
149            let Some(port) = crate::ipc::port::get_port(port_id) else {
150                push_completion_for_ring(ring, sqe.user_data, EBADF, 0);
151                return Ok(DispatchOutcome::CompletedInline);
152            };
153            match port.try_recv() {
154                Ok(Some(msg)) => {
155                    let user_msg =
156                        UserSliceWrite::new(sqe.addr, core::mem::size_of::<IpcMessage>())
157                            .map_err(|_| EFAULT)?;
158                    let mut raw = [0u8; IPC_MSG_SIZE];
159                    ipc_message_to_raw(&msg, &mut raw);
160                    user_msg.copy_from(&raw);
161                    push_completion_for_ring(ring, sqe.user_data, IPC_MSG_SIZE as i32, 0);
162                    Ok(DispatchOutcome::CompletedInline)
163                }
164                Ok(None) => {
165                    push_completion_for_ring(ring, sqe.user_data, ENOMSG, 0);
166                    Ok(DispatchOutcome::CompletedInline)
167                }
168                Err(_) => {
169                    push_completion_for_ring(ring, sqe.user_data, EBADF, 0);
170                    Ok(DispatchOutcome::CompletedInline)
171                }
172            }
173        }
174
175        // IPC call : send message + register async reply via ring CQE.
176        //
177        //   sqe.fd        = port handle (cap id)
178        //   sqe.addr      = pointer to IpcMessage (request in, reply overwrites)
179        //   sqe.user_data = correlation token (echoed in CQE on reply)
180        //
181        op if op == AsyncOp::IpcCall as u8 => {
182            let port_id = crate::ipc::PortId::from_u64(sqe.fd as u64);
183            let Some(port) = crate::ipc::port::get_port(port_id) else {
184                push_completion_for_ring(ring, sqe.user_data, EBADF, 0);
185                return Ok(DispatchOutcome::CompletedInline);
186            };
187
188            // Read the outgoing message from userspace.
189            let user_msg = UserSliceRead::new(sqe.addr, core::mem::size_of::<IpcMessage>())
190                .map_err(|_| EFAULT)?;
191            let mut raw = [0u8; IPC_MSG_SIZE];
192            user_msg.copy_to(&mut raw);
193            let mut msg = ipc_message_from_raw(&raw);
194
195            let task = crate::process::current_task_clone().ok_or(EIO)?;
196            msg.sender = task.id.as_u64();
197
198            port.try_send(msg).map_err(|_| ENOMSG)?;
199
200            crate::ipc::reply::register_ring_call(
201                task.id,
202                port.owner,
203                ring_id,
204                sqe.user_data,
205                sqe.addr,
206            );
207
208            Ok(DispatchOutcome::InFlight)
209        }
210
211        // Storage read : submit async AHCI/NVMe read, CQE arrives via the IRQ
212        op if op == AsyncOp::StorageRead as u8 => {
213            let port_idx = sqe.fd as u8;
214            let lba = sqe.off;
215            let byte_count = sqe.len;
216            let user_buf_vaddr = sqe.addr;
217
218            match crate::hardware::storage::ahci::submit_async_storage_op(
219                port_idx,
220                lba,
221                byte_count,
222                user_buf_vaddr,
223                false, // read
224                ring_id,
225                sqe.user_data,
226            ) {
227                Ok(()) => Ok(DispatchOutcome::InFlight), // in-flight; CQE comes from IRQ handler
228                Err(_) => {
229                    push_completion_for_ring(ring, sqe.user_data, EIO, 0);
230                    Ok(DispatchOutcome::CompletedInline)
231                }
232            }
233        }
234
235        // Storage write : submit async AHCI/NVMe write, CQE arrives via IRQ
236        op if op == AsyncOp::StorageWrite as u8 => {
237            let port_idx = sqe.fd as u8;
238            let lba = sqe.off;
239            let byte_count = sqe.len;
240            let user_buf_vaddr = sqe.addr;
241
242            match crate::hardware::storage::ahci::submit_async_storage_op(
243                port_idx,
244                lba,
245                byte_count,
246                user_buf_vaddr,
247                true, // write
248                ring_id,
249                sqe.user_data,
250            ) {
251                Ok(()) => Ok(DispatchOutcome::InFlight), // in-flight; CQE comes from IRQ handler
252                Err(_) => {
253                    push_completion_for_ring(ring, sqe.user_data, EIO, 0);
254                    Ok(DispatchOutcome::CompletedInline)
255                }
256            }
257        }
258
259        // Unknown opcode
260        _ => Err(EINVAL),
261    }
262}
strat9_kernel/async_io/dispatch.rs

strat9_kernel/async_io/
dispatch.rs
