java - 为什么我的自下而上合并排序在 Java 中这么慢？

Question

我花了几个小时试图弄清楚为什么我的排序算法的 Java 版本比递归合并排序慢两倍，因为 C 和 C++ 版本要快 40-50%。我一直在删除越来越多的代码，直到我将所有内容都剥离成一个简单的循环并合并，但它仍然慢了一倍。为什么这只在 Java 中这么慢？

作为引用，自下而上的合并排序可能如下所示:

public static <T> void sort(T[] a, T[] aux, Comparator<T> comp) {
    int N = a.length;
    for (int n = 1; n < N; n = n+n)
        for (int i = 0; i < N-n; i += n+n)
            merge(a, aux, i, i+n-1, Math.min(i+n+n-1, N-1), comp);
}

这是递归版本:

public static <T> void sort(T[] a, T[] aux, int lo, int hi, Comparator<T> comp) {
    int mid = lo + (hi - lo) / 2;
    sort(a, aux, lo, mid, comp);
    sort(a, aux, mid + 1, hi, comp);
    merge(a, aux, lo, mid, hi, comp);
}

这些基本上只是从算法中复制的 on this website .作为最后的手段，我想我会从网上复制和粘贴一些东西，但它的速度也是递归版本的两倍。

Java 是否有我遗漏的“特殊”之处？

编辑:根据要求，这里有一些代码:

import java.util.*;
import java.lang.*;
import java.io.*;

class Test {
    public int value;
    public int index;
}

class TestComparator implements Comparator<Test> {
    public int compare(Test a, Test b) {
        if (a.value < b.value) return -1;
        if (a.value > b.value) return 1;
        return 0;
    }
}


class Merge<T> {
    private static <T> void Merge(T[] array, int start, int mid, int end, Comparator<T> comp, T[] buffer) {
        java.lang.System.arraycopy(array, start, buffer, 0, (mid - start));
        int A_count = 0, B_count = 0, insert = 0;
        while (A_count < (mid - start) && B_count < (end - mid)) {
            if (comp.compare(array[mid + B_count], buffer[A_count]) >= 0)
                array[start + insert++] = buffer[A_count++];
            else
                array[start + insert++] = array[mid + B_count++];
        }
        java.lang.System.arraycopy(buffer, A_count, array, start + insert, (mid - start) - A_count);
    }

    private static <T> void SortR(T[] array, int start, int end, T[] buffer, Comparator<T> comp) {
        if (end - start <= 2) {
            if (end - start == 2) {
                if (comp.compare(array[start], array[end - 1]) > 0) {
                    T swap = array[start];
                    array[start] = array[end - 1];
                    array[end - 1] = swap;
                }
            }

            return;
        }

        int mid = start + (end - start)/2;
        SortR(array, start, mid, buffer, comp);
        SortR(array, mid, end, buffer, comp);
        Merge(array, start, mid, end, comp, buffer);
    }

    public static <T> void Recursive(T[] array, Comparator<T> comp) {
        @SuppressWarnings("unchecked")
        T[] buffer = (T[]) new Object[array.length];
        SortR(array, 0, array.length, buffer, comp);
    }

    public static <T> void BottomUp(T[] array, Comparator<T> comp) {
        @SuppressWarnings("unchecked")
        T[] buffer = (T[]) new Object[array.length];

        int size = array.length;
        for (int index = 0; index < size - 1; index += 2) {
            if (comp.compare(array[index], array[index + 1]) > 0) {
                T swap = array[index];
                array[index] = array[index + 1];
                array[index + 1] = swap;
            }
        }

        for (int length = 2; length < size; length += length)
            for (int index = 0; index < size - length; index += length + length)
                Merge(array, index, index + length, Math.min(index + length + length, size), comp, buffer);
    }
}


class SortRandom {
    public static Random rand;
    public static int nextInt(int max) {
        // set the seed on the random number generator
        if (rand == null) rand = new Random();
        return rand.nextInt(max);
    }
    public static int nextInt() {
        return nextInt(2147483647);
    }
}

class Sorter {
    public static void main (String[] args) throws java.lang.Exception {
        int max_size = 1500000;
        TestComparator comp = new TestComparator();

        for (int total = 0; total < max_size; total += 2048 * 16) {
            Test[] array1 = new Test[total];
            Test[] array2 = new Test[total];

            for (int index = 0; index < total; index++) {
                Test item = new Test();

                item.value = SortRandom.nextInt();
                item.index = index;

                array1[index] = item;
                array2[index] = item;
            }

            double time1 = System.currentTimeMillis();
            Merge.BottomUp(array1, comp);
            time1 = System.currentTimeMillis() - time1;

            double time2 = System.currentTimeMillis();
            Merge.Recursive(array2, comp);
            time2 = System.currentTimeMillis() - time2;

            if (time1 >= time2)
                System.out.format("%f%% as fast\n", time2/time1 * 100.0);
            else
                System.out.format("%f%% faster\n", time2/time1 * 100.0 - 100.0);

            System.out.println("verifying...");
            for (int index = 0; index < total; index++) {
                if (comp.compare(array1[index], array2[index]) != 0) throw new Exception();
                if (array2[index].index != array1[index].index) throw new Exception();
            }
            System.out.println("correct!");
        }
    }
}

这是一个 C++ 版本:

#include <iostream>
#include <cassert>
#include <cstring>
#include <ctime>

class Test {
public:
    size_t value, index;
};

bool TestCompare(Test item1, Test item2) {
    return (item1.value < item2.value);
}

namespace Merge {
    template <typename T, typename Comparison>
    void Merge(T array[], int start, int mid, int end, Comparison compare, T buffer[]) {
        std::copy(&array[start], &array[mid], &buffer[0]);
        int A_count = 0, B_count = 0, insert = 0;
        while (A_count < (mid - start) && B_count < (end - mid)) {
            if (!compare(array[mid + B_count], buffer[A_count]))
                array[start + insert++] = buffer[A_count++];
            else
                array[start + insert++] = array[mid + B_count++];
        }
        std::copy(&buffer[A_count], &buffer[mid - start], &array[start + insert]);
    }

    template <typename T, typename Comparison>
    void SortR(T array[], int start, int end, T buffer[], Comparison compare) {
        if (end - start <= 2) {
            if (end - start == 2)
                if (compare(array[end - 1], array[start]))
                    std::swap(array[start], array[end - 1]);
            return;
        }

        int mid = start + (end - start)/2;
        SortR(array, start, mid, buffer, compare);
        SortR(array, mid, end, buffer, compare);
        Merge(array, start, mid, end, compare, buffer);
    }

    template <typename T, typename Comparison>
    void Recursive(T array[], int size, Comparison compare) {
        T *buffer = new T[size];
        SortR(array, 0, size, buffer, compare);
        delete[] buffer;
    }

    template <typename T, typename Comparison>
    void BottomUp(T array[], int size, Comparison compare) {
        T *buffer = new T[size];

        for (int index = 0; index < size - 1; index += 2) {
            if (compare(array[index + 1], array[index]))
                std::swap(array[index], array[index + 1]);
        }

        for (int length = 2; length < size; length += length)
            for (int index = 0; index < size - length; index += length + length)
                Merge(array, index, index + length, std::min(index + length + length, size), compare, buffer);

        delete[] buffer;
    }
}

int main() {
    srand(time(NULL));
    int max_size = 1500000;
    for (int total = 0; total < max_size; total += 2048 * 16) {
        Test *array1 = new Test[total];
        Test *array2 = new Test[total];

        for (int index = 0; index < total; index++) {
            Test item;
            item.value = rand();
            item.index = index;

            array1[index] = item;
            array2[index] = item;
        }

        double time1 = clock() * 1.0/CLOCKS_PER_SEC;
        Merge::BottomUp(array1, total, TestCompare);
        time1 = clock() * 1.0/CLOCKS_PER_SEC;

        double time2 = clock() * 1.0/CLOCKS_PER_SEC;
        Merge::Recursive(array2, total, TestCompare);
        time2 = clock() * 1.0/CLOCKS_PER_SEC;

        if (time1 >= time2)
           std::cout << time2/time1 * 100.0 << "% as fast" << std::endl;
        else
            std::cout << time2/time1 * 100.0 - 100.0 << "% faster" << std::endl;

        std::cout << "verifying... ";
        for (int index = 0; index < total; index++) {
            assert(array1[index].value == array2[index].value);
            assert(array2[index].index == array1[index].index);
        }
        std::cout << "correct!" << std::endl;

        delete[] array1;
        delete[] array2;
    }
    return 0;
}

差异并不像原始版本那么大，但 C++ 迭代版本更快，而 Java 迭代版本更慢。

(是的，我意识到这些版本有点糟糕并且分配的内存比使用的更多)

更新 2: 当我将自下而上的合并排序切换为后序遍历时，它与递归版本中的数组访问顺序非常匹配，它最终开始比递归版本快 10%递归版本。所以看起来它与缓存未命中有关，而不是微基准测试或不可预测的 JVM。

它只影响Java版本的原因可能是因为Java缺少C++版本中使用的自定义值类型。我将在 C++ 版本中单独分配所有测试类，看看性能会发生什么变化。我正在研究的排序算法不能轻易适应这种类型的遍历，但如果 C++ 版本的性能也很差，我可能没有太多选择。

更新 3: 不，将 C++ 版本切换到分配的类似乎对其性能没有任何明显影响。它看起来确实是由 Java 引起的。

Answer 1

有趣的问题。我不明白为什么 bottomUp 版本比递归慢，而数组大小为 2 的幂时它们的工作方式相同。

至少 bottomUp 慢了一点点，而不是两倍。

Benchmark                             Mode          Mean   Mean error    Units
RecursiveVsBottomUpSort.bottomUp      avgt        64.436        0.376    us/op
RecursiveVsBottomUpSort.recursive     avgt        58.902        0.552    us/op

代码:

@OutputTimeUnit(TimeUnit.MICROSECONDS)
@BenchmarkMode(Mode.AverageTime)
@Warmup(iterations = 5, time = 1)
@Measurement(iterations = 10, time = 1)
@State(Scope.Thread)
@Threads(1)
@Fork(1)
public class RecursiveVsBottomUpSort {

    static final int N = 1024;
    int[] a = new int[N];
    int[] aux = new int[N];

    @Setup(Level.Invocation)
    public void fill() {
        Random r = ThreadLocalRandom.current();
        for (int i = 0; i < N; i++) {
            a[i] = r.nextInt();
        }
    }

    @GenerateMicroBenchmark
    public static int bottomUp(RecursiveVsBottomUpSort st) {
        int[] a = st.a, aux = st.aux;
        int N = a.length;
        for (int n = 1; n < N; n = n + n) {
            for (int i = 0; i < N - n; i += n + n) {
                merge(a, aux, i, i + n - 1, Math.min(i + n + n - 1, N - 1));
            }
        }
        return a[N - 1];
    }

    @GenerateMicroBenchmark
    public static int recursive(RecursiveVsBottomUpSort st) {
        sort(st.a, st.aux, 0, N - 1);
        return st.a[N - 1];
    }

    static void sort(int[] a, int[] aux, int lo, int hi) {
        if (lo == hi)
            return;
        int mid = lo + (hi - lo) / 2;
        sort(a, aux, lo, mid);
        sort(a, aux, mid + 1, hi);
        merge(a, aux, lo, mid, hi);
    }

    static void merge(int[] a, int[] aux, int lo, int mid, int hi) {
        System.arraycopy(a, lo, aux, lo, mid + 1 - lo);

        for (int j = mid+1; j <= hi; j++)
            aux[j] = a[hi-j+mid+1];

        int i = lo, j = hi;
        for (int k = lo; k <= hi; k++)
            if (aux[j] < aux[i]) a[k] = aux[j--];
            else                      a[k] = aux[i++];
    }
}