双流馈送以防止不必要的记忆？

Question

我是Haskell的新手,我正试图以流处理方式实现Euler的Sieve.

当我查看关于素数的Haskell Wiki页面时,我发现了一些神秘的流优化技术.在3.8维基的线性合并中:

primesLME = 2 : ([3,5..] `minus` joinL [[p*p, p*p+2*p..] | p <- primes']) 
  where
    primes' = 3 : ([5,7..] `minus` joinL [[p*p, p*p+2*p..] | p <- primes'])

joinL ((x:xs):t) = x : union xs (joinL t)

它说

" 根据Melissa O'Neill的代码,这里引入了双素数反馈,以防止不必要的记忆,从而防止内存泄漏."

怎么会这样？我无法弄清楚它是如何工作的.

Answer 1

通常,理查德·伯德(Richard Bird)对Eratosthenes筛子的描述中的素数流的定义是自我指涉的:

import Data.List.Ordered (minus, union, unionAll)

ps = ((2:) . minus [3..] . foldr (\p r -> p*p : union [p*p+p, p*p+2*p..] r) []) ps

ps此定义生成的素数用作输入.为了防止恶性循环的定义被灌注以初始值,2.这对应于埃拉托塞尼的筛作为发现在复合材料中,列举对每个素数之间的间隙的素数的数学定义p通过在步骤向上计数p,P = {2} U({3,4,...}\U {{ p ²,p ² + p,p ² + 2p,...} | p in P }).

生成的流在其自己的定义中用作输入.这导致整个素数流保留在内存中(或者无论如何都保留在内存中).这里的修复点是共享,corecursive:

fix f  = xs where xs = f xs                    -- a sharing fixpoint combinator
ps     = fix ((2:) . minus [3..] . foldr (...) [])
    -- = xs where xs = 2 : minus [3..] (foldr (...) [] xs)

然后,这个想法(由于Melissa O'Neill)将其分成两个流,内部循环将"第二个流"引入"上方":

fix2 f  = f xs where xs = f xs                 -- double-staged fixpoint combinator
ps2     = fix2 ((2:) . minus [3..] . foldr (...) [])
     -- = 2 : minus [3..] (foldr (...) [] xs) where
     --                                   xs = 2 : minus [3..] (foldr (...) [] xs)

因此,当ps2产生一些素p,其内流xs的"核心"素数仅需要被实例化至多约sqrt p,并且由产生的任何质数ps2可以被丢弃,并通过系统垃圾收集之后立即:

    \
     \
      <- ps2 <-.
                \
                 \
                  <- xs <-.
                 /         \ 
                 \_________/ 

内循环产生的素数xs不能立即丢弃,因为它们xs本身就需要它们.当xs生成素数时q,只有在计算部分sqrt q消耗之后才能丢弃其下面的foldr部分.换句话说,这序列保持返回指针到自身下降到sqrt其最大的值产生的(因为它是由消耗其消费者,像print).

因此,使用一个馈送循环(with fix)几乎整个序列必须保留在内存中,而使用double feed(with fix2)时,只需要保留内循环,只能达到当前生成的值的平方根由主流.因此,整体空间复杂度从大约O(N)减小到大约O(sqrt(N)) - 急剧减少.

为此,必须使用优化(即使用-O2交换机)编译代码,并独立运行.您可能还必须使用-fno-cse开关.并且ps2在测试代​​码中必须只有一个引用:

main = getLine >>= (read >>> (+(-1)) >>> (`drop` ps2) >>> print . take 5)

事实上,在Ideone上进行测试时,它确实显示出几乎恒定的内存消耗.

它是Eratosthenes的筛子,而不是Euler的筛子.

最初的定义是:

eratos (x:xs) = x : eratos (minus xs $ map (*x) [x..] )    -- ps = eratos [2..]
eulers (x:xs) = x : eulers (minus xs $ map (*x) (x:xs))    -- ps = eulers [2..]

由于倍数的过早处理,两者都是非常低效的.通过将枚举和枚举融合到一个更远的位置(从,到,即),可以很容易地修复第一个定义,这样它的处理可以推迟 - 因为这里每个素数的倍数是独立生成的(以固定的间隔枚举) :mapxx*x[x*x, x*x+x..]

eratos (p:ps) xs | (h,t) <- span (< p*p) xs =                 -- ps = 2 : eratos ps [2..]
                    h ++ eratos ps (minus t [p*p, p*p+p..])   -- "postponed sieve"

这篇文章的顶部与Bird的筛子相同,分段:

ps = 2 : [n | (r:q:_, px) <- (zip . tails . (2:) . map (^2) <*> inits) ps,
              n           <- [r+1..q-1] `minus` foldr union [] 
                               [[s+p, s+2*p..q-1] | p <- px, let s = r`div`p*p]]

((f <*> g) x = f x (g x)在这里用作免费的简写.)

第二个定义没有简单的解决方法,即eulers.

另外:你可以看到用Python生成器实现的相同想法,在这里进行比较.

实际上,Python代码采用了短暂的多级递归生成短暂的素数流; 在Haskell中,我们可以使用非共享,多阶段的固定点组合器来安排它_Y:

primes = 2 : _Y ((3:) . sieve 5 . unionAll . map (\p -> [p*p, p*p+2*p..]))
  where
    _Y g = g (_Y g)                                   -- == g . g . g . g . ....

    sieve k s@(x:xs) | k < x = k : sieve (k+2) s      -- == [k,k+2..] \\ s,
                     | True  =     sieve (k+2) xs     --    when s ? [k,k+2..]