Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 26 additions & 3 deletions pingora-limits/src/rate.rs
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,8 @@ impl Rate {
// should be short enough not to overflow
let now = Instant::now().duration_since(self.start).as_millis() as u64;
let last_reset = self.last_reset_time.load(Ordering::SeqCst);
let past_ms = now - last_reset;
// another thread may have stored a later `now` after we read the clock
let past_ms = now.saturating_sub(last_reset);

if past_ms < self.reset_interval_ms {
// no need to reset
Expand All @@ -164,14 +165,14 @@ impl Rate {
self.red_or_blue.store(!red_or_blue, Ordering::SeqCst);
// if current time is beyond 2 intervals, the data stored in the previous slot
// is also stale, we should clear that too
if now - last_reset >= self.reset_interval_ms * 2 {
if past_ms >= self.reset_interval_ms * 2 {
// Note that this is the previous one now because we just flipped self.red_or_blue
self.current(red_or_blue).reset();
}
}
Err(new) => {
// another thread beats us to it
assert!(new >= now - 1000); // double check that the new timestamp looks right
assert!(new >= now.saturating_sub(1000)); // double check that the new timestamp looks right
}
}

Expand Down Expand Up @@ -263,6 +264,28 @@ mod tests {
assert_approx_eq!(f64, left, right, epsilon = 0.15)
}

#[test]
fn test_concurrent_reset_does_not_panic() {
// A short interval makes threads race to reset the slots, and keeps
// every timestamp within the first second after creation.
let r = std::sync::Arc::new(Rate::new(Duration::from_millis(1)));
let threads: Vec<_> = (0..8)
.map(|_| {
let r = r.clone();
std::thread::spawn(move || {
let end = Instant::now() + Duration::from_millis(500);
while Instant::now() < end {
r.observe(&1, 1);
r.rate(&1);
}
})
})
.collect();
for t in threads {
t.join().unwrap();
}
}

#[test]
fn test_observe_rate_custom_90_10() {
let r = Rate::new(Duration::from_secs(1));
Expand Down
Loading