feat: Add support for flushing the event buffer #92
Conversation
CrabNejonas
requested review from
lucasfernog-crabnebula,
david-crabnebula,
atila-crabnebula,
tejas-crabnebula and
johann-crabnebula
as code owners
DR-591 fix: even buffer flushing and backpressue
Right now the event buffers will be emptied every 200ms. Events aren't uniformly distributed across time obviously though, instead often coming in bursts. Right now we miss a lot of data because of that. We should introduce some way for the aggregator to flush its buffers when they get too full. |
| @@ -310,7 +314,7 @@ impl<T, const CAP: usize> EventBuf<T, CAP> { | |||
| // TODO does it really make sense to track the dropped events here? | |||
There was a problem hiding this comment.
are we still tracking dropped events here? 😂
| @@ -51,6 +51,11 @@ impl Layer { | |||
| dropped.fetch_add(1, Ordering::Release); | |||
| } | |||
| } | |||
|
|
|||
| let capacity = self.tx.capacity(); | |||
There was a problem hiding this comment.
are you sure we need to check the capacity and not the len?
There was a problem hiding this comment.
Not really, the idea is to catch flush when the remaining capacity in the buffer drops below half. Also len doesn't exist on tokio::sync::mpsc::Sender 😉
| @@ -83,6 +83,10 @@ impl Aggregator { | |||
| loop { | |||
| let should_publish = tokio::select! { | |||
| _ = interval.tick() => true, | |||
| _ = self.shared.flush.notified() => { | |||
| tracing::debug!("event buffer approaching capacity, flushing..."); | |||
| false | |||
There was a problem hiding this comment.
shouldn't this return true instead so we publish the events?
There was a problem hiding this comment.
No, the flush notification is about draining the events of the internal event queue (from the Layer to the Aggregator) not about the buffer inside the aggregator itself, that one is actually fine.
Co-authored-by: Lucas Nogueira <[email protected]>
Co-authored-by: Lucas Nogueira <[email protected]>
This adds a way for the
Layerto wake up theAggregatorearly when there is high pressure on the internal even queue. This helps to reduce the number of missed dropped events (in my testing it actually reduces the number of dropped events under heavy load to 0 which is awesome)This also adds a console debug assertion for when we dropped some events (this only prints to the console but should be good enough for now)
resolves DR-591