diff --git a/util/service/BUILD.bazel b/util/service/BUILD.bazel index d92f7783..0861f33a 100644 --- a/util/service/BUILD.bazel +++ b/util/service/BUILD.bazel @@ -1,12 +1,21 @@ # Licensed under the Apache-2.0 license # SPDX-License-Identifier: Apache-2.0 -load("@rules_rust//rust:defs.bzl", "rust_library") +load("@rules_rust//rust:defs.bzl", "rust_library", "rust_test") rust_library( name = "service", - srcs = ["lib.rs"], + srcs = [ + "delayed.rs", + "lib.rs", + "loopback.rs", + ], crate_name = "util_service", edition = "2024", visibility = ["//visibility:public"], ) + +rust_test( + name = "loopback_test", + crate = ":service", +) diff --git a/util/service/README.md b/util/service/README.md index cfd904d6..e369994f 100644 --- a/util/service/README.md +++ b/util/service/README.md @@ -1,104 +1,33 @@ # util_service -The seams every IPC service is built from. `#![no_std]`, host-buildable, no -kernel dependencies. - -A service splits into three parts: wire marshalling in the service's own `api` -crate, a server that turns one request frame into one response frame, and a -transport that carries frames between them. This crate holds the traits that -seam sits on, so a service defines its wire format and nothing else. +Transport and dispatch traits for IPC services. `#![no_std]`, no kernel +dependencies. ## Traits -### [`Transport`](lib.rs) - -One round-trip, caller waits for the response. For a thread dedicated to one -service, or an in-process path before IPC exists. - -```rust -pub trait Transport { - fn transact(&mut self, req: &[u8], resp: &mut [u8]) -> Result; -} -``` - -### [`AsyncTransport`](lib.rs) - -The same round-trip split so the caller never blocks. For a caller running in -an event loop. `poll` returns `Ok(None)` while the response is outstanding. - -```rust -pub trait AsyncTransport { - fn start(&mut self, req: &[u8]) -> Result<(), TransportError>; - fn poll(&mut self, resp: &mut [u8]) -> Result, TransportError>; - fn cancel(&mut self) -> Result<(), TransportError>; -} -``` - -`poll` never waits. The caller parks its event loop until the response to this -round-trip arrives (for a kernel transport, `Signals::READABLE` on its channel -handle, registered with a WaitGroup) and polls once when it does. The handle -comes from the concrete transport at wiring time, not through the trait: a -loopback has no handle to give. - -A server nudging the client out of band, with no round-trip outstanding, is a -separate mechanism and not part of this seam. The nudge carries no payload: the -client answers it by starting a round-trip and asking what happened. - -Neither transport trait is the fallback for the other: a blocking transport has -no way to poll, and an event loop cannot wait. A type implements whichever it -can serve, or both. - -### Using a transport from an event loop - -Two pieces of code, and only one of them is generic. - -Wiring runs once at startup and knows the concrete transport, because what to -park on is a property of that transport and not of the seam. A kernel -transport hands out its channel handle; a loopback has none to give, and needs -none, because its response is ready the moment the request is. - -```rust -// Wiring: concrete type, once at startup. -let mut transport = AsyncChannelTransport::new(handle, send_buf, recv_buf); -wait_group.add(transport.as_raw(), Signals::READABLE)?; -``` - -The client is generic over `AsyncTransport` and never asks what it is talking -to. It encodes a request, starts the round-trip, and returns to the loop. - -```rust -// Client: generic over T: AsyncTransport. -let len = encode_request(&mut req, op)?; -transport.start(&req[..len])?; - -// On each wake, poll each round-trip still in flight. Ok(None) means this -// one's response has not arrived; the wake was for something else. -match transport.poll(&mut resp)? { - Some(len) => handle(decode_response(&resp[..len])?), - None => {} -} -``` +`Transport` blocks until the response arrives. `AsyncTransport` splits the +round-trip so the caller never blocks: `start`, poll `Ok(None)` until +`Ok(Some(len))`. `Dispatch` is the server end, one request frame in, one +response frame out. -That split is why registration is not on the trait. Everything a service -writes once and reuses is in the second block; only the composition root needs -the first, and it is concrete by nature. +Wake-up registration (channel handle, WaitGroup) needs the concrete +transport, so it happens at wiring time, not through the trait. -A server nudge (no round-trip outstanding) wakes the same loop through its own -signal. The client answers it by starting a round-trip and asking what changed. +## Loopback -### [`Dispatch`](lib.rs) +`Loopback` wraps a `Dispatch` and answers in-process. Both transport +traits work because the response is ready the moment the request is. `N` +sizes the async response buffer. The blocking path writes into the +caller's buffer directly. A loopback has no send buffer, so it never refuses a request at `start`. +If the request is too large, the server sees it and answers with a +protocol error frame instead. -The server end. One request frame in, one response frame out, no state between -calls beyond what the service itself owns. The same impl backs the production -channel and the in-process loopback, so host tests exercise the real server. +Because `poll` never returns `Ok(None)`, a loopback cannot test not-ready +handling on its own. Wrap it in `Delayed` (below) for that. -```rust -pub trait Dispatch { - fn dispatch(&mut self, request: &[u8], response: &mut [u8]) -> Result; -} -``` +## Delayed -A service encodes its own errors into the response frame, so a failed operation -is still a frame and still `Ok`. `DispatchError::ResponseTooSmall` is the one -case with nothing to send back: the response buffer cannot hold even an error -frame. +`Delayed` wraps any `AsyncTransport` and returns `Ok(None)` for a set +number of polls before forwarding to the inner transport. This is how a +host test reaches a client's not-ready path when the underlying transport +(like `Loopback`) always answers immediately. diff --git a/util/service/delayed.rs b/util/service/delayed.rs new file mode 100644 index 00000000..909c131e --- /dev/null +++ b/util/service/delayed.rs @@ -0,0 +1,151 @@ +// Licensed under the Apache-2.0 license +// SPDX-License-Identifier: Apache-2.0 + +//! A transport wrapper that withholds the response for a fixed number of +//! polls. Loopback answers on the first poll, so a host test cannot reach +//! a client's not-ready path without this. + +use crate::{AsyncTransport, TransportError}; + +/// Wraps a transport and returns `Ok(None)` for the first `ready_after` +/// polls of each round-trip before forwarding to the inner transport. +pub struct Delayed { + inner: T, + ready_after: usize, + // None while idle, Some(n) after n withheld polls. Keeping idle + // distinct means a poll with nothing started reaches the inner + // transport and gets WrongState, not a spurious Ok(None). + polls: Option, +} + +impl Delayed { + /// `ready_after` is how many polls return `Ok(None)` before the inner + /// transport is polled. Zero passes through on the first poll. + pub const fn new(inner: T, ready_after: usize) -> Self { + Self { + inner, + ready_after, + polls: None, + } + } + + pub fn inner(&self) -> &T { + &self.inner + } +} + +impl AsyncTransport for Delayed { + fn start(&mut self, req: &[u8]) -> Result<(), TransportError> { + self.inner.start(req)?; + self.polls = Some(0); + Ok(()) + } + + fn poll(&mut self, resp: &mut [u8]) -> Result, TransportError> { + if let Some(n) = self.polls { + if n < self.ready_after { + self.polls = Some(n + 1); + return Ok(None); + } + } + let result = self.inner.poll(resp); + // Anything but "not yet" ends the round-trip, so the next call + // is start and the count begins again. + if !matches!(result, Ok(None)) { + self.polls = None; + } + result + } + + fn cancel(&mut self) -> Result<(), TransportError> { + self.polls = None; + self.inner.cancel() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{Dispatch, DispatchError, Loopback}; + + struct Increment { + seen: usize, + } + + impl Dispatch for Increment { + fn dispatch( + &mut self, + request: &[u8], + response: &mut [u8], + ) -> Result { + self.seen += 1; + if response.is_empty() { + return Err(DispatchError::ResponseTooSmall); + } + response[0] = request[0].wrapping_add(1); + Ok(1) + } + } + + fn delayed(ready_after: usize) -> Delayed> { + Delayed::new(Loopback::new(Increment { seen: 0 }), ready_after) + } + + #[test] + fn poll_returns_none_until_the_delay_elapses() { + let mut d = delayed(3); + let mut resp = [0u8; 4]; + + assert_eq!(d.start(&[0x10]), Ok(())); + assert_eq!(d.poll(&mut resp), Ok(None)); + assert_eq!(d.poll(&mut resp), Ok(None)); + assert_eq!(d.poll(&mut resp), Ok(None)); + } + + #[test] + fn poll_after_the_delay_yields_the_response() { + let mut d = delayed(2); + let mut resp = [0u8; 4]; + + assert_eq!(d.start(&[0x10]), Ok(())); + assert_eq!(d.poll(&mut resp), Ok(None)); + assert_eq!(d.poll(&mut resp), Ok(None)); + assert_eq!(d.poll(&mut resp), Ok(Some(1))); + assert_eq!(resp[0], 0x11); + } + + #[test] + fn poll_with_nothing_in_flight_is_wrong_state() { + let mut d = delayed(2); + let mut resp = [0u8; 4]; + + assert_eq!(d.poll(&mut resp), Err(TransportError::WrongState)); + } + + #[test] + fn cancel_resets_the_delay_for_the_next_round_trip() { + let mut d = delayed(2); + let mut resp = [0u8; 4]; + + assert_eq!(d.start(&[0x10]), Ok(())); + assert_eq!(d.poll(&mut resp), Ok(None)); + assert_eq!(d.cancel(), Ok(())); + + // Next round-trip starts the delay count from scratch. + assert_eq!(d.start(&[0x20]), Ok(())); + assert_eq!(d.poll(&mut resp), Ok(None)); + assert_eq!(d.poll(&mut resp), Ok(None)); + assert_eq!(d.poll(&mut resp), Ok(Some(1))); + assert_eq!(resp[0], 0x21); + } + + #[test] + fn a_zero_delay_answers_on_the_first_poll() { + let mut d = delayed(0); + let mut resp = [0u8; 4]; + + assert_eq!(d.start(&[0x10]), Ok(())); + assert_eq!(d.poll(&mut resp), Ok(Some(1))); + assert_eq!(resp[0], 0x11); + } +} diff --git a/util/service/lib.rs b/util/service/lib.rs index a41ce3a6..2c13a3c5 100644 --- a/util/service/lib.rs +++ b/util/service/lib.rs @@ -1,42 +1,31 @@ // Licensed under the Apache-2.0 license // SPDX-License-Identifier: Apache-2.0 -//! The seams every IPC service is built from. +//! Transport and dispatch traits for IPC services. //! -//! A service splits into wire marshalling (host-buildable, in the service's -//! own `api` crate), a server that turns one request frame into one response -//! frame, and a transport that carries frames between the two. This crate -//! holds the three traits that seam sits on, so a service defines its wire -//! format and nothing else. -//! -//! A transport comes in two shapes and a type implements whichever it can -//! serve. `Transport` blocks until the response arrives, which is what a -//! dedicated server thread or an early-boot in-process path wants. -//! `AsyncTransport` starts a round-trip and returns, so a caller running in -//! an event loop never blocks. Neither is the fallback for the other: a -//! blocking transport has no way to poll, and an event loop cannot wait. -//! -//! `Dispatch` is the server end. One request frame in, one response frame -//! out, no state between calls. The same impl backs the production channel -//! and the in-process loopback, so host tests exercise the real server. +//! Three traits: `Transport` (blocking), `AsyncTransport` (split-phase), +//! and `Dispatch` (server). A service defines its wire format in its own +//! crate and plugs into these. #![no_std] -/// Why a transport round-trip failed. Small and service-neutral; -/// service-level status travels inside the response payload, not here. +mod delayed; +mod loopback; + +pub use delayed::Delayed; +pub use loopback::Loopback; + +/// Why a transport round-trip failed. /// -/// `WrongState` belongs to `AsyncTransport`: a blocking `transact` has no -/// state between calls to get wrong. The other two apply to both. +/// Service-level errors travel inside the response payload, not here. #[non_exhaustive] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum TransportError { - /// The underlying channel, syscall, or loopback call failed. + /// The channel, syscall, or loopback call failed. Failed, - /// `start` was called while a round-trip was still in flight, or - /// `poll`/`cancel` was called with nothing in flight. + /// `start` while in flight, or `poll`/`cancel` with nothing in flight. WrongState, - /// The request does not fit the transport's request buffer, or the - /// response does not fit the caller's. + /// Request or response does not fit its buffer. TooLarge, } @@ -52,63 +41,49 @@ impl core::fmt::Display for TransportError { impl core::error::Error for TransportError {} -/// Bytes in, bytes out, one round-trip, caller waits for the response. -/// -/// `transact` writes the response into `resp` and returns its length. The -/// request is one fully serialized frame and so is the response. No -/// fragmentation, no state between calls. -/// -/// Implement this when the caller can afford to wait: a thread dedicated to -/// one service, or an in-process path before IPC exists. A caller inside an -/// event loop wants `AsyncTransport` instead. +/// One round-trip, caller waits for the response. pub trait Transport { fn transact(&mut self, req: &[u8], resp: &mut [u8]) -> Result; } -/// Bytes in, bytes out, one round-trip at a time, split so the caller never -/// blocks. +/// One round-trip at a time, split so the caller never blocks. /// -/// `start` takes one fully serialized request and returns immediately; the -/// transport copies what it needs, so `req` is free afterwards. `poll` -/// returns `Ok(None)` while the response is still outstanding and -/// `Ok(Some(len))` once `resp[..len]` holds one fully serialized reply. -/// `len` is never 0: a server that cannot produce a frame at all is a -/// failed round-trip, not an empty reply. +/// `start` copies the request and returns immediately. `poll` returns +/// `Ok(None)` while the response is outstanding, `Ok(Some(len))` when +/// `resp[..len]` holds the reply. `poll` never waits. /// -/// `poll` never waits: with no response ready it returns `Ok(None)` and -/// returns. The caller is expected to be signal-driven rather than -/// spinning, parking its event loop until the response to this round-trip -/// arrives and polling once when it does. Registering for that needs the -/// concrete transport (a kernel one hands out its channel handle, a -/// loopback has none), so it happens at wiring time, not through this -/// trait. +/// One round-trip at a time: `start` while in flight or `poll`/`cancel` +/// with nothing in flight is `WrongState`. Any error from `poll` ends +/// the round-trip. /// -/// A server that raises a signal to say it has news, with no round-trip -/// outstanding, is not this trait's concern. The caller learns what the -/// news is by starting a round-trip and asking. +/// A request that is too large for the transport is caught at different +/// points depending on the implementation. A channel transport checks at +/// `start` and returns `TooLarge` before anything goes out. A loopback +/// has no send buffer to overflow, so it hands the request straight to +/// the server, which answers with a protocol error frame at `poll`. +/// Callers need to handle both cases. A response too large for the +/// caller's buffer is always `TooLarge` from `poll`. /// -/// A transport carries one round-trip at a time: `start` while another is in -/// flight, or `poll`/`cancel` with none, is `WrongState`. Any error from -/// `poll` ends the round-trip, so the next call is `start`. +/// Registering for wake-up signals needs the concrete transport (a +/// kernel channel has a handle, a loopback does not), so that happens +/// at wiring time, not through this trait. pub trait AsyncTransport { fn start(&mut self, req: &[u8]) -> Result<(), TransportError>; fn poll(&mut self, resp: &mut [u8]) -> Result, TransportError>; - /// Abandon the round-trip in flight. The response, if one arrives, is - /// discarded. + /// Abandon the in-flight round-trip. fn cancel(&mut self) -> Result<(), TransportError>; } /// Why a dispatch produced no response frame at all. /// -/// A service encodes its own errors into the response frame, so a failed -/// operation is still a frame and still `Ok`. This is the one case where -/// there is nothing to send back. +/// Service errors go in the response frame (still `Ok`). This covers +/// the case where even an error frame does not fit. #[non_exhaustive] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DispatchError { - /// `response` is too small to hold even an error frame. + /// Response buffer too small for any frame. ResponseTooSmall, } @@ -122,15 +97,18 @@ impl core::fmt::Display for DispatchError { impl core::error::Error for DispatchError {} -/// The server end: one request frame in, one response frame out. -/// -/// Returns the number of bytes written to `response`, always at least one. -/// A loopback transport writes into the caller's buffer, so it reports a -/// `DispatchError` as `TransportError::TooLarge`. +impl From for TransportError { + fn from(e: DispatchError) -> Self { + match e { + DispatchError::ResponseTooSmall => Self::TooLarge, + } + } +} + +/// Server end: one request frame in, one response frame out. /// -/// Implementations hold no state between calls beyond whatever the service -/// itself owns, so the same impl serves the production channel and the -/// in-process loopback. +/// Returns bytes written to `response`. No state between calls beyond +/// what the service itself owns. pub trait Dispatch { fn dispatch(&mut self, request: &[u8], response: &mut [u8]) -> Result; } diff --git a/util/service/loopback.rs b/util/service/loopback.rs new file mode 100644 index 00000000..3f8d2088 --- /dev/null +++ b/util/service/loopback.rs @@ -0,0 +1,209 @@ +// Licensed under the Apache-2.0 license +// SPDX-License-Identifier: Apache-2.0 + +//! In-process loopback transport for host tests. +//! +//! Calls the server's `Dispatch` directly, no kernel. The response is +//! ready on the first `poll`, so this cannot test not-ready handling. + +use crate::{AsyncTransport, Dispatch, Transport, TransportError}; + +/// In-process transport over an owned server. +/// +/// `N` sizes the async response buffer held between `start` and `poll`. +/// The blocking path writes into the caller's buffer directly. +pub struct Loopback { + server: D, + pending: Option, + held: [u8; N], +} + +impl Loopback { + pub const fn new(server: D) -> Self { + Self { + server, + pending: None, + held: [0u8; N], + } + } + + pub fn server(&self) -> &D { + &self.server + } +} + +impl Transport for Loopback { + fn transact(&mut self, req: &[u8], resp: &mut [u8]) -> Result { + if self.pending.is_some() { + return Err(TransportError::WrongState); + } + Ok(self.server.dispatch(req, resp)?) + } +} + +impl AsyncTransport for Loopback { + fn start(&mut self, req: &[u8]) -> Result<(), TransportError> { + if self.pending.is_some() { + return Err(TransportError::WrongState); + } + let len = self.server.dispatch(req, &mut self.held)?; + self.pending = Some(len); + Ok(()) + } + + fn poll(&mut self, resp: &mut [u8]) -> Result, TransportError> { + let Some(len) = self.pending else { + return Err(TransportError::WrongState); + }; + if len > resp.len() { + self.pending = None; + return Err(TransportError::TooLarge); + } + self.pending = None; + resp[..len].copy_from_slice(&self.held[..len]); + Ok(Some(len)) + } + + fn cancel(&mut self) -> Result<(), TransportError> { + if self.pending.take().is_none() { + return Err(TransportError::WrongState); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::DispatchError; + + /// Increments the first byte of the request. + struct Increment { + seen: usize, + } + + impl Dispatch for Increment { + fn dispatch( + &mut self, + request: &[u8], + response: &mut [u8], + ) -> Result { + self.seen += 1; + if response.is_empty() { + return Err(DispatchError::ResponseTooSmall); + } + response[0] = request[0].wrapping_add(1); + Ok(1) + } + } + + fn loopback() -> Loopback { + Loopback::new(Increment { seen: 0 }) + } + + #[test] + fn blocking_transact_answers_from_the_server() { + let mut lb = loopback(); + let mut resp = [0u8; 4]; + + assert_eq!(lb.transact(&[0x10], &mut resp), Ok(1)); + assert_eq!(resp[0], 0x11); + assert_eq!(lb.server().seen, 1); + } + + #[test] + fn blocking_transact_reports_a_response_buffer_that_does_not_fit() { + let mut lb = loopback(); + let mut resp = [0u8; 0]; + + assert_eq!( + lb.transact(&[0x10], &mut resp), + Err(TransportError::TooLarge) + ); + } + + #[test] + fn async_round_trip_is_ready_on_the_first_poll() { + let mut lb = loopback(); + let mut resp = [0u8; 4]; + + assert_eq!(lb.start(&[0x20]), Ok(())); + assert_eq!(lb.poll(&mut resp), Ok(Some(1))); + assert_eq!(resp[0], 0x21); + } + + #[test] + fn a_second_start_while_one_is_in_flight_is_refused() { + let mut lb = loopback(); + + assert_eq!(lb.start(&[0x20]), Ok(())); + assert_eq!(lb.start(&[0x20]), Err(TransportError::WrongState)); + assert_eq!(lb.server().seen, 1); + } + + #[test] + fn poll_with_nothing_in_flight_is_refused() { + let mut lb = loopback(); + let mut resp = [0u8; 4]; + + assert_eq!(lb.poll(&mut resp), Err(TransportError::WrongState)); + } + + #[test] + fn a_response_that_does_not_fit_ends_the_round_trip() { + let mut lb = loopback(); + let mut small = [0u8; 0]; + let mut resp = [0u8; 4]; + + assert_eq!(lb.start(&[0x20]), Ok(())); + assert_eq!(lb.poll(&mut small), Err(TransportError::TooLarge)); + + assert_eq!(lb.start(&[0x30]), Ok(())); + assert_eq!(lb.poll(&mut resp), Ok(Some(1))); + assert_eq!(resp[0], 0x31); + } + + #[test] + fn cancel_frees_the_transport_for_the_next_request() { + let mut lb = loopback(); + let mut resp = [0u8; 4]; + + assert_eq!(lb.start(&[0x20]), Ok(())); + assert_eq!(lb.cancel(), Ok(())); + assert_eq!(lb.poll(&mut resp), Err(TransportError::WrongState)); + + assert_eq!(lb.start(&[0x40]), Ok(())); + assert_eq!(lb.poll(&mut resp), Ok(Some(1))); + assert_eq!(resp[0], 0x41); + } + + #[test] + fn cancel_with_nothing_in_flight_is_refused() { + let mut lb = loopback(); + + assert_eq!(lb.cancel(), Err(TransportError::WrongState)); + } + + #[test] + fn transact_during_a_started_round_trip_is_refused() { + let mut lb = loopback(); + let mut resp = [0u8; 4]; + + assert_eq!(lb.start(&[0x20]), Ok(())); + assert_eq!( + lb.transact(&[0x30], &mut resp), + Err(TransportError::WrongState) + ); + assert_eq!(lb.poll(&mut resp), Ok(Some(1))); + assert_eq!(resp[0], 0x21); + } + + #[test] + fn a_dispatch_that_cannot_answer_is_reported_at_start() { + let mut lb = Loopback::::new(Increment { seen: 0 }); + + assert_eq!(lb.start(&[0x20]), Err(TransportError::TooLarge)); + // Failed start leaves nothing in flight. + assert_eq!(lb.cancel(), Err(TransportError::WrongState)); + } +}