use std::{
collections::VecDeque,
sync::atomic::{AtomicU64, Ordering},
time::{Duration, Instant},
};
use dashmap::{DashMap, mapref::entry::Entry};
use crate::{
BucketSize, ConditionalSetOutcome, HistoryPreservation, RateLimitComparator, RateLimitDecision,
SuppressedRateLimitSnapshot,
builder::ProviderConfig,
common::{
Bucket, HardLimitFactor, HistoryUpdateMode, RandomState, RateLimit,
SuppressionFactorCachePeriod, WindowSize,
},
};
#[derive(Debug)]
pub(crate) struct RateLimitSeries {
pub hard_window_limit: f64,
pub buckets: VecDeque<Bucket>,
pub total_count: AtomicU64,
pub total_declined_count: AtomicU64,
}
impl RateLimitSeries {
pub fn new(hard_window_limit: f64) -> Self {
Self {
hard_window_limit,
buckets: VecDeque::new(),
total_count: AtomicU64::new(0),
total_declined_count: AtomicU64::new(0),
}
}
}
#[derive(Debug)]
pub struct SuppressedLocalRateLimiter {
window_size: WindowSize,
window_size_ms: u128,
window_duration: Duration,
bucket_size: BucketSize,
pub(crate) series: DashMap<String, RateLimitSeries, RandomState>,
hard_limit_factor: HardLimitFactor,
suppression_factor_cache_period: SuppressionFactorCachePeriod,
pub(crate) suppression_factors: DashMap<String, (Instant, f64), RandomState>,
}
impl SuppressedLocalRateLimiter {
pub(crate) fn new(options: ProviderConfig) -> Self {
Self {
hard_limit_factor: options.hard_limit_factor,
window_size_ms: options.window_size.as_milliseconds(),
window_size: options.window_size,
window_duration: Duration::from_secs(options.window_size.as_seconds()),
suppression_factor_cache_period: options.suppression_factor_cache_period,
bucket_size: options.bucket_size,
series: DashMap::default(),
suppression_factors: DashMap::default(),
}
}
pub fn inc(&self, key: &str, rate_limit: &RateLimit, count: u64) -> RateLimitDecision {
let mut rng = |p: f64| rand::random_bool(p);
self.inc_with_rng(key, rate_limit, count, &mut rng)
}
#[inline(always)]
pub(crate) fn inc_with_rng(
&self,
key: &str,
rate_limit: &RateLimit,
count: u64,
random_bool: &mut impl FnMut(f64) -> bool,
) -> RateLimitDecision {
let series = match self.series.get(key) {
Some(series) => series,
None => self
.series
.entry(key.to_string())
.or_insert_with(|| {
RateLimitSeries::new(
rate_limit.as_per_second()
* self.hard_limit_factor.as_multiplier()
* self.window_size.as_seconds() as f64,
)
})
.downgrade(),
};
let series = match series.buckets.front() {
Some(oldest_bucket)
if oldest_bucket.timestamp.elapsed().as_millis() > self.window_size_ms =>
{
drop(series);
let mut series = self.series.entry(key.to_string()).or_insert_with(|| {
RateLimitSeries::new(
rate_limit.as_per_second()
* self.hard_limit_factor.as_multiplier()
* self.window_size.as_seconds() as f64,
)
});
let (_, evicted) = Self::evict_expired(&mut series, self.window_duration);
if evicted {
self.suppression_factors.remove(key);
}
series.downgrade()
}
_ => series,
};
let soft_window_limit =
(series.hard_window_limit / self.hard_limit_factor.as_multiplier()) as u64;
let hard_window_limit = series.hard_window_limit as u64;
let total_count = series.total_count.load(Ordering::Acquire);
let total_declined_count = series.total_declined_count.load(Ordering::Acquire);
let forecasted_allowed_count = total_count
.saturating_sub(total_declined_count)
.saturating_add(count);
let reached_hard_window_limit = forecasted_allowed_count == hard_window_limit;
let should_return_allowed =
forecasted_allowed_count <= soft_window_limit || reached_hard_window_limit;
let suppression_factor;
let should_allow;
if should_return_allowed {
should_allow = true;
if reached_hard_window_limit {
suppression_factor = self.persist_suppression_factor(key, 1f64).1;
} else {
suppression_factor = 0f64;
}
} else {
suppression_factor = if forecasted_allowed_count > hard_window_limit {
self.persist_suppression_factor(key, 1f64).1
} else {
self.get_suppression_factor_without_bucket_expire(key)
};
should_allow = if suppression_factor == 0f64 {
true
} else if suppression_factor == 1f64 {
false
} else {
random_bool(1f64 - suppression_factor)
};
if !should_allow {
series
.total_declined_count
.fetch_add(count, Ordering::AcqRel);
}
}
series.total_count.fetch_add(count, Ordering::AcqRel);
if let Some(latest_bucket) = series.buckets.back()
&& latest_bucket.timestamp.elapsed().as_millis()
<= self.bucket_size.as_milliseconds() as u128
{
latest_bucket.count.fetch_add(count, Ordering::AcqRel);
if !should_allow {
latest_bucket
.declined_count
.fetch_add(count, Ordering::AcqRel);
}
} else {
let hard_window_limit = series.hard_window_limit;
drop(series);
let bucket = Bucket {
count: count.into(),
declined_count: AtomicU64::new(if should_allow { 0 } else { count }),
timestamp: Instant::now(),
};
match self.series.entry(key.to_string()) {
Entry::Occupied(mut entry) => entry.get_mut().buckets.push_back(bucket),
Entry::Vacant(entry) => {
let mut series = RateLimitSeries::new(hard_window_limit);
series.total_count.store(count, Ordering::Release);
if !should_allow {
series.total_declined_count.store(count, Ordering::Release);
}
series.buckets.push_back(bucket);
entry.insert(series);
}
}
}
if should_return_allowed {
RateLimitDecision::Allowed
} else {
RateLimitDecision::Suppressed {
suppression_factor,
is_allowed: should_allow,
}
}
}
fn remove_expired_buckets(&self, key: &str) {
let Some(series) = self.series.get(key) else {
return;
};
let Some(oldest_bucket) = series.buckets.front() else {
return;
};
if oldest_bucket.timestamp.elapsed().as_millis() <= self.window_size_ms {
return;
}
drop(series);
let Some(mut series) = self.series.get_mut(key) else {
return;
};
let (_, evicted) = Self::evict_expired(&mut series, self.window_duration);
drop(series);
if evicted {
self.suppression_factors.remove(key);
}
}
pub fn get_suppression_factor(&self, key: &str) -> f64 {
self.remove_expired_buckets(key);
self.get_suppression_factor_without_bucket_expire(key)
}
fn get_suppression_factor_without_bucket_expire(&self, key: &str) -> f64 {
let suppression_factor = match self.suppression_factors.get(key) {
None => self.calculate_suppression_factor(key).1,
Some(val)
if val.0.elapsed().as_millis()
< self.suppression_factor_cache_period.as_milliseconds() as u128 =>
{
val.1
}
Some(val) => {
drop(val);
self.calculate_suppression_factor(key).1
}
};
if suppression_factor < 0f64 {
panic!(
"SuppressedLocalRateLimiter::get_suppression_factor: negative suppression factor"
);
}
if suppression_factor > 1f64 {
panic!("SuppressedLocalRateLimiter::get_suppression_factor: suppression factor > 1");
}
suppression_factor
}
#[inline(always)]
fn persist_suppression_factor(&self, key: &str, value: f64) -> (Instant, f64) {
let persist = (Instant::now(), value);
self.suppression_factors.insert(key.to_string(), persist);
persist
}
fn calculate_suppression_factor(&self, key: &str) -> (Instant, f64) {
let Some(series) = self.series.get(key) else {
return self.persist_suppression_factor(key, 0f64);
};
if series.buckets.is_empty() {
return self.persist_suppression_factor(key, 0f64);
}
let soft_window_limit =
(series.hard_window_limit / self.hard_limit_factor.as_multiplier()) as u64;
let total_count = series.total_count.load(Ordering::Acquire);
let total_declined_count = series.total_declined_count.load(Ordering::Acquire);
if total_count >= series.hard_window_limit as u64 {
return self.persist_suppression_factor(key, 1f64);
}
let accepted_count = total_count.saturating_sub(total_declined_count);
if accepted_count < soft_window_limit {
return self.persist_suppression_factor(key, 0f64);
}
if accepted_count == soft_window_limit
&& soft_window_limit == series.hard_window_limit as u64
{
return self.persist_suppression_factor(key, 1f64);
}
let mut total_in_last_second = 0u64;
for bucket in series.buckets.iter().rev() {
if bucket.timestamp.elapsed().as_millis() > 1000 {
break;
}
total_in_last_second =
total_in_last_second.saturating_add(bucket.count.load(Ordering::Acquire));
}
let average_rate_in_window: f64 = total_count as f64 / self.window_size.as_seconds() as f64;
let perceived_rate_limit = average_rate_in_window.max(total_in_last_second as f64);
let suppression_factor = 1f64
- (soft_window_limit as f64
/ self.window_size.as_seconds() as f64
/ perceived_rate_limit);
self.persist_suppression_factor(key, suppression_factor)
}
pub fn get(&self, key: &str) -> SuppressedRateLimitSnapshot {
let Some(series) = self.series.get(key) else {
return SuppressedRateLimitSnapshot::default();
};
let (total_count, contained_expired) =
Self::inspect_live_total(&series, self.window_duration);
if !contained_expired {
let total_declined_count = series.total_declined_count.load(Ordering::Acquire);
drop(series);
return SuppressedRateLimitSnapshot {
total: total_count,
total_declined: total_declined_count,
suppression_factor: self.get_suppression_factor_without_bucket_expire(key),
};
}
drop(series);
let Some(mut series) = self.series.get_mut(key) else {
return SuppressedRateLimitSnapshot::default();
};
let (total_count, evicted) = Self::evict_expired(&mut series, self.window_duration);
let total_declined_count = series.total_declined_count.load(Ordering::Acquire);
drop(series);
if evicted {
self.suppression_factors.remove(key);
}
SuppressedRateLimitSnapshot {
total: total_count,
total_declined: total_declined_count,
suppression_factor: self.get_suppression_factor_without_bucket_expire(key),
}
}
fn evict_expired(series: &mut RateLimitSeries, window_duration: Duration) -> (u64, bool) {
let now = Instant::now();
let split = series
.buckets
.partition_point(|bucket| now.duration_since(bucket.timestamp) > window_duration);
if split == 0 {
return (series.total_count.load(Ordering::Acquire), false);
}
let (removed_count, removed_declined_count) =
series
.buckets
.drain(..split)
.fold((0u64, 0u64), |(count, declined_count), bucket| {
(
count + bucket.count.load(Ordering::Acquire),
declined_count + bucket.declined_count.load(Ordering::Acquire),
)
});
let total_count = series
.total_count
.fetch_sub(removed_count, Ordering::AcqRel)
- removed_count;
series
.total_declined_count
.fetch_sub(removed_declined_count, Ordering::AcqRel);
(total_count, true)
}
fn inspect_live_total(series: &RateLimitSeries, window_duration: Duration) -> (u64, bool) {
let now = Instant::now();
let split = series
.buckets
.partition_point(|bucket| now.duration_since(bucket.timestamp) > window_duration);
if split == 0 {
return (series.total_count.load(Ordering::Acquire), split > 0);
}
let total_count = series
.buckets
.iter()
.skip(split)
.map(|bucket| bucket.count.load(Ordering::Acquire))
.sum();
(total_count, split > 0)
}
fn apply_history_update(
series: &mut RateLimitSeries,
count: u64,
old_total: u64,
mode: HistoryUpdateMode,
) {
match mode {
HistoryUpdateMode::Replace => {
series.buckets.clear();
series.buckets.push_back(Bucket {
count: count.into(),
declined_count: AtomicU64::new(0),
timestamp: Instant::now(),
});
}
HistoryUpdateMode::Preserve(preservation) if count > old_total => {
let delta = count - old_total;
let bucket = match preservation {
HistoryPreservation::PreserveNewest => series.buckets.back(),
HistoryPreservation::PreserveOldest => series.buckets.front(),
};
if let Some(bucket) = bucket {
bucket.count.fetch_add(delta, Ordering::AcqRel);
} else {
series.buckets.push_back(Bucket {
count: delta.into(),
declined_count: AtomicU64::new(0),
timestamp: Instant::now(),
});
}
}
HistoryUpdateMode::Preserve(preservation) if count < old_total => {
let mut to_remove = old_total - count;
let mut bucket;
let mut bucket_count;
let mut declined_count;
let mut retained_count;
let mut retained_declined_count;
while to_remove > 0 {
bucket = match preservation {
HistoryPreservation::PreserveNewest => series.buckets.front(),
HistoryPreservation::PreserveOldest => series.buckets.back(),
};
let Some(bucket) = bucket else {
break;
};
bucket_count = bucket.count.load(Ordering::Acquire);
if bucket_count <= to_remove {
to_remove -= bucket_count;
match preservation {
HistoryPreservation::PreserveNewest => {
series.buckets.pop_front();
}
HistoryPreservation::PreserveOldest => {
series.buckets.pop_back();
}
}
} else {
declined_count = bucket.declined_count.load(Ordering::Acquire);
retained_count = bucket_count - to_remove;
retained_declined_count =
((declined_count as u128 * retained_count as u128)
/ bucket_count as u128) as u64;
bucket.count.store(retained_count, Ordering::Release);
bucket
.declined_count
.store(retained_declined_count, Ordering::Release);
to_remove = 0;
}
}
}
HistoryUpdateMode::Preserve(_) => {}
}
let total_declined_count = series
.buckets
.iter()
.map(|bucket| bucket.declined_count.load(Ordering::Acquire))
.sum();
series.total_count.store(count, Ordering::Release);
series
.total_declined_count
.store(total_declined_count, Ordering::Release);
}
fn set_if_with_history_mode(
&self,
key: &str,
rate_limit: &RateLimit,
comparator: RateLimitComparator,
count: u64,
mode: HistoryUpdateMode,
) -> (u64, u64) {
let existing = match self.series.get(key) {
Some(series) => {
let (old_total, contains_expired) =
Self::inspect_live_total(&series, self.window_duration);
if !comparator.matches(old_total) {
return (old_total, old_total);
}
let hard_window_limit = rate_limit.as_per_second()
* self.hard_limit_factor.as_multiplier()
* self.window_size.as_seconds() as f64;
let unchanged = count > 0
&& matches!(mode, HistoryUpdateMode::Preserve(_))
&& !contains_expired
&& old_total == count
&& series.hard_window_limit == hard_window_limit;
if unchanged {
return (old_total, old_total);
}
true
}
None => false,
};
if !existing && (!comparator.matches(0) || count == 0) {
return (0, 0);
}
let (new_total, old_total) = match self.series.entry(key.to_string()) {
Entry::Occupied(mut entry) => {
let (old_total, contains_expired) =
Self::inspect_live_total(entry.get(), self.window_duration);
if !comparator.matches(old_total) {
return (old_total, old_total);
}
if count == 0 {
entry.remove();
(0, old_total)
} else {
let series = entry.get_mut();
if contains_expired {
Self::evict_expired(series, self.window_duration);
}
series.hard_window_limit = rate_limit.as_per_second()
* self.hard_limit_factor.as_multiplier()
* self.window_size.as_seconds() as f64;
Self::apply_history_update(series, count, old_total, mode);
(count, old_total)
}
}
Entry::Vacant(entry) => {
if !comparator.matches(0) || count == 0 {
return (0, 0);
}
let mut series = RateLimitSeries::new(
rate_limit.as_per_second()
* self.hard_limit_factor.as_multiplier()
* self.window_size.as_seconds() as f64,
);
Self::apply_history_update(&mut series, count, 0, mode);
entry.insert(series);
(count, 0)
}
};
self.suppression_factors.remove(key);
(new_total, old_total)
}
pub fn set_if(
&self,
key: &str,
rate_limit: &RateLimit,
comparator: RateLimitComparator,
count: u64,
) -> ConditionalSetOutcome {
let (current_total, previous_total) = self.set_if_with_history_mode(
key,
rate_limit,
comparator,
count,
HistoryUpdateMode::Replace,
);
ConditionalSetOutcome {
matched: comparator.matches(previous_total),
previous_total,
current_total,
}
}
pub fn set_if_preserve_history(
&self,
key: &str,
rate_limit: &RateLimit,
comparator: RateLimitComparator,
count: u64,
preservation: HistoryPreservation,
) -> ConditionalSetOutcome {
let (current_total, previous_total) = self.set_if_with_history_mode(
key,
rate_limit,
comparator,
count,
HistoryUpdateMode::Preserve(preservation),
);
ConditionalSetOutcome {
matched: comparator.matches(previous_total),
previous_total,
current_total,
}
}
pub(crate) fn cleanup(&self, stale_after_ms: u64) {
self.suppression_factors
.retain(|_, (instant, _)| instant.elapsed().as_millis() < stale_after_ms as u128);
self.series.retain(|_, series| match series.buckets.back() {
None => false,
Some(latest_bucket)
if latest_bucket.timestamp.elapsed().as_millis() > stale_after_ms as u128 =>
{
false
}
Some(_) => true,
});
} }