diff --git a/src/api.rs b/src/api.rs index e4e8f6f..23273dc 100644 --- a/src/api.rs +++ b/src/api.rs @@ -217,7 +217,7 @@ fn mime_for_extension(extension: &str) -> &'static str { impl EngineHandle { async fn ask(&self, query: EngineQuery) -> Result { - let read_handle = self.reader_factory.handle(); + let read_handle = self.reader.handle(); if let Some(sort_by) = query.sort_by() { let needs_build = match read_handle.enter() { Some(guard) => !guard.has_sort_helper_index(sort_by), diff --git a/src/engine.rs b/src/engine.rs index a008953..40bde36 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -149,14 +149,53 @@ impl Absorb for EngineSnapshot { } } +struct IndexWriter { + handle: left_right::WriteHandle, +} + +impl IndexWriter { + fn mutate(&mut self, op: impl FnMut(&mut EngineSnapshot) + Send + 'static) { + self.handle.append(EngineOp(Box::new(op))); + self.handle.publish(); + } + + fn peek(&mut self) -> Option> { + self.handle.enter() + } +} + +pub struct IndexReadHandle(left_right::ReadHandle); + +impl IndexReadHandle { + pub(crate) fn enter(&self) -> Option> { + self.0.enter() + } +} + +pub struct IndexReader { + factory: left_right::ReadHandleFactory, +} + +impl Clone for IndexReader { + fn clone(&self) -> Self { + IndexReader { factory: self.factory.clone() } + } +} + +impl IndexReader { + pub(crate) fn handle(&self) -> IndexReadHandle { + IndexReadHandle(self.factory.handle()) + } +} + pub struct EngineHandle { pub tx: mpsc::Sender, - pub reader_factory: left_right::ReadHandleFactory, + pub reader: IndexReader, } impl Clone for EngineHandle { fn clone(&self) -> Self { - EngineHandle { tx: self.tx.clone(), reader_factory: self.reader_factory.clone() } + EngineHandle { tx: self.tx.clone(), reader: self.reader.clone() } } } @@ -172,23 +211,19 @@ impl EngineHandle { pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle { let (tx, rx) = mpsc::channel(channel_capacity); let (write, read) = left_right::new::(); - let reader_factory = read.factory(); - tokio::spawn(run(rx, settings, write)); - EngineHandle { tx, reader_factory } + let reader = IndexReader { factory: read.factory() }; + tokio::spawn(run(rx, settings, IndexWriter { handle: write })); + EngineHandle { tx, reader } } #[cfg(test)] pub(crate) fn test_handle(channel_capacity: usize) -> (EngineHandle, mpsc::Receiver) { let (tx, rx) = mpsc::channel(channel_capacity); let (_write, read) = left_right::new::(); - (EngineHandle { tx, reader_factory: read.factory() }, rx) + (EngineHandle { tx, reader: IndexReader { factory: read.factory() } }, rx) } -async fn run( - mut rx: mpsc::Receiver, - settings: SettingsHandle, - mut write: left_right::WriteHandle, -) { +async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mut write: IndexWriter) { let mut max_logs: Option = None; let mut metric_sample_sequence: u64 = 0; let mut alert_states: HashMap = HashMap::new(); @@ -218,7 +253,7 @@ async fn run( }; let text_for_op = text.clone(); - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { snap.approx_bytes_total += approx_log_bytes(&text_for_op); snap.index.insert( IndexKey::IngestedLog { sequence, timestamp }, @@ -234,11 +269,10 @@ async fn run( snap.purge_stale_sort_helper_indexes(); snap.evict_excess_logs(max_logs); snap.ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms; - }))); - write.publish(); + }); let fired = write - .enter() + .peek() .map(|guard| { alerts::evaluate_alerts(&rules, &mut alert_states, &guard, sequence, timestamp, &text) }) @@ -257,109 +291,100 @@ async fn run( let sequence = metric_sample_sequence; let timestamp = sample.timestamp; let value = MetricPrimitive::Number(Number::Float(sample.value)); - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { snap.push_bucket_value(&metric_name, labels.clone(), sequence, timestamp, value.clone()); - }))); - write.publish(); + }); metric_sample_sequence += 1; } EngineMessage::SetMaxLogs(new_max) => { max_logs = new_max; - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { snap.evict_excess_logs(new_max); - }))); - write.publish(); + }); } EngineMessage::DeclareParam { request, respond_to } => { let result = write - .enter() + .peek() .map(|guard| guard.validate_param_declaration(&request)) .unwrap_or_else(|| Err("engine is shutting down".to_string())); let succeeded = result.is_ok(); let name = request.name.clone(); let request_for_op = request.clone(); - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { snap.sort_helper_indexes.remove(&name); if succeeded { snap.insert_param_declaration(request_for_op.clone()); } - }))); - write.publish(); + }); let _ = respond_to.send(result); } EngineMessage::RemoveParamDeclaration { name, respond_to } => { let existed = write - .enter() + .peek() .is_some_and(|guard| guard.index.contains_key(&IndexKey::ParamDeclaration { name: name.clone() })); let name_for_op = name.clone(); - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() }); snap.sort_helper_indexes.remove(&name_for_op); - }))); - write.publish(); + }); let _ = respond_to.send(existed); } EngineMessage::ClearParams { respond_to } => { let mut scratch = EngineSnapshot::from_index( - write.enter().map(|guard| guard.index.clone()).unwrap_or_default(), + write.peek().map(|guard| guard.index.clone()).unwrap_or_default(), ); let removed = scratch.clear_all_param_declarations(); let (reextracted, logs_considered) = scratch.reextract_params_from_retained_logs(); - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { snap.clear_all_param_declarations(); snap.reextract_params_from_retained_logs(); snap.sort_helper_indexes.clear(); - }))); - write.publish(); + }); let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered }); } EngineMessage::DeclareMetric { request, respond_to } => { let result = write - .enter() + .peek() .map(|guard| guard.validate_metric_request(&request).map(|_| ())) .unwrap_or_else(|| Err("engine is shutting down".to_string())); let succeeded = result.is_ok(); let request_for_op = request.clone(); - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { if succeeded { let _ = snap.declare_metric(request_for_op.clone()); } - }))); - write.publish(); + }); let _ = respond_to.send(result); } EngineMessage::UpdateMetric { name, request, respond_to } => { let result = write - .enter() + .peek() .map(|guard| guard.validate_update_metric(&name, &request).map(|_| ())) .unwrap_or_else(|| Err("engine is shutting down".to_string())); let succeeded = result.is_ok(); let existing_name = name.clone(); let request_for_op = request.clone(); - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { if succeeded { let _ = snap.update_metric(&existing_name, request_for_op.clone()); } - }))); - write.publish(); + }); let _ = respond_to.send(result); } EngineMessage::RemoveMetricDeclaration { name, respond_to } => { let existed = write - .enter() + .peek() .is_some_and(|guard| guard.index.contains_key(&IndexKey::MetricDeclaration { name: name.clone() })); let name_for_op = name.clone(); - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { snap.remove_metric_declaration(&name_for_op); - }))); - write.publish(); + }); let _ = respond_to.send(existed); } EngineMessage::EnsureSortHelperIndex { param_name, respond_to } => { - write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + write.mutate(move |snap: &mut EngineSnapshot| { snap.ensure_sort_helper_index_built(¶m_name); - }))); - write.publish(); + }); let _ = respond_to.send(()); } EngineMessage::Barrier(respond_to) => { diff --git a/src/profile.rs b/src/profile.rs index 2267e50..c71f83e 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -708,7 +708,7 @@ mod tests { } async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse { - let read_handle = engine.reader_factory.handle(); + let read_handle = engine.reader.handle(); let guard = read_handle.enter().expect("engine unavailable"); guard.handle_query_ro(query) }