How to periodically execute an AKKA Actor's routine?

Question

I need an actor to stop one of its children, so that I can possibly create a new actor with same name (UUID ?).

I've got an ActorSystem with one Actor child. And this child creates new actors with context.actorOf and context.watch. When I try to stop one of these using context.stop, I observe that its postStop method is called as expected, but no matter how long I wait (seconds... minutes...), it never sends back the Terminated message to its creator (and watching) actor.

I read this in the AKKA documentation:

Since stopping an actor is asynchronous, you cannot immediately reuse the name of the child you just stopped; this will result in an InvalidActorNameException. Instead, watch the terminating actor and create its replacement in response to the Terminated message which will eventually arrive.

I don't care waiting for normal termination, but I really need actors to eventually terminate when asked to. Am I missing something ? Should I create actors directly from the system instead of from an actor ?

EDIT:

Here is my code :

object MyApp extends App {
  def start() = {
    val system = ActorSystem("MySystem")
    val supervisor = system.actorOf(Supervisor.props(), name = "Supervisor")
  }

  override def main(args: Array[String]) {
    start()
  }
}

object Supervisor {
  def props(): Props = Props(new Supervisor())
}

case class Supervisor() extends Actor {
  private var actor: ActorRef = null

  start()

  def newActor(name: String): ActorRef = {
    try {
      actor = context.actorOf(MyActor.props(name), name)
      context.watch(actor)
    } catch {
      case iane: InvalidActorNameException =>
        println(name + " not terminated yet.")
      null
    }
  }

  def terminateActor() {
    if (actor != null) context.stop(actor)
    actor = null
  }

  def start() {
    while (true) {
      // do something
      terminateActor()
      newActor("new name possibly same name as a previously terminated one")
      Thread.sleep(5000)
    }
  }

  override def receive = {
    case Terminated(x) => println("Received termination confirmation: " + x)
    case _ => println("Unexpected message.")
  }

  override def postStop = {
    println("Supervisor called postStop().")
  }
}

object MyActor {
  def props(name: String): Props = Props(new MyActor(name))
}

case class MyActor(name: String) extends Actor {
  run()

  def run() = {
    // do something
  }

  override def receive = {
    case _ => ()
  }

  override def postStop {
    println(name + " called postStop().")
  }
}

EDIT²: As mentionned by @DanGetz, one shall not need to call Thread.sleep in an AKKA actor. Here what I needed was a periodical routine. This can be done using the AKKA context scheduler. See: http://doc.akka.io/docs/akka/2.3.3/scala/howto.html#scheduling-periodic-messages . Instead I was blocking the actor in an infinite loop, preventing it to use its asynchronous mecanisms (messages). I changed the title since the problem was actually not involving actor termination.

Could your Thread.sleep be preventing receive from ever being called? — Dan Getz
Thx @DanGetz. Actually, it is not Thread.sleep, but the fact that I use an infinite loop. So what I need is an actor which executes some routine periodically to replace this loop, and let the asynchronous mecanisms of the actor work. — ygu
This seems ok, I'll try it out : doc.akka.io/docs/akka/2.3.3/scala/… — ygu
I think what I should have said is, you should pretty much never be calling Thread.sleep from an Actor. — Dan Getz

cmbaxter cmbaxter · Accepted Answer · 2014-05-22T13:29:11

It's hard to gauge exactly what you want now that the question has changed a bit, but I'm going to take a stab anyway. Below you will find a modified version of your code that shows both periodic scheduling of a task (one that kicks off the child termination process) and also watching a child and only creating a new one with the same name when we are sure the previous one has stopped. If you run the code below, every 5 seconds you should see it kill the child and wait for the termination message before stating a new one with the exact same name. I hope this is what you were looking for:

object Supervisor {
  val ChildName = "foo"
  def props(): Props = Props(new Supervisor())
  case class TerminateChild(name:String)
}

case class Supervisor() extends Actor {
  import Supervisor._
  import scala.concurrent.duration._
  import context._

  //Start child upon creation of this actor
  newActor(ChildName)

  override def preStart = {
    //Schedule regular job to run every 5 seconds
    context.system.scheduler.schedule(5 seconds, 5 seconds, self, TerminateChild(ChildName))
  }

  def newActor(name: String): ActorRef = {
    val child = context.actorOf(MyActor.props(name), name)
    watch(child)
    println(s"created child for name $name")
    child
  }

  def terminateActor(name:String) = context.child(ChildName).foreach{ ref =>
    println(s"terminating child for name $name")
    context stop ref
  }

  override def receive = {
    case TerminateChild(name) =>
      terminateActor(name)

    case Terminated(x) => 
      println("Received termination confirmation: " + x)
      newActor(ChildName)

    case _ => println("Unexpected message.")
  }

  override def postStop = {
    println("Supervisor called postStop().")
  }
}

How to periodically execute an AKKA Actor's routine?

1 Answers