From 384332da392bc3fb0cbc71a9b2051da1073fbc83 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jonas=20Bone=CC=81r?= Date: Fri, 1 Apr 2011 14:48:04 +0200 Subject: [PATCH] removed JARs added by mistake --- .../src/main/scala/Pi.scala | 57 +++++++++++-------- 1 file changed, 34 insertions(+), 23 deletions(-) diff --git a/akka-tutorials/akka-tutorial-pi-sbt/src/main/scala/Pi.scala b/akka-tutorials/akka-tutorial-pi-sbt/src/main/scala/Pi.scala index fb7dcfcc93..787842e59e 100644 --- a/akka-tutorials/akka-tutorial-pi-sbt/src/main/scala/Pi.scala +++ b/akka-tutorials/akka-tutorial-pi-sbt/src/main/scala/Pi.scala @@ -38,7 +38,7 @@ object Main extends App { object Pi { val nrOfWorkers = 4 val nrOfMessages = 10000 - val lengthOfCalculationRange = 10000 + val nrOfElements = 10000 // ===== Messages ===== sealed trait PiMessage @@ -53,48 +53,59 @@ object Pi { } // ===== Master ===== - class Master(nrOfMessages: Int, latch: CountDownLatch) extends Actor { + class Master(latch: CountDownLatch) extends Actor { var pi: Double = _ - var count: Int = _ + var nrOfResults: Int = _ var start: Long = _ + // create the workers + val workers = { + val ws = new Array[ActorRef](nrOfWorkers) + for (i <- 0 until nrOfWorkers) ws(i) = actorOf[Worker].start + ws + } + + // wrap them with a load-balancing router + val router = Routing.loadBalancerActor(CyclicIterator(workers)).start + def receive = { + case Calculate(nrOfMessages, nrOfElements) => + // define the work + val fun = (i: Int) => { + val range = (i * nrOfElements) to ((i + 1) * nrOfElements - 1) + val results = for (j <- range) yield (4 * math.pow(-1, j) / (2 * j + 1)) + results.sum + } + // schedule work + for (arg <- 0 until nrOfMessages) router ! Work(arg, fun) + + // send a PoisonPill to all workers telling them to shut down themselves + router broadcast PoisonPill + case Result(value) => pi += value - count += 1 - if (count == nrOfMessages) self.stop + nrOfResults += 1 + if (nrOfResults == nrOfMessages) self.stop } override def preStart = start = now override def postStop = { EventHandler.info(this, "\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis".format(pi, (now - start))) - Actor.registry.shutdownAll // shut down all workers - latch.countDown + latch.nrOfResultsDown } } def calculate = { - val latch = new CountDownLatch(1) + val latch = new nrOfResultsDownLatch(1) // create the master - val master = actorOf(new Master(nrOfMessages, latch)).start + val master = actorOf(new Master(latch)).start - // the master ref is also the 'implicit sender' that the workers should reply to - implicit val replyTo = Option(master) - - // create the workers - val workers = new Array[ActorRef](nrOfWorkers) - for (i <- 0 until nrOfWorkers) workers(i) = actorOf[Worker].start - - // wrap them with a load-balancing router - val router = Routing.loadBalancerActor(CyclicIterator(workers)).start - - val fun = (x: Int) => (for (k <- (x * lengthOfCalculationRange) to ((x + 1) * lengthOfCalculationRange - 1)) yield (4 * math.pow(-1, k) / (2 * k + 1))).sum - - // schedule work - for (arg <- 0 until nrOfMessages) router ! Work(arg, fun) + // start the calculation + master ! Calculate(nrOfMessages, nrOfElements) + // wait for master to shut down latch.await } }