在 Haskell 中 Floyd-Warshall 的表现——修复空间泄漏

Posted 2023-02-19

【中文标题】在 Haskell 中 Floyd-Warshall 的表现——修复空间泄漏

【英文标题】：Performance of Floyd-Warshall in Haskell – Fixing a space leak

【发布时间】：2013-10-13 18:20:22

【问题描述】：

我想使用Vectors 在 Haskell 中编写 Floyd-Warshall 所有对最短路径算法的有效实现，以期获得良好的性能。

实现非常简单，但不是使用 3 维 |V|×|V|×|V|矩阵，使用二维向量，因为我们只读取了之前的 k 值。

因此，算法实际上只是传入一个二维向量，并生成一个新的二维向量的一系列步骤。最终的二维向量包含所有节点 (i,j) 之间的最短路径。

我的直觉告诉我，确保在每一步之前评估前一个 2D 向量很重要，所以我在 prev 函数的 prev 参数上使用了 BangPatterns 和严格的 foldl' ：

-# Language BangPatterns #-

import           Control.DeepSeq
import           Control.Monad       (forM_)
import           Data.List           (foldl')
import qualified Data.Map.Strict     as M
import           Data.Vector         (Vector, (!), (//))
import qualified Data.Vector         as V
import qualified Data.Vector.Mutable as V hiding (length, replicate, take)

type Graph = Vector (M.Map Int Double)
type TwoDVector = Vector (Vector Double)

infinity :: Double
infinity = 1/0

-- calculate shortest path between all pairs in the given graph, if there are
-- negative cycles, return Nothing
allPairsShortestPaths :: Graph -> Int -> Maybe TwoDVector
allPairsShortestPaths g v =
  let initial = fw g v V.empty 0
      results = foldl' (fw g v) initial [1..v]
  in if negCycle results
        then Nothing
        else Just results
  where -- check for negative elements along the diagonal
        negCycle a = any not $ map (\i -> a ! i ! i >= 0) [0..(V.length a-1)]

-- one step of the Floyd-Warshall algorithm
fw :: Graph -> Int -> TwoDVector -> Int -> TwoDVector
fw g v !prev k = V.create $ do                                           -- ← bang
  curr <- V.new v
  forM_ [0..(v-1)] $ \i ->
    V.write curr i $ V.create $ do
      ivec <- V.new v
      forM_ [0..(v-1)] $ \j -> do
        let d = distance g prev i j k
        V.write ivec j d
      return ivec
  return curr

distance :: Graph -> TwoDVector -> Int -> Int -> Int -> Double
distance g _ i j 0 -- base case; 0 if same vertex, edge weight if neighbours
  | i == j    = 0.0
  | otherwise = M.findWithDefault infinity j (g ! i)
distance _ a i j k = let c1 = a ! i ! j
                        c2 = (a ! i ! (k-1))+(a ! (k-1) ! j)
                        in min c1 c2

但是，当使用 1000 个节点和 47978 条边的图运行这个程序时，事情看起来一点也不好看。内存使用率非常高，程序运行时间太长。该程序是用ghc -O2编译的。

我重新构建了分析程序，并将迭代次数限制为 50：

 results = foldl' (fw g v) initial [1..50]

然后我使用+RTS -p -hc 和+RTS -p -hd 运行程序：

这……很有趣，但我想这表明它正在积累大量的 thunk。不好。

好的，所以在黑暗中拍摄了几张照片后，我在fw 中添加了一个deepseq 以确保prev 真的 被评估：

let d = prev `deepseq` distance g prev i j k

现在情况看起来好多了，我实际上可以在持续使用内存的情况下运行程序完成。很明显，prev 参数的轰动是不够的。

为了与之前的图表进行比较，这是添加deepseq后50次迭代的内存使用情况：

好的，情况好多了，但我还有一些问题：

这是解决此空间泄漏问题的正确解决方案吗？我觉得插入deepseq 有点难看是错的？ 我在这里使用Vectors 是惯用的/正确的吗？我正在为每次迭代构建一个全新的向量，并希望垃圾收集器将删除旧的 Vectors。 我还能做些什么来通过这种方法加快运行速度吗？

供参考，这里是graph.txt：http://sebsauvage.net/paste/?45147f7caf8c5f29#7tiCiPovPHWRm1XNvrSb/zNl3ujF3xB3yehrxhEdVWw=

这里是main：

main = do
  ls <- fmap lines $ readFile "graph.txt"
  let numVerts = head . map read . words . head $ ls
  let edges = map (map read . words) (tail ls)
  let g = V.create $ do
        g' <- V.new numVerts
        forM_ [0..(numVerts-1)] (\idx -> V.write g' idx M.empty)
        forM_ edges $ \[f,t,w] -> do
          -- subtract one from vertex IDs so we can index directly
          curr <- V.read g' (f-1)
          V.write g' (f-1) $ M.insert (t-1) (fromIntegral w) curr
        return g'
  let a = allPairsShortestPaths g numVerts
  case a of
    Nothing -> putStrLn "Negative cycle detected."
    Just a' -> do
      putStrLn  $ "The shortest, shortest path has length "
              ++ show ((V.minimum . V.map V.minimum) a')

【问题讨论】：

附注：any not $ map (\i -> a ! i ! i >= 0) [0..(V.length a-1)] 只是any (\i -> a ! i ! i < 0) [0..(V.length a-1)]。

您是否尝试使用可变向量将foldl' 和forM_ 计算重写为显式循环？ （如 in test0 here 所做的那样，尽管使用数组，而不是向量。和 here with loops in place of usual forM）

@WillNess：不，我唯一尝试的是用带有严格累加器的尾递归函数替换foldl'，但这似乎没有效果。看到您链接到的两个示例都充斥着unsafe* 函数，这有点令人沮丧——我真的希望在不求助于它的情况下也能获得合理的性能。 :-)

您应该使用未装箱的向量。这些将通过简单地插入向量来强制内容。这些示例中的不安全内容只是删除了边界检查。

您的TwoDVectors 只是矩阵，对吧？您是否考虑过为此使用 Repa？ Simon Marlow 在几个不同的上下文中以 FW 为例，如下所示：chimera.labs.oreilly.com/books/1230000000929/…

【参考方案1】：

首先，一些通用的代码清理：

在您的fw 函数中，您明确分配和填充可变向量。但是，有一个用于此目的的预制函数，即generate。 fw 因此可以重写为

V.generate v (\i -> V.generate v (\j -> distance g prev i j k))

同理，图生成代码可以替换为replicate和accum：

let parsedEdges = map (\[f,t,w] -> (f - 1, (t - 1, fromIntegral w))) edges
let g = V.accum (flip (uncurry M.insert)) (V.replicate numVerts M.empty) parsedEdges

请注意，这完全消除了对突变的所有需求，而不会损失任何性能。

现在，到实际问题：

根据我的经验，deepseq 非常有用，但只能快速修复像这样的空间泄漏。根本的问题不是你需要在产生结果之后强加结果。相反，deepseq 的使用意味着您应该首先更严格地构建结构。事实上，如果您在矢量创建代码中添加爆炸模式，如下所示：

let !d = distance g prev i j k

然后问题在没有deepseq 的情况下得到解决。请注意，这不适用于generate 代码，因为出于某种原因（我可能会为此创建一个功能请求），vector 没有为盒装向量提供严格的函数。但是，当我在回答问题 3 时使用未装箱的向量（它们是严格的）时，这两种方法都可以在没有严格性注释的情况下工作。

据我所知，重复生成新向量的模式是惯用的。唯一不习惯的是使用可变性 - 除非它们是绝对必要的，否则通常不鼓励使用可变向量。

有几件事要做：

最简单的方法是，您可以将Map Int 替换为IntMap。因为这并不是函数的真正慢点，所以这并不重要，但IntMap 对于繁重的工作负载可能会更快。

您可以切换到使用未装箱的向量。尽管外部向量必须保持装箱，但由于向量的向量不能取消装箱，内部向量可以。这也解决了您的严格性问题 - 因为未装箱的向量在其元素中是严格的，所以您不会出现空间泄漏。请注意，在我的机器上，这会将性能从 4.1 秒提高到 1.3 秒，因此拆箱非常有帮助。

您可以将向量展平为一个向量，并使用乘法和除法在二维索引和一维索引之间切换。我不建议这样做，因为它有点复杂，很丑陋，并且由于划分，实际上会减慢我机器上的代码。

您可以使用repa。这具有自动并行化代码的巨大优势。请注意，由于repa 将其数组展平并且显然没有正确摆脱需要很好地填充的分区（可以使用嵌套循环，但我认为它使用单个循环和一个分区），它具有与我上面提到的相同的性能损失，将运行时间从 1.3 秒提高到 1.8 秒。但是，如果您启用并行性并使用多核机器，您就会开始看到一些好处。不幸的是，您当前的测试用例太小，看不到太多好处，所以，在我的 6 核机器上，我看到它回落到 1.2 秒。如果我将大小恢复到 [1..v] 而不是 [1..50]，并行度会将其从 32 秒提高到 13 秒。大概，如果你给这个程序一个更大的输入，你可能会看到更多的好处。

如果你有兴趣，我已经发布了我的repa-ified 版本here。

编辑：使用-fllvm。在我的计算机上使用repa 进行测试，我得到了 14.7 秒没有并行性，这几乎与没有-fllvm 和并行性一样好。一般来说，LLVM 可以很好地处理这样的基于数组的代码。

【讨论】：

非常感谢！我会在接下来的几天里研究这个——这里有很多很棒的信息。 :)