enhancement(dogstatsd): batch datagram decoder dequeues

[thieman]

Summary

• Dequeue up to 32 queued DogStatsD datagrams per shared receiver lock acquisition.
• Amortize Tokio MPSC receive, async mutex, wakeup, and scheduling overhead across a small batch.
• Preserve the existing global queue, decoder worker pool, per-packet socket context, and shutdown draining behavior.

This is stacked on #2079.

Receiver::recv_many wakes as soon as one item is available; it does not wait for the batch to fill. Low-volume traffic therefore follows the same wait-for-one-message behavior as before.

Change Type

• Bug fix
• New feature
• Non-functional (chore, refactoring, docs)
• Performance

How did you test this PR?

Automated validation

• make fmt
• cargo nextest run -p saluki-components dogstatsd
  • 154 tests passed
• cargo xwin check --target x86_64-pc-windows-msvc -p saluki-components --tests
• Repository pre-commit checks, including workspace Clippy and API documentation build

Closed-loop UDS throughput benchmark

I used a local experimental Docker/Lading harness rather than committing additional benchmark infrastructure in this PR. The benchmark used:

• the converged Agent image, toggling only Core versus ADP DogStatsD ownership;
• UDS datagrams;
• the existing deterministic 3,000-context Lading DogStatsD profile;
• a two-CPU target (0-1) and separate two-CPU Lading generator (2-3);
• two decoder workers;
• a 2 GiB target memory limit;
• five seconds of warm-up;
• a 30-second measurement at a saturated 1,600 MiB/s requested rate;
• blocking Lading sends, so overload appears as reduced achieved throughput rather than client drops.

Run	One-at-a-time dequeue	Batched dequeue
1	956.2 MiB/s	1,026.6 MiB/s
2	979.3 MiB/s	1,065.7 MiB/s
Mean	967.7 MiB/s	1,046.1 MiB/s

Mean throughput improved 8.1%. Mean packet rate increased from approximately 263k datagrams/s to 284k datagrams/s.

I also prototyped Linux recvmmsg with per-message UDS credentials. It improved the realistic profile by approximately 0.9% and a small-datagram profile by approximately 1.9%, within local variance, so that additional unsafe and platform-specific code is not included.

Trade-offs

• A worker may hold up to 32 messages outside the global channel while decoding them. Highly heterogeneous per-packet decode costs could therefore create short-lived worker load imbalance.
• The 100 ms event-buffer flush timer is not polled while a batch is decoded. At the measured packet rate, a 32-packet batch represents substantially less than one millisecond of work.
• Datagram completion order is already not guaranteed across parallel decoder workers; batching does not introduce a new ordering guarantee.
• The batch vector holds at most 32 existing queued messages per worker and does not allocate additional packet buffers.

References

• fix(dogstatsd): decouple datagram reads from decoding and increase buffer limit to 256MiB #2079
• DADP-150

[pr-commenter]

Binary Size Analysis (Agent Data Plane)

Baseline: c356375 · Comparison: a7f8350 · diff
Analysis Configuration: stripped binaries · Pass/Fail Threshold: +5%
Sizes: 41.75 MiB (baseline) vs 41.86 MiB (comparison)
Size Change: +115.16 KiB (+0.27%)

✅ Binary size difference within threshold

Changes by Module

Module	File Size	Symbols
figment	-180.82 KiB	180
core	+79.40 KiB	2195
saluki_components::sources::dogstatsd	+72.01 KiB	188
tokio	+43.40 KiB	471
otlp_protos::otlp_include::opentelemetry	+38.72 KiB	107
&mut serde_json	+33.38 KiB	31
prost	-18.56 KiB	224
serde_with	-18.55 KiB	24
anon.f9217dadaa411b15d9ce5e522ea3e33e.1185.llvm.8767143410872366912	+17.80 KiB	1
anon.36a7d0d314b9701965ea5b08796fad33.1185.llvm.15807495580778693563	-17.80 KiB	1
anon.5ffef35c7768023811ac5ab5fd4f00ff.888.llvm.7247732555510565809	+17.53 KiB	1
anon.5ffef35c7768023811ac5ab5fd4f00ff.885.llvm.15862285824570942458	-17.45 KiB	1
saluki_components::transforms::dogstatsd_mapper	+16.56 KiB	8
[sections]	+14.40 KiB	8
anon.acb9f1f81b0da7c4888eb6163f96e3ae.462.llvm.10090356048528741361	+12.37 KiB	1
anon.af90fe4baeb6e6b8384b677957b04820.462.llvm.11835343198361071145	-12.37 KiB	1
tracing	+11.93 KiB	32
tokio_rustls	-11.22 KiB	14
anyhow	+10.76 KiB	231
saluki_components::sources::otlp	-9.65 KiB	36

Detailed Symbol Changes

    FILE SIZE        VM SIZE    
 --------------  -------------- 
  +0.8% +82.7Ki  +0.6% +44.0Ki    [16048 Others]
  [NEW] +21.7Ki  [NEW] +21.5Ki    _<saluki_components::sources::dogstatsd::DogStatsDConfiguration as saluki_core::components::sources::builder::SourceBuilder>::build::_{{closure}}::h4481e0171fc4f085
  [NEW] +21.1Ki  [NEW] +21.0Ki    saluki_components::sources::dogstatsd::drive_decoder::_{{closure}}::h26a33e23ae032832
  [NEW] +19.5Ki  [NEW] +19.3Ki    saluki_components::sources::dogstatsd::process_datagram_decoder::_{{closure}}::h5c05e54ed3a53573
  [NEW] +18.7Ki  [NEW] +18.6Ki    saluki_components::sources::dogstatsd::drive_stream::_{{closure}}::hbef42a8b1492df11
  [NEW] +17.8Ki  [NEW]     +76    anon.f9217dadaa411b15d9ce5e522ea3e33e.1185.llvm.8767143410872366912
  [NEW] +17.5Ki  [NEW] +17.4Ki    anon.5ffef35c7768023811ac5ab5fd4f00ff.888.llvm.7247732555510565809
  [NEW] +16.5Ki  [NEW] +16.4Ki    figment::figment::Figment::extract::h35d5a146172e5fec
  [NEW] +15.9Ki  [NEW] +15.4Ki    core::ptr::drop_in_place<datadog_agent_config::generated::datadog_configuration::DatadogConfiguration>::h12005a1fdeb0ffb6
  [NEW] +15.2Ki  [NEW] +15.0Ki    saluki_components::common::datadog::obfuscation::_::_<impl serde_core::de::Deserialize for saluki_components::common::datadog::obfuscation::ObfuscationConfig>::deserialize::hfdb45102b1c29886
  [NEW] +14.4Ki  [NEW] +14.3Ki    datadog_agent_config::generated::witness::drive::h1f4ddadd82aeab06
  [NEW] +14.1Ki  [NEW] +14.0Ki    saluki_components::sources::dogstatsd::handle_frame::h126efbbdff5749ea
  [DEL] -14.1Ki  [DEL] -14.0Ki    _<figment::value::magic::Tagged<T> as figment::value::magic::Magic>::deserialize_from::hbc114368f39b0da4
  [DEL] -14.2Ki  [DEL] -14.1Ki    datadog_agent_config::generated::witness::drive::h7d1355f2e03f9c0b
  [DEL] -14.4Ki  [DEL] -14.3Ki    _<figment::value::magic::RelativePathBuf as figment::value::magic::Magic>::deserialize_from::h57935c312d7d4142
  [DEL] -15.1Ki  [DEL] -14.9Ki    _<figment::value::de::ConfiguredValueDe<I> as serde_core::de::Deserializer>::deserialize_struct::hcd5d2f1d3886cf32
  [DEL] -15.8Ki  [DEL] -15.3Ki    core::ptr::drop_in_place<datadog_agent_config::generated::datadog_configuration::DatadogConfiguration>::h1fc411b0a78628dd
  [DEL] -17.4Ki  [DEL] -17.4Ki    anon.5ffef35c7768023811ac5ab5fd4f00ff.885.llvm.15862285824570942458
  [DEL] -17.8Ki  [DEL]     -76    anon.36a7d0d314b9701965ea5b08796fad33.1185.llvm.15807495580778693563
  [DEL] -22.3Ki  [DEL] -22.1Ki    _<saluki_components::sources::dogstatsd::DogStatsDConfiguration as saluki_core::components::sources::builder::SourceBuilder>::build::_{{closure}}::hea5210e3bf2131b7
  [DEL] -28.7Ki  [DEL] -28.6Ki    saluki_components::sources::dogstatsd::drive_stream::_{{closure}}::hb18fa47632f24940
  +0.3%  +115Ki  +0.2% +76.2Ki    TOTAL

[pr-commenter]

Regression Detector (Agent Data Plane)

Run ID: 49a314e3-21f7-445a-be02-c23a42005626
Baseline: c3563758 · Comparison: a7f83504 · diff

Optimization Goals: ❌ 3 regressions detected

experiment	goal	Δ mean %	links
quality_gates_rss_dsd_ultraheavy	memory	🔴 +69.96	metrics profiles logs
quality_gates_rss_dsd_heavy	memory	🔴 +53.43	metrics profiles logs
quality_gates_rss_dsd_medium	memory	🔴 +24.42	metrics profiles logs

Fine details of change detection per experiment (2)

Experiments configured erratic: true are tagged (ignored) and skipped when determining which experiments regressed or improved. Experiments which are detected as erratic at runtime are tagged (erratic) to flag that the run's sample dispersion was high, but their regression / improvement signal still counts.

experiment	goal	Δ mean %	links
quality_gates_rss_dsd_low	memory	⚪ +2.87	metrics profiles logs
quality_gates_rss_idle	memory	⚪ +0.63	metrics profiles logs

Bounds Checks: ❌ Failed (5)

experiment	check	replicates	observed	links
quality_gates_rss_dsd_heavy	memory_usage	0/10	❌ 228 MiB ≤ 145 MiB	metrics profiles logs
quality_gates_rss_dsd_low	memory_usage	1/10	❌ 51.1 MiB ≤ 50 MiB	metrics profiles logs
quality_gates_rss_dsd_medium	memory_usage	0/10	❌ 91.3 MiB ≤ 80 MiB	metrics profiles logs
quality_gates_rss_dsd_ultraheavy	memory_usage	0/10	❌ 388 MiB ≤ 205 MiB	metrics profiles logs
quality_gates_rss_idle	memory_usage	10/10	✅ 32.1 MiB ≤ 40 MiB	metrics profiles logs

Explanation

A change is flagged as a regression when |Δ mean %| > 5.00% in the regressing direction for its optimization goal AND SMP marks the experiment as a regression (is_regression: true). Improvements use the matching criteria for the improving direction. Experiments configured erratic: true (tagged (ignored)) are skipped outright; experiments detected as erratic at runtime (tagged (erratic)) still count, since that flag describes sample dispersion rather than directional certainty. The Δ mean % cell is colored accordingly: 🟢 = improvement, 🔴 = regression, ⚪ = neutral. Reduction in CPU or memory is an improvement; reduction in ingress throughput is a regression.

[datadog-prod-us1-5]

More details

The batched decoder preserves FIFO processing, per-datagram socket context, and shutdown draining while reducing receiver-lock overhead. No behavioral regression was identified; focused execution could not reach compilation because the sandbox lacks the git-pinned hyper-http-proxy dependency and cannot access the network.

Was this helpful? React 👍 or 👎

_{Open Bits AI session}

_{🤖 Datadog Autotest · Commit a7f8350 · What is Autotest? · @DataDog review to ask questions · Any feedback? Reach out in #autotest}

[webern]

Mostly nits other than the memory regression detector.

[webern]

I found this description slightly lacking. Here is a more suggested, fuller description.

When a completion file is requested, Millstone writes a marker after to signal that it has finished sending, then remains alive instead of exiting. This lets callers detect that sending is complete while preserving the sender’s process identity for origin detection.

[webern]

Writing this comment after the suggestion in main.rs. I am slightly confused as to whether millstone will wait indefinitely when writing a completion file or whether it is interacting with FLUSH_WAIT somehow.

[webern]

Nit: just because Pinned and Unpinned mean something specific in Rust code.

Suggested change

	Resolved(Option<EntityId>),
	Unresolved(u32),

[webern]

nit: Pinned and Unpinned are overloaded because of the Rust meanings. Also I thought a bit of documentation could help the reader here.

Suggested change

	/// The outcome of origin resolution at the time of message decoding.
	#[derive(Clone, Debug, Eq, PartialEq)]
	pub(super) enum ProcessOrigin {
	/// The resolver ran when the message was decoded.
	Resolved(Option<EntityId>),
	/// Only the PID is available, the resolver did not run.
	Unresolved(u32),
	/// The traffic is generated by replay, use the replay PID.
	Replay(u32),

[webern]

If I understand correctly, this provides backpressure before starting another receive cycle. Is it the allocations downstream from here that are causing the RSS regressions in the regression detector? Or is it the larger number of BytesBuffers themselves?