并行化合并排序时出现内存不足错误

Question

我尝试parallelize 我的merge sort 实现：http://pastebin.com/2uMGjTxr。 我想创建 Java-VM 可以提供的尽可能多的线程。我想使用 java.lang.Runtime 确定可能的最大线程数。

所以我想出了一个名为 MergeThread 的类：

public class MergeThread implements Runnable{

    public int[] list;
    int sIndex, eIndex;

    public MergeThread(int[] pArray, int pStartIndex, int pEndIndex){
        list = pArray;
        sIndex = pStartIndex;
        eIndex = pEndIndex;
    }

    public void run(){
        list = mergeSort(list, sIndex, eIndex);
    }

    /**
     * Merges two sorted int array into one new sorted array.
     * @param lhs
     * @param rhs
     * @return
     */
    private static int[] merge(int[] lhs, int[] rhs) {
        int[] result = new int[lhs.length + rhs.length];

        int leftIndex = 0;
        int rightIndex = 0;
        while(leftIndex < lhs.length && rightIndex < rhs.length) {
            if(lhs[leftIndex] <= rhs[rightIndex]) {
                result[leftIndex + rightIndex] = lhs[leftIndex];
                leftIndex++;
            } else {
                result[leftIndex + rightIndex] = rhs[rightIndex];
                rightIndex++;
            }
        }

        while(leftIndex < lhs.length) {
            result[leftIndex + rightIndex] = lhs[leftIndex];
            leftIndex++;
        }

        while(rightIndex < rhs.length) {
            result[leftIndex + rightIndex] = rhs[rightIndex];
            rightIndex++;
        }

        return result;
    }

    /**
     * Sorts an array from index <code>startIndex</code> (inclusive) to <code>endIndex</code> (exclusive).
     * @param array
     * @param startIndex
     * @param endIndex
     * @return new array that is sorted
     */
    private static int[] mergeSort(int[] array, int startIndex, int endIndex) {
        int length = endIndex - startIndex;
        if(length == 0) {
            return new int[]{};
        }
        if(length == 1) {
            return new int[]{array[startIndex]};
        }

        int halfLength = length / 2;
        //int[] sortedLeftPart = mergeSort(array, startIndex, startIndex + halfLength);
        MergeThread m1 = new MergeThread(array, startIndex, startIndex + halfLength);
        Thread t1 = new Thread(m1);
        t1.start();
        //int[] sortedRightPart = mergeSort(array, startIndex + halfLength, endIndex);
        MergeThread m2 = new MergeThread(array, startIndex + halfLength, endIndex);
        Thread t2 = new Thread(m2);
        t2.start();
        try{
        t1.join();
        t2.join();
        }catch(InterruptedException e){}
        return merge(m1.list, m2.list);     
    }
}

还有一个真正启动进程的类

import java.util.Random;

public class Aufg2 {
    public static Random random = new Random(100);

    public static void main(String[] args) {
        int[] array = createRandomArray(10000000);

        long time = System.currentTimeMillis();

        int[] sortedArray = sort(array);

        if(sortedArray.length != array.length || !isSorted(sortedArray)) {
            System.err.println("Failed to sort given array! :-(");
            return;
        }       
        System.out.println("Success! Sorting took " + (System.currentTimeMillis() - time) + "ms.");     
    }

    /**
     * Creates a randomly filled array of given length
     * @param length
     * @return
     */
    private static int[] createRandomArray(int length) {
        int[] result = new int[length];
        for(int i = 0; i < length; i++) {
            result[i] = random.nextInt();
        }
        return result;
    }

    /**
     * Checks whether a given int array is sorted in ascending order  
     * @param array
     * @return <code>true</code> if the given int array is sorted; <code>false</code> otherwise.
     */
    private static boolean isSorted(int[] array) {
        for(int i = 1; i < array.length; i++) {
            if(array[i] < array[i-1]) {
                return false;
            }
        }
        return true;
    }   

    /**
     * Sorts a given array (ascending order)
     * @param array
     * @return
     */
    private static int[] sort(int[] array){
        //TODO: use multiple threads to speed up the sorting
        MergeThread m = new MergeThread(array, 0, array.length);

        try{

        Thread t1 = new Thread(m);
        t1.start();
        t1.join();
        }catch(InterruptedException e){

        }
        return m.list;
    }
}

但是这种合并排序不起作用。控制台打印了很多java.lang.OutOfMemmoryError's unable to create new native thread。

稍后消息会更改为 java heap 之类的内容。

要使合并排序正常工作，我必须进行哪些更改以及如何使用 java.lang.Runtime 来实现？

Answer 1

分治机制让您尝试创建类似 5000000 个线程的东西 - 每个线程都需要默认 256KB (IIRC) 的堆栈内存。仍然对为什么收到OutOfMemmoryError 感到惊讶？

通过使用fixed size thread pool 来限制线程数 - 对池中的线程数进行一些实验，但任何比系统中的内核数多得多的东西都不太可能提高性能（并且可能确实会减少它）。

Answer 2

首先使用 ExecutorService 并在其中排队新任务，而不是创建数百万个线程（这应该可以解决第一个问题；如果创建数百万个线程，您迟早会耗尽资源）。 1.5 倍的内核数量通常是一个不错的猜测（通常比使用可用的内核数量提供更好的结果 - 但这是您必须尝试的）。

然后-如果您希望此算法在任何地方都具有高性能，则绝​​对重要-在合理的阈值下对叶案例使用快速排序，如果您想要较低的阈值，则使用插入排序（如果您使用插入排序，叶节点大小为 16左右应该可以正常工作）。

Answer 3

让一个线程处理数组的后半部分，而调用线程处理前半部分

    int halfLength = length / 2;
    MergeThread m2 = new MergeThread(array, startIndex + halfLength, endIndex);
    Thread t2 = new Thread(m2);
    t2.start();//let new thread handle the second half
    array = mergeSort(array, startIndex, startIndex + halfLength);//do first half ourselves
    try{
    t2.join();
    }catch(InterruptedException e){}
    return merge(array, m2.list);

这将创建的线程数量减少了一半

但快速排序更适合并行化，因为它不需要后递归步骤，允许线程（具有执行程序的可运行作业）在委派后立即返回

调用者只需要观察所有工作何时完成