控制 Akka 中消耗大量内存的 Actor 的生成答案

【问题标题】：Controlling spawning of actors in Akka who consume noticeable amounts of memory控制 Akka 中消耗大量内存的 Actor 的生成
【发布时间】：2014-01-10 21:08:49
【问题描述】：

我已经使用 akka 的 actor 模型构建了一个分布式流式机器学习模型。通过向 Actor 发送训练实例（要训练的数据）来异步训练模型。对这些数据的训练会占用计算时间并改变参与者的状态。

目前我正在使用历史数据来训练模型。我想运行一组不同配置的模型，这些模型在相同的数据上进行训练，并查看不同的集成指标如何变化。本质上，这里是对我使用 Thread.sleep(1) 和表示计算时间和状态的数据数组所做的事情的简单得多的模拟。

implicit val as = ActorSystem()

case object Report

case class Model(dataSize: Int) {
  val modelActor: ActorRef = actor(new Act {
    val data = Array.fill(dataSize)(0)
    become {
      case trainingData: Int => {
        // Screw with the state of the actor and pretend that it takes time
        Thread.sleep(1)
        data(Math.abs(Random.nextInt % dataSize)) == trainingData
      }
      case Report => {
          println(s"Finished $dataSize")
          context.stop(self)
        }
      }
    })

  def train(trainingInstance: Int) = modelActor ! trainingInstance

  def report: Unit = modelActor ! Report
}

val trainingData = Array.fill(5000)(Random.nextInt)

val dataSizeParams = (1 to 500)

接下来我使用 for 循环来改变参数（由 dataSizeParams 数组表示）

for {
  param <- dataSizeParams
} {
  // make model with params
  val model = Model(param)
  for {
    trainingInstance <- trainingData
  } {
    model.train(trainingInstance)
  }
  model.report
}

for 循环绝对是做我想做的事情的错误方法。它并行启动所有不同的模型。当 dataSizeParams 在 1 到 500 范围内时，它运行良好，但如果我将其提高到某个高度，我的模型 EACH 开始占用明显的内存块。我想出的是下面的代码。本质上，我有一个模型大师，他可以根据他收到的运行消息的数量来控制一次运行的模型数量。现在每个模型都包含对这个主actor的引用，并在他完成处理后向他发送一条消息：

// Alternative that doesn't use a for loop and instead controls concurrency through what I'm calling a master actor
case object ImDone
case object Run

case class Model(dataSize: Int, master: ActorRef) {
  val modelActor: ActorRef = actor(new Act {
    val data = Array.fill(dataSize)(0)
    become {
      case trainingData: Int => {
        // Screw with the state of the actor and pretend that it takes time
        Tread.sleep(1)
        data(Math.abs(Random.nextInt % dataSize)) == trainingData
      }
      case Report => {
          println(s"Finished $dataSize")
          master ! ImDone
          context.stop(self)
        }
      }
    })

  def train(trainingInstance: Int) = modelActor ! trainingInstance

  def report: Unit = modelActor ! Report
}

val master: ActorRef = actor(new Act {
  var paramRuns = dataSizeParams.toIterator
  become {
    case Run => {
      if (paramRuns.hasNext) {
        val model = Model(paramRuns.next(), self)
        for {
          trainingInstance <- trainingData
        } {
          model.train(trainingInstance)
        }
        model.report
      } else {
        println("No more to run")
        context.stop(self)
      }
    }
    case ImDone =>  {
      self ! Run
    }
  }
})

master ! Run

主代码没有任何问题（我可以看到）。我对一次生成的模型数量有严格的控制，但我觉得我错过了一种更容易/干净/开箱即用的方法。另外我想知道是否有任何巧妙的方法可以限制同时运行的模型数量，比如查看系统的 CPU 和内存使用情况。

【问题讨论】：

使用路由器和演员池不是您的选择吗？模型构建器池将被限制为您设置的大小，并且每个池模型构建器在构建当前模型之前不会移动到邮箱中的下一条消息，从而更好地控制内存和 CPU。
这很有趣。我们尝试不使用 Akka 手动管理线程，最终我们管理了 actor……我遇到了类似的问题。创建演员的 for 循环似乎是一个非常糟糕的主意。

标签： scala concurrency machine-learning akka

【解决方案1】：

您正在寻找工作拉动模式。我强烈推荐 Akka 开发人员的这篇博文：

http://letitcrash.com/post/29044669086/balancing-workload-across-nodes-with-akka-2

我们在 Akka 的集群功能之上使用了一个变体来避免恶意并发。通过让工作角色 pull 工作而不是让主管 push 工作，您可以通过简单地限制数量来优雅地控制工作量（以及 CPU 和内存使用）工人演员。

与纯路由器相比，这有一些优势：更容易跟踪故障（如该帖子中所述）并且工作不会在邮箱中停滞（可能会丢失）。

另外，如果您使用远程处理，我建议您不要在消息中发送大量数据。让工作节点在触发时自己从另一个源中提取数据。我们使用 S3。

【讨论】：