From ca281c9395dc8b719f1c94f1282a10c2a7593086 Mon Sep 17 00:00:00 2001 From: bjorn3 <17426603+bjorn3@users.noreply.github.com> Date: Fri, 17 Apr 2026 22:17:02 +0200 Subject: [PATCH 1/5] scheme-utils: Copy ReadinessBased from redox-scheme to scheme-utils To allow improving the implementation and interface in this repo and move it back into redox-scheme later. --- Cargo.lock | 2 + audiod/src/main.rs | 2 +- drivers/audio/ac97d/src/main.rs | 2 +- drivers/audio/ihdad/src/main.rs | 2 +- drivers/audio/sb16d/src/main.rs | 2 +- ipcd/src/main.rs | 4 +- ptyd/Cargo.toml | 1 + ptyd/src/main.rs | 2 +- scheme-utils/Cargo.toml | 1 + scheme-utils/src/lib.rs | 4 + scheme-utils/src/readiness_based.rs | 213 ++++++++++++++++++++++++++++ 11 files changed, 229 insertions(+), 6 deletions(-) create mode 100644 scheme-utils/src/readiness_based.rs diff --git a/Cargo.lock b/Cargo.lock index 3986e77567..f90cb5dba8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1604,6 +1604,7 @@ dependencies = [ "redox_event", "redox_syscall 0.7.4", "redox_termios", + "scheme-utils", ] [[package]] @@ -2029,6 +2030,7 @@ name = "scheme-utils" version = "0.0.0" dependencies = [ "libredox", + "redox-scheme", "redox_syscall 0.7.4", ] diff --git a/audiod/src/main.rs b/audiod/src/main.rs index be7d5e1e2b..8231e9a2bb 100644 --- a/audiod/src/main.rs +++ b/audiod/src/main.rs @@ -9,8 +9,8 @@ use ioslice::IoSlice; use libredox::flag; use libredox::{error::Result, Fd}; -use redox_scheme::wrappers::ReadinessBased; use redox_scheme::Socket; +use scheme_utils::ReadinessBased; use daemon::SchemeDaemon; diff --git a/drivers/audio/ac97d/src/main.rs b/drivers/audio/ac97d/src/main.rs index ca1d9b0f3a..cbba8f8296 100644 --- a/drivers/audio/ac97d/src/main.rs +++ b/drivers/audio/ac97d/src/main.rs @@ -5,8 +5,8 @@ use std::usize; use event::{user_data, EventQueue}; use pcid_interface::PciFunctionHandle; use redox_scheme::scheme::register_sync_scheme; -use redox_scheme::wrappers::ReadinessBased; use redox_scheme::Socket; +use scheme_utils::ReadinessBased; use std::cell::RefCell; #[cfg(any(target_arch = "x86", target_arch = "x86_64"))] diff --git a/drivers/audio/ihdad/src/main.rs b/drivers/audio/ihdad/src/main.rs index 9d3f6963a5..9d9b059ee3 100755 --- a/drivers/audio/ihdad/src/main.rs +++ b/drivers/audio/ihdad/src/main.rs @@ -1,6 +1,6 @@ use redox_scheme::scheme::register_sync_scheme; -use redox_scheme::wrappers::ReadinessBased; use redox_scheme::Socket; +use scheme_utils::ReadinessBased; use std::cell::RefCell; use std::io::{Read, Write}; use std::os::unix::io::AsRawFd; diff --git a/drivers/audio/sb16d/src/main.rs b/drivers/audio/sb16d/src/main.rs index 14358c35f6..88a6b49041 100644 --- a/drivers/audio/sb16d/src/main.rs +++ b/drivers/audio/sb16d/src/main.rs @@ -1,7 +1,7 @@ use libredox::{flag, Fd}; use redox_scheme::scheme::register_sync_scheme; -use redox_scheme::wrappers::ReadinessBased; use redox_scheme::Socket; +use scheme_utils::ReadinessBased; use std::cell::RefCell; use std::{env, usize}; diff --git a/ipcd/src/main.rs b/ipcd/src/main.rs index 5e07fe5b57..d6f8d7f4a1 100644 --- a/ipcd/src/main.rs +++ b/ipcd/src/main.rs @@ -1,5 +1,7 @@ use event::{EventFlags, EventQueue}; -use redox_scheme::{scheme::register_sync_scheme, wrappers::ReadinessBased, Socket}; +use redox_scheme::scheme::register_sync_scheme; +use redox_scheme::Socket; +use scheme_utils::ReadinessBased; use std::sync::Mutex; mod chan; diff --git a/ptyd/Cargo.toml b/ptyd/Cargo.toml index 7e69930aa3..0edb56a730 100644 --- a/ptyd/Cargo.toml +++ b/ptyd/Cargo.toml @@ -8,6 +8,7 @@ edition = "2021" libredox.workspace = true daemon = { path = "../daemon" } redox-scheme.workspace = true +scheme-utils = { path = "../scheme-utils" } redox_event.workspace = true redox_syscall.workspace = true redox_termios.workspace = true diff --git a/ptyd/src/main.rs b/ptyd/src/main.rs index 323e2dd97a..0581b628e7 100644 --- a/ptyd/src/main.rs +++ b/ptyd/src/main.rs @@ -3,8 +3,8 @@ use std::cell::RefCell; use event::{user_data, EventFlags, EventQueue}; use libredox::{flag, Fd}; -use redox_scheme::wrappers::ReadinessBased; use redox_scheme::{scheme::register_sync_scheme, Response, SignalBehavior, Socket}; +use scheme_utils::ReadinessBased; use syscall::data::TimeSpec; mod controlterm; diff --git a/scheme-utils/Cargo.toml b/scheme-utils/Cargo.toml index 72719c52ff..00561c9332 100644 --- a/scheme-utils/Cargo.toml +++ b/scheme-utils/Cargo.toml @@ -5,6 +5,7 @@ edition = "2024" [dependencies] libredox.workspace = true +redox-scheme.workspace = true redox_syscall.workspace = true [lints] diff --git a/scheme-utils/src/lib.rs b/scheme-utils/src/lib.rs index db07fcd7dc..7e8f9ecb4b 100644 --- a/scheme-utils/src/lib.rs +++ b/scheme-utils/src/lib.rs @@ -3,6 +3,10 @@ use std::num::Wrapping; use syscall::{EBADF, Error, Result}; +mod readiness_based; + +pub use readiness_based::ReadinessBased; + pub struct HandleMap { handles: BTreeMap, next_id: Wrapping, diff --git a/scheme-utils/src/readiness_based.rs b/scheme-utils/src/readiness_based.rs new file mode 100644 index 0000000000..9d30a4f098 --- /dev/null +++ b/scheme-utils/src/readiness_based.rs @@ -0,0 +1,213 @@ +use core::ops::{Deref, DerefMut}; +use std::collections::{HashMap, VecDeque}; +use std::ops::ControlFlow; + +use libredox::error::Error as LError; + +use syscall::Result; +use syscall::error::{self as errno, ECANCELED, EIO, EOPNOTSUPP, Error}; + +use redox_scheme::scheme::{Op, SchemeResponse, SchemeState, SchemeSync}; +use redox_scheme::{CallerCtx, Id, Request, RequestKind, Response, SignalBehavior, Socket}; + +pub struct ReadinessBased<'sock> { + // TODO: VecDeque for both when it implements spare_capacity + requests_read: Vec, + responses_to_write: VecDeque, + + states: HashMap, + ready_queue: VecDeque, + + socket: &'sock Socket, + state: SchemeState, +} +impl<'sock> ReadinessBased<'sock> { + pub fn new(socket: &'sock Socket, queue_size: usize) -> Self { + Self { + requests_read: Vec::with_capacity(queue_size), + responses_to_write: VecDeque::with_capacity(queue_size), + states: HashMap::new(), + socket, + ready_queue: VecDeque::new(), + state: SchemeState::new(), + } + } + pub fn read_requests(&mut self) -> Result { + assert!(self.requests_read.is_empty()); + + match self + .socket + .read_requests(&mut self.requests_read, SignalBehavior::Interrupt) + { + // TODO: write ECANCELED or similar for everything in self.states? + Ok(()) if self.requests_read.is_empty() => Ok(false), // EOF + Ok(()) + | Err(Error { + errno: errno::EINTR | errno::EWOULDBLOCK | errno::EAGAIN, + }) => Ok(true), + Err(err) => return Err(err), + } + } + pub fn process_requests( + &mut self, + // TODO: Maybe change this to &mut SchemeSync, requires converting e.g. audiod + mut acquire_scheme: impl FnMut() -> Guard, + ) where + Guard: Deref + DerefMut, + { + for request in self.requests_read.drain(..) { + let req = match request.kind() { + RequestKind::Call(c) => c, + RequestKind::Cancellation(req) => { + if let Some((_caller, op)) = self.states.remove(&req.id) { + self.responses_to_write + .push_back(Response::err(ECANCELED, op)); + } + continue; + } + RequestKind::OnClose { id } => { + // TODO: state.on_close() + acquire_scheme().on_close(id); + continue; + } + RequestKind::SendFd(sendfd_request) => { + let result = acquire_scheme().on_sendfd(&sendfd_request); + let response = Response::new(result, sendfd_request); + self.responses_to_write.push_back(response); + continue; + } + RequestKind::RecvFd(recvfd_request) => { + let result = acquire_scheme().on_recvfd(&recvfd_request); + let caller = recvfd_request.caller(); + + if let Err(Error { + errno: errno::EWOULDBLOCK, + }) = result + { + self.states.insert(caller.id, (caller, recvfd_request.op())); + continue; + } + let response = Response::open_dup_like(result, recvfd_request); + self.responses_to_write.push_back(response); + continue; + } + _ => continue, + }; + let caller = req.caller(); + let mut op = match req.op() { + Ok(op) => op, + Err(req) => { + self.responses_to_write + .push_back(Response::err(EOPNOTSUPP, req)); + continue; + } + }; + let resp = + match op.handle_sync_dont_consume(&caller, &mut *acquire_scheme(), &mut self.state) + { + SchemeResponse::Opened(Err(Error { + errno: errno::EWOULDBLOCK, + })) + | SchemeResponse::Regular(Err(Error { + errno: errno::EWOULDBLOCK, + })) if !op.is_explicitly_nonblock() => { + self.states.insert(caller.id, (caller, op)); + continue; + } + SchemeResponse::Regular(r) => Response::new(r, op), + SchemeResponse::RegularAndNotifyOnDetach(status) => { + Response::new_notify_on_detach(status, op) + } + SchemeResponse::Opened(o) => Response::open_dup_like(o, op), + }; + self.responses_to_write.push_back(resp); + } + } + // TODO: Doesn't scale. Instead, provide an API for some form of queue. + // TODO: panic if id isn't present? + pub fn poll_request(&mut self, id: Id, scheme: &mut impl SchemeSync) -> Result { + Ok( + match Self::poll_request_inner(id, scheme, &mut self.state, &mut self.states)? { + ControlFlow::Continue((caller, op)) => { + self.states.insert(id, (caller, op)); + false + } + ControlFlow::Break(resp) => { + self.responses_to_write.push_back(resp); + true + } + }, + ) + } + + fn poll_request_inner( + id: Id, + scheme: &mut impl SchemeSync, + state: &mut SchemeState, + states: &mut HashMap, + ) -> Result> { + let (caller, mut op) = states.remove(&id).ok_or(Error::new(EIO))?; + let resp = match op.handle_sync_dont_consume(&caller, scheme, state) { + SchemeResponse::Opened(Err(Error { + errno: errno::EWOULDBLOCK, + })) + | SchemeResponse::Regular(Err(Error { + errno: errno::EWOULDBLOCK, + })) if !op.is_explicitly_nonblock() => { + return Ok(ControlFlow::Continue((caller, op))); + } + SchemeResponse::Regular(r) => Response::new(r, op), + SchemeResponse::Opened(o) => Response::open_dup_like(o, op), + SchemeResponse::RegularAndNotifyOnDetach(status) => { + Response::new_notify_on_detach(status, op) + } + }; + Ok(ControlFlow::Break(resp)) + } + pub fn poll_ready_requests(&mut self, mut acquire_scheme: impl FnMut() -> G) -> Result<()> + where + S: SchemeSync, + G: Deref + DerefMut, + { + for id in self.ready_queue.drain(..) { + match Self::poll_request_inner( + id, + &mut *acquire_scheme(), + &mut self.state, + &mut self.states, + )? { + ControlFlow::Break(resp) => { + self.responses_to_write.push_back(resp); + } + ControlFlow::Continue((caller, op)) => { + self.states.insert(id, (caller, op)); + } + } + } + Ok(()) + } + pub fn poll_all_requests(&mut self, acquire_scheme: impl FnMut() -> G) -> Result<()> + where + S: SchemeSync, + G: Deref + DerefMut, + { + // TODO: implement waker-like API + self.ready_queue.clear(); + self.ready_queue.extend(self.states.keys().copied()); + self.poll_ready_requests(acquire_scheme) + } + pub fn write_responses(&mut self) -> Result { + match self + .socket + .write_responses(&mut self.responses_to_write, SignalBehavior::Restart) + { + // TODO: write ECANCELED or similar for everything in self.states? + //Ok(()) if !self.responses_to_write.is_empty() => Ok(false), // EOF, FIXME + Ok(()) + | Err(Error { + errno: errno::EINTR | errno::EWOULDBLOCK | errno::EAGAIN, + }) => Ok(true), + Err(err) => return Err(LError::from(err).into()), + } + } +} From 4ef2c7ad6f6c8c039082d7f1d54f3549461a98dc Mon Sep 17 00:00:00 2001 From: bjorn3 <17426603+bjorn3@users.noreply.github.com> Date: Wed, 15 Apr 2026 20:21:39 +0200 Subject: [PATCH 2/5] scheme-utils: Remove the need for RefCell wrappers around schemes for ReadinessBased --- audiod/src/main.rs | 4 +- drivers/audio/ac97d/src/main.rs | 14 +++--- drivers/audio/ihdad/src/main.rs | 13 +++--- drivers/audio/sb16d/src/main.rs | 14 +++--- ipcd/src/main.rs | 60 +++++++++---------------- ptyd/src/main.rs | 24 +++++----- scheme-utils/src/readiness_based.rs | 70 ++++++++++------------------- 7 files changed, 76 insertions(+), 123 deletions(-) diff --git a/audiod/src/main.rs b/audiod/src/main.rs index 8231e9a2bb..aa9540fe38 100644 --- a/audiod/src/main.rs +++ b/audiod/src/main.rs @@ -74,8 +74,8 @@ fn daemon(daemon: SchemeDaemon) -> anyhow::Result<()> { if !readiness.read_requests()? { break; } - readiness.process_requests(|| scheme.lock().unwrap()); - readiness.poll_all_requests(|| scheme.lock().unwrap())?; + readiness.process_requests(&mut *scheme.lock().unwrap()); + readiness.poll_all_requests(&mut *scheme.lock().unwrap())?; if !readiness.write_responses()? { break; }; diff --git a/drivers/audio/ac97d/src/main.rs b/drivers/audio/ac97d/src/main.rs index cbba8f8296..5a7551c32c 100644 --- a/drivers/audio/ac97d/src/main.rs +++ b/drivers/audio/ac97d/src/main.rs @@ -7,7 +7,6 @@ use pcid_interface::PciFunctionHandle; use redox_scheme::scheme::register_sync_scheme; use redox_scheme::Socket; use scheme_utils::ReadinessBased; -use std::cell::RefCell; #[cfg(any(target_arch = "x86", target_arch = "x86_64"))] pub mod device; @@ -46,9 +45,8 @@ fn daemon(daemon: daemon::Daemon, pcid_handle: PciFunctionHandle) -> ! { let mut irq_file = irq.irq_handle("ac97d"); let socket = Socket::nonblock().expect("ac97d: failed to create socket"); - let device = RefCell::new(unsafe { - device::Ac97::new(bar0, bar1).expect("ac97d: failed to allocate device") - }); + let mut device = + unsafe { device::Ac97::new(bar0, bar1).expect("ac97d: failed to allocate device") }; let mut readiness_based = ReadinessBased::new(&socket, 16); user_data! { @@ -74,7 +72,7 @@ fn daemon(daemon: daemon::Daemon, pcid_handle: PciFunctionHandle) -> ! { ) .unwrap(); - register_sync_scheme(&socket, "audiohw", &mut *device.borrow_mut()) + register_sync_scheme(&socket, "audiohw", &mut device) .expect("ac97d: failed to register audiohw scheme to namespace"); daemon.ready(); @@ -90,13 +88,13 @@ fn daemon(daemon: daemon::Daemon, pcid_handle: PciFunctionHandle) -> ! { let mut irq = [0; 8]; irq_file.read(&mut irq).unwrap(); - if !device.borrow_mut().irq() { + if !device.irq() { continue; } irq_file.write(&mut irq).unwrap(); readiness_based - .poll_all_requests(|| device.borrow_mut()) + .poll_all_requests(&mut device) .expect("ac97d: failed to poll requests"); if !readiness_based .write_responses() @@ -119,7 +117,7 @@ fn daemon(daemon: daemon::Daemon, pcid_handle: PciFunctionHandle) -> ! { { break; } - readiness_based.process_requests(|| device.borrow_mut()); + readiness_based.process_requests(&mut device); if !readiness_based .write_responses() .expect("ac97d: failed to write to socket") diff --git a/drivers/audio/ihdad/src/main.rs b/drivers/audio/ihdad/src/main.rs index 9d9b059ee3..43a0682b68 100755 --- a/drivers/audio/ihdad/src/main.rs +++ b/drivers/audio/ihdad/src/main.rs @@ -1,7 +1,6 @@ use redox_scheme::scheme::register_sync_scheme; use redox_scheme::Socket; use scheme_utils::ReadinessBased; -use std::cell::RefCell; use std::io::{Read, Write}; use std::os::unix::io::AsRawFd; use std::usize; @@ -57,12 +56,12 @@ fn daemon(daemon: daemon::Daemon, mut pcid_handle: PciFunctionHandle) -> ! { let event_queue = EventQueue::::new().expect("ihdad: Could not create event queue."); let socket = Socket::nonblock().expect("ihdad: failed to create socket"); - let device = RefCell::new(unsafe { + let mut device = unsafe { hda::IntelHDA::new(address, vend_prod).expect("ihdad: failed to allocate device") - }); + }; let mut readiness_based = ReadinessBased::new(&socket, 16); - register_sync_scheme(&socket, "audiohw", &mut *device.borrow_mut()) + register_sync_scheme(&socket, "audiohw", &mut device) .expect("ihdad: failed to register audiohw scheme to namespace"); daemon.ready(); @@ -94,13 +93,13 @@ fn daemon(daemon: daemon::Daemon, mut pcid_handle: PciFunctionHandle) -> ! { let mut irq = [0; 8]; irq_file.irq_handle().read(&mut irq).unwrap(); - if !device.borrow_mut().irq() { + if !device.irq() { continue; } irq_file.irq_handle().write(&mut irq).unwrap(); readiness_based - .poll_all_requests(|| device.borrow_mut()) + .poll_all_requests(&mut device) .expect("ihdad: failed to poll requests"); if !readiness_based .write_responses() @@ -123,7 +122,7 @@ fn daemon(daemon: daemon::Daemon, mut pcid_handle: PciFunctionHandle) -> ! { { break; } - readiness_based.process_requests(|| device.borrow_mut()); + readiness_based.process_requests(&mut device); if !readiness_based .write_responses() .expect("ihdad: failed to write to socket") diff --git a/drivers/audio/sb16d/src/main.rs b/drivers/audio/sb16d/src/main.rs index 88a6b49041..c224090e6b 100644 --- a/drivers/audio/sb16d/src/main.rs +++ b/drivers/audio/sb16d/src/main.rs @@ -2,7 +2,6 @@ use libredox::{flag, Fd}; use redox_scheme::scheme::register_sync_scheme; use redox_scheme::Socket; use scheme_utils::ReadinessBased; -use std::cell::RefCell; use std::{env, usize}; use event::{user_data, EventQueue}; @@ -34,12 +33,11 @@ fn daemon(daemon: daemon::Daemon) -> ! { common::acquire_port_io_rights().expect("sb16d: failed to acquire port IO rights"); let socket = Socket::nonblock().expect("sb16d: failed to create socket"); - let device = - RefCell::new(unsafe { device::Sb16::new(addr).expect("sb16d: failed to allocate device") }); + let mut device = unsafe { device::Sb16::new(addr).expect("sb16d: failed to allocate device") }; let mut readiness_based = ReadinessBased::new(&socket, 16); //TODO: error on multiple IRQs? - let irq_file = match device.borrow().irqs.first() { + let irq_file = match device.irqs.first() { Some(irq) => Fd::open(&format!("/scheme/irq/{}", irq), flag::O_RDWR, 0) .expect("sb16d: failed to open IRQ file"), None => panic!("sb16d: no IRQs found"), @@ -63,7 +61,7 @@ fn daemon(daemon: daemon::Daemon) -> ! { ) .unwrap(); - register_sync_scheme(&socket, "sb16d", &mut *device.borrow_mut()) + register_sync_scheme(&socket, "sb16d", &mut device) .expect("sb16d: failed to register audiohw scheme to namespace"); daemon.ready(); @@ -81,13 +79,13 @@ fn daemon(daemon: daemon::Daemon) -> ! { let mut irq = [0; 8]; irq_file.read(&mut irq).unwrap(); - if !device.borrow_mut().irq() { + if !device.irq() { continue; } irq_file.write(&mut irq).unwrap(); readiness_based - .poll_all_requests(|| device.borrow_mut()) + .poll_all_requests(&mut device) .expect("sb16d: failed to poll requests"); if !readiness_based .write_responses() @@ -110,7 +108,7 @@ fn daemon(daemon: daemon::Daemon) -> ! { { break; } - readiness_based.process_requests(|| device.borrow_mut()); + readiness_based.process_requests(&mut device); if !readiness_based .write_responses() .expect("sb16d: failed to write to socket") diff --git a/ipcd/src/main.rs b/ipcd/src/main.rs index d6f8d7f4a1..7590e791b7 100644 --- a/ipcd/src/main.rs +++ b/ipcd/src/main.rs @@ -2,7 +2,6 @@ use event::{EventFlags, EventQueue}; use redox_scheme::scheme::register_sync_scheme; use redox_scheme::Socket; use scheme_utils::ReadinessBased; -use std::sync::Mutex; mod chan; mod shm; @@ -41,54 +40,37 @@ fn inner(daemon: daemon::Daemon) -> anyhow::Result<()> { // Prepare chan scheme let chan_socket = Socket::nonblock().map_err(|e| anyhow::anyhow!("failed to create chan scheme: {e}"))?; - let chan = Mutex::new(ChanScheme::new(&chan_socket)); + let mut chan = ChanScheme::new(&chan_socket); let mut chan_handler = ReadinessBased::new(&chan_socket, 16); // Prepare shm scheme let shm_socket = Socket::nonblock().map_err(|e| anyhow::anyhow!("failed to create shm socket: {e}"))?; - let shm = Mutex::new(ShmScheme::new()); + let mut shm = ShmScheme::new(); let mut shm_handler = ReadinessBased::new(&shm_socket, 16); // Prepare uds stream scheme let uds_stream_socket = Socket::nonblock() .map_err(|e| anyhow::anyhow!("failed to create uds stream scheme: {e}"))?; - let uds_stream = Mutex::new( - UdsStreamScheme::new(&uds_stream_socket) - .map_err(|e| anyhow::anyhow!("failed to create uds stream scheme: {e}"))?, - ); + let mut uds_stream = UdsStreamScheme::new(&uds_stream_socket) + .map_err(|e| anyhow::anyhow!("failed to create uds stream scheme: {e}"))?; let mut uds_stream_handler = ReadinessBased::new(&uds_stream_socket, 16); // Prepare uds dgram scheme let uds_dgram_socket = Socket::nonblock() .map_err(|e| anyhow::anyhow!("failed to create uds dgram scheme: {e}"))?; - let uds_dgram = Mutex::new( - UdsDgramScheme::new(&uds_dgram_socket) - .map_err(|e| anyhow::anyhow!("failed to create uds dgram scheme: {e}"))?, - ); + let mut uds_dgram = UdsDgramScheme::new(&uds_dgram_socket) + .map_err(|e| anyhow::anyhow!("failed to create uds dgram scheme: {e}"))?; let mut uds_dgram_handler = ReadinessBased::new(&uds_dgram_socket, 16); - { - let mut chan_guard = chan.lock().unwrap(); - register_sync_scheme(&chan_socket, "chan", &mut *chan_guard) - .map_err(|e| anyhow::anyhow!("failed to register chan scheme: {e}"))?; - } - { - let mut shm_guard = shm.lock().unwrap(); - register_sync_scheme(&shm_socket, "shm", &mut *shm_guard) - .map_err(|e| anyhow::anyhow!("failed to register shm scheme: {e}"))?; - } - { - let mut uds_stream_guard = uds_stream.lock().unwrap(); - register_sync_scheme(&uds_stream_socket, "uds_stream", &mut *uds_stream_guard) - .map_err(|e| anyhow::anyhow!("failed to register uds stream scheme: {e}"))?; - } - { - let mut uds_dgram_guard = uds_dgram.lock().unwrap(); - // Register schemes to namespace - register_sync_scheme(&uds_dgram_socket, "uds_dgram", &mut *uds_dgram_guard) - .map_err(|e| anyhow::anyhow!("failed to register uds dgram scheme: {e}"))?; - } + register_sync_scheme(&chan_socket, "chan", &mut chan) + .map_err(|e| anyhow::anyhow!("failed to register chan scheme: {e}"))?; + register_sync_scheme(&shm_socket, "shm", &mut shm) + .map_err(|e| anyhow::anyhow!("failed to register shm scheme: {e}"))?; + register_sync_scheme(&uds_stream_socket, "uds_stream", &mut uds_stream) + .map_err(|e| anyhow::anyhow!("failed to register uds stream scheme: {e}"))?; + register_sync_scheme(&uds_dgram_socket, "uds_dgram", &mut uds_dgram) + .map_err(|e| anyhow::anyhow!("failed to register uds dgram scheme: {e}"))?; daemon.ready(); @@ -153,11 +135,11 @@ fn inner(daemon: daemon::Daemon) -> anyhow::Result<()> { } // 2. Process requests - chan_handler.process_requests(|| chan.lock().unwrap()); + chan_handler.process_requests(&mut chan); // 3.Poll all blocking requests chan_handler - .poll_all_requests(|| chan.lock().unwrap()) + .poll_all_requests(&mut chan) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; // 3. Write responses @@ -186,7 +168,7 @@ fn inner(daemon: daemon::Daemon) -> anyhow::Result<()> { } // 2. Process requests - shm_handler.process_requests(|| shm.lock().unwrap()); + shm_handler.process_requests(&mut shm); // shm is not a blocking scheme @@ -216,11 +198,11 @@ fn inner(daemon: daemon::Daemon) -> anyhow::Result<()> { } // 2. Process requests - uds_stream_handler.process_requests(|| uds_stream.lock().unwrap()); + uds_stream_handler.process_requests(&mut uds_stream); // 3.Poll all blocking requests uds_stream_handler - .poll_all_requests(|| uds_stream.lock().unwrap()) + .poll_all_requests(&mut uds_stream) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; // 3. Write responses @@ -249,11 +231,11 @@ fn inner(daemon: daemon::Daemon) -> anyhow::Result<()> { } // 2. Process requests - uds_dgram_handler.process_requests(|| uds_dgram.lock().unwrap()); + uds_dgram_handler.process_requests(&mut uds_dgram); // 3.Poll all blocking requests uds_dgram_handler - .poll_all_requests(|| uds_dgram.lock().unwrap()) + .poll_all_requests(&mut uds_dgram) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; // 3. Write responses diff --git a/ptyd/src/main.rs b/ptyd/src/main.rs index 0581b628e7..0a363a1443 100644 --- a/ptyd/src/main.rs +++ b/ptyd/src/main.rs @@ -1,5 +1,3 @@ -use std::cell::RefCell; - use event::{user_data, EventFlags, EventQueue}; use libredox::{flag, Fd}; @@ -39,8 +37,8 @@ fn daemon(daemon: daemon::Daemon) -> ! { let socket = redox_scheme::Socket::nonblock().expect("pty: failed to create pty scheme"); let mut handler = ReadinessBased::new(&socket, 16); - let scheme = RefCell::new(PtyScheme::new()); - register_sync_scheme(&socket, "pty", &mut *scheme.borrow_mut()) + let mut scheme = PtyScheme::new(); + register_sync_scheme(&socket, "pty", &mut scheme) .expect("ptyd: failed to register scheme to namespace"); daemon.ready(); @@ -58,15 +56,15 @@ fn daemon(daemon: daemon::Daemon) -> ! { let mut timeout_count = 0u64; - scan_requests(&mut handler, &scheme).expect("pty: could not scan requests"); - issue_events(&socket, &mut scheme.borrow_mut()); + scan_requests(&mut handler, &mut scheme).expect("pty: could not scan requests"); + issue_events(&socket, &mut scheme); for event_res in event_queue { let event = event_res.expect("pty: failed to read from event queue"); match event.user_data { EventSource::Socket => { - if scan_requests(&mut handler, &scheme).is_err() { + if scan_requests(&mut handler, &mut scheme).is_err() { break; } } @@ -75,19 +73,19 @@ fn daemon(daemon: daemon::Daemon) -> ! { timeout_count = timeout_count.wrapping_add(1); - for (_id, handle) in scheme.borrow_mut().handles.iter_mut() { + for (_id, handle) in scheme.handles.iter_mut() { if let Handle::Resource(res) = handle { res.timeout(timeout_count); } } handler - .poll_all_requests(|| scheme.borrow_mut()) + .poll_all_requests(&mut scheme) .expect("ihdad: failed to poll requests"); } } - issue_events(&socket, &mut scheme.borrow_mut()); + issue_events(&socket, &mut scheme); } std::process::exit(0); @@ -95,7 +93,7 @@ fn daemon(daemon: daemon::Daemon) -> ! { fn scan_requests( handler: &mut ReadinessBased<'_>, - scheme: &RefCell, + scheme: &mut PtyScheme, ) -> libredox::error::Result<()> { // 1. Read requests match handler @@ -109,11 +107,11 @@ fn scan_requests( } // 2. Process requests - handler.process_requests(|| scheme.borrow_mut()); + handler.process_requests(scheme); // 3. Poll all blocking requests handler - .poll_all_requests(|| scheme.borrow_mut()) + .poll_all_requests(scheme) .expect("pty: error occured in poll_all_requests"); // 4. Write responses diff --git a/scheme-utils/src/readiness_based.rs b/scheme-utils/src/readiness_based.rs index 9d30a4f098..39e7dc31fb 100644 --- a/scheme-utils/src/readiness_based.rs +++ b/scheme-utils/src/readiness_based.rs @@ -1,4 +1,3 @@ -use core::ops::{Deref, DerefMut}; use std::collections::{HashMap, VecDeque}; use std::ops::ControlFlow; @@ -48,13 +47,7 @@ impl<'sock> ReadinessBased<'sock> { Err(err) => return Err(err), } } - pub fn process_requests( - &mut self, - // TODO: Maybe change this to &mut SchemeSync, requires converting e.g. audiod - mut acquire_scheme: impl FnMut() -> Guard, - ) where - Guard: Deref + DerefMut, - { + pub fn process_requests(&mut self, scheme: &mut impl SchemeSync) { for request in self.requests_read.drain(..) { let req = match request.kind() { RequestKind::Call(c) => c, @@ -67,17 +60,17 @@ impl<'sock> ReadinessBased<'sock> { } RequestKind::OnClose { id } => { // TODO: state.on_close() - acquire_scheme().on_close(id); + scheme.on_close(id); continue; } RequestKind::SendFd(sendfd_request) => { - let result = acquire_scheme().on_sendfd(&sendfd_request); + let result = scheme.on_sendfd(&sendfd_request); let response = Response::new(result, sendfd_request); self.responses_to_write.push_back(response); continue; } RequestKind::RecvFd(recvfd_request) => { - let result = acquire_scheme().on_recvfd(&recvfd_request); + let result = scheme.on_recvfd(&recvfd_request); let caller = recvfd_request.caller(); if let Err(Error { @@ -102,24 +95,22 @@ impl<'sock> ReadinessBased<'sock> { continue; } }; - let resp = - match op.handle_sync_dont_consume(&caller, &mut *acquire_scheme(), &mut self.state) - { - SchemeResponse::Opened(Err(Error { - errno: errno::EWOULDBLOCK, - })) - | SchemeResponse::Regular(Err(Error { - errno: errno::EWOULDBLOCK, - })) if !op.is_explicitly_nonblock() => { - self.states.insert(caller.id, (caller, op)); - continue; - } - SchemeResponse::Regular(r) => Response::new(r, op), - SchemeResponse::RegularAndNotifyOnDetach(status) => { - Response::new_notify_on_detach(status, op) - } - SchemeResponse::Opened(o) => Response::open_dup_like(o, op), - }; + let resp = match op.handle_sync_dont_consume(&caller, scheme, &mut self.state) { + SchemeResponse::Opened(Err(Error { + errno: errno::EWOULDBLOCK, + })) + | SchemeResponse::Regular(Err(Error { + errno: errno::EWOULDBLOCK, + })) if !op.is_explicitly_nonblock() => { + self.states.insert(caller.id, (caller, op)); + continue; + } + SchemeResponse::Regular(r) => Response::new(r, op), + SchemeResponse::RegularAndNotifyOnDetach(status) => { + Response::new_notify_on_detach(status, op) + } + SchemeResponse::Opened(o) => Response::open_dup_like(o, op), + }; self.responses_to_write.push_back(resp); } } @@ -164,18 +155,9 @@ impl<'sock> ReadinessBased<'sock> { }; Ok(ControlFlow::Break(resp)) } - pub fn poll_ready_requests(&mut self, mut acquire_scheme: impl FnMut() -> G) -> Result<()> - where - S: SchemeSync, - G: Deref + DerefMut, - { + pub fn poll_ready_requests(&mut self, scheme: &mut impl SchemeSync) -> Result<()> { for id in self.ready_queue.drain(..) { - match Self::poll_request_inner( - id, - &mut *acquire_scheme(), - &mut self.state, - &mut self.states, - )? { + match Self::poll_request_inner(id, scheme, &mut self.state, &mut self.states)? { ControlFlow::Break(resp) => { self.responses_to_write.push_back(resp); } @@ -186,15 +168,11 @@ impl<'sock> ReadinessBased<'sock> { } Ok(()) } - pub fn poll_all_requests(&mut self, acquire_scheme: impl FnMut() -> G) -> Result<()> - where - S: SchemeSync, - G: Deref + DerefMut, - { + pub fn poll_all_requests(&mut self, scheme: &mut impl SchemeSync) -> Result<()> { // TODO: implement waker-like API self.ready_queue.clear(); self.ready_queue.extend(self.states.keys().copied()); - self.poll_ready_requests(acquire_scheme) + self.poll_ready_requests(scheme) } pub fn write_responses(&mut self) -> Result { match self From a5018ca37edbe4ab33ce77875478a2fe183a2692 Mon Sep 17 00:00:00 2001 From: bjorn3 <17426603+bjorn3@users.noreply.github.com> Date: Wed, 15 Apr 2026 22:00:52 +0200 Subject: [PATCH 3/5] scheme-utils: Merge read_requests and process_requests And simplify some matches on the resulting method. --- Cargo.toml | 2 +- audiod/src/main.rs | 3 +- drivers/audio/ac97d/src/main.rs | 3 +- drivers/audio/ihdad/src/main.rs | 3 +- drivers/audio/sb16d/src/main.rs | 3 +- ipcd/src/main.rs | 100 +++++++--------------------- ptyd/src/main.rs | 25 +++---- scheme-utils/src/readiness_based.rs | 11 +-- 8 files changed, 43 insertions(+), 107 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index cdc14bbcee..f155066248 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -94,7 +94,7 @@ redox-ioctl = { git = "https://gitlab.redox-os.org/redox-os/relibc.git" } redox-log = { git = "https://gitlab.redox-os.org/redox-os/redox-log.git" } redox-rt = { git = "https://gitlab.redox-os.org/redox-os/relibc.git", default-features = false } redox-scheme = "0.11.0" -redox_syscall = "0.7.4" +redox_syscall = { version = "0.7.4", features = ["std"] } redox_termios = "0.1.3" ron = "0.8.1" serde = { version = "1", features = ["derive"] } diff --git a/audiod/src/main.rs b/audiod/src/main.rs index aa9540fe38..1de5d2df4c 100644 --- a/audiod/src/main.rs +++ b/audiod/src/main.rs @@ -71,10 +71,9 @@ fn daemon(daemon: SchemeDaemon) -> anyhow::Result<()> { let mut readiness = ReadinessBased::new(&socket, 16); loop { - if !readiness.read_requests()? { + if !readiness.read_and_process_requests(&mut *scheme.lock().unwrap())? { break; } - readiness.process_requests(&mut *scheme.lock().unwrap()); readiness.poll_all_requests(&mut *scheme.lock().unwrap())?; if !readiness.write_responses()? { break; diff --git a/drivers/audio/ac97d/src/main.rs b/drivers/audio/ac97d/src/main.rs index 5a7551c32c..a681c165bd 100644 --- a/drivers/audio/ac97d/src/main.rs +++ b/drivers/audio/ac97d/src/main.rs @@ -112,12 +112,11 @@ fn daemon(daemon: daemon::Daemon, pcid_handle: PciFunctionHandle) -> ! { } Source::Scheme => { if !readiness_based - .read_requests() + .read_and_process_requests(&mut device) .expect("ac97d: failed to read from socket") { break; } - readiness_based.process_requests(&mut device); if !readiness_based .write_responses() .expect("ac97d: failed to write to socket") diff --git a/drivers/audio/ihdad/src/main.rs b/drivers/audio/ihdad/src/main.rs index 43a0682b68..71678e645e 100755 --- a/drivers/audio/ihdad/src/main.rs +++ b/drivers/audio/ihdad/src/main.rs @@ -117,12 +117,11 @@ fn daemon(daemon: daemon::Daemon, mut pcid_handle: PciFunctionHandle) -> ! { } Source::Scheme => { if !readiness_based - .read_requests() + .read_and_process_requests(&mut device) .expect("ihdad: failed to read from socket") { break; } - readiness_based.process_requests(&mut device); if !readiness_based .write_responses() .expect("ihdad: failed to write to socket") diff --git a/drivers/audio/sb16d/src/main.rs b/drivers/audio/sb16d/src/main.rs index c224090e6b..8060a9212d 100644 --- a/drivers/audio/sb16d/src/main.rs +++ b/drivers/audio/sb16d/src/main.rs @@ -103,12 +103,11 @@ fn daemon(daemon: daemon::Daemon) -> ! { } Source::Scheme => { if !readiness_based - .read_requests() + .read_and_process_requests(&mut device) .expect("sb16d: failed to read from socket") { break; } - readiness_based.process_requests(&mut device); if !readiness_based .write_responses() .expect("sb16d: failed to write to socket") diff --git a/ipcd/src/main.rs b/ipcd/src/main.rs index 7590e791b7..2ed4ee6a0d 100644 --- a/ipcd/src/main.rs +++ b/ipcd/src/main.rs @@ -123,130 +123,78 @@ fn inner(daemon: daemon::Daemon) -> anyhow::Result<()> { EventSource::ChanSocket => { // Channel scheme if !chan_eof { - // 1. Read requests - match chan_handler.read_requests() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - chan_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), + // 1. Read and process requests + if !chan_handler.read_and_process_requests(&mut chan)? { + chan_eof = true; } } - // 2. Process requests - chan_handler.process_requests(&mut chan); - - // 3.Poll all blocking requests + // 2. Poll all blocking requests chan_handler .poll_all_requests(&mut chan) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; // 3. Write responses // write_responses returns a Result, but currently only returns true. - match chan_handler.write_responses() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - chan_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), + if !chan_handler.write_responses()? { + chan_eof = true; } } EventSource::ShmSocket => { // Shared memory scheme if !shm_eof { - // 1. Read requests - match shm_handler.read_requests() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - shm_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), + // 1. Read and process requests + if !shm_handler.read_and_process_requests(&mut shm)? { + shm_eof = true; } } - // 2. Process requests - shm_handler.process_requests(&mut shm); - // shm is not a blocking scheme - // 3. Write responses + // 2. Write responses // write_responses returns a Result, but currently only returns true. - match shm_handler.write_responses() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - shm_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), + if !shm_handler.write_responses()? { + shm_eof = true; } } EventSource::UdsStreamSocket => { // Unix Domain Socket Stream scheme if !uds_stream_eof { - // 1. Read requests - match uds_stream_handler.read_requests() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - uds_stream_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), + // 1. Read and process requests + if !uds_stream_handler.read_and_process_requests(&mut uds_stream)? { + uds_stream_eof = true; } } - // 2. Process requests - uds_stream_handler.process_requests(&mut uds_stream); - - // 3.Poll all blocking requests + // 2. Poll all blocking requests uds_stream_handler .poll_all_requests(&mut uds_stream) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; // 3. Write responses // write_responses returns a Result, but currently only returns true. - match uds_stream_handler.write_responses() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - uds_stream_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), + if !uds_stream_handler.write_responses()? { + uds_stream_eof = true; } } EventSource::UdsDgramSocket => { // Unix Domain Socket Dgram scheme if !uds_dgram_eof { - // 1. Read requests - match uds_dgram_handler.read_requests() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - uds_dgram_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), + // 1. Read and process requests + if !uds_dgram_handler.read_and_process_requests(&mut uds_dgram)? { + uds_dgram_eof = true; } } - // 2. Process requests - uds_dgram_handler.process_requests(&mut uds_dgram); - - // 3.Poll all blocking requests + // 2. Poll all blocking requests uds_dgram_handler .poll_all_requests(&mut uds_dgram) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; // 3. Write responses // write_responses returns a Result, but currently only returns true. - match uds_dgram_handler.write_responses() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - uds_dgram_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), + if !uds_dgram_handler.write_responses()? { + uds_dgram_eof = true; } } } diff --git a/ptyd/src/main.rs b/ptyd/src/main.rs index 0a363a1443..95543faaee 100644 --- a/ptyd/src/main.rs +++ b/ptyd/src/main.rs @@ -95,34 +95,25 @@ fn scan_requests( handler: &mut ReadinessBased<'_>, scheme: &mut PtyScheme, ) -> libredox::error::Result<()> { - // 1. Read requests - match handler - .read_requests() + // 1. Read and process requests + if !handler + .read_and_process_requests(scheme) .expect("pty: failed to read from socket") { - true => {} // Read requests success - false => { - panic!("pty: channel EOF") - } + panic!("pty: channel EOF") } - // 2. Process requests - handler.process_requests(scheme); - - // 3. Poll all blocking requests + // 2. Poll all blocking requests handler .poll_all_requests(scheme) .expect("pty: error occured in poll_all_requests"); - // 4. Write responses - match handler + // 3. Write responses + if !handler .write_responses() .expect("pty: failed to write to socket") { - true => {} // Write requests success - false => { - panic!("pty: channel EOF") - } + panic!("pty: channel EOF") } Ok(()) } diff --git a/scheme-utils/src/readiness_based.rs b/scheme-utils/src/readiness_based.rs index 39e7dc31fb..50a3929e5f 100644 --- a/scheme-utils/src/readiness_based.rs +++ b/scheme-utils/src/readiness_based.rs @@ -31,7 +31,7 @@ impl<'sock> ReadinessBased<'sock> { state: SchemeState::new(), } } - pub fn read_requests(&mut self) -> Result { + pub fn read_and_process_requests(&mut self, scheme: &mut impl SchemeSync) -> Result { assert!(self.requests_read.is_empty()); match self @@ -39,15 +39,14 @@ impl<'sock> ReadinessBased<'sock> { .read_requests(&mut self.requests_read, SignalBehavior::Interrupt) { // TODO: write ECANCELED or similar for everything in self.states? - Ok(()) if self.requests_read.is_empty() => Ok(false), // EOF + Ok(()) if self.requests_read.is_empty() => return Ok(false), // EOF Ok(()) | Err(Error { errno: errno::EINTR | errno::EWOULDBLOCK | errno::EAGAIN, - }) => Ok(true), + }) => {} Err(err) => return Err(err), } - } - pub fn process_requests(&mut self, scheme: &mut impl SchemeSync) { + for request in self.requests_read.drain(..) { let req = match request.kind() { RequestKind::Call(c) => c, @@ -113,6 +112,8 @@ impl<'sock> ReadinessBased<'sock> { }; self.responses_to_write.push_back(resp); } + + Ok(true) } // TODO: Doesn't scale. Instead, provide an API for some form of queue. // TODO: panic if id isn't present? From a2bc1dd79c3cf8d157afe7d0c50c149e8724dc6f Mon Sep 17 00:00:00 2001 From: bjorn3 <17426603+bjorn3@users.noreply.github.com> Date: Wed, 15 Apr 2026 22:13:06 +0200 Subject: [PATCH 4/5] scheme-utils: Panic on scheme EOF EOF used to be triggered by unmounting a scheme, but this is no longer the case since namespaces got moved to userspace. Unmounting now only closes the scheme root fd. --- audiod/src/main.rs | 10 +--- drivers/audio/ac97d/src/main.rs | 21 +++----- drivers/audio/ihdad/src/main.rs | 21 +++----- drivers/audio/sb16d/src/main.rs | 21 +++----- ipcd/src/main.rs | 80 +++++------------------------ ptyd/src/main.rs | 19 ++----- scheme-utils/src/readiness_based.rs | 18 ++++--- 7 files changed, 47 insertions(+), 143 deletions(-) diff --git a/audiod/src/main.rs b/audiod/src/main.rs index 1de5d2df4c..51b103afa6 100644 --- a/audiod/src/main.rs +++ b/audiod/src/main.rs @@ -71,16 +71,10 @@ fn daemon(daemon: SchemeDaemon) -> anyhow::Result<()> { let mut readiness = ReadinessBased::new(&socket, 16); loop { - if !readiness.read_and_process_requests(&mut *scheme.lock().unwrap())? { - break; - } + readiness.read_and_process_requests(&mut *scheme.lock().unwrap())?; readiness.poll_all_requests(&mut *scheme.lock().unwrap())?; - if !readiness.write_responses()? { - break; - }; + readiness.write_responses()?; } - - Ok(()) } fn main() { diff --git a/drivers/audio/ac97d/src/main.rs b/drivers/audio/ac97d/src/main.rs index a681c165bd..ffa8a94b7a 100644 --- a/drivers/audio/ac97d/src/main.rs +++ b/drivers/audio/ac97d/src/main.rs @@ -96,12 +96,9 @@ fn daemon(daemon: daemon::Daemon, pcid_handle: PciFunctionHandle) -> ! { readiness_based .poll_all_requests(&mut device) .expect("ac97d: failed to poll requests"); - if !readiness_based + readiness_based .write_responses() - .expect("ac97d: failed to write to socket") - { - break; - } + .expect("ac97d: failed to write to socket"); /* let next_read = device_irq.next_read(); @@ -111,18 +108,12 @@ fn daemon(daemon: daemon::Daemon, pcid_handle: PciFunctionHandle) -> ! { */ } Source::Scheme => { - if !readiness_based + readiness_based .read_and_process_requests(&mut device) - .expect("ac97d: failed to read from socket") - { - break; - } - if !readiness_based + .expect("ac97d: failed to read from socket"); + readiness_based .write_responses() - .expect("ac97d: failed to write to socket") - { - break; - } + .expect("ac97d: failed to write to socket"); /* let next_read = device.borrow().next_read(); diff --git a/drivers/audio/ihdad/src/main.rs b/drivers/audio/ihdad/src/main.rs index 71678e645e..31a2add737 100755 --- a/drivers/audio/ihdad/src/main.rs +++ b/drivers/audio/ihdad/src/main.rs @@ -101,12 +101,9 @@ fn daemon(daemon: daemon::Daemon, mut pcid_handle: PciFunctionHandle) -> ! { readiness_based .poll_all_requests(&mut device) .expect("ihdad: failed to poll requests"); - if !readiness_based + readiness_based .write_responses() - .expect("ihdad: failed to write to socket") - { - break; - } + .expect("ihdad: failed to write to socket"); /* let next_read = device_irq.next_read(); @@ -116,18 +113,12 @@ fn daemon(daemon: daemon::Daemon, mut pcid_handle: PciFunctionHandle) -> ! { */ } Source::Scheme => { - if !readiness_based + readiness_based .read_and_process_requests(&mut device) - .expect("ihdad: failed to read from socket") - { - break; - } - if !readiness_based + .expect("ihdad: failed to read from socket"); + readiness_based .write_responses() - .expect("ihdad: failed to write to socket") - { - break; - } + .expect("ihdad: failed to write to socket"); /* let next_read = device.borrow().next_read(); diff --git a/drivers/audio/sb16d/src/main.rs b/drivers/audio/sb16d/src/main.rs index 8060a9212d..9e351629d3 100644 --- a/drivers/audio/sb16d/src/main.rs +++ b/drivers/audio/sb16d/src/main.rs @@ -87,12 +87,9 @@ fn daemon(daemon: daemon::Daemon) -> ! { readiness_based .poll_all_requests(&mut device) .expect("sb16d: failed to poll requests"); - if !readiness_based + readiness_based .write_responses() - .expect("sb16d: failed to write to socket") - { - break; - } + .expect("sb16d: failed to write to socket"); /* let next_read = device_irq.next_read(); @@ -102,18 +99,12 @@ fn daemon(daemon: daemon::Daemon) -> ! { */ } Source::Scheme => { - if !readiness_based + readiness_based .read_and_process_requests(&mut device) - .expect("sb16d: failed to read from socket") - { - break; - } - if !readiness_based + .expect("sb16d: failed to read from socket"); + readiness_based .write_responses() - .expect("sb16d: failed to write to socket") - { - break; - } + .expect("sb16d: failed to write to socket"); } } } diff --git a/ipcd/src/main.rs b/ipcd/src/main.rs index 2ed4ee6a0d..a365ae028c 100644 --- a/ipcd/src/main.rs +++ b/ipcd/src/main.rs @@ -75,7 +75,7 @@ fn inner(daemon: daemon::Daemon) -> anyhow::Result<()> { daemon.ready(); // Create event listener for both files - let mut event_queue = EventQueue::::new() + let event_queue = EventQueue::::new() .map_err(|e| anyhow::anyhow!("failed to create event queue: {e}"))?; event_queue .subscribe( @@ -108,96 +108,42 @@ fn inner(daemon: daemon::Daemon) -> anyhow::Result<()> { libredox::call::setrens(0, 0)?; - // EOF flags - let mut chan_eof = false; - let mut shm_eof = false; - let mut uds_stream_eof = false; - let mut uds_dgram_eof = false; - while !(chan_eof && shm_eof && uds_stream_eof && uds_dgram_eof) { - let Some(event_res) = event_queue.next() else { - break; - }; - let event = event_res.map_err(|e| anyhow::anyhow!("error occured in event queue: {e}"))?; + loop { + let event = event_queue + .next_event() + .map_err(|e| anyhow::anyhow!("error occured in event queue: {e}"))?; match event.user_data { EventSource::ChanSocket => { // Channel scheme - if !chan_eof { - // 1. Read and process requests - if !chan_handler.read_and_process_requests(&mut chan)? { - chan_eof = true; - } - } - - // 2. Poll all blocking requests + chan_handler.read_and_process_requests(&mut chan)?; chan_handler .poll_all_requests(&mut chan) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; - - // 3. Write responses - // write_responses returns a Result, but currently only returns true. - if !chan_handler.write_responses()? { - chan_eof = true; - } + chan_handler.write_responses()?; } EventSource::ShmSocket => { // Shared memory scheme - if !shm_eof { - // 1. Read and process requests - if !shm_handler.read_and_process_requests(&mut shm)? { - shm_eof = true; - } - } - + shm_handler.read_and_process_requests(&mut shm)?; // shm is not a blocking scheme - - // 2. Write responses - // write_responses returns a Result, but currently only returns true. - if !shm_handler.write_responses()? { - shm_eof = true; - } + shm_handler.write_responses()?; } EventSource::UdsStreamSocket => { // Unix Domain Socket Stream scheme - if !uds_stream_eof { - // 1. Read and process requests - if !uds_stream_handler.read_and_process_requests(&mut uds_stream)? { - uds_stream_eof = true; - } - } - - // 2. Poll all blocking requests + uds_stream_handler.read_and_process_requests(&mut uds_stream)?; uds_stream_handler .poll_all_requests(&mut uds_stream) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; - - // 3. Write responses - // write_responses returns a Result, but currently only returns true. - if !uds_stream_handler.write_responses()? { - uds_stream_eof = true; - } + uds_stream_handler.write_responses()?; } EventSource::UdsDgramSocket => { // Unix Domain Socket Dgram scheme - if !uds_dgram_eof { - // 1. Read and process requests - if !uds_dgram_handler.read_and_process_requests(&mut uds_dgram)? { - uds_dgram_eof = true; - } - } - - // 2. Poll all blocking requests + uds_dgram_handler.read_and_process_requests(&mut uds_dgram)?; uds_dgram_handler .poll_all_requests(&mut uds_dgram) .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; - - // 3. Write responses - // write_responses returns a Result, but currently only returns true. - if !uds_dgram_handler.write_responses()? { - uds_dgram_eof = true; - } + uds_dgram_handler.write_responses()?; } } } - Ok(()) } diff --git a/ptyd/src/main.rs b/ptyd/src/main.rs index 95543faaee..e6bf429eb8 100644 --- a/ptyd/src/main.rs +++ b/ptyd/src/main.rs @@ -95,26 +95,15 @@ fn scan_requests( handler: &mut ReadinessBased<'_>, scheme: &mut PtyScheme, ) -> libredox::error::Result<()> { - // 1. Read and process requests - if !handler + handler .read_and_process_requests(scheme) - .expect("pty: failed to read from socket") - { - panic!("pty: channel EOF") - } - - // 2. Poll all blocking requests + .expect("pty: failed to read from socket"); handler .poll_all_requests(scheme) .expect("pty: error occured in poll_all_requests"); - - // 3. Write responses - if !handler + handler .write_responses() - .expect("pty: failed to write to socket") - { - panic!("pty: channel EOF") - } + .expect("pty: failed to write to socket"); Ok(()) } diff --git a/scheme-utils/src/readiness_based.rs b/scheme-utils/src/readiness_based.rs index 50a3929e5f..45bcb0b751 100644 --- a/scheme-utils/src/readiness_based.rs +++ b/scheme-utils/src/readiness_based.rs @@ -31,15 +31,16 @@ impl<'sock> ReadinessBased<'sock> { state: SchemeState::new(), } } - pub fn read_and_process_requests(&mut self, scheme: &mut impl SchemeSync) -> Result { + pub fn read_and_process_requests(&mut self, scheme: &mut impl SchemeSync) -> Result<()> { assert!(self.requests_read.is_empty()); match self .socket .read_requests(&mut self.requests_read, SignalBehavior::Interrupt) { - // TODO: write ECANCELED or similar for everything in self.states? - Ok(()) if self.requests_read.is_empty() => return Ok(false), // EOF + Ok(()) if self.requests_read.is_empty() => { + unreachable!("blocking scheme read failed to read anything"); + } Ok(()) | Err(Error { errno: errno::EINTR | errno::EWOULDBLOCK | errno::EAGAIN, @@ -113,7 +114,7 @@ impl<'sock> ReadinessBased<'sock> { self.responses_to_write.push_back(resp); } - Ok(true) + Ok(()) } // TODO: Doesn't scale. Instead, provide an API for some form of queue. // TODO: panic if id isn't present? @@ -175,17 +176,18 @@ impl<'sock> ReadinessBased<'sock> { self.ready_queue.extend(self.states.keys().copied()); self.poll_ready_requests(scheme) } - pub fn write_responses(&mut self) -> Result { + pub fn write_responses(&mut self) -> Result<()> { match self .socket .write_responses(&mut self.responses_to_write, SignalBehavior::Restart) { - // TODO: write ECANCELED or similar for everything in self.states? - //Ok(()) if !self.responses_to_write.is_empty() => Ok(false), // EOF, FIXME + Ok(()) if !self.responses_to_write.is_empty() => { + panic!("failed to write all scheme responses"); + } Ok(()) | Err(Error { errno: errno::EINTR | errno::EWOULDBLOCK | errno::EAGAIN, - }) => Ok(true), + }) => Ok(()), Err(err) => return Err(LError::from(err).into()), } } From 07657eb50ea8f2af931bd592608d97d6e5858323 Mon Sep 17 00:00:00 2001 From: bjorn3 <17426603+bjorn3@users.noreply.github.com> Date: Wed, 15 Apr 2026 22:39:40 +0200 Subject: [PATCH 5/5] scheme-utils: Panic on nonblocking scheme write --- scheme-utils/src/readiness_based.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/scheme-utils/src/readiness_based.rs b/scheme-utils/src/readiness_based.rs index 45bcb0b751..0eb929d333 100644 --- a/scheme-utils/src/readiness_based.rs +++ b/scheme-utils/src/readiness_based.rs @@ -184,10 +184,12 @@ impl<'sock> ReadinessBased<'sock> { Ok(()) if !self.responses_to_write.is_empty() => { panic!("failed to write all scheme responses"); } - Ok(()) - | Err(Error { + Ok(()) => Ok(()), + Err(Error { errno: errno::EINTR | errno::EWOULDBLOCK | errno::EAGAIN, - }) => Ok(()), + }) => { + panic!("scheme response writing should always block"); + } Err(err) => return Err(LError::from(err).into()), } }