diff --git a/quinn/src/endpoint.rs b/quinn/src/endpoint.rs index 274a501bf..8eba9dac1 100644 --- a/quinn/src/endpoint.rs +++ b/quinn/src/endpoint.rs @@ -190,7 +190,7 @@ impl Endpoint { } else { addr }; - let (ch, conn) = endpoint.inner.connect(config, addr, server_name)?; + let (ch, conn) = self.inner.shared.inner.connect(config, addr, server_name)?; let udp_state = endpoint.udp_state.clone(); Ok(endpoint .connections @@ -224,9 +224,7 @@ impl Endpoint { /// Useful for e.g. refreshing TLS certificates without disrupting existing connections. pub fn set_server_config(&self, server_config: Option) { self.inner - .state - .lock() - .unwrap() + .shared .inner .set_server_config(server_config.map(Arc::new)) } @@ -321,7 +319,7 @@ impl Future for EndpointDriver { let now = Instant::now(); let mut keep_going = false; - keep_going |= endpoint.drive_recv(cx, now)?; + keep_going |= endpoint.drive_recv(cx, now, &self.0.shared)?; keep_going |= endpoint.handle_events(cx, &self.0.shared); keep_going |= endpoint.drive_send(cx)?; @@ -365,7 +363,6 @@ pub(crate) struct EndpointInner { pub(crate) struct State { socket: Box, udp_state: Arc, - inner: proto::Endpoint, outgoing: VecDeque, incoming: VecDeque, driver: Option, @@ -385,10 +382,16 @@ pub(crate) struct State { pub(crate) struct Shared { incoming: Notify, idle: Notify, + inner: proto::Endpoint, } impl State { - fn drive_recv<'a>(&'a mut self, cx: &mut Context, now: Instant) -> Result { + fn drive_recv<'a>( + &'a mut self, + cx: &mut Context, + now: Instant, + shared: &Shared, + ) -> Result { self.recv_limiter.start_cycle(); let mut metas = [RecvMeta::default(); BATCH_SIZE]; let mut iovs = MaybeUninit::<[IoSliceMut<'a>; BATCH_SIZE]>::uninit(); @@ -410,7 +413,7 @@ impl State { let mut data: BytesMut = buf[0..meta.len].into(); while !data.is_empty() { let buf = data.split_to(meta.stride.min(data.len())); - match self.inner.handle( + match shared.inner.handle( now, meta.addr, meta.dst_ip, @@ -513,7 +516,7 @@ impl State { shared.idle.notify_waiters(); } } - if let Some(event) = self.inner.handle_event(ch, e) { + if let Some(event) = shared.inner.handle_event(ch, e) { // Ignoring errors from dropped connections that haven't yet been cleaned up let _ = self .connections @@ -665,11 +668,11 @@ impl EndpointRef { shared: Shared { incoming: Notify::new(), idle: Notify::new(), + inner, }, state: Mutex::new(State { socket, udp_state, - inner, ipv6, events, outgoing: VecDeque::new(),