c# - 使用并行查找素数

Question

所以我创建了下面的方法来查找不超过某个数字的所有素数。关于如何加快速度的任何建议？

我是这样调用它的；

interval = (value + NOOFTHREADS - 1) / NOOFTHREADS;
int max = interval * NOOFTHREADS;

tickets = new List<int>(NOOFTHREADS);
for (int i = 1; i <= NOOFTHREADS; i++)
{
    tickets.Add(i * (max / NOOFTHREADS));
}

Enumerable.Range(1, NOOFTHREADS)
.AsParallel()
.ForAll(_ => findPrimes());

---

带有一些全局变量；

static List<int> vals = new List<int>();
static List<int> tickets;
static int interval = new int();

---

和方法；

public static void findPrimes()
    {
        int myTicket;
        lock (tickets)
        {
            myTicket = (int)tickets.Last();
            tickets.RemoveAt(tickets.Count - 1);
        }
        var max = myTicket;
        int min = max - interval +1;
        int num;

        var maxSquareRoot = Math.Sqrt(max);
        var eliminated = new System.Collections.BitArray(max + 1);
        eliminated[0] = true;
        eliminated[1] = true;
        for (int i = 2; i < (max) / 2; i++)
        {
            if (!eliminated[i])
            {
                if (i < maxSquareRoot)
                {
                    num = ((min + i -1 )/i)*i;

                    if (num == i)
                        num = num + i;

                    for (int j =num; j <= max; j += i)
                        eliminated[j] = true;
                }
            }
        }
        for (int b = (int)min; b < max; b++)
        {
            if (!eliminated[b])
                lock(vals)
                    vals.Add(b);
        }
    }

Answer 1

Eratosthenes 筛法可以很容易地并行化，您只需将它分成单独的 block ，然后单独筛选每个 block 。您已经开始进行拆分，但还没有走得足够远，无法获得良好的结果。看看findPrimes()有什么问题

var max = myTicket;
int min = max - interval +1;
int num;

var maxSquareRoot = Math.Sqrt(max);
var eliminated = new System.Collections.BitArray(max + 1);

您创建一个新的 BitArray对于涵盖从 0 到 max 的所有数字的每个线程.对于筛选第一个 block 的线程，这很好，但对于后面的线程，您分配的内存比需要的多得多。由于上限较高且线程较多，这本身就是一个问题，您分配的大致是 (NOOFTHREADS + 1) * limit / 2只有大约 limit 的位需要位。对于更少的线程和/或更低的限制，您仍然会恶化局部性并且会有更多的缓存未命中。

eliminated[0] = true;
eliminated[1] = true;
for (int i = 2; i < (max) / 2; i++)

你应该在 i > maxSquareRoot 时停止外循环.然后循环体不再做任何有成效的事情，它只执行一次读取和一两次检查。每次迭代都不会花很长时间，但对所有 i 都这样做来自 √max至 max如果 max 加起来是例如10¹¹。仅对最后一个 block 执行此操作可能比单线程单 block 筛选需要更长的时间。

{
    if (!eliminated[i])

eliminated[i]只能对 i >= min 成立(或 i < 2 )，您只会在 i <= maxSquareRoot 的第一个 block 中遇到这种情况(除非限制低得离谱)。因此，对于其他 block ，您还消除了 4、6、8、9、10、12、14、... 的倍数。大量浪费的工作。

    {
        if (i < maxSquareRoot)

如果maxSquareRoot恰好是一个质数，你没有消除它的平方，比较应该是<= .

        {
            num = ((min + i -1 )/i)*i;

            if (num == i)
                num = num + i;

            for (int j =num; j <= max; j += i)
                eliminated[j] = true;
        }
    }
}

现在，当筛分完成后，您逐步浏览 BitArray 的 block

for (int b = (int)min; b < max; b++)
{
    if (!eliminated[b])
        lock(vals)
            vals.Add(b);
}

只要找到素数，就锁定列表 vals并向其添加素数。如果有两个或更多线程几乎同时完成筛分，它们将在那里互相踩到脚趾，锁定和等待将进一步减慢过程。

为了减少空间使用，每个线程应该创建一个素数列表到maxSquareRoot ，并使用它来消除其 block 中的复合 Material ，以便 BitArray只需要 max - min + 1位。每个创建自己的列表的线程都会重复一些工作，但是由于这里的上限很小，因此不会有太多额外的工作。我不知道并发读取访问是如何处理的，如果这不会增加同步开销，您也可以只为所有线程使用一个列表，但我怀疑这会有所收获。

代码大致如下:

List<int> sievePrimes = simpleSieve(maxSquareRoot);
// simpleSieve is a standard SoE returning a list of primes not exceeding its argument
var sieve = new System.Collections.BitArray(max - min + 1);
int minSquareRoot = (int)Math.Sqrt(min);
foreach(int p in sievePrimes)
{
    int num = p > minSquareRoot ? p*p : ((min + p - 1)/p)*p;
    num -= min;
    for(; num <= max-min; num += p)
    {
        sieve[num] =true;
    }
}

现在，为了避免线程在将素数添加到列表时踩到彼此的脚趾，每个线程都应该创建自己的素数列表并一步添加它(我不是 100% 确定这比添加更快每个素数都有自己的锁，但如果没有我会感到惊讶)

List<int> primes = new List<int>();
for(int offset = 0; offset <= max-min; ++offset)
{
    if (!sieve[offset])
    {
        primes.Add(min + offset);
    }
}
lock(vals) vals.AddRange(primes);

(和 vals 应该以大约预期素数数量的初始容量创建，以避免为每个 block 重新分配)