diff --git a/runtime/core/src/client.rs b/runtime/core/src/client.rs index 04c4b84..c8c6b58 100644 --- a/runtime/core/src/client.rs +++ b/runtime/core/src/client.rs @@ -15,6 +15,8 @@ use alloc::vec::Vec; +use core::time::Duration; + use serde::Serialize; use crate::contract::{Envelope, RuntimeError, WireFormat}; @@ -54,6 +56,34 @@ impl Client { /// response must be decoded (or dropped) before the next call. A pipelined /// / multiplexed client is a later, additive layer. pub fn call

(&mut self, call_id: u16, params: &P) -> Result<(Envelope<'_>, &W), RuntimeError> + where + P: Serialize + ?Sized, + { + self.request_response(call_id, params, None) + } + + /// [`call`](Self::call), but give up after `timeout` waiting for the + /// response — `Err(RuntimeError::Timeout)`. What a generated stub emits for + /// a `@timeout_ms` function annotation. Honoured only by transports that + /// override [`Transport::recv_timeout`] (the `std` ones do); others block. + pub fn call_with_timeout

( + &mut self, + call_id: u16, + params: &P, + timeout: Duration, + ) -> Result<(Envelope<'_>, &W), RuntimeError> + where + P: Serialize + ?Sized, + { + self.request_response(call_id, params, Some(timeout)) + } + + fn request_response

( + &mut self, + call_id: u16, + params: &P, + timeout: Option, + ) -> Result<(Envelope<'_>, &W), RuntimeError> where P: Serialize + ?Sized, { @@ -66,7 +96,14 @@ impl Client { self.transport.send(&self.request)?; self.response.clear(); - self.transport.recv(&mut self.response)?; + match timeout { + None => self.transport.recv(&mut self.response)?, + Some(d) => { + if !self.transport.recv_timeout(&mut self.response, d)? { + return Err(RuntimeError::Timeout); + } + } + } let (echoed, envelope) = wire::decode_response(&self.response).ok_or(RuntimeError::Framing)?; diff --git a/runtime/core/src/transport.rs b/runtime/core/src/transport.rs index b67b4fb..9041cf6 100644 --- a/runtime/core/src/transport.rs +++ b/runtime/core/src/transport.rs @@ -6,6 +6,8 @@ //! natively. A byte stream ([`Tcp`]) has no message boundaries, so it adds a //! `u32` length prefix per frame. +use core::time::Duration; + use alloc::vec::Vec; use crate::contract::RuntimeError; @@ -18,6 +20,19 @@ pub trait Transport { /// Receive the next frame into `buf` (the caller clears and reuses it). /// `Err(RuntimeError::Transport)` once the peer is gone. fn recv(&mut self, buf: &mut Vec) -> Result<(), RuntimeError>; + + /// Receive the next frame, waiting at most `timeout`. `Ok(true)` — a + /// frame was read into `buf`; `Ok(false)` — the timeout elapsed first. + /// + /// The default **blocks**, ignoring `timeout` — a transport with no clock + /// still works, it just can't honour a per-call deadline. `std` transports + /// override this. [`Client::call_with_timeout`](crate::client::Client::call_with_timeout) + /// is the caller. + fn recv_timeout(&mut self, buf: &mut Vec, timeout: Duration) -> Result { + let _ = timeout; + self.recv(buf)?; + Ok(true) + } } #[cfg(feature = "std")] @@ -72,6 +87,23 @@ mod in_memory { buf.extend_from_slice(&frame); Ok(()) } + + fn recv_timeout( + &mut self, + buf: &mut Vec, + timeout: super::Duration, + ) -> Result { + use std::sync::mpsc::RecvTimeoutError; + match self.rx.recv_timeout(timeout) { + Ok(frame) => { + buf.clear(); + buf.extend_from_slice(&frame); + Ok(true) + } + Err(RecvTimeoutError::Timeout) => Ok(false), + Err(RecvTimeoutError::Disconnected) => Err(RuntimeError::Transport), + } + } } } @@ -80,10 +112,10 @@ pub use in_memory::{duplex, InMemory}; #[cfg(feature = "std")] mod tcp { - use std::io::{Read, Write}; + use std::io::{ErrorKind, Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; - use super::{Transport, Vec}; + use super::{Duration, Transport, Vec}; use crate::contract::RuntimeError; /// Reject a length prefix larger than this before allocating for it — a @@ -144,6 +176,49 @@ mod tcp { .read_exact(buf) .map_err(|_| RuntimeError::Transport) } + + /// Best-effort: sets a read timeout for the length-prefix and body + /// reads, then clears it. A timeout part-way through a frame leaves the + /// stream desynced — treat a timed-out call as fatal to the connection + /// and drop the [`Tcp`]. + fn recv_timeout( + &mut self, + buf: &mut Vec, + timeout: Duration, + ) -> Result { + self.stream + .set_read_timeout(Some(timeout)) + .map_err(|_| RuntimeError::Transport)?; + let outcome = self.recv_within(buf); + let _ = self.stream.set_read_timeout(None); + outcome + } + } + + impl Tcp { + fn recv_within(&mut self, buf: &mut Vec) -> Result { + let timed_out = |e: &std::io::Error| { + matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) + }; + if let Err(e) = self.stream.read_exact(&mut self.len) { + return if timed_out(&e) { + Ok(false) + } else { + Err(RuntimeError::Transport) + }; + } + let len = u32::from_le_bytes(self.len) as usize; + if len > MAX_FRAME { + return Err(RuntimeError::Framing); + } + buf.clear(); + buf.resize(len, 0); + match self.stream.read_exact(buf) { + Ok(()) => Ok(true), + Err(e) if timed_out(&e) => Ok(false), + Err(_) => Err(RuntimeError::Transport), + } + } } } diff --git a/runtime/core/tests/timeout_roundtrip.rs b/runtime/core/tests/timeout_roundtrip.rs new file mode 100644 index 0000000..ac34e4e --- /dev/null +++ b/runtime/core/tests/timeout_roundtrip.rs @@ -0,0 +1,97 @@ +//! `Client::call_with_timeout` — what a generated stub emits for a +//! `@timeout_ms` function annotation. Over `InMemory` (which overrides +//! `Transport::recv_timeout`): the call gives up when no reply comes, and +//! succeeds normally when one does before the deadline. +#![cfg(feature = "std")] + +use std::thread; +use std::time::Duration; + +use comline_runtime::client::Client; +use comline_runtime::contract::{BufMut, Dispatch, Envelope, Kind, RuntimeError, WireFormat}; +use comline_runtime::format::MsgPack; +use comline_runtime::serve::Server; +use comline_runtime::transport::duplex; +use serde::{Deserialize, Serialize}; + +// protocol Echo { function ping(n: u32) -> u32; } + +#[derive(Serialize, Deserialize)] +struct PingParams { + n: u32, +} + +const CALLS: &[&str] = &["ping"]; + +trait Echo { + fn ping(&self, n: u32) -> u32; +} + +struct EchoDispatcher(T); + +impl Dispatch for EchoDispatcher { + fn dispatch( + &self, + call: Kind, + params: &[u8], + fmt: &W, + out: &mut dyn BufMut, + ) -> Result<(), RuntimeError> { + match call.resolve(CALLS).ok_or(RuntimeError::UnknownCall)? { + 0 => { + let p: PingParams = fmt.decode(params)?; + let mut body = Vec::new(); + fmt.encode(&self.0.ping(p.n), &mut body)?; + Envelope::encode_ok(&body, out); + Ok(()) + } + _ => Err(RuntimeError::UnknownCall), + } + } +} + +struct Inc; +impl Echo for Inc { + fn ping(&self, n: u32) -> u32 { + n + 1 + } +} + +#[test] +fn a_call_that_gets_no_reply_times_out() { + // `_provider` end stays bound (so the channel isn't disconnected) but + // nothing serves it. + let (client_side, _provider) = duplex(); + let mut client = Client::new(client_side, MsgPack); + + let err = client + .call_with_timeout(0, &PingParams { n: 1 }, Duration::from_millis(50)) + .unwrap_err(); + assert_eq!(err, RuntimeError::Timeout); +} + +#[test] +fn a_call_answered_before_the_deadline_succeeds() { + let (client_side, provider_side) = duplex(); + + let provider = thread::spawn(move || { + let mut provider_side = provider_side; + Server::new(EchoDispatcher(Inc), MsgPack) + .serve(&mut provider_side) + .unwrap(); + }); + + let mut client = Client::new(client_side, MsgPack); + + let (reply, fmt) = client + .call_with_timeout(0, &PingParams { n: 41 }, Duration::from_secs(5)) + .unwrap(); + let n: u32 = match reply { + Envelope::Ok(payload) => fmt.decode(payload).unwrap(), + Envelope::Err { .. } => panic!("unexpected error frame"), + }; + assert_eq!(n, 42); + + drop(client); + provider.join().unwrap(); +}