// 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<java.util.concurrent.Future<?>> 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());
+ }
+ }
}
/** 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
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)
}
/**
- * 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;
}
/**
- * 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
*/
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);
}
* (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
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<AbstractCoordinateShape> 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<Future<?>> 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<AbstractCoordinateShape> cmp,
- final ExecutorService executor,
- final int runCount, final int runSize) {
- final java.util.List<Future<?>> 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<AbstractCoordinateShape> 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<Future<?>> futures, final String what) {
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();
+ }
+ }
}
projects / aukio-3d.git / commitdiff