在退出之前等待完成子级中的所有工作的演员

Question

无法弄清楚如何解决以下问题：我有一些演员（工人）在收到（我的意思是反应）任务时以某种方式执行任务。主要参与者（领班）控制此过程并可以接收停止工作的任务。在这种情况下，主要参与者必须停止创建新任务，并等待工作人员完成所有现有任务，然后主要参与者才应退出。

import actors.Actor
import actors.Actor._

class Foreman extends Actor{
  val workerA = new WorkerA
  val workerB = new WorkerB
  val workerC = new WorkerC
  self.link(workerA)
  self.link(workerB)
  self.link(workerC)
  def act{
    workerA.start
    workerB.start
    workerC.start

    // adding tasks to workers somehow
    //...
    loop{
      case ResultOfTask(res) => //...
      case Stop => //workers mustn't immediately stop but must finish their tasks and then exit
      case ProductionAccident => //...
    }


  }
}

case class Task(activity:String)
case class ResultOfTask(result:String)

trait Worker extends Actor{

  def act{
    loop{
      react{
        case Task(activity) => sender ! processTask(activity)
      }
    }
  }

  def processTask(activity:String):ResultOfTask
}

为了解决这个问题，我编写了以下代码：

def condition = workerA.getState!=State.Suspended  && workerB.getState!=State.Suspended && workerC.getState!=State.Suspended && mailboxSize == 0
case Stop => {
  if(condition) exit("sweet dreams") else continue
}

检查主要演员是否应该退出。另一种变体是在“Worker”特征中包含计算者，当工作人员收到消息时增加它，并在它响应时减少它。

trait Worker extends Actor {
  private var count = 0
  def act {
    loop{
      react{
        case Task(activity) => {
          count += 1
          sender ! processTask(activity)
          count -= 1
        }
      }
    }
  }
  def hasDoneAllTasks = count == 0

  def processTask(activity: String): ResultOfTask
}

“工头”中的“条件”功能将是

def condition = workerA.hasDoneAllTasks   && workerB.hasDoneAllTasks  && workerC.hasDoneAllTasks  && mailboxSize == 0

我希望有更好的解决方案，并且您会提出它们。

Answer 1

如果工头总是期望从工人那里得到答案，那么解决方案很简单：工头维护一个计数器，每次发送一条消息时，它都会递增它，每次从工人那里收到一条消息时，它都会递减它。只要计数器为零，它就可以自行停止（假设没有其他人向工作人员发送消息）。

如果工头并不总是期待工人的答复，您可以通过发送无内容的消息

case object Done { }

并让工人在完成后回复该消息来实现这种情况。然后，参见上文。

如果工头不是唯一与工人交谈的人，或者你希望后台少一些闲聊，那么工头和工人就必须进行谈判。工头可以发送消息

case object RequestStop { }

，工人们会做一些优雅的事情，并在完成后回复 Done。当工头收到的 Done 消息与发送的 RequestStop 消息一样多时，它就可以自由退出。

Answer 2

为什么在向 Workers 发送作业时不包含对 Foreman actor 的引用？然后，当工人关闭时，他们可以向工头发送通知。每次 Foreman 收到工人关闭消息时，它都会记录该消息并查看所有工人是否已完成。如果是这样，它也会自行关闭。

Answer 3

我的方法是在 Foreman 中完成所有计算。

您没有写领班如何创建任务，所以我假设它是为了响应消息

class Foreman extends Actor{

  var numTasks: Int = 0
  var shouldExit = false

  def act = loop {
   react {
     case t: Task =>
       if (!shouldExit)  {
         numTasks += 1
         selectWorker ! t
       } else {
         // send some kind of error status to sender
       }
     case ResultOfTask(rest) =>
       numTasks -= 1
       // ....
       if (numTasks == 0 && shouldExit) exit
     case Stop() => shoudExit = true
}

然后改进是优先考虑停止，以便即使队列中有任务消息也能处理它

  def act = loop {
   reactWithin(0) {
     case Stop() => shouldStop = true
     case TIMEOUT => react {
       case t: Task =>
         if (!shouldExit)  {
           numTasks += 1
           selectWorker ! t
         } else {
           // send some kind of error status to sender
         }
       case ResultOfTask(rest) =>
         numTasks -= 1
         // ....
         if (numTasks == 0 && shouldExit) exit
       case Stop() => shoudExit = true
  }
}

Answer 4

您可以利用的一件事是 trapExit、case Exit 和 exit、system。在你的主要参与者中，你可以将 trapExit 设置为 true：

// Foreman

def act() {
  trapExit = true
  link(workerA)
  link(workerB)
  ...
}

这意味着当工作进程终止时，你的领班参与者将收到退出消息，其中包括原因：

// Foreman

def act() {
  ....
  loop { react {
    case Exit (worker: Actor, reason: AnyRef) => {
      // decrement counter, or list of workers, and exit if empty
    }
    ...
   }}
}

你可以对原因参数进行模式匹配。例如，在您的工人类中，您可能会使用各种案例类来指示工头应该做什么：

// Worker:

exit(WorkComplete)
exit(Emergency)

等等，等等。当您的工作线程抛出异常时，它将终止并向链接的进程发送包含异常的退出消息。考虑到这种情况，你最终可能会得到这样的结果：

// Foreman

def act() {
  ....
  loop { react {
    case Exit (worker: Actor, reason: WorkComplete) => {
      // decrement counter, or list of workers, and exit if empty
    }

    case Exit (worker: Actor, reason: TasksExhausted) => {
      // log something, or make shut down worker if too many are exhausted
    }

    case Exit (worker: Actor, reason: Exception) => {
      // log the exception, then restart the actor
    }

    ...
   }}
}

从你最初的问题来看，你是否希望工人继续工作，即使他们已经完成，直到工头告诉他们应该在时间到了时退出。如果是这种情况，可以向工作人员发送一条消息，告诉他们“完成后退出”，并且您可以使用 trapExit 机制告诉他们已经完成。

希望这能激发出一个有趣的解决方案！