raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Merge branch 'master' of github.com:jboner/akka

Browse source
This commit is contained in:
Viktor Klang 2011-12-14 02:39:22 +01:00
commit a9fe796eae
4 changed files with 49 additions and 198 deletions
Download patch file Download diff file Expand all files Collapse all files

132
akka-docs/intro/code/tutorials/first/Pi.scala
View file

@ -1,132 +0,0 @@
// /**
// * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
// */
// //#imports
// package akka.tutorial.first.scala
// import akka.actor.{ Actor, ActorSystem, PoisonPill }
// import akka.routing.Routing.Broadcast
// import akka.routing.{ RoutedProps, Routing }
// import java.util.concurrent.CountDownLatch
// //#imports
// //#system
// object Pi extends App {
// val system = ActorSystem()
// calculate(nrOfWorkers = 4, nrOfElements = 10000, nrOfMessages = 10000)
// //#actors-and-messages
// // ====================
// // ===== Messages =====
// // ====================
// //#messages
// sealed trait PiMessage
// case object Calculate extends PiMessage
// case class Work(start: Int, nrOfElements: Int) extends PiMessage
// case class Result(value: Double) extends PiMessage
// //#messages
// // ==================
// // ===== Worker =====
// // ==================
// //#worker
// class Worker extends Actor {
// // define the work
// //#calculatePiFor
// def calculatePiFor(start: Int, nrOfElements: Int): Double = {
// var acc = 0.0
// for (i ← start until (start + nrOfElements))
// acc += 4.0 * (1 - (i % 2) * 2) / (2 * i + 1)
// acc
// }
// //#calculatePiFor
// def receive = {
// case Work(start, nrOfElements) ⇒ sender ! Result(calculatePiFor(start, nrOfElements)) // perform the work
// }
// }
// //#worker
// // ==================
// // ===== Master =====
// // ==================
// //#master
// class Master(nrOfWorkers: Int, nrOfMessages: Int, nrOfElements: Int, latch: CountDownLatch) extends Actor {
// var pi: Double = _
// var nrOfResults: Int = _
// var start: Long = _
// //#create-workers
// // create the workers
// val workers = Vector.fill(nrOfWorkers)(system.actorOf(Props[Worker])
// // wrap them with a load-balancing router
// val router = system.actorOf(RoutedProps().withRoundRobinRouter.withLocalConnections(workers), "pi")
// //#create-workers
// //#master-receive
// // message handler
// def receive = {
// //#handle-messages
// case Calculate ⇒
// // schedule work
// for (i ← 0 until nrOfMessages) router ! Work(i * nrOfElements, nrOfElements)
// // send a PoisonPill to all workers telling them to shut down themselves
// router ! Broadcast(PoisonPill)
// // send a PoisonPill to the router, telling him to shut himself down
// router ! PoisonPill
// case Result(value) ⇒
// // handle result from the worker
// pi += value
// nrOfResults += 1
// if (nrOfResults == nrOfMessages) self.stop()
// //#handle-messages
// }
// //#master-receive
// override def preStart() {
// start = System.currentTimeMillis
// }
// override def postStop() {
// // tell the world that the calculation is complete
// println(
// "\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis"
// .format(pi, (System.currentTimeMillis - start)))
// latch.countDown()
// }
// }
// //#master
// //#actors-and-messages
// // ==================
// // ===== Run it =====
// // ==================
// def calculate(nrOfWorkers: Int, nrOfElements: Int, nrOfMessages: Int) {
// // this latch is only plumbing to know when the calculation is completed
// val latch = new CountDownLatch(1)
// // create the master
// val master = system.actorOf(Props(new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
// // start the calculation
// master ! Calculate
// // wait for master to shut down
// latch.await()
// }
// }
// //#system

56
akka-tutorials/akka-tutorial-first/src/main/java/akka/tutorial/first/java/Pi.java
View file

@ -4,17 +4,13 @@
package akka.tutorial.first.java; package akka.tutorial.first.java;
import akka.actor.Props; //#imports
import akka.actor.ActorRef; import akka.actor.*;
import akka.actor.ActorSystem;
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
import akka.routing.RoundRobinRouter; import akka.routing.RoundRobinRouter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
//#imports
//#app
public class Pi { public class Pi {
public static void main(String[] args) throws Exception { public static void main(String[] args) throws Exception {
@ -22,9 +18,8 @@ public class Pi {
pi.calculate(4, 10000, 10000); pi.calculate(4, 10000, 10000);
} }
// ==================== //#actors-and-messages
// ===== Messages ===== //#messages
// ====================
static class Calculate { static class Calculate {
} }
@ -57,13 +52,12 @@ public class Pi {
return value; return value;
} }
} }
//#messages
// ================== //#worker
// ===== Worker =====
// ==================
public static class Worker extends UntypedActor { public static class Worker extends UntypedActor {
// define the work //#calculatePiFor
private double calculatePiFor(int start, int nrOfElements) { private double calculatePiFor(int start, int nrOfElements) {
double acc = 0.0; double acc = 0.0;
for (int i = start * nrOfElements; i <= ((start + 1) * nrOfElements - 1); i++) { for (int i = start * nrOfElements; i <= ((start + 1) * nrOfElements - 1); i++) {
@ -71,25 +65,22 @@ public class Pi {
} }
return acc; return acc;
} }
//#calculatePiFor
// message handler
public void onReceive(Object message) { public void onReceive(Object message) {
if (message instanceof Work) { if (message instanceof Work) {
Work work = (Work) message; Work work = (Work) message;
// perform the work
double result = calculatePiFor(work.getStart(), work.getNrOfElements()); double result = calculatePiFor(work.getStart(), work.getNrOfElements());
// reply with the result
getSender().tell(new Result(result)); getSender().tell(new Result(result));
} else throw new IllegalArgumentException("Unknown message [" + message + "]"); } else throw new IllegalArgumentException("Unknown message [" + message + "]");
} }
} }
//#worker
// ================== //#master
// ===== Master =====
// ==================
public static class Master extends UntypedActor { public static class Master extends UntypedActor {
private final int nrOfMessages; private final int nrOfMessages;
private final int nrOfElements; private final int nrOfElements;
@ -106,28 +97,27 @@ public class Pi {
this.nrOfElements = nrOfElements; this.nrOfElements = nrOfElements;
this.latch = latch; this.latch = latch;
router = this.getContext().actorOf(new Props().withCreator(Worker.class).withRouter(new RoundRobinRouter(5)), "pi"); //#create-router
router = this.getContext().actorOf(new Props().withCreator(Worker.class).withRouter(new RoundRobinRouter(nrOfWorkers)), "pi");
//#create-router
} }
// message handler //#master-receive
public void onReceive(Object message) { public void onReceive(Object message) {
//#handle-messages
if (message instanceof Calculate) { if (message instanceof Calculate) {
// schedule work
for (int start = 0; start < nrOfMessages; start++) { for (int start = 0; start < nrOfMessages; start++) {
router.tell(new Work(start, nrOfElements), getSelf()); router.tell(new Work(start, nrOfElements), getSelf());
} }
} else if (message instanceof Result) { } else if (message instanceof Result) {
// handle result from the worker
Result result = (Result) message; Result result = (Result) message;
pi += result.getValue(); pi += result.getValue();
nrOfResults += 1; nrOfResults += 1;
if (nrOfResults == nrOfMessages) getSelf().stop(); if (nrOfResults == nrOfMessages) getSelf().stop();
} else throw new IllegalArgumentException("Unknown message [" + message + "]"); } else throw new IllegalArgumentException("Unknown message [" + message + "]");
//#handle-messages
} }
//#master-receive
@Override @Override
public void preStart() { public void preStart() {
@ -136,19 +126,18 @@ public class Pi {
@Override @Override
public void postStop() { public void postStop() {
// tell the world that the calculation is complete
System.out.println(String.format( System.out.println(String.format(
"\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis", "\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis",
pi, (System.currentTimeMillis() - start))); pi, (System.currentTimeMillis() - start)));
latch.countDown(); latch.countDown();
} }
} }
//#master
//#actors-and-messages
// ==================
// ===== Run it =====
// ==================
public void calculate(final int nrOfWorkers, final int nrOfElements, final int nrOfMessages) public void calculate(final int nrOfWorkers, final int nrOfElements, final int nrOfMessages)
throws Exception { throws Exception {
// Create an Akka system
final ActorSystem system = ActorSystem.create(); final ActorSystem system = ActorSystem.create();
// this latch is only plumbing to know when the calculation is completed // this latch is only plumbing to know when the calculation is completed
@ -171,3 +160,4 @@ public class Pi {
system.stop(); system.stop();
} }
} }
//#app

49
akka-tutorials/akka-tutorial-first/src/main/scala/Pi.scala
View file

@ -3,49 +3,45 @@
*/ */
package akka.tutorial.first.scala package akka.tutorial.first.scala
//#imports
import java.util.concurrent.CountDownLatch import java.util.concurrent.CountDownLatch
import akka.actor._ import akka.actor._
import akka.routing._ import akka.routing._
import com.typesafe.config.{ ConfigFactory, Config } //#imports
//#app
object Pi extends App { object Pi extends App {
// Initiate the calculation
calculate(nrOfWorkers = 4, nrOfElements = 10000, nrOfMessages = 10000) calculate(nrOfWorkers = 4, nrOfElements = 10000, nrOfMessages = 10000)
// ==================== //#actors-and-messages
// ===== Messages ===== //#messages
// ====================
sealed trait PiMessage sealed trait PiMessage
case object Calculate extends PiMessage case object Calculate extends PiMessage
case class Work(start: Int, nrOfElements: Int) extends PiMessage case class Work(start: Int, nrOfElements: Int) extends PiMessage
case class Result(value: Double) extends PiMessage case class Result(value: Double) extends PiMessage
//#messages
// ================== //#worker
// ===== Worker =====
// ==================
class Worker extends Actor { class Worker extends Actor {
// define the work //#calculatePiFor
def calculatePiFor(start: Int, nrOfElements: Int): Double = { def calculatePiFor(start: Int, nrOfElements: Int): Double = {
var acc = 0.0 var acc = 0.0
for (i ← start until (start + nrOfElements)) for (i ← start until (start + nrOfElements))
acc += 4.0 * (1 - (i % 2) * 2) / (2 * i + 1) acc += 4.0 * (1 - (i % 2) * 2) / (2 * i + 1)
acc acc
} }
//#calculatePiFor
def receive = { def receive = {
case Work(start, nrOfElements) ⇒ case Work(start, nrOfElements) ⇒
sender ! Result(calculatePiFor(start, nrOfElements)) // perform the work sender ! Result(calculatePiFor(start, nrOfElements)) // perform the work
} }
} }
//#worker
// ================== //#master
// ===== Master =====
// ==================
class Master(nrOfWorkers: Int, nrOfMessages: Int, nrOfElements: Int, latch: CountDownLatch) class Master(nrOfWorkers: Int, nrOfMessages: Int, nrOfElements: Int, latch: CountDownLatch)
extends Actor { extends Actor {
@ -53,40 +49,40 @@ object Pi extends App {
var nrOfResults: Int = _ var nrOfResults: Int = _
var start: Long = _ var start: Long = _
// create a round robin router for the workers //#create-router
val router = context.actorOf(Props(new Worker).withRouter(RoundRobinRouter(nrOfInstances = 5)), "pi") val router = context.actorOf(Props(new Worker).withRouter(RoundRobinRouter(nrOfInstances = nrOfWorkers)), "pi")
//#create-router
// message handler //#master-receive
def receive = { def receive = {
//#handle-messages
case Calculate ⇒ case Calculate ⇒
// schedule work
for (i ← 0 until nrOfMessages) router ! Work(i * nrOfElements, nrOfElements) for (i ← 0 until nrOfMessages) router ! Work(i * nrOfElements, nrOfElements)
case Result(value) ⇒ case Result(value) ⇒
// handle result from the worker
pi += value pi += value
nrOfResults += 1 nrOfResults += 1
// Stops this actor and all its supervised children
// Stop this actor and all its supervised children
if (nrOfResults == nrOfMessages) self.stop() if (nrOfResults == nrOfMessages) self.stop()
//#handle-messages
} }
//#master-receive
override def preStart() { override def preStart() {
start = System.currentTimeMillis start = System.currentTimeMillis
} }
override def postStop() { override def postStop() {
// tell the world that the calculation is complete
println( println(
"\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis" "\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis"
.format(pi, (System.currentTimeMillis - start))) .format(pi, (System.currentTimeMillis - start)))
latch.countDown() latch.countDown()
} }
} }
//#master
//#actors-and-messages
// ==================
// ===== Run it =====
// ==================
def calculate(nrOfWorkers: Int, nrOfElements: Int, nrOfMessages: Int) { def calculate(nrOfWorkers: Int, nrOfElements: Int, nrOfMessages: Int) {
// Create an Akka system
val system = ActorSystem("PiSystem") val system = ActorSystem("PiSystem")
// this latch is only plumbing to know when the calculation is completed // this latch is only plumbing to know when the calculation is completed
@ -105,3 +101,4 @@ object Pi extends App {
system.stop() system.stop()
} }
} }
//#app

10
project/Dist.scala
View file

@ -66,9 +66,9 @@ object Dist {
val libAkka = lib / "akka" val libAkka = lib / "akka"
val src = base / "src" / "akka" val src = base / "src" / "akka"
IO.delete(unzipped) IO.delete(unzipped)
//createStructure(doc, docJars, lib, src) // TODO: re-enable bin and config dirs, and add deploy dir, when akka-kernel is enabled
copyFilesTo(scripts, bin, setExecutable = true) //copyFilesTo(scripts, bin, setExecutable = true)
IO.copyDirectory(configSources, config) //IO.copyDirectory(configSources, config)
IO.copyDirectory(allSources.api, api) IO.copyDirectory(allSources.api, api)
IO.copyDirectory(allSources.docs, docs) IO.copyDirectory(allSources.docs, docs)
copyFilesTo(allSources.docJars, docJars) copyFilesTo(allSources.docJars, docJars)
@ -83,10 +83,6 @@ object Dist {
} }
} }
def createStructure(dirs: File*): Unit = {
dirs foreach IO.createDirectory
}
def copyFilesTo(files: Seq[File], dir: File, setExecutable: Boolean = false): Unit = { def copyFilesTo(files: Seq[File], dir: File, setExecutable: Boolean = false): Unit = {
IO.createDirectory(dir) IO.createDirectory(dir)
for (file <- files) { for (file <- files) {