如何在 Haskell 中创建通道列表（或数组），例如 Go

Question

我正在尝试比较 Haskell 和 Go 的一些并发示例，以下代码是 Go 中使用 Goroutines 和通道的简单 map-reduce 示例。下面的 Go 代码计算平方和：

1^2+2^2+3^2....1024^2

为了测试 Go 和 Haskell 的性能，我重复计算 R 次 (10) 的平方和。

package main

import "fmt"

func mapper(in chan int, out chan int) {
    for v := range in {out <- v*v}
}

func reducer(in1, in2 chan int, out chan int) {
    for i1 := range in1 {i2 := <- in2; out <- i1 + i2}
}

func main() {
    const N = 1024  // calculate sum of squares up to N; N must be power of 2
    const R = 10  // number of repetitions to fill the "pipe"

    var r [N*2]chan int
    for i := range r {r[i] = make(chan int)}
    var m [N]chan int
    for i := range m {m[i] = make(chan int)}

    for i := 0; i < N; i++ {go mapper(m[i], r[i + N])}
    for i := 1; i < N; i++ {go reducer(r[i * 2], r[i *2 + 1], r[i])}

    go func () {
        for j := 0; j < R; j++ {
            for i := 0; i < N; i++ {m[i] <- i + 1} 
        }
    } ()

    for j := 0; j < R; j++ {
        <- r[1]
    } 
}

问题是如何在 Haskell 中高效地实现这个 mapreduce 示例。以下 Haskell 代码尝试在 main 函数中计算 10^2 + 7^2。我的问题是如何创建一个通道数组（或列表），例如 Go，然后在 main 函数中将映射器和减速器线程连接在一起。

import Control.Concurrent
data MRchannel = MRchannel !(MVar MRcmd)
data MRcmd = Pass !Int | Add !Int
  deriving (Show)

mapper:: MRchannel -> MRchannel -> IO ()
mapper left_C@(MRchannel left) right_C@(MRchannel right) = do
    v <- takeMVar left
    case v of
        Pass x -> do
            putMVar right (Add (x*x))
            mapper left_C right_C
        otherwise -> do
            putStrLn "Error!"
            return ()

reducer::  MRchannel -> MRchannel -> MRchannel -> IO ()
reducer left_1_C@(MRchannel left_1) left_2_C@(MRchannel left_2) 
right_C@(MRchannel right) = do
    v1 <- takeMVar left_1
    case v1 of
        Add x1 -> do
            v2 <- takeMVar left_2
            case v2 of
                Add x2 -> do 
                    putMVar right (Add (x1+x2))
                    reducer left_1_C left_2_C right_C
                otherwise -> do
                    putStrLn "Error!"
                    return ()
        otherwise -> do
            putStrLn "Error!"
            return ()

main = do
m1_l <- newEmptyMVar
m2_l <- newEmptyMVar
r1_l1 <- newEmptyMVar
r1_l2 <- newEmptyMVar
r1_r <- newEmptyMVar
    let m1_input = MRchannel m1_l
    let m2_input = MRchannel m2_l
    let r1_input1 = MRchannel r1_l1
    let r1_input2 = MRchannel r1_l2
    let r1_output = MRchannel r1_r
    forkIO $ mapper m1_input r1_input1
    forkIO $ mapper m2_input r1_input2
    forkIO $ reducer r1_input1 r1_input2 r1_output

    putMVar m1_l (Pass 10)
    putMVar m2_l (Pass 7)

    y <- takeMVar r1_r
    case y of 
        Add kvalue  -> do
            putStrLn $ show kvalue
        otherwise -> do
            putStrLn "Error"
            return () 

Answer 1

我已经在 Haskell 中为这个 MapReduce 示例制定了一个解决方案。请指出此代码是否有任何部分可以优化。非常感谢您提供的任何帮助。

import Control.Concurrent

mapper:: (MVar Int) -> (MVar Int) -> IO ()
mapper left right = do
    v <- takeMVar left
    --putMVar right (v*v)
    putMVar right $! v*v
    mapper left right

reducer::  (MVar Int) -> (MVar Int) -> (MVar Int) -> IO ()
reducer left_1 left_2 right = do
    v1 <- takeMVar left_1
    v2 <- takeMVar left_2
    --putMVar right  (v1+v2)
    putMVar right $! v1+v2
    reducer left_1 left_2 right        

repeats:: Int -> [MVar Int] -> [MVar Int] ->IO ()
repeats 0 m r = do        
    --mapM_ (\ x -> putMVar (m!!x) (x+1) ) [0..1023]
    mapM_ (\ (e,x) -> putMVar e (x+1)) $ zip m [0..]
    result <- takeMVar (r!!1)
    return ()
    --putStrLn (show result)

repeats n m r = do
    --mapM_ (\ x -> putMVar (m!!x) (x+1) ) [0..1023]
    mapM_ (\ (e,x) -> putMVar e (x+1)) $ zip m [0..]
    result <- takeMVar (r!!1)
    --putStrLn (show result)
    repeats (n-1) m r

main = do
    --m <- sequence $ take 1024 $ repeat newEmptyMVar
    m <- replicateM 1024 newEmptyMVar
    --r <- sequence $ take 2048 $ repeat newEmptyMVar
    r <- replicateM 2048 newEmptyMVar

    mapM_ (\ x -> forkIO (mapper (m!!x) (r!!(1024+x) ))) [0..1023]
    mapM_ (\ x -> forkIO (reducer (r!!(x*2)) (r!!(x*2+1))  (r!!x))) [0..1023]
    repeats 1024 m r

Answer 2

这里有一个替代方案。在我看来，它并不是非常优雅，但比原始代码要简单一些。

主要变化：

• 我们没有两个MVars 列表并使用慢速!!
• 我们使用forkAll 生成映射器/缩减器的整个二叉树（它的参数是深度：2^10 = 1024）
• forkAll 仍然不是最理想的，因为它使用的是速度较慢的 ++，尽管
• 主要变化：在执行putMVar 时，我们强制使用$! 的值，这样我们就不会在MVar 中存储未计算的表达式。这是一个重大的性能提升。
• 我们只打印它们的总和，而不是打印所有结果。否则，文本 IO 会降低基准测试速度。 （请注意，原来的repeats 会循环n+1，所以我必须使用10241 而不是下面的10240 来匹配结果。）

在这里，新代码的速度提高了 54%。

import Control.Concurrent
import Control.Monad

mapper :: MVar Int -> MVar Int -> IO ()
mapper left right = forever $ do
    v <- takeMVar left
    putMVar right $! v*v

reducer :: MVar Int -> MVar Int -> MVar Int -> IO ()
reducer left_1 left_2 right = forever $ do
    v1 <- takeMVar left_1
    v2 <- takeMVar left_2
    putMVar right $! v1+v2

forkAll :: Int -> MVar Int -> IO [MVar Int]
forkAll 0 res = do
  v <- newEmptyMVar
  _ <- forkIO $ mapper v res
  return [v]
forkAll depth res = do
  v1 <- newEmptyMVar
  r1 <- forkAll (depth - 1) v1
  v2 <- newEmptyMVar
  r2 <- forkAll (depth - 1) v2
  _ <- forkIO $ reducer v1 v2 res
  return (r1++r2)

main :: IO ()
main = do
  v <- newEmptyMVar
  res <- forkAll 10 v

  nums <- replicateM 10241 $ do
     mapM_ (\ (r,i) -> putMVar r (i+1)) $ zip res [0..]
     takeMVar v
  print $ sum nums

Answer 3

这将返回与您的 Haskell 代码相同的内容；因为它是纯计算，所以它使用parallel 包。

import Data.Vector ((!), fromList, Vector)
import Control.Parallel (par, pseq)

reducer :: Int -> Int -> Int
reducer a b = a `par` b `pseq` a + b

result :: Vector Int -> Int
result src =
  let mr = fromList $
               map (\i arr -> reducer (arr ! (2 * i)) (arr ! (2 * i + 1))) [0..1023]
            ++ map (\i _ -> src ! i * src ! i) [0..1023]
            :: Vector (Vector Int -> Int)
      -- See https://github.com/quchen/articles/blob/master/loeb-moeb.md
      res = fmap ($ res) mr
  in res ! 1

main :: IO ()
main = do
  let src = fromList [1..1024]
  print (result src)