Paralelismo en el algoritmo de dividir y conquistar

Question

Tengo problemas para hacer que mi código se ejecute en paralelo. Es un generador 3D Delaunay que utiliza un algoritmo de conquista de & dividido denominado DeWall.

La función principal es:

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

llama a la función "recursividad" que podría llamar a la función DeWall espalda. Y es aquí donde aparece la oportunidad de parallización. El siguiente código muestra la solución secuencial.

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 

mónadas estatales y IO se utilizan para tubo del estado y para generar UID para cada tetraedro encontrado usando MVar de. Mi primer intento fue agregar un forkIO pero no funciona. Da un resultado incorrecto debido a una falta de control durante la parte de fusión que no espera a que terminen ambos subprocesos. No sé cómo hacerlo esperar por ellos.

  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) 

lo tanto, mi siguiente intento fue usar una mejor estrategia paralela "par" y "PSEQ" que da el resultado correcto, pero no la ejecución en paralelo de acuerdo con 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) 

¿Qué estoy haciendo mal?

ACTUALIZACIÓN: De alguna manera este problema parece estar relacionado con la presencia de IO mónadas. En otra versión (anterior) sin mónada IO, solo Mónada de estado, la ejecución paralela se ejecuta con 'par' y 'pseq'. El GHC -sstderr da 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 

¿Alguien en la nube explica eso?

Answer 1

La manera más fácil de hacer este trabajo sería utilizar algo como:

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) 

Esto funciona, pero hay algunos gastos indirectos innecesarios. Una mejor solución es:

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) 

O sea posible esto si los resultados no están siendo evaluados en la que desea que sean:

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) 

(éstas son las respuestas que di al cartel de #haskell que pensé que sería útil aquí también)

Editar: eliminado innecesaria evaluar.

Answer 2

El uso de par y pseq debe ocurrir en la "ruta de ejecución", es decir, no dentro de un local let. Prueba esto (modificar su último fragmento)

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

Una evaluación case fuerzas de su argumento a débil cabeza de forma normal (WHNF) antes de continuar en una de sus ramas. WHNF significa que el argumento se evalúa hasta que el constructor más externo sea visible. Los campos aún pueden estar sin evaluar.

Para forzar la evaluación completa de un argumento, use deepseq. Tenga cuidado con eso, sin embargo, porque deepseq a veces puede hacer las cosas más lentas haciendo demasiado trabajo.

Un enfoque más ligero para añadir rigidez es hacer campos estricta:

data Foo = Foo !Int String 

Ahora, cada vez que un valor de tipo Foo se evalúa a WHNF, por lo que es su primer argumento (pero no el segundo).

Answer 3

Si desea continuar con los subprocesos explícitos, en lugar de pseq, como ha indicado, necesita alguna forma de esperar a que finalicen los subprocesos de trabajo. Ese es un gran caso de uso para un semáforo de cantidad. Después de dividir el trabajo que se va a realizar, haga que cada hilo trabajador, al finalizar, le indique al semáforo cuánto trabajo ha realizado.

Luego espere a que se completen todas las unidades de trabajo.

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

Editar: algunos pseudocódigo para ayudar a explicar la noción

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.