Commit 723689ae authored by Dominik Charousset's avatar Dominik Charousset

scala remote actors benchmark

parent 7fdfbfb4
import akka.actor.{ Props, Actor, ActorRef, ActorSystem } import akka.actor.{ Props, Actor => AkkaActor, ActorRef => AkkaActorRef, ActorSystem }
import scala.actors.{Actor, AbstractActor, OutputChannel}
import scala.actors.remote.RemoteActor
import scala.actors.remote.RemoteActor._
import scala.actors.remote.Node
import com.typesafe.config.ConfigFactory import com.typesafe.config.ConfigFactory
import Console.println import Console.println
...@@ -13,18 +19,134 @@ case class AddPong(path: String) ...@@ -13,18 +19,134 @@ case class AddPong(path: String)
case object Hello case object Hello
case object Olleh case object Olleh
case class PongDidNotRespond(pong: ActorRef, ping: ActorRef) case class PongDidNotRespond(pong: AkkaActorRef, ping: AkkaActorRef)
case class RunClient(pongs: List[ActorRef], numPings: Int) case class RemoteActorPath(uri: String, host: String, port: Int)
case class RunClient(pongs: List[RemoteActorPath], numPings: Int)
case class RunAkkaClient(pongs: List[AkkaActorRef], numPings: Int)
object global { object global {
val latch = new java.util.concurrent.CountDownLatch(1) val latch = new java.util.concurrent.CountDownLatch(1)
} }
class PingActor(parent: ActorRef, pong: ActorRef) extends Actor { class PingActor(parent: OutputChannel[Any], pong: AbstractActor) extends Actor {
override def act() = react {
case Pong(0) => {
parent ! Done
}
case Pong(value) => {
reply(Ping(value - 1))
act
}
case KickOff(value) => {
pong ! Ping(value)
act
}
}
}
class ServerActor(port: Int) extends Actor {
private var pongs: List[Pair[String,AbstractActor]] = Nil
final def msgLoop(): Nothing = react {
case Ping(value) => {
reply(Pong(value))
msgLoop
}
case AddPong(path) => {
if (pongs.exists(x => x._1 == path)) {
sender ! Ok
}
else {
try {
path.split(":") match {
case Array(node, port) => {
val pong = select(new Node(node, port.toInt), 'Pong)
sender ! Ok
pongs = Pair(path, pong) :: pongs
}
}
}
catch {
case e => {
// catches match error, connection error and
// integer conversion failure
reply(Error(e.toString))
}
}
}
msgLoop
}
case KickOff(value) => {
val client = sender
pongs.foreach(x => (new PingActor(client, x._2)).start() ! KickOff(value))
msgLoop
}
}
override def act() {
RemoteActor.classLoader = getClass().getClassLoader
alive(port)
register('Pong, Actor.self)
msgLoop
}
}
class ClientActor extends Actor {
private var left: Int = 0
private var numPings: Int = 0
private var pongs: List[AbstractActor] = Nil
def collectDoneMessages(): Nothing = react {
case Done => {
if (left > 1) {
left -= 1
collectDoneMessages
}
}
}
def collectOkMessages(): Nothing = react {
case Ok => {
if (left == 1) {
pongs.foreach(x => x ! KickOff(numPings))
left = pongs.length * (pongs.length - 1)
collectDoneMessages
} else {
left -= 1
collectOkMessages
}
}
case Error(what) => {
println("Error: " + what)
}
}
override def act() = {
RemoteActor.classLoader = getClass().getClassLoader()
react {
case RunClient(pongPaths, numPings) => {
this.numPings = numPings
pongs = pongPaths.map(x => {
val pong = select(new Node(x.host, x.port), 'Pong)
pong ! AddPong(x.uri)
pong
})
left = pongs.length * (pongs.length - 1)
collectOkMessages
}
}
}
}
class PingAkkaActor(parent: AkkaActorRef, pong: AkkaActorRef) extends AkkaActor {
def receive = { def receive = {
case Pong(0) => { case Pong(0) => {
//println(parent + " ! Done")
parent ! Done parent ! Done
context.stop(self) context.stop(self)
} }
...@@ -32,18 +154,16 @@ class PingActor(parent: ActorRef, pong: ActorRef) extends Actor { ...@@ -32,18 +154,16 @@ class PingActor(parent: ActorRef, pong: ActorRef) extends Actor {
sender ! Ping(value - 1) sender ! Ping(value - 1)
} }
case KickOff(value) => { case KickOff(value) => {
//println("PingActor::KickOff " + value)
pong ! Ping(value) pong ! Ping(value)
} }
} }
} }
class ServerActor(system: ActorSystem) extends Actor { class ServerAkkaActor(system: ActorSystem) extends AkkaActor {
var pongs = List[ActorRef]() import context.become
var pendingPongs = List[Pair[ActorRef, ActorRef]]()
def receive = { def recvLoop(pongs: List[AkkaActorRef], pendingPongs: List[Pair[AkkaActorRef, AkkaActorRef]]): Receive = {
case Ping(value) => { case Ping(value) => {
sender ! Pong(value) sender ! Pong(value)
} }
...@@ -52,8 +172,7 @@ class ServerActor(system: ActorSystem) extends Actor { ...@@ -52,8 +172,7 @@ class ServerActor(system: ActorSystem) extends Actor {
case Some((pong, ping)) => { case Some((pong, ping)) => {
println("added actor " + pong.path) println("added actor " + pong.path)
ping ! Ok ping ! Ok
pendingPongs = pendingPongs.filterNot(x => x._1 == sender) become(recvLoop(pong :: pongs, pendingPongs.filterNot(x => x._1 == sender)))
pongs = pong :: pongs
} }
case None => { case None => {
// operation already timed out // operation already timed out
...@@ -64,24 +183,25 @@ println("added actor " + pong.path) ...@@ -64,24 +183,25 @@ println("added actor " + pong.path)
pendingPongs.find(x => x._1 == sender) match { pendingPongs.find(x => x._1 == sender) match {
case Some(Pair(_, y)) => { case Some(Pair(_, y)) => {
ping ! Error(pong + " did not respond") ping ! Error(pong + " did not respond")
pendingPongs = pendingPongs.filterNot(x => x._1 == sender) become(recvLoop(pongs, pendingPongs.filterNot(x => x._1 == sender)))
} }
case None => { case None => {
// operation already succeeded // operation already succeeded (received Olleh)
} }
} }
} }
case AddPong(path) => { case AddPong(path) => {
if (pongs.exists((x) => x.path == path)) { if (pongs.exists((x) => x.path == path)) {
sender ! Ok sender ! Ok
} else { }
else {
import akka.util.duration._ import akka.util.duration._
println("try to add actor " + path) println("try to add actor " + path)
val pong = system.actorFor(path) val pong = system.actorFor(path)
// wait at most 5sec. for response // wait at most 5sec. for response
pong ! Hello pong ! Hello
pendingPongs = Pair(pong, sender) :: pendingPongs
system.scheduler.scheduleOnce(5 seconds, self, PongDidNotRespond(pong, sender)) system.scheduler.scheduleOnce(5 seconds, self, PongDidNotRespond(pong, sender))
become(recvLoop(pongs, Pair(pong, sender) :: pendingPongs))
//pong ! Hello //pong ! Hello
//pongs = pong :: pongs //pongs = pong :: pongs
//sender ! Ok //sender ! Ok
...@@ -90,16 +210,18 @@ println("try to add actor " + path) ...@@ -90,16 +210,18 @@ println("try to add actor " + path)
case KickOff(value) => { case KickOff(value) => {
val client = sender val client = sender
//println("KickOff(" + value + ") from " + client) //println("KickOff(" + value + ") from " + client)
pongs.foreach((x) => context.actorOf(Props(new PingActor(client, x))) ! KickOff(value)) pongs.foreach((x) => context.actorOf(Props(new PingAkkaActor(client, x))) ! KickOff(value))
} }
case Hello => { case Hello => {
sender ! Olleh sender ! Olleh
} }
} }
def receive = recvLoop(Nil, Nil)
} }
class ClientActor extends Actor { class ClientAkkaActor extends AkkaActor {
import context._ import context._
...@@ -115,12 +237,12 @@ class ClientActor extends Actor { ...@@ -115,12 +237,12 @@ class ClientActor extends Actor {
left = left - 1 left = left - 1
} }
} }
case x => { case _ => {
// ignore any other message // ignore any other message
} }
} }
def collectOkMessages(pongs: List[ActorRef], numPings: Int): Receive = { def collectOkMessages(pongs: List[AkkaActorRef], numPings: Int): Receive = {
case Ok => { case Ok => {
//println("Ok") //println("Ok")
if (left == 1) { if (left == 1) {
...@@ -142,7 +264,7 @@ class ClientActor extends Actor { ...@@ -142,7 +264,7 @@ class ClientActor extends Actor {
} }
def receive = { def receive = {
case RunClient(pongs, numPings) => { case RunAkkaClient(pongs, numPings) => {
import akka.util.duration._ import akka.util.duration._
pongs.foreach(x => pongs.foreach(y => if (x != y) { pongs.foreach(x => pongs.foreach(y => if (x != y) {
x ! AddPong(y.path.toString) x ! AddPong(y.path.toString)
...@@ -155,45 +277,86 @@ class ClientActor extends Actor { ...@@ -155,45 +277,86 @@ class ClientActor extends Actor {
} }
object DistributedClientApp { object DistributedClientApp {
def main(args: Array[String]): Unit = {
val system = ActorSystem("benchmark", ConfigFactory.load.getConfig("benchmark"))
var numPings: Int = 0
var pongs = List[ActorRef]()
val NumPings = "num_pings=([0-9]+)".r val NumPings = "num_pings=([0-9]+)".r
args.foreach(arg => arg match { val SimpleUri = "([0-9a-zA-Z\\.]+):([0-9]+)".r
case NumPings(num) => {
numPings = num.toInt def runAkka(system: ActorSystem, args: List[String], pongs: List[AkkaActorRef], numPings: Option[Int]): Unit = args match {
case NumPings(num) :: tail => numPings match {
case Some(x) => {
println("\"num_pings\" already defined, first value = " + x + ", second value = " + num)
} }
case _ => { case None => runAkka(system, tail, pongs, Some(num.toInt))
//println("add actor " + arg)
pongs = system.actorFor(arg) :: pongs
} }
}) case path :: tail => {
runAkka(system, tail, system.actorFor(path) :: pongs, numPings)
system.actorOf(Props[ClientActor]) ! RunClient(pongs, numPings) }
case Nil => numPings match {
case Some(x) => {
if (pongs isEmpty) throw new RuntimeException("No pong no fun")
system.actorOf(Props[ClientAkkaActor]) ! RunAkkaClient(pongs, x)
global.latch.await global.latch.await
system.shutdown system.shutdown
System.exit(0) System.exit(0)
} }
} case None => {
throw new RuntimeException("no \"num_pings\" found")
object DistributedServerApp { }
def main(args: Array[String]): Unit = { }
}
val sysName = if (args.length > 0) args(0) else "pongServer" def runRemoteActors(args: List[String], pongs: List[RemoteActorPath], numPings: Option[Int]): Unit = args match {
case NumPings(num) :: tail => numPings match {
case Some(x) => {
println("\"num_pings\" already defined, first value = " + x + ", second value = " + num)
}
case None => runRemoteActors(tail, pongs, Some(num.toInt))
}
case arg :: tail => arg match {
case SimpleUri(host, port) => {
runRemoteActors(tail, RemoteActorPath(arg, host, port.toInt) :: pongs, numPings)
}
case _ => {
throw new IllegalArgumentException("illegal argument: " + arg)
}
}
case Nil => numPings match {
case Some(x) => {
(new ClientActor).start ! RunClient(pongs, x)
}
case None => {
throw new RuntimeException("no \"num_pings\" found")
}
}
}
val system = ActorSystem(sysName, ConfigFactory.load.getConfig("pongServer")) def main(args: Array[String]): Unit = args.toList match {
case "akka" :: akkaArgs => {
val system = ActorSystem("benchmark", ConfigFactory.load.getConfig("benchmark"))
runAkka(system, akkaArgs, Nil, None)
}
case "remote_actors" :: tail => {
runRemoteActors(tail, Nil, None)
}
case Nil => {
println("usage: ...")
}
}
}
val pong = system.actorOf(Props(new ServerActor(system)), "pong") object DistributedServerApp {
//val subSystem = ActorSystem(sysName + "Client", ConfigFactory.load.getConfig("benchmark")) def usage = println("usage: (akka [configName]) | (remote_actors {port})")
//val pong = system.actorOf(Props(new ServerActor(subSystem)), "pong")
def main(args: Array[String]): Unit = args.toList match {
case "akka" :: tail if tail.length < 2 => {
val system = ActorSystem(if (tail.isEmpty) "pongServer" else tail.head, ConfigFactory.load.getConfig("pongServer"))
val pong = system.actorOf(Props(new ServerAkkaActor(system)), "pong")
}
case "remote_actors" :: port :: Nil => {
(new ServerActor(port.toInt)).start
}
case _ => usage
} }
} }
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment