分治算法的并行性

Question

我在让我的代码并行运行时遇到了问题。它是一个 3D Delaunay 生成器，使用名为 DeWall 的分治算法。

主要功能是：

deWall::[SimplexPointer] -> SetSimplexFace -> Box -> StateT DeWallSets IO ([Simplex], [Edge])
deWall p afl box = do
   ...
   ...
   get >>= recursion box1 box2 p1 p2 sigma edges
   ...
   ...

它调用可能回调 dewall 函数的“递归”函数。正是在这里出现了并行化机会。以下代码显示了顺序解决方案。

recursion::Box -> Box -> [SimplexPointer] -> [SimplexPointer] -> [Simplex] -> [Edge] -> DeWallSets -> StateT DeWallSets IO ([Simplex], [Edge])    
recursion box1 box2 p1 p2 sigma edges deWallSet
        | null afl1 && null afl2 = return (sigma, edges)
        | (null) afl1 = do
            (s, e) <- deWall p2 afl2 box2
            return (s ++ sigma, e ++ edges)
        | (null) afl2 = do
            (s,e) <- deWall p1 afl1 box1
            return (s ++ sigma, e ++ edges)
        | otherwise   = do
            x <- get
            liftIO $ do
                (s1, e1) <- evalStateT (deWall p1 afl1 box1) x
                (s2, e2) <- evalStateT (deWall p2 afl2 box2) x
                return (s1 ++ s2 ++ sigma, e1 ++ e2 ++ edges)

        where   afl1 = aflBox1 deWallSet
                afl2 = aflBox2 deWallSet

State 和 IO monad 用于传递状态并为使用 MVar 找到的每个四面体生成 UID。我的第一次尝试是添加一个 forkIO，但它不起作用。由于在不等待两个线程完成的合并部分期间缺乏控制，它给出了错误的输出。我不知道如何让它等待他们。

            liftIO $ do
                let 
                    s1 = evalStateT (deWall p1 afl1 box1) x
                    s2 = evalStateT (deWall p2 afl2 box2) x
                    concatThread var (a1, b1) = takeMVar var >>= \(a2, b2) -> putMVar var (a1 ++ a2, b1 ++ b2)
                mv <- newMVar ([],[])
                forkIO (s1 >>= concatThread mv)
                forkIO (s2 >>= concatThread mv)
                takeMVar mv >>= \(s, e) -> return (s ++ sigma, e ++ edges)

所以，我的下一个尝试是使用更好的并行策略“par”和“pseq”，它给出了正确的结果，但根据 threadScope 没有并行执行。

        liftIO $ do
            let
                s1 = evalStateT (deWall p1 afl1 box1) x
                s2 = evalStateT (deWall p2 afl2 box2) x
                conc = liftM2 (\(a1, b1) (a2, b2) -> (a1 ++ a2, b1 ++ b2))
            (stotal, etotal) = s1 `par` (s2 `pseq` (s1 `conc` s2))
            return (stotal ++ sigma, etotal ++ edges)

我做错了什么？

更新：不知何故，这个问题似乎与 IO monad 的存在有关。在没有 IO monad、只有 State monad 的另一个（旧）版本中，并行执行使用 'par' 和 'pseq' 运行。 GHC -sstderr 给出SPARKS: 1160 (69 converted, 1069 pruned)。

recursion::Box -> Box -> [SimplexPointer] -> [SimplexPointer] -> [Simplex] -> [Edge] -> DeWallSets -> State DeWallSets ([Simplex], [Edge])  
recursion p1 p2 sigma deWallSet
     | null afl1 && null afl2 = return sigma
     | (null) afl1 = do
         s <- deWall p2 afl2 box2
         return (s ++ sigma)
     | (null) afl2 = do
         s <- deWall p1 afl1 box1
         return (s ++ sigma)
     | otherwise   = do
                     x <- get
                     let s1 = evalState (deWall p1 afl1 box1) x
                     let s2 = evalState (deWall p2 afl2 box2) x
                     return $ s1 `par` (s2 `pseq` (s1 ++ s2 ++ sigma))
     where   afl1 = aflBox1 deWallSet
             afl2 = aflBox2 deWallSet

云有人解释一下吗？

Answer 1

par 和 pseq 的使用应该发生在“执行路径”上，即，不在本地 let 内。试试这个（修改你的最后一个 sn-p）

let s1 = ...
    s2 = ...
    conc = ...
case s1 `par` (s2 `pseq` (s1 `conc` s2)) of
  (stotal, etotal) ->
     return (stotal ++ sigma, etotal ++ edges)

case 强制对其参数的评估为弱头范式 (WHNF)，然后继续其分支之一。 WHNF 意味着对参数进行评估，直到最外面的构造函数可见。字段可能仍未评估。

要强制对参数进行全面评估，请使用deepseq。不过要小心，因为deepseq 有时会因为做太多工作而使事情变慢。

增加严格性的一种更轻量级的方法是使字段严格：

data Foo = Foo !Int String

现在，每当 Foo 类型的值被评估为 WHNF 时，它的第一个参数也是如此（但不是第二个）。

Answer 2

完成这项工作的最简单方法是使用以下内容：

liftIO $ do
            let 
                s1 = evalStateT (deWall p1 afl1 box1) x
                s2 = evalStateT (deWall p2 afl2 box2) x
            mv1 <- newMVar ([],[])
            mv2 <- newMVar ([],[])
            forkIO (s1 >>= putMVar mv1)
            forkIO (s2 >>= putMVar mv2)
            (a1,b1) <- takeMVar mv1
            (a2,b2) <- takeMVar mv2
            return (a1++a2++sigma, b1++b2++edges)

这可行，但有一些不必要的开销。更好的解决方案是：

liftIO $ do
            let 
                s1 = evalStateT (deWall p1 afl1 box1) x
                s2 = evalStateT (deWall p2 afl2 box2) x
            mv <- newMVar ([],[])
            forkIO (s2 >>= putMVar mv2)
            (a1,b1) <- s1
            (a2,b2) <- takeMVar mv2
             return (a1++a2++sigma, b1++b2++edges)

或者，如果结果没有在您希望的位置进行评估，则可能是这样：

liftIO $ do
        let 
            s1 = evalStateT (deWall p1 afl1 box1) x
            s2 = evalStateT (deWall p2 afl2 box2) x
        mv <- newMVar ([],[])
        forkIO (s2 >>= evaluate >>= putMVar mv2)
        (a1,b1) <- s1
        (a2,b2) <- takeMVar mv2
         return (a1++a2++sigma, b1++b2++edges)

（这些是我在#haskell 中给发帖人的答案，我认为在这里也很有用）

编辑：删除了不必要的评估。

Answer 3

如果您想坚持使用显式线程而不是 pseq，正如您所指出的，您需要一些方法来等待工作线程结束。这是数量信号量的一个很好的用例。在您划分要完成的工作后，让每个工作线程在终止时向信号量发出信号，告知其已完成的工作量。

然后等待所有工作单元完成。

http://www.haskell.org/ghc/docs/6.8.3/html/libraries/base/Control-Concurrent-QSemN.html

编辑：一些伪代码来帮助解释这个概念

do
 let workchunks :: [(WorkChunk, Size)]
     workchunks = dividework work

  let totalsize = sum $ map snd workchunks

 sem <- newQSem 0

 let forkworkThread (workchunk, size) = do
        executeWorkChunk workchunk
        signalQSem size

 mapM_ forkWorkThread workchunks
 waitQSem totalsize

 -- now all your work is done.