compute: make index peek processing cooperative - #38040
Conversation
40be927 to
a69646f
Compare
DAlperin
left a comment
DAlperin
left a comment
There was a problem hiding this comment.
Still reading but an initial thought is that this may change peak memory behavior of clusterds in slightly unexpected ways. Previously when a worker would drive its scan to completion there would only be one in progress results vec per worker. Now if a user issues multiple peeks (and they aren't streamable) we can end up holding refs to more batches for longer. Might not be an issue but wonder if we have a good way to test cc @dennis.felsing
DAlperin
left a comment
DAlperin
left a comment
There was a problem hiding this comment.
Added some minor. My big question is around the complexity this introduces (and convinces me we need a real serving layer). Sitting with it for a few to think.
| /// Non-zero while the budget is not spent, and zero once it is. A caller | ||
| /// that must make progress regardless has to impose its own floor: a spec | ||
| /// of `work:0` yields a budget that is spent before any work happens. | ||
| pub fn allowance(&self) -> usize { |
There was a problem hiding this comment.
I think you need to check expired to make the contract described in the doc comment true. "Non-zero while the budget is not spent, and zero once it is."
There was a problem hiding this comment.
Good catch, and it was hiding a real bug rather than just an inaccurate doc.
allowance is fine on its own here (the scan re-checks is_spent at the end of
each slice, so it does yield on time). The damage is in NestedBudget, whose
allowance is the minimum of the two. With the deadline only consulted by
is_spent, an expired shared budget kept reporting until_clock_check, so
every peek reached after the activation deadline still got a full
peek_yielding turn. peek_yielding_total's time bound therefore didn't cap
the activation at all, which is the one thing it exists to do.
Fixed by returning 0 from allowance once expired, which makes
allowance() == 0 and is_spent() agree exactly rather than by convention:
work == Some(0) gives 0 through the min, expired gives 0 directly, and
until_clock_check is never observed at 0 because charge resets it. The
forward-progress floor in PeekScan::scan is allowance().max(1), so a spent
budget still advances the cursor by one and can't livelock.
Two tests added, both of which fail without the change:
expired_budget_hands_out_no_allowance for the contract you quoted, and
expired_shared_budget_leaves_nothing_to_nest for the nested case that was
actually broken.
|
Some duped content |
a69646f to
4919d27
Compare
Serving a ready index peek walked the whole arrangement in one go. For a large arrangement that pins the worker thread for the full duration of the scan, so dataflows aren't scheduled, commands aren't handled, and the peek can't even observe its own cancellation. A peek now scans in bounded slices. `PeekScan` owns the cursor, the rows collected so far, and the size accounting, and its `step` spends one budget before handing the worker back. The cursor owns the batches it reads rather than borrowing them from the trace, so a scan is self-contained and parking one between activations is safe. `PeekResultIterator::step` charges fuel per cursor position rather than per row returned. Counting rows would let a selective `map_filter_project` over a large arrangement run arbitrarily long without ever reaching a yield point. Fuel bounds how often we get to yield, not the length of any one slice, which the docs now say. Budgets nest. Each peek gets its own turn (`peek_yielding`), bounded by what all peeks together may spend in one activation (`peek_yielding_total`). Peeks that don't get a turn are served first on the next activation, so a long peek can't starve the ones behind it. A budget reports no allowance once spent, on the time bound as much as the work bound. Since a nested allowance is the minimum of the two, checking the deadline only when asked whether to yield would let every peek reached after the shared deadline still take a full turn, and the shared time bound would not cap the activation at all. A slice always advances the cursor at least once, even on a spent budget. A yielded peek keeps the worker from parking, so a slice that does no work at all would be a livelock rather than a slow peek. Making that a property of the loop means a budget of `work:0` degrades to a slow peek instead of hanging one. A yielded peek is work the worker owes itself, so `run_client` doesn't park while any peek has work left. That is also why `handle_peek` no longer serves the peek inline: `process_peeks` runs later in the same iteration, so latency is unchanged, and routing everything through there means a burst of peeks shares one budget instead of each getting its own. Peek timings now accumulate across activations and are reported once the peek is done. Reporting per activation would turn one slow peek into a string of fast ones. The cost is that a peek cancelled mid-scan reports nothing, which the help texts now say. That leaves nothing measuring how long peeks hold the worker per activation, which is the quantity the budgets bound, so `mz_peek_processing_seconds` times one pass over the pending peeks. `YieldSpec` moves out of the linear join into `crate::yielding` so both callers share one policy type and config format. The new `peek-count` script command in the clusterd test driver peeks an index directly and reports only the row count. `count` tallies through an ephemeral reduce dataflow and then peeks that dataflow's single-row output, so it cannot exercise a scan over many rows. It also takes optional literal constraints, which reach the other cursor path, the one that seeks from one literal to the next rather than stepping.
4919d27 to
a5dea7b
Compare
antiguru
left a comment
antiguru
left a comment
There was a problem hiding this comment.
Leaving some comments inline. The main mechanical comment is that we shouldn't support yielding by time, and hence also not extract the YieldSpec from joins. I need to think more about implications on latency, throughput of the whole computation, so just a high-level review for now.
| /// the peek walks `id`'s whole arrangement, which is what makes it useful | ||
| /// for exercising the peek scan over many rows without a golden that has to | ||
| /// spell every one of them out. | ||
| PeekCount { |
There was a problem hiding this comment.
I think this could be folded into Peek by adding a finishing parameter. By default it's emit, and could take count to just emit the count. hash might be a third option.
| }, | ||
| /// Peek `id` at `ts` and emit the returned rows (sorted, one per line). The | ||
| /// generic output assertion: the script's `----` block holds the expected rows. | ||
| Peek { |
| } | ||
|
|
||
| /// Parse a `[a,b,c]` list of `usize`s (`[]` is empty). | ||
| fn parse_i64_list(s: &str) -> anyhow::Result<Vec<i64>> { |
There was a problem hiding this comment.
Prefer making parse_usize_list generic instead of adding another variant. Also this skips a comment.
| /// The yielding behavior with which a single index peek should be processed. | ||
| pub const PEEK_YIELDING: Config<&str> = Config::new( | ||
| "peek_yielding", | ||
| "work:100000,time:10", |
There was a problem hiding this comment.
Please do not use time-based yielding by default. I believe it's a dangerous anti-pattern. Raises the question why support it at all.
There was a problem hiding this comment.
What's the background for time-based yielding being bad?
| /// The yielding behavior with which index peeks as a whole should be processed. | ||
| pub const PEEK_YIELDING_TOTAL: Config<&str> = Config::new( | ||
| "peek_yielding_total", | ||
| "work:1000000,time:100", |
There was a problem hiding this comment.
Same here, don't yield based on time.
| // arrangement still comes back here to have the budget checked. | ||
| let allowance = budget.allowance().max(1); | ||
| let mut fuel = allowance; | ||
| let step = self.iter.step(&mut fuel); |
There was a problem hiding this comment.
This assumes that step never increments fuel, which would cause overflows. Not sure why it would, but not documented as an invariant.
| /// Fuel is charged per cursor position, not per row returned, so a | ||
| /// selective `map_filter_project` cannot starve the caller of yield | ||
| /// points. Returning with fuel left over means the cursor is exhausted. |
There was a problem hiding this comment.
This is indeed a deviation of how we use fuel in other places, where it serves to guard downstream operators from overflowing their memory.
3 participants
Motivation
Part 1 of 2 in a stack that makes index peeks stop monopolizing the compute
worker. A replica crash fix in the code this commit rewrites went first, in
#38039, and has landed.
Serving a ready index peek walked the whole arrangement cursor in one call.
The compute worker is a single thread, so for a large arrangement that pins it
for the full duration of the scan: dataflows aren't scheduled, commands aren't
handled, and the peek can't even observe its own cancellation.
Part of CPU-195.
Description
A peek now scans in bounded slices.
PeekScanowns the cursor, the rowscollected so far, and the size accounting, and its
stepspends one budgetbefore handing the worker back. The cursor owns the batches it reads rather
than borrowing them from the trace, so a scan is self-contained and parking one
between activations is safe. This is the same shape the peek stash path already
uses to pump rows across ticks.
Fuel is charged per cursor position, not per row returned. Counting rows would
let a selective
map_filter_projectover a large arrangement run arbitrarilylong without ever reaching a yield point, which is exactly the case we care
about.
Budgets nest. Each peek gets its own turn (
peek_yielding), bounded by whatall peeks together may spend in one activation (
peek_yielding_total). Peeksthat don't get a turn are served first on the next activation, so a long peek
can't starve the ones behind it.
A slice always advances the cursor at least once, even on a spent budget. A
yielded peek keeps the worker from parking, so a slice that did no work at all
would be a livelock rather than a slow peek. Making that a property of the loop
means a budget of
work:0degrades to a slow peek instead of hanging a worker,rather than relying on config validation to rule the value out.
A yielded peek is work the worker owes itself, so
run_clientdoesn't parkwhile any peek has work left. That is also why
handle_peekno longer servesthe peek inline:
process_peeksruns later in the same loop iteration, solatency is unchanged, and routing everything through there means a burst of
peeks shares one budget instead of each getting a full one.
Cancellation of a long scan now actually works. Previously the worker could not
observe a
CancelPeekwhile it was inside the scan.YieldSpecmoves out of the linear join intocrate::yieldingso both callersshare one policy type and config format. That move is pure, the linear join
behaves the same.
Metrics
Peek timings now accumulate across activations and are reported once the peek is
done. Reporting per activation would turn one slow peek into a string of fast
ones. The cost is that a peek cancelled mid-scan reports nothing, which the help
texts now say.
That accumulation removes any view of how long a peek holds the worker in one
go, which is the quantity the budgets actually bound, so
mz_peek_processing_secondstimes one pass over the pending peeks.Known limits, deliberately left
The error-trace scan stays unbudgeted. It walks every key of the errs trace,
which in practice holds a handful of rows. There is a
NOTE:on it.Cursor setup is unbudgeted, and a single unit of fuel is not a bounded amount of
work. Seeking across a run of literal constraints that match no key happens
inside one unit. So the budget bounds how often we get to yield, not the length
of any one slice. Both have
NOTE:s.Verification
New unit tests in
src/compute/src/yielding.rscover the yield-spec parser andthe budget arithmetic: that a nested budget is bounded by both its own and the
shared allowance, that an unbounded budget never spends, and that a budget that
is not yet spent never hands out a zero allowance. That last one would spin the
scan loop, and writing it caught a real bug.
test/clusterd-test-driver/scripts/peek_yielding.specis the targeted test. Itserves peeks over a 2000-row index under
work:1, so the scan cannot finish inone activation and has to resume on the order of 2000 times. It pins that no row
is lost or duplicated across a yield, that a finished scan leaves nothing behind
that corrupts the next peek, and that a peek spanning many activations still
produces exactly one response.
Two new driver capabilities were needed to write it.
peek-countpeeks an indexdirectly and reports the row count, because the existing
counttallies throughan ephemeral reduce dataflow and then peeks that dataflow's single-row output,
so it walks one cursor position and cannot exercise a scan at all. And
peek-counttakes optional literal constraints, which reach the other cursorpath, the one that seeks from one literal to the next rather than stepping. The
literal cases cover unsorted input, literals matching no key, ending by
exhausting the literal list rather than the cursor, and an empty list.
Broader coverage comes from running the suite with a budget below production, so
that any peek over ~1000 cursor positions resumes at least once.
peek_yieldingandpeek_yielding_totalare wired intoget_variable_system_parametersfor that, and into parallel-workload's flagflipping.
That default is kept within an order of magnitude of production on purpose. It
reaches every mzcompose composition, and a very small value buys little extra
coverage while multiplying the timely steps a large peek needs. The randomized
runs go lower. Both parameters are also pinned to their production values in
ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS, because the benchmarks measureagainst an older image that does not know them, so a non-production value there
would only slow down one side of the comparison.
Not covered by a test. The round robin across concurrent peeks, and with it
peek_yielding_total, which only binds once more than one peek is pending. Thedriver awaits each peek before sending the next, so expressing this needs a
concurrency primitive in the driver rather than another command.