Skip to content

[dbsp] Fix broken metrics - #6859

Merged
ryzhyk merged 7 commits into
mainfrom
fix-circuit-wait-time
Aug 14, 2026
Merged

[dbsp] Fix broken metrics#6859
ryzhyk merged 7 commits into
mainfrom
fix-circuit-wait-time

Conversation

@ryzhyk

@ryzhyk ryzhyk commented Aug 13, 2026

Copy link
Copy Markdown
Contributor

Fix two metrics that became inaccurate after recent changes: circuit wait time and merger backpressure wait time.

Also, add information about remaining batch count to samply markers related to merger events.

Fix spine attribution in the streaming exchange (confused the samply profile, but could also have impact on performance).

See individual commits for details.

Describe Manual Test Plan

Checklist

  • Unit tests added/updated
  • Integration tests added/updated
  • Documentation updated
  • Changelog updated

Breaking Changes?

Mark if you think the answer is yes for any of these components:

Describe Incompatible Changes

ryzhyk added 6 commits August 13, 2026 14:50
circuit_wait_time_seconds has read zero since June. It was fed by the
scheduler's WaitStart/WaitEnd events, which fire only when the scheduler has no
runnable task, and that requires a DBSP-asynchronous operator. Since 50ff767
turned ExchangeReceiver into a Rust-async operator, a blocked operator is a
pending task rather than an idle scheduler, so the events stop happening: only
RebalancingExchangeSender and GatherConsumer still declare themselves
asynchronous, and neither appears in an ordinary circuit.

Fortunately we have an easy way to measure how long the circuit waits by asking
the tokio runtime on each step.

Also fill in the step profile's CPU time, which was hardcoded to zero, and
report it as circuit_cpu_time_seconds plus circuit_nonblocking_percent,
mirroring what operators already report. A step's wall time now decomposes:

    circuit_runtime_seconds
      - circuit_cpu_time_seconds     on cpu
      - circuit_wait_time_seconds    nothing to run
      = blocked in the kernel or descheduled

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
A merge marker said what the merge consumed and produced but not what it left
behind? The marker now ends with "spine now holds N loose of M
batches".

The marker moves after the merged batch is added to the spine, so the counts
describe the state the merge leaves rather than the state mid-way through it.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
merge_backpressure_wait_time_seconds read zero on a workload where workers
demonstrably waited for merging, and no backpressure-wait marker appeared in
its profile. There are two ways to wait for the spine to drain and only one of
them was instrumented.

`backpressure_wait` polls in a loop, adds the elapsed time to the spine's
stats, and records a marker. `backpressure_waiter` just hands out a `Notified`
to await, and recorded nothing; the sharded accumulator uses that one
(ShardedAccumulatorLocalWaiter), so its waits reached neither the metric nor
the profile. They were visible only as unexplained time inside the operator.

The waiter now carries what a report needs, its caller reports the wait once
awaited, and both paths funnel into a single `record_backpressure_wait`, so a
wait counts the same however it was taken. The marker reports the batch counts
either side of the wait, loose and total, which says whether waiting achieved
anything.

Splitting the waiter into a future and a report keeps the `Notified` out of the
value the caller holds across the await.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
Every worker shares one `ShardedAccumulator`, built by whichever of them
reaches the constructor first, and that worker's index went to all of the
receive queues. Their compaction then ran on the constructing worker's
background buffer cache and slab allocator, which a spine's worker index
picks, rather than spreading over its peers', and a profile reported all of
the operator's merges under that one worker.

The wrong worker also made concurrent merges of distinct queues look like
repeated merges of a single spine in samply markers: every queue names its
spines after the same operator, so the owner is the only field that tells them
apart.

Construction now takes no worker index at all, so the mistake cannot recur.
A spine carries its owner, so the backpressure-wait marker can name it
alongside the blocked worker that records the wait.

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
Its documentation links to the private `BackpressureWait` it is split from,
which rustdoc reports because that link resolves only while private items are
being documented and breaks in the public docs.

The report, and the two `Spine` methods that hand it out and take it back, are
used only by the sharded accumulator inside this crate, so `pub(crate)`
describes them accurately and the link resolves.

Also stop pointing at `RuntimeIdle`'s private `base` field, which rustdoc
flags for the same reason.
`cargo doc -p dbsp` reported eleven warnings that predate this branch, and the
crate's documentation now builds without any.

Four links named items that no longer exist or never did: `StaticScheduler` is
gone, `group_transform` is now `dyn_group_transform`, `size_from_level` is now
`size_to_level`, and `eof` is `BulkRows::at_eof` rather than a method of the
index level whose documentation cited it. Two more spelled out a target that
the link's own label already resolved to, and one wrote `std::mem::transmute`
with a single colon.

The remaining four are prose indexing an array, `weight_times[current_index]`
and `MERGE_COUNTS[level]`, which rustdoc reads as a link. Backticks make them
code, which is what they are.
@ryzhyk
ryzhyk requested a review from blp August 13, 2026 22:30

@blp blp left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Good catches, great improvements, thank you!

Comment on lines +3225 to +3235
let tokio_runtime = {
let (park, unpark) = (runtime_idle.clone(), runtime_idle.clone());
let mut builder = tokio::runtime::Builder::new_current_thread();
builder.on_thread_park(move || park.park());
builder.on_thread_unpark(move || unpark.unpark());
builder.build().map_err(|e| {
DbspError::Scheduler(SchedulerError::TokioError {
error: e.to_string(),
})
})?;
})?
};

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I see tokio has a function for reporting the amount of time that a worker is busy; presumably we could subtract. I don't know whether that would be accurate enough, but it might be easier.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Apparently it doesn't work well, but I didn't fully understand the reason. Part of it is that we want to measure wait time within a step, not between steps, but there are other issues as well. Apparently this counter doesn't get updated at the right times, making it difficult to use in our accounting.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Makes sense.

Maybe a comment would be a good idea.

Comment thread crates/dbsp/src/trace/spine_async.rs Outdated
// counts describe the spine as the merge leaves it. `loose` is what
// backpressure measures, and it is what grows when merging cannot keep
// up with the batches each step adds.
let loose = self.batch_count().0;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Hmm, maybe we should rename this function to count_loose_batches. The name makes it sound like it's all batches.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Also, I wonder if our heuristic for backpressure should be based on the total number of batches instead of the number of loose batches. Just starting a merge does nothing to relieve backpressure but we currently pretend it does.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

huh, I didn't know the function was not counting all batches, and yes, this can definitely go wrong. I'll rename the function, but I'm weary to change the heuristic in this PR. It's not easy to check that it doesn't make things worse.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I'll check on the heuristic myself after this merges.

Comment thread crates/dbsp/src/trace/spine_async.rs Outdated
`batch_count` counted only the loose batches, not the batches being merged, so
every caller read as if it covered the whole spine. It is now
`count_loose_batches`, returning a `LooseBatchCount`, and the locals that held
its result are named for what they hold: the type carried the same ambiguity as
the function, and so did calling a loose count `batch_count`.

The total, which really is every batch, was spelled out as a fold over the
slots in six places. It is now `count_all_batches`, so the two counts read as
the pair they are.
@ryzhyk
ryzhyk added this pull request to the merge queue Aug 14, 2026
Merged via the queue into main with commit 5e76148 Aug 14, 2026
1 check passed
@ryzhyk
ryzhyk deleted the fix-circuit-wait-time branch August 14, 2026 22:29
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants