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/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 be7d5e1e2b..51b103afa6 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; @@ -71,17 +71,10 @@ fn daemon(daemon: SchemeDaemon) -> anyhow::Result<()> { let mut readiness = ReadinessBased::new(&socket, 16); loop { - if !readiness.read_requests()? { - break; - } - readiness.process_requests(|| scheme.lock().unwrap()); - readiness.poll_all_requests(|| scheme.lock().unwrap())?; - if !readiness.write_responses()? { - break; - }; + readiness.read_and_process_requests(&mut *scheme.lock().unwrap())?; + readiness.poll_all_requests(&mut *scheme.lock().unwrap())?; + readiness.write_responses()?; } - - Ok(()) } fn main() { diff --git a/drivers/audio/ac97d/src/main.rs b/drivers/audio/ac97d/src/main.rs index ca1d9b0f3a..ffa8a94b7a 100644 --- a/drivers/audio/ac97d/src/main.rs +++ b/drivers/audio/ac97d/src/main.rs @@ -5,9 +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 std::cell::RefCell; +use scheme_utils::ReadinessBased; #[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,20 +88,17 @@ 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 + 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(); @@ -113,19 +108,12 @@ fn daemon(daemon: daemon::Daemon, pcid_handle: PciFunctionHandle) -> ! { */ } Source::Scheme => { - if !readiness_based - .read_requests() - .expect("ac97d: failed to read from socket") - { - break; - } - readiness_based.process_requests(|| device.borrow_mut()); - if !readiness_based + readiness_based + .read_and_process_requests(&mut device) + .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 9d3f6963a5..31a2add737 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::wrappers::ReadinessBased; use redox_scheme::Socket; -use std::cell::RefCell; +use scheme_utils::ReadinessBased; 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,20 +93,17 @@ 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 + 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(); @@ -117,19 +113,12 @@ fn daemon(daemon: daemon::Daemon, mut pcid_handle: PciFunctionHandle) -> ! { */ } Source::Scheme => { - if !readiness_based - .read_requests() - .expect("ihdad: failed to read from socket") - { - break; - } - readiness_based.process_requests(|| device.borrow_mut()); - if !readiness_based + readiness_based + .read_and_process_requests(&mut device) + .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 14358c35f6..9e351629d3 100644 --- a/drivers/audio/sb16d/src/main.rs +++ b/drivers/audio/sb16d/src/main.rs @@ -1,8 +1,7 @@ use libredox::{flag, Fd}; use redox_scheme::scheme::register_sync_scheme; -use redox_scheme::wrappers::ReadinessBased; use redox_scheme::Socket; -use std::cell::RefCell; +use scheme_utils::ReadinessBased; 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,20 +79,17 @@ 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 + 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(); @@ -104,19 +99,12 @@ fn daemon(daemon: daemon::Daemon) -> ! { */ } Source::Scheme => { - if !readiness_based - .read_requests() - .expect("sb16d: failed to read from socket") - { - break; - } - readiness_based.process_requests(|| device.borrow_mut()); - if !readiness_based + readiness_based + .read_and_process_requests(&mut device) + .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 5e07fe5b57..a365ae028c 100644 --- a/ipcd/src/main.rs +++ b/ipcd/src/main.rs @@ -1,6 +1,7 @@ use event::{EventFlags, EventQueue}; -use redox_scheme::{scheme::register_sync_scheme, wrappers::ReadinessBased, Socket}; -use std::sync::Mutex; +use redox_scheme::scheme::register_sync_scheme; +use redox_scheme::Socket; +use scheme_utils::ReadinessBased; mod chan; mod shm; @@ -39,59 +40,42 @@ 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(); // 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( @@ -124,148 +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 requests - match chan_handler.read_requests() { - Ok(true) => {} // Read requests success - Ok(false) => { - // EOF - chan_eof = true; - } - Err(err) => return Err(anyhow::anyhow!("{err}")), - } - } - - // 2. Process requests - chan_handler.process_requests(|| chan.lock().unwrap()); - - // 3.Poll all blocking requests + chan_handler.read_and_process_requests(&mut chan)?; 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 - // 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}")), - } + chan_handler.write_responses()?; } 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}")), - } - } - - // 2. Process requests - shm_handler.process_requests(|| shm.lock().unwrap()); - + shm_handler.read_and_process_requests(&mut shm)?; // shm is not a blocking scheme - - // 3. 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}")), - } + shm_handler.write_responses()?; } 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}")), - } - } - - // 2. Process requests - uds_stream_handler.process_requests(|| uds_stream.lock().unwrap()); - - // 3.Poll all blocking requests + uds_stream_handler.read_and_process_requests(&mut uds_stream)?; 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 - // 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}")), - } + uds_stream_handler.write_responses()?; } 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}")), - } - } - - // 2. Process requests - uds_dgram_handler.process_requests(|| uds_dgram.lock().unwrap()); - - // 3.Poll all blocking requests + uds_dgram_handler.read_and_process_requests(&mut uds_dgram)?; 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 - // 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}")), - } + uds_dgram_handler.write_responses()?; } } } - Ok(()) } 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..e6bf429eb8 100644 --- a/ptyd/src/main.rs +++ b/ptyd/src/main.rs @@ -1,10 +1,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; @@ -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,37 +93,17 @@ fn daemon(daemon: daemon::Daemon) -> ! { fn scan_requests( handler: &mut ReadinessBased<'_>, - scheme: &RefCell, + scheme: &mut PtyScheme, ) -> libredox::error::Result<()> { - // 1. Read requests - match handler - .read_requests() - .expect("pty: failed to read from socket") - { - true => {} // Read requests success - false => { - panic!("pty: channel EOF") - } - } - - // 2. Process requests - handler.process_requests(|| scheme.borrow_mut()); - - // 3. Poll all blocking requests handler - .poll_all_requests(|| scheme.borrow_mut()) + .read_and_process_requests(scheme) + .expect("pty: failed to read from socket"); + handler + .poll_all_requests(scheme) .expect("pty: error occured in poll_all_requests"); - - // 4. Write responses - match handler + handler .write_responses() - .expect("pty: failed to write to socket") - { - true => {} // Write requests success - false => { - panic!("pty: channel EOF") - } - } + .expect("pty: failed to write to socket"); Ok(()) } 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..0eb929d333 --- /dev/null +++ b/scheme-utils/src/readiness_based.rs @@ -0,0 +1,196 @@ +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_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) + { + 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, + }) => {} + Err(err) => return Err(err), + } + + 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() + scheme.on_close(id); + continue; + } + RequestKind::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 = 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, 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); + } + + Ok(()) + } + // 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, scheme: &mut impl SchemeSync) -> Result<()> { + for id in self.ready_queue.drain(..) { + match Self::poll_request_inner(id, 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, 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(scheme) + } + pub fn write_responses(&mut self) -> Result<()> { + match self + .socket + .write_responses(&mut self.responses_to_write, SignalBehavior::Restart) + { + Ok(()) if !self.responses_to_write.is_empty() => { + panic!("failed to write all scheme responses"); + } + Ok(()) => Ok(()), + Err(Error { + errno: errno::EINTR | errno::EWOULDBLOCK | errno::EAGAIN, + }) => { + panic!("scheme response writing should always block"); + } + Err(err) => return Err(LError::from(err).into()), + } + } +}