Commit 9b4e5f3d authored by Dominik Charousset's avatar Dominik Charousset

fixed akka benchmark

parent 24dc87d0
...@@ -163,15 +163,19 @@ class ClientActor(pongPaths: List[RemoteActorPath], numPings: Int) extends Actor ...@@ -163,15 +163,19 @@ class ClientActor(pongPaths: List[RemoteActorPath], numPings: Int) extends Actor
} }
} }
class AkkaPingActor(parent: AkkaActorRef, pongs: List[AkkaActorRef]) extends AkkaActor { case class SetParent(parent: AkkaActorRef)
class AkkaPingActor(pongs: List[AkkaActorRef]) extends AkkaActor {
import context.become import context.become
private var parent: AkkaActorRef = null
private var left = pongs.length private var left = pongs.length
private def recvLoop: Receive = { private def recvLoop: Receive = {
case Pong(0) => { case Pong(0) => {
parent ! Done parent ! Done
//println(parent.toString + " ! Done")
if (left > 1) left -= 1 if (left > 1) left -= 1
else context.stop(self) else context.stop(self)
} }
...@@ -179,6 +183,7 @@ class AkkaPingActor(parent: AkkaActorRef, pongs: List[AkkaActorRef]) extends Akk ...@@ -179,6 +183,7 @@ class AkkaPingActor(parent: AkkaActorRef, pongs: List[AkkaActorRef]) extends Akk
} }
def receive = { def receive = {
case SetParent(p) => parent = p
case KickOff(value) => pongs.foreach(_ ! Ping(value)); become(recvLoop) case KickOff(value) => pongs.foreach(_ ! Ping(value)); become(recvLoop)
} }
} }
...@@ -194,12 +199,16 @@ class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPro ...@@ -194,12 +199,16 @@ class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPro
protected def reply(what: Any): Unit = sender ! what protected def reply(what: Any): Unit = sender ! what
protected def kickOff(peers: AkkaPeers, value: Int): Unit = { protected def kickOff(peers: AkkaPeers, value: Int): Unit = {
context.actorOf(Props(new AkkaPingActor(sender, peers.connected map (_.channel)))) ! KickOff(value) val ping = context.actorOf(Props(new AkkaPingActor(peers.connected map (_.channel))))
ping ! SetParent(sender)
ping ! KickOff(value)
//println("[" + peers + "]: KickOff(" + value + ")")
} }
protected def connectionEstablished(peers: AkkaPeers, x: Any): AkkaPeers = x match { protected def connectionEstablished(peers: AkkaPeers, x: Any): AkkaPeers = x match {
case PendingAkkaPeer(path, channel, client, token) => { case PendingAkkaPeer(path, channel, client, token) => {
client ! Ok(token) client ! Ok(token)
//println("connected to " + path)
AkkaPeers(AkkaPeer(path, channel) :: peers.connected, peers.pending filterNot (_.clientToken == token)) AkkaPeers(AkkaPeer(path, channel) :: peers.connected, peers.pending filterNot (_.clientToken == token))
} }
} }
...@@ -209,7 +218,7 @@ class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPro ...@@ -209,7 +218,7 @@ class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPro
channel ! Hello(token) channel ! Hello(token)
import akka.util.duration._ import akka.util.duration._
system.scheduler.scheduleOnce(5 seconds, self, AddPongTimeout(path, token)) system.scheduler.scheduleOnce(5 seconds, self, AddPongTimeout(path, token))
//println("recv[" + peers + "]: sent 'Hello' to " + path) //println("[" + peers + "]: sent 'Hello' to " + path)
AkkaPeers(peers.connected, PendingAkkaPeer(path, channel, sender, token) :: peers.pending) AkkaPeers(peers.connected, PendingAkkaPeer(path, channel, sender, token) :: peers.pending)
} }
...@@ -217,6 +226,7 @@ class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPro ...@@ -217,6 +226,7 @@ class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPro
peers.pending find (x => x.path == path && x.clientToken == token) match { peers.pending find (x => x.path == path && x.clientToken == token) match {
case Some(PendingAkkaPeer(_, channel, client, _)) => { case Some(PendingAkkaPeer(_, channel, client, _)) => {
client ! Error(path + " did not respond", token) client ! Error(path + " did not respond", token)
//println(path + " did not respond")
(true, AkkaPeers(peers.connected, peers.pending filterNot (x => x.path == path && x.clientToken == token))) (true, AkkaPeers(peers.connected, peers.pending filterNot (x => x.path == path && x.clientToken == token)))
} }
case None => (false, peers) case None => (false, peers)
...@@ -237,7 +247,7 @@ class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPro ...@@ -237,7 +247,7 @@ class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPro
} }
case class TokenTimeout(token: String) case class TokenTimeout(token: String)
case class RunAkkaClient(pongs: List[AkkaActorRef], numPings: Int) case class RunAkkaClient(paths: List[String], numPings: Int)
class AkkaClientActor(system: ActorSystem) extends AkkaActor { class AkkaClientActor(system: ActorSystem) extends AkkaActor {
...@@ -258,21 +268,23 @@ class AkkaClientActor(system: ActorSystem) extends AkkaActor { ...@@ -258,21 +268,23 @@ class AkkaClientActor(system: ActorSystem) extends AkkaActor {
} }
} }
def collectOkMessages(pongs: List[AkkaActorRef], tokens: List[String], numPings: Int): Receive = { def collectOkMessages(pongs: List[AkkaActorRef], left: Int, receivedTokens: List[String], numPings: Int): Receive = {
case Ok(token) => { case Ok(token) => {
//println("Ok") //println("Ok")
if (tokens.length + 1 == pongs.length) { if (left == 1) {
val left = pongs.length * (pongs.length - 1) //println("collected all Ok messages (wait for Done messages)")
pongs.foreach(x => x ! KickOff(numPings)) pongs foreach (_ ! KickOff(numPings))
become(collectDoneMessages(left)) become(collectDoneMessages(pongs.length * (pongs.length - 1)))
} }
else { else {
become(collectOkMessages(pongs, token :: tokens, numPings)) become(collectOkMessages(pongs, left - 1, token :: receivedTokens, numPings))
} }
} }
case TokenTimeout(token) => { case TokenTimeout(token) => {
if (!tokens.contains(token)) { if (!receivedTokens.contains(token)) {
println("Error: " + token + " did not reply within 10 seconds"); println("Error: " + token + " did not reply within 10 seconds")
global.latch.countDown
context.stop(self)
} }
} }
case Error(what, token) => { case Error(what, token) => {
...@@ -283,14 +295,20 @@ class AkkaClientActor(system: ActorSystem) extends AkkaActor { ...@@ -283,14 +295,20 @@ class AkkaClientActor(system: ActorSystem) extends AkkaActor {
} }
def receive = { def receive = {
case RunAkkaClient(pongs, numPings) => { case RunAkkaClient(paths, numPings) => {
//println("RunAkkaClient(" + paths.toString + ", " + numPings + ")")
import akka.util.duration._ import akka.util.duration._
pongs.foreach(x => pongs.foreach(y => if (x != y) { val pongs = paths map (x => {
val token = x.toString val pong = system.actorFor(x)
x ! AddPong(y.path.toString, token) paths foreach (y => if (x != y) {
system.scheduler.scheduleOnce(10 seconds, self, TokenTimeout(token)) val token = x + " -> " + y
})) pong ! AddPong(y, token)
become(collectOkMessages(pongs, Nil, numPings)) //println(x + " ! AddPong(" + y + ", " + token + ")")
system.scheduler.scheduleOnce(10 seconds, self, TokenTimeout(token))
})
pong
})
become(collectOkMessages(pongs, pongs.length * (pongs.length - 1), Nil, numPings))
} }
} }
} }
...@@ -323,7 +341,7 @@ object DistributedClientApp { ...@@ -323,7 +341,7 @@ object DistributedClientApp {
})) }))
case "akka" => run(args.toList.drop(1), Nil, None, ((paths, x) => { case "akka" => run(args.toList.drop(1), Nil, None, ((paths, x) => {
val system = ActorSystem("benchmark", ConfigFactory.load.getConfig("benchmark")) val system = ActorSystem("benchmark", ConfigFactory.load.getConfig("benchmark"))
system.actorOf(Props(new AkkaClientActor(system))) ! RunAkkaClient(paths map (system.actorFor(_)), x) system.actorOf(Props(new AkkaClientActor(system))) ! RunAkkaClient(paths, x)
global.latch.await global.latch.await
system.shutdown system.shutdown
System.exit(0) System.exit(0)
......
...@@ -3,11 +3,12 @@ benchmark { ...@@ -3,11 +3,12 @@ benchmark {
loglevel = ERROR loglevel = ERROR
actor.provider = "akka.remote.RemoteActorRefProvider" actor.provider = "akka.remote.RemoteActorRefProvider"
remote { remote {
transport = "akka.remote.netty.NettyRemoteTransport"
untrusted-mode = on untrusted-mode = on
remote-daemon-ack-timeout = 300s # remote-daemon-ack-timeout = 300s
netty { # netty {
connection-timeout = 1800s # connection-timeout = 1800s
} # }
} }
} }
} }
...@@ -18,13 +19,14 @@ pongServer { ...@@ -18,13 +19,14 @@ pongServer {
provider = "akka.remote.RemoteActorRefProvider" provider = "akka.remote.RemoteActorRefProvider"
} }
remote { remote {
transport = "akka.remote.netty.NettyRemoteTransport"
untrusted-mode = on untrusted-mode = on
remote-daemon-ack-timeout = 300s # remote-daemon-ack-timeout = 300s
netty { netty {
backoff-timeout = 0ms # backoff-timeout = 0ms
connection-timeout = 1800s connection-timeout = 1800s
read-timeout = 1800s # read-timeout = 1800s
write-timeout = 10s # write-timeout = 10s
all-timeout = 1800s all-timeout = 1800s
#hostname = "mobi10" #hostname = "mobi10"
port = 2244 port = 2244
......
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