Skip to content

fix(group): fold keyless reduction partials in chunk order - #758

Closed
ser-vasilich wants to merge 2 commits into
devfrom
fix/f64-sum-determinism
Closed

ser-vasilich wants to merge 2 commits into
devfrom
fix/f64-sum-determinism

Conversation

@ser-vasilich

Copy link
Copy Markdown
Collaborator

What & why

A parallel scalar reduction over F64 returned different results from call to call:

(set P (take [1.7976931348623157e308 -1.7976931348623157e308 3.2] 131073))
(sum P)   ;; -c 4: 3.2, 6.4, 0 or 0Nf from call to call; -c 1: always 3.2
(set C (take [1e16 1.0 -1e16] 131073))
(sum C)   ;; -c 4: 0, 1, 2 or 8; -c 1: always 0

Two keyless paths kept one accumulator per worker while ray_pool_dispatch hands 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 the sum/avg/prod/min/max builtins, lazy reductions such as (sum (+ p 1.3)), lazy var/dev, and selects whose aggregates all go through the sum shortcut.
  • The keyless select scan in exec_group (scalar_*_fn → accums[worker_id]), with and without where:.

Affected: F64/F32 sum, avg, var/stddev (also on integer inputs, which accumulate in a double) and F64 prod, plus min/max ties between +0 and -0. Integer sum/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:

  • below RAY_PARALLEL_THRESHOLD, one chunk, as before;
  • above it, multiples of 8192 rows, at most RAY_POOL_INIT_TASKS chunks.

Each chunk keeps its own partial (reduce_acc_t in exec_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/last now 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 avg equals the vector avg of the same column.

Validation

  • New tests in 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.
      • Covers vector sum/avg/var_pop, the keyless SUM through the fast and the generic kernel, and where: through both exec_reduction and exec_group.
      • Data: overflow, cancellation, an absorbing 2^53-then-1s column, ±0 min/max ties, and first/last with nulls spanning whole chunks.
    • 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, including where:, must equal the first call.
  • Sabotage. Deriving the chunk shape from the worker count, collapsing either serial path into a single pass, or reversing the keyless fold each makes a test fail.
  • Speed, release build, 20M and 100M rows at -c 1/8/24. Vector sum/avg/min/max and single and multi-aggregate selects with and without where: are equal within noise. A select with first/last goes from 256 to 17 ms at 24 cores (20M rows).
  • make test (ASan/UBSan): 4112 of 4112 passed. make tsan-test is clean on the reduction, keyless and stress filters.

Checklist

  • PR targets dev
  • Conventional Commits
  • make builds cleanly (no new warnings)
  • make test passes; tests added for behaviour changes

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.
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.
@singaraiona

Copy link
Copy Markdown
Collaborator

Every hunk of this branch is in dev now through #744 (f91f52f), which carried this head; a reverse apply of this PR's diff against dev succeeds cleanly. It can be closed as included.

@singaraiona

Copy link
Copy Markdown
Collaborator

Closing as included: this branch's commits were carried into #744 and are on dev as f91f52f, with authorship kept in the squash commit's co-author trailer.

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