【问题标题】:Sieving prime numbers with Haskell用 Haskell 筛选素数
【发布时间】:2012-08-06 19:55:06
【问题描述】:

好的,所以我正在尝试编写一个 Haskell 程序,它可以非常快速地计算素数。大概我不是第一个尝试这样做的人。 (特别是,我该死的确定我看到了一些现有技术,但我现在找不到它......)

最初,我想计算小于 10^11 的素数。目前我已经让我的程序运行了大约 15 分钟,而且还不到一半。一些狂热的 C++ 程序员声称他的程序只需要 8 分钟。很明显,我做错了可怕的事情。

(以防万一,我当前的实现使用IOUArray Integer Bool 和多个线程来处理搜索空间的独立子范围。目前需要几秒钟才能从 10MB 数组块中删除所有 2 的倍数...)

请注意,10^11 对于 32 位算术来说太大了。此外,10^11 位 = 12.5 GB, 太多数据无法放入 Haskell 的 32 位地址空间。所以你不能一次将整个位图放在内存中。最后,请注意,小于 10^11 的素数只是小于 2^32 的阴影,因此您也无法一次存储所有实际整数。


编辑:显然我误读了时间信息。 C++ 家伙实际上声称的是:

  • 仅使用一个核心计算

  • 计数

抱歉这个错误...

编辑:我的源代码可以在这里找到:http://hpaste.org/72898

【问题讨论】:

  • 有其他人尝试过这样做吗?有什么代码可以看吗?
  • 您的解决方案与stackoverflow.com/questions/2221378/…相比如何?
  • “Haskell 的 32 位地址空间”是什么意思?这对我来说是个新闻!
  • @MathematicalOrchid 就 GHC 而言,64 位与 32 位的巨大影响并不是由于更大的地址空间或内存(尽管有时这也是有益的),它是一个) 64 位架构中的更多寄存器,这非常重要,因为 GHC 产生大量寄存器代码并且通常会产生 2 倍或更多的差异,并且 b) 64 位 Int#s,这意味着对 GMP 的 C 调用要少得多在整数计算中,当然Int 通常在 32 位系统上就足够了。这通常也会产生巨大的影响。

标签: haskell primes


【解决方案1】:

这是一个奇怪的问题/答案,因为接受的答案与问题不匹配:问题要求帮助提高筛子的速度(正确选择 Eratosthenes 实施的页面分段筛子)but the accepted answer 不' t 使用筛子,而是使用数值分析技术,只是一个库。尽管这对于非常快速地找到最大范围内的素数总数是很好的(并且在其他语言such as Kim Walisch's primecount in C++for quickly calculating the sums of primes over a range 中有更快和更广泛的版本可以做到这一点,但筛子对于处理特定类型很有用分析,例如查找素数间隙、素数双倍、三元组等 (generally K-Tuple primes) 等。事实上,一般的数值分析技术,如the Meissel–Lehmer algorithm,其中大多数都是基于这些技术,需要一个来源“种子素”开始,它最好由优化的埃拉托色尼筛产生。

事实上,根据上述链接,Kim Walisch 的 primecount 已经为其构建了 GHC/Haskell API,并且可以通过外部函数接口 (FFI) 轻松调用,因此比 arithmoi 库更好,因为它是比较快的。它是如此之快,以至于它是目前计算素数高达 1e28 的记录保持者!如果必须将这样的值提供给 Haskell 程序并且不在乎他们是否理解它是如何获取它的,它会在几十毫秒内计算出 1e11 的素数。

以类似的方式,如果确实需要筛子,那么 Kim Walisch's primesieve 也有 GHC Haskell FFI,也可以直接调用。

虽然使用库可以完成工作,但仅仅通过使用它们并不能了解如何实现它们;因此,Daniel Fischer's (DF's) very good tutorial answer 的原因以及他开始的这个后续系列。 DF 的回答显示了如何改进问题的代码,但是没有一个总结可以显示在完成他的所有建议后代码应该是什么样子;这一点尤其重要,因为 OP 的 hpaste 中的原始问题代码已经消失(幸运的是,这只是一个很好的例子,说明如何不这样做,但也许代码应该嵌入问题中以供参考),我们只能在他的回答中重建它通过 DF 的 cmets 所做的事情。这一系列答案旨在纠正如果有人需要纯 GHC Haskell 中的这种筛子,首先是对 DF 教学所引导的代码的摘要,然后是进一步的阶段性改进。

TLDR; 跳到我发布的 Haskell 代码的最后一个答案的末尾,它实际上几乎与 Kim Walisch 的 Primesieve 一样快或一样快,这可能是迄今为止最快的世界上,至少可以达到 10 到 1000 亿左右的更小范围,除了非常优化的 YAFU 版本之外,其他任何东西都不会超过这个范围,对于大范围来说,它可能会快 5% 左右。 DF 在车轮分解之前的最终代码据说比原始问题代码快 40 倍,我将其扩展到我的代码再次快 30 到 32 倍,总共比原来的代码快 1200 到 1280 倍原始问题代码!

原题代码

这将是我唯一一次引用它,因为它不再可用(而且我认为无论如何都不值得修改):我唯一喜欢该代码的地方是线程池的实现,但即便如此缺陷在于它使用mapM 将要由线程处理的整个作业队列提供给通道,这是一个非延迟函数,因此可能会将大量工作推送到作业通道上,从而消耗大量内存,而不是仅仅推动足够的工作来保持所有线程忙碌,然后再为从结果Channel 返回的每一个人提供一份工作。我将在此答案底部的代码和后续代码中更正这一点。实际上,只需要一个结果 MVar 池,因为 GHC 运行时分叉新线程的速度比将新工作通过管道传输到等待池的速度要快。

原始代码和 DF 改进代码的一个问题是它们都没有使用“-fllvm”编译器标志来使用 LLVM 后端代码生成器。对于我们在这里尝试编写的紧密循环,LLVM 可以将每个循环的时间减少大约两倍。原始代码的循环非常不紧密,LLVM 无能为力,但 DF 的代码确实有紧密的循环,并且可以通过将循环时间减少到大约 60% 来受益。

使用MVar(以及Chan)的另一个问题是它们不是特别快,每次激活的开销约为三毫秒。我们在 DF 在他的最终分析中的回答中得到了这个问题的证据,他说“使用四个线程,我用不可变数组得到 4.55 秒,而用可变数组得到 5.3 秒”,而使用两个线程则需要 3.75 秒。现在他的机器只有两个核心,额外的两个线程是超线程的,与另外两个共享大部分相同的资源,所以使用它们的人不期望有更好的性能,但也不期望性能更差,因为这里。原因是开销太大,以至于添加效率低下的内核实际上会增加额外的工作并减慢最终结果。在使用 LLVM 后端提高效率后,我还在我的四个“真实”线程/核心机器中看到了这一点。在使用所有四个代码时,我只将时间减少到大约 55%,这与我只使用两个内核时总执行时间实际上增加的情况一致。由于 `MVar's 是我们可以使用 GHC 实现“等待结果”的唯一方法,因此解决方案是使工作切片更大(更粗粒度的多线程),因此这种开销变得可以忽略不计,这是我在第二次回答中的算法改进。

一旦重载问题得到解决,也不需要像在无限深度的地方那样使用通道来接收工作和返回结果,所以我消除了它们,只支持一个“循环”数组,它的元素数量为池中的进程数。

测试环境

我不确定 DF 是否还有他的 Sandy Bridge 笔记本电脑,虽然我有一个 Sandy Bridge CPU,但它目前正在停机进行维护。但是,the online IDE website, Wandbox 使用的 Broadwell CPU 的额定值与 DF 的机器在 2.3 GHz 的回答中使用的额定值大致相同,涡轮增压到 2.9 GHz,用于单线程使用,具有两个内核/四个线程(超线程)。这与 DF 的机器具有相同的性能,正如我采用他引用的“arithmoi”库的内部结构并强制它运行十亿的范围所证明的那样。 this Wandbox link 显示它的运行时间与他在回答中提到的几乎完全相同的 2.92 秒。我没有费心去计算结果(使用弹出计数只有大约 0.01 毫秒),因为这不会改变比较,而只是强制它在默认的 128 KB 缓冲区大小的范围内运行。

因此,在 Wandbox 中,我们有一个易于引用的可比机器;但是,它有一个限制,它不支持使用 LLVM 后端,它的使用对于优化我们将使用的紧密剔除循环很重要。因此,我将在我自己的机器上进行有无 LLVM 的比较,这是一个英特尔 Skylake i5-2500,主频为 3.2 GHz,单线程使用时可提升至 3.6 GHz。这有一个轻微的限制,因为 Skylake 在架构上进行了进一步的改进,可以更好地预测分支,并且在正确预测分支时将分支省略至零时间,因此结果不会直接按使用的时钟速度进行缩放;由于我们正在开发的循环几乎将所有时间都花在紧密循环中,这可以使筛子实现的执行时间减少 10% 到 15%。

快速筛算法背后的原理

这些原则只有两条,如下:

  1. 在给定的筛分范围内保持较低的操作总数非常重要。
  2. 每个操作平均必须占用少量 CPU 时钟周期。

那么最终的执行效率就是这两者的乘积。

绩效目标

DF 似乎认为 Atkin 和 Bernstein 的 "primegen" 是筛分性能的“黄金标准”。这不是主要原因,它不会也不能平均每个操作占用少量 CPU 时钟周期(原则 2),并且它消耗的周期数随着范围的增加而增加,速度快于我认为的“黄金”标准”- TLDR 中引用的 Kim Walisch 的主筛。虽然阿特金筛 (SoA) 的这种实现完全正确,但它通过了上述原则 1),因为它快速收敛到恒定操作次数,即 0.2587 倍的范围,范围低至约 100,范围约为 1e11这低于埃拉托色尼 (Eratosthenes)as per the combo sieve here 的最大轮分解筛的最佳实际实现,由网页上该点上方的公式估计(1e11 为 0.2984,随着范围的增加而更高),它没有达到预期,因为到效率。 SoA 文档进行的比较与 Eratosthenese (SoE) 的比较存在缺陷:在与“primegen”代码相同的下载中是“eratspeed”的代码,它是 SoE 的参考版本,筛分超过十亿。然而,他们削弱了该参考版本,因为他们将其限制为与烘焙到 SoA 中相同的 2/3/5 轮分解(无法增加),而不是使用最大轮分解组合筛(有他们知道的文件中的证据)。这使得该范围内的操作数量略高于 4 亿次,而 SoA 的操作数约为 2.587 亿次。接下来,他们似乎进一步削弱了参考 SoE,使筛子缓冲区比用于 SoA 的缓冲区更小,从而增加了 SoE 的每次操作时间,因此与 SoA 的时间差不多,尽管这些操作比那些操作更简单的 SoA。通过这种方式,他们声称 SoA 比 SoA 快了大约 40%。

Berstein 以类似的方式对这两者的紧密内部操作循环进行了一些手动优化,这可能是由于当时的 C 编译器无法完全优化这些循环,并且为了让这些编译器不撤消这些手动优化,他在注释中指出,编译器应该只使用第一级优化来运行。这对于今天的“gcc”版本不再适用,因为两者的性能都随着“-O3”高级优化而提高。如果两者都设置为 8192 32 位字(32 KB),并按照上述优化进行编译,它们在 DF 型机器上都在大约 0.49/0.50 秒内筛选到十亿,表明 CPU 的数量对于“eratspeed”,每次剔除的周期减少了约 36%。如果应用了最大轮分解组合原则,那么它应该再快 40%,这是在对紧密的内部剔除循环进行优化之前; SoA 无法对剔除循环进行这些优化,因为与 SoE 的每个循环的固定跨度相比,它必须使用每个操作的可变跨度循环。这个参考“eratspeed”SoE 实现将在答案后面进一步讨论,因为这是导致我改进算法答案的方法。

作为关于 SoA“primegen”的最后一点说明,Bernstein 似乎认为筛缓冲区需要限制为小于 CPU L1 缓存大小。虽然这对于他开发这项工作的 CPU 来说可能是正确的,但对于现代 CPU 来说不再正确,因为现代 CPU 的 L2 缓存性能比 SoA 快得多,并且接近参考 SoE 紧凑的内部循环时间。因此,如果使筛选缓冲区等于 CPU L2 高速缓存大小(在这些 CPU 的情况下为 256 Kiloytes),“primegen”的筛选到十亿的时间几乎不会发生变化,大约为 0.50 秒,但时间筛分到 1000 亿个几乎线性缩放到大约 52 秒(应该如此)。这是有效的,因为缓冲区已增加,因此 SoA 不会很快受到操作跨度溢出的困扰,但它并不能解决问题,它只是将其移到更远的范围内,SoA 仍然不会比即使在最高的实用范围内也能最大限度地优化 SoE。

作为参考,“primesieve”类型的算法在大约 0.18 秒内筛选到十亿个,单线程时大约 25 秒到 1000 亿个范围内,使用两个线程时两个时间都减少了大约两倍在 CPU 的 Wandbox/DF 范围内。

DF 对他的回答的最终结果和提出的进一步工作的状态

DF 表示,当单线程/分别在两个线程上运行时,他的最终答案代码将在大约 7.5/3.75 秒内筛选到十亿。这代表了大约 25.514 亿次操作,在 2.9 GHz 的 CPU 时钟下,每个剔除 (CpC) 代表大约 9 个 CPU 时钟。这不好,因为基本的剔除循环需要大约 3.5 CpC。这是上面讨论的不使用 LLVM 的结果。

他建议第一个进一步的改进是轮子因式分解,仅用于将速度提高约 2.5 倍,并且使用更多扩展的轮子分解进行进一步改进是相当容易的。这是真实的。然而,他在“arithmoi”库的素数函数中进行扩展轮因式分解的尝试非常失败:在 2.9 GHz 下 2.92 秒完成 4.048 亿次剔除约为 21 CpC,就像 340 秒完成 460.7 亿次剔除一样筛选到 1000 亿个的筛选也非常糟糕,每次筛选的时钟都差不多。这太慢了,没有理由进行这种扩展的 2/3/5 轮子分解,因为结果将与仅使用赔率相同或慢,即使在 9 CpC 时也是如此。这种糟糕的效率的原因是他使用了一些复杂的,因此很慢的数学来减少车轮分解的剔除,但是这些计算需要大量的机器时间。有一些查找表方法可以做到这一点,每次剔除大约 12 个 CPU 时钟周期,速度大约是前者的两倍,但在如此小的范围内使用它们仍然太慢;它们的使用应仅限于为非常大的范围增加有效范围,而它们所花费的时间百分比只占总时间的一小部分。

为了显示这些结果有多糟糕,here is a reference Wandbox Javascript odds-only version 在大约 2.14 秒或大约 6.15 CpC 内筛分到十亿;这在我的 Skylake 机器上运行时间为 1.54 秒,由于上述 5.4 CpC 架构的改进,时间缩短了高于时钟速率的比率。

此外,在 Haskell 中,here is a odds-only version from my submission to RosettaCode, the second faster versionruns on the reference Wandbox CPU 在 2.26 秒内筛选到十亿,在没有 LLVM 的情况下编译大约 6.4 CpC。在我的 Skylake 机器上,这在没有/使用 LLVM 的情况下分别运行 1.83 秒和 1.023 秒(分别为 6.4/3.6 CpC),并且在没有 LLVM/LLVM 的情况下分别在大约 210/127 秒内筛分到 1000 亿(分别为 6.7/4.0 CpC )。请注意,这些比“arithmoi”库轮分解版本更快。这将成为我在第二个答案中进一步改进算法的基础。

因此,以下代码是一个仅赔率算法,按照 DF 提到的作为他的最终答案执行:

-- Multi-threaded Page-Segmented Bit-Packed Odds-Only Sieve of Eratosthenes... 
-- "Running a modern CPU single threaded is like
--  running a race car on one cylinder" me ...

-- compile with "-threaded" to use maximum available cores and threads...
-- compile with "-fllvm" for highest speed by a factor of up to two times.

{-# LANGUAGE FlexibleContexts, ScopedTypeVariables #-} -- , BangPatterns, MagicHash, UnboxedTuples, Strict
{-# OPTIONS_GHC -O2 -fllvm #-} -- or -O3 -keep-s-files -fno-cse -rtsopts

import Data.Int ( Int32, Int64 )
import Data.Word ( Word32, Word64 )
import Data.Bits ( (.&.), (.|.), shiftL, shiftR, popCount )
import Data.Array.Base (
         UArray(..), listArray, assocs, unsafeAt, elems,
         STUArray(..), newArray,
         unsafeRead, unsafeWrite,
         unsafeThaw, unsafeFreezeSTUArray, castSTUArray )
import Data.Array.ST ( runSTUArray )
import Control.Monad.ST ( ST, runST )
import Data.Time.Clock.POSIX ( getPOSIXTime )

-- imports to do with multi-threading...
import Data.Array (Array)
import Control.Monad ( forever, when )
import GHC.Conc ( getNumProcessors )
import Control.Monad.Cont ( join )
import Control.Concurrent
    ( ThreadId,
      forkIO,
      getNumCapabilities,
      myThreadId,
      setNumCapabilities )
import Control.Concurrent.MVar ( MVar, newEmptyMVar, putMVar, takeMVar )
import System.IO.Unsafe ( unsafePerformIO )

type Prime = Word64
type PrimeNdx = Int64
type StartAddr = Int32
type StartAddrArr = UArray Int StartAddr
type BasePrimeRep = Word32
type BasePrimeRepArr = UArray Int BasePrimeRep
type SieveBuffer = UArray Int Bool -- no point to artificial index!
 
-- constants related to odds-only...
cWHLPRMS :: [Prime]
cWHLPRMS = [2] -- excludes even numbers other than 2
cFRSTSVPRM :: Prime
cFRSTSVPRM = 3 -- start at first prime past the wheel prime(s)
 
makeSieveBuffer :: Int -> SieveBuffer
{-# INLINE makeSieveBuffer #-}
makeSieveBuffer szbts = runSTUArray $ do
  newArray (0, szbts - 1) False

-- count the remaining un-marked composite bits using very fast popcount...
{-# INLINE countSieveBuffer #-}
countSieveBuffer :: Int -> SieveBuffer -> Int
countSieveBuffer lstndx sb = runST $ do
  cmpsts <- unsafeThaw sb -- :: ST s (STUArray s PrimeNdx Bool)
  wrdcmpsts <-
    (castSTUArray :: STUArray s Int Bool ->
                      ST s (STUArray s Int Word64)) cmpsts
  let lstwrd = lstndx `shiftR` 6
  let lstmsk = 0xFFFFFFFFFFFFFFFE `shiftL` (lstndx .&. 63)
  let loopwi wi cnt =
        if wi < lstwrd then do
          v <- unsafeRead wrdcmpsts wi
          case cnt - popCount v of
            ncnt -> ncnt `seq` loopwi (wi + 1) ncnt
        else do
          v <- unsafeRead wrdcmpsts lstwrd
          return $ fromIntegral (cnt - popCount (v .|. lstmsk))
  loopwi 0 (lstwrd * 64 + 64)

cWHLPTRNLEN64 :: Int
cWHLPTRNLEN64 = 2048

cWHLPTRN :: SieveBuffer -- twice as big to allow for overflow...
cWHLPTRN = makeSieveBuffer (131072 + 131072)

-- could be faster using primitive copyByteArray#...
-- in preparation for filling with pre-cull pattern...
fillSieveBuffer :: PrimeNdx -> SieveBuffer -> SieveBuffer
fillSieveBuffer lwi sb@(UArray _ _ rng _) = runSTUArray $ do
  ptrn <- unsafeThaw cWHLPTRN :: ST s (STUArray s Int Bool)
  ptrnu64 <- (castSTUArray :: STUArray s Int Bool ->
                                  ST s (STUArray s Int Word64)) ptrn
  cmpsts <- unsafeThaw sb :: ST s (STUArray s Int Bool)
  cmpstsu64 <- (castSTUArray :: STUArray s Int Bool ->
                                  ST s (STUArray s Int Word64)) cmpsts
  let lmt = rng `shiftR` 6
      lwi64 = lwi `shiftR` 6
      loop i | i >= lmt = return cmpsts
             | otherwise =
                 let mdlo = fromIntegral $ lwi64 `mod` fromIntegral cWHLPTRNLEN64
                     sloop j
                       | j >= cWHLPTRNLEN64 = loop  (i + cWHLPTRNLEN64)
                       | otherwise = do
                          v <- unsafeRead ptrnu64 (mdlo + j)
                          unsafeWrite cmpstsu64 (i + j) v; sloop (j + 1) in sloop 0
  loop 0

cullSieveBuffer :: PrimeNdx -> [BasePrimeRepArr] -> SieveBuffer -> SieveBuffer
cullSieveBuffer lwi bpras sb@(UArray _ _ rng _) = runSTUArray $ do
  cmpsts <- unsafeThaw sb :: ST s (STUArray s Int Bool)
  let limi = lwi + fromIntegral rng - 1
      loopbpras [] = return cmpsts -- stop warning incomplete pattern match!
      loopbpras (bpra@(UArray _ _ bprrng _) : bprastl) =
        let loopbpi bpi
              | bpi >= bprrng = loopbpras bprastl
              | otherwise =
                let bp = unsafeAt bpra bpi
                    bpndx = (fromIntegral bp - cFRSTSVPRM) `shiftR` 1
                    rsqri = fromIntegral ((bpndx + bpndx) * (bpndx + cFRSTSVPRM)
                                             + cFRSTSVPRM) - lwi in
                if rsqri >= fromIntegral rng then return cmpsts else
                let bpint = fromIntegral bp
                    bppn = fromIntegral bp
                    cullbits c | c >= rng = loopbpi (bpi + 1)
                               | otherwise = do unsafeWrite cmpsts c True
                                                cullbits (c + bpint)
                    s = if rsqri >= 0 then fromIntegral rsqri else
                        let r = fromIntegral (-rsqri `rem` bppn)
                        in if r == 0 then 0 else fromIntegral (bppn - r)
                in cullbits s in loopbpi 0
  loopbpras bpras

-- multithreading goes here...

{-# NOINLINE cNUMPROCS #-}
cNUMPROCS :: Int -- force to the maximum number of threads available
cNUMPROCS = -- 1
-- {-
  unsafePerformIO $ do -- no side effects because global!
  np <- getNumProcessors; setNumCapabilities np
  getNumCapabilities
--}

-- list of culled soeve buffers from index with give bit size...
makePrimePagesFrom :: forall r. PrimeNdx -> Int ->
                                (PrimeNdx -> SieveBuffer -> r) -> Bool -> [r]
makePrimePagesFrom stwi szbts cnvrtrf thrdd =
  -- great, we can make an extra thread pool whenever we might need more, and
  -- it should die and be collected whenever this goes out of scope!
  let bpras = makeBasePrimeRepArrs thrdd
      jbparms() =
        let loop lwi szb =
              (lwi, szb) : loop (lwi + fromIntegral szb) szb
        in loop stwi szbts in
  if thrdd then
    let

      {-# NOINLINE strttsk #-}
      strttsk lwi szbts bpras mvr = -- do some strict work but define it non-strictly,
        forkIO $ do -- else it will run in forground before threading!
          -- and return it using a MVar; force strict execution in thread...
          putMVar mvr $! cnvrtrf lwi $ cullSieveBuffer lwi bpras
                           $ fillSieveBuffer lwi $ makeSieveBuffer szbts

      -- start a result pool, initialized to start with the first tasks...
      {-# NOINLINE rsltpool #-}
      rsltpool :: Array Int (MVar r) = unsafePerformIO $! do
        mvlst <- mapM (const newEmptyMVar) [ 1 .. cNUMPROCS ] -- unique copies
        mapM_ (\ (mvr, (lwi, szb)) -> strttsk lwi szb bpras mvr)
                                          $ zip mvlst $ jbparms()
        return $! listArray (0, cNUMPROCS - 1) mvlst

      -- lazily loop over the entire job list...
      loop (fdhd : fdtl) =
        let {-# NOINLINE getnxt #-}
            getnxt ((lwi, szb), i) = unsafePerformIO $! do -- wait for and get result of next page
              let mvr = unsafeAt rsltpool i
              r <- takeMVar mvr -- recycle mvr for next
              strttsk lwi szb bpras mvr; return $! r
        in getnxt fdhd : loop fdtl

  -- lazily cycle over the rest of the jobs forever...
  in rsltpool `seq` loop $ zip (drop cNUMPROCS $ jbparms())
                               (cycle [ 0 .. cNUMPROCS - 1 ]) else 

-- back to non multi-threaded functions...

  let loop ((lwi, szb) : jbpmstl) =
        (cnvrtrf lwi . cullSieveBuffer lwi bpras . fillSieveBuffer lwi .
           makeSieveBuffer) szb : loop jbpmstl
  in loop $ jbparms()

makeBasePrimeRepArrs :: Bool -> [BasePrimeRepArr]
makeBasePrimeRepArrs thrdd = 
  let sb2bpra :: PrimeNdx -> SieveBuffer -> BasePrimeRepArr
      sb2bpra lwi sb@(UArray _ _ rng _) =
        let len = countSieveBuffer (rng - 1) sb
            bpbs = fromIntegral cFRSTSVPRM + fromIntegral (lwi + lwi) in
        listArray (0, len - 1) [ bpbs + fromIntegral (i + i) |
                                          (i, False) <- assocs sb ]                
      fkbpras = [ sb2bpra 0 $ makeSieveBuffer 512 ]
      bpra0 = sb2bpra 0 $ cullSieveBuffer 0 fkbpras $ makeSieveBuffer 131072
  in bpra0 : makePrimePagesFrom 131072 131072 sb2bpra thrdd

-- result functions are here...

-- prepends the wheel factorized initial primes to the sieved primes output...
-- some faster not useing higher-order-functions, but still slow so who cares?
primes :: Int -> Bool -> [Prime]
primes szbts thrdd = cWHLPRMS ++ concat prmslsts where
  -- convert a list of sieve buffers to a UArray of primes...
  sb2prmsa :: PrimeNdx -> SieveBuffer -> UArray Int Prime
  sb2prmsa lwi sb@(UArray _ _ rng _) = -- bsprm `seq` loop 0 where
    let bsprm = cFRSTSVPRM + fromIntegral (lwi + lwi)
        len = countSieveBuffer (rng - 1) sb in
    bsprm `seq` len `seq`
      listArray (0, len - 1)
                [ bsprm + fromIntegral (i + i) | (i, False) <- assocs sb ]
  prmslsts = map elems $ makePrimePagesFrom 0 szbts sb2prmsa thrdd

-- count the primes from the sieved page list to the limit...
countPrimesTo :: Prime -> Int -> Bool -> Int64
countPrimesTo limit szbts thrdd =
  let lmtndx = fromIntegral $ (limit - cFRSTSVPRM) `shiftR` 1 :: PrimeNdx
      sb2cnt lwi sb@(UArray _ _ rng _) =
        let nlwi = lwi + fromIntegral rng in
        if nlwi < lmtndx then (countSieveBuffer (rng - 1) sb, nlwi)
        else (countSieveBuffer (fromIntegral (lmtndx - lwi)) sb, nlwi)
      loop [] cnt = cnt
      loop ((cnt, nxtlwi) : cntstl) ocnt =
        if nxtlwi > lmtndx then ocnt + fromIntegral cnt
        else loop cntstl $ ocnt + fromIntegral cnt
  in if limit < cFRSTSVPRM then
       if limit < 2 then 0 else 1
     else loop (makePrimePagesFrom 0 szbts sb2cnt thrdd) 1
 
-- test it...
main :: IO ()
main = do
  let limit = 10^9 :: Prime
  -- page segmentation sized for most efficiency;
  -- fastest with CPU L1 cache size but more address calculation overhead;
  -- a little slower with CPU L2 cache size but just about enough to
  -- cancell out the gain from reduced page start address calculations...
  let cSIEVEPGSZ = (2^18) * 8 :: Int -- CPU L2 cache size in bits
  let threaded = True

  putStrLn $ "There are " ++ show cNUMPROCS ++ " threads available."
 
  strt <- getPOSIXTime
--  let answr = length $ takeWhile (<= limit) $ primes cSIEVEPGSZ threaded -- slow way
  let answr = countPrimesTo limit cSIEVEPGSZ threaded -- fast way
  stop <- answr `seq` getPOSIXTime -- force evaluation of answr b4 stop time!
  let elpsd = round $ 1e3 * (stop - strt) :: Int64
 
  putStr $ "Found " ++ show answr
  putStr $ " primes up to " ++ show limit
  putStrLn $ " in " ++ show elpsd ++ " milliseconds."

这已经从我上面提到的 RosettaCode 提交中进行了重构,通过使主筛循环和辅助基础主要进料循环具有不同的筛缓冲区大小成为可能,以及添加多线程(如上所述在 DF 之上进行了改进) )。它的运行速度与 DF 在 CpC 中提到的最终答案的速度大致相同(分别为 3.7/4.1 CpC)在我的带有 LLVM 单线程的 Skylake 机器上,由于如上所述的不够“粗粒度”的问题,大约有一半是多线程的。

这个答案只比 DF 的代码快不到两倍,主要是由于推荐使用 LLVM 后端。

【讨论】:

    【解决方案2】:

    一些狂热的 C++ 程序员声称他的程序只需要 8 秒。

    这是挂钟时间还是 CPU 时间?

    如果是挂钟,并且任务被分成 100 个 CPU,比如说,它不是很令人印象深刻(它还不错),如果分成 1000 个,那就太可怜了。

    如果是 CPU 时间:

    我很确定实际上 筛分 到 1011 还没有达到时间。

    在此之前,如果有几个超过 4×109 个素数,假设一个正常的 2-3GHz CPU,每个素数有 4-6 个周期。

    使用 Eratosthenes 的筛子或 Atkin 的筛子无法实现这一目标。每个素数都必须被检查和计数,每个复合材料都必须这样标记和检查。这给出了筛子中每个数字两个循环的理论下限,不包括例如数组初始化、循环边界检查、循环变量更新、冗余标记。你不会接近那个理论界限。

    几个数据点:

    Daniel Bernstein's primegen(阿特金筛),调整筛分块以充分利用我的 32KB L1 缓存,需要 90 秒将素数筛分到 1011 并计算它们(234在我的 Core i5 2410M (2.3GHz) 上,默认筛块大小为 8K 字的秒数。 (它针对高达 232 的范围进行了优化,但在此范围内,它变得明显更慢,对于 109 的限制,时间分别为 0.49 和 0.64 秒。)

    My segmented Sieve of Eratosthenes,使用一些未暴露的内部结构来避免创建列表,在 340 秒内筛选并计数到 1011(嗅探:-/,但是嘿,对于 109sup> 花了 2.92 秒 - 它越来越近了,在 1012 和 1013 之间的某个地方,它超过了primegen :) 使用暴露的界面大致创建了一个素数列表用 32 位 GHC 编译它所花费的时间加倍。

    所以我敢打赌,如果是 CPU 时间,那么报告的 8 秒时间对于计算素数数量的算法来说是正确的,而无需实际筛选整个过程。正如applicative's answer 所指出的,这可以更快更快地完成。

    dafis@schwartz:~/Haskell/Repos/arithmoi> time tests/primeCount 100000000000
    4118054813
    
    real    0m0.145s
    user    0m0.139s
    sys     0m0.006s
    

    请注意,10^11 对于 32 位算术来说太大了。此外,10^11 位 = 12.5 GB,太多的数据无法放入 Haskell 的 32 位地址空间。所以你不能一次把整个位图放在内存中。

    要筛选该范围,您必须使用分段筛。即使您不受 32 位地址空间的限制,使用如此大的数组也会由于频繁的缓存未命中而产生糟糕的性能。您的程序将花费大部分时间来等待从主内存传输的数据。筛选适合您的 L2 缓存的块(我没有成功尝试通过使筛子适合 L1 来使其更快,我猜 GHC 运行时的开销太大而无法使其工作)。

    此外,从筛子中消除一些小素数的倍数,这减少了所需的工作,并通过使筛子更小来进一步提高性能。消除偶数很简单,3 的倍数很容易,5 的倍数不是很困难。

    最后,请注意,小于 10^11 的素数只是小于 2^32 的阴影,因此您也无法一次存储所有实际整数。

    如果您将筛子存储为位数组列表,并删除 2、3 和 5 的倍数,则需要大约 3.3GB 来存储块,所以如果您真的可以拥有高达 4GB 的空间,它会适合。但是你应该让你不再需要的块立即被垃圾收集。

    (以防万一,我当前的实现使用 IOUArray Integer Bool 和多个线程来处理搜索空间的独立子范围。目前从 10MB 数组块中删除所有 2 的倍数需要几秒钟...)

    这很重要。

    • 使用Int 作为索引,使用unsafeRead/unsafeWrite 读取和修改数组。 Integer 计算比 Int 计算慢得多,并且您使用 readArray/writeArray really hurts 获得的边界检查。
    • 10MB 块太大了,你会失去缓存局部性。最多使用几百 KB 的块(L2 缓存减去一些空间用于其他需要的东西)。
    • 不过,即使使用 Integer 索引、边界检查和 10MB 块,删除 2 的倍数也不应该花费几秒钟。我们可以看看你的代码吗?

    假期后更新:

    8 分钟 筛出最多 1011 的素数是可能的,无需深奥的魔法。我看不出从 1 到 4 核如何能产生 8 倍的加速,因为这里不应该有缓存效应,但无论如何,如果没有看到代码,我无法调查。

    让我们看看你的代码。

    首先,一个错误:

    vs <-
      mapM
        (\ start -> do
          let block = (start, start + block_size)
          v <- newEmptyMVar
          writeChan queue $ do
            say $ "New chunk " ++ show block
            chunk <- chunk_new block
            sieve_top base chunk
            c <- chunk_count chunk
            putMVar v c
          return v
        )
        (takeWhile (< target) $ iterate (+block_size) base_max)
    

    数字base_max + k*block_size 出现在两个块中,如果其中任何一个是素数,则该素数被计算两次,你也应该将上限限制在target

    现在到性能方面:

    跳出来的一件事是它真实很健谈,一旦你将block_size 调整到缓存(我为 512KB L2 缓存占用了 256KB 块),它就可以测量了 - 然后通过为stdout 争夺if prime &lt; 100 then say $ "Sieve " ++ show prime else return () 消息而使线程变慢。

    让我们看看您的(静音)筛分循环:

    chunk_sieve :: Chunk -> Integer -> IO ()
    chunk_sieve array prime = do
      (min, max) <- getBounds array
      let n0 = min `mod` prime
      let n1 = if n0 == 0 then min else min - n0 + prime
      mapM_
        (\ n -> writeArray array n (n == prime))
        (takeWhile (<= max) $ iterate (+prime) n1)
    

    花费时间的一件事是,将每个索引与已标出倍数的素数进行比较。每一个比较都很便宜(虽然比Int 比较贵得多),但是大量的比较,其中只有一个可能产生True,加起来。在循环后无条件写入False 并在必要时在素数索引处写入True 会产生相当大的加速。

    出于计时目的,我将目标减少到 109 并在两个内核上运行。原始代码花费了 155 秒(经过,292 秒用户),减少了 block_size 148 秒,静音 143 秒。省略比较,

    mapM_
      (\ n -> writeArray array n False)
      (takeWhile (<= max) $ iterate (+prime) n1)
    when (min <= prime && prime <= max) $ writeArray array prime True
    

    它在 131 秒内运行。

    现在是时候进行更大的加速了。我是否已经提到bounds-checking costs a lot of time?由于循环条件保证不会尝试越界访问(并且素数足够小,不会发生Int-overflow),我们应该真正使用未经检查的访问:

    chunk_sieve :: Chunk -> Integer -> IO ()
    chunk_sieve array prime = do
      (min, max) <- getBounds array
      let n0 = min `mod` prime
          n1 = if n0 == 0 then min else min - n0 + prime
          n2 = fromInteger (n1 - min)
          mx = fromInteger (max - min)
          pr = fromInteger prime
      mapM_
        (\ n -> unsafeWrite array n False)
        (takeWhile (<= mx) $ iterate (+pr) n2)
      when (min <= prime && prime <= max) $ writeArray array prime True
    

    这将运行时间减少到 96 秒。好多了,但仍然很糟糕。罪魁祸首是

    takeWhile (<= mx) $ iterate (+pr) n2
    

    GHC 不能很好地融合该组合,并且您会得到一个已遍历的装箱 Ints 列表。将其替换为算术序列[n2, n2+pr .. mx],GHC 会愉快地使用未装箱的Int#s 创建一个循环,时间为 37 秒。

    好多了,但仍然很糟糕。现在最大的时间消耗是

    chunk_count :: Chunk -> IO Integer
    chunk_count array = do
        (min, max) <- getBounds array
        work min max 0
      where
        work i max count = do
          b <- readArray array i
          let count' = count + if b then 1 else 0
          evaluate count'
          let i' = i+1
          if i' > max
            then return count'
            else work i' max count'
    

    同样,边界检查会花费 大量 时间。与

    chunk_count :: Chunk -> IO Integer
    chunk_count array = do
        (min, max) <- getBounds array
        work 0 (fromInteger (max-min)) 0
      where
        work i max count = do
          b <- unsafeRead array i
          let count' = count + if b then 1 else 0
          evaluate count'
          let i' = i+1
          if i' > max
            then return count'
            else work i' max count'
    

    我们只剩 15 秒了。现在,evaluate count' 是一种使work 严格限制在count 中的方法有点昂贵。在最后一行使用else work i' max $! count' 代替evaluate 将运行时间减少到13 秒。以更适合(至少对于 GHC)的方式定义 work

    chunk_count :: Chunk -> IO Integer
    chunk_count array = do
        (min, max) <- getBounds array
        let mx = fromInteger (max-min)
            work i !ct
                | mx < i    = return ct
                | otherwise = do
                    b <- unsafeRead array i
                    work (i+1) (if b then ct+1 else ct)
        work 0 0
    

    将时间缩短至 6.55 秒。现在我们处于say $ "New chunk " ++ show block 产生显着差异的情况下,禁用它使我们缩短到 6.18 秒。

    但是,通过从数组中读取一个字节来计算设置位、屏蔽不需要的位并将每个单独的位与 0 进行比较并不是最有效的方法。从数组中读取整个Words 更快(通过castIOUArray)并使用popCount,如果“你知道你在做什么......”,这可以让我们缩短到4.25秒;当素数的平方大于块的上限时停止标记

    sieve_top :: Chunk -> Chunk -> IO ()
    sieve_top base chunk = work 2
      where
        work prime = do
          chunk_sieve chunk prime
          mp <- chunk_next_prime base prime
          case mp of
            Nothing -> return ()
            Just p' -> do
                (_,mx) <- getBounds chunk
                when (p'*p' <= mx) $ work p'
    

    到 3.9 秒。仍然不壮观,但考虑到我们从哪里开始,还不错。只是为了说明在减少其他不良行为后缓存局部性的重要性:具有原始 10MB 块大小的相同代码需要 8.5 秒。

    代码中的另一个小问题是所有线程都使用相同的可变小素数数组进行筛选。由于它是可变的,因此必须同步对其的访问,这会增加一些开销。只有两个线程,开销不会太大,使用 immutable 副本进行筛选只会将时间减少到 3.75 秒,但我希望更多线程的效果会更大。 (我只有两个物理内核 - 带有超线程 - 所以使用两个以上的线程来做同样的工作会导致减速,这可能会使从中得出的结论无效,但是使用四个线程,我用不可变数组得到 4.55 秒而不是 5.3 秒使用可变数组。这似乎证实了不断增长的同步开销。)

    通过消除更多的Integer 计算和为 GHC 的优化器编写代码(更多的工作器/包装器转换,一些静态参数转换),仍有一些收获,但不是很多,可能 10-15%。

    下一个重大改进是通过从筛子中消除偶数来获得。这将工作、分配和运行时间减少了一半以上。任何素筛都不应该考虑偶数,真的,那只是无意义的浪费工作。

    【讨论】:

    • 感谢您提供广泛而彻底的回答。在我的时区有点晚了,但希望明天我可以尝试你的一些建议。
    • @MathematicalOrchid 你有没有尝试过任何建议,如果有,你从他们那里得到了什么?我从他们那里得到了很多;)
    • @DanielFischer,感谢您详细描述了您的 arithmoi 包的 Eratosthenes Sieve 实现所使用的编程方法。在该软件包的“待办事项”列表中,您提到想要尝试阿特金筛法算法。由于my answer to a C# question 中列出的原因,我建议您最好将时间花在进一步扩展车轮分解和多处理上;与 SoA 的小经验运行时间优势相比,最大车轮分解对 SoE 的影响更大。
    • @GordonBGood 是的,我不指望能写出能击败像样的 Eratosthenes 筛子的 Atkin Sieve。尽管如此,原则上我很好奇它的表现如何,所以它仍然是“当我有时间和兴趣时,我会尝试一下”。但在此之前,当我可以让自己去做时,Eratosthenes 筛子已经在生产线上进行了改进。
    • @DanielFischer,完全优化的阿特金筛 (SoA) 可能会击败您目前在 arithmoi 中实施的埃拉托色尼筛 (SoE),它只有 2、3、5 轮分解,但是SoE 不仅限于此 - 2、3、5、7 轮非常容易实现,并且可以使用更多轮。一种简单的优化是使用大轮模式(例如 2、3、5、7、11、13、17 轮)预初始化可变数组,并使用较小的轮进一步剔除以显着减少总数的剔除操作。 SoA 已将 2、3、5 轮烘焙到二次曲线中。
    【解决方案3】:

    使用 StackOverflow 老师 Daniel Fischer 的包arithmoi

    import Math.NumberTheory.Primes.Counting
    
    main = print $ primeCount (10^11)
    
    -- $ time ./arith
    -- 4118054813
    -- 
    -- real 0m0.209s
    -- user 0m0.198s
    -- sys  0m0.008s
    

    这比你“狂热”的 C++ 朋友写的任何东西都要快 40 倍;也许他可以通过查看 Haskell 源代码来学习一两件事...这里是 the haddocks

    【讨论】:

    • 如果您至少不知道每个系统的硬件规格,那么将这个程序的运行时间与 C++ 版本进行比较是没有意义的。
    • 我的“硬件”是一个半古董的 macbook。如果MathematicalOrchid 像我一样发布任何代码,我当然会给出更科学的比较。机器之间的差异远远超过了一个数量级——在 MathematicalOrchids Haskell 程序的情况下有许多数量级——表明 arithmoi 可能会有所作为。
    • @applicative,当然,您实际上阅读了“黑线鳕”链接并意识到您从“arithmoi”包中使用的函数“primeCount”对于大于 30,000 的范围不是筛子,应该t 在执行时间上与此处的筛子进行比较;它是由 D. H. Lehmer 开发并在 'arithmoi' 包中实现的特殊素数计数函数。
    • @starflyer,虽然您是对的,我们应该将同类与同类进行比较,但您为什么认为相同的算法(在本例中为“Meissel-Lehmer 方法”)如果用 C++ 编写会更快比在 Haskell 中?以我的经验,Haskell 编译器非常高效,当与同样非常高效的 LLVM 后端结合使用时,Haskell 通常可以生成至少与 C 中的等效代码一样快的机器代码 只要选择正确的 Haskell 功能以优化速度而不是程序简洁
    • @applicative,调用外部包似乎没有教任何人任何东西,除了它存在。如果要调用外部程序,不妨将同一作者的最新技术primecount 称为“primesieve”,并在 10 毫秒内将素数计数为 10^11,将素数计数为64 位 nuber 范围(大约 1.8 * 10^19)在几分钟内。如果坚持从 Haskell 调用它,则编写一个包装器并通过外部函数接口 (FFI) 调用它。但我再说一遍,一个人学到了什么?
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2021-12-19
    • 2017-11-14
    • 2012-07-30
    • 2020-07-22
    • 1970-01-01
    • 2014-06-12
    相关资源
    最近更新 更多