2
\$\begingroup\$

(See the previous and first iteration.)

Adaptive Mergesort

This algorithm is from a paper "Sublinear Merging and Natural Mergesort" by Svante Carlsson, Christos Levcopoulos and Ola Petersson. It is a natural mergesort that runs in \$\mathcal{O}(n \log m)\$ time, where \$n\$ is the length of the array range to sort, and \$m\$ is the number of runs in it. (Note that \$m\$ is always at most \$\lceil n/2 \rceil\$, which gives us the commonly known upper bound of \$\mathcal{O}(n \log n)\$.) The idea behind natural mergesort is scanning the input range in linear time in order to establish a queue of ascending and strictly descending runs. The strictly descending runs are reversed in-place into strictly ascending runs. We require the descending runs to be strictly descending in order to retain the stability of the sorting algorithm: if we reverse \$\langle 3, 2, 2', 1\rangle\$, the result is \$\langle 1, 2', 2, 3\rangle\$, which is not stable.

What comes to sublinear time merging, the algorithm uses exponential and binary searches in order to get to the "splitting" array component faster.

The below figures summarize the technique:

Adaptive Mergesort

What's new

I fixed the stability issue in the previous iteration. Also, I incorporated nice points by Peter Taylor

Code

AdaptiveMergesort.java

package net.coderodde.util;

import java.util.Arrays;
import java.util.Objects;

public final class AdaptiveMergesort {

    private AdaptiveMergesort() {}

    /**
     * Sorts stably the entire input array.
     * 
     * @param <T>   the array component type.
     * @param array the array to sort.
     */
    public static <T extends Comparable<? super T>> void sort(T[] array) {
        Objects.requireNonNull(array, "The input array is null.");
        sort(array, 0, array.length);
    }

    /**
     * Sorts stably the input subarray {@code array[fromIndex], 
     * array[fromIndex + 1], ..., array[toIndex - 2], array[toIndex - 1]}.
     * 
     * @param <T>       the array component type.
     * @param array     the array holding the target subarray.
     * @param fromIndex the index of the leftmost array component belonging to 
     *                  the requested array range.
     * @param toIndex   the index of the largest array component in the range   
     *                  plus one.
     */
    public static <T extends Comparable<? super T>> void sort(T[] array,
                                                              int fromIndex,
                                                              int toIndex) {
        Objects.requireNonNull(array, "The input array is null.");
        checkIndices(array.length, fromIndex, toIndex);

        int rangeLength = toIndex - fromIndex;

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

        T[] aux = Arrays.copyOfRange(array, fromIndex, toIndex);
        RunQueue queue = new RunLengthQueueBuilder<>(aux).run();

        // Number of runs not yet processed in the current merge pass over the
        // data:
        int runsLeft = queue.size();

        while (queue.size() > 1) {
            switch (runsLeft) {
                case 1:
                    // Bounce the lonely leftover run back to the tail of the 
                    // queue:
                    queue.enqueue(queue.dequeue());
                    // Fall through!

                case 0:
                    // Get to know how many runs there is to process in the 
                    // next merge pass:
                    runsLeft = queue.size();
                    continue;
            }

            // Remove the first two consecutive runs, merge them and append the
            // resulting merged run to the tail of the run queue:
            queue.enqueue(merge(aux, queue.dequeue(), queue.dequeue()));
            // Update the number of runs not yet processed in this merge pass:
            runsLeft -= 2;
        }

        // Put the elements in their correct positions such that the input array
        // range becomes stabily sorted:
        int arrayIndex = fromIndex;

        for (Interval interval = queue.dequeue().first; 
                interval != null; 
                interval = interval.next) {
            for (int i = interval.from; i <= interval.to; ++i) {
                array[arrayIndex++] = aux[i];
            }
        }
    }

    private static <T extends Comparable<? super T>> Run merge(T[] aux,
                                                               Run run1, 
                                                               Run run2) {
        Interval headInterval1 = run1.first;
        Interval headInterval2 = run2.first;
        Interval mergedRunHead = null;
        Interval mergedRunTail = null;

        // While both the left and right runs have intervals to offer, do:
        while (headInterval1 != null && headInterval2 != null) {
            T head1 = aux[headInterval1.from];
            T head2 = aux[headInterval2.from];

            if (head1.compareTo(head2) <= 0) {
                T tail1 = aux[headInterval1.to];

                if (tail1.compareTo(head2) <= 0) {
                    // Easy case, just append one interval to the other:
                    if (mergedRunHead == null) {
                        mergedRunHead = headInterval1;
                        mergedRunTail = headInterval1;
                    } else {
                        mergedRunTail.next = headInterval1;
                        headInterval1.prev = mergedRunTail;
                        mergedRunTail = headInterval1;
                    }

                    headInterval1 = headInterval1.next;
                    continue;
                }

                // Cannot append. We need to split the left interval:
                int index = findUpperBound(aux,
                                           headInterval1.from,
                                           headInterval1.to + 1,
                                           head2);

                Interval newInterval = new Interval(headInterval1.from,
                                                    index - 1);

                // Remove some head elements from first interval:
                headInterval1.from = index;

                // Append a split interval to the tail of the merged run:
                if (mergedRunHead == null) {
                    mergedRunHead = newInterval;
                    mergedRunTail = newInterval;
                } else {
                    mergedRunTail.next = newInterval;
                    newInterval.prev = mergedRunTail;
                    mergedRunTail = newInterval;
                }
            } else {
                T tail2 = aux[headInterval2.to];

                if (tail2.compareTo(head1) < 0) {
                    // Easy case, just append one interval to the other:
                    if (mergedRunHead == null) {
                        mergedRunHead = headInterval2;
                        mergedRunTail = headInterval2;
                    } else {
                        mergedRunTail.next = headInterval2;
                        headInterval2.prev = mergedRunTail;
                        mergedRunTail = headInterval2;
                    }

                    headInterval2 = headInterval2.next;
                    continue;
                }

                // Cannot append. We need to split the right interval:
                int index = findLowerBound(aux, 
                                           headInterval2.from,
                                           headInterval2.to + 1,
                                           head1);

                Interval newInterval = new Interval(headInterval2.from,
                                                    index - 1);

                // Remove some head elements from second interval:
                headInterval2.from = index; 

                // Append a split interval to the tail of the merge run:
                if (mergedRunHead == null) {
                    mergedRunHead = newInterval;
                    mergedRunTail = newInterval;
                } else {
                    mergedRunTail.next = newInterval;
                    newInterval.prev = mergedRunTail;
                    mergedRunTail = newInterval;
                }
            }
        }

        // Append the leftover intervals of a currently non-empty run to the
        // tail of the merged run:
        mergedRunTail.next = headInterval1 != null ? headInterval1 :
                                                     headInterval2;
        mergedRunTail.next.prev = mergedRunTail;
        mergedRunTail = mergedRunTail.next;

        // Reuse 'run1' in order not to abuse the heap memory too often:
        run1.first = mergedRunHead;
        run1.last = mergedRunTail;
        return run1;
    }

    private static void checkIndices(int arrayLength, 
                                     int fromIndex, 
                                     int toIndex) {
        if (fromIndex > toIndex) {
            throw new IllegalArgumentException(
                    "fromIndex(" + fromIndex + ") > toIndex(" + toIndex + ")");
        }

        if (fromIndex < 0) {
            throw new ArrayIndexOutOfBoundsException(
                    "fromIndex = " + fromIndex);
        }

        if (toIndex > arrayLength) {
            throw new ArrayIndexOutOfBoundsException(
                    "toIndex = " + toIndex);
        }
    }

    /**
     * This class represents a sorted ascending interval. In other words,
     * {@code aux[from], ..., aux[to]} is a sorted ascending sequence (block).
     */
    private static final class Interval {
        int from;
        int to;
        Interval prev;
        Interval next;

        Interval(int from, int to) {
            this.from = from;
            this.to = to;
        }
    }

    /**
     * Run represents a doubly-linked list of intervals such that the list
     * represents a sorted run.
     */
    private static final class Run {
        Interval first;
        Interval last;

        Run(int from, int to) {
            first = new Interval(from, to);
            last = first;
        }
    }

    /**
     * This class holds a queue of runs yet to merge.
     */
    private static final class RunQueue {

        private final Run[] runArray;
        // Used for bit level modulo arithmetic. Instead of
        // 'index % runArray.length' we can write 'index & mask'.
        private final int mask; 
        private int head;
        private int tail;
        private int size;

        RunQueue(int capacity) {
            capacity = ceilCapacityToPowerOfTwo(capacity);
            this.mask = capacity - 1;
            this.runArray = new Run[capacity];
        }

        void enqueue(Run run) {
            runArray[tail] = run;
            tail = (tail + 1) & mask;
            ++size;
        }

        /**
         * Extends the length of the tail run by {@code runLength} elements.
         * 
         * @param runLength the number of elements to add to the tail run.
         */
        void addToLastRun(int runLength) {
            runArray[(tail - 1) & mask].first.to += runLength;
        }

        Run dequeue() {
            Run run = runArray[head];
            head = (head + 1) & mask;
            --size;
            return run;
        }

        int size() {
            return size;
        }

        /**
         * If {@code capacity} is not a power of two, this method ceils it up 
         * towards the smallest power of two no less than {@code capacity}.
         * 
         * @param capacity the candidate capacity.
         * @return a smallest power of two no less than {@code capacity}.
         */
        private static int ceilCapacityToPowerOfTwo(int capacity) {
            int ret = Integer.highestOneBit(capacity);
            return ret != capacity ? ret << 1 : ret;
        }
    }

    private static final class 
            RunLengthQueueBuilder<T extends Comparable<? super T>> {

        private final RunQueue queue;
        private final T[] array;
        private int head;
        private int left;
        private int right;
        private final int last;
        private boolean previousRunWasDesending;

        RunLengthQueueBuilder(T[] array) {
            this.queue = new RunQueue((array.length >>> 1) + 1);
            this.array = array;
            this.left  = 0;
            this.right = 1;
            this.last  = array.length - 1;
        }

        RunQueue run() {
            while (left < last) {
                head = left;

                if (array[left++].compareTo(array[right++]) <= 0) {
                    // The next run is ascending:
                    scanAscendingRun();
                } else {
                    // The next run is descending:
                    scanDescendingRun();
                }

                ++left;
                ++right;
            }

            if (left == last) {
                // Deal with a single element run at the very tail of the input
                // array range:
                if (array[last - 1].compareTo(array[last]) <= 0) {
                    queue.addToLastRun(1);
                } else {
                    queue.enqueue(new Run(left, left));
                }
            }

            return queue;
        }

        void scanAscendingRun() {
            while (left < last && array[left].compareTo(array[right]) <= 0) {
                ++left;
                ++right;
            }

            Run run = new Run(head, left);

            if (previousRunWasDesending) {
                if (array[head - 1].compareTo(array[head]) <= 0) {
                    // We can just extend the previous run:
                    queue.addToLastRun(right - head);
                } else {
                    queue.enqueue(run);
                }
            } else {
                queue.enqueue(run);
            }

            previousRunWasDesending = false;
        }

        void scanDescendingRun() {
            while (left < last && array[left].compareTo(array[right]) > 0) {
                ++left;
                ++right;
            }

            Run run = new Run(head, left);
            reverseRun(array, head, left);

            if (previousRunWasDesending) {
                if (array[head - 1].compareTo(array[head]) <= 0) {
                    // We can just extend the previous run:
                    queue.addToLastRun(right - head);
                } else {
                    queue.enqueue(run);
                }
            } else {
                queue.enqueue(run);
            }

            previousRunWasDesending = true;
        }

        private void reverseRun(T[] array, int i, int j) {
            for (; i < j; ++i, --j) {
                T tmp = array[i];
                array[i] = array[j];
                array[j] = tmp;
            }
        }
    }

    /**
     * Returns the smallest index of an array component that does not compare 
     * less than {@code value}. 
     * 
     * @param <T>       the array component type.
     * @param array     the array holding the target range.
     * @param fromIndex the lowest index of the array range to process.
     * @param toIndex   the largest index of the array range to process plus
     *                  one.
     * @param value     the target value.
     * @return          the array index.
     */
    private static <T extends Comparable<? super T>> 
        int lowerBound(T[] array, int fromIndex, int toIndex, T value) {
        int count = toIndex - fromIndex;
        int it;

        while (count > 0) {
            it = fromIndex;
            int step = count >>> 1;
            it += step;

            if (array[it].compareTo(value) < 0) {
                fromIndex = ++it;
                count -= step + 1;
            } else {
                count = step;
            }
        }

        return fromIndex;
    }

    /**
     * Returns the smallest index of an array component that compares greater 
     * than {@code value}.
     * 
     * @param <T>       the array component type.
     * @param array     the array holding the target range.
     * @param fromIndex the lowest index of the array range to process.
     * @param toIndex   the largest index of the array range to process plus
     *                  one.
     * @param value     the target value.
     * @return          the array index.
     */
    private static <T extends Comparable<? super T>>
        int upperBound(T[] array, int fromIndex, int toIndex, T value) {
        int count = toIndex - fromIndex;
        int it;

        while (count > 0) {
            it = fromIndex; 
            int step = count >>> 1;
            it += step;

            if (array[it].compareTo(value) <= 0) {
                fromIndex = ++it;
                count -= step + 1;
            } else {
                count = step;
            }
        }

        return fromIndex;
    }

    private static <T extends Comparable<? super T>> 
        int findLowerBound(T[] array, int fromIndex, int toIndex, T value) {
        int bound = 1;
        int rangeLength = toIndex - fromIndex;

        // Do the exponential search in order to find faster the array subrange
        // that might contain 'value':
        while (bound < rangeLength &&
                array[bound + fromIndex].compareTo(value) < 0) {
            bound <<= 1;
        }

        // The containing range found. Now search in it with binary search:
        return lowerBound(array, 
                          fromIndex + (bound >>> 1), 
                          Math.min(toIndex, fromIndex + bound), 
                          value);
    }

    private static <T extends Comparable<? super T>> 
        int findUpperBound(T[] array, int fromIndex, int toIndex, T value) {
        int bound = 1;
        int rangeLength = toIndex - fromIndex;

        // Do the exponential search in order to find faster the array subrange
        // that might contain 'value':
        while (bound < rangeLength 
                && array[bound + fromIndex].compareTo(value) < 0) {
            bound <<= 1;
        }

        // The containing range found. Now search in it with binary search:
        return upperBound(array, 
                          fromIndex + (bound >>> 1), 
                          Math.min(toIndex, fromIndex + bound),
                          value);
    }
}

Demo.java

package net.coderodde.util;

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

public final class Demo {

    private static final int FROM_INDEX = 8;
    private static final int SKIP_RIGHT = 9;
    private static final int ARRAY_LENGTH = 50_000;
    private static final int BLOCKS = 1000;
    private static final int MIN_ELEMENT = -10_000;
    private static final int MAX_ELEMENT = 10_000;
    private static final int MAX_RUN_LENGTH = 100;
    private static final int RUNS = 1000;

    public static void main(String[] args) {
        long seed = System.currentTimeMillis();
        Random random = new Random(seed);
        System.out.println("Seed = " + seed);

        warmup(random);
        benchmark(random);
    }

    private static void warmup(Random random) {
        System.out.println("Warming up...");

        Integer[] array = getBlockedArray(ARRAY_LENGTH, BLOCKS, random);
        warmup(array);

        array = getRandomArray(ARRAY_LENGTH, random);
        warmup(array);

        array = getFunnyArray(ARRAY_LENGTH, random);
        warmup(array);

        array = getRunnyArray(ARRAY_LENGTH, RUNS, random);
        warmup(array);

        array = getZigZagArray(ARRAY_LENGTH);
        warmup(array);

        System.out.println("Warming up done!");
    }

    private static void benchmark(Random random) {
        Integer[] array = getBlockedArray(ARRAY_LENGTH, BLOCKS, random);
        System.out.println("\n--- Blocked array ---");
        benchmark(array);

        array = getRandomArray(ARRAY_LENGTH, random);
        System.out.println("\n--- Random array ----");
        benchmark(array);

        array = getFunnyArray(ARRAY_LENGTH, random);
        System.out.println("\n--- Funny array -----");
        benchmark(array);

        array = getRunnyArray(ARRAY_LENGTH, RUNS, random);
        System.out.println("\n--- Runny array -----");
        benchmark(array);

        array = getZigZagArray(ARRAY_LENGTH);
        System.out.println("\n--- Zig zag array ---");
        benchmark(array);
    }

    private static void warmup(Integer[] array1) {
        perform(false, array1);
    }

    private static void benchmark(Integer[] array1) {
        perform(true, array1);
    }

    private static void perform(boolean output, 
                                Integer[] array1) {
        Integer[] array2 = array1.clone();
        int length = array1.length;

        long startTime = System.currentTimeMillis();
        Arrays.sort(array1, FROM_INDEX, length - SKIP_RIGHT);
        long endTime = System.currentTimeMillis();

        if (output) {
            System.out.println("Arrays.sort in " + (endTime - startTime) + 
                               " milliseconds.");
        }

        startTime = System.currentTimeMillis();
        AdaptiveMergesort.sort(array2, FROM_INDEX, length - SKIP_RIGHT);
        endTime = System.currentTimeMillis();

        if (output) {
            System.out.println("AdaptiveMergesort.sort in " + 
                               (endTime - startTime) +
                               " milliseconds.");

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

    private static final Integer[] getBlockedArray(int length, 
                                                   int blocks,
                                                   Random random) {
        Integer[] array = getAscendingArray(length);
        blockify(array, blocks, random);
        return array;
    }

    private static final Integer[] getRandomArray(int length, Random random) {
        Integer[] array = new Integer[length];

        for (int i = 0; i < length; ++i) {
            array[i] = random.nextInt(MAX_ELEMENT - MIN_ELEMENT + 1) + MIN_ELEMENT;
        }

        return array;
    }

    private static final Integer[] getFunnyArray(int length, Random random) {
        Integer[] array = new Integer[length];

        int index = 0;

        while (index < array.length) {
            int remaining = array.length - index;
            int next = random.nextInt(MAX_RUN_LENGTH);
            int actual = Math.min(remaining, next);
            boolean direction = random.nextBoolean();

            Integer first = 
                    MIN_ELEMENT + 
                    random.nextInt(MAX_ELEMENT - MIN_ELEMENT + 1);

            array[index++] = first;
            int step = 1 + random.nextInt(5);

            if (direction) {
                for (int i = 1; i < actual; ++i) {
                    array[index++] = first + i * step;
                }
            } else {
                for (int i = 1; i < actual; ++i) {
                    array[index++] = first - i * step;
                }
            }
        }

        return array;
    }

    private static final Integer[] getRunnyArray(int length,
                                                 int runLength, 
                                                 Random random) {
        Integer[] array = getRandomArray(length, random);

        int index = 0;

        while (index < length) {
            int remaining = length - index;
            int requested = random.nextInt(runLength);
            int actual = Math.min(remaining, requested);

            Arrays.sort(array, index, index + actual);
            index += actual;
        }

        return array;
    }

    private static final Integer[] getAscendingArray(int length) {
        Integer[] array = new Integer[length];

        for (int i = 0; i < length; ++i) {
            array[i] = i;
        }

        return array;
    }

    private static final Integer[] getZigZagArray(int length) {
        Integer[] array = getAscendingArray(length);

        for (int i = 0; i + 1 < length; i += 2) {
            Integer tmp = array[i];
            array[i] = array[i + 1];
            array[i + 1] = tmp;
        }

        return array;
    }

    private static void blockify(Integer[] array,
                                 int numberOfBlocks, 
                                 Random random) {
        int blockSize = array.length / numberOfBlocks;
        Integer[][] blocks = new Integer[numberOfBlocks][];

        for (int i = 0; i < numberOfBlocks - 1; ++i) {
            blocks[i] = new Integer[blockSize];
        }

        blocks[numberOfBlocks - 1] = 
                new Integer[blockSize + array.length % blockSize];

        int index = 0;

        for (Integer[] block : blocks) {
            for (int i = 0; i < block.length; ++i) {
                block[i] = array[index++];
            }
        }

        shuffle(blocks, random);

        index = 0;

        for (Integer[] block : blocks) {
            for (int i = 0; i < block.length; ++i) {
                array[index++] = block[i];
            }
        }
    }

    private static void shuffle(Integer[][] blocks, Random random) {
        for (int i = 0; i < blocks.length; ++i) {
            int index1 = random.nextInt(blocks.length);
            int index2 = random.nextInt(blocks.length);
            Integer[] block = blocks[index1];
            blocks[index1] = blocks[index2];
            blocks[index2] = block;
        }
    }

    private static boolean arraysEqual(Integer[] array1, Integer[] array2) {
        if (array1.length != array2.length) {
            return false;
        }

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

        return true;
    }
}

Critique request

Please tell me anything that comes to mind.

\$\endgroup\$
  • 1
    \$\begingroup\$ Looks nice and tidy! First thing that comes to mind: I think you could use Java Microbenchmark Harness for you demo :) \$\endgroup\$ – RobAu Oct 18 '17 at 7:45
1
\$\begingroup\$

Figures 1 .. 3 are beautiful and educational, thank you for including them. Tiny nit on fig. 3: perhaps there could be some \qquad or &nbsp; mixed into the horizontal legend, so "(5, 5)" aligns more closely with the tallest item. Not looking for the precision of fig. 2, just closer is all.

package net.coderodde.util;

A fine name, but consider breaking out "net.coderodde.sort".

 * Sorts stably the entire input array.

I love one-line javadocs. "Stably" is a strong guarantee. I like that you viewed "adaptive merge" as obvious or redundant given the class name.

 * Sorts stably the input subarray {@code array[fromIndex], 
 * array[fromIndex + 1], ..., array[toIndex - 2], array[toIndex - 1]}.

Maybe the fromIndex + 1 and toIndex -2 improve clarity, but I feel they are verbose.

        return; // Trivially sorted.

Again, thank you for brief, accurate, helpful comments.

    // Number of runs not yet processed in the current merge pass over the
    // data:

The "in the current merge pass" phrase seems confusing / clunky, and "over the data" is redundant. The comment leaves me wondering about the semantics of runsLeft, and also wondering why it wasn't queue's responsibility to manage it. Ideally I'm looking for an invariant optionally evaluated by an assert.

                // Fall through!

Useful comment - thank you for the helpful reminder.

    // range becomes stabily sorted:

Typo: stably

    // While both the left and right runs have intervals to offer, do:

Thank you for this helpful comment.

        if (head1.compareTo(head2) <= 0) {

The if / else clauses are copy-n-paste-n-edit clauses, and are somewhat long. This code might benefit from a level of indirection so an index controls whether we're working on 1st or 2nd interval. Or at least push some of the code into helper functions?

    private static int ceilCapacityToPowerOfTwo(int capacity) {

This is generic math, which elsewhere happens to be applied to a capacity problem. I would prefer to see the input parameter named simply n, and to remove "capacity" from the name of the function.

    void scanAscendingRun() {

I suppose scanDescendingRun() might be merged with this via indexing, as well. But here the code duplication is less troubling, as each method is short and clear, one can view all the code without scrolling.

 * Returns the smallest index of an array component that does not compare 
 * less than {@code value}. 

Rather than "not compare", please use "greater or equal". (If you're sensitive to -0.0 < 0.0, and NaN, maybe reference the java8 sort or tack on a sentence as they do.) Please mention "binary search". I find "array element" more natural than "array component", but whatever. There is an ordering constraint on (some of) the array elements - please document that.

    int count = toIndex - fromIndex;

count is a bit vague. I prefer width, or the rangeLength identifier that you later used.

        int step = count >>> 1;

I don't understand why this is "better" than assigning count / 2. Do we have benchmark timings showing we should not trust the JIT?

Usual complaint: perhaps there is no need for upperBound(), if lowerBound() accepted another parameter? No biggie, as each is nicely compact. Similarly for find{Upper,Lower}Bound.

The exponential vs. binary search wrinkle is an interesting one, that might be worth a sentence or a paragraph of comments, or a pointer to a blog post that shows benchmark results.

Overall this is a solid piece of code. Ship it.

\$\endgroup\$
  • \$\begingroup\$ Also, what comes to shipping it, unfortunately there is no reason for doing that since Adaptive mergesort has impractically large constant factors hidden. Some algorithms are efficient on paper but not in practice, and that is one of them. \$\endgroup\$ – coderodde Oct 24 '17 at 10:08

Your Answer

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

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