从 Parallel.ForEach 到多线程

Question

所以我将递归函数转换为迭代函数，然后使用 Parallel.ForEach，但是当我通过 VTune 运行它时，它的大部分运行时间实际上只使用了 2 个逻辑内核。

我决定尝试改用托管线程，并转换了这段代码：

for (int N = 2; N <= length; N <<= 1)
{
    int maxThreads = 4;
    var workGroup = Enumerable.Range(0, maxThreads);

    Parallel.ForEach(workGroup, i =>
    {
        for (int j = ((i / maxThreads) * length); j < (((i + 1) / maxThreads) * length); j += N)
        {
            for (int k = 0; k < N / 2; k++)
            {
                int evenIndex = j + k;
                int oddIndex = j + k + (N / 2);

                var even = output[evenIndex];
                var odd = output[oddIndex];

                output[evenIndex] = even + odd * twiddles[k * (length / N)];
                output[oddIndex] = even + odd * twiddles[(k + (N / 2)) * (length / N)];
            }
        }
    });
}

进入这个：

for (int N = 2; N <= length; N <<= 1)
{
    int maxThreads = 4;

    Thread one = new Thread(() => calculateChunk(0, maxThreads, length, N, output));
    Thread two = new Thread(() => calculateChunk(1, maxThreads, length, N, output));
    Thread three = new Thread(() => calculateChunk(2, maxThreads, length, N, output));
    Thread four = new Thread(() => calculateChunk(3, maxThreads, length, N, output));

    one.Start();
    two.Start();
    three.Start();
    four.Start();
}

public void calculateChunk(int i, int maxThreads, int length, int N, Complex[] output)
{
    for (int j = ((i / maxThreads) * length); j < (((i + 1) / maxThreads) * length); j += N)
    {
        for (int k = 0; k < N / 2; k++)
        {
            int evenIndex = j + k;
            int oddIndex = j + k + (N / 2);
            var even = output[evenIndex];
            var odd = output[oddIndex];

            output[evenIndex] = even + odd * twiddles[k * (length / N)];
            output[oddIndex] = even + odd * twiddles[(k + (N / 2)) * (length / N)];
        }
    }
}

问题出在 N 循环的最后一次迭代的第四个线程中，我得到一个输出数组的索引越界异常，其中索引尝试访问 length 的等效项。

我无法通过调试来查明原因，但我相信这与线程有关，我在没有线程的情况下运行代码并且它按预期工作。

如果有任何代码需要更改，请告诉我，我通常会有一些人建议修改。感谢您的帮助，我已尝试自己对其进行排序，并且相当确定问题出现在我的线程中，但我看不出是如何发生的。

PS：预期目的是并行化这段代码。

Answer 1

观察到的行为几乎可以肯定是由于使用了捕获的循环迭代变量N。我可以通过一个简单的测试重现你的情况：

ConcurrentBag<int> numbers = new ConcurrentBag<int>();

for (int i = 0; i < 10000; i++)
{
    Thread t = new Thread(() => numbers.Add(i));

    t.Start();
    //t.Join(); // Uncomment this to get expected behaviour.
}

// You'd not expect this assert to be true, but most of the time it will be.
Assert.True(numbers.Contains(10000));

简而言之，您的for 循环正在竞相递增N，然后执行calculateChunk 调用的委托可以复制N 的值。结果calculateChunk 看到N 的几乎随机值上升到（包括）length <<= 1 - 这就是导致您的IndexOutOfRangeException 的原因。

您将获得的输出值也将是垃圾，因为您永远不能依赖 N 的值是正确的。

如果您想安全地重写原始代码以利用更多内核，请将Parallel.ForEach 从内循环移动到外循环。如果外部循环迭代的次数很高，负载均衡器将能够正常工作（在您当前的 workGroup 计数为 4 时它无法正常工作 - 元素数量太少了）。