java - 并行中值滤波器

Question

所以我一直在研究中值滤波器*(比较顺序与并行，因为我正在学习如何同时编码)，但是并行版本对于大输入(大约 100k；它运行得更少)无限期地运行。

本质上，代码接收一个文件，根据给定的窗口大小对其进行过滤，然后将其写入一个新文件。并行编程新手，所以当谈到这里可能出现的问题时我有点迷失。

//import everything

public class SecondMedianFilter extends RecursiveAction {
    float[] numbers;
    static int filter;
    int window;
    int length;
    int lo;
    int hi;

    static final int SEQUENTIAL_CUTOFF = 500;


    float[] outArray;

    public SecondMedianFilter(float[] numbers, int filter, int lo, int hi) {
        this.numbers = numbers;
        this.filter = filter;
        this.lo = lo;
        this.hi = hi;
        length = numbers.length;
        window = (filter - 1) / 2;
    }

    public float[] getRes() {
        return result;
    }


    protected void compute() {

        result = new float[length];

        if ((hi - lo) < SEQUENTIAL_CUTOFF) {

            for (int a = lo; a < hi; a++) {


                for (int i = 0; i < length; i++) {
                    if (i < window || i >= length - window) {
                        result[i] = numbers[i];
                    } else {
                        float[] subArray = Arrays.copyOfRange(numbers, i - window, i + window + 1);
                        Arrays.sort(subArray);
                        float median = subArray[(subArray.length / 2)];
                        result[i] = median;

                    }
                }

            }
        } else {

            SecondMedianFilter left = new SecondMedianFilter(filtered, filter, lo, (hi + lo) / 2);
            SecondMedianFilter right = new SecondMedianFilter(filtered, filter, (hi + lo) / 2, hi);
            left.fork();
            right.compute();
            left.join();
        }
    }


    public static void main(String[] args) {
        //reads in a file, processes each line into an array of floats that
        // I call inArray, which gets filtered into outIntArray


        float[] outIntArray = new float[inArray.length];

        if (window < 3 || window > 21 || window % 2 == 0) {
            System.out.println("Window size error.");
        } else {

            SecondMedianFilter smf = new SecondMedianFilter(inArray, window, 0, inArray.length);
            smf.compute();
            outIntArray = smf.getRes();


            // loops through outIntArray and writes to file.
        }//end main           
    }
}

按顺序执行似乎可以工作(大约 1,000,000 个元素不到一秒)，但我的并发版本仅需要 4 个小时才能执行 10,000 个元素。正如我所说，并行编程是全新的，所以我很迷失。有没有一种我缺少的并行中值滤波器的方法？

(* 中值过滤器 = 获取数组的某个窗口，对它们进行排序，并用排序后的子数组的中值替换该索引处的原始元素；例如:2, 80, 6, 3, 1 结果为 2, 6, 6, 3, 1。)

*例如:

Taking in this file: 
5.0
13.2
-2.6
22.3
12.4
-0.21
23.1
-0.2454

它将把它读入数组 [5.0, 13.2, -2.6, 22.3, 12.4, -0.21, 23.1, -0.2454] 假设窗口大小为 3。为了适用于过滤，元素之前必须有 n 个元素，之后必须有 n 个元素，其中 n = (window - 1)/2；因此，在 window = 3 的情况下，元素的两侧必须有 1 个元素。如果不满足该条件，则按原样采用该元素。

因此 5.0 将保留，因为它前面没有元素。但 13.2 满足条件 - 因此，采用子数组 [5.0, 13.2, -2.6]。然后对该数组进行排序(使用 .sort(): [-2.6, 5.0, 13.2])，然后将中位数取为 5.0。随后，最终数组中的 13.2 被替换为 5.0，现在看起来像 [5.0, 5.0,...]。

接下来它移动到 -2.6 - 它前面有一个元素，后面有一个元素，因此对子数组 [-2.6, 22.3, 12.4] 进行排序，并将 12.4 的中位数添加到最终数组:[5.0, 5.0, 12.4,...]。它重复此过程，直到访问了原始数组中的所有元素。然后它将最终的数组写入文件，但这并不是特别相关(除非这可以以某种方式并行完成 - 我对此表示怀疑，但正如我所说:这里完全是菜鸟)。

Answer 1

您正在处理的是完全可并行化的。事实上，您可以为每个元素子集计算中值过滤器(因为您在计算中不会更改原始数组的内容)。所以，是的，你可以并行进行。

话虽如此，即使我不知道RecursiveAction的所有细节，我也会说你错过了整个线程部分。我检查了一下，任何递归操作都应该由 ForkJoinPool 类调用。然后这个线程处理线程并 fork /连接它们。

此外，将操作拆分为右侧和左侧(使用两个不同的调用)需要您稍后fork并join它们。

我希望这样的东西有一个工作配置:

public class SecondMedianFilter extends RecursiveAction {
... 

protected void compute() {

    result = new float[length];

    if ((hi - lo) < SEQUENTIAL_CUTOFF) {

        for (int a = lo; a < hi; a++) {


            for (int i = 0; i < length; i++) {
                if (i < window || i >= length - window) {
                    result[i] = numbers[i];
                } else {
                    float[] subArray = Arrays.copyOfRange(numbers, i - window, i + window + 1);
                    Arrays.sort(subArray);
                    float median = subArray[(subArray.length / 2)];
                    result[i] = median;

                }
            }

        }
    } else {

        SecondMedianFilter left = new SecondMedianFilter(filtered, filter, lo, (hi + lo) / 2);
        SecondMedianFilter right = new SecondMedianFilter(filtered, filter, (hi + lo) / 2, hi);
        left.fork();
        //CODE CHANGES FROM HERE
        right.fork();
        //right.compute(); <- IS IMPLICIT IN THE FORK 
        left.join();
        right.join();
        //TO HERE
    }
}


public static void main(String[] args) {
    //reads in a file, processes each line into an array of floats that
    // I call inArray, which gets filtered into outIntArray


    float[] outIntArray = new float[inArray.length];

    if (window < 3 || window > 21 || window % 2 == 0) {
        System.out.println("Window size error.");
    } else {
        // CODE CHANGES FROM HERE
        ForkJoinPool pool = new ForkJoinPool(); // WHO HANDLES THE THREADS 
        SecondMedianFilter smf = new SecondMedianFilter(inArray, window, 0, inArray.length);
        //smf.compute(); <- DUTY OF THREAD POOL
        pool.invoke(smf); //  START OF PROCESSING 
        outIntArray = smf.getRes();


        // loops through outIntArray and writes to file.
    }//end main           
}

}

我将向您提供两个链接，它们将足够详细地解释整个过程如何工作(其中一个显示并行方法和顺序方法)。

http://www.concretepage.com/java/jdk7/example-of-recursiveaction-in-java

http://www.logicbig.com/how-to/java/fork-and-join-recursive-action/

编辑我还没有检查你的算法是如何工作的(我希望它将数组分成两部分，直到涉及到绑定(bind)情况)，所以我只是根据你所写的内容添加了代码片段。