vasilito/RedBear-OS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'scheme_cleanups' into 'main'

Browse Source 
scheme-utils: Copy ReadinessBased from redox-scheme to scheme-utils

See merge request redox-os/base!230
This commit is contained in:
Jeremy Soller
2026-04-18 08:52:34 -06:00
parent 0669f41847 07657eb50e
commit 80233d0d62
12 changed files with 297 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -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",
]
Cargo.toml
+1 -1
View File
@@ -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"] }
audiod/src/main.rs
+4 -11
View File
@@ -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() {
drivers/audio/ac97d/src/main.rs
+13 -25
View File
@@ -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();
drivers/audio/ihdad/src/main.rs
+13 -24
View File
@@ -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::<Source>::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();
drivers/audio/sb16d/src/main.rs
+13 -25
View File
@@ -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");
}
}
}
ipcd/src/main.rs
+33 -155
View File
@@ -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::<EventSource>::new()
let event_queue = EventQueue::<EventSource>::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<bool>, 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<bool>, 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<bool>, 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<bool>, 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(())
}
ptyd/Cargo.toml
+1
View File
@@ -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
ptyd/src/main.rs
+16 -38
View File
@@ -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<PtyScheme>,
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(())
}
scheme-utils/Cargo.toml
+1
View File
@@ -5,6 +5,7 @@ edition = "2024"
[dependencies]
libredox.workspace = true
redox-scheme.workspace = true
redox_syscall.workspace = true
[lints]
scheme-utils/src/lib.rs
+4
View File
@@ -3,6 +3,10 @@ use std::num::Wrapping;
use syscall::{EBADF, Error, Result};
mod readiness_based;
pub use readiness_based::ReadinessBased;
pub struct HandleMap<T> {
handles: BTreeMap<usize, T>,
next_id: Wrapping<usize>,
scheme-utils/src/readiness_based.rs
+196
View File
@@ -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<Request>,
responses_to_write: VecDeque<Response>,
states: HashMap<Id, (CallerCtx, Op)>,
ready_queue: VecDeque<Id>,
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<bool> {
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<Id, (CallerCtx, Op)>,
) -> Result<ControlFlow<Response, (CallerCtx, Op)>> {
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()),
}
}
}