线程池中的并行递归任务

Question

我的确切示例是并行化这种树聚合，其中信息从叶子流向根：

aggregate :: ([a] -> a) -> Tree a -> Tree a
aggregate _ (Node x []) = Node x []
aggregate aggregator (Node _ children) =
  let agChildren = map (aggregate aggregator) children in
  Node (aggregator $ map (\(Node y _) -> y) agChildren) agChildren

我希望聚合器函数的每个应用程序都在不同的线程上进行处理。所以我想改变上面的代码，让它生成一个依赖任务树并将它们提供给线程池。

我不希望节点上的线程受到影响，等待子线程完成。相反，这个等待线程应该去计算树中其他可用的子节点。同时为每个节点运行一个线程也太慢了。我的树可以有数百个节点，而我的机器只有 8 个核心：他们会花时间调度而不是计算。我需要一个仅在其他任务完成时才使用任务的线程池。

正如 ErikR 下面提到的，parMap 似乎正是这样做的。我尝试了它并用strat 64 +RTS -N2 执行它，以获得完全相同的计算时间。这是代码（为了测试性能而进行愚蠢的计算），你明白为什么时间没有改变吗？

slowAggregate :: [Int] -> Int
slowAggregate l = let s = sum l in
  sum [a + b + c | a <- [0..s], b <- [0..s], c <- [0..s] ]

bigTree :: Tree Int
bigTree = Node 0 $ map (\x -> Node x []) [71..78]

aggregate :: NFData a => ([a] -> a) -> Tree a -> Tree a
aggregate _ (Node x []) = Node x []
aggregate aggregator (Node _ children) =
  let agChildren = parMap rdeepseq (aggregate aggregator) children in 
  Node (aggregator $ map (\(Node y _) -> y) agChildren) agChildren

main = timeIt $ let (Node y _) = aggregate slowAggregate bigTree in print y

Answer 1

在您的 slowAggregate 示例中，函数 slowAggregate 仅被调用一次 - 用于顶部节点。

它不会被任何子节点调用，因为这些节点本身没有任何子节点。这些节点由aggregate 的第一个保护子句处理：

aggregate _ (Node x []) = Node x []

此外，当我使用 +RTS -s -N2 运行程序时，所有 8 个创建的火花都失败了：

  SPARKS: 8 (0 converted, 0 overflowed, 0 dud, 0 GC'd, 8 fizzled)

这些火花对应于列表 [71..78] 中的 8 个元素。所有这些都失败了，因为没有一个人叫slowAggregate。

Answer 2

并行运行时已经为 spark 管理了一个线程池。 来自this SO answer：

火花不是线程。 forkIO 引入了 Haskell 线程（映射到更少的真实操作系统线程）。 Sparks 在每个线程的工作队列中创建条目，如果线程空闲，它们将从这些条目中执行任务。

所以我会首先尝试使用parMap strat，看看它是否适合你。

事实上，如果你的树是 Traversable，我会考虑使用 parTraversable：

parTraversable :: Traversable t => Strategy a -> Strategy (t a)