使用HXT的并行Haskell

Question

我正在尝试提高解析XML的程序的性能。该程序可以解析多个XML文件，所以我认为我可以让它并行运行，但我的所有尝试都导致了较低的性能！

对于XML解析，我使用HXT。

我有一个如下定义的run函数：

run printTasks xs = pExec xs >>= return . concat >>= doPrint printTasks 1

'pExec‘被赋予一个文件名列表，并被定义为：

pExec xs = do
   ex <- mapM exec xs
   as <- ex `usingIO` parList rdeepseq
   return as

其中'exec‘定义为：

exec = runX . process

线程作用域显示只有一个线程在使用(直到最后)。

有没有人能解释一下为什么我在并行化这段代码时失败得如此惨烈？

如果有帮助的话：

exec :: FilePath -> [CV_scene]
pExec :: [FilePath] -> IO [[CV_scene]]

data CV_scene = Scene [CV_layer] Time deriving (Show)
data CV_layer = Layer [DirtyRects] SourceCrop deriving (Show)
data Rect     = Rect Int Int Int Int deriving (Show)-- Left Top Width Height

instance NFData CV_scene where
  rnf = foldScene reduceScene
    where reduceScene l t = rnf (seq t l)

instance NFData CV_layer where
  rnf = foldLayer reduceLayer
    where reduceLayer d s = rnf (seq s d)

instance NFData Rect where
  rnf = foldRect reduceRect
    where reduceRect l t w h = rnf [l,t,w,h]

type SourceCrop = Rect
type DirtyRect  = Rect
type Time       = Int64

提前感谢您的帮助！

Answer 1

首先，看起来您错误地标记了exec的签名，这应该是：

exec :: FilePath -> IO [CV_scene]

现在是重要的部分。我已经内联地评论了我认为您正在进行的事情。

pExec xs = do
   -- A. Parse the file found at each location via exec.
   ex <- mapM exec xs
   -- B. Force the lazy parsing in parallel.
   as <- ex `usingIO` parList rdeepseq
   return as

请注意，A行不会以并行方式出现，你可能认为这是可以的，因为它只会设置在B中并行强制的解析块。这是一个合理的假设，也是对懒惰的巧妙使用，但结果让我对此产生了疑问。

我怀疑exec的实现甚至在达到B行之前就强制执行了大部分解析，所以深度seq不会做太多事情。这与我的解析经验非常吻合，分析支持这一解释。

如果不能测试您的代码，我只能提出以下建议。首先尝试将文件的解析与IO分离，并将解析放在并行执行策略中。在这种情况下，A和B行类似于：

ex <- mapM readFile xs
as <- ex `usingIO` parList (rdeepseq . exec')

对于exec'，从磁盘读取文件后的exec部分。

    exec' :: FilePath -> [CVScene]

此外，在进行此更改后，您甚至可能不需要rdeepSeq。

作为替代方案，您可以使用软件事务内存并行执行IO和解析。STM方法通常用于单独的IO线程，这些线程的行为更像是服务，而不是纯粹的计算。但是如果由于某些原因你不能使用基于策略的方法，这可能是值得一试的。

import Control.Concurrent.STM.TChan --(from stm package)
import Control.Concurrent(forkIO)

pExec'' :: [FilePath] -> IO [[CVSene]]
pExec'' xs = do
  -- A. create [(Filename,TChan [CVScene])]
  tcx <- mapM (\x -> (x,) <$> newTChanIO) xs
  -- B. do the reading/parsing in separate threads
  mapM_ (forkIO . exec'') tcx
  -- C. Collect the results
  cvs <- mapM (atomically . readTChan . snd) tcx

exec'' :: [(FilePath,TChan [CVScene])] -> IO ()
exec'' (x,tch) = do
  --D. The original exec function
  cv <- exec x
  --E. Put on the channel fifo buffer
  atomically $ writeTChan tch cv

祝好运!