优化从文件读取数据的Haskell代码

Question

我正在尝试实现Kosaraju的图算法，处理一个包含350万行数据的文件，每一行都是两个整数（以空格分隔）表示图的边。首先，我需要创建一个汇总数据结构，其中包含节点及其入边和出边列表。下面的代码实现了这一点，但需要一分钟以上的时间，而我可以从MOOC论坛上的帖子中看到，使用其他语言的人只需要不到10秒钟。 (getLines需要10秒，而我读到的基准测试中只需不到1秒。)

我对Haskell还很陌生，并使用foldl'实现了累加方法('是使其终止的关键)，但感觉风格相当命令式，并希望这就是它运行缓慢的原因。此外，我当前计划使用类似的模式进行深度优先搜索，担心所有这些操作都会变得太慢。

我找到了这篇演示文稿和这篇博客，讨论了这些问题，但对于我来说太高级了。

import System.IO
import Control.Monad
import Data.Map.Strict as Map
import Data.List as L

type NodeName = Int
type Edges = [NodeName]
type Explored = Bool

data Node = Node Explored (Edges, Edges) deriving (Show)

type Graph1 = Map NodeName Node

getLines :: FilePath -> IO [[Int]]
getLines = liftM (fmap (fmap read . words) . lines) . readFile

getLines' :: FilePath -> IO [(Int,Int)]
getLines' = liftM (fmap (tuplify2 . fmap read . words) . lines) . readFile

tuplify2 :: [a] -> (a,a)
tuplify2 [x,y] = (x,y)

main = do
    list <- getLines "testdata.txt"  -- [String]
    --list <- getLines "SCC.txt"  -- [String]   
    let
        list' = createGraph list
    return list'

createGraph :: [[Int]] -> Graph1
createGraph xs = L.foldl' build Map.empty xs
    where
        build :: Graph1-> [Int] -> Graph1
        build = \acc (x:y:_) ->
            let tmpAcc = case Map.lookup x acc of
                Nothing -> Map.insert x (Node False ([y],[])) acc
                Just a -> Map.adjust (\(Node _ (fwd, bck)) -> (Node False ((y:fwd), bck))) x acc
            in case Map.lookup y tmpAcc of
                Nothing -> Map.insert y (Node False ([],[x])) tmpAcc
                Just a -> Map.adjust (\(Node _ (fwd, bck)) -> (Node False (fwd, (x:bck)))) y tmpAcc

- Simon H

---

你应该考虑使用数组。此外，仅包含两个元素的列表应立即引起怀疑。 - n. m.

将查看数组。这个两个元素的列表直接来自源文件，表示一个边缘。 - Simon H

节点索引的范围是否事先已知？ - András Kovács

不要猜测，进行个人资料描述。 - Don Stewart

@DonStewart 当我看到这个问题时，我想起了你。 - AndrewC

Answer 1

使用地图：

• 尽可能使用IntMap或HashMap。对于Int键，两者都比Map快得多。HashMap通常比IntMap更快，但使用的RAM更多，库也不如IntMap丰富。
• 不要进行不必要的查找。 containers包有大量专门的函数。与问题中的createGraph实现相比，使用alter可以将查找次数减半。

createGraph的示例：

import Data.List (foldl')
import qualified Data.IntMap.Strict as IM

type NodeName = Int
type Edges = [NodeName]
type Explored = Bool

data Node = Node Explored Edges Edges deriving (Eq, Show)
type Graph1 = IM.IntMap Node

createGraph :: [(Int, Int)] -> Graph1
createGraph xs = foldl' build IM.empty xs
    where
        addFwd y (Just (Node _ f b)) = Just (Node False (y:f) b)
        addFwd y _                   = Just (Node False [y] [])
        addBwd x (Just (Node _ f b)) = Just (Node False f (x:b))
        addBwd x _                   = Just (Node False [] [x])

        build :: Graph1 -> (Int, Int) -> Graph1
        build acc (x, y) = IM.alter (addBwd x) y $ IM.alter (addFwd y) x acc 

使用vectors:

• 考虑使用高效的构建函数（累加器、展开、generate、iterate、constructN等）。这些函数可能在幕后使用了变异，但比实际可变向量更方便使用。
• 在更一般的情况下，使用盒式向量的惰性来在构建向量时启用自我引用。
• 尽可能使用非盒式向量。
• 当您绝对确定边界时，请使用不安全的函数。
• 仅在没有纯粹的替代品时使用可变向量。在这种情况下，优先使用ST单子而不是IO。此外，避免创建many mutable heap objects（即首选可变向量而不是可变引用的不可变向量）。

createGraph的示例：

import qualified Data.Vector as V

type NodeName = Int
type Edges = [NodeName]
type Explored = Bool

data Node = Node Explored Edges Edges deriving (Eq, Show)
type Graph1 = V.Vector Node

createGraph :: Int -> [(Int, Int)] -> Graph1
createGraph maxIndex edges = graph'' where
    graph    = V.replicate maxIndex (Node False [] [])
    graph'   = V.accum (\(Node e f b) x -> Node e (x:f) b) graph  edges
    graph''  = V.accum (\(Node e f b) x -> Node e f (x:b)) graph' (map (\(a, b) -> (b, a)) edges)

请注意，如果节点索引范围中存在间隙，则最好在执行任何其他操作之前连续重新标记索引。或者，在Node中引入一个空构造函数来表示缺少的索引。

更快的I/O：

使用Data.Text或Data.ByteString中的IO函数。在这两种情况下，还有将输入分成行或单词的高效函数。

示例：

import qualified Data.ByteString.Char8 as BS
import System.IO

getLines :: FilePath -> IO [(Int, Int)]
getLines path = do
    lines <- (map BS.words . BS.lines) `fmap` BS.readFile path
    let pairs = (map . map) (maybe (error "can't read Int") fst . BS.readInt) lines
    return [(a, b) | [a, b] <- pairs]

基准测试：

始终进行基准测试，不像我在这个答案中。使用criterion。

- András Kovács

---

这太棒了，我将逐步学习。学习标准可能在短期内有些过多，但是我尝试了你的IntMap解决方案，它将运行时间从113秒减少到100秒（而且你的代码还包括tuplify，相对于我的基准测试增加了一些时间）。更多内容将会跟进。 - Simon H

Criterion 其实非常简单。它的最简单用法包括导入 Criterion.Main，然后使用 defaultMain、bench，以及你需要的 nf、whnf、nfIO 或者 whnfIO 中的任何一个。 - Carl

即使您只是在一些预定义的输入上运行createGraph的简单测试，第三方criterion也非常重要。能够看到运行的分布情况，并进行准确的测量（而不是使用time）将为您节省大量时间和头痛。一旦您设置了第一个测试，就可以很容易地添加其他程序部分，以确保您对代码性能有准确的视图。 - jberryman

我在我的答案中添加了一些 Criterion 代码（但没有更改主要部分）。我得到了一堆数字，但找不到一个简单的运行时间。 - Simon H

Answer 2

根据András的建议，我已将一个需时113秒的任务缩短至24秒（使用秒表测量，因为我还没能让Criterion起作用）（通过编译-O2又缩短至10秒）!!! 去年我参加了一些课程，讲述了针对大型数据集进行优化的挑战，但这是我第一次面临一个涉及到大型数据集并非常规的问题。这比我的教练所建议的还要棘手。这是我的成果:

import System.IO
import Control.Monad
import Data.List (foldl')
import qualified Data.IntMap.Strict as IM
import qualified Data.ByteString.Char8 as BS

type NodeName = Int
type Edges = [NodeName]
type Explored = Bool

data Node = Node Explored Edges Edges deriving (Eq, Show)
type Graph1 = IM.IntMap Node

-- DFS uses a stack to store next points to explore, a list can do this
type Stack = [(NodeName, NodeName)]

getBytes :: FilePath -> IO [(Int, Int)]
getBytes path = do
    lines <- (map BS.words . BS.lines) `fmap` BS.readFile path
    let
        pairs = (map . map) (maybe (error "Can't read integers") fst . BS.readInt) lines
    return [(a,b) | [a,b] <- pairs]

main = do
    --list <- getLines' "testdata.txt"  -- [String]
    list <- getBytes "SCC.txt"  -- [String] 
    let list' = createGraph' list
    putStrLn $ show $ list' IM.! 66
    -- return list'


bmark = defaultMain [
    bgroup "1" [
        bench "Sim test" $ whnf bmark' "SCC.txt"
        ]
    ]

bmark' :: FilePath -> IO ()
bmark' path = do
    list <- getLines path
    let
        list' = createGraph list
    putStrLn $ show $ list' IM.! 2


createGraph' :: [(Int, Int)] -> Graph1
createGraph' xs = foldl' build IM.empty xs
    where
        addFwd y (Just (Node _ f b)) = Just (Node False (y:f) b)
        addFwd y _                   = Just (Node False [y] [])
        addBwd x (Just (Node _ f b)) = Just (Node False f (x:b))
        addBwd x _                   = Just (Node False [] [x])

        build :: Graph1 -> (Int, Int) -> Graph1
        build acc (x, y) = IM.alter (addBwd x) y $ IM.alter (addFwd y) x acc 

现在继续进行剩下的练习...

- Simon H

---

干得好！顺便说一下，我看了你的 SCC.txt，它实际上有一个连续的节点范围，只缺少节点“0”。因此，我可以使用几乎相同的 vector 代码来描述它。这是它的要点。此外，它在我的计算机上运行时间为4.7秒。 - András Kovács

另外，你是否使用优化编译（-O2，可能还包括-fllvm）？我也运行了你刚刚在这里发布的代码，对我来说它在6.3秒内完成了（或者可能是你的电脑速度较慢...我有Core i7 3770 CPU）。 - András Kovács

哇 - 五分钟前我还只是一个简单的Prelude用户 - 现在在第一次使用 ghc -O2 <filename> 进行编译后，时间缩短到了10秒。 - Simon H

你可以将缺失的节点存储为没有入边和出边的节点。你可以在所有缺失的节点之间共享空节点，这样它们只会占用向量中指针的空间。 - Piezoid

好的，我改正了。只需要决定向量是否能在接下来的练习中帮助我（我认为可以），然后重新实现代码 - 这真的是一个很好的学习练习，这就是为什么有这么多人在跟随它的原因。 - Simon H

Answer 3

这并不是一个真正的答案，我更愿意评论András Kovács的帖子，如果我加上那50分......

我已经在IntMap和MVector中实现了图形的加载，试图对可变性与不可变性进行基准测试。

两个程序都使用Attoparsec进行解析。肯定有更经济的方法来做到这一点，但是相对于其高抽象级别（解析器可以占据一行），Attoparsec相对较快。指导方针是避免String和read。 read是部分和缓慢的，[Char]是缓慢而且不够节省内存，除非被正确融合。

正如András Kovács所指出的，IntMap比Map更适合用于Int键。我的代码提供了另一个alter用法示例。如果节点标识符映射是密集的，则还可以使用Vector和Array。它们允许通过标识符进行O(1)索引。

可变版本按需处理MVector的指数增长。这避免了对节点标识符设定上限，但引入了更多复杂性（向量的引用可能会改变）。

我使用标识符范围为[0..2^16]的5M边文件进行了基准测试。 MVector版本比IntMap代码快大约2倍（在我的计算机上为12秒与25秒）。

代码在这里[Gist]。

我会在我这边完成更多分析后进行编辑。

- Piezoid

---

我的解析器出了问题：它累积了很多字节串。我试图让它更严格，但仍然是同样的情况。 - Piezoid