Merge branch 'scheme_cleanups2' into 'main'

Introduce Blocking as counterpart to ReadinessBased for blocking schemes

See merge request redox-os/base!232

Cargo.lock

@@ -735,6 +735,7 @@ dependencies = [
  "redox-scheme",
  "redox_event",
  "redox_syscall 0.7.4",
+ "scheme-utils",
 ]
 
 [[package]]
@@ -2911,6 +2912,7 @@ dependencies = [
  "redox_event",
  "redox_syscall 0.7.4",
  "regex",
+ "scheme-utils",
  "serde",
  "serde_json",
  "smallvec 1.15.1",
@@ -2946,6 +2948,7 @@ dependencies = [
  "libredox",
  "redox-scheme",
  "redox_syscall 0.7.4",
+ "scheme-utils",
 ]
 
 [[package]]

drivers/acpid/src/main.rs

@@ -1,16 +1,14 @@
 use std::convert::TryFrom;
 use std::fs::File;
 use std::mem;
+use std::ops::ControlFlow;
 use std::os::unix::io::AsRawFd;
 use std::sync::Arc;
 
 use ::acpi::aml::op_region::{RegionHandler, RegionSpace};
 use event::{EventFlags, RawEventQueue};
-use redox_scheme::{
-    scheme::{register_sync_scheme, SchemeState, SchemeSync},
-    RequestKind, Response, SignalBehavior, Socket,
-};
-use syscall::{EAGAIN, EWOULDBLOCK};
+use redox_scheme::{scheme::register_sync_scheme, Socket};
+use scheme_utils::Blocking;
 
 mod acpi;
 mod aml_physmem;
@@ -85,8 +83,8 @@ fn daemon(daemon: daemon::Daemon) -> ! {
     let mut event_queue = RawEventQueue::new().expect("acpid: failed to create event queue");
     let socket = Socket::nonblock().expect("acpid: failed to create disk scheme");
 
-    let mut state = SchemeState::new();
     let mut scheme = self::scheme::AcpiScheme::new(&acpi_context, &socket);
+    let mut handler = Blocking::new(&socket, 16);
 
     event_queue
         .subscribe(shutdown_pipe.as_raw_fd() as usize, 0, EventFlags::READ)
@@ -114,40 +112,12 @@ fn daemon(daemon: daemon::Daemon) -> ! {
 
         if event.fd == socket.inner().raw() {
             loop {
-                let req = match socket.next_request(SignalBehavior::Interrupt) {
-                    Ok(None) => {
-                        mounted = false;
-                        break;
-                    }
-                    Ok(Some(req)) => req,
-                    Err(err) => {
-                        if err.errno == EWOULDBLOCK || err.errno == EAGAIN {
-                            break;
-                        } else {
-                            panic!("acpid: failed to read next request: {}", err);
-                        }
-                    }
-                };
-                match req.kind() {
-                    RequestKind::Call(call) => {
-                        let response = call.handle_sync(&mut scheme, &mut state);
-                        socket
-                            .write_response(response, SignalBehavior::Restart)
-                            .expect("acpid: failed to write response");
-                    }
-                    RequestKind::OnClose { id } => {
-                        scheme.on_close(id);
-                    }
-                    RequestKind::SendFd(sendfd_request) => {
-                        let result = scheme.on_sendfd(&sendfd_request);
-                        socket
-                            .write_response(
-                                Response::new(result, sendfd_request),
-                                SignalBehavior::Restart,
-                            )
-                            .expect("acpid: failed to write response");
-                    }
-                    _ => (),
+                match handler
+                    .process_requests_nonblocking(&mut scheme)
+                    .expect("acpid: failed to process requests")
+                {
+                    ControlFlow::Continue(()) => {}
+                    ControlFlow::Break(()) => break,
                 }
             }
         } else if event.fd == shutdown_pipe.as_raw_fd() as usize {

drivers/graphics/driver-graphics/src/lib.rs

@@ -978,9 +978,7 @@ impl<T: GraphicsAdapter> SchemeSync for GraphicsSchemeInner<T> {
         let ptr = T::map_dumb_buffer(&mut self.adapter, framebuffer);
         Ok(unsafe { ptr.add(offset as usize) } as usize)
     }
-}
 
-impl<T: GraphicsAdapter> GraphicsSchemeInner<T> {
     fn on_close(&mut self, id: usize) {
         self.handles.remove(id);
     }

drivers/graphics/fbbootlogd/Cargo.toml

@@ -11,6 +11,7 @@ ransid.workspace = true
 redox_event.workspace = true
 redox_syscall.workspace = true
 redox-scheme.workspace = true
+scheme-utils = { path = "../../../scheme-utils" }
 
 console-draw = { path = "../console-draw" }
 daemon = { path = "../../../daemon" }

drivers/graphics/fbbootlogd/src/main.rs

@@ -7,13 +7,14 @@
 //! In the future fbbootlogd may also pull from logd as opposed to have logd push logs to it. And it
 //! it could display a boot splash like plymouth instead of a boot log when booting in quiet mode.
 
+use std::ops::ControlFlow;
 use std::os::fd::AsRawFd;
 
 use event::EventQueue;
 use inputd::ConsumerHandleEvent;
-use libredox::errno::EAGAIN;
 use orbclient::Event;
-use redox_scheme::{scheme::SchemeState, RequestKind, SignalBehavior, Socket};
+use redox_scheme::Socket;
+use scheme_utils::Blocking;
 
 use crate::scheme::FbbootlogScheme;
 
@@ -34,8 +35,8 @@ fn daemon(daemon: daemon::SchemeDaemon) -> ! {
 
     let socket = Socket::nonblock().expect("fbbootlogd: failed to create fbbootlog scheme");
 
-    let mut state = SchemeState::new();
     let mut scheme = FbbootlogScheme::new();
+    let mut handler = Blocking::new(&socket, 16);
 
     event_queue
         .subscribe(
@@ -77,33 +78,15 @@ fn daemon(daemon: daemon::SchemeDaemon) -> ! {
 
     for event in event_queue {
         match event.expect("fbbootlogd: failed to get event").user_data {
-            Source::Scheme => {
-                loop {
-                    let request = match socket.next_request(SignalBehavior::Restart) {
-                        Ok(Some(request)) => request,
-                        Ok(None) => {
-                            // Scheme likely got unmounted
-                            std::process::exit(0);
-                        }
-                        Err(err) if err.errno == EAGAIN => break,
-                        Err(err) => panic!("fbbootlogd: failed to read display scheme: {err:?}"),
-                    };
-
-                    match request.kind() {
-                        RequestKind::Call(call) => {
-                            let response = call.handle_sync(&mut scheme, &mut state);
-
-                            socket
-                                .write_response(response, SignalBehavior::Restart)
-                                .expect("pcid: failed to write next scheme response");
-                        }
-                        RequestKind::OnClose { id } => {
-                            scheme.on_close(id);
-                        }
-                        _ => (),
-                    }
+            Source::Scheme => loop {
+                match handler
+                    .process_requests_nonblocking(&mut scheme)
+                    .expect("fbbootlogd: failed to process requests")
+                {
+                    ControlFlow::Continue(()) => {}
+                    ControlFlow::Break(()) => break,
                 }
-            }
+            },
             Source::Input => {
                 let mut events = [Event::new(); 16];
                 loop {

drivers/graphics/fbbootlogd/src/scheme.rs

@@ -249,7 +249,3 @@ impl SchemeSync for FbbootlogScheme {
         Ok(buf.len())
     }
 }
-
-impl FbbootlogScheme {
-    pub fn on_close(&mut self, _id: usize) {}
-}

drivers/inputd/src/main.rs

@@ -15,16 +15,17 @@ use core::mem::size_of;
 use std::borrow::Cow;
 use std::collections::BTreeSet;
 use std::mem::transmute;
+use std::ops::ControlFlow;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 use inputd::{ControlEvent, VtEvent, VtEventKind};
 
 use libredox::errno::ESTALE;
-use redox_scheme::scheme::{SchemeState, SchemeSync};
-use redox_scheme::{CallerCtx, OpenResult, RequestKind, Response, SignalBehavior, Socket};
+use redox_scheme::scheme::SchemeSync;
+use redox_scheme::{CallerCtx, OpenResult, Response, SignalBehavior, Socket};
 
 use orbclient::{Event, EventOption};
-use scheme_utils::HandleMap;
+use scheme_utils::{Blocking, HandleMap};
 use syscall::schemev2::NewFdFlags;
 use syscall::{Error as SysError, EventFlags, EACCES, EBADF, EEXIST, EINVAL};
 
@@ -525,29 +526,16 @@ impl SchemeSync for InputScheme {
 fn deamon(daemon: daemon::SchemeDaemon) -> anyhow::Result<()> {
     // Create the ":input" scheme.
     let socket_file = Socket::create()?;
-    let mut state = SchemeState::new();
     let mut scheme = InputScheme::new();
+    let mut handler = Blocking::new(&socket_file, 16);
 
     let _ = daemon.ready_sync_scheme(&socket_file, &mut scheme);
 
     loop {
         scheme.has_new_events = false;
-        let Some(request) = socket_file.next_request(SignalBehavior::Restart)? else {
-            // Scheme likely got unmounted
-            return Ok(());
-        };
-
-        match request.kind() {
-            RequestKind::Call(call_request) => {
-                socket_file.write_response(
-                    call_request.handle_sync(&mut scheme, &mut state),
-                    SignalBehavior::Restart,
-                )?;
-            }
-            RequestKind::OnClose { id } => {
-                scheme.on_close(id);
-            }
-            _ => {}
+        match handler.process_requests_nonblocking(&mut scheme)? {
+            ControlFlow::Continue(()) => {}
+            ControlFlow::Break(()) => unreachable!("scheme should be non-blocking"),
         }
 
         if !scheme.has_new_events {

drivers/pcid/src/main.rs

@@ -10,10 +10,8 @@ use pci_types::{
     Bar as TyBar, CommandRegister, EndpointHeader, HeaderType, PciAddress,
     PciHeader as TyPciHeader, PciPciBridgeHeader,
 };
-use redox_scheme::{
-    scheme::{register_sync_scheme, SchemeState},
-    RequestKind, SignalBehavior,
-};
+use redox_scheme::scheme::register_sync_scheme;
+use scheme_utils::Blocking;
 
 use crate::cfg_access::Pcie;
 use pcid_interface::{FullDeviceId, LegacyInterruptLine, PciBar, PciFunction, PciRom};
@@ -252,9 +250,9 @@ fn daemon(daemon: daemon::Daemon) -> ! {
 
     info!("PCI SG-BS:DV.F VEND:DEVI CL.SC.IN.RV");
 
-    let mut state = SchemeState::new();
     let mut scheme = scheme::PciScheme::new(pcie);
     let socket = redox_scheme::Socket::create().expect("failed to open pci scheme socket");
+    let handler = Blocking::new(&socket, 16);
 
     {
         match libredox::Fd::open("/scheme/acpi/register_pci", libredox::flag::O_WRONLY, 0) {
@@ -311,30 +309,9 @@ fn daemon(daemon: daemon::Daemon) -> ! {
 
     let _ = daemon.ready();
 
-    loop {
-        let Some(request) = socket
-            .next_request(SignalBehavior::Restart)
-            .expect("pcid: failed to get next scheme request")
-        else {
-            break;
-        };
-        match request.kind() {
-            RequestKind::Call(call) => {
-                let response = call.handle_sync(&mut scheme, &mut state);
-
-                socket
-                    .write_response(response, SignalBehavior::Restart)
-                    .expect("pcid: failed to write next scheme response");
-            }
-            RequestKind::OnClose { id } => {
-                scheme.on_close(id);
-            }
-            _ => (),
-        }
-    }
-
-    println!("pcid: exit");
-    std::process::exit(0);
+    handler
+        .process_requests_blocking(scheme)
+        .expect("pcid: failed to process requests");
 }
 
 fn scan_device(

drivers/pcid/src/scheme.rs

@@ -304,10 +304,8 @@ impl SchemeSync for PciScheme {
             _ => Err(Error::new(EBADF)),
         }
     }
-}
 
-impl PciScheme {
-    pub fn on_close(&mut self, id: usize) {
+    fn on_close(&mut self, id: usize) {
         match self.handles.remove(id) {
             Some(HandleWrapper {
                 inner: Handle::Channel { addr, .. },

drivers/usb/xhcid/Cargo.toml

@@ -22,6 +22,7 @@ lazy_static = "1.4"
 log.workspace = true
 redox_event.workspace = true
 redox-scheme.workspace = true
+scheme-utils = { path = "../../../scheme-utils" }
 redox_syscall.workspace = true
 serde.workspace = true
 serde_json.workspace = true

drivers/usb/xhcid/src/main.rs

@@ -37,10 +37,8 @@ use pcid_interface::irq_helpers::{
 };
 use pcid_interface::{PciFeature, PciFeatureInfo, PciFunctionHandle};
 
-use redox_scheme::{
-    scheme::{register_sync_scheme, SchemeState},
-    RequestKind, SignalBehavior, Socket,
-};
+use redox_scheme::{scheme::register_sync_scheme, Socket};
+use scheme_utils::Blocking;
 
 use crate::xhci::{InterruptMethod, Xhci};
 
@@ -147,8 +145,8 @@ fn daemon_with_context_size<const N: usize>(
 
     let scheme_name = format!("usb.{}", name);
     let socket = Socket::create().expect("xhcid: failed to create usb scheme");
+    let handler = Blocking::new(&socket, 16);
 
-    let mut state = SchemeState::new();
     let hci = Arc::new(
         Xhci::<N>::new(scheme_name.clone(), address, interrupt_method, pcid_handle)
             .expect("xhcid: failed to allocate device"),
@@ -163,28 +161,9 @@ fn daemon_with_context_size<const N: usize>(
 
     hci.poll();
 
-    loop {
-        let Some(request) = socket
-            .next_request(SignalBehavior::Restart)
-            .expect("xhcid: failed to read scheme")
-        else {
-            // Scheme likely got unmounted
-            std::process::exit(0);
-        };
-
-        match request.kind() {
-            RequestKind::Call(call_request) => {
-                let resp = call_request.handle_sync(&mut &*hci, &mut state);
-                socket
-                    .write_response(resp, SignalBehavior::Restart)
-                    .expect("xhcid: failed to write scheme");
-            }
-            RequestKind::OnClose { id } => {
-                hci.on_close(id);
-            }
-            _ => {}
-        }
-    }
+    handler
+        .process_requests_blocking(&*hci)
+        .expect("xhcid: failed to process requests");
 }
 
 fn main() {

drivers/usb/xhcid/src/xhci/scheme.rs

@@ -2350,12 +2350,13 @@ impl<const N: usize> SchemeSync for &Xhci<N> {
             _ => Err(Error::new(EBADF)),
         }
     }
-}
-impl<const N: usize> Xhci<N> {
-    pub fn on_close(&self, fd: usize) {
+
+    fn on_close(&mut self, fd: usize) {
         self.handles.remove(&fd);
     }
+}
 
+impl<const N: usize> Xhci<N> {
     pub fn get_endp_status(&self, port_num: PortId, endp_num: u8) -> Result<EndpointStatus> {
         let port_state = self.port_states.get(&port_num).ok_or(Error::new(EBADFD))?;
 

logd/src/main.rs

@@ -1,8 +1,5 @@
-use redox_scheme::{
-    scheme::{SchemeState, SchemeSync},
-    RequestKind, Response, SignalBehavior, Socket,
-};
-use std::process;
+use redox_scheme::Socket;
+use scheme_utils::Blocking;
 
 use crate::scheme::LogScheme;
 
@@ -11,38 +8,16 @@ mod scheme;
 fn daemon(daemon: daemon::SchemeDaemon) -> ! {
     let socket = Socket::create().expect("logd: failed to create log scheme");
 
-    let mut state = SchemeState::new();
     let mut scheme = LogScheme::new(&socket);
+    let handler = Blocking::new(&socket, 16);
 
     let _ = daemon.ready_sync_scheme(&socket, &mut scheme);
 
     libredox::call::setrens(0, 0).expect("logd: failed to enter null namespace");
 
-    while let Some(request) = socket
-        .next_request(SignalBehavior::Restart)
-        .expect("logd: failed to read events from log scheme")
-    {
-        match request.kind() {
-            RequestKind::Call(call) => {
-                let response = call.handle_sync(&mut scheme, &mut state);
-                socket
-                    .write_response(response, SignalBehavior::Restart)
-                    .expect("logd: failed to write responses to log scheme");
-            }
-            RequestKind::OnClose { id } => {
-                scheme.on_close(id);
-            }
-            RequestKind::SendFd(sendfd_request) => {
-                let result = scheme.on_sendfd(&sendfd_request);
-                let resp = Response::new(result, sendfd_request);
-                socket
-                    .write_response(resp, SignalBehavior::Restart)
-                    .expect("logd: failed to write responses to log scheme");
-            }
-            _ => {}
-        };
-    }
-    process::exit(0);
+    handler
+        .process_requests_blocking(scheme)
+        .expect("logd: failed to process requests");
 }
 
 fn main() {

ramfs/src/main.rs

@@ -1,9 +1,9 @@
-use std::{env, process};
+use std::env;
 
 mod filesystem;
 mod scheme;
 
-use redox_scheme::{scheme::SchemeState, RequestKind, SignalBehavior};
+use scheme_utils::Blocking;
 
 use self::scheme::Scheme;
 
@@ -16,34 +16,14 @@ fn daemon(daemon: daemon::SchemeDaemon) -> ! {
 
     let socket = redox_scheme::Socket::create().expect("ramfs: failed to create socket");
 
-    let mut state = SchemeState::new();
     let mut scheme = Scheme::new(scheme_name.clone()).expect("ramfs: failed to initialize scheme");
+    let handler = Blocking::new(&socket, 16);
 
     let _ = daemon.ready_sync_scheme(&socket, &mut scheme);
 
     libredox::call::setrens(0, 0).expect("ramfs: failed to enter null namespace");
 
-    loop {
-        let Some(request) = socket
-            .next_request(SignalBehavior::Restart)
-            .expect("ramfs: failed to get next scheme request")
-        else {
-            break;
-        };
-        match request.kind() {
-            RequestKind::Call(call) => {
-                let response = call.handle_sync(&mut scheme, &mut state);
-
-                socket
-                    .write_response(response, SignalBehavior::Restart)
-                    .expect("ramfs: failed to write next scheme response");
-            }
-            RequestKind::OnClose { id } => {
-                scheme.on_close(id);
-            }
-            _ => (),
-        }
-    }
-
-    process::exit(0);
+    handler
+        .process_requests_blocking(scheme)
+        .expect("ramfs: failed to process events from zero scheme");
 }

ramfs/src/scheme.rs

@@ -677,9 +677,8 @@ impl SchemeSync for Scheme {
             _ => Err(Error::new(EOPNOTSUPP)),
         }
     }
-}
-impl Scheme {
-    pub fn on_close(&mut self, fd: usize) {
+
+    fn on_close(&mut self, fd: usize) {
         let Some(Handle::Inode(inode)) = self.handles.remove(fd) else {
             return;
         };

randd/src/main.rs

@@ -1,5 +1,3 @@
-use std::process;
-
 use std::arch::asm;
 
 use rand_chacha::ChaCha20Rng;
@@ -13,11 +11,8 @@ pub const MODE_READ: u16 = 0o4;
 #[cfg(target_arch = "x86_64")]
 use raw_cpuid::CpuId;
 
-use redox_scheme::{
-    scheme::{SchemeState, SchemeSync},
-    CallerCtx, OpenResult, RequestKind, SignalBehavior, Socket,
-};
-use scheme_utils::HandleMap;
+use redox_scheme::{scheme::SchemeSync, CallerCtx, OpenResult, Socket};
+use scheme_utils::{Blocking, HandleMap};
 use syscall::data::Stat;
 use syscall::flag::{EventFlags, O_CREAT, O_EXCL, O_RDONLY, O_RDWR, O_WRONLY};
 use syscall::schemev2::NewFdFlags;
@@ -463,30 +458,16 @@ impl SchemeSync for RandScheme {
 fn daemon(daemon: daemon::SchemeDaemon) -> ! {
     let socket = Socket::create().expect("randd: failed to create rand scheme");
 
-    let mut state = SchemeState::new();
     let mut scheme = RandScheme::new();
+    let handler = Blocking::new(&socket, 16);
 
     let _ = daemon.ready_sync_scheme(&socket, &mut scheme);
 
     libredox::call::setrens(0, 0).expect("randd: failed to enter null namespace");
 
-    while let Some(request) = socket
-        .next_request(SignalBehavior::Restart)
-        .expect("error reading packet")
-    {
-        match request.kind() {
-            RequestKind::Call(call) => {
-                let response = call.handle_sync(&mut scheme, &mut state);
-                socket
-                    .write_response(response, SignalBehavior::Restart)
-                    .expect("error writing packet");
-            }
-            RequestKind::OnClose { id } => scheme.on_close(id),
-            _ => {}
-        }
-    }
-
-    process::exit(0);
+    handler
+        .process_requests_blocking(scheme)
+        .expect("randd: failed to process events from zero scheme");
 }
 
 fn main() {

scheme-utils/src/blocking.rs

@@ -0,0 +1,104 @@
+use std::collections::VecDeque;
+use std::ops::ControlFlow;
+
+use libredox::error::Error as LError;
+
+use syscall::Result;
+use syscall::error::{self as errno, Error};
+
+use redox_scheme::scheme::{SchemeState, SchemeSync};
+use redox_scheme::{Request, RequestKind, Response, SignalBehavior, Socket};
+
+pub struct Blocking<'sock> {
+    // TODO: VecDeque for both when it implements spare_capacity
+    requests_read: Vec<Request>,
+    responses_to_write: VecDeque<Response>,
+
+    socket: &'sock Socket,
+    state: SchemeState,
+}
+
+impl<'sock> Blocking<'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),
+            socket,
+            state: SchemeState::new(),
+        }
+    }
+
+    pub fn process_requests_nonblocking(
+        &mut self,
+        scheme: &mut impl SchemeSync,
+    ) -> Result<ControlFlow<()>> {
+        assert!(self.requests_read.is_empty());
+        assert!(self.responses_to_write.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,
+            }) => return Ok(ControlFlow::Break(())),
+            Err(err) => return Err(err),
+        }
+
+        for request in self.requests_read.drain(..) {
+            match request.kind() {
+                RequestKind::Call(req) => {
+                    let response = req.handle_sync(scheme, &mut self.state);
+                    self.responses_to_write.push_back(response);
+                }
+                RequestKind::Cancellation(_req) => {}
+                RequestKind::OnClose { id } => {
+                    // TODO: state.on_close()
+                    scheme.on_close(id);
+                }
+                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);
+                }
+                RequestKind::RecvFd(recvfd_request) => {
+                    let result = scheme.on_recvfd(&recvfd_request);
+                    let response = Response::open_dup_like(result, recvfd_request);
+                    self.responses_to_write.push_back(response);
+                }
+                _ => {}
+            }
+        }
+
+        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(ControlFlow::Continue(())),
+            Err(Error {
+                errno: errno::EINTR | errno::EWOULDBLOCK | errno::EAGAIN,
+            }) => {
+                panic!("scheme response writing should always block");
+            }
+            Err(err) => return Err(LError::from(err).into()),
+        }
+    }
+
+    pub fn process_requests_blocking(mut self, mut scheme: impl SchemeSync) -> Result<!> {
+        loop {
+            match self.process_requests_nonblocking(&mut scheme)? {
+                ControlFlow::Continue(()) => {}
+                ControlFlow::Break(()) => {
+                    panic!("process_requests_blocking should not be used on non-blocking schemes");
+                }
+            }
+        }
+    }
+}

scheme-utils/src/lib.rs

@@ -1,10 +1,14 @@
+#![feature(never_type)]
+
 use std::collections::{BTreeMap, btree_map};
 use std::num::Wrapping;
 
 use syscall::{EBADF, Error, Result};
 
+mod blocking;
 mod readiness_based;
 
+pub use blocking::Blocking;
 pub use readiness_based::ReadinessBased;
 
 pub struct HandleMap<T> {

zerod/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2021"
 [dependencies]
 redox_syscall.workspace = true
 redox-scheme.workspace = true
+scheme-utils = { path = "../scheme-utils" }
 daemon = { path = "../daemon" }
 libredox.workspace = true
 

zerod/src/main.rs

@@ -1,6 +1,7 @@
-use redox_scheme::{scheme::SchemeState, RequestKind, SignalBehavior, Socket};
+use redox_scheme::Socket;
 
 use scheme::ZeroScheme;
+use scheme_utils::Blocking;
 
 mod scheme;
 
@@ -21,29 +22,14 @@ fn daemon(daemon: daemon::SchemeDaemon) -> ! {
     };
 
     let socket = Socket::create().expect("zerod: failed to create zero scheme");
-    let mut state = SchemeState::new();
     let mut zero_scheme = ZeroScheme(ty);
+    let zero_handler = Blocking::new(&socket, 16);
 
     let _ = daemon.ready_sync_scheme(&socket, &mut zero_scheme);
 
     libredox::call::setrens(0, 0).expect("zerod: failed to enter null namespace");
 
-    loop {
-        let Some(request) = socket
-            .next_request(SignalBehavior::Restart)
-            .expect("zerod: failed to read events from zero scheme")
-        else {
-            std::process::exit(0);
-        };
-        match request.kind() {
-            RequestKind::Call(request) => {
-                let response = request.handle_sync(&mut zero_scheme, &mut state);
-
-                socket
-                    .write_response(response, SignalBehavior::Restart)
-                    .expect("zerod: failed to write responses to zero scheme");
-            }
-            _ => (),
-        }
-    }
+    zero_handler
+        .process_requests_blocking(zero_scheme)
+        .expect("zerod: failed to process events from zero scheme");
 }