如何优化并行排序以提高时间性能？

Posted

技术标签:

【中文标题】如何优化并行排序以提高时间性能？【英文标题】：How can I optimize parallel sorting to improve temporal performance? 【发布时间】：2019-04-23 12:00:30 【问题描述】：

我有一个算法可以对给定长度的列表进行并行排序：

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

【问题讨论】：

密切相关：***.com/q/19752983/745903 force 没有做你认为它正在做的事情：它只强制它给出的列表的脊椎，而不是内容。对于随机数生成，强制内容是实际工作。此外，force result `seq` result 并没有做您希望它做的事情。在开始计时之前尝试将evaluate (last input) 放入main，您会立即看到差异。我还建议使用mkStdGen 0（或其他一些常量）而不是getStdGen，以使您的时间更加一致。 （不过，这些都不能解决您的实际问题。） 至于您的实际问题：您可能创建了太多火花，而火花跟踪开销主导了运行时。您可以尝试一个更简单的策略：将列表拆分为一些小块，例如 60，或者可能是传递给 -N 的数字，并行排序每个块，然后合并结果。我自己没有测试过，所以我不会写这个作为答案，因为我不能确定它是正确的，但我敢打赌它会表现得更好。 并行化快速排序的一个问题是它不能建立平衡的子列表。您可能会通过并行化合并排序获得更好的结果，它会在输入时保持平衡，在输出时进行排序。 @ReinHenrichs 是的，快速排序不适合并行化，但即使平衡不好，它仍然应该有可能获得 一些 性能提升，尤其是在完全随机数的情况下列表。 【参考方案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

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

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

-- | 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 []

【讨论】：

我们能否有一些基准来看看这是否真的导致并行化的性能提高？ @DanRobertson，我做了一些非正式的基准测试，它似乎确实有效（pSort 2 比pSort 1 在[1,-1,2,-2,...,2*10^6] 之类的列表上快很多），但我只有不适合那种东西的中型笔记本电脑。 Threadscope 似乎暗示了一个相当平衡的负载，但我以前从未使用过它，所以我不能保证我正确地解释了结果。 @DanRobertson，它似乎可以很好地扩展到至少 4 个带有伪随机输入的内核。我怀疑还有更多的调整要做（例如，使用通过monad-par 提供的更精细的并行控制可能比依赖内置火花池更好），但一般方法似乎很好。跨度>

以上是关于如何优化并行排序以提高时间性能？的主要内容，如果未能解决你的问题，请参考以下文章

优化 SQL：如何重写此查询以提高性能？ （使用子查询，摆脱 GROUP BY？）

.NET中并行开发优化

rt下降40%？程序并行优化六步法

如何优化QTableView的性能

MongoDB如何优化查询性能？

怎么优化电脑性能