Scala Future[T] 开销很大？

Question

我写了一个合并排序来测试scala Future[T]类型的异步计算性能。

我有一个 4 核 CPU，所以我预计异步计算比同步计算快大约 4 倍，因为我使用完整的 cpu 功能(由于子任务的大小相同，停顿时间应该很小)。然而结果表明，异步合并排序比普通合并排序慢。

是我并发写得不好还是只是因为 Future[T] 开销？谁能帮我解释一下吗？

package kai.concurrent

import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.Random

object MergeSort {
  lazy val regressThreadhold = 10000

  def mergeSortedList[T](a: Seq[T], b: Seq[T])(implicit ord: Ordering[T]): Seq[T] = {
    def loop(a: Seq[T], b: Seq[T], acc: Seq[T]): Seq[T] = {
      if (a.isEmpty && b.isEmpty) acc
      else if (a.isEmpty) b.reverse ++: acc
      else if (b.isEmpty) a.reverse ++: acc
      else if (ord.lt(a.head, b.head)) loop(a.tail, b, a.head +: acc)
      else loop(a, b.tail, b.head +: acc)
    }

    loop(a, b, Seq()).reverse
  }

  def mergeSortAsync0[T](x: Seq[T])(implicit ord: Ordering[T]): Future[Seq[T]] =
    if (x.size <= regressThreadhold) Future(mergeSort(x)) else {
      val (left, right) = x.splitAt(x.size / 2)
      val Seq(leftSorted, rightSorted) = Seq(left, right).map(seq => Future(mergeSortAsync0(seq)).flatten)
      leftSorted.zip(rightSorted).map(pair => mergeSortedList(pair._1, pair._2))
    }

  def mergeSortAsync[T](x: Seq[T])(implicit ord: Ordering[T]): Seq[T] =
    Await.result(mergeSortAsync0(x), Duration.Inf)

  def mergeSort[T](x: Seq[T])(implicit ord: Ordering[T]): Seq[T] =
    if (x.size <= 1) x else {
      val (left, right) = x.splitAt(x.size / 2)
      val (leftSorted, rightSorted) = (mergeSort(left), mergeSort(right))
      mergeSortedList(leftSorted, rightSorted)
    }
}

object MergeSortTest extends App {

  import kai.util.ProfileUtil.TimeResult

  val seq: Vector[Double] = (1 to 1000000).map(i => Random.nextDouble()).toVector
  val seqMergeSortAsync = MergeSort.mergeSortAsync(seq) withWallTimePrinted "mergeSortAsync"
  val seqMergeSort = MergeSort.mergeSort(seq) withWallTimePrinted "mergeSort"
  val seqSort = seq.sorted withWallTimePrinted "sorted"
  println(seqSort == seqMergeSort && seqMergeSort == seqMergeSortAsync)
}

输出:

mergeSortAsync elapsed time: 3186 ms

mergeSort elapsed time: 3300 ms

sorted elapsed time: 581 ms

true

Answer 1

我已复制您的测试并通过 JMH 运行它(使用 sbt-jmh )。我使用预定义的 scala.concurrent.ExecutionContext.Implicits.global 来作为测试中的底层执行上下文。

结果:

[info] Benchmark                          Mode  Cnt  Score   Error  Units
[info] MergeSortTest.benchMergeSortAsync  avgt   25  1.534 +–’ 0.212   s/op
[info] MergeSortTest.benchMergeSortSync   avgt   25  2.325 +–’ 0.437   s/op
[info] MergeSortTest.benchScalaSort       avgt   25  0.382 +–’ 0.006   s/op

您可以在此处看到，运行并行版本大约比顺序版本快 1.5 倍，而 Scala 排序比顺序合并排序快 6 倍。

需要记住，在进行此类微观基准测试时，需要考虑很多因素。通常最好让 JMH 为您处理 JVM 运行时的微妙之处。

插件.sbt:

addSbtPlugin("pl.project13.scala" % "sbt-jmh" % "0.2.27")

构建.sbt:

enablePlugins(JmhPlugin)

测试代码:

import java.util.concurrent.TimeUnit

import org.openjdk.jmh.annotations._

import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.util.Random
import scala.concurrent.ExecutionContext.Implicits.global

/**
  * Created by Yuval.Itzchakov on 21/08/2017.
  */
@State(Scope.Thread)
@Warmup(iterations = 3, time = 1)
@Measurement(iterations = 5, timeUnit = TimeUnit.MILLISECONDS)
@BenchmarkMode(Array(Mode.AverageTime))
@Fork(5)
class MergeSortTest {

  var seq: Seq[Double] = _

  @Setup
  def setup(): Unit = {
    seq = (1 to 1000000).map(i => Random.nextDouble()).toVector
  }

  lazy val regressThreadhold = 10000

  def mergeSortedList[T](a: Seq[T], b: Seq[T])(implicit ord: Ordering[T]): Seq[T] = {
    def loop(a: Seq[T], b: Seq[T], acc: Seq[T]): Seq[T] = {
      if (a.isEmpty && b.isEmpty) acc
      else if (a.isEmpty) b.reverse ++: acc
      else if (b.isEmpty) a.reverse ++: acc
      else if (ord.lt(a.head, b.head)) loop(a.tail, b, a.head +: acc)
      else loop(a, b.tail, b.head +: acc)
    }

    loop(a, b, Seq()).reverse
  }

  def mergeSortAsync0[T](x: Seq[T])(implicit ord: Ordering[T]): Future[Seq[T]] =
    if (x.size <= regressThreadhold) Future(mergeSort(x)) else {
      val (left, right) = x.splitAt(x.size / 2)
      val Seq(leftSorted, rightSorted) = Seq(left, right).map(seq => Future(mergeSortAsync0(seq)).flatten)
      leftSorted.zip(rightSorted).map(pair => mergeSortedList(pair._1, pair._2))
    }

  def mergeSortAsync[T](x: Seq[T])(implicit ord: Ordering[T]): Seq[T] =
    Await.result(mergeSortAsync0(x), Duration.Inf)

  def mergeSort[T](x: Seq[T])(implicit ord: Ordering[T]): Seq[T] =
    if (x.size <= 1) x else {
      val (left, right) = x.splitAt(x.size / 2)
      val (leftSorted, rightSorted) = (mergeSort(left), mergeSort(right))
      mergeSortedList(leftSorted, rightSorted)
    }

  @Benchmark
  def benchMergeSortSync(): Unit = {
    mergeSort(seq)
  }

  @Benchmark
  def benchMergeSortAsync(): Unit = {
    mergeSortAsync(seq)
  }

  @Benchmark
  def benchScalaSort(): Unit = {
    seq.sorted
  }
}