诊断并行单子性能

Question

我使用 Attoparsec 库编写了一个 Bytestring 解析器：

import qualified Data.ByteString.Char8 as B
import qualified Data.Attoparsec.ByteString.Char8 as P

parseComplex :: P.Parser Complex

我的意图是使用此解析大型（> 5 Gb）文件，因此实现懒惰地使用此解析器：

import qualified Data.ByteString.Lazy.Char8 as LB
import qualified Data.Attoparsec.ByteString.Lazy as LP

extr :: LP.Result a -> a

main = do
    rawData <- liftA LB.words (LB.readFile "/mnt/hgfs/outputs/out.txt")
    let formatedData = map (extr.LP.parse parseComplex) rawData
    ...

在带有-O2 和-s 标志的测试文件上执行此操作，我明白了：

 3,509,019,048 bytes allocated in the heap
     2,086,240 bytes copied during GC
        58,256 bytes maximum residency (30 sample(s))
       126,240 bytes maximum slop
             2 MB total memory in use (0 MB lost due to fragmentation)

                                  Tot time (elapsed)  Avg pause  Max pause
Gen  0      6737 colls,     0 par    0.03s    0.03s     0.0000s    0.0001s
Gen  1        30 colls,     0 par    0.00s    0.00s     0.0001s    0.0002s

INIT    time    0.00s  (  0.00s elapsed)
MUT     time    0.83s  (  0.83s elapsed)
GC      time    0.04s  (  0.04s elapsed)
EXIT    time    0.00s  (  0.00s elapsed)
Total   time    0.87s  (  0.86s elapsed)

%GC     time       4.3%  (4.3% elapsed)

Alloc rate    4,251,154,493 bytes per MUT second

Productivity  95.6% of total user, 95.8% of total elapsed

由于我是独立地将一个函数映射到一个列表上，我认为这段代码可能会受益于并行化。我之前在 Haskell 中从来没有做过任何类似的事情，但在使用 Control.Monad.Par 库时，我编写了一个简单、天真的、静态分区函数，我认为它可以并行映射我的解析：

import Control.Monad.Par

parseMap :: [LB.ByteString] -> [Complex]
parseMap x = runPar $ do
    let (as, bs) = force $ splitAt (length x `div` 2) x
    a <- spawnP $ map (extr.LP.parse parseComplex) as 
    b <- spawnP $ map (extr.LP.parse parseComplex) bs
    c <- get a
    d <- get b
    return $ c ++ d

我并没有对这个函数有太多期望，但是并行计算的性能比顺序计算差得多。下面是main函数和结果，用-O2 -threaded -rtsopts编译，用+RTS -s -N2执行：

main = do
    rawData <- liftA LB.words (LB.readFile "/mnt/hgfs/outputs/out.txt")
    let formatedData = parseMap rawData
    ...

---

 3,641,068,984 bytes allocated in the heap
   356,490,472 bytes copied during GC
    82,325,144 bytes maximum residency (10 sample(s))
    14,182,712 bytes maximum slop
           253 MB total memory in use (0 MB lost due to fragmentation)

                                  Tot time (elapsed)  Avg pause  Max pause
Gen  0      4704 colls,  4704 par    0.50s    0.25s     0.0001s    0.0006s
Gen  1        10 colls,     9 par    0.57s    0.29s     0.0295s    0.1064s

Parallel GC work balance: 19.77% (serial 0%, perfect 100%)

TASKS: 4 (1 bound, 3 peak workers (3 total), using -N2)

SPARKS: 0 (0 converted, 0 overflowed, 0 dud, 0 GC'd, 0 fizzled)

INIT    time    0.00s  (  0.00s elapsed)
MUT     time    1.11s  (  0.72s elapsed)
GC      time    1.07s  (  0.54s elapsed)
EXIT    time    0.02s  (  0.02s elapsed)
Total   time    2.20s  (  1.28s elapsed)

Alloc rate    3,278,811,516 bytes per MUT second

Productivity  51.2% of total user, 88.4% of total elapsed

gc_alloc_block_sync: 149514
whitehole_spin: 0
gen[0].sync: 0
gen[1].sync: 32

如您所见，在并行情况下似乎有很多垃圾收集器活动，并且负载平衡很差。我使用 threadscope 分析了执行情况并得到以下信息：

我可以非常清楚地看到在 HEC 1 上运行的垃圾收集器正在中断 HEC 2 上的计算。此外，HEC 1 分配的工作显然比 HEC 2 少。作为测试，我尝试调整两个拆分的相对大小列出重新平衡负载的列表，但这样做后我没有看到程序行为的明显差异。我还尝试在不同大小的输入上运行它，使用更大的最小堆分配，并且只使用 Control.Monad.Par 库中包含的 parMap 函数，但这些努力也对结果没有影响。

我假设某处存在空间泄漏，可能来自let (as,bs) = ... 分配，因为在并行情况下内存使用率要高得多。这是问题吗？如果是这样，我应该如何解决它？

---

编辑：按照建议手动拆分输入数据，我现在看到时间上有一些小的改进。对于 6m 点输入文件，我手动将文件拆分为两个 3m 点文件和三个 2m 点文件，并分别使用 2 核和 3 核重新运行代码。大致时间如下：

1 个核心：6.5 秒

2 核：5.7 秒

3 核：4.5 秒

新的 threadscope 配置文件如下所示：

一开始的奇怪行为已经消失，但现在仍然有一些在我看来仍然存在一些明显的负载平衡问题。

Answer 1

首先，我建议您参考您的代码审查帖子(link)，以便为人们提供有关您正在尝试做的事情的更多背景信息。

您的基本问题是您强制 Haskell 使用 length x 将整个文件读入内存。您要做的是将结果流式传输，以便随时在内存中保留尽可能少的文件。

您拥有的是典型的 map-reduce 计算，因此要将工作负载分为两部分，我的建议是：

1. 两次打开输入文件，创建两个文件句柄。
2. 将第二个句柄放在文件的“中间”。
3. 创建两个计算 - 每个文件句柄一个。
4. 第一个计算将从其句柄中读取，直到到达“中间”；第二个将从其句柄中读取，直到到达文件末尾。
5. 每次计算都会创建一个Vector Int
6. 每次计算完成后，我们将两个向量组合在一起（将向量按元素相加。）

当然，文件的“中间”是靠近文件中间的一行的开头。

棘手的部分是第 4 步，因此为了简化事情，我们假设输入文件已经被分成两个单独的文件 part1 和 part2。那么你的计算可能如下所示：

main = do
    content1 <- LB.readFile "part1"
    content2 <- LB.readFile "part2"
    let v = runPar $ do a <- spawnP $ computeVector content1
                        b <- spawnP $ computeVector content2
                        vec1 <- get a
                        vec2 <- get b
                        -- combine vec1 and vec2
                        let vec3 = ...vec1 + vec2...
                        return vec3
    ...

您应该尝试这种方法并确定加速比是多少。如果看起来不错，那么我们可以弄清楚如何虚拟地将文件拆分为多个部分，而无需实际复制数据。

注意 - 我实际上并没有运行它，所以我不知道是否有 w.r.t 的怪癖。 lazy-IO 和 Par monad，但这种想法在某种形式下应该可行。