Haskell 性能——拓扑排序不够快

Question

我是 Haskell 初学者，并选择它来解决我的班级的编程任务，但是我的解决方案太慢并且没有被接受。我正在尝试对其进行分析，并希望我可以从更高级的 Haskellers 那里得到一些指导。

到目前为止，我班上唯一被接受的其他解决方案是用 Rust 编写的。我确信我应该能够在 Haskell 中实现类似的性能，并且我编写了可怕的命令式代码以希望提高性能，可惜没有效果。

我的第一个怀疑与 相关work，我用来forever遍历入度数组，直到出现越界异常。我希望这是尾递归并编译为while (true)样式循环。

我的第二个怀疑是 I/O 可能会减慢速度。

编辑：这个问题可能与我的算法有关，因为我没有保留入度为 0 的节点队列。谢谢@luqui。

EDIT2：看来真正的瓶颈是 I/O，感谢@Davislor，我解决了这个问题。

该任务基于此： http: //www.spoj.com/UKCPLAD/problems/TOPOSORT/，我只能使用 Haskell 平台中的库。

{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE LambdaCase #-}
{-# OPTIONS_GHC -O3 #-}

import Control.Monad
import Data.Array.IO
import Data.IORef
import Data.Int
import Control.Exception

type List = []
type Node = Int32
type Edge = (Node, Node)
type Indegree = Int32

main = do
  (numNodes, _) <- readPair <$> getLine
  edges <- map readPair . lines <$> getContents
  topo numNodes edges

-- lower bound
{-# INLINE lb #-}
lb = 1

topo :: Node -> List Edge -> IO ()
topo numNodes edges = do
    result <- newIORef []
    count <- newIORef 0
    indegrees <- newArray (lb,numNodes) 0 :: IO (IOUArray Node Indegree)
    neighbours <- newArray (lb,numNodes) [] :: IO (IOArray Node (List Node))
    forM_ edges $ \(from,to) -> do
      update indegrees to (+1)
      update neighbours from (to:)
    let work = forever $ do
          z <- getNext indegrees
          modifyIORef' result (z:)
          modifyIORef' count (+1)
          ns <- readArray neighbours z
          forM_ ns $ \n -> update indegrees n pred
    work `catch`
      \(_ :: SomeException) -> do
        count <- readIORef count
        if numNodes == count
          then (mapM_ (\n -> putStr (show n ++ " ")) . reverse) =<< readIORef result
          else putStrLn "Sandro fails."


{-# INLINE update #-}
update a i f = do
  x <- readArray a i
  writeArray a i (f x)

{-# INLINE getNext #-}
getNext indegrees = getNext' indegrees =<< getBounds indegrees

{-# INLINE getNext' #-}
getNext' indegrees (lb,ub) = readArray indegrees lb >>= \case
    0 -> writeArray indegrees lb (-1) >> return lb
    _ -> getNext' indegrees (lb+1,ub)

readPair :: String -> (Node,Node)
{-# INLINE readPair #-}
readPair = toPair . map read . words
  where toPair [x,y] = (x,y)
        toPair _ = error "Only two entries per line allowed"

输出示例

$ ./topo
8 9
1 4
1 2
4 2
4 3
3 2
5 2
3 5
8 2
8 6
^D
1 4 3 5 7 8 2 6

Answer 1

如果您已经\xe2\x80\x99t，请通过使用命令行选项编译然后执行来分析您的程序。这将生成一个配置文件-prof -fprof-auto+RTS -p*.prof，告诉您程序将所有时间都花在哪些函数上。但是，我可以立即看到最浪费时间的地方。你的直觉是对的：它\xe2\x80\x99是I/O。

\n\n

做了很多次之后，我可以向您保证，您\xe2\x80\x99 会发现它\xe2\x80\x99 的绝大多数时间都花在了 I/O 上。为了加速你的程序，你应该做的第一件事就是重写它以使用快速 I/O。当您使用正确的数据结构时，Haskell 是一种快速语言。 Prelude 中的默认 I/O 库使用带有惰性求值 thunk 的单链表，其中每个节点保存一个 Unicode 字符。这在 C 语言中也会很慢！

\n\n

Data.ByteString.Lazy.Char8当输入为 ASCII 并Data.ByteString.Builder生成输出时， I\xe2\x80\x99 获得了最佳结果。（另一种方法是Data.Text。）这会为您提供输入时严格字符缓冲区的延迟评估列表（因此交互式输入和输出仍然有效），并在输出时填充单个缓冲区。

\n\n

在您\xe2\x80\x99 编写了具有快速 I/O 的程序骨架后，下一步是查看您的算法，尤其是您的数据结构。使用分析来查看时间都花在哪里了。但我\xe2\x80\x99d 建议你使用函数式算法，而不是尝试使用 Haskell 编写命令式程序do。

\n\n

在 Haskell 中，我几乎总是用更函数式的风格来处理这样的问题：特别是，我的main函数几乎总是类似于：

\n\n

import qualified Data.ByteString.Lazy.Char8 as B8\n\nmain :: IO()\nmain = B8.interact ( output . compute . input )\n

\n\n

这使得除了对纯函数的调用之外的所有内容都被隔离interact，并且隔离了解析代码和格式化代码，因此compute中间的部分可以独立于它。

\n\n

由于这是一项作业，并且您想自己解决问题，因此我\xe2\x80\x99不会为您重构程序，但这里\xe2\x80\x99是我在另一个论坛上回答问题时写的一个例子执行计数排序。它应该适合作为其他类型问题的骨架。

\n\n

import Data.Array.IArray (accumArray, assocs)\nimport Data.Array.Unboxed (UArray)\nimport Data.ByteString.Builder (Builder, char7, intDec, toLazyByteString)\nimport qualified Data.ByteString.Lazy.Char8 as B8\nimport Data.Monoid ((<>))\n\nmain :: IO()\nmain = B8.interact ( output . compute . input ) where\n  input :: B8.ByteString -> [Int]\n  input = map perLine . tail . B8.lines where\n    perLine = decode . B8.readInt\n\n    decode (Just (x, _)) = x\n    decode Nothing = error "Invalid input: expected integer."\n\n  compute :: [Int] -> [Int]\n  compute = concatMap expand . assocs . countingSort . map encode where\n    encode i = (i, 1)\n\n    countingSort :: [(Int, Int)] -> UArray Int Int\n    countingSort = accumArray (+) 0 (lower, upper)\n\n    lower = 0\n    upper = 1000000\n\n    expand (i,c) = replicate c i\n\n  output :: [Int] -> B8.ByteString\n  output = toLazyByteString . foldMap perCase where\n    perCase :: Int -> Builder\n    perCase x = intDec x <> char7 \'\\n\'\n

\n\n

目前，这个版本的运行时间还不到其他人解决相同问题的 Haskell 解决方案的一半时间，对于我使用它来解决的实际竞赛问题也是如此，并且方法概括。

\n\n

因此，我建议首先将 I/O 更改为与此类似，然后进行分析，如果 \xe2\x80\x99 没有产生足够的差异，则返回分析输出。这也可能是一个很好的代码审查问题。

\n