Recreate timers less often

With the current timer setup, Quinn destroys a tokio timer each time
the timer is disabled, and recreates it when a timeout is required again.
This has a certain attached cost, since timer creation requires a memory
allocation, and both creating the timer as well as resetting and polling
it incur some synchronization costs.

With this change, we keep one timer around and reuse it - which saves
some of these costs.

In addition to this, the change checks the connection first on whether
a timer is still required before polling the timeout. This can remove
some unnecessary wakeups.

quinn/src/connection.rs

@@ -655,6 +655,7 @@ where
             on_connected: Some(on_connected),
             connected: false,
             timer: None,
+            timer_deadline: None,
             conn_events,
             endpoint_events,
             blocked_writers: HashMap::new(),
@@ -726,6 +727,7 @@ where
     on_connected: Option<oneshot::Sender<bool>>,
     connected: bool,
     timer: Option<Delay>,
+    timer_deadline: Option<TokioInstant>,
     conn_events: mpsc::UnboundedReceiver<ConnectionEvent>,
     endpoint_events: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>,
     pub(crate) blocked_writers: HashMap<StreamId, Waker>,
@@ -863,39 +865,49 @@ where
     }
 
     fn drive_timer(&mut self, cx: &mut Context) -> bool {
-        let mut keep_going = false;
-        loop {
-            if let Some(ref mut delay) = self.timer {
-                if delay.poll_unpin(cx) == Poll::Ready(()) {
-                    // We must get a fresh `now` each iteration. If the value were cached and tokio
-                    // deems a timer for a later time than the cached value to be expired, we'd get
-                    // stuck in an infinite loop resetting it.
-                    self.inner.handle_timeout(Instant::now());
-                    self.timer = None;
-                    // A timer expired, so the caller needs to check for new transmits, which might
-                    // cause new timers to be set.
-                    keep_going = true;
+        // Check whether we need to (re)set the timer. If so, we must poll again to ensure the
+        // timer is registered with the runtime (and check whether it's already
+        // expired).
+        match self.inner.poll_timeout().map(TokioInstant::from_std) {
+            Some(deadline) => {
+                if let Some(delay) = &mut self.timer {
+                    // There is no need to reset the tokio timer if the deadline
+                    // did not change
+                    if self
+                        .timer_deadline
+                        .map(|current_deadline| current_deadline != deadline)
+                        .unwrap_or(true)
+                    {
+                        delay.reset(deadline);
+                    }
+                } else {
+                    self.timer = Some(delay_until(deadline));
                 }
+                // Store the actual expiration time of the timer
+                self.timer_deadline = Some(deadline);
             }
-            // Check whether we need to (re)set the timer. If so, we must poll again to ensure the
-            // timer is registered with the runtime (and check whether it's already
-            // expired). Otherwise, exit the loop.
-            match (
-                self.inner.poll_timeout().map(TokioInstant::from_std),
-                &mut self.timer,
-            ) {
-                (Some(timeout), &mut None) => self.timer = Some(delay_until(timeout)),
-                (Some(timeout), &mut Some(ref mut delay)) if delay.deadline() != timeout => {
-                    delay.reset(timeout);
-                }
-                (None, _) => {
-                    self.timer = None;
-                    break;
-                }
-                _ => break,
+            None => {
+                self.timer_deadline = None;
+                return false;
             }
         }
-        keep_going
+
+        if self.timer_deadline.is_none() {
+            return false;
+        }
+
+        let delay = self.timer.as_mut().expect("timer must exist in this state");
+        if delay.poll_unpin(cx).is_pending() {
+            // Since there wasn't a timeout event, there is nothing new
+            // for the connection to do
+            return false;
+        }
+
+        // A timer expired, so the caller needs to check for
+        // new transmits, which might cause new timers to be set.
+        self.inner.handle_timeout(Instant::now());
+        self.timer_deadline = None;
+        true
     }
 
     /// Wake up a blocked `Driver` task to process I/O