diff --git a/src/arrow.rs b/src/arrow.rs index 046f020..3eca9e0 100644 --- a/src/arrow.rs +++ b/src/arrow.rs @@ -35,9 +35,8 @@ use std::time::Instant; use crate::format::{Format, FormatType}; use crate::parse::{ - Entry, LabelSpan, METRIC_NAME_LABEL, MetricType, OmScratch, OpenMetricsParser, - OpenMetricsParserState, ParseError, Parser, PromParser, PromParserState, parse_go_float, - unescape_bytes_into, + Entry, LabelSpan, METRIC_NAME_LABEL, MetricType, OpenMetricsParser, OpenMetricsParserState, + Parser, PromParser, PromParserState, parse_go_float, unescape_bytes_into, }; use self::columns::{ @@ -86,54 +85,24 @@ pub enum DecodeStep { Done, } -enum ActiveParserState { - Prom(PromParserState), - OpenMetrics(OpenMetricsParserState), +#[derive(Clone, Copy)] +enum ParserKind { + Prom, + OpenMetrics, Invalid, } -impl ActiveParserState { - #[inline] - fn into_parser(self, input: &[u8]) -> ActiveParser<'_> { - match self { - Self::Prom(state) => ActiveParser::Prom(PromParser::from_state(input, state)), - Self::OpenMetrics(state) => { - ActiveParser::OpenMetrics(OpenMetricsParser::from_state(input, state)) - } - Self::Invalid => ActiveParser::Invalid, - } - } - - fn take_scratch_into(&mut self, decoder: &mut ArrowDecoder) { - match std::mem::replace(self, Self::Invalid) { - Self::Prom(mut state) => { - let scratch = state.take_scratch(); - if scratch.capacity() > decoder.prom_spans.capacity() { - decoder.prom_spans = scratch; - } - } - Self::OpenMetrics(mut state) => { - let scratch = state.take_scratch(); - if scratch.retained_memory_bytes() > decoder.om_scratch.retained_memory_bytes() { - decoder.om_scratch = scratch; - } - } - Self::Invalid => {} - } - } -} - /// Lifetime-free incremental decode cursor for driving [`ArrowDecoder::step_decode`] across /// async suspension points without borrowing the input slice. /// /// Created by [`ArrowDecoder::start_decode`] (single-batch mode) or /// [`ArrowDecoder::start_decode_stream`] (chunked streaming mode). Because `DecodeCursor` stores -/// only byte-offset spans and parser state rather than references into the scrape payload, an -/// owned task or `otel-arrow` `PdataDecodeJob` struct can hold an owned buffer (`Bytes`, -/// `Vec`, or `Rc<[u8]>`), an [`ArrowDecoder`], and a `DecodeCursor` side by side under -/// `#![forbid(unsafe_code)]`. +/// only scalar byte offsets and options while [`ArrowDecoder`] retains the reusable parser scratch +/// buffers in place, an owned task or `otel-arrow` `PdataDecodeJob` struct can hold an owned +/// buffer (`Bytes`, `Vec`, or `Rc<[u8]>`), an [`ArrowDecoder`], and a compact `DecodeCursor` +/// side by side under `#![forbid(unsafe_code)]`. pub struct DecodeCursor { - parser_state: ActiveParserState, + parser_kind: ParserKind, opts: DecodeOptions, default_ts_ns: Option, max_data_points: usize, @@ -171,122 +140,73 @@ impl DecodeCursor { } } -enum ActiveParser<'input> { - Prom(PromParser<'input>), - OpenMetrics(OpenMetricsParser<'input>), - Invalid, +// Internal parser specialization trait so `ingest_entries` monomorphizes directly over +// `PromParser` and `OpenMetricsParser` without per-entry enum dispatch or unused stack state. +trait ScrapeParser<'input>: Parser<'input> { + const IS_OPENMETRICS: bool; + + fn parser_pos(&self) -> usize; + fn name_offset(&self) -> usize; + fn raw_help(&self) -> (&'input [u8], &'input [u8]); + fn raw_series_parts(&self) -> (&'input [u8], &[LabelSpan], &'input [u8], Option, f64); + fn raw_exemplar_parts(&self) -> Option<(f64, Option, &[LabelSpan])>; } -impl<'input> ActiveParser<'input> { +impl<'input> ScrapeParser<'input> for PromParser<'input> { + const IS_OPENMETRICS: bool = false; + #[inline] - fn is_openmetrics(&self) -> bool { - matches!(self, Self::OpenMetrics(_)) + fn parser_pos(&self) -> usize { + self.pos() } #[inline] - fn pos(&self) -> usize { - match self { - Self::Prom(p) => p.pos(), - Self::OpenMetrics(p) => p.pos(), - Self::Invalid => 0, - } + fn name_offset(&self) -> usize { + PromParser::name_offset(self) } #[inline] - fn into_state(self) -> ActiveParserState { - match self { - Self::Prom(p) => ActiveParserState::Prom(p.into_state()), - Self::OpenMetrics(p) => ActiveParserState::OpenMetrics(p.into_state()), - Self::Invalid => ActiveParserState::Invalid, - } + fn raw_help(&self) -> (&'input [u8], &'input [u8]) { + PromParser::raw_help(self) } #[inline] - fn next_entry(&mut self) -> Result, ParseError> { - match self { - Self::Prom(p) => p.next_entry(), - Self::OpenMetrics(p) => p.next_entry(), - Self::Invalid => Ok(None), - } + fn raw_series_parts(&self) -> (&'input [u8], &[LabelSpan], &'input [u8], Option, f64) { + PromParser::raw_series_parts(self) } #[inline] - fn name_offset(&self) -> usize { - match self { - Self::Prom(p) => p.name_offset(), - Self::OpenMetrics(p) => p.name_offset(), - Self::Invalid => 0, - } + fn raw_exemplar_parts(&self) -> Option<(f64, Option, &[LabelSpan])> { + None } +} + +impl<'input> ScrapeParser<'input> for OpenMetricsParser<'input> { + const IS_OPENMETRICS: bool = true; #[inline] - fn raw_help(&self) -> (&'input [u8], &'input [u8]) { - match self { - Self::Prom(p) => p.raw_help(), - Self::OpenMetrics(p) => p.raw_help(), - Self::Invalid => (b"", b""), - } + fn parser_pos(&self) -> usize { + self.pos() } #[inline] - fn metric_type(&self) -> (&[u8], MetricType) { - match self { - Self::Prom(p) => p.metric_type(), - Self::OpenMetrics(p) => p.metric_type(), - Self::Invalid => (b"", MetricType::Unknown), - } + fn name_offset(&self) -> usize { + OpenMetricsParser::name_offset(self) } #[inline] - fn unit(&self) -> (&[u8], &[u8]) { - match self { - Self::Prom(p) => p.unit(), - Self::OpenMetrics(p) => p.unit(), - Self::Invalid => (b"", b""), - } + fn raw_help(&self) -> (&'input [u8], &'input [u8]) { + OpenMetricsParser::raw_help(self) } #[inline] fn raw_series_parts(&self) -> (&'input [u8], &[LabelSpan], &'input [u8], Option, f64) { - match self { - Self::Prom(p) => p.raw_series_parts(), - Self::OpenMetrics(p) => p.raw_series_parts(), - Self::Invalid => (b"", &[], b"", None, 0.0), - } + OpenMetricsParser::raw_series_parts(self) } #[inline] fn raw_exemplar_parts(&self) -> Option<(f64, Option, &[LabelSpan])> { - match self { - Self::Prom(_) | Self::Invalid => None, - Self::OpenMetrics(p) => p.raw_exemplar_parts(), - } - } - - #[inline] - fn start_timestamp(&mut self) -> Option { - match self { - Self::Prom(_) | Self::Invalid => None, - Self::OpenMetrics(p) => p.start_timestamp(), - } - } - - fn take_scratch_into(&mut self, decoder: &mut ArrowDecoder) { - match self { - Self::Prom(p) => { - let scratch = p.take_scratch(); - if scratch.capacity() > decoder.prom_spans.capacity() { - decoder.prom_spans = scratch; - } - } - Self::OpenMetrics(p) => { - let scratch = p.take_scratch(); - if scratch.retained_memory_bytes() > decoder.om_scratch.retained_memory_bytes() { - decoder.om_scratch = scratch; - } - } - Self::Invalid => {} - } + OpenMetricsParser::raw_exemplar_parts(self) } } @@ -298,8 +218,8 @@ impl<'input> ActiveParser<'input> { /// Arrow `Buffer`s returned inside [`OtapMetricRecords`]. pub struct ArrowDecoder { scrape_id: u64, - prom_spans: Vec, - om_scratch: OmScratch, + prom_state: PromParserState, + om_state: OpenMetricsParserState, unescape_buf: Vec, series_name_buf: Vec, current_family: CurrentFamily, @@ -342,8 +262,8 @@ impl ArrowDecoder { pub fn with_capacity(metrics: usize, data_points: usize, attrs: usize) -> Self { Self { scrape_id: 0, - prom_spans: Vec::with_capacity(16), - om_scratch: OmScratch::default(), + prom_state: PromParserState::with_capacity(16), + om_state: OpenMetricsParserState::default(), unescape_buf: Vec::new(), series_name_buf: Vec::with_capacity(64), current_family: CurrentFamily::with_capacity(64, 128), @@ -364,8 +284,8 @@ impl ArrowDecoder { /// reusable scratch structures. #[must_use] pub fn retained_memory_bytes(&self) -> usize { - self.prom_spans.capacity() * size_of::() - + self.om_scratch.retained_memory_bytes() + self.prom_state.retained_memory_bytes() + + self.om_state.retained_memory_bytes() + self.unescape_buf.capacity() + self.series_name_buf.capacity() + self.current_family.retained_memory_bytes() @@ -410,20 +330,51 @@ impl ArrowDecoder { opts: &DecodeOptions, ) -> Result { self.reset_scrape(); - let (mut parser, default_ts_ns) = self.init_scrape_parser(input, format, opts)?; - + let default_ts_ns = match opts.timestamp_ms { + Some(ms) => Some(ms_to_ns(ms)?), + None => None, + }; let mut pending_series = false; let mut done = false; - let res = self.ingest_entries::( - &mut parser, - opts, - default_ts_ns, - 0, - None, - &mut pending_series, - &mut done, - ); - parser.take_scratch_into(self); + let res = match format.format_type() { + FormatType::TextPlain => { + let spans = std::mem::take(&mut self.prom_state.labels); + let mut parser = PromParser::with_scratch(input, spans); + let res = self.ingest_entries::<_, false, false>( + &mut parser, + input.len(), + opts, + default_ts_ns, + 0, + None, + &mut pending_series, + &mut done, + ); + self.prom_state.labels = parser.take_scratch(); + res + } + FormatType::OpenMetrics if !format.version().starts_with("2.") => { + // TODO(bwplotka): Dispatch OpenMetrics 2.0 in Phase 3. + let scratch = self.om_state.take_scratch(); + let mut parser = OpenMetricsParser::with_scratch(input, scratch); + let res = self.ingest_entries::<_, false, false>( + &mut parser, + input.len(), + opts, + default_ts_ns, + 0, + None, + &mut pending_series, + &mut done, + ); + self.om_state.put_scratch(parser.take_scratch()); + res + } + _ => { + // TODO(bwplotka): Dispatch Proto* in Phase 3. + return Err(DecodeError::UnsupportedFormat(format.as_str().to_owned())); + } + }; match res? { DecodeStep::Batch(records) => Ok(records), DecodeStep::Done => Ok(OtapMetricRecords::default()), @@ -525,14 +476,14 @@ impl ArrowDecoder { /// `step_decode` call for a given `cursor`, because the cursor and staged parser state store /// byte offsets into `input`. /// - /// Internal parser scratch buffers are automatically returned to `self` once `cursor` reaches - /// end-of-input or encounters an error. If `cursor` is abandoned early before completion, - /// call [`ArrowDecoder::finish_decode`] to reclaim its parser scratch buffers. + /// If `cursor` is abandoned early before completion, call [`ArrowDecoder::finish_decode`] to + /// reset partial scrape state in `self` while preserving internal scratch capacities. /// /// # Errors /// /// Returns [`DecodeError`] if the exposition payload fails validation or an OTAP table exceeds /// its integer ID width. + #[inline] pub fn step_decode( &mut self, cursor: &mut DecodeCursor, @@ -553,7 +504,7 @@ impl ArrowDecoder { } if cursor.scrape_id != self.scrape_id { cursor.done = true; - cursor.parser_state.take_scratch_into(self); + cursor.parser_kind = ParserKind::Invalid; return Err(DecodeError::InvalidCursor( "cursor does not belong to current decoder scrape", )); @@ -565,11 +516,74 @@ impl ArrowDecoder { "scrape input buffer length changed across decode steps", )); } - let state = std::mem::replace(&mut cursor.parser_state, ActiveParserState::Invalid); - let mut parser = state.into_parser(input); - let res = if cursor.chunked { - self.ingest_entries::( - &mut parser, + match cursor.parser_kind { + ParserKind::Prom => { + let state = std::mem::take(&mut self.prom_state); + let mut parser = PromParser::from_state(input, state); + let res = + self.run_cursor_step::<_, BUDGETED>(&mut parser, input.len(), cursor, budget); + cursor.pos = parser.parser_pos(); + self.prom_state = parser.into_state(); + match res { + Ok(step) => { + if cursor.done { + cursor.parser_kind = ParserKind::Invalid; + self.prom_state.reset(); + } + Ok(step) + } + Err(err) => { + cursor.done = true; + cursor.parser_kind = ParserKind::Invalid; + self.prom_state.reset(); + self.reset_scrape(); + Err(err) + } + } + } + ParserKind::OpenMetrics => { + let state = std::mem::take(&mut self.om_state); + let mut parser = OpenMetricsParser::from_state(input, state); + let res = + self.run_cursor_step::<_, BUDGETED>(&mut parser, input.len(), cursor, budget); + cursor.pos = parser.parser_pos(); + self.om_state = parser.into_state(); + match res { + Ok(step) => { + if cursor.done { + cursor.parser_kind = ParserKind::Invalid; + self.om_state.reset(); + } + Ok(step) + } + Err(err) => { + cursor.done = true; + cursor.parser_kind = ParserKind::Invalid; + self.om_state.reset(); + self.reset_scrape(); + Err(err) + } + } + } + ParserKind::Invalid => { + cursor.done = true; + Ok(DecodeStep::Done) + } + } + } + + #[inline] + fn run_cursor_step<'input, P: ScrapeParser<'input>, const BUDGETED: bool>( + &mut self, + parser: &mut P, + input_len: usize, + cursor: &mut DecodeCursor, + budget: Option, + ) -> Result { + if cursor.chunked { + self.ingest_entries::( + parser, + input_len, &cursor.opts, cursor.default_ts_ns, cursor.max_data_points, @@ -578,8 +592,9 @@ impl ArrowDecoder { &mut cursor.done, ) } else { - self.ingest_entries::( - &mut parser, + self.ingest_entries::( + parser, + input_len, &cursor.opts, cursor.default_ts_ns, 0, @@ -587,33 +602,18 @@ impl ArrowDecoder { &mut cursor.pending_series, &mut cursor.done, ) - }; - cursor.pos = parser.pos(); - match res { - Ok(step) => { - if cursor.done { - parser.take_scratch_into(self); - } else { - cursor.parser_state = parser.into_state(); - } - Ok(step) - } - Err(err) => { - cursor.done = true; - parser.take_scratch_into(self); - self.reset_scrape(); - Err(err) - } } } - /// Ends an active [`DecodeCursor`] and reclaims its internal parser scratch buffers into - /// `self`. + /// Ends an active [`DecodeCursor`] and resets partial scrape state in `self` while preserving + /// internal parser scratch capacities. pub fn finish_decode(&mut self, cursor: &mut DecodeCursor) { - cursor.parser_state.take_scratch_into(self); + cursor.parser_kind = ParserKind::Invalid; if !cursor.done { cursor.done = true; if cursor.scrape_id == self.scrape_id { + self.prom_state.reset(); + self.om_state.reset(); self.reset_scrape(); } } @@ -626,8 +626,8 @@ impl ArrowDecoder { /// Empty or metadata-only scrapes yield zero items. When a contiguous metric family spans a /// chunk boundary, [`ArrowDecoder`] retains its `# HELP`, `# TYPE`, and `# UNIT` metadata /// across the flush and re-emits the family row in the next chunk only if further samples - /// follow. Dropping [`DecodeStream`] early returns the internal parser scratch buffer to - /// `self`. + /// follow. Dropping [`DecodeStream`] early resets partial scrape state in `self` while + /// preserving internal scratch capacities. pub fn decode_stream<'dec, 'input>( &'dec mut self, input: &'input [u8], @@ -653,7 +653,7 @@ impl ArrowDecoder { decoder: self, input, cursor: DecodeCursor { - parser_state: ActiveParserState::Invalid, + parser_kind: ParserKind::Invalid, opts: *opts, default_ts_ns: None, max_data_points, @@ -683,47 +683,37 @@ impl ArrowDecoder { } else { 0 }; - let (parser, default_ts_ns) = self.init_scrape_parser(b"", format, opts)?; - Ok(DecodeCursor { - parser_state: parser.into_state(), - opts: *opts, - default_ts_ns, - max_data_points, - scrape_id: self.scrape_id, - input_len: None, - pos: 0, - chunked, - pending_series: false, - done: false, - }) - } - - fn init_scrape_parser<'input>( - &mut self, - input: &'input [u8], - format: &Format, - opts: &DecodeOptions, - ) -> Result<(ActiveParser<'input>, Option), DecodeError> { let default_ts_ns = match opts.timestamp_ms { Some(ms) => Some(ms_to_ns(ms)?), None => None, }; - let parser = match format.format_type() { + let parser_kind = match format.format_type() { FormatType::TextPlain => { - let spans = std::mem::take(&mut self.prom_spans); - ActiveParser::Prom(PromParser::with_scratch(input, spans)) + self.prom_state.reset(); + ParserKind::Prom } FormatType::OpenMetrics if !format.version().starts_with("2.") => { // TODO(bwplotka): Dispatch OpenMetrics 2.0 in Phase 3. - let scratch = std::mem::take(&mut self.om_scratch); - ActiveParser::OpenMetrics(OpenMetricsParser::with_scratch(input, scratch)) + self.om_state.reset(); + ParserKind::OpenMetrics } _ => { // TODO(bwplotka): Dispatch Proto* in Phase 3. return Err(DecodeError::UnsupportedFormat(format.as_str().to_owned())); } }; - Ok((parser, default_ts_ns)) + Ok(DecodeCursor { + parser_kind, + opts: *opts, + default_ts_ns, + max_data_points, + scrape_id: self.scrape_id, + input_len: None, + pos: 0, + chunked, + pending_series: false, + done: false, + }) } fn reset_scrape(&mut self) { @@ -801,15 +791,21 @@ impl ArrowDecoder { Ok(()) } - // Monomorphizing on `CHUNKED` and `BUDGETED` shares one entry dispatch loop across + // Monomorphizing on `P`, `CHUNKED`, and `BUDGETED` shares one entry dispatch loop across // synchronous single-batch `decode` (`false, false`), synchronous `decode_stream` // (`true, false`), cooperative single-batch `decode_cooperative` (`false, true`), and - // cooperative streaming `DecodeStream::step` (`true, true`), compiling out unused checks on - // each path. + // cooperative streaming `DecodeStream::step` (`true, true`), compiling out per-entry parser + // enum dispatch and unused budget/chunk checks on each path. #[allow(clippy::too_many_arguments)] - fn ingest_entries( + fn ingest_entries< + 'input, + P: ScrapeParser<'input>, + const CHUNKED: bool, + const BUDGETED: bool, + >( &mut self, - parser: &mut ActiveParser<'_>, + parser: &mut P, + input_len: usize, opts: &DecodeOptions, default_ts_ns: Option, max_data_points: usize, @@ -817,18 +813,35 @@ impl ArrowDecoder { pending_series: &mut bool, done: &mut bool, ) -> Result { - let start_pos = if BUDGETED { parser.pos() } else { 0 }; - let max_entries = if let Some(b) = budget.filter(|_| BUDGETED) { - b.max_entries.max(1) - } else { - usize::MAX - }; - let max_bytes = if let Some(b) = budget.filter(|_| BUDGETED) { - b.max_bytes.max(1) + let start_pos = if BUDGETED { parser.parser_pos() } else { 0 }; + let (max_entries, byte_limit, max_duration, check_every_entry, started) = if let Some(b) = + budget.filter(|_| BUDGETED) + { + let max_entries = b.max_entries.max(1); + let byte_limit = start_pos.saturating_add(b.max_bytes.max(1)); + let check_every = b.max_duration.is_zero(); + // Skip `Instant::now()` when the step is guaranteed to finish or hit `max_entries` + // before the first periodic duration check at `DECODE_BUDGET_DURATION_CHECK_INTERVAL`. + let need_clock = check_every + || (max_entries > DECODE_BUDGET_DURATION_CHECK_INTERVAL + && b.max_bytes.max(1) > DECODE_BUDGET_DURATION_CHECK_INTERVAL + && input_len.saturating_sub(start_pos) > DECODE_BUDGET_DURATION_CHECK_INTERVAL); + ( + max_entries, + byte_limit, + b.max_duration, + check_every, + need_clock.then(Instant::now), + ) } else { - usize::MAX + ( + usize::MAX, + usize::MAX, + std::time::Duration::ZERO, + false, + None, + ) }; - let started = if BUDGETED { Some(Instant::now()) } else { None }; let mut entries = 0usize; loop { @@ -903,17 +916,20 @@ impl ArrowDecoder { }; if CHUNKED { - let split_assembled = !self.assembly.is_empty() - && self.batch_data_points() >= max_data_points - && !self.series_matches_staged_group(parser, role); - let split_scalar = self.assembly.is_empty() - && (self.total_data_points() >= max_data_points + if self.assembly.is_empty() { + if self.total_data_points() >= max_data_points || (self.current_family.batch_id.is_none() - && self.metrics.len() >= MAX_METRICS_PER_BATCH)); - if split_assembled || split_scalar { - if split_assembled { - self.flush_pending_family(opts, default_ts_ns)?; + && self.metrics.len() >= MAX_METRICS_PER_BATCH) + { + *pending_series = true; + let batch = self.finish_batches(default_ts_ns.is_some())?; + self.reset_batch(); + return Ok(DecodeStep::Batch(batch)); } + } else if self.batch_data_points() >= max_data_points + && !self.series_matches_staged_group(parser, role) + { + self.flush_pending_family(opts, default_ts_ns)?; *pending_series = true; let batch = self.finish_batches(default_ts_ns.is_some())?; self.reset_batch(); @@ -934,13 +950,15 @@ impl ArrowDecoder { } } - if let Some(budget) = budget.filter(|_| BUDGETED) { + if BUDGETED { entries += 1; - let bytes_consumed = parser.pos().saturating_sub(start_pos); - let duration_exceeded = (budget.max_duration.is_zero() + if entries >= max_entries || parser.parser_pos() >= byte_limit { + return Ok(DecodeStep::Pending); + } + if (check_every_entry || entries.is_multiple_of(DECODE_BUDGET_DURATION_CHECK_INTERVAL)) - && started.is_some_and(|t| t.elapsed() >= budget.max_duration); - if entries >= max_entries || bytes_consumed >= max_bytes || duration_exceeded { + && started.is_some_and(|t| t.elapsed() >= max_duration) + { return Ok(DecodeStep::Pending); } } @@ -948,9 +966,9 @@ impl ArrowDecoder { } #[inline] - fn ingest_help_entry( + fn ingest_help_entry<'input, P: ScrapeParser<'input>>( &mut self, - parser: &ActiveParser<'_>, + parser: &P, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -975,8 +993,7 @@ impl ArrowDecoder { metric: self.current_family.name_string(), }); } - let unescaped = - unescape_bytes_into(raw_help, parser.is_openmetrics(), &mut self.unescape_buf); + let unescaped = unescape_bytes_into(raw_help, P::IS_OPENMETRICS, &mut self.unescape_buf); self.current_family.help.clear(); self.current_family.help.extend_from_slice(unescaped); self.current_family.has_help = true; @@ -984,9 +1001,9 @@ impl ArrowDecoder { } #[inline] - fn ingest_type_entry( + fn ingest_type_entry<'input, P: ScrapeParser<'input>>( &mut self, - parser: &ActiveParser<'_>, + parser: &P, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -1007,9 +1024,9 @@ impl ArrowDecoder { } #[inline] - fn ingest_unit_entry( + fn ingest_unit_entry<'input, P: ScrapeParser<'input>>( &mut self, - parser: &ActiveParser<'_>, + parser: &P, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -1035,9 +1052,9 @@ impl ArrowDecoder { } #[inline] - fn ingest_series_entry( + fn ingest_series_entry<'input, P: ScrapeParser<'input>>( &mut self, - parser: &mut ActiveParser<'_>, + parser: &mut P, role: SeriesRole, opts: &DecodeOptions, default_ts_ns: Option, @@ -1054,19 +1071,24 @@ impl ArrowDecoder { } #[inline] - fn append_scalar_series( + fn append_scalar_series<'input, P: ScrapeParser<'input>>( &mut self, - parser: &mut ActiveParser<'_>, + parser: &mut P, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { - let start_ts_ms = if self.current_family.metric_type == MetricType::Counter { - parser.start_timestamp() + let start_ts_ms = + if P::IS_OPENMETRICS && self.current_family.metric_type == MetricType::Counter { + parser.start_timestamp() + } else { + None + }; + let (_, label_spans, full_input, sample_ts_ms, value) = parser.raw_series_parts(); + let exemplar_parts = if P::IS_OPENMETRICS { + parser.raw_exemplar_parts() } else { None }; - let (_, label_spans, full_input, sample_ts_ms, value) = parser.raw_series_parts(); - let exemplar_parts = parser.raw_exemplar_parts(); // Always validate series labels first so duplicate/reserved labels are caught even when a // translation policy drops the metric family. @@ -1089,7 +1111,9 @@ impl ArrowDecoder { self.extract_series_labels_into_temp(label_spans, full_input, None)?; self.num_dp.attrs.extend_for_parent(dp_id, &self.temp_attrs); - if let Some((ex_val, ex_ts_ms, ex_spans)) = exemplar_parts { + if P::IS_OPENMETRICS + && let Some((ex_val, ex_ts_ms, ex_spans)) = exemplar_parts + { self.append_number_exemplar( dp_id, ex_val, @@ -1115,9 +1139,9 @@ impl ArrowDecoder { Ok(()) } - fn append_assembled_series( + fn append_assembled_series<'input, P: ScrapeParser<'input>>( &mut self, - parser: &mut ActiveParser<'_>, + parser: &mut P, role: SeriesRole, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -1140,7 +1164,7 @@ impl ArrowDecoder { }; let om_start_ts_ns = if !self.assembly.groups[group_idx].start_ts_checked { self.assembly.groups[group_idx].start_ts_checked = true; - if parser.is_openmetrics() { + if P::IS_OPENMETRICS { parser .start_timestamp() .and_then(|ms| ms.checked_mul(1_000_000)) @@ -1177,10 +1201,12 @@ impl ArrowDecoder { SeriesRole::Scalar => {} } - if matches!( - self.current_family.metric_type, - MetricType::Histogram | MetricType::GaugeHistogram - ) && let Some((ex_val, ex_ts_ms, ex_spans)) = parser.raw_exemplar_parts() + if P::IS_OPENMETRICS + && matches!( + self.current_family.metric_type, + MetricType::Histogram | MetricType::GaugeHistogram + ) + && let Some((ex_val, ex_ts_ms, ex_spans)) = parser.raw_exemplar_parts() { let full_input = parser.raw_series_parts().2; self.stage_pending_histogram_exemplar( @@ -1198,7 +1224,11 @@ impl ArrowDecoder { /// Checks whether the current assembled series line belongs to a data-point group already /// staged in `self.assembly` without interning new keys or values into the active batch. - fn series_matches_staged_group(&mut self, parser: &ActiveParser<'_>, role: SeriesRole) -> bool { + fn series_matches_staged_group<'input, P: ScrapeParser<'input>>( + &mut self, + parser: &P, + role: SeriesRole, + ) -> bool { let special_label = match role { SeriesRole::HistogramBucket => Some(b"le" as &[u8]), SeriesRole::SummaryQuantile => Some(b"quantile" as &[u8]), @@ -1648,6 +1678,7 @@ impl DecodeStream<'_, '_> { /// UTF-8 validation, duplicate or late metadata lines are encountered within a contiguous /// family, duplicate label names appear on a sample, or an OTAP table exceeds its integer ID /// width. + #[inline] pub fn step(&mut self, budget: DecodeBudget) -> Result { if self.cursor.done { return Ok(DecodeStep::Done); @@ -1681,6 +1712,7 @@ impl DecodeStream<'_, '_> { impl Iterator for DecodeStream<'_, '_> { type Item = Result; + #[inline] fn next(&mut self) -> Option { if self.cursor.done { return None; diff --git a/src/arrow/options.rs b/src/arrow/options.rs index 2b66367..05357c7 100644 --- a/src/arrow/options.rs +++ b/src/arrow/options.rs @@ -37,7 +37,7 @@ pub const DEFAULT_DECODE_BUDGET_MAX_DURATION: Duration = Duration::from_millis(1 /// Number of Layer 0 `Entry` items processed between wall-clock `Instant::elapsed` checks in /// budgeted decode steps. -pub const DECODE_BUDGET_DURATION_CHECK_INTERVAL: usize = 16; +pub const DECODE_BUDGET_DURATION_CHECK_INTERVAL: usize = 64; /// Cooperative step budget bounding how much entry ingestion work [`super::DecodeStream::step`], /// [`super::ArrowDecoder::step_decode`], [`super::ArrowDecoder::decode_cooperative`], and diff --git a/src/arrow/tests.rs b/src/arrow/tests.rs index 7b0b074..2a50b0c 100644 --- a/src/arrow/tests.rs +++ b/src/arrow/tests.rs @@ -444,7 +444,7 @@ fn decode_stream_splits_mid_family_preserves_metadata_and_handles_edges() { let mut stream = dec.decode_stream(labeled, &Format::TEXT, &zero_opts); assert!(stream.next().unwrap().is_ok()); } - assert!(dec.prom_spans.capacity() >= 4); + assert!(dec.prom_state.labels.capacity() >= 4); // Consecutive assembled families (histograms/summaries) preceded by `# TYPE` flush at // `max_data_points_per_batch` when the next family begins, and a single assembled family @@ -710,9 +710,11 @@ fn decode_stream_step_respects_entry_byte_and_duration_budgets() { assert_eq!(mid_err_stream.step(step1).unwrap(), DecodeStep::Done); drop(mid_err_stream); - // Non-zero `max_duration` checks elapsed time every 16 entries (`entries.is_multiple_of(16)`). + // Non-zero `max_duration` checks elapsed time every `DECODE_BUDGET_DURATION_CHECK_INTERVAL` + // entries. + let total_heavy_lines = DECODE_BUDGET_DURATION_CHECK_INTERVAL + 16; let mut heavy_lines = String::new(); - for i in 0..20 { + for i in 0..total_heavy_lines { heavy_lines.push_str("m{"); for k in 0..32 { if k > 0 { @@ -726,7 +728,7 @@ fn decode_stream_step_respects_entry_byte_and_duration_budgets() { let mut dur_stream = dec.decode_stream( heavy_lines.as_bytes(), &Format::TEXT, - &DecodeOptions::new().with_max_data_points_per_batch(64), + &DecodeOptions::new().with_max_data_points_per_batch(total_heavy_lines * 2), ); assert_eq!(dur_stream.step(dur_budget).unwrap(), DecodeStep::Pending); let consumed_lines = heavy_lines.as_bytes()[..dur_stream.pos()] @@ -735,7 +737,7 @@ fn decode_stream_step_respects_entry_byte_and_duration_budgets() { .count(); assert_eq!(consumed_lines, DECODE_BUDGET_DURATION_CHECK_INTERVAL); match dur_stream.step(dur_budget).unwrap() { - DecodeStep::Batch(batch) => assert_eq!(batch.num_data_points(), 20), + DecodeStep::Batch(batch) => assert_eq!(batch.num_data_points(), total_heavy_lines), other => panic!("expected DecodeStep::Batch, got {other:?}"), } assert_eq!(dur_stream.step(dur_budget).unwrap(), DecodeStep::Done); @@ -850,12 +852,12 @@ fn lifetime_free_decode_cursor_supports_owned_job_across_steps() { DecodeStep::Pending ); job.decoder.finish_decode(&mut aborted); - let retained_after_first = job.decoder.om_scratch.retained_memory_bytes(); + let retained_after_first = job.decoder.om_state.retained_memory_bytes(); assert!(aborted.is_done()); assert!(retained_after_first > 0); job.decoder.finish_decode(&mut aborted); assert_eq!( - job.decoder.om_scratch.retained_memory_bytes(), + job.decoder.om_state.retained_memory_bytes(), retained_after_first ); @@ -1005,11 +1007,11 @@ fn lifetime_free_decode_cursor_supports_owned_job_across_steps() { DecodeStep::Batch(records) => assert_eq!(records.num_data_points(), 2), other => panic!("expected active cursor to finish cleanly, got {other:?}"), } - let grown_prom_cap = job.decoder.prom_spans.capacity(); + let grown_prom_cap = job.decoder.prom_state.labels.capacity(); assert!(grown_prom_cap >= 18); job.decoder.finish_decode(&mut stale_after_active); assert!(stale_after_active.is_done()); - assert_eq!(job.decoder.prom_spans.capacity(), grown_prom_cap); + assert_eq!(job.decoder.prom_state.labels.capacity(), grown_prom_cap); let mut om_holder = job .decoder @@ -1021,12 +1023,12 @@ fn lifetime_free_decode_cursor_supports_owned_job_across_steps() { .unwrap(); job.decoder.finish_decode(&mut om_holder); assert_eq!( - job.decoder.om_scratch.retained_memory_bytes(), + job.decoder.om_state.retained_memory_bytes(), retained_after_first ); job.decoder.finish_decode(&mut stale_om_cursor); assert_eq!( - job.decoder.om_scratch.retained_memory_bytes(), + job.decoder.om_state.retained_memory_bytes(), retained_after_first ); @@ -1153,7 +1155,7 @@ fn cooperative_async_helpers_respect_fast_path_and_yield_budgets() { let mut cx = Context::from_waker(Waker::noop()); assert!(matches!(pinned.as_mut().poll(&mut cx), Poll::Pending)); } - assert!(dec.prom_spans.capacity() > 0); + assert!(dec.prom_state.labels.capacity() > 0); assert!(dec.metrics.is_empty()); let bad_coop = b"m{a=\"1\",b=\"2\",c=\"3\",d=\"4\"} 1\nbad_line\n"; diff --git a/src/parse.rs b/src/parse.rs index e0858bf..d1d56ca 100644 --- a/src/parse.rs +++ b/src/parse.rs @@ -321,6 +321,7 @@ pub trait Parser<'a> { fn comment(&self) -> &'a [u8]; /// Returns the metric name and unit of the current [`Entry::Unit`]. + #[inline] fn unit(&self) -> (&[u8], &[u8]) { (b"", b"") } diff --git a/src/parse/omlex.rs b/src/parse/omlex.rs index 3521c2e..627a99a 100644 --- a/src/parse/omlex.rs +++ b/src/parse/omlex.rs @@ -17,8 +17,9 @@ use super::ByteSpan; /// Lexer states, named after the start conditions in `openmetricslex.l`. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] enum State { + #[default] Init, Comment, /// After `# HELP`, `# TYPE` or `# UNIT`, before the metric name. @@ -35,7 +36,7 @@ enum State { } /// Lifetime-free cursor state of [`Lexer`]. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(crate) struct LexerState { pub(super) pos: usize, pub(super) token_start: usize, diff --git a/src/parse/omparse.rs b/src/parse/omparse.rs index 2334937..2b81f0b 100644 --- a/src/parse/omparse.rs +++ b/src/parse/omparse.rs @@ -64,6 +64,7 @@ impl OmScratch { /// Lifetime-free state of [`OpenMetricsParser`] that can be stored across suspension points /// alongside an owned input buffer without self-referential lifetimes. +#[derive(Default)] pub(crate) struct OpenMetricsParserState { lexer: LexerState, done: bool, @@ -85,7 +86,28 @@ pub(crate) struct OpenMetricsParserState { } impl OpenMetricsParserState { - /// Takes the reusable scratch buffers in place for `Drop` handlers. + /// Resets cursor state for a new scrape while preserving internal scratch capacities in place. + #[inline] + pub(crate) fn reset(&mut self) { + self.lexer = LexerState::default(); + self.done = false; + self.peeking = false; + self.line_start = 0; + self.series = ByteSpan::default(); + self.text = ByteSpan::default(); + self.name = None; + self.family_name_len = 0; + self.metric_type = MetricType::Unknown; + self.value = 0.0; + self.timestamp = None; + self.labels.clear(); + self.exemplar = None; + self.exemplar_labels.clear(); + self.peek_labels.clear(); + self.start_timestamp_key.clear(); + self.start_timestamp = None; + } + pub(crate) fn take_scratch(&mut self) -> OmScratch { self.labels.clear(); self.exemplar_labels.clear(); @@ -98,6 +120,20 @@ impl OpenMetricsParserState { start_timestamp_key: std::mem::take(&mut self.start_timestamp_key), } } + + pub(crate) fn put_scratch(&mut self, mut scratch: OmScratch) { + scratch.clear(); + self.labels = scratch.labels; + self.exemplar_labels = scratch.exemplar_labels; + self.peek_labels = scratch.peek_labels; + self.start_timestamp_key = scratch.start_timestamp_key; + } + + pub(crate) fn retained_memory_bytes(&self) -> usize { + (self.labels.capacity() + self.exemplar_labels.capacity() + self.peek_labels.capacity()) + * size_of::() + + self.start_timestamp_key.capacity() + } } /// Zero-copy parser for the OpenMetrics 1.0 text exposition format. @@ -140,6 +176,7 @@ impl<'a> OpenMetricsParser<'a> { } /// Creates a parser over `input` reusing pre-allocated scratch buffers. + #[inline] pub(crate) fn with_scratch(input: &'a [u8], mut scratch: OmScratch) -> Self { scratch.clear(); Self { @@ -217,6 +254,7 @@ impl<'a> OpenMetricsParser<'a> { } /// Takes the reusable scratch buffers in place for `Drop` handlers. + #[inline] pub(crate) fn take_scratch(&mut self) -> OmScratch { self.labels.clear(); self.exemplar_labels.clear(); @@ -243,12 +281,14 @@ impl<'a> OpenMetricsParser<'a> { } /// Returns the raw metric name and raw (escaped) help text for `Entry::Help`. + #[inline] pub(crate) fn raw_help(&self) -> (&'a [u8], &'a [u8]) { (self.metric_name(), &self.lexer.input[self.text]) } /// Returns the raw metric name, validated UTF-8 label spans, full input slice, optional /// timestamp in milliseconds, and sample value for `Entry::Series`. + #[inline] pub(crate) fn raw_series_parts(&self) -> (&'a [u8], &[LabelSpan], &'a [u8], Option, f64) { ( self.metric_name(), @@ -261,6 +301,7 @@ impl<'a> OpenMetricsParser<'a> { /// Returns the exemplar value, optional timestamp in milliseconds, and label spans for the /// current `Entry::Series`, if present. + #[inline] pub(crate) fn raw_exemplar_parts(&self) -> Option<(f64, Option, &[LabelSpan])> { self.exemplar .map(|(val, ts)| (val, ts, self.exemplar_labels.as_slice())) @@ -557,6 +598,7 @@ impl<'a> OpenMetricsParser<'a> { } impl<'a> Parser<'a> for OpenMetricsParser<'a> { + #[inline] fn next_entry(&mut self) -> Result, ParseError> { if self.done { return Ok(None); @@ -599,10 +641,12 @@ impl<'a> Parser<'a> for OpenMetricsParser<'a> { (self.metric_name(), unescape(&self.lexer.input[self.text])) } + #[inline] fn metric_type(&self) -> (&'a [u8], MetricType) { (self.metric_name(), self.metric_type) } + #[inline] fn unit(&self) -> (&'a [u8], &'a [u8]) { (self.metric_name(), &self.lexer.input[self.text]) } diff --git a/src/parse/promlex.rs b/src/parse/promlex.rs index 62f1284..be459a1 100644 --- a/src/parse/promlex.rs +++ b/src/parse/promlex.rs @@ -17,8 +17,9 @@ use super::ByteSpan; /// Lexer states, named after the start conditions in `promlex.l`. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] enum State { + #[default] Init, Comment, /// After `# HELP` or `# TYPE`, before the metric name. @@ -32,7 +33,7 @@ enum State { } /// Lifetime-free cursor state of [`Lexer`]. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(crate) struct LexerState { pub(super) pos: usize, pub(super) token_start: usize, diff --git a/src/parse/promparse.rs b/src/parse/promparse.rs index cd81aa5..ea1edf3 100644 --- a/src/parse/promparse.rs +++ b/src/parse/promparse.rs @@ -71,6 +71,7 @@ pub(crate) struct LabelSpan { /// Lifetime-free state of [`PromParser`] that can be stored across suspension points alongside an /// owned input buffer without self-referential lifetimes. +#[derive(Default)] pub(crate) struct PromParserState { lexer: LexerState, series: ByteSpan, @@ -80,14 +81,33 @@ pub(crate) struct PromParserState { timestamp: Option, line_start: usize, name: Option, - labels: Vec, + pub(crate) labels: Vec, } impl PromParserState { - /// Takes the reusable label span buffer in place for `Drop` handlers. - pub(crate) fn take_scratch(&mut self) -> Vec { + pub(crate) fn with_capacity(label_cap: usize) -> Self { + Self { + labels: Vec::with_capacity(label_cap), + ..Self::default() + } + } + + /// Resets cursor state for a new scrape while preserving `labels` capacity in place. + #[inline] + pub(crate) fn reset(&mut self) { + self.lexer = LexerState::default(); + self.series = ByteSpan::default(); + self.text = ByteSpan::default(); + self.last_type = MetricType::Unknown; + self.value = 0.0; + self.timestamp = None; + self.line_start = 0; + self.name = None; self.labels.clear(); - std::mem::take(&mut self.labels) + } + + pub(crate) fn retained_memory_bytes(&self) -> usize { + self.labels.capacity() * size_of::() } } @@ -115,6 +135,7 @@ impl<'a> PromParser<'a> { } /// Creates a parser over `input` reusing a caller-owned `labels` span buffer. + #[inline] pub(crate) fn with_scratch(input: &'a [u8], mut labels: Vec) -> Self { labels.clear(); Self { @@ -169,6 +190,7 @@ impl<'a> PromParser<'a> { } /// Takes the reusable label span buffer in place for `Drop` handlers. + #[inline] pub(crate) fn take_scratch(&mut self) -> Vec { self.labels.clear(); std::mem::take(&mut self.labels) @@ -188,12 +210,14 @@ impl<'a> PromParser<'a> { /// Returns the raw metric name and raw (escaped) help text for `Entry::Help` without /// allocating a `Cow::Owned` buffer. + #[inline] pub(crate) fn raw_help(&self) -> (&'a [u8], &'a [u8]) { (self.metric_name(), &self.lexer.input[self.text]) } /// Returns the raw metric name, validated UTF-8 label spans, full input slice, optional /// timestamp in milliseconds, and sample value for `Entry::Series`. + #[inline] pub(crate) fn raw_series_parts(&self) -> (&'a [u8], &[LabelSpan], &'a [u8], Option, f64) { ( self.metric_name(), @@ -395,6 +419,7 @@ impl<'a> PromParser<'a> { } impl<'a> Parser<'a> for PromParser<'a> { + #[inline] fn next_entry(&mut self) -> Result, ParseError> { loop { self.line_start = self.lexer.pos; @@ -424,6 +449,7 @@ impl<'a> Parser<'a> for PromParser<'a> { ) } + #[inline] fn metric_type(&self) -> (&'a [u8], MetricType) { (self.metric_name(), self.last_type) }