scala - Akka Actor no termina si se lanza una excepción

routing fault-tolerance

Question

Actualmente estoy tratando de comenzar con Akka y estoy enfrentando un problema extraño. Tengo el siguiente código para mi Actor:

class AkkaWorkerFT extends Actor {
  def receive = {
    case Work(n, c) if n < 0 => throw new Exception("Negative number")
    case Work(n, c) => self reply n.isProbablePrime(c);
  }
}

Y así es como comienzo a mis trabajadores:

val workers = Vector.fill(nrOfWorkers)(actorOf[AkkaWorkerFT].start());
val router = Routing.loadBalancerActor(SmallestMailboxFirstIterator(workers)).start()

Y así es como lo cierro todo:

futures.foreach( _.await )
router ! Broadcast(PoisonPill)
router ! PoisonPill

Ahora, lo que sucede es que si envío mensajes a los trabajadores con n> 0 (no se lanza ninguna excepción), todo funciona bien y la aplicación se apaga correctamente. Sin embargo, tan pronto como le envío un mensaje único que da como resultado una excepción, la aplicación no finaliza porque todavía hay un actor ejecutándose, pero no puedo averiguar de dónde viene.

En caso de que ayude, esta es la pila del hilo en cuestión:

  Thread [akka:event-driven:dispatcher:event:handler-6] (Suspended) 
    Unsafe.park(boolean, long) line: not available [native method]  
    LockSupport.park(Object) line: 158  
    AbstractQueuedSynchronizer$ConditionObject.await() line: 1987   
    LinkedBlockingQueue<E>.take() line: 399 
    ThreadPoolExecutor.getTask() line: 947  
    ThreadPoolExecutor$Worker.run() line: 907   
    MonitorableThread(Thread).run() line: 680   
    MonitorableThread.run() line: 182   

PD: El hilo que no termina no es ninguno de los subprocesos de trabajo, porque he agregado una devolución de llamada post-parada, cada uno de ellos se detiene correctamente.

PPS: Actors.registry.shutdownAll soluciones al problema, pero creo que shutdownAll solo debe usarse como último recurso, ¿no es así?

Answer 1

Apagar el registro para asegurarse de que las cosas terminen, como lo propuso Viktor, es un poco extraño. Lo que puedes hacer en cambio es:

EventHandler.shutdown()

que apaga limpiamente todos los oyentes (registradores) que mantienen el mundo en funcionamiento después de la excepción:

def shutdown() {
  foreachListener(_.stop())
  EventHandlerDispatcher.shutdown()
}

Answer 2

Gire el registrador en el akka.conf

Answer 3

La forma correcta de manejar los problemas dentro de los actores Akka no es arrojar una excepción, sino establecer jerarquías de supervisor

"Lanzar una excepción en código concurrente (supongamos que estamos usando actores no vinculados), simplemente hará explotar el hilo que actualmente ejecuta el actor.

No hay forma de descubrir que las cosas salieron mal (aparte de inspeccionar el seguimiento de la pila). No hay nada que puedas hacer al respecto ".

vea Tolerancia a fallas a través de jerarquías de supervisor (1.2)

* note * lo anterior es cierto para las versiones anteriores de Akka (1.2). En las versiones más nuevas (por ejemplo, 2.2) aún establecería una jerarquía de supervisor, pero atrapará las excepciones lanzadas por procesos secundarios. p.ej

class Child extends Actor {
    var state = 0
    def receive = {
      case ex: Exception ⇒ throw ex
      case x: Int        ⇒ state = x
      case "get"         ⇒ sender ! state
    }
  }

y en el supervisor:

class Supervisor extends Actor {
    import akka.actor.OneForOneStrategy
    import akka.actor.SupervisorStrategy._
    import scala.concurrent.duration._

    override val supervisorStrategy =
      OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
        case _: ArithmeticException      ⇒ Resume
        case _: NullPointerException     ⇒ Restart
        case _: IllegalArgumentException ⇒ Stop
        case _: Exception                ⇒ Escalate
      }

    def receive = {
      case p: Props ⇒ sender ! context.actorOf(p)
    }
  }

vea Tolerancia a fallas a través de jerarquías de supervisor (2.2)