6

我正在运行一些并行处理的测试，并制作了一个程序，给定一个整数矩阵，根据邻居重新计算每个位置的值。

CyclicBarrier我需要一个矩阵的副本，以便在解决部分问题后不会覆盖这些值并使用 a来合并结果：

CyclicBarrier cyclic_barrier = new CyclicBarrier(n_tasks + 1, new Runnable() {
    public void run() {
        ParallelProcess.mergeResult();
    }
});
ParallelProcess p = new ParallelProcess(cyclic_barrier, n_rows, r_cols); // init

每个任务都被分配了矩阵的一部分：我将它逐行分成相等的部分。但是可能会发生分区不准确的情况，因此会有一小块对应于最后一行不会提交到线程池。

示例：如果我有16行并且n_tasks = 4没有问题，则所有 4 个都将提交到池中。但如果我有18，前 16 个将被提交，而不是最后两个。

因此，如果发生这种情况，我将强制提交。好吧，我实际上并没有提交，因为我使用的是这样创建的固定线程池ExecutorService e = Executors.newFixedThreadPool(n_tasks)。由于池中的所有槽都被占用并且线程被屏障阻塞（mybarrier.await()在方法中调用run）我无法将它提交到池中，所以我只使用了Thread.start().

让我们进入正题。由于我需要考虑CyclicBarrier该块剩余的可能性，因此必须将参与方的数量增加 1。

但如果这种情况没有发生，我将是一个不足触发障碍的一方。

我的解决方案是什么？：

if (lower_limit != n_rows) { // the remaining chunk to be processed
    Thread t = new Thread(new ParallelProcess(lower_limit, n_rows));
    t.start();
    t.join();
}
else {
    cyclic_barrier.await();
}

cyclic_barrier.await()使用绝招强行提升结界的时候，感觉像是在作弊。

有没有其他方法可以解决这个问题，所以我不必做我正在做的事情？

4

1 回答 1

1

虽然这不能回答您关于 CyclicBarriers 的问题，但我可以推荐使用Phaser吗？这确实能够包含参与方的数量，并且它还允许您mergeResult在一个阶段被触发时运行。

因此，在执行异步计算之前，只需register. 然后在该计算中让线程到达phaser. 当所有线程都到达时，它将推进阶段并可以调用覆盖的方法onAdvance。

提交：

ParallelProcess process = new ParallelProcess(lower_limit, n_rows));
phaser.register();
executor.submit(process);

处理器

public void run(){
   //do stuff
   phaser.arrive();
}

移相器

Phaser phaser = new Phaser(){
    protected boolean onAdvance(int phase, int registeredParties) {
        ParallelProcess.mergeResult(); 
        return true;
    }
}
于 2014-06-23T19:35:34.977 回答