fix(server): keep unmarkable listeners serving
This commit is contained in:
3 files changed
+116
-26
No files matched your search
@@ -194,12 +194,26 @@ session policy lives.
|
|||||||
suspended. Do not introduce concurrent sends without replacing this with
|
suspended. Do not introduce concurrent sends without replacing this with
|
||||||
per-packet control messages.
|
per-packet control messages.
|
||||||
- Applying the class and sending are both per-packet transport steps: a failure
|
- Applying the class and sending are both per-packet transport steps: a failure
|
||||||
of either drops that one reply and the listener keeps serving. A class that
|
of either drops that one reply and the listener keeps serving. A marking that
|
||||||
could not be applied means the reply is not sent at all, because sending it
|
could not be applied means the reply is not sent at all, because sending it
|
||||||
would put it on the wire under the previous reply's marking.
|
would put it on the wire under the previous reply's marking.
|
||||||
|
- **One exception, and it is what keeps the server usable on hosts without the
|
||||||
|
option**: an *unmarked* reply may still be sent when this server has never
|
||||||
|
successfully applied a nonzero class to the socket, because then there is no
|
||||||
|
marking of ours for it to inherit. Some Windows builds do not support
|
||||||
|
`IP_TOS`, and a few targets have no safe setter at all; without this, such a
|
||||||
|
host would drop every open reply and no session could ever start. The flag
|
||||||
|
behind it is consulted only on failure — it never elides a call.
|
||||||
|
- The platform helper must report failure honestly, including for zero. "The
|
||||||
|
socket was cleared" and "nothing was written" are different statements, and a
|
||||||
|
socket passed to `Server::from_socket` may have been marked by its creator.
|
||||||
|
Whether an unappliable zero is nevertheless safe is the runtime's decision,
|
||||||
|
made from what this server itself applied.
|
||||||
- The option itself comes from `socket2`'s safe API, chosen by the listener's
|
- The option itself comes from `socket2`'s safe API, chosen by the listener's
|
||||||
bound address family rather than the peer's. The crate forbids unsafe code;
|
bound address family rather than the peer's. The crate forbids unsafe code;
|
||||||
there is no raw `setsockopt` and no `sendmsg`/`cmsg` machinery.
|
there is no raw `setsockopt` and no `sendmsg`/`cmsg` machinery. Keep the
|
||||||
|
target lists identical to the pinned `socket2` version's own gates — check
|
||||||
|
its source rather than copying another crate's older matrix.
|
||||||
|
|
||||||
## Rate, lifetime and server-initiated close
|
## Rate, lifetime and server-initiated close
|
||||||
|
|
||||||
|
|||||||
@@ -44,6 +44,14 @@ pub struct Server {
|
|||||||
/// queried at construction, in which case the send path asks again rather
|
/// queried at construction, in which case the send path asks again rather
|
||||||
/// than guessing a family.
|
/// than guessing a family.
|
||||||
listener_is_ipv4: Option<bool>,
|
listener_is_ipv4: Option<bool>,
|
||||||
|
/// Whether this server has ever successfully applied a nonzero traffic
|
||||||
|
/// class to the socket.
|
||||||
|
///
|
||||||
|
/// This is not a cache and never elides a call — the class is applied
|
||||||
|
/// before every send regardless. It is consulted only when applying one
|
||||||
|
/// *fails*, to tell "the socket may be carrying a marking of ours" from
|
||||||
|
/// "it cannot be carrying one".
|
||||||
|
socket_is_marked: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Server {
|
impl Server {
|
||||||
@@ -65,16 +73,36 @@ impl Server {
|
|||||||
core: ServerCore::new(config),
|
core: ServerCore::new(config),
|
||||||
recv_buffer: vec![0; RECEIVE_BUFFER_LEN],
|
recv_buffer: vec![0; RECEIVE_BUFFER_LEN],
|
||||||
listener_is_ipv4,
|
listener_is_ipv4,
|
||||||
|
socket_is_marked: false,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Applies one reply's raw traffic class to the listener socket.
|
/// Applies one reply's raw traffic class to the listener socket, and
|
||||||
|
/// reports whether that reply may be sent.
|
||||||
///
|
///
|
||||||
/// This runs before *every* send, and zero is applied as deliberately as
|
/// The class is applied before *every* send, and zero as deliberately as
|
||||||
/// any other value: a listener serves many sessions from one socket, so
|
/// any other value: a listener serves many sessions from one socket, so
|
||||||
/// skipping the call for an unmarked reply — an open reply, or a session
|
/// skipping the call for an unmarked reply — an open reply, or a session
|
||||||
/// that negotiated nothing — would send it under whichever marking the
|
/// that negotiated nothing — would send it under whichever marking the
|
||||||
/// previous reply left behind.
|
/// previous reply left behind.
|
||||||
|
///
|
||||||
|
/// When the option cannot be applied, [`may_send_unappliable`] decides. A
|
||||||
|
/// marking this server asked for and did not get must never be replaced by
|
||||||
|
/// silently sending under the previous one, so those replies are dropped;
|
||||||
|
/// but a host that refuses the option outright — some Windows builds do not
|
||||||
|
/// support `IP_TOS`, and a few targets have no safe setter at all — must
|
||||||
|
/// still be able to run a server, and it can, because a socket this server
|
||||||
|
/// has never marked has no marking of ours to clear.
|
||||||
|
fn prepare_reply_traffic_class(&mut self, traffic_class: u8) -> bool {
|
||||||
|
match self.apply_reply_traffic_class(traffic_class) {
|
||||||
|
Ok(()) => {
|
||||||
|
self.socket_is_marked = traffic_class != 0;
|
||||||
|
true
|
||||||
|
}
|
||||||
|
Err(_) => may_send_unappliable(traffic_class, self.socket_is_marked),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn apply_reply_traffic_class(&self, traffic_class: u8) -> io::Result<()> {
|
fn apply_reply_traffic_class(&self, traffic_class: u8) -> io::Result<()> {
|
||||||
let is_ipv4 = match self.listener_is_ipv4 {
|
let is_ipv4 = match self.listener_is_ipv4 {
|
||||||
Some(is_ipv4) => is_ipv4,
|
Some(is_ipv4) => is_ipv4,
|
||||||
@@ -95,13 +123,16 @@ impl Server {
|
|||||||
/// Graceful shutdown sends no session-close packets and leaves no hidden
|
/// Graceful shutdown sends no session-close packets and leaves no hidden
|
||||||
/// receive task behind, even while a reply send is waiting for socket
|
/// receive task behind, even while a reply send is waiting for socket
|
||||||
/// writability. Per-packet send failures and short sends drop only that
|
/// writability. Per-packet send failures and short sends drop only that
|
||||||
/// reply, and so does a failure to apply that reply's traffic class —
|
/// reply, and so does a failure to apply a marking that reply needed —
|
||||||
/// preparing one packet is not a reason to stop serving every other
|
/// preparing one packet is not a reason to stop serving every other
|
||||||
/// session, and the next reply makes its own independent attempt. Receive
|
/// session, and the next reply makes its own independent attempt. Receive
|
||||||
/// failures and internal core failures terminate the loop with an error.
|
/// failures and internal core failures terminate the loop with an error.
|
||||||
///
|
///
|
||||||
/// A reply whose class could not be applied is never sent anyway: sending it
|
/// A reply whose marking could not be applied is not sent instead: sending
|
||||||
/// would put it on the wire under the previous reply's marking.
|
/// it would put it on the wire under the previous reply's marking. An
|
||||||
|
/// *unmarked* reply on a socket this server has never marked is the one
|
||||||
|
/// exception, so a host that does not support the option at all still runs
|
||||||
|
/// a working server. See [`Server::prepare_reply_traffic_class`].
|
||||||
pub async fn run<F>(&mut self, shutdown: F) -> Result<(), ServerRuntimeError>
|
pub async fn run<F>(&mut self, shutdown: F) -> Result<(), ServerRuntimeError>
|
||||||
where
|
where
|
||||||
F: Future<Output = ()>,
|
F: Future<Output = ()>,
|
||||||
@@ -126,7 +157,7 @@ impl Server {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? {
|
if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? {
|
||||||
if self.apply_reply_traffic_class(reply.traffic_class()).is_err() {
|
if !self.prepare_reply_traffic_class(reply.traffic_class()) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
let send = self.socket.send_to(reply.bytes(), peer);
|
let send = self.socket.send_to(reply.bytes(), peer);
|
||||||
@@ -148,6 +179,21 @@ impl Server {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Whether a reply may still be sent after its traffic class could not be
|
||||||
|
/// applied.
|
||||||
|
///
|
||||||
|
/// Only an unmarked reply on a socket this server has never successfully marked
|
||||||
|
/// qualifies. Both halves matter: a reply that wanted a marking must not go out
|
||||||
|
/// without one, and once this server has put a nonzero class on the socket, a
|
||||||
|
/// failure to clear it means the socket may still be carrying it.
|
||||||
|
///
|
||||||
|
/// A socket handed to [`Server::from_socket`] pre-marked by its creator is
|
||||||
|
/// outside what this can know; explicit marking is the server's own to manage
|
||||||
|
/// from a fresh listener.
|
||||||
|
fn may_send_unappliable(traffic_class: u8, socket_is_marked: bool) -> bool {
|
||||||
|
traffic_class == 0 && !socket_is_marked
|
||||||
|
}
|
||||||
|
|
||||||
/// Failure to create or run a Tokio UDP server.
|
/// Failure to create or run a Tokio UDP server.
|
||||||
#[derive(Debug, Error)]
|
#[derive(Debug, Error)]
|
||||||
pub enum ServerRuntimeError {
|
pub enum ServerRuntimeError {
|
||||||
@@ -323,4 +369,33 @@ mod tests {
|
|||||||
fn unthrottled() -> ServerConfig {
|
fn unthrottled() -> ServerConfig {
|
||||||
ServerConfig::default().with_min_send_interval(Duration::ZERO)
|
ServerConfig::default().with_min_send_interval(Duration::ZERO)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The rule that decides what an unappliable traffic class means, which no
|
||||||
|
/// normal interface can reach: it needs a host that refuses the socket
|
||||||
|
/// option, and the hosts these tests run on do not.
|
||||||
|
#[test]
|
||||||
|
fn only_an_unmarked_reply_on_a_never_marked_socket_survives_an_apply_failure() {
|
||||||
|
for (traffic_class, socket_is_marked, sendable, why) in [
|
||||||
|
(
|
||||||
|
0,
|
||||||
|
false,
|
||||||
|
true,
|
||||||
|
"no marking wanted, and none of ours to clear",
|
||||||
|
),
|
||||||
|
(0, true, false, "our own marking may still be on the socket"),
|
||||||
|
(
|
||||||
|
0xb8,
|
||||||
|
false,
|
||||||
|
false,
|
||||||
|
"a marking that was wanted but not applied",
|
||||||
|
),
|
||||||
|
(0xb8, true, false, "and likewise over a marking already set"),
|
||||||
|
] {
|
||||||
|
assert_eq!(
|
||||||
|
may_send_unappliable(traffic_class, socket_is_marked),
|
||||||
|
sendable,
|
||||||
|
"{why}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
@@ -34,8 +34,8 @@ pub(crate) fn set_reply_traffic_class(
|
|||||||
target_os = "fuchsia",
|
target_os = "fuchsia",
|
||||||
target_os = "redox",
|
target_os = "redox",
|
||||||
target_os = "solaris",
|
target_os = "solaris",
|
||||||
target_os = "illumos",
|
|
||||||
target_os = "haiku",
|
target_os = "haiku",
|
||||||
|
target_os = "wasi",
|
||||||
)))]
|
)))]
|
||||||
fn set_ipv4_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
|
fn set_ipv4_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
|
||||||
socket.set_tos_v4(traffic_class)
|
socket.set_tos_v4(traffic_class)
|
||||||
@@ -45,8 +45,8 @@ fn set_ipv4_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result
|
|||||||
target_os = "fuchsia",
|
target_os = "fuchsia",
|
||||||
target_os = "redox",
|
target_os = "redox",
|
||||||
target_os = "solaris",
|
target_os = "solaris",
|
||||||
target_os = "illumos",
|
|
||||||
target_os = "haiku",
|
target_os = "haiku",
|
||||||
|
target_os = "wasi",
|
||||||
))]
|
))]
|
||||||
fn set_ipv4_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
|
fn set_ipv4_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
|
||||||
unsupported_traffic_class(traffic_class, "IPv4 TOS socket options")
|
unsupported_traffic_class(traffic_class, "IPv4 TOS socket options")
|
||||||
@@ -62,6 +62,7 @@ fn set_ipv4_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Resul
|
|||||||
target_os = "netbsd",
|
target_os = "netbsd",
|
||||||
target_os = "openbsd",
|
target_os = "openbsd",
|
||||||
target_os = "cygwin",
|
target_os = "cygwin",
|
||||||
|
target_os = "illumos",
|
||||||
))]
|
))]
|
||||||
fn set_ipv6_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
|
fn set_ipv6_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
|
||||||
socket.set_tclass_v6(traffic_class)
|
socket.set_tclass_v6(traffic_class)
|
||||||
@@ -77,6 +78,7 @@ fn set_ipv6_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result
|
|||||||
target_os = "netbsd",
|
target_os = "netbsd",
|
||||||
target_os = "openbsd",
|
target_os = "openbsd",
|
||||||
target_os = "cygwin",
|
target_os = "cygwin",
|
||||||
|
target_os = "illumos",
|
||||||
)))]
|
)))]
|
||||||
fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
|
fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
|
||||||
unsupported_traffic_class(traffic_class, "IPv6 Traffic Class socket options")
|
unsupported_traffic_class(traffic_class, "IPv6 Traffic Class socket options")
|
||||||
@@ -84,18 +86,20 @@ fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Resul
|
|||||||
|
|
||||||
/// A target where the safe API cannot express this option.
|
/// A target where the safe API cannot express this option.
|
||||||
///
|
///
|
||||||
/// Requesting *unmarked* still succeeds, because a platform that can never
|
/// This reports [`io::ErrorKind::Unsupported`] for *every* value, zero
|
||||||
/// apply a nonzero class also never left one on the socket for the next reply to
|
/// included. Returning success for zero would claim the socket had been cleared
|
||||||
/// inherit — the invariant the explicit zero exists to protect. Asking for a
|
/// when nothing was written to it, which is not the same statement — a caller
|
||||||
/// real marking reports [`io::ErrorKind::Unsupported`] rather than pretending:
|
/// handing [`Server::from_socket`](crate::Server::from_socket) an already
|
||||||
/// the runtime drops that one reply and stays alive, and no reply ever goes out
|
/// prepared socket may have marked it by some other means. Whether an
|
||||||
/// carrying a class the server did not intend.
|
/// unappliable zero is nevertheless safe to send is the runtime's decision, and
|
||||||
|
/// it is made from what this server itself has applied; this helper's job is
|
||||||
|
/// only to report honestly.
|
||||||
#[cfg(any(
|
#[cfg(any(
|
||||||
target_os = "fuchsia",
|
target_os = "fuchsia",
|
||||||
target_os = "redox",
|
target_os = "redox",
|
||||||
target_os = "solaris",
|
target_os = "solaris",
|
||||||
target_os = "illumos",
|
|
||||||
target_os = "haiku",
|
target_os = "haiku",
|
||||||
|
target_os = "wasi",
|
||||||
not(any(
|
not(any(
|
||||||
target_os = "android",
|
target_os = "android",
|
||||||
target_os = "dragonfly",
|
target_os = "dragonfly",
|
||||||
@@ -106,15 +110,12 @@ fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Resul
|
|||||||
target_os = "netbsd",
|
target_os = "netbsd",
|
||||||
target_os = "openbsd",
|
target_os = "openbsd",
|
||||||
target_os = "cygwin",
|
target_os = "cygwin",
|
||||||
|
target_os = "illumos",
|
||||||
)),
|
)),
|
||||||
))]
|
))]
|
||||||
fn unsupported_traffic_class(traffic_class: u32, option: &'static str) -> io::Result<()> {
|
fn unsupported_traffic_class(_traffic_class: u32, option: &'static str) -> io::Result<()> {
|
||||||
if traffic_class == 0 {
|
Err(io::Error::new(
|
||||||
Ok(())
|
io::ErrorKind::Unsupported,
|
||||||
} else {
|
format!("{option} are unsupported on this target"),
|
||||||
Err(io::Error::new(
|
))
|
||||||
io::ErrorKind::Unsupported,
|
|
||||||
format!("{option} are unsupported on this target"),
|
|
||||||
))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user