Scala 并行集合运行时令人费解的问题

Question

编辑：我的样本量太小。当我在 8 个 CPU 上针对真实数据运行它时，我发现速度提高了 7.2 倍。在我的代码中添加 4 个字符还算不错;)

我目前正在尝试向管理人员推销使用 Scala 的好处，尤其是在使用 CPU 进行扩展时。为此，我创建了一个简单的测试应用程序，该应用程序执行大量向量数学运算，并且有点惊讶地发现，在我的四核机器上，运行时间并没有明显更好。有趣的是，我发现第一次查看集合时运行时间是最差的，并且在后续调用中会变得更好。并行集合中是否有一些懒惰的事情导致了这种情况，或者我只是做错了？应该指出的是，我来自 C++/C# 世界，所以我完全有可能以某种方式弄乱了我的配置。无论如何，这是我的设置：

InteliJ Scala Plugin

Scala 2.9.1.final

Windows 7 64 位，四核处理器（无超线程）

import util.Random

  // simple Vector3D class that has final x,y,z components a length, and a '-' function
  class Vector3D(val x:Double,  val y:Double, val z:Double)
  {
    def length = math.sqrt(x*x+y*y+z*z)
    def -(rhs : Vector3D ) = new Vector3D(x - rhs.x, y - rhs.y, z - rhs.z)
  }

object MainClass {

  def main(args : Array[String]) =
  {
    println("Available CPU's: " + Runtime.getRuntime.availableProcessors())
    println("Parallelism Degree set to: " + collection.parallel.ForkJoinTasks.defaultForkJoinPool.getParallelism);
    // my position
    val myPos = new Vector3D(0,0,0);

    val r = new Random(0);

    // define a function nextRand that gets us a random between 0 and 100
    def nextRand = r.nextDouble() * 100;

    // make 10 million random targets
    val targets = (0 until 10000000).map(_ => new Vector3D(nextRand, nextRand, nextRand)).toArray
    // take the .par hit before we start profiling
    val parTargets = targets.par

    println("Created " + targets.length + " vectors")

    // define a range function
    val rangeFunc : (Vector3D => Double) = (targetPos) => (targetPos - myPos).length

    // we'll select ones that are <50
    val within50 : (Vector3D => Boolean) = (targetPos) => rangeFunc(targetPos) < 50

    // time it sequentially
    val startTime_sequential = System.currentTimeMillis()
    val numTargetsInRange_sequential = targets.filter(within50)
    val endTime_sequential = System.currentTimeMillis()
    println("Sequential (ms): " + (endTime_sequential - startTime_sequential))

    // do the parallel version 10 times
    for(i <- 1 to 10)
    {

      val startTime_par = System.currentTimeMillis()
      val numTargetsInRange_parallel = parTargets.filter(within50)
      val endTime_par = System.currentTimeMillis()

      val ms = endTime_par - startTime_par;
      println("Iteration[" + i + "] Executed in " + ms + " ms")
    }
  }
}

该程序的输出是：

Available CPU's: 4
Parallelism Degree set to: 4
Created 10000000 vectors
Sequential (ms): 216
Iteration[1] Executed in 227 ms
Iteration[2] Executed in 253 ms
Iteration[3] Executed in 76 ms
Iteration[4] Executed in 78 ms
Iteration[5] Executed in 77 ms
Iteration[6] Executed in 80 ms
Iteration[7] Executed in 78 ms
Iteration[8] Executed in 78 ms
Iteration[9] Executed in 79 ms
Iteration[10] Executed in 82 ms

那么这里发生了什么？前两次我们进行过滤，速度较慢，然后速度加快？我知道本质上存在并行启动成本，我只是想弄清楚在我的应用程序中表达并行性的意义在哪里，特别是我希望能够向管理人员展示一个运行 3-4 次的程序在四核盒子上速度更快。这难道不是一个好问题吗？

有想法吗？

Answer 1

你患有微基准病。您很可能正在对 JIT 编译阶段进行基准测试。您需要首先通过预运行来预热 JIT。

最好的想法可能是使用微基准测试框架，例如 http://code.google.com/p /caliper/ 为您处理所有这些。

编辑：有一个很好的 SBT 模板，用于 Caliper 基准测试 Scala项目，如此博文中引用的

Answer 2

事情确实会加速，但这与并行与顺序无关，你不是在比较苹果与苹果。 JVM 有一个 JIT（即时）编译器，只有在代码使用一定次数后才会编译一些字节代码。因此，您在第一次迭代中看到的是尚未进行 JIT 编译的代码的执行速度较慢，以及正在进行的 JIT 编译本身的时间也较慢。删除.par，使其全部顺序，这是我在我的机器上看到的（迭代次数减少了 10 倍，因为我使用的是较旧的机器）：

Sequential (ms): 312
Iteration[1] Executed in 117 ms
Iteration[2] Executed in 112 ms
Iteration[3] Executed in 112 ms
Iteration[4] Executed in 112 ms
Iteration[5] Executed in 114 ms
Iteration[6] Executed in 113 ms
Iteration[7] Executed in 113 ms
Iteration[8] Executed in 117 ms
Iteration[9] Executed in 113 ms
Iteration[10] Executed in 111 ms

但它都是顺序的！您可以使用 JVM -XX:+PrintCompilation（在 JAVA_OPTS 中设置）查看 JVM 在 JIT 方面所做的工作，或者使用 -J-XX:+PrintCompilation scala 选项。在第一次迭代中，您将看到大量 JVM 打印语句，显示正在执行 JIT 的内容，然后它会稳定下来，

因此要进行逐个比较，您首先要在没有 的情况下运行 。 par，然后添加 par 并运行相同的程序。在我的双核上，当使用 .par 时，我得到：

Sequential (ms): 329
Iteration[1] Executed in 197 ms
Iteration[2] Executed in 60 ms
Iteration[3] Executed in 57 ms
Iteration[4] Executed in 58 ms
Iteration[5] Executed in 59 ms
Iteration[6] Executed in 73 ms
Iteration[7] Executed in 56 ms
Iteration[8] Executed in 60 ms
Iteration[9] Executed in 58 ms
Iteration[10] Executed in 57 ms

，一旦稳定，或多或少会有 2 倍的加速。

因此 需要小心的是装箱和拆箱，特别是如果您只与 Java 库的高阶函数（如 filter）进行基本类型的装箱和拆箱比较，这通常是那些最初感到失望的原因。将代码从 Java 转换为 Scala。

虽然它不适用于本例，因为 for 超出了计时范围，但使用 for 而不是 while 也有一些成本，但是使用 -optimize scalac 标志时，2.9.1 编译器应该表现得不错。

Answer 3

除了前面提到的 JIT 优化之外，您需要评估的一个关键概念是您的问题是否倾向于并行化：拆分、线程协调和连接的固有成本会抵消并行处理的优势。 Scala 向您隐藏了这种复杂性，但您确实需要知道何时应用它以获得良好的结果。

就您而言，尽管您正在执行大量操作，但每个操作本身对于 CPU 来说几乎是微不足道的。要查看并行集合的运行情况，请尝试以单位为单位进行大量操作。

对于类似的 Scala 演示，我使用了一个简单（效率低下）的算法来计算一个数字是否是素数：
def isPrime(x:Int) = (2 to x/2).forall(y=>x%y!=0)

然后使用您提供的相同逻辑来确定集合中的数字 ：

val col = 1 to 1000000
col.filter(isPrime(_))  // sequential
col.par.filter(isPrime(_)) // parallel

CPU 行为确实显示了两者之间的差异

4 核 CPU 中并行收集的时间大约缩短了 3.5 倍。

Answer 4

怎么样

val numTargetsInRange_sequential = parTargets.filter(within50)

？

此外，使用映射而不是过滤操作可能会获得更令人印象深刻的结果。