moq_net/
driver.rs

1//! Drive a session on the caller's executor.
2
3use std::task::Poll;
4
5use crate::Error;
6use crate::time::{Clock, Instant};
7
8/// Drives a session with caller-supplied time.
9///
10/// Returned by [`crate::Client::connect`] and [`crate::Server::accept`]. Call
11/// [`poll`](Self::poll) when external activity wakes the waiter or when the
12/// returned deadline is reached, supplying nondecreasing instants; see
13/// [`crate::time::Driver`] for the contract. Completion is cached, so
14/// subsequent polls return the same error.
15///
16/// It holds no session handle: dropping the last [`crate::Session`] requests
17/// closure on the next poll. Dropping the driver cancels the session, and
18/// [`crate::Session::closed`] resolves with [`Error::Cancel`]. Its `Send`-ness
19/// follows its transport.
20#[must_use = "the session makes no progress unless its driver is polled"]
21pub struct Driver<S: crate::transport::poll::Session> {
22	state: State<S>,
23	clock: Clock,
24}
25
26/// The protocol half of a machine, one variant per negotiated wire protocol.
27///
28/// The lite driver is a named machine, so the machine's `Send`-ness follows
29/// the transport (a pinned `!Send` transport yields a `!Send` machine that
30/// stays on its thread). The ietf driver is still a boxed future; the box
31/// demands `Send` on native, which is why the ietf path requires a
32/// [`Boxable`](crate::transport::poll::Boxable) transport until it too becomes
33/// a named machine.
34pub(crate) enum Protocol<S: crate::transport::poll::Session> {
35	/// Boxed for size only: a concrete box, so `Send` stays inferred.
36	Lite(Box<crate::lite::Driver<S>>),
37	Ietf(crate::util::MaybeSendBox<'static, Result<(), Error>>),
38}
39
40/// Protocol and lifecycle work owned by the driver.
41pub(crate) struct State<S: crate::transport::poll::Session> {
42	pub(crate) protocol: Protocol<S>,
43	// The session supervisor, polled alongside the protocol: it executes the
44	// handles' close requests, publishes the transport's terminal error, and
45	// samples stats. It finishes once the transport reports closed, and the
46	// machine is not done until it has: the protocol's terminal transport close
47	// is what `Session::closed` observes, so resolving before it is published
48	// would leave waiters parked on a machine nobody polls again. `None` once
49	// finished, since a completed machine must not be polled again.
50	pub(crate) supervisor: Option<crate::session::Supervisor<S>>,
51	// Cached so a poll after completion doesn't re-poll a finished protocol.
52	pub(crate) result: Option<Result<(), Error>>,
53}
54
55impl<S: crate::transport::poll::Session> Driver<S> {
56	pub(crate) fn new(clock: Clock, state: State<S>) -> Self {
57		Self { state, clock }
58	}
59
60	/// Process ready work at `now`, registering for external activity.
61	///
62	/// Panics if `now` is earlier than the previous poll or construction time.
63	pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
64		self.clock.advance(now);
65		match self.state.poll(waiter) {
66			Poll::Ready(Ok(())) => Err(Error::Closed),
67			Poll::Ready(Err(err)) => Err(err),
68			Poll::Pending => Ok(self.clock.timeout()),
69		}
70	}
71}
72
73impl<S: crate::transport::poll::Session> Protocol<S> {
74	fn poll(&mut self, waiter: &kio::Waiter) -> Poll<Result<(), Error>> {
75		match self {
76			Self::Lite(driver) => driver.poll(waiter),
77			Self::Ietf(driver) => waiter.poll_future(driver.as_mut()),
78		}
79	}
80}
81
82impl<S: crate::transport::poll::Session> State<S> {
83	fn poll(&mut self, waiter: &kio::Waiter) -> Poll<Result<(), Error>> {
84		if let Some(supervisor) = &mut self.supervisor
85			&& supervisor.poll(waiter).is_ready()
86		{
87			self.supervisor = None;
88		}
89
90		if self.result.is_none()
91			&& let Poll::Ready(result) = self.protocol.poll(waiter)
92		{
93			self.result = Some(result);
94			// The protocol's last act was closing the transport, which wakes the
95			// supervisor's close watch; poll it now instead of waiting a turn.
96			if let Some(supervisor) = &mut self.supervisor
97				&& supervisor.poll(waiter).is_ready()
98			{
99				self.supervisor = None;
100			}
101		}
102
103		match (&self.result, &self.supervisor) {
104			(Some(result), None) => Poll::Ready(result.clone()),
105			_ => Poll::Pending,
106		}
107	}
108}
109
110impl<S: crate::transport::poll::Session> crate::time::Driver for Driver<S> {
111	fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
112		self.poll(now, waiter)
113	}
114}
115
116impl<S: crate::transport::poll::Session> std::fmt::Debug for Driver<S> {
117	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
118		f.debug_struct("Driver")
119			.field("done", &self.state.result.is_some())
120			.finish()
121	}
122}
moq_net/driver.rs

moq_net/
driver.rs
