在 Scala 中映射并行序列时的异构运行时间答案

【问题标题】：Heterogeneous running times when mapping over parallel sequence in Scala在 Scala 中映射并行序列时的异构运行时间
【发布时间】：2017-10-24 10:44:17
【问题描述】：

在并行序列上映射时，Scala 似乎为每个计划的计算元素分配了一个固定的 CPU 内核，然后防止这个初始分配发生变化。

List(a, b, c, d, e, f, g, h, i, j, k, l).par.map(someComputation)

// (Simplified) initial assignment for a 2-cores machine:
// Core 1: a, b, c, d, e, f
// Core 2: g, h, i, j, k, l

问题是，如果 (a .. f) 非常快，而 (g .. l) 每个需要 1 小时，我们最终会导致 Core 1 闲置 6 小时，而 Core 2 则试图完成其工作。

有没有办法进行并行计算，如果 Core X 空闲并且还有未启动的作业，则向 Core X 提供这些作业之一？

编辑

要重现，请启动 Scala REPL 并输入：

scala> :paste

def compute(x: Int) = {
  if (x > 10)
    (0 to 10e9.toInt).foreach(n => n + 1) // loads 100% of a core for 4-5 seconds on a typical iMac
  println(x)
  x * 2
}
// hit Ctrl+D
compute: (x: Int)Int

scala> def foo = (0 to 20).toList.par.map(compute)
foo: scala.collection.parallel.immutable.ParSeq[Int]

scala> foo

然后看看会发生什么。一开始，所有核心都处于 100%，随着您的进步，您可以清楚地看到每个进程在开始之前就分配了自己的核心。即使另一个内核当前空闲，它也不会更改内核。到最后，只有一两个核心处理队列中的所有剩余作业，而其他 3 个或更多核心闲置的情况并不罕见（取决于初始分配）。

【问题讨论】：

标签： scala parallel-processing

【解决方案1】：

最简单的方法是使用Futures：

val tasks = List("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l")
val resultFutures: List[Future[Something]] = tasks.map(t => Future {
  someComputation(t)
}(ExecutionContext.Implicits.global))

// wait for result somehow
resultFutures.foreach(f => Await.result(f))

您还可以查看ForkJoinTask API。

【讨论】：

【解决方案2】：

对于par，默认使用defaultTaskSupport，由ExecutionContext.Implicits.global实现

而这个ExecutionContextExecutor默认由ForkJoinPool实现，它是一个work-stealing线程池，空闲线程会从繁忙线程中窃取FutureTasks，所以它应该使用操作系统所有处理器。

对于您的示例，我认为您可以尝试覆盖 tasksupport 来验证并尝试不使用 repl 来测试它（也许这是一个 repl 错误 :)）。喜欢：

  val par: ParSeq[Int] = (0 to 20).toList.par
  //par.tasksupport = new ForkJoinTaskSupport(new java.util.concurrent.ForkJoinPool(1)) // test run with one processor
  //default it will use all processors of OS
  par.tasksupport = new ForkJoinTaskSupport(new java.util.concurrent.ForkJoinPool())
  par.map(compute)

【讨论】：