为什么这个 scala prime 生成这么慢/内存密集？

Question

我在查找第 10,001 个素数时内存不足。

object Euler0007 {
  def from(n: Int): Stream[Int] = n #:: from(n + 1)
  def sieve(s: Stream[Int]): Stream[Int] = s.head #:: sieve(s.filter(_ % s.head != 0))
  def primes = sieve(from(2))
  def main(args: Array[String]): Unit = {
    println(primes(10001))
  }
}

这是因为在primes 的每次“迭代”（在这种情况下这是正确的术语吗？）之后，我将要调用的函数堆栈增加一个以获取下一个元素？

我在网上找到的一个不采用迭代解决方案的解决方案（我想避免进入函数式编程/惯用 scala）是this（问题 7）：

lazy val ps: Stream[Int] = 2 #:: Stream.from(3).filter(i => ps.takeWhile(j => j * j <= i).forall(i % _ > 0))

据我所知，这不会导致这种类似递归的方式。这是一个好方法吗，还是你知道更好的方法？

Answer 1

这很慢的一个原因是它不是埃拉托色尼的筛子。阅读http://www.cs.hmc.edu/~oneill/papers/Sieve-JFP.pdf 以获得详细解释（示例在 Haskell 中，但可以直接翻译成 Scala）。

我对欧拉问题 #7 的旧解决方案也不是“真正的”筛子，但它似乎对小数字足够好：

object Sieve {

    val primes = 2 #:: sieve(3)

    def sieve(n: Int) : Stream[Int] =
          if (primes.takeWhile(p => p*p <= n).exists(n % _ == 0)) sieve(n + 2)
          else n #:: sieve(n + 2)

    def main(args: Array[String]) {
      println(primes(10000)) //note that indexes are zero-based
    }
}

我认为你的第一个版本的问题是你只有defs，没有val，它收集结果并且可以被生成函数查询，所以你总是从头开始重新计算。

Answer 2

FWIW，这是一个真正的埃拉托色尼筛法：

def sieve(n: Int) = (2 to math.sqrt(n).toInt).foldLeft((2 to n).toSet) { (ps, x) => 
    if (ps(x)) ps -- (x * x to n by x) 
    else ps
}

这是一个无限的素数流，使用了保留其基本属性的埃拉托色尼筛的变体：

case class Cross(next: Int, incr: Int)

def adjustCrosses(crosses: List[Cross], current: Int) = {
  crosses map {
    case cross @ Cross(`current`, incr) => cross copy (next = current + incr)
    case unchangedCross                 => unchangedCross
  }
}

def notPrime(crosses: List[Cross], current: Int) = crosses exists (_.next == current)

def sieve(s: Stream[Int], crosses: List[Cross]): Stream[Int] = {
    val current #:: rest = s

    if (notPrime(crosses, current)) sieve(rest, adjustCrosses(crosses, current))
    else current #:: sieve(rest, Cross(current * current, current) :: crosses)
}

def primes = sieve(Stream from 2, Nil)

然而，这有点难以使用，因为Stream 的每个元素都是使用crosses 列表组成的，该列表中的数字与一个数字的素数一样多，而且看起来，对于出于某种原因，这些列表针对Stream 中的每个数字保存在内存中。

例如，在评论提示下，primes take 6000 contains 56993 会抛出 GC 异常，而 primes drop 5000 take 1000 contains 56993 在我的测试中会很快返回结果。

Answer 3

是的，它是，因为您“在每次“迭代”之后将要调用以获取下一个元素的函数堆栈增加一个 em> " - 即每次获得每个素数后，在过滤器堆栈顶部添加一个新过滤器。这过滤器太多了。

这意味着每个生成的素数都经过其所有前面的素数的测试 - 但只有低于其平方根的素数才是真正需要的。例如，要获得第 10001 个素数 104743，将在运行时创建 10000 个过滤器。但是在323（104743 的平方根）下面只有 66 个素数，所以真正需要的只有 66 个过滤器。所有其他 9934 人将不必要地在那里，占用内存，努力工作，绝对不会产生任何附加值。

这是“功能筛”的关键缺陷，它似乎起源于 1970 年代 David Turner 的代码，后来又进入了 @ 987654321@等地。 不是，它是一个试验划分筛子（而不是埃拉托色尼的筛子）。这太遥远了。试除法在优化实施时完全能够非常快速地产生第 10000 个素数。

该代码的主要缺陷是它没有将过滤器的创建推迟到正确的时间，并最终创建了太多的过滤器。

现在谈论复杂性，“旧筛子”代码是 O(n²)，在n 产生的素数中。最优试划分为O(n^1.5/log^0.5(n))，Eratosthenes的筛子为O(n *log(n)*log(log(n)))。作为empirical orders of growth，第一个通常被视为~ n^2，第二个被视为~ n^1.45，第三个被视为~ n^1.2。

您可以找到基于 Python 生成器的代码，以实现优化试验划分in this answer (2nd half of it)。这是originally discussed here 处理与您的筛子功能等效的Haskell。

---

作为一个例子，旧筛子的"readable pseudocode" :) 是

primes = sieve [2..]  where
   sieve (x:xs) = x : sieve [ y | y <- xs, rem y x > 0 ]
                         -- list of 'y's, drawn from 'xs',
                         --         such that (y % x > 0)

对于最优试除法 (TD) 筛，在素数方格上同步，

primes = sieve [2..] primes   where
  sieve (x:xs) ps = x : (h ++ sieve [ y | y <- t, rem y p > 0 ] qs)
          where
            (p:qs) = ps     -- 'p' is head elt in 'ps', and 'qs' the rest
            (h,t)  = span (< p*p) xs    -- 'h' are elts below p^2 in 'xs'
                                        -- and 't' are the rest

对于a sieve of Eratosthenes、devised by Richard Bird，正如此处另一个答案中提到的 JFP 文章中所见，

primes = 2 : minus [3..] 
               (foldr (\p r-> p*p : union [p*p+p, p*p+2*p..] r) [] primes)
                      -- function of 'p' and 'r', that returns 
                      -- a list with p^2 as its head elt, ...

短和快。 （minus a b 是一个列表 a，其中 b 的所有 elt 逐渐从中删除；union a b 是一个列表 a，b 的所有 elt 逐渐添加到其中，没有重复；两者都处理有序的，非递减列表）。 foldr 是列表的the right fold。因为它是线性的，所以它在~ n^1.33 处运行，要使其在~ n^1.2 处运行，可以使用tree-like folding 函数foldi）。

---

第二个问题的答案也是是。你的第二个代码，用相同的“伪代码”重写，

ps = 2 : [i | i <- [3..], all ((> 0).rem i) (takeWhile ((<= i).(^2)) ps)]

与上面的最佳 TD 筛非常相似 - 两者都安排每个候选者通过其平方根以下的所有素数进行测试。虽然筛子通过延迟过滤器的运行时间序列来安排它，但后一个定义为每个候选者重新获取所需的素数。一个可能比另一个更快，具体取决于编译器，但两者本质上是相同的。

第三个也是是的：埃拉托色尼的筛子更好，

ps = 2 : 3 : minus [5,7..] (unionAll [[p*p, p*p+2*p..] | p <- drop 1 ps])

unionAll = foldi union' []          -- one possible implementation
union' (x:xs) ys = x : union xs ys
   -- unconditionally produce first elt of the 1st arg 
   -- to avoid run-away access to infinite lists

从其他代码sn-ps的相似性来看，看起来它也可以在Scala中实现。 （虽然我不知道 Scala）。 unionAll 这里实现了树状折叠结构 (click for a picture and full code) 但也可以用滑动数组实现，沿着素数的倍数流逐段工作。

TL;DR：是的，是的，是的。