java - 并行对两个数组的每个元素求和

Question

我有两个二维数组，我想将它们逐个元素求和。两个数组的大小相同，行数和列数也相同)。它应该返回一个最终数组，其大小与逐个元素的总和相同。

如何使用 Java 的 Fork-Join 框架或一般的并行性来完成这样的任务？使用并行性来解决这个问题有意义吗？

下面是我对 Java 的 Fork-Join 框架未完成的尝试:

public class SumArray extends RecursiveTask<int[][]> {

    private static final int ROW_CUTOFF = 10;
    private static final int COL_CUTOFF = 10;

    int[][] left_;
    int[][] right_;
    int rowLo_;
    int rowHi_;
    int colLo_;
    int colHi_;

    SumArray(int[][] left, int[][] right, int rowLo, int rowHi, int colLo, int colHi) {
        left_ = left;
        right_ = right;
        rowLo_ = rowLo;
        rowHi_ = rowHi;
        colLo_ = colLo;
        colHi_ = colHi;
    }

    @Override
    protected int[][] compute() {
        if (rowHi_ - rowLo_ <= ROW_CUTOFF && colHi_ - colLo_ <= COL_CUTOFF) {
            for (int i = rowLo_; i < rowHi_; i++) {
                for (int j = colLo_; j < colHi_; j++) {
                    left_[i][j] += right_[i][j];
                }
            }
            return left_;
        }
        int rowMid = rowLo_ + ((rowHi_ - rowLo_) / 2); 
        int colMid = colLo_ + ((colHi_ - colLo_) / 2);
        SumArray topLeft = new SumArray(left_, right_, rowLo_, rowMid, colLo_, colMid);
            SumArray topRight = new SumArray(left_, right_, rowMid, rowHi_, colLo_, colMid);
            topLeft.fork()
            int[][] topRightSummed = topRight.compute();
            int[][] topLeftSummed = topLeft.join();
            // ???

我可以类似地找到左下角和右下角的数组，但是如何在保持并行性性能的同时连接这些数组？我应该使用共享内存吗？

Answer 1

在抛出线程解决此问题之前，请优化单核的使用。在这种情况下，CPU 缓存未命中会产生明显的差异。例如，考虑此示例代码，在一种情况下，它对 array[i][j] 和另一个 array[j][i] 中的值求和。其中一个的 CPU 缓存未命中次数要少得多，因此比另一个要快得多。以下代码可用于演示该行为。

public class Sum2D {

    public static void main( String[] args ) {
         int[][] data = createGrid(100);

        long sum = 0;
        long start1 = System.currentTimeMillis();
        for ( int i=0; i<100000; i++ ) {
            sum += sumAcrossFirst(data);
        }

        long end1 = System.currentTimeMillis();

        long start2 = System.currentTimeMillis();
        for ( int i=0; i<100000; i++ ) {
            sum += sumAcrossSecond(data);
        }

        long end2 = System.currentTimeMillis();

        double duration1 = (end1-start1)/1000.0;
        double duration2 = (end2-start2)/1000.0;
        System.out.println("duration1 = " + duration1);
        System.out.println("duration2 = " + duration2);
        System.out.println("sum = " + sum);
    }

    private static int[][] createGrid(int size) {
        int[][] data = new int[size][size];

        for ( int x=0; x<size; x++ ) {
            for ( int y=0; y<size; y++ ) {
                data[x][y] = 1;
            }
        }

        return data;
    }

    private static long sumAcrossFirst(int[][] data) {
        long sum = 0;

        int size = data.length;
        for ( int x=0; x<size; x++ ) {
            for ( int y=0; y<size; y++ ) {
                sum += data[x][y];
            }
        }

        return sum;
    }

    private static long sumAcrossSecond(int[][] data) {
        long sum = 0;

        int size = data.length;
        for ( int x=0; x<size; x++ ) {
            for ( int y=0; y<size; y++ ) {
                sum += data[y][x];
            }
        }

        return sum;
    }


}

另一个优化是将 int[][] 减少为 int[]，这将减少指针追逐，现代 CPU 预取器将启动并将数组的下一部分保留在缓存中。

为了并行，您必须考虑相同的缓存行为，并认识到使用多个线程会产生开销。因此，较小的数组在单个线程上求和的速度会更快。最好测量此阈值，因为它随 CPU 的不同而变化，但一般来说它会在 1000 左右或更多。也就是说，我通常会等待输入数据通过一百万个单元格，然后再担心额外的复杂性。跨数组求和的速度很快。

对数组求和的最快方法是使用 SIMD 指令，不幸的是，如果不使用 JNI 或类似的东西，它们不能直接在 Java 中使用。 Fork/Join 的工作令人钦佩，但在加快速度之前它有一些开销。这意味着并行和单核需要多少int才能实现收支平衡的阈值会更高。

让多个线程写入同一个单个数组是有意义的。请注意，从多个 CPU 核心写入可能会导致核心之间的缓存失效，如果有两个单独的核心访问同一内存页，则可能会导致系统抖动。

因此，为了开始工作，您可以随意使用以下方法。它演示了如何使用 Java Executor；这是位于 Fork/Join 框架下方的线程池。

private static Executor pool = Executors.newFixedThreadPool( Runtime.getRuntime().availableProcessors() );

private static int[][] sumParallel( int[][] a, int[][] b ) throws InterruptedException {
    int[][] result = createGrid(a.length);
    CountDownLatch latch = new CountDownLatch(a.length);

    for ( int i=0; i<a.length; i++ ) {
        pool.execute( new SumTask(latch, a,b,i, result) );
    }

    latch.await();

    return result;
}

public static class SumTask implements Runnable {
    private CountDownLatch latch;

    private int[][] a;
    private int[][] b;
    private int     row;
    private int[][] result;

    public SumTask(CountDownLatch latch, int[][] a, int[][] b, int row, int[][] result) {
        this.latch = latch;

        this.a = a;
        this.b = b;
        this.row = row;
        this.result = result;
    }

    public void run() {
        for ( int y=0; y<a.length; y++ ) {
            result[row][y] = a[row][y] + b[row][y];
        }

        latch.countDown();
    }
}

为了更有趣，这里有一个 ForkJoin 等效项:

public class Sum2DFJ {

    public static void main( String[] args ) throws ExecutionException, InterruptedException {
        int[][] data = {{1,2,3},{1,2,3},{1,2,3}};

        SumTask task = new SumTask(data, data);
        ForkJoinPool pool = new ForkJoinPool();


        pool.execute(task);

        int[][] result = task.get();

        for ( int x=0; x<data.length; x++ ) {
            for ( int y=0; y<data.length; y++ ) {
                System.out.println("result[x][y] = " + result[x][y]);
            }
        }
    }

}


@SuppressWarnings("unchecked")
class SumTask extends RecursiveTask<int[][]> {

    private int[][] a;
    private int[][] b;

    public SumTask( int[][] a, int[][] b ) {

        this.a = a;
        this.b = b;
    }

    protected int[][] compute() {
        int[][] result = createGrid(a.length);

        List<ForkJoinTask> children = new ArrayList();

        for ( int i=0; i<a.length; i++ ) {
            children.add( new SumChildTask(a,b,i, result) );
        }

        invokeAll(children);

        return result;
    }

    private static int[][] createGrid(int size) {
        int[][] data = new int[size][size];

        for ( int x=0; x<size; x++ ) {
            for ( int y=0; y<size; y++ ) {
                data[x][y] = 0;
            }
        }

        return data;
    }
}

class SumChildTask extends RecursiveAction {


    private int[][] a;
    private int[][] b;
    private int row;
    private int[][] result;

    public SumChildTask(int[][] a, int[][] b, int row, int[][] result) {
        this.a = a;
        this.b = b;
        this.row = row;
        this.result = result;
    }

    protected void compute() {
        for ( int i=0; i<b.length; i++ ) {
            result[row][i] = a[row][i] + b[row][i];
        }
    }
}