1
\$\begingroup\$

Introduction

In this post, I will present a parallel sorting algorithm for sorting primitive integer arrays.

Treesort

Treesort is an algorithm which iterates over the input array and constructs a binary search tree from the array components. As soon as the input range is processed, it traverses the nodes in-order and dumps them into the input array. The algorithm I am presenting here relies on treesort. However, instead of building a balanced binary search tree, I shuffle the input range in linear time, and build an unbalanced tree, which on average should have a logarithmic height (according to a lemma in Introduction to Algorithms book). Also, I am making two optimisations:

  1. In each node, I cache the number of occurrences of its key.
  2. I maintain a hashtable mapping each key to its corresponding tree node.

The above arrangement allows me to reduce the total tree construction phase to time \$\mathcal{O}(k \log k)\$ (where \$k\$ is the number of distinct integers), assuming that shuffling was not bad.

The algorithm

The actual algorithm computes some "reasonable" amount of threads \$T\$, splits the input range into \$T\$ contiguous subsequences, conquers them, and finally merges them.

Performance

The resulting sort is not comparable to Arrays.parallelSort on average, yet is more efficient on arrays of size around 2 000 000 elements with relatively small set of distinct integers. For example, I get the following performance figures on arrays of 2 million elements and 11 000 distinct values in the array:


[STATUS] Warming up...
[STATUS] Warming up done.
Seed = 168244858505017
ParallelTreesort in 76 milliseconds.
Arrays.parallelSort in 240 milliseconds.
Algorithms agree: true

Implementation

The code snippet follows:

ParallelTreesort.java:

package net.coderodde.util;

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

/**
 * This class implements a parallel tree sort for primitive integer arrays. The 
 * algorithm splits the input range into a particular number of substrings, 
 * sorts each in its own thread using an unbalanced tree sort algorithm, and 
 * merges the resulting sorted substrings. 
 * 
 * @author Rodion "rodde" Efremov
 * @version 1.6 (May 23, 2016)
 */
public class ParallelTreesort {

    private static final int MINIMUM_THREAD_WORK_LOAD = 1;

    private ParallelTreesort() {}

    public static void sort(final int[] array) {
        sort(array, 0, array.length);
    }

    public static void sort(final int[] array, 
                            final int fromIndex, 
                            final int toIndex) {
        final int rangeLength = toIndex - fromIndex;

        if (rangeLength < 2) {
            // Trivially sorted.
            return;
        }

        final int cores = Runtime.getRuntime().availableProcessors();
        final int commitThreads = 
                Math.max(1, rangeLength / MINIMUM_THREAD_WORK_LOAD);

        int tentativeThreads = Math.min(cores, commitThreads);
        // Make sure that the number of threads is a power of two after all.
        tentativeThreads = fixThreadCount(tentativeThreads);

        final TreesortThread[] sorterThreads =
          new TreesortThread[tentativeThreads];

        int tmpFromIndex = fromIndex;
        final int basicChunkSize = rangeLength / tentativeThreads;

        for (int i = 0; i < sorterThreads.length - 1; ++i) {
            sorterThreads[i] =
                    new TreesortThread(array,
                                       tmpFromIndex,
                                       tmpFromIndex += basicChunkSize);
            sorterThreads[i].start();
        }

        sorterThreads[sorterThreads.length - 1] = 
                new TreesortThread(array, tmpFromIndex, toIndex);

        // Run the last sorter thread in this thread.
        sorterThreads[sorterThreads.length - 1].run();

        for (int i = 0; i < sorterThreads.length - 1; ++i) {
            try {
                sorterThreads[i].join();
            } catch (final InterruptedException ex) {
                throw new IllegalStateException(
                        "" + sorterThreads[i].getClass().getSimpleName() +
                        " \"" + sorterThreads[i].getName() + "\" " +
                        "threw an " + ex.getClass().getSimpleName(), ex);
            }
        }

        if (sorterThreads.length == 1) {
            // Single threaded sorting; no need to merge sort results from 
            // multiple threads.
            return;
        }

        final int[] aux = Arrays.copyOfRange(array, fromIndex, toIndex);

        new MergerThread(aux,
                         array,
                         0,
                         fromIndex,
                         sorterThreads,
                         0,
                         sorterThreads.length).run();
    }

    private static int getNumberOfMergePasses(final int threads) {
        return 32 - Integer.numberOfLeadingZeros(threads - 1);
    }

    private static int fixThreadCount(final int threads) {
        int ret = 1;

        while (ret < threads) {
            ret <<= 1;
        }

        return ret;
    }

    private static final class TreesortThread extends Thread {

        private final int[] array;
        private final int fromIndex;
        private final int toIndex;

        private TreeNode root;
        private final HashTableEntry[] hashtable;
        private final int mask;
        private final int rangeLength;

        TreesortThread(final int[] array,
                       final int fromIndex,
                       final int toIndex) {
            this.array       = array;
            this.fromIndex   = fromIndex;
            this.toIndex     = toIndex;
            this.rangeLength = toIndex - fromIndex;

            final int tableCapacity = fixCapacity(rangeLength);

            this.mask = tableCapacity - 1;
            this.hashtable = new HashTableEntry[tableCapacity];
        }

        int getFromIndex() {
            return fromIndex;
        }

        int getToIndex() {
            return toIndex;
        }

        int getRunLength() {
            return toIndex - fromIndex;
        }

        @Override
        public void run() {
            shuffle();
            constructTree();
            dump();
        }

        private void shuffle() {
            final Random random = new Random();
            final int rangeLength = toIndex - fromIndex;
            final int to = fromIndex + rangeLength / 2;

            for (int i = fromIndex; i < to; ++i) {
                final int randomIndex = fromIndex + random.nextInt(rangeLength);
                swap(i, randomIndex);
            }
        }

        private void constructTree() {
            final int initialKey = array[fromIndex];
            root = new TreeNode(initialKey);
            hashtable[getHashTableIndex(initialKey)] =
                    new HashTableEntry(initialKey, root, null);

            for (int i = fromIndex + 1; i < toIndex; ++i) {
                final int currentArrayComponent = array[i];
                final int hashtableIndex = 
                        getHashTableIndex(currentArrayComponent);

                final HashTableEntry entry = 
                        findEntry(currentArrayComponent,
                                  hashtable[hashtableIndex]);

                if (entry != null) {
                    entry.treeNode.count++;
                } else {
                    final TreeNode newnode = 
                            new TreeNode(currentArrayComponent);

                    hashtable[hashtableIndex] = 
                            new HashTableEntry(currentArrayComponent,
                                               newnode,
                                               hashtable[hashtableIndex]);

                    insertTreeNode(newnode);
                }
            }
        }

        private void dump() {
            int index = fromIndex;
            TreeNode node = root.getMinimum();

            while (node != null) {
                final int count = node.count;
                final int key   = node.key;

                for (int i = 0; i < count; ++i) {
                    array[index++] = key;
                }

                node = node.getSuccessor();
            }
        }

        private void insertTreeNode(final TreeNode node) {
            final int key = node.key;

            TreeNode current = root;
            TreeNode parentOfCurrent = null;

            while (current != null) {
                parentOfCurrent = current;

                if (key < current.key) {
                    current = current.left;
                } else {
                    // We don't check 'key > current.key' as there is no risk
                    // of duplicate keys in the tree.
                    current = current.right;
                }
            }

            if (key < parentOfCurrent.key) {
                parentOfCurrent.left = node;
            } else {
                parentOfCurrent.right = node;
            }

            node.parent = parentOfCurrent;
        }

        private HashTableEntry 
            findEntry(final int key, final HashTableEntry collisionChainHead) {
            HashTableEntry currentEntry = collisionChainHead;

            while (currentEntry != null && currentEntry.key != key) {
                currentEntry = currentEntry.next;
            }

            return currentEntry;
        }

        private int fixCapacity(final int capacity) {
            int ret = 1;

            while (ret < capacity) {
                ret <<= 1;
            }

            return ret;
        }

        private int getHashTableIndex(final int key) {
            return key & mask;
        }

        private void swap(final int index1, final int index2) {
            final int tmp = array[index1];
            array[index1] = array[index2];
            array[index2] = tmp;
        }

        private static final class TreeNode {
            TreeNode left;
            TreeNode right;
            TreeNode parent;

            final int key;
            int count = 1;

            TreeNode(final int key) {
                this.key = key;
            }

            TreeNode getMinimum() {
                TreeNode minimumNode = this;

                while (minimumNode.left != null) {
                    minimumNode = minimumNode.left;
                }

                return minimumNode;
            }

            TreeNode getSuccessor() {
                if (this.right != null) {
                    return this.right.getMinimum();
                }

                TreeNode parentNode  = this.parent;
                TreeNode currentNode = this;

                while (parentNode != null && parentNode.right == currentNode) {
                    currentNode = parentNode;
                    parentNode  = parentNode.parent;
                }

                return parentNode;
            }
        }

        private static final class HashTableEntry {
            int key;
            TreeNode treeNode;
            HashTableEntry next;

            HashTableEntry(final int key,
                           final TreeNode treeNode,
                           final HashTableEntry next) {
                this.key = key;
                this.treeNode = treeNode;
                this.next = next;
            }
        }
    }

    private static final class MergerThread extends Thread {

        private final int[] source;
        private final int[] target;
        private final int sourceOffset;
        private final int targetOffset;
        private final TreesortThread[] threads;
        private final int threadStartIndex;
        private final int threadEndIndex;

        MergerThread(final int[] source,
                     final int[] target,
                     final int sourceOffset,
                     final int targetOffset,
                     final TreesortThread[] threads,
                     final int threadStartIndex,
                     final int threadEndIndex) {
            this.source           = source;
            this.target           = target;
            this.sourceOffset     = sourceOffset;
            this.targetOffset     = targetOffset;
            this.threads          = threads;
            this.threadStartIndex = threadStartIndex;
            this.threadEndIndex   = threadEndIndex;
        }

        @Override
        public void run() {
            final int threadCount = threadEndIndex - threadStartIndex;

            if (threadCount == 1) {
                return;
            }

            final MergerThread leftMergerThread =
              new MergerThread(target,
                               source,
                               targetOffset,
                               sourceOffset,
                               threads,
                               threadStartIndex,
                               threadStartIndex + threadCount / 2);

            final int leftMergerThreadCoverage = 
                    leftMergerThread.getCoverageLength();

            final MergerThread rightMergerThread =
              new MergerThread(target,
                               source,
                               targetOffset + leftMergerThreadCoverage,
                               sourceOffset + leftMergerThreadCoverage,
                               threads,
                               threadStartIndex + threadCount / 2,
                               threadEndIndex);

            rightMergerThread.start();
            leftMergerThread.run();

            try {
                rightMergerThread.join();
            } catch (final InterruptedException ex) {
                throw new IllegalStateException(
                        "" + rightMergerThread.getClass().getSimpleName() +
                        " \"" + rightMergerThread.getName() + "\" " +
                        "threw an " + ex.getClass().getSimpleName(), ex);

            }

            final int leftRunLength  = leftMergerThread.getCoverageLength();
            final int rightRunLength = rightMergerThread.getCoverageLength();

            int left  = sourceOffset;
            int right = sourceOffset + leftRunLength;

            final int leftEnd  = right;
            final int rightEnd = right + rightRunLength;

            int targetIndex = targetOffset;

            while (left < leftEnd && right < rightEnd) {
                target[targetIndex++] = 
                        source[right] < source[left] ?
                        source[right++] :
                        source[left++];
            }

            System.arraycopy(source, 
                             left, 
                             target, 
                             targetIndex, 
                             leftEnd - left);

            System.arraycopy(source, 
                             right, 
                             target, 
                             targetIndex, 
                             rightEnd - right);
        }

        private int getCoverageLength() {
            return threads[threadEndIndex - 1].getToIndex() - 
                   threads[threadStartIndex].getFromIndex();
        }
    }

    private static final class Warmup {
        static final int ITERATIONS = 100;
        static final int ARRAY_LENGTH = 500_000;
        static final int MINIMUM_VALUE = -100;
        static final int MAXIMUM_VALUE = 100;
    }

    private static final class Demo {
        static final int ARRAY_LENGTH = 2_000_000;
        static final int MINIMUM_VALUE = -1000;
        static final int MAXIMUM_VALUE = 10_000;
        static final int FROM_INDEX = 10;
        static final int TO_INDEX = ARRAY_LENGTH - 15;
    }

    public static void main(final String... args) {
        final long seed = System.nanoTime();
        final Random random = new Random(seed);
        final int[] array1 = random.ints(Demo.ARRAY_LENGTH, 
                                         Demo.MINIMUM_VALUE, 
                                         Demo.MAXIMUM_VALUE).toArray();
        final int[] array2 = array1.clone();

        System.out.println("[STATUS] Warming up...");
        warmup(random);
        System.out.println("[STATUS] Warming up done.");

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

        long startTime = System.nanoTime();
        sort(array1, Demo.FROM_INDEX, Demo.TO_INDEX);
        long endTime = System.nanoTime();

        System.out.printf("ParallelTreesort in %.0f milliseconds.\n",
                          (endTime - startTime) / 1e6);

        startTime = System.nanoTime();
        Arrays.parallelSort(array2, Demo.FROM_INDEX, Demo.TO_INDEX);
        endTime = System.nanoTime();

        System.out.printf("Arrays.parallelSort in %.0f milliseconds.\n",
                          (endTime - startTime) / 1e6);

        System.out.println("Algorithms agree: " + Arrays.equals(array1,
                                                                array2));
    }

    private static void warmup(final Random random) {
        for (int i = 0; i < Warmup.ITERATIONS; ++i) {
            final int[] array1 = random.ints(Warmup.ARRAY_LENGTH,
                                             Warmup.MINIMUM_VALUE, 
                                             Warmup.MAXIMUM_VALUE).toArray();

            final int[] array2 = array1.clone();

            ParallelTreesort.sort(array1);
            Arrays.parallelSort(array2);
        }
    }
}

Critique request

I would like to receive critique on naming conventions, coding style, API design, and, especially, optimization opportunities.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.