Commit 823b722f authored by Dominik Charousset's avatar Dominik Charousset

fixed scala benchmark

parent 98e2a20a
...@@ -4,6 +4,10 @@ import scala.actors.{Actor, AbstractActor, OutputChannel} ...@@ -4,6 +4,10 @@ import scala.actors.{Actor, AbstractActor, OutputChannel}
import scala.actors.remote.RemoteActor import scala.actors.remote.RemoteActor
import scala.actors.remote.RemoteActor._ import scala.actors.remote.RemoteActor._
import scala.actors.remote.Node import scala.actors.remote.Node
import scala.actors.TIMEOUT
import scala.annotation.tailrec
import com.typesafe.config.ConfigFactory import com.typesafe.config.ConfigFactory
import Console.println import Console.println
...@@ -11,9 +15,10 @@ import Console.println ...@@ -11,9 +15,10 @@ import Console.println
case object Done case object Done
case object OkTimeout case object OkTimeout
case class Error(msg: String, token: String) case class Error(msg: String, token: String)
case class KickOff(value: Int)
case class Ping(value: Int) case class Ping(value: Int)
case class Pong(value: Int) case class Pong(value: Int)
case class KickOff(value: Int)
case class Ok(token: String) case class Ok(token: String)
case class AddPong(path: String, token: String) case class AddPong(path: String, token: String)
...@@ -34,18 +39,28 @@ object global { ...@@ -34,18 +39,28 @@ object global {
val latch = new java.util.concurrent.CountDownLatch(1) val latch = new java.util.concurrent.CountDownLatch(1)
} }
class PingActor(parent: OutputChannel[Any], pong: OutputChannel[Any]) extends Actor { class PingActor(parent: OutputChannel[Any], pongs: List[OutputChannel[Any]]) extends Actor {
override def act() = react {
var left = pongs.length
private def recvLoop: Nothing = react {
case Pong(0) => { case Pong(0) => {
parent ! Done parent ! Done
if (left > 1) {
left -= 1
recvLoop
}
} }
case Pong(value) => { case Pong(value) => {
reply(Ping(value - 1)) sender ! Ping(value - 1)
act recvLoop
} }
}
override def act() = react {
case KickOff(value) => { case KickOff(value) => {
pong ! Ping(value) pongs.foreach(x => x ! Ping(value))
act recvLoop
} }
} }
} }
...@@ -53,144 +68,152 @@ class PingActor(parent: OutputChannel[Any], pong: OutputChannel[Any]) extends Ac ...@@ -53,144 +68,152 @@ class PingActor(parent: OutputChannel[Any], pong: OutputChannel[Any]) extends Ac
class DelayActor(msec: Long, parent: AbstractActor, msg: Any) extends Actor { class DelayActor(msec: Long, parent: AbstractActor, msg: Any) extends Actor {
override def act() { override def act() {
reactWithin(msec) { reactWithin(msec) {
case scala.actors.TIMEOUT => parent ! msg case TIMEOUT => parent ! msg
} }
} }
} }
class ServerActor(port: Int) extends Actor { class ServerActor(port: Int) extends Actor {
private var pongs: List[(String,OutputChannel[Any])] = Nil type PeerList = List[(String,OutputChannel[Any])]
private var pendingPongs: List[(String,AbstractActor,OutputChannel[Any],String)] = Nil type PendingPeerList = List[(String,AbstractActor,OutputChannel[Any],String)]
type PListPair = (PeerList, PendingPeerList)
final def msgLoop(): Nothing = react { private def recv(peers: PListPair): PListPair = receive {
case Ping(value) => { case Ping(value) => {
reply(Pong(value)) sender ! Pong(value)
msgLoop peers
} }
case Hello(token) => { case Hello(token) => {
sender ! Olleh(token) sender ! Olleh(token)
msgLoop peers
} }
case Olleh(token) => { case Olleh(token) => {
pendingPongs.filter(x => x match { peers._2.find(x => x._4 == token) match {
case (path, pong, ping, `token`) => { case Some((path, pong, ping, _)) => {
ping ! Ok(token) ping ! Ok(token)
pongs = (path, pong) :: pongs //println("recv[" + peers + "]: received Olleh from " + path)
true ((path, pong) :: peers._1, peers._2.filterNot(y => y._4 == token))
} }
case _ => false case None => {
}) peers
msgLoop }
}
} }
case PongDidNotRespond(pong) => { case PongDidNotRespond(pong) => {
pendingPongs.filter(x => x match { peers._2.find(x => x._2 == pong) match {
case (_, `pong`, ping, token) => { case Some ((path, _, ping, token)) => {
ping ! Error(pong + " did not respond within 5 seconds", token) //println("recv[" + peers + "]: " + path + " did not respond")
true ping ! Error(path + " did not respond within 5 seconds", token)
(peers._1, peers._2.filterNot(y => y._2 == pong))
} }
case _ => false case None => {
}) peers
msgLoop }
}
} }
case AddPong(path, token) => { case AddPong(path, token) => {
if (pongs.exists(x => x._1 == path)) { //println("received AddPong(" + path + ", " + token + ")")
if (peers._1.exists(x => x._1 == path)) {
sender ! Ok(token) sender ! Ok(token)
//println("recv[" + peers + "]: " + path + " cached (replied 'Ok')")
peers
} }
else { else {
try { try {
path.split(":") match { path.split(":") match {
case Array(node, port) => { case Array(node, port) => {
val pong = select(new Node(node, port.toInt), 'Pong) val pong = select(new Node(node, port.toInt), 'Pong)
(new DelayActor(5000, Actor.self, PongDidNotRespond(pong))).start (new DelayActor(5000, Actor.self, PongDidNotRespond(pong))).start
pendingPongs = (path, pong, sender, token) :: pendingPongs
pong ! Hello(token) pong ! Hello(token)
//println("recv[" + peers + "]: sent 'Hello' to " + path)
(peers._1, (path, pong, sender, token) :: peers._2)
} }
} }
} }
catch { catch {
case e => { case e => {
// catches match error and integer conversion failure // catches match error and integer conversion failure
reply(Error(e.toString, token)) sender ! Error(e.toString, token)
peers
} }
} }
} }
msgLoop
} }
case KickOff(value) => { case KickOff(value) => {
val client = sender (new PingActor(sender, peers._1.map(x => x._2))).start ! KickOff(value)
pongs.foreach(x => (new PingActor(client, x._2)).start() ! KickOff(value)) //println("recv[" + peers + "]: KickOff(" + value + ")")
msgLoop peers
} }
} }
@tailrec private def recvLoop(peers: PListPair): Nothing = {
recvLoop(recv(peers))
}
override def act() { override def act() {
RemoteActor.classLoader = getClass().getClassLoader RemoteActor.classLoader = getClass().getClassLoader
alive(port) alive(port)
register('Pong, Actor.self) register('Pong, Actor.self)
msgLoop recvLoop((Nil, Nil))
} }
} }
class ClientActor extends Actor { class ClientActor(pongPaths: List[RemoteActorPath], numPings: Int) extends Actor {
override def act() = {
private var left: Int = 0 RemoteActor.classLoader = getClass().getClassLoader
private var numPings: Int = 0 val pongs = pongPaths.map(x => {
private var pongs: List[AbstractActor] = Nil val pong = select(new Node(x.host, x.port), 'Pong)
private var tokens: List[String] = Nil pongPaths.foreach(y => {
if (x != y) {
def collectDoneMessages(): Nothing = react { val token = x.uri + " -> " + y.uri
case Done => { pong ! AddPong(y.uri, token)
if (left > 1) { }
left -= 1 })
collectDoneMessages pong
})
def receiveOkMessage(receivedTokens: List[String]): List[String] = {
receiveWithin(10*1000) {
case Ok(token) => {
token :: receivedTokens
}
case TIMEOUT => {
throw new RuntimeException("no Ok within 10sec.\nreceived tokens:\n" + receivedTokens.reduceLeft(_ + "\n" + _))
}
} }
} }
} def collectOkMessages(left: Int, receivedTokens: List[String]): Unit = {
if (left > 1) collectOkMessages(left - 1, receiveOkMessage(receivedTokens))
def collectOkMessages(): Nothing = react {
case Ok(token) => {
tokens = token :: tokens
if (tokens.length == left) {
pongs.foreach(x => x ! KickOff(numPings))
collectDoneMessages
} else {
collectOkMessages
}
} }
case NoTokenReceived(token) => { collectOkMessages(pongs.length * (pongs.length - 1), Nil)
//println("NoTokenReceived("+token+")") // collect ok messages
if (!tokens.contains(token)) { var receivedTokens: List[String] = Nil
println("Error: " + token + " did not respond within 10 seconds") for (_ <- 1 until (pongs.length * (pongs.length - 1))) {
receiveWithin(10*1000) {
case Ok(token) => {
receivedTokens = token :: receivedTokens
}
case TIMEOUT => {
println("no Ok within 10sec.")
println("received tokens: " + receivedTokens)
System.exit(1)
}
} }
} }
case Error(what, token) => { // kickoff
println("Error [from " + token + "]: " + what) pongs.foreach(x => x ! KickOff(numPings))
} // collect done messages
} for (_ <- 1 until (pongs.length * (pongs.length - 1))) {
receiveWithin(30*60*1000) {
override def act() = { case Done => {
RemoteActor.classLoader = getClass().getClassLoader() }
react { case TIMEOUT => {
case RunClient(pongPaths, numPings) => { println("no Done within 30min")
this.numPings = numPings System.exit(1)
pongs = pongPaths.map(x => { }
//println("select pong ...")
val pong = select(new Node(x.host, x.port), 'Pong)
pongPaths.foreach(y => if (x != y) {
//println("send AddPong")
pong ! AddPong(y.uri, x.uri)
//println("spawn DelayActor")
(new DelayActor(10000, Actor.self, NoTokenReceived(x.uri))).start
})
pong
})
left = pongs.length * (pongs.length - 1)
collectOkMessages
} }
} }
} }
} }
class PingAkkaActor(parent: AkkaActorRef, pong: AkkaActorRef) extends AkkaActor { class PingAkkaActor(parent: AkkaActorRef, pong: AkkaActorRef) extends AkkaActor {
...@@ -372,7 +395,8 @@ object DistributedClientApp { ...@@ -372,7 +395,8 @@ object DistributedClientApp {
} }
case Nil => numPings match { case Nil => numPings match {
case Some(x) => { case Some(x) => {
(new ClientActor).start ! RunClient(pongs, x) if (pongs.length < 2) throw new RuntimeException("at least two hosts required")
(new ClientActor(pongs, x)).start
} }
case None => { case None => {
throw new RuntimeException("no \"num_pings\" found") throw new RuntimeException("no \"num_pings\" found")
......
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