平行中值滤波器

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的所有细节，我也会说你错过了整个threads部分。我检查了一下，任何递归操作都应该由 ForkJoinPool 类调用。然后这个处理线程并分叉/加入它们。

此外，拆分左右操作（使用两个不同的调用）需要您稍后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/

编辑 我还没有检查过你的算法是如何工作的（我希望它在遇到绑定情况时将数组分成两部分），所以我只是根据你写的内容添加了一些代码。