Lift proto::Endpoint out of the endpoint driver mutex

Locking the whole thing is no longer necessary now that Endpoint uses
interior mutability.

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<ServerConfig>) {
         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<dyn AsyncUdpSocket>,
     udp_state: Arc<UdpState>,
-    inner: proto::Endpoint,
     outgoing: VecDeque<udp::Transmit>,
     incoming: VecDeque<Connecting>,
     driver: Option<Waker>,
@@ -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<bool, io::Error> {
+    fn drive_recv<'a>(
+        &'a mut self,
+        cx: &mut Context,
+        now: Instant,
+        shared: &Shared,
+    ) -> Result<bool, io::Error> {
         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(),