Repository navigation
fix(group): fold keyless reduction partials in chunk order - #758
Closed
ser-vasilich wants to merge 2 commits into
Closed
ser-vasilich wants to merge 2 commits into
ser-vasilich wants to merge 2 commits into
Conversation
Scalar reductions and the keyless select scan kept one accumulator per worker while the pool hands out tasks dynamically, so F64 sum, avg, var and prod changed from run to run. Partials are now kept per chunk, with chunk boundaries derived from the row count only, and folded in chunk order, which gives the same bits at every core count.
4 tasks done
Blocks of chunks scanned at the same time sat on neighbouring 64-byte lines, which the adjacent-line prefetcher moves as a pair, so they still bounced between cores on every row of the generic kernel. Five aggregates over 70K rows on a two-socket box took 5.7 ms against 0.58 ms before the chunked fold.
Collaborator
Collaborator
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What & why
A parallel scalar reduction over F64 returned different results from call to call:
Two keyless paths kept one accumulator per worker while
ray_pool_dispatchhands out 8192-row tasks dynamically, then merged the partials in worker order. The rows behind each partial, and so the float association order, changed on every call:exec_reduction(par_reduce_fn→accs[worker_id]). It serves thesum/avg/prod/min/maxbuiltins, lazy reductions such as(sum (+ p 1.3)), lazyvar/dev, and selects whose aggregates all go through the sum shortcut.exec_group(scalar_*_fn→accums[worker_id]), with and withoutwhere:.Affected: F64/F32
sum,avg,var/stddev(also on integer inputs, which accumulate in a double) and F64prod, plus min/max ties between +0 and -0. Integersum/avg, ALL/ANY and integer min/max are order-independent and were already stable.Fix. The rows are split into chunks whose boundaries depend only on the row count:
RAY_PARALLEL_THRESHOLD, one chunk, as before;RAY_POOL_INIT_TASKSchunks.Each chunk keeps its own partial (
reduce_acc_tinexec_reduction, a cache-line-padded accumulator block in the keyless scan), and the partials are folded in chunk order. When the pool cannot dispatch (one core, or a nested call), the same chunks run serially. The result is the same bits at every core count.Since the fold is now ordered, the keyless scan no longer forces a serial pass for FIRST/LAST, so a select with
first/lastnow runs in parallel. A cancelled dispatch now returns the cancel error. The hot loops are unchanged: no compensated summation and no -0.0 handling. Extra memory is at most 1024 partials, about 155 KB at 100M rows.Results change only in the last few significant digits of large F64 aggregates. They are now identical across runs and core counts, and a select
avgequals the vectoravgof the same column.Validation
test_group_extra.c. They fail on the previous code on every run.group_extra/reduction_bits_any_pool: 131,073 rows, pools 1/2/5/8, two runs each.where:through bothexec_reductionandexec_group.group_extra/reduction_bits_wide_chunks: 8.4M rows, so each chunk spans several 8192-row blocks.test/rfl/agg/keyless_pool_accum.rfl: repeated vector and select calls, includingwhere:, must equal the first call.-c 1/8/24. Vector sum/avg/min/max and single and multi-aggregate selects with and withoutwhere:are equal within noise. A select withfirst/lastgoes from 256 to 17 ms at 24 cores (20M rows).make test(ASan/UBSan): 4112 of 4112 passed.make tsan-testis clean on the reduction, keyless and stress filters.Checklist
devmakebuilds cleanly (no new warnings)make testpasses; tests added for behaviour changes