为 fork-join 递归添加停止条件

Question

为了简化我的例子，假设我正在使用 Java 的 Fork-Join 框架实现二进制搜索。我的目标是在整数数组中找到一个特定的整数值（目标整数）。这可以通过将数组分成两半来完成，直到它小到可以执行串行搜索。算法的结果需要是一个布尔值，表示是否在数组中找到目标整数。

在幻灯片 28 之后的Klaus Kreft's presentation 中探讨了类似的问题。然而，Kreft 的目标是找到数组中的最大数，因此必须扫描所有条目。就我而言，没有必要扫描整个数组，因为一旦找到目标整数，就可以停止搜索。

我的问题是，一旦我遇到目标整数，许多任务已经被插入到线程池中，我需要取消它们，因为继续搜索没有意义。我尝试从 RecursiveTask 内部调用 getPool().terminate() 但这并没有太大帮助，因为许多任务已经排队，我什至注意到即使在调用 shutdown 之后新的一次也排队..

我当前的解决方案是使用一个静态的 volatile 布尔值，该布尔值初始化为“false”，并在任务开始时检查其值。如果仍然为“假”，则任务开始工作，如果为“真”，则任务立即返回。我实际上可以为此使用 RecursiveAction。

所以我认为这个解决方案应该可以工作，但我想知道框架是否提供了一些处理此类情况的标准方法 - 即为递归定义一个停止条件，从而取消所有排队的任务。

请注意，如果我想在找到目标整数时立即停止所有正在运行的任务（通过其中一个正在运行的任务），我必须检查这些任务中每一行之后的布尔值，这可能会影响性能，因为它的值boolean 不能被缓存（它被定义为 volatile）。

确实，我认为需要一些标准解决方案，并且可以以清除队列和中断正在运行的任务的形式提供。但是我还没有找到这样的解决方案，我想知道是否有其他人知道它或有更好的想法。

感谢您的宝贵时间， 阿萨夫

编辑：这是我的测试代码：

package xxx;

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public class ForkJoinTest {

    static final int ARRAY_SIZE = 1000;
    static final int THRESHOLD = 10;

    static final int MIN_VALUE = 0;
    static final int MAX_VALUE = 100;

    static Random rand = new Random();


    // a function for retrieving a random int in a specific range
    public static int randInt(int min, int max) {
        return rand.nextInt((max - min) + 1) + min;
    }

    static volatile boolean result = false;
    static int[] array = new int[ARRAY_SIZE];
    static int target;

    @SuppressWarnings("serial")
    static class MyAction extends RecursiveAction {

        int startIndex, endIndex;

        public MyAction(int startIndex, int endIndex) {
            this.startIndex = startIndex;
            this.endIndex = endIndex;
        }

        // if the target integer was not found yet: we first check whether 
        // the entries to search are too few. In that case, we perform a 
        // sequential search and update the result if the target was found. 
        // Otherwise, we break the search into two parts and invoke the 
        // search in these two tasks.
        @Override
        protected void compute() {
            if (!result) {
                if (endIndex-startIndex<THRESHOLD) { 
                    // 
                    for (int i=startIndex ; i<endIndex ; i++) {
                        if (array[i]==target) {
                            result = true;
                        }
                    }
                } else {
                    int middleIndex = (startIndex + endIndex) / 2;
                    RecursiveAction action1 = new MyAction(startIndex, middleIndex);
                    RecursiveAction action2 = new MyAction(middleIndex+1, endIndex);
                    invokeAll(Arrays.asList(action1,action2));
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        for (int i=0 ; i<ARRAY_SIZE ; i++) {
            array[i] = randInt(MIN_VALUE, MAX_VALUE);
        }
        target = randInt(MIN_VALUE, MAX_VALUE);
        ForkJoinPool pool = new ForkJoinPool();
        pool.invoke(new MyAction(0,ARRAY_SIZE));
        System.out.println(result);
    }

}

Answer 1

我认为您可能正在为正确的解决方案设置障碍。

您说您的boolean stop 标志必须是volatile，因此会影响解决方案的速度——嗯，是的，也不是——访问volatile 确实会刷新缓存，但您是否考虑过@987654325 @?

我相信正确的解决方案是使用AtomicBoolean 标志来停止所有进程。您应该检查是否合理，以使您的系统快速停止。

尝试清除所有队列并中断所有线程是错误的 - 这会导致可怕的混乱。

    static AtomicBoolean finished = new AtomicBoolean();
    ....

        protected void compute() {
            if (!finished.get()) {
                if (endIndex - startIndex < THRESHOLD) {
                    //
                    for (int i = startIndex; i < endIndex && !finished.get(); i++) {
                        if (array[i] == target) {
                            finished.set(true);
                            System.out.print("Found at " + i);
                        }
                    }
                } else {
                    ...
                }
            }
        }

Answer 2

我通过查看在许多内置功能中执行此操作的开源产品，在上面留下了关于如何执行此操作的评论。让我在这里说一些细节。

如果您想取消正在开始或正在执行的任务，则每个任务都需要了解其他所有任务。当一个任务找到它想要的东西时，该任务需要通知其他所有任务停止。你不能用二元递归除法（RecursiveTask 等）来做到这一点，因为你递归地创建新任务，而旧任务永远不会知道新任务。我敢肯定，您可以为每个新任务传递对停止我的字段的引用，但它会变得非常混乱，并且调试会“很有趣”。

您可以使用 Java8 CountedCompleter() 来完成此操作。为了支持这个类，框架被砍掉了，所以应该由框架完成的事情需要手动完成，但它可以工作。

每个任务都需要一个可变布尔值和一个将其设置为 true 的方法。每个任务都需要一个对所有其他任务的引用数组。预先创建所有任务，每个任务都有一个空数组，其中包含对其他任务的引用。填写对所有其他任务的引用数组。现在提交每个任务（请参阅此类的文档，fork() addPendingCount() 等）

当一个任务找到它想要的东西时，它使用对其他任务的引用数组将它们的布尔值设置为 true。如果存在多个线程的竞争条件，则无关紧要，因为所有线程都设置为“真”。您还需要处理 tryComplete()、onCompletion() 等。这个类非常混乱。它用于 Java8 流处理，这本身就是一个故事。

你不能做的是在它们开始之前从双端队列中清除待处理的任务。您需要等到任务开始并检查布尔值是否为真。如果执行时间很长，那么您可能还需要定期检查布尔值是否为真。 volatile 读取的开销并没有那么糟糕，而且确实没有其他方法。