scala - 在其他长期运行的 Actor 在场的情况下，如何防止 Actor 饥饿？

Question

这是使用 Scala 2.8 Actors。我有一个可以并行化的长时间运行的工作。它由大约 650,000 个工作单元组成。我将它分成 2600 个不同的独立子任务，并为每个子任务创建一个新角色:

actor {
  val range = (0L to total by limit)
  val latch = new CountDownLatch(range.length)
  range.foreach { offset =>
    actor {
      doExpensiveStuff(offset,limit)
      latch.countDown
    }
  }
  latch.await
}

这工作得相当好，但总体上需要 2+h 才能完成。问题是，与此同时，我创建的任何其他执行正常任务的 Actor 似乎都被最初的 2600 个 Actor 饿死了，这些 Actor 也在耐心地等待他们在线程上运行的时间，但等待的时间比任何新 Actor 都长一起来。

我怎样才能避免这种饥饿？

初步想法:

不要使用 2600 名 Actor ，而是使用一名 Actor ，依次翻阅大量作品。我不喜欢这个，因为我希望通过拆分来完成这项工作。

使用两个 Actor 代替 2600 个 Actor ，每个 Actor 处理总工作集的不同一半。这可能会更好，但如果我的机器有 8 个内核呢？我可能想利用更多。

UPDATE

有些人根本就质疑 Actors 的使用，特别是因为消息传递功能没有在工作人员中使用。我曾假设 Actor 是围绕 ThreadPool 的一个非常轻量级的抽象，其性能水平与简单地手动编码基于 ThreadPool 的执行的性能水平相同或接近。所以我写了一个小基准:

import testing._
import java.util.concurrent._
import actors.Futures._

val count = 100000
val poolSize = 4
val numRuns = 100

val ActorTest = new Benchmark {
  def run = {
    (1 to count).map(i => future {
      i * i
    }).foreach(_())
  }
}

val ThreadPoolTest = new Benchmark {
  def run = {
    val queue = new LinkedBlockingQueue[Runnable]
    val pool = new ThreadPoolExecutor(
          poolSize, poolSize, 1, TimeUnit.SECONDS, queue)
    val latch = new CountDownLatch(count)
    (1 to count).map(i => pool.execute(new Runnable {
      override def run = {
        i * i
        latch.countDown
      }
    }))
    latch.await
  }
}

List(ActorTest,ThreadPoolTest).map { b =>
  b.runBenchmark(numRuns).sum.toDouble / numRuns
}

// List[Double] = List(545.45, 44.35)

我在 ActorTest 中使用了 Future 抽象来避免将消息传递回另一个 Actor 来表示工作已完成。我惊讶地发现我的 Actor 代码慢了 10 倍以上。请注意，我还使用初始池大小创建了我的 ThreadPoolExecutor，用于创建默认 Actor 池。

回想起来，似乎我可能过度使用了 Actor 抽象。我将考虑为这些不同的、昂贵的、长时间运行的任务使用单独的线程池。

Answer 1

不管你有多少 Actor ，如果你没有明确地配置你的日程安排，所有的 Actor 都得到了 的支持。单例 fork/join 调度程序(针对容量为 4 的线程池运行，如果我没记错的话)。这就是饥饿的来源。

您应该为您的参与者池尝试不同的调度程序，以找到显示最佳性能的调度程序(尝试 ResizableThreadPoolScheduler，如果您想使用尽可能多的线程来最大化并行性)

您需要为庞大的 Actor 池设置一个单独的调度程序(系统中的其他 Actor 不应使用它)

正如@DaGGeRRz 所建议的那样，您可以尝试提供可配置调度程序的 Akka 框架(例如，工作窃取负载平衡调度程序将事件从繁忙参与者的邮箱移动到空闲参与者)

从评论到默认 Actor执行:

The run-time system can be configured to use a larger thread pool size (for example, by setting the actors.corePoolSize JVM property). The scheduler method of the Actor trait can be overridden to return a ResizableThreadPoolScheduler, which resizes its thread pool to avoid starvation caused by actors that invoke arbitrary blocking methods. The actors.enableForkJoin JVM property can be set to false, in which case a ResizableThreadPoolScheduler is used by default to execute actors.

另外:一个有趣的 thread on schedulers在斯卡拉朗。