CODEWITHSUNDEEP
algorithmscalaconcurrencyakkaactor
Zim-Zam O'Pootertoot
Zim-Zam O'Pootertoot

Reputation: 18158

How do I determine that all actors have received a broadcast message

I have a single ActorA that reads from an input stream and sends messages to a group of ActorB's. When ActorA reaches the end of the input stream it cleans up its resources, broadcasts a Done message to the ActorB's, and shuts itself down.

I have approx 12 ActorB's that send messages to a group of ActorC's. When an ActorB receives a Done message from ActorA then it cleans up its resources and shuts itself down, with the exception of the last surviving ActorB which broadcasts a Done message to the ActorC's before it shuts itself down.

I have approx 24 ActorC's that send messages to a single ActorD. Similar to the ActorB's, when each ActorC gets a Done message it cleans up its resources and shuts itself down, with the exception of the last surviving ActorC which sends a Done message to ActorD.

When ActorD gets a Done message it cleans up its resources and shuts itself down.

Initially I had the ActorB's and ActorC's immediately propagate the Done message when they received it, but this might cause the ActorC's to shut down before all of the ActorB's have finished processing their queues; likewise the ActorD might shut down before the ActorC's have finished processing their queues.

My solution is to use an AtomicInteger that is shared among the ActorB's

class ActorB(private val actorCRouter: ActorRef,
             private val actorCount: AtomicInteger) extends Actor {
  private val init = {
    actorCount.incrementAndGet()
    ()
  }

  def receive = {
    case Done => {
      if(actorCount.decrementAndGet() == 0) {
        actorCRouter ! Broadcast(Done)
      }
      // clean up resources
      context.stop(self)
    }
  }
}

ActorC uses similar code, with each ActorC sharing an AtomicInteger.

At present all actors are initialized in a web service method, with the downstream ActorRef's passed in the upstream actors' constructors.

Is there a preferred way to do this, e.g. using calls to Akka methods instead of an AtomicInteger?

Edit: I'm considering the following as a possible alternative: when an actor receives a Done message it sets the receive timeout to 5 seconds (the program will take over an hour to run, so delaying cleanup/shutdown by a few seconds won't impact the performance); when the actor gets a ReceiveTimeout it broadcasts Done to the downstream actors, cleans up, and shuts down. (The routers for ActorB and ActorC are using a SmallestMailboxRouter)

class ActorB(private val actorCRouter: ActorRef) extends Actor {

  def receive = {
    case Done => {
      context.setReceiveTimeout(Duration.create(5, SECONDS))
    }

    case ReceiveTimeout => {
      actorCRouter ! Broadcast(Done)
      // clean up resources
      context.stop(self)
    }
  }
}

Upvotes: 2

Views: 182

Answers (1)

Mithun Ruikar
Mithun Ruikar

Reputation: 66

Sharing actorCount among related actors is not good thing to do. Actor should only be using its own state to handle messages. How about having ActorBCompletionHanlder actor for actor of type ActorB. All ActorB will have reference to ActorBCompletionHanlder actor. Every time ActorB receives Done message it can do necessay cleanup and simply pass done message to ActorBCompletionHanlder. ActorBCompletionHanlder will maintain state variale for maintaining counts. Everytime it receives done message it can simply update counter. As this is solely state variable for this actor no need to have it atomic and that way no need for any explicit locking. ActorBCompletionHanlder will send done message to ActorC once it receives last done message. This way sharing of activeCount is not among actors but only managed by ActorBCompletionHanlder. Same thing can be repeated for other types.

A-> B's -> BCompletionHanlder -> C's -> CCompletionHandler -> D

Other approach could be to have one monitoring actor for evey related group of actors. And using watch api and child terminated event on monitor you can chose to decide what to do once you receive last done message.

val child = context.actorOf(Props[ChildActor])
    context.watch(child)

    case Terminated(child) => {
        log.info(child + " Child actor terminated")
    }

Upvotes: 1

Related Questions