3
\$\begingroup\$

I have this parallel implementation of MSD radix sort, which processes the entries by one particular byte. At each byte index, it has three phases:

  1. Count the bucket sizes.
  2. Insert each entry to its bucket.
  3. Recur on each resulting bucket, if there are less-significant bytes to process.

The only synchronization primitive in this implementation is joining the threads upon the ends of each phase 1, 2, 3.

CoderoddeArrays.java:

package net.coderodde.util;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;

public class CoderoddeArrays {

    private static final int BITS_PER_BUCKET = 8;
    private static final int BUCKETS = 1 << BITS_PER_BUCKET;
    private static final int BUCKET_MASK = BUCKETS - 1;
    private static final long SIGN_MASK = 1L << 63;
    private static final int THREAD_THRESHOLD = 65536;
    private static final int MERGESORT_THRESHOLD = 4096;

    public static <E> void parallelSort(final Entry<E>[] array) {
        parallelSort(array, 0, array.length);
    }

    public static <E> void parallelSort(final Entry<E>[] array,
                                        final int fromIndex,
                                        final int toIndex) {
        final int RANGE_LENGTH = toIndex - fromIndex;

        if (RANGE_LENGTH < 2) {
            return;
        }

        final Entry<E>[] buffer = array.clone();
        final int threads = Math.min(RANGE_LENGTH / THREAD_THRESHOLD, 
                                     Runtime.getRuntime()
                                            .availableProcessors());
        parallelSortImpl(array, buffer, threads, 0, fromIndex, toIndex);
    }

    public static final <E> boolean areEqual(final Entry<E>[]... arrays) {
        for (int i = 0; i < arrays.length - 1; ++i) {
            if (arrays[i].length != arrays[i + 1].length) {
                return false;
            }
        }

        for (int i = 0; i < arrays[0].length; ++i) {
            for (int j = 0; j < arrays.length - 1; ++j) {
                if (!Objects.equals(arrays[j][i], arrays[j + 1][i])) {
                    return false;
                }
            }
        }

        return true;
    }

    public static final <E extends Comparable<? super E>> 
        boolean isSorted(final E[] array, 
                         final int fromIndex,
                         final int toIndex) {
        for (int i = fromIndex; i < toIndex - 1; ++i) {
            if (array[i].compareTo(array[i + 1]) > 0) {
                return false;
            }
        }

        return true;
    }

    public static final <E extends Comparable<? super E>>
        boolean isSorted(final E[] array) {
        return isSorted(array, 0, array.length);       
    }

    private static final <E> void sortImpl(final Entry<E>[] source,
                                           final Entry<E>[] target,
                                           final int recursionDepth,
                                           final int fromIndex,
                                           final int toIndex) {
        // Try merge sort.
        if (toIndex - fromIndex <= MERGESORT_THRESHOLD) {
            mergesortAndCleanUp(source, 
                                target, 
                                recursionDepth, 
                                fromIndex,
                                toIndex);
            return;
        }

        final int[] bucketSizeMap = new int[BUCKETS];
        final int[] startIndexMap = new int[BUCKETS];
        final int[] processedMap  = new int[BUCKETS];

        // Compute the size of each bucket.
        for (int i = fromIndex; i < toIndex; ++i) {
            bucketSizeMap[getBucket(source[i].key(), recursionDepth)]++;
        }

        // Initialize the start index map.
        startIndexMap[0] = fromIndex;

        // Compute the start index map in its entirety.
        for (int i = 1; i != BUCKETS; ++i) {
            startIndexMap[i] = startIndexMap[i - 1] +
                               bucketSizeMap[i - 1];
        }

        // Insert the entries from 'source' into their respective 'target'.
        for (int i = fromIndex; i < toIndex; ++i) {
            final Entry<E> e = source[i];
            final int index = getBucket(source[i].key(), recursionDepth);
            target[startIndexMap[index] + processedMap[index]++] = e;
        }

        if (recursionDepth == 7) {
            // There is nowhere to recur, return.
            return;
        }

        // Recur to sort each bucket.
        for (int i = 0; i != BUCKETS; ++i) {
            if (bucketSizeMap[i] != 0) {
                sortImpl(target,
                         source,
                         recursionDepth + 1,
                         startIndexMap[i],
                         startIndexMap[i] + bucketSizeMap[i]);
            }
        }
    }

    private static final <E> boolean mergesort(final Entry<E>[] source,
                                               final Entry<E>[] target,
                                               final int fromIndex,
                                               final int toIndex) {
        final int RANGE_LENGTH = toIndex - fromIndex;

        Entry<E>[] s = source;
        Entry<E>[] t = target;

        int passes = 0;

        for (int width = 1; width < RANGE_LENGTH; width <<= 1) {
            ++passes;
            int c = 0;

            for (; c < RANGE_LENGTH / width; c += 2) {
                int left = fromIndex + c * width;
                int right = left + width;
                int i = left;

                final int leftBound = right;
                final int rightBound = Math.min(toIndex, right + width);

                while (left < leftBound && right < rightBound) {
                    t[i++] = s[right].key() < s[left].key() ?
                             s[right++] :
                             s[left++];
                }

                while (left < leftBound)   { t[i++] = s[left++]; }
                while (right < rightBound) { t[i++] = s[right++]; }
            }

            if (c * width < RANGE_LENGTH) {
                for (int i = fromIndex + c * width; i < toIndex; ++i) {
                    t[i] = s[i];
                }
            }

            final Entry<E>[] tmp = s;
            s = t;
            t = tmp;
        }

        return (passes & 1) == 0;
    }

    private static final <E> 
        void mergesortAndCleanUp(final Entry<E>[] source,
                                 final Entry<E>[] target,
                                 final int recursionDepth,
                                 final int fromIndex, 
                                 final int toIndex) {
        final boolean even = mergesort(source, target, fromIndex, toIndex);

        if (even) {
            // source contains the sorted range.
            if ((recursionDepth & 1) == 1) {
                // source is buffer, copy to target.
                System.arraycopy(source,
                                 fromIndex, 
                                 target,
                                 fromIndex, 
                                 toIndex - fromIndex);
            }
        } else {
            // target contains the sorted range.
            if ((recursionDepth & 1) == 0) {
                // target is buffer, copy to source.
                System.arraycopy(target, 
                                 fromIndex,
                                 source, 
                                 fromIndex, 
                                 toIndex - fromIndex);
            }
        }
    }

    private static final class BucketSizeCounter<E> extends Thread {

        int[] localBucketSizeMap;
        private final Entry<E>[] source;
        private final int recursionDepth;
        private final int fromIndex;
        private final int toIndex;

        BucketSizeCounter(final Entry<E>[] source,
                          final int recursionDepth,
                          final int fromIndex,
                          final int toIndex) {
            this.source = source;
            this.recursionDepth = recursionDepth;
            this.fromIndex = fromIndex;
            this.toIndex = toIndex;
        }

        @Override
        public void run() {
            this.localBucketSizeMap = new int[BUCKETS];

            for (int i = fromIndex; i < toIndex; ++i) {
                localBucketSizeMap[getBucket(source[i].key(), 
                                   recursionDepth)]++;
            }
        }
    }

    private static final class BucketInserter<E> extends Thread {

        private final int[] startIndexMap;
        private final int[] processedMap;
        private final Entry<E>[] source;
        private final Entry<E>[] target;
        private final int recursionDepth;
        private final int fromIndex;
        private final int toIndex;

        BucketInserter(final int[] startIndexMap,
                       final int[] processedMap,
                       final Entry<E>[] source,
                       final Entry<E>[] target,
                       final int recursionDepth,
                       final int fromIndex,
                       final int toIndex) {
            this.startIndexMap = startIndexMap;
            this.processedMap = processedMap;
            this.source = source;
            this.target = target;
            this.recursionDepth = recursionDepth;
            this.fromIndex = fromIndex;
            this.toIndex = toIndex;
        }

        @Override
        public void run() {
            for (int i = fromIndex; i < toIndex; ++i) {
                final Entry<E> e = source[i];
                final int index = getBucket(e.key(), recursionDepth);
                target[startIndexMap[index] + processedMap[index]++] = e;
            }
        }
    }

    private static final class Sorter<E> extends Thread {

        private final List<Task<E>> taskList;

        Sorter(final List<Task<E>> taskList) {
            this.taskList = taskList;
        }

        @Override
        public void run() {
            for (final Task task : taskList) {
                // Choose parallel or sequential.
                if (task.threads > 1) {
                    parallelSortImpl(task.source,
                                     task.target,
                                     task.threads,
                                     task.recursionDepth,
                                     task.fromIndex,
                                     task.toIndex);
                } else {
                    sortImpl(task.source,
                             task.target,
                             task.recursionDepth,
                             task.fromIndex,
                             task.toIndex);
                }
            }
        }
    }

    private static final class Task<E> {

        private final Entry<E>[] source;
        private final Entry<E>[] target;
        private final int threads;
        private final int recursionDepth;
        private final int fromIndex;
        private final int toIndex;

        Task(final Entry<E>[] source,
             final Entry<E>[] target,
             final int threads,
             final int recursionDepth,
             final int fromIndex,
             final int toIndex) {
            this.source = source;
            this.target = target;
            this.threads = threads;
            this.recursionDepth = recursionDepth;
            this.fromIndex = fromIndex;
            this.toIndex = toIndex;
        }
    }

    private static final <E> void parallelSortImpl(final Entry<E>[] source,
                                                   final Entry<E>[] target,
                                                   final int threads,
                                                   final int recursionDepth,
                                                   final int fromIndex,
                                                   final int toIndex) {
        final int RANGE_LENGTH = toIndex - fromIndex;

        if (RANGE_LENGTH <= MERGESORT_THRESHOLD) {
            mergesortAndCleanUp(source, 
                                target, 
                                recursionDepth, 
                                fromIndex, 
                                toIndex);
            return;
        }

        if (threads < 2) {
            sortImpl(source, target, recursionDepth, fromIndex, toIndex);
            return;
        }

        // Create the bucket size counter threads.
        final BucketSizeCounter[] counters = new BucketSizeCounter[threads];
        final int SUB_RANGE_LENGTH = RANGE_LENGTH / threads;
        int start = fromIndex;

        for (int i = 0; i != threads - 1; ++i, start += SUB_RANGE_LENGTH) {
            counters[i] = new BucketSizeCounter<>(source,
                                                  recursionDepth,
                                                  start,
                                                  start + SUB_RANGE_LENGTH);
            counters[i].start();
        }

        counters[threads - 1] = 
                new BucketSizeCounter<>(source,
                                        recursionDepth,
                                        start,
                                        toIndex);

        // Run the last counter in this thread while other are already on their
        // way.
        counters[threads - 1].run();

        try {
            for (int i = 0; i != threads - 1; ++i) {
                counters[i].join();
            }
        } catch (final InterruptedException ie) {
            ie.printStackTrace();
            return;
        }

        final int[] bucketSizeMap = new int[BUCKETS];
        final int[] startIndexMap = new int[BUCKETS];

        // Count the size of each processed bucket.
        for (int i = 0; i != threads; ++i) {
            for (int j = 0; j != BUCKETS; ++j) {
                bucketSizeMap[j] += counters[i].localBucketSizeMap[j];
            }
        }

        // Prepare the starting indices of each bucket.
        startIndexMap[0] = fromIndex;

        for (int i = 1; i != BUCKETS; ++i) {
            startIndexMap[i] = startIndexMap[i - 1] +
                               bucketSizeMap[i - 1];
        }

        // Create the inserter threads.
        final BucketInserter<E>[] inserters = new BucketInserter[threads - 1];
        final int[][] processedMaps = new int[threads][BUCKETS];

        // Make processedMaps of each thread independent of the other.
        for (int i = 1; i != threads; ++i) {
            int[] partialBucketSizeMap = counters[i - 1].localBucketSizeMap;

            for (int j = 0; j != BUCKETS; ++j) {
                processedMaps[i][j] = 
                        processedMaps[i - 1][j] + partialBucketSizeMap[j];
            }
        }

        int startIndex = fromIndex;

        for (int i = 0; i != threads - 1; ++i, startIndex += SUB_RANGE_LENGTH) {
            inserters[i] =
                    new BucketInserter<>(startIndexMap,
                                         processedMaps[i],
                                         source,
                                         target,
                                         recursionDepth,
                                         startIndex,
                                         startIndex + SUB_RANGE_LENGTH);
            inserters[i].start();
        }

        // Run the last inserter in this thread while other are on their ways.
        new BucketInserter<>(startIndexMap,
                             processedMaps[threads - 1],
                             source,
                             target,
                             recursionDepth,
                             startIndex,
                             toIndex).run();

        try {
            for (int i = 0; i != threads - 1; ++i) {
                inserters[i].join();
            }
        } catch (final InterruptedException ie) {
            ie.printStackTrace();
            return;
        }

        if (recursionDepth == 7) {
            // Nowhere to recur.
            return;
        }

        int nonEmptyBucketAmount = 0;

        for (int i : bucketSizeMap) {
            if (i != 0) {
                ++nonEmptyBucketAmount;
            }
        }

        final int SPAWN_DEGREE = Math.min(nonEmptyBucketAmount, threads);
        final List<Integer>[] bucketIndexListArray = new List[SPAWN_DEGREE];

        for (int i = 0; i != SPAWN_DEGREE; ++i) {
            bucketIndexListArray[i] = new ArrayList<>(nonEmptyBucketAmount);
        }

        final int[] threadCountMap = new int[SPAWN_DEGREE];

        for (int i = 0; i != SPAWN_DEGREE; ++i) {
            threadCountMap[i] = threads / SPAWN_DEGREE;
        }

        for (int i = 0; i != threads % SPAWN_DEGREE; ++i) {
            ++threadCountMap[i];
        }

        final List<Integer> nonEmptyBucketIndices = 
                new ArrayList<>(nonEmptyBucketAmount);


        for (int i = 0; i != BUCKETS; ++i) {
            if (bucketSizeMap[i] != 0) {
                nonEmptyBucketIndices.add(i);
            }
        }

        Collections.sort(nonEmptyBucketIndices, 
                         new BucketSizeComparator(bucketSizeMap));

        final int OPTIMAL_SUBRANGE_LENGTH = RANGE_LENGTH / SPAWN_DEGREE;
        int listIndex = 0;
        int packed = 0;
        int f = 0;
        int j = 0;

        while (j < nonEmptyBucketIndices.size()) {
            int tmp = bucketSizeMap[nonEmptyBucketIndices.get(j++)];
            packed += tmp;

            if (packed >= OPTIMAL_SUBRANGE_LENGTH
                    || j == nonEmptyBucketIndices.size()) {
                packed = 0;

                for (int i = f; i < j; ++i) {
                    bucketIndexListArray[listIndex]
                            .add(nonEmptyBucketIndices.get(i));
                }

                ++listIndex;
                f = j;
            }
        }

        final Sorter[] sorters = new Sorter[SPAWN_DEGREE];
        final List<List<Task<E>>> llt = new ArrayList<>(SPAWN_DEGREE);

        for (int i = 0; i != SPAWN_DEGREE; ++i) {
            final List<Task<E>> lt = new ArrayList<>();

            for (int idx : bucketIndexListArray[i]) {
                lt.add(new Task<>(target,
                                  source,
                                  threadCountMap[i],
                                  recursionDepth + 1,
                                  startIndexMap[idx],
                                  startIndexMap[idx] + bucketSizeMap[idx]));
            }

            llt.add(lt);
        }

        for (int i = 0; i != SPAWN_DEGREE - 1; ++i) {
            sorters[i] = new Sorter<>(llt.get(i));
            sorters[i].start();
        }

        new Sorter<>(llt.get(SPAWN_DEGREE - 1)).run();

        try {
            for (int i = 0; i != SPAWN_DEGREE - 1; ++i) {
                sorters[i].join();
            }
        } catch (final InterruptedException ie) {
            ie.printStackTrace();
            return;
        }
    }

    private static final class BucketSizeComparator 
    implements Comparator<Integer> {
        private final int[] bucketSizeMap;

        BucketSizeComparator(final int[] bucketSizeMap) {
            this.bucketSizeMap = bucketSizeMap;
        }

        @Override
        public int compare(final Integer i1, final Integer i2) {
            final int sz1 = bucketSizeMap[i1];
            final int sz2 = bucketSizeMap[i2];
            return sz2 - sz1;
        }
    }

    private static final int getBucket(final long key, 
                                       final int recursionDepth) {
        final int bitShift = 64 - (recursionDepth + 1) * BITS_PER_BUCKET;
        return (int)((key ^ SIGN_MASK) >>> bitShift) & BUCKET_MASK;
    }
}

Entry.java:

package net.coderodde.util;

public final class Entry<E> implements Comparable<Entry<E>> {

    private final long key;
    private final E satelliteData;

    public Entry(final long key, final E satelliteData) {
        this.key = key;
        this.satelliteData = satelliteData;
    }

    public long key() {
        return key;
    }

    public E satelliteData() {
        return satelliteData;
    }

    @Override
    public int compareTo(Entry<E> o) {
        return Long.compare(key, o.key);
    }
}

Demo.java:

package net.coderodde.util;

import java.util.Arrays;
import java.util.Random;

public class Demo {

    private static final int N = 10000000;

    public static void main(final String... args) {
        final long seed = System.currentTimeMillis();
        final Random rnd = new Random(seed);
        final Entry<Integer>[] array1 = getRandomEntryArray(N, rnd);
        final Entry<Integer>[] array2 = array1.clone();
        final Entry<Integer>[] array3 = array1.clone();

        System.out.println("Seed: " + seed);

        long ta = System.currentTimeMillis();
        net.coderodde.util.CoderoddeArrays.parallelSort(array1);
        long tb = System.currentTimeMillis();

        System.out.println("net.coderodde.util.CoderoddeArrays.parallelSort " +
                           "in " + (tb - ta) + " ms.");

        ta = System.currentTimeMillis();
        Arrays.parallelSort(array2);
        tb = System.currentTimeMillis();

        System.out.println("java.util.Arrays.parallelSort in " + 
                           (tb - ta) + " ms.");

        ta = System.currentTimeMillis();
        Arrays.sort(array3);
        tb = System.currentTimeMillis();

        System.out.println("java.util.Arrays.sort in " + (tb - ta) + " ms.");

        System.out.println("Arrays are equal: " + 
                           CoderoddeArrays.areEqual(array1, array2, array3));
        System.out.println("Sorted: " + CoderoddeArrays.isSorted(array1));
    }

    private static Entry<Integer>[] getRandomEntryArray(final int size,
                                                        final Random rnd) {
        final Entry<Integer>[] array = new Entry[size];

        for (int i = 0; i < size; ++i) {
            array[i] = new Entry<>(rnd.nextLong(), null);
        }

        return array;
    }
}

Any suggestions?

Improve this question
\$\endgroup\$

1 Answer 1

Reset to default
2
\$\begingroup\$

Very nice.

private static final int THREAD_THRESHOLD = 65536;
private static final int MERGESORT_THRESHOLD = 4096;

These seem arbitrary. THREAD_THRESHOLD is perhaps "infinite" with a sanity check, that's fine. But MERGESORT_THRESHOLD seems like it wants an accompanying comment, a reference that describes how some benchmark fared worse with higher or lower thresholds. I'm especially concerned that the optimal threshold would be sensitive to cache size, and no guidance is offered on how to compare the system I'm running on today with the historic benchmark system.

A broader observation is that there's no guidance offered for how, on data of interest, this sort stacks up against the competition, such as in https://docs.oracle.com/javase/8/docs/api/java/util/Arrays.html .

Rather than using Runtime.getRuntime().availableProcessors(), consider assigning that (as default value) to some config parameter. This offers control to a caller that already has some busy BG threads.

The final loops in areEqual() are fairly pedestrian. Even though modern JITs can do wonders, it seems like Arrays or some other module should offer direct access to memcmp(), since we just need to know if every bit of src matches the bits of dst. I have a concern about whether the current approach of iterating first over array items, then over arrays, would be as cache friendly as the order memcmp() would use. Benching it both ways would be instructive.

ParallelSortImpl might break out helpers to explicitly name each of the three phases. In three places it attempts .join() and then:

    } catch (final InterruptedException ie) {
        ie.printStackTrace();
        return;
    }

It seems likely this allows corrupted (unsorted) results to be passed back to callers. Consider, at a minimum, setting a dirty flag in the handler. Then top-level return to caller can consult flag, do final linear pass to check that all is sorted, and raise exception if not. Or, simply re-throw ie as a RuntimeException if this is a "can't happen" clause that you wrote just to keep the compiler happy about checked exceptions.

Improve this answer
\$\endgroup\$
1
  • \$\begingroup\$ A nice review once again, J_H! Unfortunately, my life became more hectic and (most unfortunate) codereview is no longer my first priority. I might rip off some time to actually reiterate the radix sort in question, yet, alas, cannot guarantee that either. \$\endgroup\$
    – coderodde
    Mar 29, 2018 at 16:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.