在有其他长时间运行的演员在场的情况下，如何防止演员饥饿？

Question

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

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 名演员。我不喜欢这个，因为我希望通过拆分来尽快完成这项工作。
• 使用两个演员，而不是 2600 个演员，每个演员处理总工作集的不同一半。这可能会更好，但如果我的机器有 8 个核心怎么办？我可能想利用更多。

更新

有些人根本质疑 Actor 的使用，特别是因为消息传递功能没有在工作人员中使用。我曾假设 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

无论您有多少参与者，如果您没有显式配置调度，那么所有参与者都由一个单个 fork/join 调度程序支持（针对容量为 4 的线程池运行，如果我'我没有记错）。这就是饥饿的来源。

1. 您应该为您的 Actor 池尝试不同的调度程序，以找到表现出最佳性能的调度程序（如果您想使用尽可能多的线程来最大化并行性，请尝试 ResizingThreadPoolScheduler）
2. 您需要为庞大的 Actor 池拥有一个单独的调度程序（系统中的其他参与者不应该使用它）
3. 正如@DaGGeRRz所建议的，你可以尝试提供可配置调度程序的Akka框架（例如，工作窃取负载平衡调度程序将事件从繁忙参与者的邮箱移动到空闲参与者

）默认注释 Actor 实现：

运行时系统可配置
使用更大的线程池大小（例如
例如，通过设置
actors.corePoolSize JVM 属性）。
Actor 的 scheduler 方法
可以覆盖特征以返回
ResizableThreadPoolScheduler，其中
调整其线程池的大小以避免
演员造成的饥饿
调用任意阻塞方法。这
actors.enableForkJoin JVM 属性
可以设置为 false，在这种情况下
使用ResizableThreadPoolScheduler
默认情况下执行参与者。

另外：scala-lang 上有一个有趣的关于调度程序的线程。

Answer 2

从您的示例来看，您实际上根本不需要使用参与者，因为您没有将消息传递给您的工作单位，或者回复，甚至循环。

为什么不直接创建一批 Future 然后等待它们完成呢？这样，底层的 Fork Join Pool 就可以完全自由地为您的系统决定适当的并行级别（即线程数）：

import actors.Futures._
def mkFuture(i : Int) = future {
  doExpensiveStuff(i, limit)
}
val fs = (1 to range by limit).map(mkFuture)
awaitAll(timeout, fs) //wait on the work all finishing

请注意，您只能通过处理更多任务来从并行性中受益。如果昂贵的工作不受 CPU 限制（可能是 IO 限制），那么您的系统同时拥有多个核心。

Answer 3

我没有使用具有该语法的参与者，但默认情况下我认为 scala 中的所有参与者都使用线程池。

请参阅如何为 actor 指定线程池