pingora_timeout/
lib.rs

1// Copyright 2026 Cloudflare, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![warn(clippy::all)]
16
17//! A drop-in replacement of [tokio::time::timeout] which is much more efficient.
18//!
19//! Similar to [tokio::time::timeout] but more efficient on busy concurrent IOs where timeouts are
20//! created and canceled very frequently.
21//!
22//! This crate provides the following optimizations
23//! - The timeouts lazily initializes their timer when the Future is pending for the first time.
24//! - There is no global lock for creating and cancelling timeouts.
25//! - Timeout timers are rounded to the next 10ms tick and timers are shared across all timeouts with the same deadline.
26//!
27//! Benchmark:
28//!
29//! 438.302µs total, 4ns avg per iteration
30//!
31//! v.s. Tokio timeout():
32//!
33//! 10.716192ms total, 107ns avg per iteration
34//!
35
36pub mod fast_timeout;
37pub mod timer;
38
39pub use fast_timeout::fast_sleep as sleep;
40pub use fast_timeout::fast_timeout as timeout;
41
42use pin_project_lite::pin_project;
43use std::future::Future;
44use std::pin::Pin;
45use std::task::{self, Poll};
46use tokio::time::{sleep as tokio_sleep, Duration};
47
48/// The interface to start a timeout
49///
50/// Users don't need to interact with this trait
51pub trait ToTimeout {
52    fn timeout(&self) -> Pin<Box<dyn Future<Output = ()> + Send + Sync>>;
53    fn create(d: Duration) -> Self;
54}
55
56/// The timeout generated by [tokio_timeout()].
57///
58/// Users don't need to interact with this object.
59pub struct TokioTimeout(Duration);
60
61impl ToTimeout for TokioTimeout {
62    fn timeout(&self) -> Pin<Box<dyn Future<Output = ()> + Send + Sync>> {
63        Box::pin(tokio_sleep(self.0))
64    }
65
66    fn create(d: Duration) -> Self {
67        TokioTimeout(d)
68    }
69}
70
71/// The error type returned when the timeout is reached.
72#[derive(Debug)]
73pub struct Elapsed;
74
75impl std::fmt::Display for Elapsed {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        write!(f, "Timeout Elapsed")
78    }
79}
80
81impl std::error::Error for Elapsed {}
82
83/// The [tokio::time::timeout] with just lazy timer initialization.
84///
85/// The timer is created the first time the `future` is pending. This avoids unnecessary timer
86/// creation and cancellation on busy IOs with a good chance to be already ready (e.g., reading
87/// data from TCP where the recv buffer already has a lot of data to read right away).
88pub fn tokio_timeout<T>(duration: Duration, future: T) -> Timeout<T, TokioTimeout>
89where
90    T: Future,
91{
92    Timeout::<T, TokioTimeout>::new_with_delay(future, duration)
93}
94
95pin_project! {
96    /// The timeout future returned by the timeout functions
97    #[must_use = "futures do nothing unless you `.await` or poll them"]
98    pub struct Timeout<T, F> {
99        #[pin]
100        value: T,
101        #[pin]
102        delay: Option<Pin<Box<dyn Future<Output = ()> + Send + Sync>>>,
103        callback: F, // callback to create the timer
104    }
105}
106
107impl<T, F> Timeout<T, F>
108where
109    F: ToTimeout,
110{
111    pub(crate) fn new_with_delay(value: T, d: Duration) -> Timeout<T, F> {
112        Self::new_with_callback(value, F::create(d))
113    }
114
115    pub(crate) fn new_with_callback(value: T, callback: F) -> Timeout<T, F> {
116        Timeout {
117            value,
118            delay: None,
119            callback,
120        }
121    }
122}
123
124impl<T, F> Future for Timeout<T, F>
125where
126    T: Future,
127    F: ToTimeout,
128{
129    type Output = Result<T::Output, Elapsed>;
130
131    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
132        let mut me = self.project();
133
134        // First, try polling the future
135        if let Poll::Ready(v) = me.value.poll(cx) {
136            return Poll::Ready(Ok(v));
137        }
138
139        let delay = me
140            .delay
141            .get_or_insert_with(|| Box::pin(me.callback.timeout()));
142
143        match delay.as_mut().poll(cx) {
144            Poll::Pending => Poll::Pending,
145            Poll::Ready(()) => Poll::Ready(Err(Elapsed {})),
146        }
147    }
148}
149
150#[cfg(test)]
151mod tests {
152    use super::*;
153
154    #[tokio::test]
155    async fn test_timeout() {
156        let fut = tokio_sleep(Duration::from_secs(1000));
157        let to = timeout(Duration::from_secs(1), fut);
158        assert!(to.await.is_err())
159    }
160
161    #[tokio::test]
162    async fn test_instantly_return() {
163        let fut = async { 1 };
164        let to = timeout(Duration::from_secs(1), fut);
165        assert_eq!(to.await.unwrap(), 1)
166    }
167
168    #[tokio::test]
169    async fn test_delayed_return() {
170        let fut = async {
171            tokio_sleep(Duration::from_secs(1)).await;
172            1
173        };
174        let to = timeout(Duration::from_secs(1000), fut);
175        assert_eq!(to.await.unwrap(), 1)
176    }
177}
pingora_timeout/lib.rs

pingora_timeout/
lib.rs
