From 618ff60e06a2a91c3caffe58ac19d9ff45b61190 Mon Sep 17 00:00:00 2001 From: Svjatoslav Agejenko Date: Sun, 20 Sep 2026 05:29:37 +0300 Subject: [PATCH] Parallel radix sort for large render queues MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit The radix path now uses the paint executor: key build, the 8 LSD passes (per-chunk histograms, digit-major/chunk-minor offsets, per-chunk scatter) and the final permute all run chunked in parallel. Stability is preserved exactly, so the output is bit-identical to the serial sort for any chunk count (pinned by a new RadixLongSortTest property test over sizes 0..250k x chunk counts 1..8) — golden-image determinism intact. Also deletes the now-unreachable parallelMergeSort/runSortTasks/ mergeRuns (~100 LOC) and routes the no-executor (headless) path through the serial radix instead of Arrays.parallelSort. Measured on a 24-core desktop (450k pairs, SortBench/SortSweep harnesses): sort phase ~7 ms serial -> ~4-5 ms parallel (1.2-1.8x; bandwidth-bound, frequency-capped test machine). Key-build/permute parallelization comes on top in the full pipeline. --- .../e3d/renderer/raster/RadixLongSort.java | 121 ++++++++++++ .../e3d/renderer/raster/RenderAggregator.java | 183 ++++++------------ .../renderer/raster/RadixLongSortTest.java | 48 +++++ 3 files changed, 230 insertions(+), 122 deletions(-) diff --git a/src/main/java/eu/svjatoslav/aukio/e3d/renderer/raster/RadixLongSort.java b/src/main/java/eu/svjatoslav/aukio/e3d/renderer/raster/RadixLongSort.java index 0c8d150..ed0f794 100644 --- a/src/main/java/eu/svjatoslav/aukio/e3d/renderer/raster/RadixLongSort.java +++ b/src/main/java/eu/svjatoslav/aukio/e3d/renderer/raster/RadixLongSort.java @@ -96,4 +96,125 @@ final class RadixLongSort { // 8 passes: after the final swap the sorted pairs are back in the // caller's keys/idx arrays. } + + /** + * Parallel variant of {@link #sortPairs}: each pass builds per-chunk + * histograms concurrently, then computes scatter offsets digit-major / + * chunk-minor (chunk t's elements precede chunk t+1's within a digit), + * then scatters per chunk concurrently. That offset order preserves + * LSD stability exactly, so the result is IDENTICAL to the serial + * sort for any chunk count — the work partitioning is invisible to + * the output. Tasks are recorded on the thread-activity timeline like + * the rest of the sort machinery. + * + * @param histScratch scratch for per-chunk histograms and scatter + * offsets, length ≥ {@code threads * 512} + * @param threads chunk count (1 = fall back to the serial sort) + */ + static void sortPairsParallel(final long[] keys, final int[] idx, final int n, + final long[] keyTmp, final int[] idxTmp, + final int[] histScratch, + final java.util.concurrent.ExecutorService executor, + final int threads) { + if (threads <= 1 || executor == null) { + sortPairs(keys, idx, n, keyTmp, idxTmp); + return; + } + long[] srcK = keys; + long[] dstK = keyTmp; + int[] srcI = idx; + int[] dstI = idxTmp; + final int chunk = (n + threads - 1) / threads; + for (int shift = 0; shift < 64; shift += 8) { + final int s = shift; + final long[] sk = srcK; + final long[] dk = dstK; + final int[] si = srcI; + final int[] di = dstI; + + // Phase A: per-chunk histograms (concurrent) + runChunks(executor, n, chunk, (t, from, to) -> { + java.util.Arrays.fill(histScratch, t * 256, t * 256 + 256, 0); + for (int i = from; i < to; i++) + histScratch[t * 256 + (int) ((sk[i] >>> s) & 0xFF)]++; + }); + + // Serial combine: digit-major, chunk-minor offsets — this is + // what keeps the parallel sort stable and bit-identical to + // the serial one. + int pos = 0; + final int offsetsBase = threads * 256; + for (int d = 0; d < 256; d++) + for (int t = 0; t < threads; t++) { + final int c = histScratch[t * 256 + d]; + histScratch[offsetsBase + t * 256 + d] = pos; + pos += c; + } + + // Phase B: per-chunk scatter (concurrent; each chunk owns its + // private offset row) + runChunks(executor, n, chunk, (t, from, to) -> { + final int base = offsetsBase + t * 256; + for (int i = from; i < to; i++) { + final int d = (int) ((sk[i] >>> s) & 0xFF); + final int p = histScratch[base + d]++; + dk[p] = sk[i]; + di[p] = si[i]; + } + }); + + final long[] tk = srcK; + srcK = dstK; + dstK = tk; + final int[] ti = srcI; + srcI = dstI; + dstI = ti; + } + } + + /** One unit of chunk work: chunk index t and its [from, to) range. */ + interface ChunkWork { + void run(int t, int from, int to); + } + + /** + * Runs {@code work} for every non-empty chunk concurrently on + * {@code executor} and awaits completion; each task is recorded as + * KIND_SORT on the thread-activity timeline. Package-visible: the + * aggregator reuses it for the parallel key-build and permute. + */ + static void runChunks(final java.util.concurrent.ExecutorService executor, + final int n, final int chunk, + final ChunkWork work) { + final java.util.List> futures = + new java.util.ArrayList<>(); + for (int t = 0, from = 0; from < n; t++, from += chunk) { + final int ti = t; + final int f = from; + final int to = Math.min(n, from + chunk); + futures.add(executor.submit(() -> { + final boolean trace = + eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.isEnabled(); + final long t0 = trace ? System.nanoTime() : 0; + try { + work.run(ti, f, to); + } finally { + if (trace) + eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.record( + eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.KIND_SORT, + t0, System.nanoTime()); + } + })); + } + try { + for (final java.util.concurrent.Future future : futures) + future.get(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted during parallel radix sort", e); + } catch (final java.util.concurrent.ExecutionException e) { + throw new RuntimeException("Task failed during parallel radix sort", + e.getCause()); + } + } } diff --git a/src/main/java/eu/svjatoslav/aukio/e3d/renderer/raster/RenderAggregator.java b/src/main/java/eu/svjatoslav/aukio/e3d/renderer/raster/RenderAggregator.java index ad82495..698a51b 100644 --- a/src/main/java/eu/svjatoslav/aukio/e3d/renderer/raster/RenderAggregator.java +++ b/src/main/java/eu/svjatoslav/aukio/e3d/renderer/raster/RenderAggregator.java @@ -83,7 +83,7 @@ public class RenderAggregator { /** Valid entries of {@link #queueArray} after a merge. */ private int pendingMergeCount; - /** Reusable merge-sort scratch (grow-only), see parallelMergeSort. */ + /** Reusable permutation scratch (grow-only), see tryRadixSort. */ private AbstractCoordinateShape[] sortScratch; /** Radix-sort scratch (grow-only): keys, indices, and their swap @@ -93,6 +93,8 @@ public class RenderAggregator { private long[] tiePack; private int[] radixIdx; private int[] radixIdxTmp; + /** Per-chunk histogram/offset scratch for the parallel radix passes. */ + private int[] radixHist; private static long[] ensureCapacity(final long[] array, final int capacity) { return (array != null && array.length >= capacity) @@ -122,8 +124,8 @@ public class RenderAggregator { } /** - * Above this many queued shapes, {@link #sort()} uses a parallel sort - * on the fork/join common pool instead of a single-threaded sort. + * Above this many queued shapes, {@link #sort()} uses the radix path + * (parallel when an executor is available) instead of a comparator sort. */ private static final int PARALLEL_SORT_THRESHOLD = 8192; @@ -137,12 +139,10 @@ public class RenderAggregator { } /** - * Sorts the queue by (Z, shapeId), using an instrumented parallel - * merge sort on the given executor for large queues. Unlike - * {@code Arrays.parallelSort}, every subtask is recorded on the - * thread-activity timeline, so the sort does not appear as phantom - * idle time on the worker rows. Deterministic: (Z, shapeId) is a - * total order, so any merge schedule yields the same result. + * Sorts the queue by (Z, shapeId). Large queues go through the radix + * path (see {@link #tryRadixSort}), which is parallel when an executor + * is given; small queues use a plain comparator sort. Deterministic: + * (Z, shapeId) is a total order, so every path yields the same result. * * @param executor executor for parallel sorting, or null for serial */ @@ -163,13 +163,8 @@ public class RenderAggregator { ensureCapacity(queueArray, sortedCount)); queueArray = sortedArray; } - if (executor != null && sortedCount >= PARALLEL_SORT_THRESHOLD) { - if (!tryRadixSort(sortedArray, sortedCount)) - parallelMergeSort(sortedArray, sortedCount, comparator, - executor); - } else if (sortedCount >= PARALLEL_SORT_THRESHOLD) { - Arrays.parallelSort(sortedArray, 0, sortedCount, - comparator); + if (sortedCount >= PARALLEL_SORT_THRESHOLD) { + tryRadixSort(sortedArray, sortedCount, executor); } else { Arrays.sort(sortedArray, 0, sortedCount, comparator); } @@ -184,22 +179,52 @@ public class RenderAggregator { * (Z descending, shapeId ascending) exactly, without a single * comparator call. Sequential memory throughout: key build and the * final permute stream the queue array, the radix passes stream - * long/int arrays. - * - * @return true when the radix path sorted the queue + * long/int arrays. With an executor the key build, radix passes and + * permute run chunked in parallel (bit-identical to serial: stability + * makes the partitioning invisible). */ - private boolean tryRadixSort(final AbstractCoordinateShape[] array, - final int length) { + private void tryRadixSort(final AbstractCoordinateShape[] array, + final int length, + final ExecutorService executor) { radixKeys = ensureCapacity(radixKeys, length); radixKeysTmp = ensureCapacity(radixKeysTmp, length); radixIdx = ensureCapacity(radixIdx, length); radixIdxTmp = ensureCapacity(radixIdxTmp, length); - for (int i = 0; i < length; i++) { - radixKeys[i] = RadixLongSort.zSortKey(array[i].getZ(slot)); - radixIdx[i] = i; + + // Sweet spot measured on a 24-core desktop (450k pairs, SortSweep + // harness): ~16-24 chunks; below ~4 chunks bandwidth stays + // underutilized, and every chunk costs 2 task submissions per pass. + final int threads = executor == null ? 1 + : Math.max(1, Math.min( + Runtime.getRuntime().availableProcessors(), + length / 16384)); + + final long[] keys = radixKeys; + final int[] idx = radixIdx; + if (threads > 1) { + RadixLongSort.runChunks(executor, length, (length + threads - 1) / threads, + (t, from, to) -> { + for (int i = from; i < to; i++) { + keys[i] = RadixLongSort.zSortKey(array[i].getZ(slot)); + idx[i] = i; + } + }); + } else { + for (int i = 0; i < length; i++) { + radixKeys[i] = RadixLongSort.zSortKey(array[i].getZ(slot)); + radixIdx[i] = i; + } } - RadixLongSort.sortPairs(radixKeys, radixIdx, length, - radixKeysTmp, radixIdxTmp); + + if (threads > 1) { + radixHist = ensureCapacity(radixHist, threads * 512); + RadixLongSort.sortPairsParallel(radixKeys, radixIdx, length, + radixKeysTmp, radixIdxTmp, radixHist, executor, threads); + } else { + RadixLongSort.sortPairs(radixKeys, radixIdx, length, + radixKeysTmp, radixIdxTmp); + } + // Equal-key runs must resolve by ascending shapeId (the // comparator's tie-break; queue order is NOT construction order). // Runs are almost always singletons — the pack array only @@ -225,104 +250,18 @@ public class RenderAggregator { runStart = runEnd; } sortScratch = ensureCapacity(sortScratch, length); - for (int i = 0; i < length; i++) - sortScratch[i] = array[radixIdx[i]]; - System.arraycopy(sortScratch, 0, array, 0, length); - return true; - } - - /** - * Parallel merge sort over our own executor: chunk the array, - * sort chunks concurrently, then merge runs pairwise in a tree — - * every task recorded as KIND_SORT on the activity timeline. - */ - private void parallelMergeSort(final AbstractCoordinateShape[] array, - final int length, - final Comparator cmp, - final ExecutorService executor) { - final int cores = Runtime.getRuntime().availableProcessors(); - final int runCount = Math.min(length, cores * 4); - final int runSize = (length + runCount - 1) / runCount; - - // Phase 1: sort runs concurrently - runSortTasks(array, length, cmp, executor, runCount, runSize); - - // Phase 2: pairwise merge tree, in-place into the REUSED scratch - // array (previously a fresh multi-MB array per frame) - AbstractCoordinateShape[] from = array; - sortScratch = ensureCapacity(sortScratch, length); final AbstractCoordinateShape[] scratch = sortScratch; - int width = runSize; - while (width < length) { - final int w = width; - final AbstractCoordinateShape[] src = from; - final AbstractCoordinateShape[] dst = (from == array) ? scratch : array; - final java.util.List> futures = new java.util.ArrayList<>(); - for (int start = 0; start < length; start += 2 * w) { - final int left = start; - final int mid = Math.min(start + w, length); - final int right = Math.min(start + 2 * w, length); - futures.add(executor.submit(() -> { - final boolean trace = eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.isEnabled(); - final long t0 = trace ? System.nanoTime() : 0; - try { - mergeRuns(src, dst, cmp, left, mid, right); - } finally { - if (trace) - eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.record( - eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.KIND_SORT, - t0, System.nanoTime()); - } - })); - } - awaitAll(futures, "parallel sort merge"); - from = dst; - width *= 2; - } - if (from != array) - System.arraycopy(from, 0, array, 0, length); - } - - /** Sorts {@code runCount} consecutive runs of the array concurrently. */ - private void runSortTasks(final AbstractCoordinateShape[] array, - final int length, - final Comparator cmp, - final ExecutorService executor, - final int runCount, final int runSize) { - final java.util.List> futures = new java.util.ArrayList<>(runCount); - for (int r = 0; r < runCount; r++) { - final int from = r * runSize; - final int to = Math.min(length, from + runSize); - if (from >= to) - break; - futures.add(executor.submit(() -> { - final boolean trace = eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.isEnabled(); - final long t0 = trace ? System.nanoTime() : 0; - try { - Arrays.sort(array, from, to, cmp); - } finally { - if (trace) - eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.record( - eu.svjatoslav.aukio.e3d.gui.ThreadActivityRecorder.KIND_SORT, - t0, System.nanoTime()); - } - })); + if (threads > 1) { + RadixLongSort.runChunks(executor, length, (length + threads - 1) / threads, + (t, from, to) -> { + for (int i = from; i < to; i++) + scratch[i] = array[radixIdx[i]]; + }); + } else { + for (int i = 0; i < length; i++) + sortScratch[i] = array[radixIdx[i]]; } - awaitAll(futures, "parallel sort runs"); - } - - /** Merges two adjacent sorted runs [left,mid) and [mid,right) into dst. */ - private static void mergeRuns(final AbstractCoordinateShape[] src, - final AbstractCoordinateShape[] dst, - final Comparator cmp, - final int left, final int mid, final int right) { - int i = left, j = mid, k = left; - while (i < mid && j < right) - dst[k++] = cmp.compare(src[i], src[j]) <= 0 ? src[i++] : src[j++]; - while (i < mid) - dst[k++] = src[i++]; - while (j < right) - dst[k++] = src[j++]; + System.arraycopy(sortScratch, 0, array, 0, length); } private static void awaitAll(final java.util.List> futures, final String what) { diff --git a/src/test/java/eu/svjatoslav/aukio/e3d/renderer/raster/RadixLongSortTest.java b/src/test/java/eu/svjatoslav/aukio/e3d/renderer/raster/RadixLongSortTest.java index 56180c0..98d318c 100644 --- a/src/test/java/eu/svjatoslav/aukio/e3d/renderer/raster/RadixLongSortTest.java +++ b/src/test/java/eu/svjatoslav/aukio/e3d/renderer/raster/RadixLongSortTest.java @@ -92,4 +92,52 @@ public class RadixLongSortTest { for (int i = 1; i < n; i++) assertTrue(Long.compareUnsigned(keys[i - 1], keys[i]) <= 0); } + + /** + * The parallel pair sort must produce output BIT-IDENTICAL to the + * serial one for any chunk count (stability makes the partitioning + * invisible) — golden-image determinism depends on it. + */ + @Test + public void parallelMatchesSerialBitExactly() throws Exception { + final java.util.concurrent.ExecutorService executor = + java.util.concurrent.Executors.newFixedThreadPool(4); + try { + final Random random = new Random(1234); + for (final int n : new int[]{0, 1, 7, 1000, 65536, 250000}) { + for (final int threads : new int[]{1, 2, 3, 8}) { + final long[] keys = new long[Math.max(n, 1)]; + final int[] idx = new int[Math.max(n, 1)]; + // duplicate-heavy + full-range mix exercises both + // stability and unsigned digit handling + for (int i = 0; i < n; i++) { + keys[i] = (i & 1) == 0 + ? random.nextInt(37) + : random.nextLong(); + idx[i] = i; + } + final long[] serialKeys = keys.clone(); + final int[] serialIdx = idx.clone(); + RadixLongSort.sortPairs(serialKeys, serialIdx, n, + new long[Math.max(n, 1)], new int[Math.max(n, 1)]); + + final long[] parKeys = keys.clone(); + final int[] parIdx = idx.clone(); + RadixLongSort.sortPairsParallel(parKeys, parIdx, n, + new long[Math.max(n, 1)], new int[Math.max(n, 1)], + new int[Math.max(threads, 1) * 512], + executor, threads); + + org.junit.Assert.assertArrayEquals( + "keys n=" + n + " threads=" + threads, + serialKeys, parKeys); + org.junit.Assert.assertArrayEquals( + "idx n=" + n + " threads=" + threads, + serialIdx, parIdx); + } + } + } finally { + executor.shutdownNow(); + } + } } -- 2.20.1