diff --git a/pr/426/docs/help.html b/pr/426/docs/help.html index 09d0ca8e7..cda0c7f93 100644 --- a/pr/426/docs/help.html +++ b/pr/426/docs/help.html @@ -1 +1 @@ -Help

All

Rustdoc help

Back
\ No newline at end of file +Help

All

Rustdoc help

Back
\ No newline at end of file diff --git a/pr/426/docs/iroh_quinn/struct.Accept.html b/pr/426/docs/iroh_quinn/struct.Accept.html index 30f43129c..05eef9a44 100644 --- a/pr/426/docs/iroh_quinn/struct.Accept.html +++ b/pr/426/docs/iroh_quinn/struct.Accept.html @@ -1,6 +1,6 @@ -Accept in iroh_quinn - Rust

Accept

iroh_quinn

Struct Accept 

Source 
pub struct Accept<'a> { /* private fields */ }
Expand description

Future produced by Endpoint::accept

-

Trait Implementations§

Source§

impl Future for Accept<'_>

Source§

type Output = Option<Incoming>

The type of value produced on completion.
Source§

fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output>

Attempts to resolve the future to a final value, registering -the current task for wakeup if the value is not yet available. Read more
Source§

impl<'__pin, 'a> Unpin for Accept<'a>
where +Accept in iroh_quinn - Rust

Accept

iroh_quinn

Struct Accept 

Source 
pub struct Accept<'a> { /* private fields */ }
Expand description

Future produced by Endpoint::accept

+

Trait Implementations§

Source§

impl Future for Accept<'_>

Source§

type Output = Option<Incoming>

The type of value produced on completion.
Source§

fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output>

Attempts to resolve the future to a final value, registering +the current task for wakeup if the value is not yet available. Read more
Source§

impl<'__pin, 'a> Unpin for Accept<'a>
where PinnedFieldsOf<__Origin<'__pin, 'a>>: Unpin,

Auto Trait Implementations§

§

impl<'a> !Freeze for Accept<'a>

§

impl<'a> !RefUnwindSafe for Accept<'a>

§

impl<'a> Send for Accept<'a>

§

impl<'a> Sync for Accept<'a>

§

impl<'a> !UnwindSafe for Accept<'a>

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where diff --git a/pr/426/docs/settings.html b/pr/426/docs/settings.html index fc24471ee..1adf3f67d 100644 --- a/pr/426/docs/settings.html +++ b/pr/426/docs/settings.html @@ -1 +1 @@ -Settings

All

Rustdoc settings

Back
\ No newline at end of file +Settings

All

Rustdoc settings

Back
\ No newline at end of file diff --git a/pr/426/docs/src/iroh_quinn/endpoint.rs.html b/pr/426/docs/src/iroh_quinn/endpoint.rs.html index b5fb2703b..49f6d5f02 100644 --- a/pr/426/docs/src/iroh_quinn/endpoint.rs.html +++ b/pr/426/docs/src/iroh_quinn/endpoint.rs.html @@ -37,7 +37,7 @@ 37))] 38use socket2::{Domain, Protocol, Socket, Type}; 39use tokio::sync::{Notify, futures::Notified, mpsc}; -40use tracing::{Instrument, Span}; +40use tracing::{Instrument, Span, trace}; 41use udp::{BATCH_SIZE, RecvMeta}; 42 43use crate::{ @@ -416,548 +416,549 @@ 416 self.0.shared.incoming.notify_waiters(); 417 } 418 -419 // Stop the driver once: -420 // * all `Endpoint` structs are dropped and all connections are drained -421 // * the endpoint was closed via `Endpoint::close` and all connections are drained +419 // Stop the driver if either: +420 // - all `Endpoint` structs are dropped and all connections are drained, +421 // - or `Endpoint::close` has been called and all connections are drained. 422 if endpoint.recv_state.connections.is_empty() 423 && (endpoint.ref_count == 0 || endpoint.recv_state.connections.close.is_some()) 424 { -425 Poll::Ready(Ok(())) -426 } else { -427 drop(endpoint); -428 // If there is more work to do schedule the endpoint task again. -429 // `wake_by_ref()` is called outside the lock to minimize -430 // lock contention on a multithreaded runtime. -431 if keep_going { -432 cx.waker().wake_by_ref(); -433 } -434 Poll::Pending -435 } -436 } -437} -438 -439impl Drop for EndpointDriver { -440 fn drop(&mut self) { -441 let mut endpoint = self.0.state.lock().unwrap(); -442 endpoint.driver_lost = true; -443 self.0.shared.incoming.notify_waiters(); -444 // Drop all outgoing channels, signaling the termination of the endpoint to the associated -445 // connections. -446 endpoint.recv_state.connections.senders.clear(); -447 } -448} -449 -450#[derive(Debug)] -451pub(crate) struct EndpointInner { -452 pub(crate) state: Mutex<State>, -453 pub(crate) shared: Shared, -454} -455 -456impl EndpointInner { -457 pub(crate) fn accept( -458 &self, -459 incoming: proto::Incoming, -460 server_config: Option<Arc<ServerConfig>>, -461 ) -> Result<Connecting, ConnectionError> { -462 let mut state = self.state.lock().unwrap(); -463 let mut response_buffer = Vec::new(); -464 let now = state.runtime.now(); -465 match state -466 .inner -467 .accept(incoming, now, &mut response_buffer, server_config) -468 { -469 Ok((handle, conn)) => { -470 state.stats.accepted_handshakes += 1; -471 let sender = state.socket.create_sender(); -472 let runtime = state.runtime.clone(); -473 Ok(state -474 .recv_state -475 .connections -476 .insert(handle, conn, sender, runtime)) -477 } -478 Err(error) => { -479 if let Some(transmit) = error.response { -480 respond(transmit, &response_buffer, &mut state.sender); -481 } -482 Err(error.cause) -483 } -484 } -485 } -486 -487 pub(crate) fn refuse(&self, incoming: proto::Incoming) { -488 let mut state = self.state.lock().unwrap(); -489 state.stats.refused_handshakes += 1; -490 let mut response_buffer = Vec::new(); -491 let transmit = state.inner.refuse(incoming, &mut response_buffer); -492 respond(transmit, &response_buffer, &mut state.sender); -493 } -494 -495 pub(crate) fn retry(&self, incoming: proto::Incoming) -> Result<(), proto::RetryError> { -496 let mut state = self.state.lock().unwrap(); -497 let mut response_buffer = Vec::new(); -498 let transmit = state.inner.retry(incoming, &mut response_buffer)?; -499 respond(transmit, &response_buffer, &mut state.sender); -500 Ok(()) -501 } -502 -503 pub(crate) fn ignore(&self, incoming: proto::Incoming) { -504 let mut state = self.state.lock().unwrap(); -505 state.stats.ignored_handshakes += 1; -506 state.inner.ignore(incoming); -507 } -508} -509 -510#[derive(Debug)] -511pub(crate) struct State { -512 socket: Box<dyn AsyncUdpSocket>, -513 sender: Pin<Box<dyn UdpSender>>, -514 /// During an active migration, abandoned_socket receives traffic -515 /// until the first packet arrives on the new socket. -516 prev_socket: Option<Box<dyn AsyncUdpSocket>>, -517 inner: proto::Endpoint, -518 recv_state: RecvState, -519 driver: Option<Waker>, -520 ipv6: bool, -521 events: mpsc::UnboundedReceiver<(ConnectionHandle, EndpointEvent)>, -522 /// Number of live handles that can be used to initiate or handle I/O; excludes the driver -523 ref_count: usize, -524 driver_lost: bool, -525 runtime: Arc<dyn Runtime>, -526 stats: EndpointStats, -527 default_client_config: Option<ClientConfig>, -528} -529 -530#[derive(Debug)] -531pub(crate) struct Shared { -532 incoming: Notify, -533 idle: Notify, -534} -535 -536impl State { -537 fn drive_recv(&mut self, cx: &mut Context<'_>, now: Instant) -> Result<bool, io::Error> { -538 let get_time = || self.runtime.now(); -539 self.recv_state.recv_limiter.start_cycle(get_time); -540 if let Some(socket) = &mut self.prev_socket { -541 // We don't care about the `PollProgress` from old sockets. -542 let poll_res = self.recv_state.poll_socket( -543 cx, -544 &mut self.inner, -545 &mut **socket, -546 &mut self.sender, -547 &*self.runtime, -548 now, -549 ); -550 if poll_res.is_err() { -551 self.prev_socket = None; -552 } -553 }; -554 let poll_res = self.recv_state.poll_socket( -555 cx, -556 &mut self.inner, -557 &mut *self.socket, -558 &mut self.sender, -559 &*self.runtime, -560 now, -561 ); -562 self.recv_state.recv_limiter.finish_cycle(get_time); -563 let poll_res = poll_res?; -564 if poll_res.received_connection_packet { -565 // Traffic has arrived on self.socket, therefore there is no need for the abandoned -566 // one anymore. TODO: Account for multiple outgoing connections. -567 self.prev_socket = None; -568 } -569 Ok(poll_res.keep_going) -570 } -571 -572 fn handle_events(&mut self, cx: &mut Context<'_>, shared: &Shared) -> bool { -573 for _ in 0..IO_LOOP_BOUND { -574 let (ch, event) = match self.events.poll_recv(cx) { -575 Poll::Ready(Some(x)) => x, -576 Poll::Ready(None) => unreachable!("EndpointInner owns one sender"), -577 Poll::Pending => { -578 return false; -579 } -580 }; -581 -582 if event.is_drained() { -583 self.recv_state.connections.senders.remove(&ch); -584 if self.recv_state.connections.is_empty() { -585 shared.idle.notify_waiters(); -586 } -587 } -588 let Some(event) = self.inner.handle_event(ch, event) else { -589 continue; -590 }; -591 // Ignoring errors from dropped connections that haven't yet been cleaned up -592 let _ = self -593 .recv_state -594 .connections -595 .senders -596 .get_mut(&ch) -597 .unwrap() -598 .send(ConnectionEvent::Proto(event)); -599 } -600 -601 true -602 } -603} -604 -605impl Drop for State { -606 fn drop(&mut self) { -607 for incoming in self.recv_state.incoming.drain(..) { -608 self.inner.ignore(incoming); -609 } -610 } -611} -612 -613fn respond( -614 transmit: proto::Transmit, -615 response_buffer: &[u8], -616 sender: &mut Pin<Box<dyn UdpSender>>, -617) { -618 // Send if there's kernel buffer space; otherwise, drop it -619 // -620 // As an endpoint-generated packet, we know this is an -621 // immediate, stateless response to an unconnected peer, -622 // one of: -623 // -624 // - A version negotiation response due to an unknown version -625 // - A `CLOSE` due to a malformed or unwanted connection attempt -626 // - A stateless reset due to an unrecognized connection -627 // - A `Retry` packet due to a connection attempt when -628 // `use_retry` is set -629 // -630 // In each case, a well-behaved peer can be trusted to retry a -631 // few times, which is guaranteed to produce the same response -632 // from us. Repeated failures might at worst cause a peer's new -633 // connection attempt to time out, which is acceptable if we're -634 // under such heavy load that there's never room for this code -635 // to transmit. This is morally equivalent to the packet getting -636 // lost due to congestion further along the link, which -637 // similarly relies on peer retries for recovery. -638 -639 // Copied from rust 1.85's std::task::Waker::noop() implementation for backwards compatibility -640 const NOOP: RawWaker = { -641 const VTABLE: RawWakerVTable = RawWakerVTable::new( -642 // Cloning just returns a new no-op raw waker -643 |_| NOOP, -644 // `wake` does nothing -645 |_| {}, -646 // `wake_by_ref` does nothing -647 |_| {}, -648 // Dropping does nothing as we don't allocate anything -649 |_| {}, -650 ); -651 RawWaker::new(std::ptr::null(), &VTABLE) -652 }; -653 // SAFETY: Copied from rust stdlib, the NOOP waker is thread-safe and doesn't violate the RawWakerVTable contract, -654 // it doesn't access the data pointer at all. -655 let waker = unsafe { Waker::from_raw(NOOP) }; -656 let mut cx = Context::from_waker(&waker); -657 _ = sender.as_mut().poll_send( -658 &udp_transmit(&transmit, &response_buffer[..transmit.size]), -659 &mut cx, -660 ); -661} -662 -663#[inline] -664fn proto_ecn(ecn: udp::EcnCodepoint) -> proto::EcnCodepoint { -665 match ecn { -666 udp::EcnCodepoint::Ect0 => proto::EcnCodepoint::Ect0, -667 udp::EcnCodepoint::Ect1 => proto::EcnCodepoint::Ect1, -668 udp::EcnCodepoint::Ce => proto::EcnCodepoint::Ce, -669 } -670} -671 -672#[derive(Debug)] -673struct ConnectionSet { -674 /// Senders for communicating with the endpoint's connections -675 senders: FxHashMap<ConnectionHandle, mpsc::UnboundedSender<ConnectionEvent>>, -676 /// Stored to give out clones to new ConnectionInners -677 sender: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>, -678 /// Set if the endpoint has been manually closed -679 close: Option<(VarInt, Bytes)>, -680} -681 -682impl ConnectionSet { -683 fn insert( -684 &mut self, -685 handle: ConnectionHandle, -686 conn: proto::Connection, -687 sender: Pin<Box<dyn UdpSender>>, -688 runtime: Arc<dyn Runtime>, -689 ) -> Connecting { -690 let (send, recv) = mpsc::unbounded_channel(); -691 if let Some((error_code, ref reason)) = self.close { -692 send.send(ConnectionEvent::Close { -693 error_code, -694 reason: reason.clone(), -695 }) -696 .unwrap(); -697 } -698 self.senders.insert(handle, send); -699 Connecting::new(handle, conn, self.sender.clone(), recv, sender, runtime) -700 } -701 -702 fn is_empty(&self) -> bool { -703 self.senders.is_empty() -704 } -705} -706 -707pub(crate) fn ensure_ipv6(x: SocketAddr) -> SocketAddrV6 { -708 match x { -709 SocketAddr::V6(x) => x, -710 SocketAddr::V4(x) => SocketAddrV6::new(x.ip().to_ipv6_mapped(), x.port(), 0, 0), -711 } -712} -713 -714pin_project! { -715 /// Future produced by [`Endpoint::accept`] -716 pub struct Accept<'a> { -717 endpoint: &'a Endpoint, -718 #[pin] -719 notify: Notified<'a>, -720 } -721} -722 -723impl Future for Accept<'_> { -724 type Output = Option<Incoming>; -725 fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> { -726 let mut this = self.project(); -727 let mut endpoint = this.endpoint.inner.state.lock().unwrap(); -728 if endpoint.driver_lost { -729 return Poll::Ready(None); -730 } -731 if let Some(incoming) = endpoint.recv_state.incoming.pop_front() { -732 // Release the mutex lock on endpoint so cloning it doesn't deadlock -733 drop(endpoint); -734 let incoming = Incoming::new(incoming, this.endpoint.inner.clone()); -735 return Poll::Ready(Some(incoming)); -736 } -737 if endpoint.recv_state.connections.close.is_some() { -738 return Poll::Ready(None); -739 } -740 loop { -741 match this.notify.as_mut().poll(ctx) { -742 // `state` lock ensures we didn't race with readiness -743 Poll::Pending => return Poll::Pending, -744 // Spurious wakeup, get a new future -745 Poll::Ready(()) => this -746 .notify -747 .set(this.endpoint.inner.shared.incoming.notified()), -748 } -749 } -750 } -751} -752 -753#[derive(Debug)] -754pub(crate) struct EndpointRef(Arc<EndpointInner>); -755 -756impl EndpointRef { -757 pub(crate) fn new( -758 socket: Box<dyn AsyncUdpSocket>, -759 inner: proto::Endpoint, -760 ipv6: bool, -761 runtime: Arc<dyn Runtime>, -762 ) -> Self { -763 let (sender, events) = mpsc::unbounded_channel(); -764 let recv_state = RecvState::new(sender, socket.max_receive_segments(), &inner); -765 let sender = socket.create_sender(); -766 Self(Arc::new(EndpointInner { -767 shared: Shared { -768 incoming: Notify::new(), -769 idle: Notify::new(), -770 }, -771 state: Mutex::new(State { -772 socket, -773 sender, -774 prev_socket: None, -775 inner, -776 ipv6, -777 events, -778 driver: None, -779 ref_count: 0, -780 driver_lost: false, -781 recv_state, -782 runtime, -783 stats: EndpointStats::default(), -784 default_client_config: None, -785 }), -786 })) -787 } -788} -789 -790impl Clone for EndpointRef { -791 fn clone(&self) -> Self { -792 self.0.state.lock().unwrap().ref_count += 1; -793 Self(self.0.clone()) -794 } -795} -796 -797impl Drop for EndpointRef { -798 fn drop(&mut self) { -799 let endpoint = &mut *self.0.state.lock().unwrap(); -800 if let Some(x) = endpoint.ref_count.checked_sub(1) { -801 endpoint.ref_count = x; -802 if x == 0 { -803 // If the driver is about to be on its own, ensure it can shut down if the last -804 // connection is gone. -805 if let Some(task) = endpoint.driver.take() { -806 task.wake(); -807 } -808 } -809 } -810 } -811} -812 -813impl std::ops::Deref for EndpointRef { -814 type Target = EndpointInner; -815 fn deref(&self) -> &Self::Target { -816 &self.0 -817 } -818} -819 -820/// State directly involved in handling incoming packets -821struct RecvState { -822 incoming: VecDeque<proto::Incoming>, -823 connections: ConnectionSet, -824 recv_buf: Box<[u8]>, -825 recv_limiter: WorkLimiter, -826} -827 -828impl RecvState { -829 fn new( -830 sender: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>, -831 max_receive_segments: NonZeroUsize, -832 endpoint: &proto::Endpoint, -833 ) -> Self { -834 let recv_buf = vec![ -835 0; -836 endpoint.config().get_max_udp_payload_size().min(64 * 1024) as usize -837 * max_receive_segments.get() -838 * BATCH_SIZE -839 ]; -840 Self { -841 connections: ConnectionSet { -842 senders: FxHashMap::default(), -843 sender, -844 close: None, -845 }, -846 incoming: VecDeque::new(), -847 recv_buf: recv_buf.into(), -848 recv_limiter: WorkLimiter::new(RECV_TIME_BOUND), -849 } -850 } -851 -852 fn poll_socket( -853 &mut self, -854 cx: &mut Context<'_>, -855 endpoint: &mut proto::Endpoint, -856 socket: &mut dyn AsyncUdpSocket, -857 sender: &mut Pin<Box<dyn UdpSender>>, -858 runtime: &dyn Runtime, -859 now: Instant, -860 ) -> Result<PollProgress, io::Error> { -861 let mut received_connection_packet = false; -862 let mut metas = [RecvMeta::default(); BATCH_SIZE]; -863 let mut iovs: [IoSliceMut<'_>; BATCH_SIZE] = { -864 let mut bufs = self -865 .recv_buf -866 .chunks_mut(self.recv_buf.len() / BATCH_SIZE) -867 .map(IoSliceMut::new); -868 -869 // expect() safe as self.recv_buf is chunked into BATCH_SIZE items -870 // and iovs will be of size BATCH_SIZE, thus from_fn is called -871 // exactly BATCH_SIZE times. -872 std::array::from_fn(|_| bufs.next().expect("BATCH_SIZE elements")) -873 }; -874 loop { -875 match socket.poll_recv(cx, &mut iovs, &mut metas) { -876 Poll::Ready(Ok(msgs)) => { -877 self.recv_limiter.record_work(msgs); -878 for (meta, buf) in metas.iter().zip(iovs.iter()).take(msgs) { -879 let mut data: BytesMut = buf[0..meta.len].into(); -880 while !data.is_empty() { -881 let buf = data.split_to(meta.stride.min(data.len())); -882 let mut response_buffer = Vec::new(); -883 let addresses = FourTuple { -884 remote: meta.addr, -885 local_ip: meta.dst_ip, -886 }; -887 match endpoint.handle( -888 now, -889 addresses, -890 meta.ecn.map(proto_ecn), -891 buf, -892 &mut response_buffer, -893 ) { -894 Some(DatagramEvent::NewConnection(incoming)) => { -895 if self.connections.close.is_none() { -896 self.incoming.push_back(incoming); -897 } else { -898 let transmit = -899 endpoint.refuse(incoming, &mut response_buffer); -900 respond(transmit, &response_buffer, sender); -901 } -902 } -903 Some(DatagramEvent::ConnectionEvent(handle, event)) => { -904 // Ignoring errors from dropped connections that haven't yet been cleaned up -905 received_connection_packet = true; -906 let _ = self -907 .connections -908 .senders -909 .get_mut(&handle) -910 .unwrap() -911 .send(ConnectionEvent::Proto(event)); -912 } -913 Some(DatagramEvent::Response(transmit)) => { -914 respond(transmit, &response_buffer, sender); -915 } -916 None => {} -917 } -918 } -919 } -920 } -921 Poll::Pending => { -922 return Ok(PollProgress { -923 received_connection_packet, -924 keep_going: false, -925 }); -926 } -927 // Ignore ECONNRESET as it's undefined in QUIC and may be injected by an -928 // attacker -929 Poll::Ready(Err(ref e)) if e.kind() == io::ErrorKind::ConnectionReset => { -930 continue; -931 } -932 Poll::Ready(Err(e)) => { -933 return Err(e); -934 } -935 } -936 if !self.recv_limiter.allow_work(|| runtime.now()) { -937 return Ok(PollProgress { -938 received_connection_packet, -939 keep_going: true, -940 }); -941 } -942 } -943 } -944} -945 -946impl fmt::Debug for RecvState { -947 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { -948 f.debug_struct("RecvState") -949 .field("incoming", &self.incoming) -950 .field("connections", &self.connections) -951 // recv_buf too large -952 .field("recv_limiter", &self.recv_limiter) -953 .finish_non_exhaustive() -954 } -955} -956 -957#[derive(Default)] -958struct PollProgress { -959 /// Whether a datagram was routed to an existing connection -960 received_connection_packet: bool, -961 /// Whether datagram handling was interrupted early by the work limiter for fairness -962 keep_going: bool, -963}

\ No newline at end of file +425 trace!("endpoint driver stopping"); +426 Poll::Ready(Ok(())) +427 } else { +428 drop(endpoint); +429 // If there is more work to do schedule the endpoint task again. +430 // `wake_by_ref()` is called outside the lock to minimize +431 // lock contention on a multithreaded runtime. +432 if keep_going { +433 cx.waker().wake_by_ref(); +434 } +435 Poll::Pending +436 } +437 } +438} +439 +440impl Drop for EndpointDriver { +441 fn drop(&mut self) { +442 let mut endpoint = self.0.state.lock().unwrap(); +443 endpoint.driver_lost = true; +444 self.0.shared.incoming.notify_waiters(); +445 // Drop all outgoing channels, signaling the termination of the endpoint to the associated +446 // connections. +447 endpoint.recv_state.connections.senders.clear(); +448 } +449} +450 +451#[derive(Debug)] +452pub(crate) struct EndpointInner { +453 pub(crate) state: Mutex<State>, +454 pub(crate) shared: Shared, +455} +456 +457impl EndpointInner { +458 pub(crate) fn accept( +459 &self, +460 incoming: proto::Incoming, +461 server_config: Option<Arc<ServerConfig>>, +462 ) -> Result<Connecting, ConnectionError> { +463 let mut state = self.state.lock().unwrap(); +464 let mut response_buffer = Vec::new(); +465 let now = state.runtime.now(); +466 match state +467 .inner +468 .accept(incoming, now, &mut response_buffer, server_config) +469 { +470 Ok((handle, conn)) => { +471 state.stats.accepted_handshakes += 1; +472 let sender = state.socket.create_sender(); +473 let runtime = state.runtime.clone(); +474 Ok(state +475 .recv_state +476 .connections +477 .insert(handle, conn, sender, runtime)) +478 } +479 Err(error) => { +480 if let Some(transmit) = error.response { +481 respond(transmit, &response_buffer, &mut state.sender); +482 } +483 Err(error.cause) +484 } +485 } +486 } +487 +488 pub(crate) fn refuse(&self, incoming: proto::Incoming) { +489 let mut state = self.state.lock().unwrap(); +490 state.stats.refused_handshakes += 1; +491 let mut response_buffer = Vec::new(); +492 let transmit = state.inner.refuse(incoming, &mut response_buffer); +493 respond(transmit, &response_buffer, &mut state.sender); +494 } +495 +496 pub(crate) fn retry(&self, incoming: proto::Incoming) -> Result<(), proto::RetryError> { +497 let mut state = self.state.lock().unwrap(); +498 let mut response_buffer = Vec::new(); +499 let transmit = state.inner.retry(incoming, &mut response_buffer)?; +500 respond(transmit, &response_buffer, &mut state.sender); +501 Ok(()) +502 } +503 +504 pub(crate) fn ignore(&self, incoming: proto::Incoming) { +505 let mut state = self.state.lock().unwrap(); +506 state.stats.ignored_handshakes += 1; +507 state.inner.ignore(incoming); +508 } +509} +510 +511#[derive(Debug)] +512pub(crate) struct State { +513 socket: Box<dyn AsyncUdpSocket>, +514 sender: Pin<Box<dyn UdpSender>>, +515 /// During an active migration, abandoned_socket receives traffic +516 /// until the first packet arrives on the new socket. +517 prev_socket: Option<Box<dyn AsyncUdpSocket>>, +518 inner: proto::Endpoint, +519 recv_state: RecvState, +520 driver: Option<Waker>, +521 ipv6: bool, +522 events: mpsc::UnboundedReceiver<(ConnectionHandle, EndpointEvent)>, +523 /// Number of live handles that can be used to initiate or handle I/O; excludes the driver +524 ref_count: usize, +525 driver_lost: bool, +526 runtime: Arc<dyn Runtime>, +527 stats: EndpointStats, +528 default_client_config: Option<ClientConfig>, +529} +530 +531#[derive(Debug)] +532pub(crate) struct Shared { +533 incoming: Notify, +534 idle: Notify, +535} +536 +537impl State { +538 fn drive_recv(&mut self, cx: &mut Context<'_>, now: Instant) -> Result<bool, io::Error> { +539 let get_time = || self.runtime.now(); +540 self.recv_state.recv_limiter.start_cycle(get_time); +541 if let Some(socket) = &mut self.prev_socket { +542 // We don't care about the `PollProgress` from old sockets. +543 let poll_res = self.recv_state.poll_socket( +544 cx, +545 &mut self.inner, +546 &mut **socket, +547 &mut self.sender, +548 &*self.runtime, +549 now, +550 ); +551 if poll_res.is_err() { +552 self.prev_socket = None; +553 } +554 }; +555 let poll_res = self.recv_state.poll_socket( +556 cx, +557 &mut self.inner, +558 &mut *self.socket, +559 &mut self.sender, +560 &*self.runtime, +561 now, +562 ); +563 self.recv_state.recv_limiter.finish_cycle(get_time); +564 let poll_res = poll_res?; +565 if poll_res.received_connection_packet { +566 // Traffic has arrived on self.socket, therefore there is no need for the abandoned +567 // one anymore. TODO: Account for multiple outgoing connections. +568 self.prev_socket = None; +569 } +570 Ok(poll_res.keep_going) +571 } +572 +573 fn handle_events(&mut self, cx: &mut Context<'_>, shared: &Shared) -> bool { +574 for _ in 0..IO_LOOP_BOUND { +575 let (ch, event) = match self.events.poll_recv(cx) { +576 Poll::Ready(Some(x)) => x, +577 Poll::Ready(None) => unreachable!("EndpointInner owns one sender"), +578 Poll::Pending => { +579 return false; +580 } +581 }; +582 +583 if event.is_drained() { +584 self.recv_state.connections.senders.remove(&ch); +585 if self.recv_state.connections.is_empty() { +586 shared.idle.notify_waiters(); +587 } +588 } +589 let Some(event) = self.inner.handle_event(ch, event) else { +590 continue; +591 }; +592 // Ignoring errors from dropped connections that haven't yet been cleaned up +593 let _ = self +594 .recv_state +595 .connections +596 .senders +597 .get_mut(&ch) +598 .unwrap() +599 .send(ConnectionEvent::Proto(event)); +600 } +601 +602 true +603 } +604} +605 +606impl Drop for State { +607 fn drop(&mut self) { +608 for incoming in self.recv_state.incoming.drain(..) { +609 self.inner.ignore(incoming); +610 } +611 } +612} +613 +614fn respond( +615 transmit: proto::Transmit, +616 response_buffer: &[u8], +617 sender: &mut Pin<Box<dyn UdpSender>>, +618) { +619 // Send if there's kernel buffer space; otherwise, drop it +620 // +621 // As an endpoint-generated packet, we know this is an +622 // immediate, stateless response to an unconnected peer, +623 // one of: +624 // +625 // - A version negotiation response due to an unknown version +626 // - A `CLOSE` due to a malformed or unwanted connection attempt +627 // - A stateless reset due to an unrecognized connection +628 // - A `Retry` packet due to a connection attempt when +629 // `use_retry` is set +630 // +631 // In each case, a well-behaved peer can be trusted to retry a +632 // few times, which is guaranteed to produce the same response +633 // from us. Repeated failures might at worst cause a peer's new +634 // connection attempt to time out, which is acceptable if we're +635 // under such heavy load that there's never room for this code +636 // to transmit. This is morally equivalent to the packet getting +637 // lost due to congestion further along the link, which +638 // similarly relies on peer retries for recovery. +639 +640 // Copied from rust 1.85's std::task::Waker::noop() implementation for backwards compatibility +641 const NOOP: RawWaker = { +642 const VTABLE: RawWakerVTable = RawWakerVTable::new( +643 // Cloning just returns a new no-op raw waker +644 |_| NOOP, +645 // `wake` does nothing +646 |_| {}, +647 // `wake_by_ref` does nothing +648 |_| {}, +649 // Dropping does nothing as we don't allocate anything +650 |_| {}, +651 ); +652 RawWaker::new(std::ptr::null(), &VTABLE) +653 }; +654 // SAFETY: Copied from rust stdlib, the NOOP waker is thread-safe and doesn't violate the RawWakerVTable contract, +655 // it doesn't access the data pointer at all. +656 let waker = unsafe { Waker::from_raw(NOOP) }; +657 let mut cx = Context::from_waker(&waker); +658 _ = sender.as_mut().poll_send( +659 &udp_transmit(&transmit, &response_buffer[..transmit.size]), +660 &mut cx, +661 ); +662} +663 +664#[inline] +665fn proto_ecn(ecn: udp::EcnCodepoint) -> proto::EcnCodepoint { +666 match ecn { +667 udp::EcnCodepoint::Ect0 => proto::EcnCodepoint::Ect0, +668 udp::EcnCodepoint::Ect1 => proto::EcnCodepoint::Ect1, +669 udp::EcnCodepoint::Ce => proto::EcnCodepoint::Ce, +670 } +671} +672 +673#[derive(Debug)] +674struct ConnectionSet { +675 /// Senders for communicating with the endpoint's connections +676 senders: FxHashMap<ConnectionHandle, mpsc::UnboundedSender<ConnectionEvent>>, +677 /// Stored to give out clones to new ConnectionInners +678 sender: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>, +679 /// Set if the endpoint has been manually closed +680 close: Option<(VarInt, Bytes)>, +681} +682 +683impl ConnectionSet { +684 fn insert( +685 &mut self, +686 handle: ConnectionHandle, +687 conn: proto::Connection, +688 sender: Pin<Box<dyn UdpSender>>, +689 runtime: Arc<dyn Runtime>, +690 ) -> Connecting { +691 let (send, recv) = mpsc::unbounded_channel(); +692 if let Some((error_code, ref reason)) = self.close { +693 send.send(ConnectionEvent::Close { +694 error_code, +695 reason: reason.clone(), +696 }) +697 .unwrap(); +698 } +699 self.senders.insert(handle, send); +700 Connecting::new(handle, conn, self.sender.clone(), recv, sender, runtime) +701 } +702 +703 fn is_empty(&self) -> bool { +704 self.senders.is_empty() +705 } +706} +707 +708pub(crate) fn ensure_ipv6(x: SocketAddr) -> SocketAddrV6 { +709 match x { +710 SocketAddr::V6(x) => x, +711 SocketAddr::V4(x) => SocketAddrV6::new(x.ip().to_ipv6_mapped(), x.port(), 0, 0), +712 } +713} +714 +715pin_project! { +716 /// Future produced by [`Endpoint::accept`] +717 pub struct Accept<'a> { +718 endpoint: &'a Endpoint, +719 #[pin] +720 notify: Notified<'a>, +721 } +722} +723 +724impl Future for Accept<'_> { +725 type Output = Option<Incoming>; +726 fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> { +727 let mut this = self.project(); +728 let mut endpoint = this.endpoint.inner.state.lock().unwrap(); +729 if endpoint.driver_lost { +730 return Poll::Ready(None); +731 } +732 if let Some(incoming) = endpoint.recv_state.incoming.pop_front() { +733 // Release the mutex lock on endpoint so cloning it doesn't deadlock +734 drop(endpoint); +735 let incoming = Incoming::new(incoming, this.endpoint.inner.clone()); +736 return Poll::Ready(Some(incoming)); +737 } +738 if endpoint.recv_state.connections.close.is_some() { +739 return Poll::Ready(None); +740 } +741 loop { +742 match this.notify.as_mut().poll(ctx) { +743 // `state` lock ensures we didn't race with readiness +744 Poll::Pending => return Poll::Pending, +745 // Spurious wakeup, get a new future +746 Poll::Ready(()) => this +747 .notify +748 .set(this.endpoint.inner.shared.incoming.notified()), +749 } +750 } +751 } +752} +753 +754#[derive(Debug)] +755pub(crate) struct EndpointRef(Arc<EndpointInner>); +756 +757impl EndpointRef { +758 pub(crate) fn new( +759 socket: Box<dyn AsyncUdpSocket>, +760 inner: proto::Endpoint, +761 ipv6: bool, +762 runtime: Arc<dyn Runtime>, +763 ) -> Self { +764 let (sender, events) = mpsc::unbounded_channel(); +765 let recv_state = RecvState::new(sender, socket.max_receive_segments(), &inner); +766 let sender = socket.create_sender(); +767 Self(Arc::new(EndpointInner { +768 shared: Shared { +769 incoming: Notify::new(), +770 idle: Notify::new(), +771 }, +772 state: Mutex::new(State { +773 socket, +774 sender, +775 prev_socket: None, +776 inner, +777 ipv6, +778 events, +779 driver: None, +780 ref_count: 0, +781 driver_lost: false, +782 recv_state, +783 runtime, +784 stats: EndpointStats::default(), +785 default_client_config: None, +786 }), +787 })) +788 } +789} +790 +791impl Clone for EndpointRef { +792 fn clone(&self) -> Self { +793 self.0.state.lock().unwrap().ref_count += 1; +794 Self(self.0.clone()) +795 } +796} +797 +798impl Drop for EndpointRef { +799 fn drop(&mut self) { +800 let endpoint = &mut *self.0.state.lock().unwrap(); +801 if let Some(x) = endpoint.ref_count.checked_sub(1) { +802 endpoint.ref_count = x; +803 if x == 0 { +804 // If the driver is about to be on its own, ensure it can shut down if the last +805 // connection is gone. +806 if let Some(task) = endpoint.driver.take() { +807 task.wake(); +808 } +809 } +810 } +811 } +812} +813 +814impl std::ops::Deref for EndpointRef { +815 type Target = EndpointInner; +816 fn deref(&self) -> &Self::Target { +817 &self.0 +818 } +819} +820 +821/// State directly involved in handling incoming packets +822struct RecvState { +823 incoming: VecDeque<proto::Incoming>, +824 connections: ConnectionSet, +825 recv_buf: Box<[u8]>, +826 recv_limiter: WorkLimiter, +827} +828 +829impl RecvState { +830 fn new( +831 sender: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>, +832 max_receive_segments: NonZeroUsize, +833 endpoint: &proto::Endpoint, +834 ) -> Self { +835 let recv_buf = vec![ +836 0; +837 endpoint.config().get_max_udp_payload_size().min(64 * 1024) as usize +838 * max_receive_segments.get() +839 * BATCH_SIZE +840 ]; +841 Self { +842 connections: ConnectionSet { +843 senders: FxHashMap::default(), +844 sender, +845 close: None, +846 }, +847 incoming: VecDeque::new(), +848 recv_buf: recv_buf.into(), +849 recv_limiter: WorkLimiter::new(RECV_TIME_BOUND), +850 } +851 } +852 +853 fn poll_socket( +854 &mut self, +855 cx: &mut Context<'_>, +856 endpoint: &mut proto::Endpoint, +857 socket: &mut dyn AsyncUdpSocket, +858 sender: &mut Pin<Box<dyn UdpSender>>, +859 runtime: &dyn Runtime, +860 now: Instant, +861 ) -> Result<PollProgress, io::Error> { +862 let mut received_connection_packet = false; +863 let mut metas = [RecvMeta::default(); BATCH_SIZE]; +864 let mut iovs: [IoSliceMut<'_>; BATCH_SIZE] = { +865 let mut bufs = self +866 .recv_buf +867 .chunks_mut(self.recv_buf.len() / BATCH_SIZE) +868 .map(IoSliceMut::new); +869 +870 // expect() safe as self.recv_buf is chunked into BATCH_SIZE items +871 // and iovs will be of size BATCH_SIZE, thus from_fn is called +872 // exactly BATCH_SIZE times. +873 std::array::from_fn(|_| bufs.next().expect("BATCH_SIZE elements")) +874 }; +875 loop { +876 match socket.poll_recv(cx, &mut iovs, &mut metas) { +877 Poll::Ready(Ok(msgs)) => { +878 self.recv_limiter.record_work(msgs); +879 for (meta, buf) in metas.iter().zip(iovs.iter()).take(msgs) { +880 let mut data: BytesMut = buf[0..meta.len].into(); +881 while !data.is_empty() { +882 let buf = data.split_to(meta.stride.min(data.len())); +883 let mut response_buffer = Vec::new(); +884 let addresses = FourTuple { +885 remote: meta.addr, +886 local_ip: meta.dst_ip, +887 }; +888 match endpoint.handle( +889 now, +890 addresses, +891 meta.ecn.map(proto_ecn), +892 buf, +893 &mut response_buffer, +894 ) { +895 Some(DatagramEvent::NewConnection(incoming)) => { +896 if self.connections.close.is_none() { +897 self.incoming.push_back(incoming); +898 } else { +899 let transmit = +900 endpoint.refuse(incoming, &mut response_buffer); +901 respond(transmit, &response_buffer, sender); +902 } +903 } +904 Some(DatagramEvent::ConnectionEvent(handle, event)) => { +905 // Ignoring errors from dropped connections that haven't yet been cleaned up +906 received_connection_packet = true; +907 let _ = self +908 .connections +909 .senders +910 .get_mut(&handle) +911 .unwrap() +912 .send(ConnectionEvent::Proto(event)); +913 } +914 Some(DatagramEvent::Response(transmit)) => { +915 respond(transmit, &response_buffer, sender); +916 } +917 None => {} +918 } +919 } +920 } +921 } +922 Poll::Pending => { +923 return Ok(PollProgress { +924 received_connection_packet, +925 keep_going: false, +926 }); +927 } +928 // Ignore ECONNRESET as it's undefined in QUIC and may be injected by an +929 // attacker +930 Poll::Ready(Err(ref e)) if e.kind() == io::ErrorKind::ConnectionReset => { +931 continue; +932 } +933 Poll::Ready(Err(e)) => { +934 return Err(e); +935 } +936 } +937 if !self.recv_limiter.allow_work(|| runtime.now()) { +938 return Ok(PollProgress { +939 received_connection_packet, +940 keep_going: true, +941 }); +942 } +943 } +944 } +945} +946 +947impl fmt::Debug for RecvState { +948 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { +949 f.debug_struct("RecvState") +950 .field("incoming", &self.incoming) +951 .field("connections", &self.connections) +952 // recv_buf too large +953 .field("recv_limiter", &self.recv_limiter) +954 .finish_non_exhaustive() +955 } +956} +957 +958#[derive(Default)] +959struct PollProgress { +960 /// Whether a datagram was routed to an existing connection +961 received_connection_packet: bool, +962 /// Whether datagram handling was interrupted early by the work limiter for fairness +963 keep_going: bool, +964}

\ No newline at end of file