Skip to content
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.

Sign up for GitHub

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

Jump to bottom

fix(buffer): Cycle through unready projects #3968

Merged
merged 3 commits into from
Aug 30, 2024
Merged

fix(buffer): Cycle through unready projects #3968

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3c5e16c
fix(buffer): Cycle through unready projects
jjbayer Aug 30, 2024
fa0eae6
sleep
jjbayer Aug 30, 2024
b3035bc
fix: Reset sleep from handle_message
jjbayer Aug 30, 2024
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
3 changes: 0 additions & 3 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,6 @@
- Add `EnvelopeStore` trait and implement `DiskUsage` for tracking disk usage. ([#3925](https://github.com/getsentry/relay/pull/3925))
- Increase replay recording limit to two hours. ([#3961](https://github.com/getsentry/relay/pull/3961))
- Make EnvelopeBuffer a Service. ([#3965](https://github.com/getsentry/relay/pull/3965))

**Internal**:

- No longer send COGS data to dedicated Kafka topic. ([#3953](https://github.com/getsentry/relay/pull/3953))

## 24.8.0
Expand Down
100 changes: 60 additions & 40 deletions relay-server/src/services/buffer/envelope_buffer/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,18 @@ impl PolymorphicEnvelopeBuffer {
}
}

/// Marks a stack as seen.
///
/// Non-ready stacks are deprioritized when they are marked as seen, such that
/// the next call to `.peek()` will look at a different stack. This prevents
/// head-of-line blocking.
pub fn mark_seen(&mut self, stack_key: &StackKey) {
match self {
Self::Sqlite(buffer) => buffer.mark_seen(stack_key),
Self::InMemory(buffer) => buffer.mark_seen(stack_key),
}
}

/// Returns `true` whether the buffer has capacity to accept new [`Envelope`]s.
pub fn has_capacity(&self) -> bool {
match self {
Expand Down Expand Up @@ -179,7 +191,7 @@ where
pub async fn peek(&mut self) -> Result<Peek, EnvelopeBufferError> {
let Some((
QueueItem {
key: _,
key: stack_key,
value: stack,
},
Priority { readiness, .. },
Expand All @@ -193,7 +205,7 @@ where
Ok(match (stack.peek().await?, ready) {
(None, _) => Peek::Empty,
(Some(envelope), true) => Peek::Ready(envelope),
(Some(envelope), false) => Peek::NotReady(envelope),
(Some(envelope), false) => Peek::NotReady(*stack_key, envelope),
})
}

Expand Down Expand Up @@ -256,6 +268,17 @@ where
changed
}

/// Marks a stack as seen.
///
/// Non-ready stacks are deprioritized when they are marked as seen, such that
/// the next call to `.peek()` will look at a different stack. This prevents
/// head-of-line blocking.
pub fn mark_seen(&mut self, stack_key: &StackKey) {
self.priority_queue.change_priority_by(stack_key, |stack| {
stack.last_peek = Instant::now();
});
}

fn push_stack(&mut self, envelope: Box<Envelope>) {
let received_at = envelope.meta().start_time();
let stack_key = StackKey::from_envelope(&envelope);
Expand Down Expand Up @@ -302,7 +325,7 @@ where
pub enum Peek<'a> {
Empty,
Ready(&'a Envelope),
NotReady(&'a Envelope),
NotReady(StackKey, &'a Envelope),
}

#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq, PartialOrd, Ord)]
Expand Down Expand Up @@ -457,7 +480,7 @@ mod tests {
fn envelope(&self) -> Option<&Envelope> {
match self {
Peek::Empty => None,
Peek::Ready(envelope) | Peek::NotReady(envelope) => Some(envelope),
Peek::Ready(envelope) | Peek::NotReady(_, envelope) => Some(envelope),
}
}
}
Expand Down Expand Up @@ -793,7 +816,6 @@ mod tests {
assert_eq!(buffer.priority_queue.len(), 2);
}

#[ignore = "last_peek disabled for now"]
#[tokio::test]
async fn test_last_peek_internal_order() {
let mut buffer = EnvelopeBuffer::<MemoryStackProvider>::new(mock_memory_checker());
Expand All @@ -804,44 +826,42 @@ mod tests {

let project_key_2 = ProjectKey::parse("b56ae32be2584e0bbd7a4cbb95971fed").unwrap();
let event_id_2 = EventId::new();
let mut envelope2 = new_envelope(project_key_2, None, Some(event_id_2));
envelope2.set_start_time(envelope1.meta().start_time());
let envelope2 = new_envelope(project_key_2, None, Some(event_id_2));

buffer.push(envelope1).await.unwrap();
buffer.push(envelope2).await.unwrap();

assert_eq!(
buffer
.peek()
.await
.unwrap()
.envelope()
.unwrap()
.event_id()
.unwrap(),
event_id_1
);
assert_eq!(
buffer
.peek()
.await
.unwrap()
.envelope()
.unwrap()
.event_id()
.unwrap(),
event_id_2
);
assert_eq!(
buffer
.peek()
.await
.unwrap()
.envelope()
.unwrap()
.event_id()
.unwrap(),
event_id_1
);
// event_id_1 is first element:
let Peek::NotReady(_, envelope) = buffer.peek().await.unwrap() else {
panic!();
};
assert_eq!(envelope.event_id(), Some(event_id_1));

// Second peek returns same element:
let Peek::NotReady(stack_key, envelope) = buffer.peek().await.unwrap() else {
panic!();
};
assert_eq!(envelope.event_id(), Some(event_id_1));

buffer.mark_seen(&stack_key);

// After mark_seen, event 2 is on top:
let Peek::NotReady(_, envelope) = buffer.peek().await.unwrap() else {
panic!();
};
assert_eq!(envelope.event_id(), Some(event_id_2));

let Peek::NotReady(stack_key, envelope) = buffer.peek().await.unwrap() else {
panic!();
};
assert_eq!(envelope.event_id(), Some(event_id_2));

buffer.mark_seen(&stack_key);

// After another mark_seen, cycle back to event 1:
let Peek::NotReady(_, envelope) = buffer.peek().await.unwrap() else {
panic!();
};
assert_eq!(envelope.event_id(), Some(event_id_1));
}
}
41 changes: 18 additions & 23 deletions relay-server/src/services/buffer/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
//! Types for buffering envelopes.

use std::future;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;

use relay_base_schema::project::ProjectKey;
use relay_config::Config;
Expand Down Expand Up @@ -83,9 +83,10 @@ pub struct EnvelopeBufferService {
buffer: PolymorphicEnvelopeBuffer,
project_cache: Addr<ProjectCache>,
has_capacity: Arc<AtomicBool>,
changes: bool,
}

const DEFAULT_SLEEP: Duration = Duration::from_millis(100);

impl EnvelopeBufferService {
/// Creates a memory or disk based [`EnvelopeBufferService`], depending on the given config.
///
Expand All @@ -100,7 +101,6 @@ impl EnvelopeBufferService {
buffer: PolymorphicEnvelopeBuffer::from_config(config, memory_checker),
project_cache,
has_capacity: Arc::new(AtomicBool::new(true)),
changes: true,
})
}

Expand All @@ -113,23 +113,16 @@ impl EnvelopeBufferService {
}
}

/// Return immediately if changes were flagged, otherwise sleep forever.
///
/// NOTE: This function sleeps indefinitely if no changes were flagged.
/// Only use in combination with [`tokio::select!`].
async fn wait_for_changes(&mut self) {
if !self.changes {
let _: () = future::pending().await; // wait until cancelled
}
}

/// Tries to pop an envelope for a ready project.
async fn try_pop(&mut self) -> Result<(), EnvelopeBufferError> {
///
/// Returns the amount of time we should wait until next pop
async fn try_pop(&mut self) -> Result<Duration, EnvelopeBufferError> {
relay_log::trace!("EnvelopeBufferService peek");
let next_sleep;
match self.buffer.peek().await? {
Peek::Empty => {
relay_log::trace!("EnvelopeBufferService empty");
self.changes = false;
next_sleep = DEFAULT_SLEEP
}
Peek::Ready(_) => {
relay_log::trace!("EnvelopeBufferService pop");
Expand All @@ -139,9 +132,9 @@ impl EnvelopeBufferService {
.await?
.expect("Element disappeared despite exclusive excess");
self.project_cache.send(DequeuedEnvelope(envelope));
self.changes = true;
next_sleep = Duration::ZERO; // try next pop immediately
}
Peek::NotReady(envelope) => {
Peek::NotReady(stack_key, envelope) => {
relay_log::trace!("EnvelopeBufferService request update");
let project_key = envelope.meta().public_key();
self.project_cache.send(UpdateProject(project_key));
Expand All @@ -152,10 +145,12 @@ impl EnvelopeBufferService {
self.project_cache.send(UpdateProject(sampling_key));
}
}
self.changes = false;
// deprioritize the stack to prevent head-of-line blocking
self.buffer.mark_seen(&stack_key);
next_sleep = DEFAULT_SLEEP;
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now that we cycle through unready stacks, we should not wait forever or until a new message comes along. Instead, we only sleep for a short default interval. As before, the sleep gets canceled when a new message comes in.

}
}
Ok(())
Ok(next_sleep)
}

async fn handle_message(&mut self, message: EnvelopeBuffer) {
Expand All @@ -179,7 +174,6 @@ impl EnvelopeBufferService {
self.buffer.mark_ready(&project_key, true);
}
};
self.changes = true;
}

async fn push(&mut self, envelope: Box<Envelope>) {
Expand All @@ -203,20 +197,21 @@ impl Service for EnvelopeBufferService {
fn spawn_handler(mut self, mut rx: Receiver<Self::Interface>) {
tokio::spawn(async move {
relay_log::info!("EnvelopeBufferService start");
let mut sleep = Duration::ZERO;
loop {
relay_log::trace!("EnvelopeBufferService loop");
tokio::select! {
// NOTE: we do not select a bias here.
// On the one hand, we might want to prioritize dequeing over enqueing
// so we do not exceed the buffer capacity by starving the dequeue.
// on the other hand, prioritizing old messages violates the LIFO design.
() = self.wait_for_changes() => {
if let Err(e) = self.try_pop().await {
() = tokio::time::sleep(sleep) => {
sleep = self.try_pop().await.map_err(|e| {
relay_log::error!(
error = &e as &dyn std::error::Error,
"failed to pop envelope"
);
}
}).unwrap_or(DEFAULT_SLEEP);
}
Some(message) = rx.recv() => {
self.handle_message(message).await;
Expand Down
Loading