-
Notifications
You must be signed in to change notification settings - Fork 21
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Support pending writes within CachedCounterValue #299
Changes from 1 commit
8c48bff
df3b556
f3a7f07
620c021
826b8a9
87f69a7
9d2b2c9
49dd4c4
83ba6a7
2b22301
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -4,10 +4,12 @@ use crate::storage::redis::{ | |
DEFAULT_MAX_CACHED_COUNTERS, DEFAULT_MAX_TTL_CACHED_COUNTERS_SEC, | ||
DEFAULT_TTL_RATIO_CACHED_COUNTERS, | ||
}; | ||
use dashmap::DashMap; | ||
use moka::sync::Cache; | ||
use std::collections::HashMap; | ||
use std::sync::atomic::{AtomicI64, Ordering}; | ||
use std::sync::{Arc, Mutex}; | ||
use std::future::Future; | ||
use std::sync::atomic::{AtomicBool, AtomicI64, Ordering}; | ||
use std::sync::Arc; | ||
use std::time::{Duration, SystemTime}; | ||
use tokio::select; | ||
use tokio::sync::Notify; | ||
|
@@ -20,46 +22,91 @@ pub struct CachedCounterValue { | |
} | ||
|
||
pub struct Batcher { | ||
updates: Mutex<HashMap<Counter, Arc<CachedCounterValue>>>, | ||
updates: DashMap<Counter, Arc<CachedCounterValue>>, | ||
notifier: Notify, | ||
interval: Duration, | ||
priority_flush: AtomicBool, | ||
} | ||
|
||
impl Batcher { | ||
fn new(period: Duration) -> Self { | ||
Self { | ||
updates: Mutex::new(Default::default()), | ||
updates: Default::default(), | ||
notifier: Default::default(), | ||
interval: period, | ||
priority_flush: AtomicBool::new(false), | ||
} | ||
} | ||
|
||
pub fn is_empty(&self) -> bool { | ||
self.updates.lock().unwrap().is_empty() | ||
self.updates.is_empty() | ||
} | ||
|
||
pub async fn consume(&self, min: usize) -> HashMap<Counter, Arc<CachedCounterValue>> { | ||
pub async fn consume<F, Fut, O>(&self, min: usize, consumer: F) -> O | ||
where | ||
F: FnOnce(HashMap<Counter, Arc<CachedCounterValue>>) -> Fut, | ||
Fut: Future<Output = O>, | ||
{ | ||
let mut interval = interval(self.interval); | ||
let mut ready = self.updates.lock().unwrap().len() >= min; | ||
let mut ready = self.updates.len() >= min; | ||
loop { | ||
if ready { | ||
return self.consume_all(); | ||
let mut batch = Vec::with_capacity(min); | ||
let mut probably_fake = Vec::with_capacity(min); | ||
for entry in &self.updates { | ||
if entry.value().value.ttl() < self.interval { | ||
batch.push(entry.key().clone()); | ||
if batch.len() == min { | ||
break; | ||
} | ||
} | ||
if entry.value().expiry.duration() == Duration::from_secs(entry.key().seconds()) | ||
alexsnaps marked this conversation as resolved.
Show resolved
Hide resolved
|
||
{ | ||
probably_fake.push(entry.key().clone()); | ||
if probably_fake.len() == min { | ||
break; | ||
} | ||
} | ||
} | ||
if let Some(remaining) = min.checked_sub(batch.len()) { | ||
let take = probably_fake.into_iter().take(remaining); | ||
batch.append(&mut take.collect()); | ||
} | ||
if let Some(remaining) = min.checked_sub(batch.len()) { | ||
let take = self.updates.iter().take(remaining); | ||
batch.append(&mut take.map(|e| e.key().clone()).collect()); | ||
} | ||
let mut result = HashMap::new(); | ||
for counter in &batch { | ||
let value = self.updates.get(counter).unwrap().clone(); | ||
result.insert(counter.clone(), value); | ||
} | ||
let result = consumer(result).await; | ||
for counter in &batch { | ||
self.updates | ||
.remove_if(counter, |_, v| v.no_pending_writes()); | ||
} | ||
Comment on lines
+75
to
+78
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If after having flushed to redis no additional pending writes were added, we can remove these entries from the queue. |
||
return result; | ||
} else { | ||
ready = select! { | ||
_ = self.notifier.notified() => self.updates.lock().unwrap().len() >= min, | ||
_ = self.notifier.notified() => { | ||
self.updates.len() >= min || | ||
self.priority_flush | ||
.compare_exchange(true, false, Ordering::Release, Ordering::Acquire) | ||
.is_ok() | ||
}, | ||
_ = interval.tick() => true, | ||
} | ||
} | ||
} | ||
} | ||
|
||
pub fn consume_all(&self) -> HashMap<Counter, Arc<CachedCounterValue>> { | ||
let mut batch = self.updates.lock().unwrap(); | ||
std::mem::take(&mut *batch) | ||
} | ||
|
||
pub fn add(&self, counter: Counter, value: Arc<CachedCounterValue>) { | ||
self.updates.lock().unwrap().entry(counter).or_insert(value); | ||
pub fn add(&self, counter: Counter, value: Arc<CachedCounterValue>, priority: bool) { | ||
let priority = priority || value.value.ttl() < self.interval; | ||
self.updates.entry(counter).or_insert(value); | ||
if priority { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. if we start constraining the batch size sent to Redis, we may want to add the priority flag in the value too so that the first batch we execute contains the priority items first. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yeah, I do this "hack" where I check the TTL on entry… to decide the things that are "important", but I think I should maybe just waste the additional bit… |
||
self.priority_flush.store(true, Ordering::Release); | ||
} | ||
self.notifier.notify_one(); | ||
} | ||
} | ||
|
@@ -140,6 +187,12 @@ impl CachedCounterValue { | |
} | ||
} | ||
|
||
fn no_pending_writes(&self) -> bool { | ||
let start = self.initial_value.load(Ordering::SeqCst); | ||
let value = self.value.value_at(SystemTime::now()); | ||
value - start == 0 | ||
} | ||
|
||
pub fn hits(&self, _: &Counter) -> i64 { | ||
self.value.value_at(SystemTime::now()) | ||
} | ||
|
@@ -240,15 +293,17 @@ impl CountersCache { | |
} | ||
|
||
pub fn increase_by(&self, counter: &Counter, delta: i64) { | ||
let mut priority = false; | ||
let val = self.cache.get_with_by_ref(counter, || { | ||
priority = true; | ||
Arc::new( | ||
// this TTL is wrong, it needs to be the cache's TTL, not the time window of our limit | ||
// todo fix when introducing the Batcher type! | ||
CachedCounterValue::from(counter, 0, Duration::from_secs(counter.seconds())), | ||
) | ||
}); | ||
val.delta(counter, delta); | ||
self.batcher.add(counter.clone(), val.clone()); | ||
self.batcher.add(counter.clone(), val.clone(), priority); | ||
} | ||
|
||
fn ttl_from_redis_ttl( | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
DashMap makes it much cleaner and easier to consume <3