1208 lines
44 KiB
Rust
1208 lines
44 KiB
Rust
use std::{
|
|
future::{poll_fn, Future},
|
|
io,
|
|
net::SocketAddr,
|
|
pin::Pin,
|
|
task::{Context, Poll},
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
#[cfg(test)]
|
|
use std::{
|
|
cell::{Cell, RefCell},
|
|
collections::VecDeque,
|
|
};
|
|
|
|
use tokio::time;
|
|
|
|
use crate::{
|
|
client::{
|
|
echo_sent_event,
|
|
schedule::{instant_abs_diff, ProbeSchedule},
|
|
validate_datagram_length,
|
|
},
|
|
config::ClientConfig,
|
|
error::ClientError,
|
|
event::{ClientEvent, OpenOutcome},
|
|
receive::{drain_tx_timestamps, try_enable_tx_timestamping, try_recv_tokio_datagram},
|
|
session::machine::{
|
|
recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest,
|
|
PreparedProbe, SessionMachine, TimeoutBatch, MAX_OPEN_PACKET_SIZE,
|
|
},
|
|
socket::{connect_tokio_udp_socket, resolve_remote_tokio, validate_open_timeouts},
|
|
socket_options::{apply_traffic_class_to_tokio_socket, clear_dscp_on_tokio_socket},
|
|
timing::ClientTimestamp,
|
|
};
|
|
|
|
/// Maximum amount of immediately available opening work performed by one poll.
|
|
///
|
|
/// Opening may consume ignored UDP datagrams, roll over an expired attempt, or
|
|
/// encounter readiness false positives without ever awaiting. Bound that work
|
|
/// so one managed target cannot retain the current-thread runtime.
|
|
const OPEN_POLL_WORK_BUDGET: usize = 64;
|
|
|
|
/// Maximum immediately available receive work performed by one
|
|
/// [`AsyncClient::poll_recv`] poll.
|
|
///
|
|
/// A `WouldBlock` readiness false positive is naturally bounded: `try_io`
|
|
/// clears the reactor's readiness on `WouldBlock`, so the next
|
|
/// `poll_recv_ready` genuinely awaits a fresh edge. An `Interrupted`
|
|
/// (`EINTR`) nonblocking `recvmsg` does not clear that readiness — the
|
|
/// syscall was interrupted, not told there was nothing to read — so without
|
|
/// an explicit bound a sustained signal delivery rate could otherwise retry
|
|
/// in a tight loop on this executor thread. Bounding both cases the same way
|
|
/// this crate already bounds open-attempt work keeps that impossible.
|
|
const RECV_POLL_WORK_BUDGET: usize = 64;
|
|
|
|
#[cfg(test)]
|
|
use crate::client::ProbeSendTimestamps;
|
|
|
|
#[derive(Debug)]
|
|
struct PreparedAsyncOpen {
|
|
machine: PreparedOpenAcceptance,
|
|
schedule: Option<ProbeSchedule>,
|
|
recv_buffer_len: Option<usize>,
|
|
negotiated_traffic_class: Option<u8>,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
struct PreparedAsyncOpenFailure {
|
|
primary: ClientError,
|
|
machine: PreparedOpenAcceptance,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
enum AsyncOpenCleanup {
|
|
Acceptance {
|
|
primary: ClientError,
|
|
packet: Option<Box<[u8]>>,
|
|
},
|
|
Adapter(Box<PreparedAsyncOpenFailure>),
|
|
}
|
|
|
|
impl AsyncOpenCleanup {
|
|
fn packet(&self) -> Option<&[u8]> {
|
|
match self {
|
|
Self::Acceptance { packet, .. } => packet.as_deref(),
|
|
Self::Adapter(failure) => failure.machine.cleanup_close_packet(),
|
|
}
|
|
}
|
|
|
|
fn into_primary(self) -> ClientError {
|
|
match self {
|
|
Self::Acceptance { primary, .. } => primary,
|
|
Self::Adapter(failure) => failure.primary,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Owned state for one asynchronous open transaction.
|
|
///
|
|
/// This state deliberately contains no reference to [`AsyncClient`], allowing
|
|
/// a future managed target to store the client and opening state side by side.
|
|
#[derive(Debug)]
|
|
pub(crate) struct AsyncOpenState {
|
|
attempt: usize,
|
|
deadline: Option<Instant>,
|
|
deadline_timer: Option<Pin<Box<time::Sleep>>>,
|
|
request_submitted: bool,
|
|
stop_after_current_attempt: bool,
|
|
stopped_after_current_attempt: bool,
|
|
buffer: [u8; MAX_OPEN_PACKET_SIZE],
|
|
cleanup: Option<AsyncOpenCleanup>,
|
|
}
|
|
|
|
impl AsyncOpenState {
|
|
pub(crate) fn new() -> Self {
|
|
Self {
|
|
attempt: 0,
|
|
deadline: None,
|
|
deadline_timer: None,
|
|
request_submitted: false,
|
|
stop_after_current_attempt: false,
|
|
stopped_after_current_attempt: false,
|
|
buffer: [0_u8; MAX_OPEN_PACKET_SIZE],
|
|
cleanup: None,
|
|
}
|
|
}
|
|
|
|
fn start_attempt(&mut self, timeout: Duration) -> Result<(), ClientError> {
|
|
let deadline = Instant::now()
|
|
.checked_add(timeout)
|
|
.ok_or(ClientError::DurationOverflow)?;
|
|
self.deadline = Some(deadline);
|
|
self.deadline_timer = Some(Box::pin(time::sleep_until(deadline.into())));
|
|
self.request_submitted = false;
|
|
Ok(())
|
|
}
|
|
|
|
fn finish_attempt(&mut self) {
|
|
self.attempt += 1;
|
|
self.deadline = None;
|
|
self.deadline_timer = None;
|
|
self.request_submitted = false;
|
|
self.stopped_after_current_attempt = self.stop_after_current_attempt;
|
|
}
|
|
|
|
pub(crate) fn has_in_flight_work(&self) -> bool {
|
|
self.request_submitted || self.cleanup.is_some()
|
|
}
|
|
|
|
pub(crate) fn request_stop_after_current_attempt(&mut self) {
|
|
self.stop_after_current_attempt = true;
|
|
}
|
|
|
|
pub(crate) fn stopped_after_current_attempt(&self) -> bool {
|
|
self.stopped_after_current_attempt
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn has_retained_cleanup(&self) -> bool {
|
|
self.cleanup.is_some()
|
|
}
|
|
|
|
fn deadline(&self) -> Instant {
|
|
self.deadline
|
|
.expect("an open attempt has an active deadline")
|
|
}
|
|
|
|
fn deadline_timer(&mut self) -> &mut Pin<Box<time::Sleep>> {
|
|
self.deadline_timer
|
|
.as_mut()
|
|
.expect("an open attempt has an active deadline timer")
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[derive(Debug)]
|
|
enum InjectedSend {
|
|
WouldBlock,
|
|
Error,
|
|
ReportedLength(usize),
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[derive(Debug, Default)]
|
|
struct AsyncClientTestHooks {
|
|
sends: RefCell<VecDeque<InjectedSend>>,
|
|
send_attempts: Cell<usize>,
|
|
receive_would_block: Cell<usize>,
|
|
/// Number of subsequent receive attempts (`poll_open` or `poll_recv`)
|
|
/// that should report `io::ErrorKind::Interrupted` instead of performing
|
|
/// a real nonblocking receive, deterministically exercising interrupted-
|
|
/// syscall retry behavior.
|
|
inject_recv_interrupted: Cell<usize>,
|
|
fail_open_dscp: Cell<bool>,
|
|
fail_cleanup_send: Cell<bool>,
|
|
fail_peer_close_dscp: Cell<bool>,
|
|
fail_dscp_restore: Cell<bool>,
|
|
pause_open_cleanup_before_send: Cell<bool>,
|
|
pause_open_before_writable: Cell<bool>,
|
|
pause_open_before_readable: Cell<bool>,
|
|
pause_probe_before_writable: Cell<bool>,
|
|
pause_close_before_writable: Cell<bool>,
|
|
pause_close_after_would_block: Cell<bool>,
|
|
probe_timestamps: RefCell<VecDeque<ProbeSendTimestamps>>,
|
|
close_sent_at: Cell<Option<ClientTimestamp>>,
|
|
}
|
|
|
|
#[cfg(test)]
|
|
impl AsyncClientTestHooks {
|
|
fn try_send(&self, socket: &tokio::net::UdpSocket, packet: &[u8]) -> io::Result<usize> {
|
|
self.send_attempts.set(self.send_attempts.get() + 1);
|
|
if let Some(injected) = self.sends.borrow_mut().pop_front() {
|
|
return match injected {
|
|
InjectedSend::WouldBlock => Err(io::Error::from(io::ErrorKind::WouldBlock)),
|
|
InjectedSend::Error => Err(io::Error::other("injected send failure")),
|
|
InjectedSend::ReportedLength(bytes) => Ok(bytes),
|
|
};
|
|
}
|
|
socket.try_send(packet)
|
|
}
|
|
|
|
fn take_probe_timestamps(&self) -> Option<ProbeSendTimestamps> {
|
|
self.probe_timestamps.borrow_mut().pop_front()
|
|
}
|
|
|
|
/// Consumes one pending injected interruption, if any, reporting whether
|
|
/// the caller should report `io::ErrorKind::Interrupted` instead of
|
|
/// attempting a real receive.
|
|
fn take_injected_interrupted(&self) -> bool {
|
|
let remaining = self.inject_recv_interrupted.get();
|
|
if remaining == 0 {
|
|
return false;
|
|
}
|
|
self.inject_recv_interrupted.set(remaining - 1);
|
|
true
|
|
}
|
|
}
|
|
|
|
/// Low-level Tokio IRTT client for one connected UDP target.
|
|
///
|
|
/// `AsyncClient` does not construct, own, or store a Tokio runtime. Its async
|
|
/// methods are polled by the caller and require a current Tokio runtime with I/O
|
|
/// and time enabled.
|
|
///
|
|
/// # Example
|
|
///
|
|
/// Use this low-level adapter when the application already owns Tokio and wants
|
|
/// to drive the session lifecycle itself. This example type-checks without
|
|
/// contacting a server.
|
|
///
|
|
/// ```no_run
|
|
/// use irtt_client::{AsyncClient, ClientConfig};
|
|
///
|
|
/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
|
|
/// let mut client = AsyncClient::connect(ClientConfig::default()).await?;
|
|
/// let outcome = client.open().await?;
|
|
/// println!("opened: {outcome:?}");
|
|
/// let sent = client.send_probe().await?;
|
|
/// println!("sent: {sent:?}");
|
|
/// let received = client.recv().await?;
|
|
/// println!("received: {received:?}");
|
|
/// let timed_out = client.poll_timeouts()?;
|
|
/// println!("timeouts: {timed_out:?}");
|
|
/// let closed = client.close().await?;
|
|
/// println!("closed: {closed:?}");
|
|
/// # Ok(())
|
|
/// # }
|
|
/// ```
|
|
#[derive(Debug)]
|
|
pub struct AsyncClient {
|
|
socket: tokio::net::UdpSocket,
|
|
machine: SessionMachine,
|
|
schedule: Option<ProbeSchedule>,
|
|
remote: SocketAddr,
|
|
recv_buffer: Vec<u8>,
|
|
applied_traffic_class: Option<u8>,
|
|
prepared_open: Option<PreparedOpenRequest>,
|
|
prepared_probe: Option<PreparedProbe>,
|
|
/// See `Client::tx_timestamping_enabled` in `client.rs` for the
|
|
/// invariant this tracks; it is the same adapter-owned capability flag.
|
|
tx_timestamping_enabled: bool,
|
|
#[cfg(test)]
|
|
test_hooks: AsyncClientTestHooks,
|
|
}
|
|
|
|
impl AsyncClient {
|
|
/// Resolve the configured server and construct one connected Tokio UDP
|
|
/// socket.
|
|
///
|
|
/// Polling this future without a current Tokio runtime returns
|
|
/// [`ClientError::NoTokioRuntime`]. A runtime without enabled I/O or time
|
|
/// drivers is outside this type's runtime contract.
|
|
pub async fn connect(config: ClientConfig) -> Result<Self, ClientError> {
|
|
tokio::runtime::Handle::try_current().map_err(|_| ClientError::NoTokioRuntime)?;
|
|
validate_open_timeouts(&config.open_timeouts)?;
|
|
let remote = resolve_remote_tokio(&config).await?;
|
|
let machine = SessionMachine::new(config.clone(), remote)?;
|
|
let prepared_open = machine.prepare_open_request()?;
|
|
let socket = connect_tokio_udp_socket(&config.socket_config, remote)?;
|
|
|
|
Ok(Self {
|
|
socket,
|
|
machine,
|
|
schedule: None,
|
|
remote,
|
|
recv_buffer: vec![0_u8; recv_buffer_size(false, None)?],
|
|
applied_traffic_class: None,
|
|
prepared_open: Some(prepared_open),
|
|
prepared_probe: None,
|
|
tx_timestamping_enabled: false,
|
|
#[cfg(test)]
|
|
test_hooks: AsyncClientTestHooks::default(),
|
|
})
|
|
}
|
|
|
|
/// Perform the IRTT open exchange.
|
|
///
|
|
/// Each configured attempt uses one absolute deadline for one successful
|
|
/// request submission and all replies inspected during that attempt.
|
|
/// Malformed, unrelated, wrong-direction, and unauthenticated datagrams are
|
|
/// ignored without resending the request.
|
|
///
|
|
/// Dropping this future before trusted acceptance leaves the local machine
|
|
/// connected with no committed schedule or negotiated adapter state. If it
|
|
/// is dropped during best-effort post-token cleanup, cleanup may remain
|
|
/// incomplete because this low-level client never detaches cleanup work.
|
|
pub async fn open(&mut self) -> Result<OpenOutcome, ClientError> {
|
|
let mut state = AsyncOpenState::new();
|
|
poll_fn(|cx| self.poll_open(&mut state, cx)).await
|
|
}
|
|
|
|
/// Send one caller-paced probe with transactional protocol and schedule
|
|
/// commits.
|
|
///
|
|
/// A prepared packet is retained across readiness false positives,
|
|
/// cancellation, and socket errors so a later call retries the same logical
|
|
/// probe without advancing state before kernel acceptance.
|
|
pub async fn send_probe(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
|
|
poll_fn(|cx| self.poll_send_probe(cx)).await
|
|
}
|
|
|
|
/// Await and classify one complete UDP datagram.
|
|
///
|
|
/// State is validated before readiness is awaited. Readiness false positives
|
|
/// retry without changing protocol state. Authenticated peer-close events
|
|
/// remain authoritative even if best-effort DSCP cleanup fails.
|
|
pub async fn recv(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
|
|
poll_fn(|cx| self.poll_recv(cx)).await
|
|
}
|
|
|
|
/// Poll protocol timeouts using a newly captured monotonic timestamp.
|
|
pub fn poll_timeouts(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
|
|
self.poll_timeouts_at(Instant::now())
|
|
}
|
|
|
|
/// Poll protocol timeouts using the caller's monotonic timestamp.
|
|
pub fn poll_timeouts_at(&mut self, now: Instant) -> Result<Vec<ClientEvent>, ClientError> {
|
|
self.drain_tx_timestamps()?;
|
|
self.machine.poll_timeouts_at(now)
|
|
}
|
|
|
|
pub(crate) fn poll_timeouts_bounded_at(
|
|
&mut self,
|
|
now: Instant,
|
|
limit: usize,
|
|
) -> Result<TimeoutBatch, ClientError> {
|
|
self.drain_tx_timestamps()?;
|
|
self.machine.poll_timeouts_bounded_at(now, limit)
|
|
}
|
|
|
|
/// Send the retained close packet and commit local close exactly once.
|
|
///
|
|
/// DSCP is cleared only after write readiness and immediately before the
|
|
/// nonblocking send attempt. A `WouldBlock` result restores it before the
|
|
/// next await, so dropping this future at a suspension point cannot leave
|
|
/// an otherwise-open session with cleared DSCP.
|
|
pub async fn close(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
|
|
poll_fn(|cx| self.poll_close(cx)).await
|
|
}
|
|
|
|
pub(crate) fn poll_open(
|
|
&mut self,
|
|
state: &mut AsyncOpenState,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Result<OpenOutcome, ClientError>> {
|
|
if self.prepared_open.is_none() {
|
|
if let Err(error) = self.machine.prepare_open_request() {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
unreachable!("connected async clients retain an open request");
|
|
}
|
|
|
|
for _ in 0..OPEN_POLL_WORK_BUDGET {
|
|
if state.cleanup.is_some() {
|
|
return self.poll_open_cleanup(state, cx);
|
|
}
|
|
|
|
let attempt_count = self.machine.config().open_timeouts.len();
|
|
if state.attempt >= attempt_count {
|
|
return Poll::Ready(Err(ClientError::OpenTimeout));
|
|
}
|
|
if state.stop_after_current_attempt && state.deadline.is_none() {
|
|
state.stopped_after_current_attempt = true;
|
|
return Poll::Ready(Err(ClientError::OpenTimeout));
|
|
}
|
|
if state.deadline.is_none() {
|
|
let timeout = self.machine.config().open_timeouts[state.attempt];
|
|
if let Err(error) = state.start_attempt(timeout) {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
}
|
|
|
|
if !state.request_submitted {
|
|
#[cfg(test)]
|
|
if self.test_hooks.pause_open_before_writable.replace(false) {
|
|
return Poll::Pending;
|
|
}
|
|
match poll_writable_until(&self.socket, state.deadline_timer(), cx) {
|
|
Poll::Pending => return Poll::Pending,
|
|
Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
|
|
Poll::Ready(Ok(false)) => {
|
|
state.finish_attempt();
|
|
continue;
|
|
}
|
|
Poll::Ready(Ok(true)) => {}
|
|
}
|
|
|
|
let request = self
|
|
.prepared_open
|
|
.as_ref()
|
|
.expect("connected clients retain their prepared open request");
|
|
#[cfg(test)]
|
|
let send_result = self.test_hooks.try_send(&self.socket, &request.bytes);
|
|
#[cfg(not(test))]
|
|
let send_result = self.socket.try_send(&request.bytes);
|
|
match send_result {
|
|
Ok(bytes) => {
|
|
state.request_submitted = true;
|
|
if let Err(error) = validate_datagram_length(request.bytes.len(), bytes) {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
}
|
|
Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
|
|
Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
if self.test_hooks.pause_open_before_readable.replace(false) {
|
|
return Poll::Pending;
|
|
}
|
|
match poll_readable_until(&self.socket, state.deadline_timer(), cx) {
|
|
Poll::Pending => return Poll::Pending,
|
|
Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
|
|
Poll::Ready(Ok(false)) => {
|
|
state.finish_attempt();
|
|
if state.stopped_after_current_attempt() {
|
|
return Poll::Ready(Err(ClientError::OpenTimeout));
|
|
}
|
|
continue;
|
|
}
|
|
Poll::Ready(Ok(true)) => {}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
if self.test_hooks.take_injected_interrupted() {
|
|
continue;
|
|
}
|
|
let datagram = match try_recv_tokio_datagram(&self.socket, &mut state.buffer) {
|
|
Ok(datagram) => datagram,
|
|
// WouldBlock is a readiness false positive; Interrupted
|
|
// (EINTR) is a transient nonblocking-recvmsg interruption.
|
|
// Both retry through the surrounding `OPEN_POLL_WORK_BUDGET`-
|
|
// bounded loop above, which re-checks readiness before the
|
|
// next attempt and yields back to the executor once that
|
|
// budget is exhausted.
|
|
Err(error)
|
|
if matches!(
|
|
error.kind(),
|
|
io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
|
|
) =>
|
|
{
|
|
continue
|
|
}
|
|
Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
|
|
};
|
|
if datagram.received_at.mono > state.deadline() {
|
|
state.finish_attempt();
|
|
if state.stopped_after_current_attempt() {
|
|
return Poll::Ready(Err(ClientError::OpenTimeout));
|
|
}
|
|
continue;
|
|
}
|
|
|
|
let reply = match self
|
|
.machine
|
|
.inspect_open_datagram(&state.buffer[..datagram.len])
|
|
{
|
|
Ok(OpenDatagramDisposition::Ignore) => continue,
|
|
Ok(OpenDatagramDisposition::Trusted(reply)) => reply,
|
|
Err(error) => return Poll::Ready(Err(error)),
|
|
};
|
|
let machine = match self
|
|
.machine
|
|
.prepare_open_acceptance(reply, datagram.received_at)
|
|
{
|
|
Ok(machine) => machine,
|
|
Err(failure) => {
|
|
state.cleanup = Some(AsyncOpenCleanup::Acceptance {
|
|
primary: failure.primary,
|
|
packet: failure.cleanup_close,
|
|
});
|
|
continue;
|
|
}
|
|
};
|
|
let prepared = match self.prepare_async_open(machine, datagram.received_at) {
|
|
Ok(prepared) => prepared,
|
|
Err(failure) => {
|
|
state.cleanup = Some(AsyncOpenCleanup::Adapter(failure));
|
|
continue;
|
|
}
|
|
};
|
|
return Poll::Ready(Ok(self.commit_async_open(prepared)));
|
|
}
|
|
|
|
// We deliberately truncated known immediate opening work. Arrange for
|
|
// another poll even if no new socket-readiness edge occurs.
|
|
cx.waker().wake_by_ref();
|
|
Poll::Pending
|
|
}
|
|
|
|
pub(crate) fn poll_send_probe(
|
|
&mut self,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Result<Vec<ClientEvent>, ClientError>> {
|
|
if let Err(error) = self.machine.ensure_open() {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
if self.prepared_probe.is_none() {
|
|
match self.machine.prepare_probe() {
|
|
Ok(prepared) => self.prepared_probe = prepared,
|
|
Err(error) => return Poll::Ready(Err(error)),
|
|
}
|
|
}
|
|
if self.prepared_probe.is_none() {
|
|
return Poll::Ready(Ok(Vec::new()));
|
|
}
|
|
|
|
loop {
|
|
#[cfg(test)]
|
|
if self.test_hooks.pause_probe_before_writable.replace(false) {
|
|
return Poll::Pending;
|
|
}
|
|
match self.socket.poll_send_ready(cx) {
|
|
Poll::Pending => return Poll::Pending,
|
|
Poll::Ready(Ok(())) => {}
|
|
Poll::Ready(Err(error)) => {
|
|
return Poll::Ready(Err(ClientError::Socket(error)));
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
let test_timestamps = self.test_hooks.take_probe_timestamps();
|
|
#[cfg(not(test))]
|
|
let permission_at = Instant::now();
|
|
#[cfg(test)]
|
|
let permission_at = test_timestamps
|
|
.map(|timestamps| timestamps.permission_at)
|
|
.unwrap_or_else(Instant::now);
|
|
|
|
let schedule = self
|
|
.schedule
|
|
.as_mut()
|
|
.expect("open sessions always have a probe schedule");
|
|
if !schedule.permit_probe_at(permission_at) {
|
|
self.prepared_probe = None;
|
|
return Poll::Ready(Ok(Vec::new()));
|
|
}
|
|
let prepared = self
|
|
.prepared_probe
|
|
.as_ref()
|
|
.expect("prepared probe was retained across readiness");
|
|
let machine_preflight = match self.machine.preflight_probe_commit(prepared) {
|
|
Ok(preflight) => preflight,
|
|
Err(error) => return Poll::Ready(Err(error)),
|
|
};
|
|
let schedule_commit =
|
|
match schedule.preflight_caller_commit(machine_preflight.next_packets_sent) {
|
|
Ok(commit) => commit,
|
|
Err(error) => return Poll::Ready(Err(error)),
|
|
};
|
|
let mut events = Vec::new();
|
|
if let Err(source) = events.try_reserve(1) {
|
|
return Poll::Ready(Err(ClientError::AllocationFailed {
|
|
operation: "probe event result",
|
|
source,
|
|
}));
|
|
}
|
|
|
|
let expected_bytes = prepared.bytes.len();
|
|
let scheduled_at = schedule_commit.scheduled_at;
|
|
// Private pre-send anchor: finalizes all fallible commit work
|
|
// (timeout deadline arithmetic, kernel TX lower bound) before
|
|
// the send attempt. A WouldBlock retry samples a fresh one on
|
|
// the next loop iteration; only the successful attempt's anchor
|
|
// is retained (via `machine_commit`/`commit_probe_sent` below).
|
|
#[cfg(not(test))]
|
|
let send_anchor = ClientTimestamp::now();
|
|
#[cfg(test)]
|
|
let send_anchor = test_timestamps
|
|
.map(|timestamps| timestamps.send_anchor)
|
|
.unwrap_or_else(ClientTimestamp::now);
|
|
let machine_commit = match self
|
|
.machine
|
|
.finalize_probe_commit(machine_preflight, send_anchor)
|
|
{
|
|
Ok(commit) => commit,
|
|
Err(error) => return Poll::Ready(Err(error)),
|
|
};
|
|
#[cfg(not(test))]
|
|
let send_call_start = Instant::now();
|
|
#[cfg(test)]
|
|
let send_call_start = test_timestamps
|
|
.map(|timestamps| timestamps.send_call_start)
|
|
.unwrap_or_else(Instant::now);
|
|
#[cfg(test)]
|
|
let send_result = self.test_hooks.try_send(&self.socket, &prepared.bytes);
|
|
#[cfg(not(test))]
|
|
let send_result = self.socket.try_send(&prepared.bytes);
|
|
let bytes = match send_result {
|
|
Ok(bytes) => bytes,
|
|
// No sent_at measurement, no machine commit, no schedule
|
|
// commit: the transaction is fully discarded and retried
|
|
// with a fresh send_anchor on the next loop iteration.
|
|
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
|
|
self.machine.invalidate_kernel_tx_correlation();
|
|
continue;
|
|
}
|
|
Err(error) => {
|
|
self.machine.invalidate_kernel_tx_correlation();
|
|
return Poll::Ready(Err(ClientError::Socket(error)));
|
|
}
|
|
};
|
|
|
|
#[cfg(not(test))]
|
|
let send_finished_at = Instant::now();
|
|
#[cfg(test)]
|
|
let send_finished_at = test_timestamps
|
|
.map(|timestamps| timestamps.send_finished_at)
|
|
.unwrap_or_else(Instant::now);
|
|
// Public post-send measurement, captured only after the
|
|
// successful send above. The subsequent machine/schedule commit
|
|
// is infallible.
|
|
#[cfg(not(test))]
|
|
let sent_at = ClientTimestamp::now();
|
|
#[cfg(test)]
|
|
let sent_at = test_timestamps
|
|
.map(|timestamps| timestamps.sent_at)
|
|
.unwrap_or_else(ClientTimestamp::now);
|
|
let sent = self
|
|
.machine
|
|
.commit_probe_sent(machine_commit, sent_at, bytes);
|
|
schedule.commit(schedule_commit);
|
|
self.prepared_probe = None;
|
|
|
|
let send_call = send_finished_at.saturating_duration_since(send_call_start);
|
|
let timer_error = instant_abs_diff(sent_at.mono, scheduled_at);
|
|
if let Err(error) = validate_datagram_length(expected_bytes, bytes) {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
if let Err(error) = self.drain_tx_timestamps() {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
events.push(echo_sent_event(
|
|
self.remote,
|
|
sent,
|
|
send_call,
|
|
scheduled_at,
|
|
timer_error,
|
|
));
|
|
return Poll::Ready(Ok(events));
|
|
}
|
|
}
|
|
|
|
pub(crate) fn poll_recv(
|
|
&mut self,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Result<Vec<ClientEvent>, ClientError>> {
|
|
if let Err(error) = self.machine.ensure_open() {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
|
|
for _ in 0..RECV_POLL_WORK_BUDGET {
|
|
match self.socket.poll_recv_ready(cx) {
|
|
Poll::Pending => return Poll::Pending,
|
|
Poll::Ready(Ok(())) => {}
|
|
Poll::Ready(Err(error)) => {
|
|
return Poll::Ready(Err(ClientError::Socket(error)));
|
|
}
|
|
}
|
|
#[cfg(test)]
|
|
if self.test_hooks.receive_would_block.get() > 0 {
|
|
self.test_hooks
|
|
.receive_would_block
|
|
.set(self.test_hooks.receive_would_block.get() - 1);
|
|
continue;
|
|
}
|
|
#[cfg(test)]
|
|
if self.test_hooks.take_injected_interrupted() {
|
|
continue;
|
|
}
|
|
let datagram = match try_recv_tokio_datagram(&self.socket, &mut self.recv_buffer) {
|
|
Ok(datagram) => datagram,
|
|
// See `RECV_POLL_WORK_BUDGET`: both a readiness false
|
|
// positive and an interrupted nonblocking recvmsg retry
|
|
// within this bounded loop rather than failing the receive.
|
|
Err(error)
|
|
if matches!(
|
|
error.kind(),
|
|
io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
|
|
) =>
|
|
{
|
|
continue
|
|
}
|
|
Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
|
|
};
|
|
|
|
// One final bounded drain before processing the reply, so a TX
|
|
// timestamp that raced the reply is still associated with its
|
|
// probe before that probe is looked up and possibly removed
|
|
// below.
|
|
if let Err(error) = self.drain_tx_timestamps() {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
|
|
let events = match self.machine.process_received_echo_packet(
|
|
&self.recv_buffer[..datagram.len],
|
|
datagram.received_at,
|
|
datagram.meta,
|
|
) {
|
|
Ok(events) => events,
|
|
Err(error) => return Poll::Ready(Err(error)),
|
|
};
|
|
if self.machine.is_peer_closed() {
|
|
self.schedule = None;
|
|
self.prepared_probe = None;
|
|
if self.clear_peer_close_dscp().is_ok() {
|
|
self.applied_traffic_class = None;
|
|
}
|
|
}
|
|
return Poll::Ready(Ok(events));
|
|
}
|
|
|
|
// Truncated known immediate receive work, exactly as `poll_open`
|
|
// does: arrange for another poll even if no new socket-readiness
|
|
// edge occurs, rather than spinning this executor thread further.
|
|
cx.waker().wake_by_ref();
|
|
Poll::Pending
|
|
}
|
|
|
|
pub(crate) fn poll_close(
|
|
&mut self,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Result<Vec<ClientEvent>, ClientError>> {
|
|
loop {
|
|
let prepared = match self.machine.prepare_close() {
|
|
Ok(prepared) => prepared,
|
|
Err(error) => return Poll::Ready(Err(error)),
|
|
};
|
|
let mut events = Vec::new();
|
|
if let Err(source) = events.try_reserve(1) {
|
|
return Poll::Ready(Err(ClientError::AllocationFailed {
|
|
operation: "close event result",
|
|
source,
|
|
}));
|
|
}
|
|
let previous_traffic_class = self.applied_traffic_class;
|
|
let expected_bytes = prepared.bytes.len();
|
|
|
|
#[cfg(test)]
|
|
if self.test_hooks.pause_close_before_writable.replace(false) {
|
|
return Poll::Pending;
|
|
}
|
|
match self.socket.poll_send_ready(cx) {
|
|
Poll::Pending => return Poll::Pending,
|
|
Poll::Ready(Ok(())) => {}
|
|
Poll::Ready(Err(error)) => {
|
|
return Poll::Ready(Err(ClientError::Socket(error)));
|
|
}
|
|
}
|
|
if let Err(error) = self.clear_close_dscp() {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
let mut rollback = DscpRollback::armed(
|
|
&self.socket,
|
|
self.remote,
|
|
previous_traffic_class,
|
|
#[cfg(test)]
|
|
&self.test_hooks.fail_dscp_restore,
|
|
);
|
|
#[cfg(test)]
|
|
let send_result = self.test_hooks.try_send(&self.socket, prepared.bytes);
|
|
#[cfg(not(test))]
|
|
let send_result = self.socket.try_send(prepared.bytes);
|
|
let bytes = match send_result {
|
|
Ok(bytes) => bytes,
|
|
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
|
|
if let Err(error) = rollback.restore() {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
#[cfg(test)]
|
|
if self.test_hooks.pause_close_after_would_block.replace(false) {
|
|
return Poll::Pending;
|
|
}
|
|
continue;
|
|
}
|
|
Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
|
|
};
|
|
|
|
#[cfg(not(test))]
|
|
let close_sent_at = ClientTimestamp::now();
|
|
#[cfg(test)]
|
|
let close_sent_at = self
|
|
.test_hooks
|
|
.close_sent_at
|
|
.take()
|
|
.unwrap_or_else(ClientTimestamp::now);
|
|
let event = self
|
|
.machine
|
|
.commit_local_close(prepared.commit, close_sent_at);
|
|
rollback.disarm();
|
|
self.schedule = None;
|
|
self.prepared_probe = None;
|
|
self.applied_traffic_class = None;
|
|
|
|
if let Err(error) = validate_datagram_length(expected_bytes, bytes) {
|
|
return Poll::Ready(Err(error));
|
|
}
|
|
events.push(event);
|
|
return Poll::Ready(Ok(events));
|
|
}
|
|
}
|
|
|
|
/// Return the next scheduled probe deadline.
|
|
pub fn next_send_deadline(&self) -> Option<Instant> {
|
|
if !self.machine.is_open() {
|
|
return None;
|
|
}
|
|
self.schedule.as_ref()?.next_send_deadline()
|
|
}
|
|
|
|
pub(crate) fn probe_interval(&self) -> Option<Duration> {
|
|
if !self.machine.is_open() {
|
|
return None;
|
|
}
|
|
self.schedule.as_ref().map(ProbeSchedule::interval)
|
|
}
|
|
|
|
/// Return the configured local probe timeout.
|
|
pub fn probe_timeout(&self) -> Duration {
|
|
self.machine.probe_timeout()
|
|
}
|
|
|
|
/// Return whether the current run has completed.
|
|
pub fn is_run_complete(&self) -> bool {
|
|
self.machine.is_terminal()
|
|
|| self
|
|
.schedule
|
|
.as_ref()
|
|
.is_some_and(|schedule| schedule.is_finished() && self.machine.pending_is_empty())
|
|
}
|
|
|
|
/// Return whether an authenticated peer close ended the session.
|
|
pub fn is_peer_closed(&self) -> bool {
|
|
self.machine.is_peer_closed()
|
|
}
|
|
|
|
pub(crate) fn remote_addr(&self) -> SocketAddr {
|
|
self.remote
|
|
}
|
|
|
|
pub(crate) fn packets_sent(&self) -> u64 {
|
|
self.machine.packets_sent()
|
|
}
|
|
|
|
pub(crate) fn next_probe_timeout_deadline(&self) -> Option<Instant> {
|
|
self.machine.next_probe_timeout_deadline()
|
|
}
|
|
|
|
pub(crate) fn latest_probe_timeout_deadline(&self) -> Option<Instant> {
|
|
self.machine.latest_probe_timeout_deadline()
|
|
}
|
|
|
|
pub(crate) fn discard_prepared_probe(&mut self) {
|
|
self.prepared_probe = None;
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn retain_open_cleanup_once(&self) {
|
|
self.test_hooks.fail_open_dscp.set(true);
|
|
self.test_hooks.pause_open_cleanup_before_send.set(true);
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn poll_recv_ready_for_test(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
|
self.socket.poll_recv_ready(cx)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn remove_pending_for_test(
|
|
&mut self,
|
|
wire_seq: u32,
|
|
) -> Option<crate::probe::PendingProbe> {
|
|
self.machine.remove_pending_for_test(wire_seq)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn replace_pending_for_test(&mut self, probe: crate::probe::PendingProbe) {
|
|
self.machine.replace_pending_for_test(probe);
|
|
}
|
|
|
|
pub(crate) fn skip_missed_probe_slots_at(&mut self, now: Instant) -> Result<(), ClientError> {
|
|
let Some(schedule) = self.schedule.as_mut() else {
|
|
return Ok(());
|
|
};
|
|
let Some(deadline) = schedule.next_send_deadline() else {
|
|
return Ok(());
|
|
};
|
|
if deadline > now {
|
|
return Ok(());
|
|
}
|
|
let commit = schedule.preflight_managed_commit(deadline, now)?;
|
|
schedule.commit(commit);
|
|
Ok(())
|
|
}
|
|
|
|
fn prepare_async_open(
|
|
&mut self,
|
|
machine: PreparedOpenAcceptance,
|
|
opened_at: ClientTimestamp,
|
|
) -> Result<PreparedAsyncOpen, Box<PreparedAsyncOpenFailure>> {
|
|
let Some(negotiated) = machine.normal_negotiated() else {
|
|
return Ok(PreparedAsyncOpen {
|
|
machine,
|
|
schedule: None,
|
|
recv_buffer_len: None,
|
|
negotiated_traffic_class: None,
|
|
});
|
|
};
|
|
let schedule = match ProbeSchedule::new(opened_at.mono, negotiated) {
|
|
Ok(schedule) => schedule,
|
|
Err(primary) => {
|
|
return Err(Box::new(PreparedAsyncOpenFailure { primary, machine }));
|
|
}
|
|
};
|
|
let recv_buffer_len = match recv_buffer_size(self.machine.has_hmac(), Some(negotiated)) {
|
|
Ok(size) => size,
|
|
Err(primary) => {
|
|
return Err(Box::new(PreparedAsyncOpenFailure { primary, machine }));
|
|
}
|
|
};
|
|
let negotiated_traffic_class = match u8::try_from(negotiated.params.dscp) {
|
|
Ok(traffic_class) => traffic_class,
|
|
Err(_) => {
|
|
return Err(Box::new(PreparedAsyncOpenFailure {
|
|
primary: ClientError::InvalidConfig {
|
|
reason: "negotiated dscp must be in range 0..=255".to_owned(),
|
|
},
|
|
machine,
|
|
}));
|
|
}
|
|
};
|
|
let additional = recv_buffer_len.saturating_sub(self.recv_buffer.len());
|
|
if let Err(source) = self.recv_buffer.try_reserve(additional) {
|
|
return Err(Box::new(PreparedAsyncOpenFailure {
|
|
primary: ClientError::AllocationFailed {
|
|
operation: "negotiated receive buffer",
|
|
source,
|
|
},
|
|
machine,
|
|
}));
|
|
}
|
|
#[cfg(test)]
|
|
let dscp_result = if self.test_hooks.fail_open_dscp.replace(false) {
|
|
Err(ClientError::SocketOption {
|
|
operation: "set negotiated DSCP",
|
|
remote: self.remote,
|
|
source: io::Error::other("injected negotiated DSCP failure"),
|
|
})
|
|
} else {
|
|
apply_traffic_class_to_tokio_socket(&self.socket, self.remote, negotiated_traffic_class)
|
|
};
|
|
#[cfg(not(test))]
|
|
let dscp_result = apply_traffic_class_to_tokio_socket(
|
|
&self.socket,
|
|
self.remote,
|
|
negotiated_traffic_class,
|
|
);
|
|
if let Err(primary) = dscp_result {
|
|
self.restore_dscp_best_effort(self.applied_traffic_class);
|
|
return Err(Box::new(PreparedAsyncOpenFailure { primary, machine }));
|
|
}
|
|
|
|
Ok(PreparedAsyncOpen {
|
|
machine,
|
|
schedule: Some(schedule),
|
|
recv_buffer_len: Some(recv_buffer_len),
|
|
negotiated_traffic_class: Some(negotiated_traffic_class),
|
|
})
|
|
}
|
|
|
|
fn commit_async_open(&mut self, prepared: PreparedAsyncOpen) -> OpenOutcome {
|
|
if let Some(recv_buffer_len) = prepared.recv_buffer_len {
|
|
self.recv_buffer.resize(recv_buffer_len, 0);
|
|
}
|
|
let outcome = self.machine.commit_open(prepared.machine);
|
|
self.schedule = prepared.schedule;
|
|
self.applied_traffic_class = prepared.negotiated_traffic_class;
|
|
self.prepared_open = None;
|
|
self.tx_timestamping_enabled = try_enable_tx_timestamping(&self.socket);
|
|
outcome
|
|
}
|
|
|
|
/// See `Client::drain_tx_timestamps` in `client.rs`: same bounded,
|
|
/// nonblocking, best-effort drain, adapted to the Tokio socket.
|
|
fn drain_tx_timestamps(&mut self) -> Result<(), ClientError> {
|
|
if !self.tx_timestamping_enabled {
|
|
return Ok(());
|
|
}
|
|
let machine = &mut self.machine;
|
|
drain_tx_timestamps(&self.socket, |id, timestamp| {
|
|
machine.record_kernel_tx_timestamp(id, timestamp);
|
|
})
|
|
.map_err(ClientError::Socket)
|
|
}
|
|
|
|
fn poll_open_cleanup(
|
|
&self,
|
|
state: &mut AsyncOpenState,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Result<OpenOutcome, ClientError>> {
|
|
let Some(packet) = state
|
|
.cleanup
|
|
.as_ref()
|
|
.expect("cleanup polling requires a retained primary error")
|
|
.packet()
|
|
else {
|
|
return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary()));
|
|
};
|
|
|
|
#[cfg(test)]
|
|
if self
|
|
.test_hooks
|
|
.pause_open_cleanup_before_send
|
|
.replace(false)
|
|
{
|
|
return Poll::Pending;
|
|
}
|
|
|
|
#[cfg(test)]
|
|
if self.test_hooks.fail_cleanup_send.replace(false) {
|
|
return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary()));
|
|
}
|
|
#[cfg(test)]
|
|
let send_result = self.test_hooks.try_send(&self.socket, packet);
|
|
#[cfg(not(test))]
|
|
let send_result = self.socket.try_send(packet);
|
|
match send_result {
|
|
Ok(_) => return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())),
|
|
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
|
|
Err(_) => return Poll::Ready(Err(state.cleanup.take().unwrap().into_primary())),
|
|
}
|
|
|
|
match poll_writable_until(&self.socket, state.deadline_timer(), cx) {
|
|
Poll::Pending => Poll::Pending,
|
|
Poll::Ready(Ok(true)) => {
|
|
// Retain cleanup state and resume the send in a fresh poll. A
|
|
// writable readiness false positive must not spin this future.
|
|
cx.waker().wake_by_ref();
|
|
Poll::Pending
|
|
}
|
|
Poll::Ready(Ok(false) | Err(_)) => {
|
|
Poll::Ready(Err(state.cleanup.take().unwrap().into_primary()))
|
|
}
|
|
}
|
|
}
|
|
|
|
fn clear_peer_close_dscp(&self) -> Result<(), ClientError> {
|
|
#[cfg(test)]
|
|
if self.test_hooks.fail_peer_close_dscp.replace(false) {
|
|
return Err(ClientError::SocketOption {
|
|
operation: "clear DSCP before close",
|
|
remote: self.remote,
|
|
source: io::Error::other("injected peer-close DSCP cleanup failure"),
|
|
});
|
|
}
|
|
clear_dscp_on_tokio_socket(&self.socket, self.remote)
|
|
}
|
|
|
|
fn clear_close_dscp(&self) -> Result<(), ClientError> {
|
|
clear_dscp_on_tokio_socket(&self.socket, self.remote)
|
|
}
|
|
|
|
fn restore_dscp_best_effort(&self, previous_traffic_class: Option<u8>) {
|
|
let _ = match previous_traffic_class {
|
|
Some(traffic_class) => {
|
|
apply_traffic_class_to_tokio_socket(&self.socket, self.remote, traffic_class)
|
|
}
|
|
None => clear_dscp_on_tokio_socket(&self.socket, self.remote),
|
|
};
|
|
}
|
|
}
|
|
|
|
struct DscpRollback<'a> {
|
|
socket: &'a tokio::net::UdpSocket,
|
|
remote: SocketAddr,
|
|
previous_traffic_class: Option<u8>,
|
|
armed: bool,
|
|
#[cfg(test)]
|
|
fail_restore: &'a Cell<bool>,
|
|
}
|
|
|
|
impl<'a> DscpRollback<'a> {
|
|
fn armed(
|
|
socket: &'a tokio::net::UdpSocket,
|
|
remote: SocketAddr,
|
|
previous_traffic_class: Option<u8>,
|
|
#[cfg(test)] fail_restore: &'a Cell<bool>,
|
|
) -> Self {
|
|
Self {
|
|
socket,
|
|
remote,
|
|
previous_traffic_class,
|
|
armed: true,
|
|
#[cfg(test)]
|
|
fail_restore,
|
|
}
|
|
}
|
|
|
|
fn restore(&mut self) -> Result<(), ClientError> {
|
|
if !self.armed {
|
|
return Ok(());
|
|
}
|
|
#[cfg(test)]
|
|
if self.fail_restore.replace(false) {
|
|
return Err(ClientError::SocketOption {
|
|
operation: "restore negotiated DSCP",
|
|
remote: self.remote,
|
|
source: io::Error::other("injected DSCP restore failure"),
|
|
});
|
|
}
|
|
match self.previous_traffic_class {
|
|
Some(traffic_class) => {
|
|
apply_traffic_class_to_tokio_socket(self.socket, self.remote, traffic_class)?
|
|
}
|
|
None => clear_dscp_on_tokio_socket(self.socket, self.remote)?,
|
|
}
|
|
self.armed = false;
|
|
Ok(())
|
|
}
|
|
|
|
fn disarm(&mut self) {
|
|
self.armed = false;
|
|
}
|
|
}
|
|
|
|
impl Drop for DscpRollback<'_> {
|
|
fn drop(&mut self) {
|
|
if self.armed {
|
|
let _ = self.restore();
|
|
}
|
|
}
|
|
}
|
|
|
|
fn poll_writable_until(
|
|
socket: &tokio::net::UdpSocket,
|
|
deadline: &mut Pin<Box<time::Sleep>>,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Result<bool, ClientError>> {
|
|
match socket.poll_send_ready(cx) {
|
|
Poll::Ready(Ok(())) => Poll::Ready(Ok(true)),
|
|
Poll::Ready(Err(error)) => Poll::Ready(Err(ClientError::Socket(error))),
|
|
Poll::Pending => match deadline.as_mut().poll(cx) {
|
|
Poll::Ready(()) => Poll::Ready(Ok(false)),
|
|
Poll::Pending => Poll::Pending,
|
|
},
|
|
}
|
|
}
|
|
|
|
fn poll_readable_until(
|
|
socket: &tokio::net::UdpSocket,
|
|
deadline: &mut Pin<Box<time::Sleep>>,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Result<bool, ClientError>> {
|
|
match socket.poll_recv_ready(cx) {
|
|
Poll::Ready(Ok(())) => Poll::Ready(Ok(true)),
|
|
Poll::Ready(Err(error)) => Poll::Ready(Err(ClientError::Socket(error))),
|
|
Poll::Pending => match deadline.as_mut().poll(cx) {
|
|
Poll::Ready(()) => Poll::Ready(Ok(false)),
|
|
Poll::Pending => Poll::Pending,
|
|
},
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests;
|