如何优化并行排序以提高时态性能?

Question

我有一个算法来并行排序给定长度的列表：

import Control.Parallel (par, pseq)
import Data.Time.Clock (diffUTCTime, getCurrentTime)
import System.Environment (getArgs)
import System.Random (StdGen, getStdGen, randoms)

parSort :: (Ord a) => [a] -> [a]
parSort (x:xs)    = force greater `par` (force lesser `pseq`
(lesser ++ x:greater))
where lesser  = parSort [y | y <- xs, y <  x]
greater = parSort [y | y <- xs, y >= x]
parSort _         = []
sort :: (Ord a) => [a] -> [a]
sort (x:xs) = lesser ++ x:greater
where lesser  = sort [y | y <- xs, y <  x]
greater = sort [y | y <- xs, y >= x]
sort _ = []
parSort2 :: (Ord a) => Int -> [a] -> [a]
parSort2 d list@(x:xs)
| d <= 0     = sort list
| otherwise = force greater `par` (force lesser `pseq`
(lesser ++ x:greater))
where lesser      = parSort2 d' [y | y <- xs, y <  x]
greater     = parSort2 d' [y | y <- xs, y >= x]
d' = d - 1
parSort2 _ _              = []
force :: [a] -> ()
force xs = go xs `pseq` ()
where go (_:xs) = go xs
go [] = 1

randomInts :: Int -> StdGen -> [Int]
randomInts k g = let result = take k (randoms g)
in force result `seq` result
testFunction = parSort
main = do
args <- getArgs
let count | null args = 500000
| otherwise = read (head args)
input <- randomInts count `fmap` getStdGen
start <- getCurrentTime
let sorted = testFunction input
putStrLn $ "Sort list N = " ++ show (length sorted)
end <- getCurrentTime
putStrLn $ show (end `diffUTCTime` start) 

我想有时间对少于 1 个内核的 2、3 和 4 个处理器内核执行并行排序。 目前，这个结果我无法达到。 以下是我的计划启动：

1. SortList +RTS -N1 -RTS 10000000
time = 41.2 s
2.SortList +RTS -N3 -RTS 10000000
time = 39.55 s
3.SortList +RTS -N4 -RTS 10000000
time = 54.2 s

我能做什么？

更新 1：

testFunction = parSort2 60

Answer 1

这里有一个你可以玩的想法，使用Data.Map。为了简单和性能，我假设元素类型具有替代性，因此我们可以计算出现次数而不是存储元素列表。我相信您可以使用一些花哨的数组算法获得更好的结果，但这很简单并且(本质上)功能齐全。

在编写并行算法时，我们希望尽量减少必须按顺序完成的工作量。在对列表进行排序时，有一件事我们真的无法避免按顺序进行：将列表拆分为多个线程进行处理。我们希望以尽可能少的努力完成这项工作，然后从那时起尝试主要并行工作。

让我们从一个简单的顺序算法开始。

{-# language BangPatterns, TupleSections #-}
import qualified Data.Map.Strict as M
import Data.Map (Map)
import Data.List
import Control.Parallel.Strategies
type Bag a = Map a Int
ssort :: Ord a => [a] -> [a]
ssort xs =
let m = M.fromListWith (+) $ (,1) <$> xs
in concat [replicate c x | (x,c) <- M.toList m]

我们如何并行化呢？首先，让我们将列表分解为多个部分。有多种方法可以做到这一点，但没有一种很好。假设功能数量很少，我认为让每个功能自己走过列表是合理的。随意尝试其他方法。

-- | Every Nth element, including the first
everyNth :: Int -> [a] -> [a]
everyNth n | n <= 0 = error "What you doing?"
everyNth n = go 0 where
go !_ [] = []
go 0 (x : xs) = x : go (n - 1) xs
go k (_ : xs) = go (k - 1) xs
-- | Divide up a list into N pieces fairly. Walking each list in the
-- result will walk the original list.
splatter :: Int -> [a] -> [[a]]
splatter n = map (everyNth n) . take n . tails

现在我们有了列表，我们激发线程将它们转换为袋子。

parMakeBags :: Ord a => [[a]] -> Eval [Bag a]
parMakeBags xs = 
traverse (rpar . M.fromListWith (+)) $ map (,1) <$> xs

现在我们可以反复合并成对的袋子，直到我们只有一个。

parMergeBags_ :: Ord a => [Bag a] -> Eval (Bag a)
parMergeBags_ [] = pure M.empty
parMergeBags_ [t] = pure t
parMergeBags_ q = parMergeBags_ =<< go q where
go [] = pure []
go [t] = pure [t]
go (t1:t2:ts) = (:) <$> rpar (M.unionWith (+) t1 t2) <*> go ts

但。。。有问题。在每一轮合并中，我们只使用前一轮一半的功能，并且只使用一种功能执行最终合并。哎哟！要解决此问题，我们需要并行化unionWith。幸运的是，这很容易！

import Data.Map.Internal (Map (..), splitLookup, link)
parUnionWith
:: Ord k
=> (v -> v -> v)
-> Int -- Number of threads to spark
-> Map k v
-> Map k v
-> Eval (Map k v)
parUnionWith f n t1 t2 | n <= 1 = rseq $ M.unionWith f t1 t2
parUnionWith _ !_ Tip t2 = rseq t2
parUnionWith _ !_ t1 Tip = rseq t1
parUnionWith f n (Bin _ k1 x1 l1 r1) t2 = case splitLookup k1 t2 of
(l2, mb, r2) -> do
l1l2 <- parEval $ parUnionWith f (n `quot` 2) l1 l2
r1r2 <- parUnionWith f (n `quot` 2) r1 r2
case mb of
Nothing -> rseq $ link k1 x1 l1l2 r1r2
Just x2 -> rseq $ link k1 fx1x2 l1l2 r1r2
where !fx1x2 = f x1 x2

现在我们可以完全并行化袋合并：

-- Uses the given number of capabilities per merge, initially,
-- doubling for each round.
parMergeBags :: Ord a => Int -> [Bag a] -> Eval (Bag a)
parMergeBags !_ [] = pure M.empty
parMergeBags !_ [t] = pure t
parMergeBags n q = parMergeBags (n * 2) =<< go q where
go [] = pure []
go [t] = pure [t]
go (t1:t2:ts) = (:) <$> parEval (parUnionWith (+) n t1 t2) <*> go ts

然后，我们可以像这样实现并行合并：

parMerge :: Ord a => [[a]] -> Eval [a]
parMerge xs = do
bags <- parMakeBags xs
-- Why 2 and not one? We only have half as many
-- pairs as we have lists (capabilities we want to use)
-- so we double up.
m <- parMergeBags 2 bags
pure $ concat [replicate c x | (x,c) <- M.toList m]

把碎片放在一起，

parSort :: Ord a => Int -> [a] -> Eval [a]
parSort n = parMerge . splatter n
pSort :: Ord a => Int -> [a] -> [a]
pSort n = runEval . parMerge . splatter n

---

只剩下一个可以并行化的连续部分：将最终包转换为列表。值得并行化吗？我很确定在实践中并非如此。但无论如何，让我们这样做，只是为了好玩！为了避免相当大的额外复杂性，我将假设没有大量相等的元素;结果中的重复元素将导致结果列表中保留一些工作(麻烦)。

我们需要一个基本的部分列表脊柱强制器：

-- | Force the first n conses of a list
walkList :: Int -> [a] -> ()
walkList n _ | n <= 0 = ()
walkList _ [] = ()
walkList n (_:xs) = walkList (n - 1) xs

现在我们可以将包转换为并行块的列表，而无需支付连接费用：

-- | Use up to the given number of threads to convert a bag
-- to a list, appending the final list argument.
parToListPlus :: Int -> Bag k -> [k] -> Eval [k]
parToListPlus n m lst | n <= 1 = do
rseq (walkList (M.size m) res)
pure res
-- Note: the concat and ++ should fuse away when compiling with
-- optimization.
where res = concat [replicate c x | (x,c) <- M.toList m] ++ lst
parToListPlus _ Tip lst = pure lst
parToListPlus n (Bin _ x c l r) lst = do
r' <- parEval $ parToListPlus (n `quot` 2) r lst
res <- parToListPlus (n `quot` 2) l $ replicate c x ++ r'
rseq r' -- make sure the right side is finished
pure res

然后我们相应地修改合并：

parMerge :: Ord a => Int -> [[a]] -> Eval [a]
parMerge n xs = do
bags <- parMakeBags xs
m <- parMergeBags 2 bags
parToListPlus n m []