1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
/// Performance Tuning API
use std::{thread, time::Duration};
use tokio::time::{sleep_until, Instant};
use super::SimpleRateLimiter;
impl SimpleRateLimiter {
/// Returns the capacity of the underyling `VecDeque<Instant>`.
pub fn capacity(&self) -> usize {
let mut history = self
.history
.lock()
.expect("rate limiter mutex was poisoned");
self.clean_history(&mut history, Instant::now());
history.capacity()
}
/// Reserve the specified capacity in the underlying `VecDeque<Instant>`.
pub fn reserve(&self, capacity: usize) {
let mut history = self
.history
.lock()
.expect("rate limiter mutex was poisoned");
self.clean_history(&mut history, Instant::now());
history.reserve(capacity)
}
/// Shrink the underlying `VecDeque<Instant>` to the standard capacity of `2 * limit`.
pub fn shrink(&self) {
self.shrink_to(2 * self.limit)
}
/// Shrink the underlying `VecDeque<Instant>` to the standard capacity of `2 * limit`
/// after all currently scheduled requests have cleared the window.
pub async fn shrink_later(&self) {
self.shrink_to_later(2 * self.limit).await
}
/// Shrink the underlying `VecDeque<Instant>` to the specified capacity.
pub fn shrink_to(&self, capacity: usize) {
let mut history = self
.history
.lock()
.expect("rate limiter mutex was poisoned");
self.clean_history(&mut history, Instant::now());
history.shrink_to(capacity)
}
/// Shrink the underlying `VecDeque<Instant>` to the specified capacity after
/// all currently scheduled requests have cleared the window.
pub async fn shrink_to_later(&self, capacity: usize) {
let maybe_youngest = {
let mut history = self
.history
.lock()
.expect("rate limiter mutex was poisoned");
self.clean_history(&mut history, Instant::now());
history.back().copied()
};
if let Some(youngest) = maybe_youngest {
sleep_until(youngest + self.window + Duration::from_nanos(1)).await;
}
self.shrink_to(capacity);
}
/// Force the rate limiter to pause by sleeping for the supplied duration while holding the
/// lock on the underlying `VecDeque<Instant>`. Useful for introducing random jitter into a
/// rate limiter in the steady state.
/// This function is blocking.
pub fn pause(&self, duration: Duration) {
let mut history = self
.history
.lock()
.expect("rate limiter mutex was poisoned");
thread::sleep(duration);
self.clean_history(&mut history, Instant::now());
}
/// Force the rate limiter to pause if it is in the steady
/// state. Otherwise, it returns early without sleeping for the supplied duration.
/// Returns true if the sleep was called. Useful for introducing random jitter.
/// This function is blocking.
pub fn pause_if_saturated(&self, duration: Duration) -> bool {
let saturated = self.len() >= self.limit;
if saturated {
self.pause(duration);
}
saturated
}
}