Take the 2-minute tour ×
Code Review Stack Exchange is a question and answer site for peer programmer code reviews. It's 100% free, no registration required.

I'm wondering if there is anything i'm overlooking in the code below. I've taken the Pi calculation example, simplified it and converted it to Akka 2.1.

More specifically, i use the ask pattern to replace the Listener actor.

So is this

  1. A proper and valid conversion?
  2. A reasonable approach?
  3. And is there a better approach to waiting for a result from an optionally distributed parallel reduction of a collection? In other words, is this an Akka equivalent of collection.par.map(work).par.reduce(result) or is there a better way?
object Work extends App {
  // This is just some collection of data to process (possibly remotely).
  val collection = 1 to 10000
  val start = System.currentTimeMillis
  val system = ActorSystem(f"Work${start}%08x${hashCode}%04x")
  // The collection is converted into a stream since it could
  // potentially be huge.
  val master = system.actorOf(Props(new Master(collection.toStream)), name = "master")

  // This replaces what the Listener was doing, i think.
  implicit val timeout = Timeout(12.seconds)
  val result = Await.result(master ? Run, timeout.duration)
  println(s"\n${result}")
  system.shutdown()

  sealed trait WorkMessage
  case object Run extends WorkMessage
  case class Work(start: Int) extends WorkMessage
  case class Result(value: Double) extends WorkMessage

  sealed trait ControlMessage
  case class Complete(elapsed:Long) extends ControlMessage

  class Worker extends Actor {
    def doWork(element: Int): Double = {
      print(s" ${element}")
      return element.toDouble
    }
    def receive = {
      case Work(element) ⇒
        sender ! Result(doWork(element))
    }
  }

  class Master(work: Seq[Int]) extends Actor {
    val nrOfCPUs = Runtime.getRuntime().availableProcessors();
    val nrOfWorkers = nrOfCPUs * 4
    val workSize = work.size
    var nrOfResults: Int = _
    val workerRouter = context.actorOf(
      Props[Worker].withRouter(RoundRobinRouter(nrOfWorkers)), name = "workerRouter")

    var done: ActorRef = _
    def receive = {
      case Run ⇒
        // Save the sender for responding to the ask once all the work is processed.
        done = sender
        work.foreach(element ⇒ workerRouter ! Work(element))
      case Result(value) ⇒
        nrOfResults += 1
        if (nrOfResults >= workSize) {
          done ! Complete(System.currentTimeMillis - start)
          context.stop(self)
        }
    }

  }

}

Here's a gist with the full code and a build.sbt to try:

curl -ks https://gist.github.com/nicerobot/4751388/raw/run.sh | sh
share|improve this question

1 Answer 1

I'm not an akka expert but I did used akka in a project so these are just some of my observations:

  1. result is blocking.

    val result = Await.result(master ? Run, timeout.duration)
    

    On case the master takes longer due to load or whatever, the pending result will exceed the timeout. Try async ask directly.

    val result ask(actor, msg).mapTo[String]
    
  2. In order to remain async you may carry on with an onComplete as Future callback:

    result onComplete {
        case Success(result) => doSomethingOnSuccess(result)
        case Failure(failure) => doSomethingOnFailure(failure)
    }
    

    In this case no operation ever blocks the running thread. Read the Akka docs for more details

And is there a better approach to waiting for a result from an optionally distributed parallel reduction of a collection? In other words, is this an Akka equivalent of collection.par.map(work).par.reduce(result) or is there a better way?

Operating on distributed data-sets is much better done on an in-memory data processing engine. Using the distributed word count as example, in Spark you do:

   file.flatMap(line => line.split(" "))
    .map(word => (word, 1))
    .reduceByKey(_ + _) 

Actors are made for message processing such as Twitter or WhatsApp but for computing tasks that must be fast, you use either MapReduce (Hadoop) for offline crunching of Spark/Storm for real-time processing. Hope that answers your question.

share|improve this answer

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

Not the answer you're looking for? Browse other questions tagged or ask your own question.