Bottom up mergesort in Java

Question

See the next iteration.

I have this iterative mergesort (a.k.a. bottom-up mergesort).

BottomUpMergesort.java:

package net.coderodde.util.sorting;

import java.util.Arrays;
import java.util.Comparator;

/**
 * This class provides static methods for sorting object arrays using 
 * bottom-up merge sort. The algorithm used is a bottom-up merge sort.
 * 
 * @author Rodion "rodde" Efremov
 * @version 1.6
 */
public class BottomUpMergesort {

    /**
     * Specifies the maximum length of a chunk which is sorted using insertion
     * sort.
     */
    private static final int INSERTIONSORT_THRESHOLD = 13;

    /**
     * Sorts the range {@code array[fromIndex], array[fromIndex + 1], ...,
     * array[toIndex - 2], array[toIndex - 1]}.
     * 
     * @param <T>       the actual array component type.
     * @param array     the array holding the target range.
     * @param fromIndex the starting (inclusive) index of the range to sort.
     * @param toIndex   the ending (exclusive) index of the range to sort. 
     * @param cmp       the object comparator.
     */
    public static <T> void sort(T[] array,
                                int fromIndex,
                                int toIndex,
                                Comparator<? super T> cmp) {
        if (toIndex - fromIndex < 2) {
            // Trivially sorted or indices ass-basckwards.
            return;
        }

        // Create the auxiliary buffer.
        int rangeLength = toIndex - fromIndex;
        T[] buffer = Arrays.copyOfRange(array, fromIndex, toIndex);

        // Find out how many merge passes we need to do over the input range.
        int runs = rangeLength / INSERTIONSORT_THRESHOLD +
                  (rangeLength % INSERTIONSORT_THRESHOLD != 0 ? 1 : 0);
        int mergePasses = getMergePassAmount(runs);

        // Set up the state.
        T[] source;
        T[] target;
        int sourceOffset;
        int targetOffset;

        if (mergePasses % 2 == 0) {
            // If here, we will do an even amount of merge passes.
            source = array;
            target = buffer;
            sourceOffset = fromIndex;
            targetOffset = 0;
        } else {
            // If here, we will do an odd amount of merge passes.
            source = buffer;
            target = array;
            sourceOffset = 0;
            targetOffset = fromIndex;
        }

        // Create the initial runs.
        for (int i = 0; i < runs - 1; ++i) {
            int tmpIndex = sourceOffset + i * INSERTIONSORT_THRESHOLD;
            insertionSort(source, 
                          tmpIndex, 
                          tmpIndex + INSERTIONSORT_THRESHOLD,
                          cmp);
        }

        // Do not forget the last (the righmost) run. Note, that the length of
        // the last run may vary between 1 and INSERTIONS_SORT_THRESHOLD, 
        // inclusively.
        int lastRunStartIndex = sourceOffset + (runs - 1) * 
                                INSERTIONSORT_THRESHOLD;
        insertionSort(source, 
                      lastRunStartIndex,
                      Math.min(lastRunStartIndex + INSERTIONSORT_THRESHOLD, 
                               sourceOffset + rangeLength),
                      cmp);

        // Initial runs are ready to be merged. 'runWidth <<= 1' multiplies
        // 'runWidth' by 2.
        for (int runWidth = INSERTIONSORT_THRESHOLD; 
                 runWidth < rangeLength;
                 runWidth <<= 1) {

            int runIndex = 0;

            for (; runIndex < runs - 1; runIndex += 2) {
                // Set up the indices.
                int leftIndex = sourceOffset + runIndex * runWidth;
                int leftBound = leftIndex + runWidth;
                int rightBound = Math.min(leftBound + runWidth, 
                                          rangeLength + sourceOffset);
                int targetIndex = targetOffset + runIndex * runWidth;

                // Perform the actual merging.
                merge(source,
                      target,
                      leftIndex,
                      leftBound,
                      rightBound,
                      targetIndex,
                      cmp);
            }

            if (runIndex < runs) {
                // 'runIndex' is the index of the "orphan" run at the end of the
                // range being sorted. Since it may appear in the opposite 
                // array as two non-merged runs, we have to simply copy this 
                // orphan run to the opposite array.
                System.arraycopy(source,
                                 sourceOffset + runIndex * runWidth,
                                 target,
                                 targetOffset + runIndex * runWidth,
                                 rangeLength - runIndex * runWidth);
            }

            runs = (runs >>> 1) + (runs % 2 == 0 ? 0 : 1);
            // Change the roles of the arrays.
            T[] tmparr = source;
            source = target;
            target = tmparr;
            int tmp = sourceOffset;
            sourceOffset = targetOffset;
            targetOffset = tmp;
        }
    }

    /**
     * Sorts the entire array.
     * 
     * @param <T>   the array component type.
     * @param array the array to sort.
     * @param cmp   the comparator.
     */
    public static <T> void sort(T[] array, Comparator<? super T> cmp) {
        sort(array, 0, array.length, cmp);
    }

    /**
     * Sorts the range {@code array[fromIndex,], array[fromIndex + 1], ...,
     * array[toIndex - 2], array[toIndex - 1]} using insertion sort. This 
     * implementation is <b>stable</b>.
     * 
     * @param <T>       the array component type.
     * @param array     the array holding the requested range.
     * @param fromIndex the starting (inclusive) index.
     * @param toIndex   the ending (exclusive) index.
     * @param cmp       the array component comparator.
     */
    public static <T> void insertionSort(T[] array,
                                         int fromIndex,
                                         int toIndex,
                                         Comparator<? super T> cmp) {
        for (int i = fromIndex + 1; i < toIndex; ++i) {
            T element = array[i];
            int j = i;

            for (; j > fromIndex && cmp.compare(array[j - 1], element) > 0; --j) {
                array[j] = array[j - 1];
            }

            array[j] = element;
        }
    }

    /**
     * Returns the amount of merge passes needed to sort a range containing 
     * {@code runs} runs. (A run is any contiguous, strictly descending or
     * ascending subsequence.
     * 
     * @param  runs the amount of runs in the target range.
     * @return the amount of needed merge passes.
     */
    private static int getMergePassAmount(int runs) {
        return 32 - Integer.numberOfLeadingZeros(runs - 1);
    }

    /**
     * Merges the sorted ranges {@code source[leftIndex, leftBound)} and
     * {@code source[rightIndex, rightBound)} putting the result to 
     * {@code target} starting from component with index {@code targetIndex}.
     * 
     * @param <T>         the array component type.
     * @param source      the source array.
     * @param target      the target array.
     * @param leftIndex   the (inclusive) starting index of the left run.
     * @param leftBound   the (exclusive) ending index of the left run.
     * @param rightIndex  the (inclusive) starting index of the right run.
     * @param rightBound  the (exclusive) ending index of the right run.
     * @param targetIndex the starting index of the result run in the target
     *                     array.
     * @param cmp         the element comparator.
     */
    private static <T> void merge(T[] source,
                                  T[] target,
                                  int leftIndex,
                                  int leftBound,
                                  int rightBound,
                                  int targetIndex,
                                  Comparator<? super T> cmp) {
        int rightIndex = leftBound;

        while (leftIndex < leftBound && rightIndex < rightBound) {
            target[targetIndex++] = 
                    cmp.compare(source[rightIndex], source[leftIndex]) < 0 ?
                    source[rightIndex++] :
                    source[leftIndex++];
        }

        System.arraycopy(source, 
                         leftIndex, 
                         target, 
                         targetIndex, 
                         leftBound - leftIndex);

        System.arraycopy(source, 
                         rightIndex, 
                         target, 
                         targetIndex, 
                         rightBound - rightIndex);
    }
}

Utils.java:

package net.coderodde.util.sorting;

import java.util.Comparator;
import java.util.Random;

/**
 * This class contains some static utility methods for working with arrays.
 * 
 * @author Rodion "rodde" Efremov
 * @version 1.6
 */
public class Utils {


    /**
     * Tests whether the range {@code array[fromIndex], array[fromIndex + 1],
     * ..., array[toIndex - 2], array[toIndex - 1]} is sorted into ascending 
     * order as specified by {@code cmp}.
     * 
     * @param <T>       the array component type.
     * @param array     the array holding the target range.
     * @param fromIndex the starting (inclusive) index.
     * @param toIndex   the ending (exclusive) index.
     * @param cmp       the element comparator.
     * @return          {@code true} only if the requested range is sorted.
     */
    public static <T> boolean isSorted(T[] array,
                                       int fromIndex,
                                       int toIndex,
                                       Comparator<? super T> cmp) {
        for (int i = fromIndex; i < toIndex - 1; ++i) {
            if (cmp.compare(array[i], array[i + 1]) > 0) {
                return false;
            }
        }

        return true;
    }

    /**
     * Tests whether the array {@code array} is sorted into ascending order as
     * specified by {@code cmp}.
     * 
     * @param <T>   the array component type.
     * @param array the array holding the target range.
     * @param cmp   the element comparator.
     * @return      {@code true} only if the entire array is sorted.
     */
    public static <T> boolean isSorted(T[] array, Comparator<? super T> cmp) {
        return isSorted(array, 0, array.length, cmp);
    }

    /**
     * Returns {@code true} if the two input arrays are of the same length, and
     * both have identical array components.
     * 
     * @param <T>  the array component type.
     * @param arr1 the first array.
     * @param arr2 the second array.
     * @return {@code true} if the two arrays have identical contents.
     */
    public static <T> boolean arraysIdentical(T[] arr1, T[] arr2) {
        if (arr1.length != arr2.length) {
            return false;
        }

        for (int i = 0; i < arr1.length; ++i) {
            if (arr1[i] != arr2[i]) {
                return false;
            }
        }

        return true;
    }

    /**
     * This method creates a random array of integers.
     * 
     * @param size the length of the result array.
     * @param random the instance of {@link java.util.Random}.
     * @return a random array.
     */
    public static Integer[] createRandomIntegerArray(int size, Random random) {
        Integer[] ret = new Integer[size];

        for (int i = 0; i < size; ++i) {
            ret[i] = random.nextInt();
        }

        return ret;
    }
}

Demo.java:

import java.util.Arrays;
import java.util.Random;
import static net.coderodde.util.sorting.Utils.isSorted;
import static net.coderodde.util.sorting.Utils.arraysIdentical;
import static net.coderodde.util.sorting.Utils.createRandomIntegerArray;
import net.coderodde.util.sorting.BottomUpMergesort;

public class Demo {

    private static final int SIZE = 2000000;

    public static void main(String[] args) {
        long seed = System.currentTimeMillis();
        Random random = new Random(seed);
        Integer[] array1 = createRandomIntegerArray(SIZE, random);
        Integer[] array2 = array1.clone();
        System.out.println("Seed: " + seed);

        //// java.util.Arrays.sort
        long ta = System.currentTimeMillis();
        Arrays.sort(array1, Integer::compare);
        long tb = System.currentTimeMillis();

        System.out.println(
                "java.util.Arrays.sort() in " + (tb - ta) + " ms. Sorted: " +
                isSorted(array1, 2, 9, Integer::compare));

        //// net.coderodde.util.sorting.BottomUpMergesort.sort
        ta = System.currentTimeMillis();
        BottomUpMergesort.sort(array2, Integer::compare);
        tb = System.currentTimeMillis();

        System.out.println(
                "net.coderodde.util.sorting.BottomUpMergesort.sort() " +
                (tb - ta) + " ms. Sorted: " + 
                isSorted(array2, 2, 9, Integer::compare));

        System.out.println(
                "Arrays identical: " + arraysIdentical(array1, array2));
    }
}

And the entire story is here:

git clone [email protected]:coderodde/BottomUpMergesort.git && cd BottomUpMergesort && mvn test && mvn exec:java

So, what do you think?

Answer 1

Generally speaking, this looks pretty good. A few comments:

Method Length

Your primary method, sort, is really, really long. A sign of a method that is too long is if you feel the need to comment what's happening rather than just creating a new private method. For example, you could refactor the code that handles the rightmost run to this:

sortRightmostRun(cmp, rangeLength, runs, source, sourceOffset);

See? No need for a comment now, it's very clear what that line does. Though of course you can include it in the method itself.

private static <T> void sortRightmostRun(Comparator<? super T> cmp,
    int rangeLength, int runs, T[] source, int sourceOffset) {
  // Do not forget the last (the righmost) run. Note, that the length of
  // the last run may vary between 1 and INSERTIONS_SORT_THRESHOLD, 
  // inclusively.
  int lastRunStartIndex = sourceOffset + (runs - 1) * 
                          INSERTIONSORT_THRESHOLD;
  insertionSort(source, 
                lastRunStartIndex,
                Math.min(lastRunStartIndex + INSERTIONSORT_THRESHOLD, 
                         sourceOffset + rangeLength),
                cmp);
}

Commenting of private methods

This may be a bit of a controversial topic, but I prefer not to comment private methods. This goes hand in hand with the first part; code can change, but if you don't remember to also change the comments, comments can become out of date. This can lead to the classic example

/**
 * Always returns true.
 */
public boolean isAvailable() {
    return false;
}

That code probably did, at one time, return true. The point is, with private methods:

• Only future developers are ever going to look at them
• They create additional work that must be maintained when refactoring
• Javadoc is generally not generated for private methods
• If the code itself is not clear, it is better to refactor to make it clear by using descriptive method names rather than using a comment to do the explanation.

Use of static methods

I try to avoid static methods as much as possible. It's always better to create an instance, and then have the instance do the work. This way, you can configure the instance separately, and reuse different versions. For example, you have this:

private static final int INSERTIONSORT_THRESHOLD = 13;

You could, with an instance based system, do this:

private static final int DEFAULT_THRESHOLD = 13;
private final int insertionSortThreshold;

public BottomUpMergeSort() {
    this(DEFAULT_THRESHOLD);
}

public BottomUpMergeSort(int insertionSortThreshold) {
    this.insertionSortThreshold = insertionSortThreshold;
}

This would allow you to do more configuration for different types of runs.

Spelling errors

You have a few misspellings in the comments.

// Trivially sorted or indices ass-basckwards.
// Do not forget the last (the righmost) run. Note, that the length of