Commit 66d357a8 authored by Matthias Vallentin's avatar Matthias Vallentin

Merge branch 'unstable' into topic/libprocess

parents 635412a2 f33696eb
import akka.actor.{ Props, Actor => AkkaActor, ActorRef => AkkaActorRef, ActorSystem }
import scala.actors.{Actor, AbstractActor, OutputChannel}
import scala.actors._
import scala.actors.Actor.self
import scala.actors.remote.RemoteActor
import scala.actors.remote.RemoteActor._
import scala.actors.remote.RemoteActor.{alive, select, register}
import scala.actors.remote.Node
import scala.actors.TIMEOUT
import akka.actor.{ Props, Actor => AkkaActor, ActorRef => AkkaActorRef, ActorSystem }
import com.typesafe.config.ConfigFactory
import scala.annotation.tailrec
import com.typesafe.config.ConfigFactory
import Console.println
case object Done
case object OkTimeout
case class Error(msg: String, token: String)
case class Ping(value: Int)
case class Pong(value: Int)
case class KickOff(value: Int)
case object Done
case class Ok(token: String)
case class AddPong(path: String, token: String)
case class Hello(token: String)
case class Olleh(token: String)
case class Error(msg: String, token: String)
case class AddPong(path: String, token: String)
case class NoTokenReceived(token: String)
case class PongDidNotRespond(pong: AbstractActor)
case class AkkaPongDidNotRespond(pong: AkkaActorRef, ping: AkkaActorRef)
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)
case class AddPongTimeout(path: String, token: String)
object global {
val latch = new java.util.concurrent.CountDownLatch(1)
}
trait ServerActorPrototype[T] {
protected def reply(what: Any): Unit
protected def kickOff(old: T, value: Int): Unit
protected def connectionEstablished(peers: T, pending: Any): T
protected def newPending(peers: T, path: String, token: String) : T
protected def handleTimeout(peers: T): (Boolean, T) = throw new RuntimeException("unsupported timeout")
protected def handleAddPongTimeout(peers: T, path: String, token: String): (Boolean, T) = throw new RuntimeException("unsupported timeout")
def recvFun(peers: T { def connected: List[{ def path: String }]; def pending: List[{ def clientToken: String }] }): PartialFunction[Any, (Boolean, T)] = {
case Ping(value) => reply(Pong(value)); (false, peers)
case Hello(token) => reply(Olleh(token)); (false, peers)
case Olleh(token) => peers.pending find (_.clientToken == token) match {
case Some(x) => (true, connectionEstablished(peers, x))
case None => (false, peers)
}
case AddPong(path, token) => {
//println("received AddPong(" + path + ", " + token + ")")
if (peers.connected exists (_.path == path)) {
reply(Ok(token))
//println("recv[" + peers + "]: " + path + " cached (replied 'Ok')")
(false, peers)
}
else {
try { (true, newPending(peers, path, token)) }
catch {
// catches match error and integer conversion failure
case e => reply(Error(e.toString, token)); (false, peers)
}
}
}
case KickOff(value) => kickOff(peers, value); (false, peers)
case AddPongTimeout(path, token) => handleAddPongTimeout(peers, path, token)
case TIMEOUT => handleTimeout(peers)
}
}
case class RemoteActorPath(uri: String, host: String, port: Int)
class PingActor(parent: OutputChannel[Any], pongs: List[OutputChannel[Any]]) extends Actor {
var left = pongs.length
private var left = pongs.length
private def recvLoop: Nothing = react {
case Pong(0) => {
......@@ -51,251 +79,179 @@ class PingActor(parent: OutputChannel[Any], pongs: List[OutputChannel[Any]]) ext
recvLoop
}
}
case Pong(value) => {
sender ! Ping(value - 1)
recvLoop
}
case Pong(value) => sender ! Ping(value - 1); recvLoop
}
override def act() = react {
case KickOff(value) => {
pongs.foreach(x => x ! Ping(value))
recvLoop
}
case KickOff(value) => pongs.foreach(_ ! Ping(value)); recvLoop
}
}
class DelayActor(msec: Long, parent: AbstractActor, msg: Any) extends Actor {
override def act() {
reactWithin(msec) {
case TIMEOUT => parent ! msg
}
}
}
case class Peer(path: String, channel: OutputChannel[Any])
case class PendingPeer(path: String, channel: OutputChannel[Any], client: OutputChannel[Any], clientToken: String)
case class Peers(connected: List[Peer], pending: List[PendingPeer])
class ServerActor(port: Int) extends Actor {
class ServerActor(port: Int) extends Actor with ServerActorPrototype[Peers] {
type PeerList = List[(String,OutputChannel[Any])]
type PendingPeerList = List[(String,AbstractActor,OutputChannel[Any],String)]
type PListPair = (PeerList, PendingPeerList)
//def reply(what: Any): Unit = sender ! what // inherited from ReplyReactor
private def recv(peers: PListPair): PListPair = receive {
case Ping(value) => {
sender ! Pong(value)
peers
}
case Hello(token) => {
sender ! Olleh(token)
peers
}
case Olleh(token) => {
peers._2.find(x => x._4 == token) match {
case Some((path, pong, ping, _)) => {
ping ! Ok(token)
//println("recv[" + peers + "]: received Olleh from " + path)
((path, pong) :: peers._1, peers._2.filterNot(y => y._4 == token))
}
case None => {
peers
}
}
}
case PongDidNotRespond(pong) => {
peers._2.find(x => x._2 == pong) match {
case Some ((path, _, ping, token)) => {
//println("recv[" + peers + "]: " + path + " did not respond")
ping ! Error(path + " did not respond within 5 seconds", token)
(peers._1, peers._2.filterNot(y => y._2 == pong))
}
case None => {
peers
}
}
}
case AddPong(path, token) => {
//println("received AddPong(" + path + ", " + token + ")")
if (peers._1.exists(x => x._1 == path)) {
sender ! Ok(token)
//println("recv[" + peers + "]: " + path + " cached (replied 'Ok')")
peers
}
else {
try {
path.split(":") match {
case Array(node, port) => {
val pong = select(new Node(node, port.toInt), 'Pong)
(new DelayActor(5000, Actor.self, PongDidNotRespond(pong))).start
pong ! Hello(token)
//println("recv[" + peers + "]: sent 'Hello' to " + path)
(peers._1, (path, pong, sender, token) :: peers._2)
}
}
}
catch {
case e => {
// catches match error and integer conversion failure
sender ! Error(e.toString, token)
peers
}
}
}
protected override def kickOff(peers: Peers, value: Int) = {
(new PingActor(sender, peers.connected map (_.channel))).start ! KickOff(value)
}
protected override def connectionEstablished(peers: Peers, x: Any) = x match {
case PendingPeer(path, channel, client, token) => {
client ! Ok(token)
Peers(Peer(path, channel) :: peers.connected, peers.pending filterNot (_.clientToken == token))
}
case KickOff(value) => {
(new PingActor(sender, peers._1.map(x => x._2))).start ! KickOff(value)
//println("recv[" + peers + "]: KickOff(" + value + ")")
peers
}
protected def newPending(peers: Peers, path: String, token: String) : Peers = path split ":" match {
case Array(node, port) => {
val channel = select(new Node(node, port.toInt), 'Pong)
channel ! Hello(token)
//println("recv[" + peers + "]: sent 'Hello' to " + path)
Peers(peers.connected, PendingPeer(path, channel, sender, token) :: peers.pending)
}
}
@tailrec private def recvLoop(peers: PListPair): Nothing = {
recvLoop(recv(peers))
protected override def handleTimeout(peers: Peers) = {
peers.pending foreach (x => x.client ! Error("cannot connect to " + x.path, x.clientToken))
(true, Peers(peers.connected, Nil))
}
override def act() {
RemoteActor.classLoader = getClass().getClassLoader
RemoteActor classLoader = getClass().getClassLoader
alive(port)
register('Pong, Actor.self)
recvLoop((Nil, Nil))
register('Pong, self)
@tailrec def recvLoop(peers: Peers): Nothing = {
def recv(peers: Peers, receiveFun: PartialFunction[Any, (Boolean, Peers)] => (Boolean, Peers)): Peers = receiveFun(recvFun(peers))._2
recvLoop(recv(peers, if (peers.pending isEmpty) receive else receiveWithin(5000)))
}
recvLoop(Peers(Nil, Nil))
}
}
class ClientActor(pongPaths: List[RemoteActorPath], numPings: Int) extends Actor {
override def act() = {
RemoteActor.classLoader = getClass().getClassLoader
val pongs = pongPaths.map(x => {
RemoteActor classLoader = getClass().getClassLoader
val pongs = pongPaths map (x => {
val pong = select(new Node(x.host, x.port), 'Pong)
pongPaths.foreach(y => {
if (x != y) {
val token = x.uri + " -> " + y.uri
pong ! AddPong(y.uri, token)
}
})
pongPaths foreach (y => if (x != y) pong ! AddPong(y.uri, x.uri + " -> " + y.uri))
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))
@tailrec def collectOkMessages(left: Int, receivedTokens: List[String]): Unit = {
if (left > 0)
collectOkMessages(left - 1, receiveWithin(10000) {
case Ok(token) => token :: receivedTokens
case Error(msg, token) => throw new RuntimeException("connection failed: " + token + ", message from server: " + msg)
case TIMEOUT => throw new RuntimeException("no Ok within 10sec.\nreceived tokens:\n" + receivedTokens.sortWith(_.compareTo(_) < 0).mkString("\n"))
})
}
collectOkMessages(pongs.length * (pongs.length - 1), Nil)
// collect ok messages
var receivedTokens: List[String] = Nil
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)
}
}
}
// kickoff
pongs.foreach(x => x ! KickOff(numPings))
pongs foreach (_ ! KickOff(numPings))
// collect done messages
for (_ <- 1 until (pongs.length * (pongs.length - 1))) {
receiveWithin(30*60*1000) {
case Done => {
}
case TIMEOUT => {
println("no Done within 30min")
System.exit(1)
}
case Done => Unit
case TIMEOUT => throw new RuntimeException("no Done within 30min")
case x => throw new RuntimeException("Unexpected message: " + x.toString)
}
}
}
}
class PingAkkaActor(parent: AkkaActorRef, pong: AkkaActorRef) extends AkkaActor {
def receive = {
case class SetParent(parent: AkkaActorRef)
class AkkaPingActor(pongs: List[AkkaActorRef]) extends AkkaActor {
import context.become
private var parent: AkkaActorRef = null
private var left = pongs.length
private def recvLoop: Receive = {
case Pong(0) => {
parent ! Done
context.stop(self)
}
case Pong(value) => {
sender ! Ping(value - 1)
}
case KickOff(value) => {
pong ! Ping(value)
//println(parent.toString + " ! Done")
if (left > 1) left -= 1
else context.stop(self)
}
case Pong(value) => sender ! Ping(value - 1)
}
def receive = {
case SetParent(p) => parent = p
case KickOff(value) => pongs.foreach(_ ! Ping(value)); become(recvLoop)
}
}
class ServerAkkaActor(system: ActorSystem) extends AkkaActor {
case class AkkaPeer(path: String, channel: AkkaActorRef)
case class PendingAkkaPeer(path: String, channel: AkkaActorRef, client: AkkaActorRef, clientToken: String)
case class AkkaPeers(connected: List[AkkaPeer], pending: List[PendingAkkaPeer])
class AkkaServerActor(system: ActorSystem) extends AkkaActor with ServerActorPrototype[AkkaPeers] {
import context.become
def recvLoop(pongs: List[AkkaActorRef], pendingPongs: List[(AkkaActorRef, AkkaActorRef, String)]): Receive = {
case Ping(value) => {
sender ! Pong(value)
}
case Olleh(_) => {
pendingPongs.find(x => x == sender) match {
case Some((pong, ping, token)) if pong == sender => {
println("added actor " + pong.path)
ping ! Ok(token)
become(recvLoop(pong :: pongs, pendingPongs.filterNot(x => x._1 == sender)))
}
case None => {
// operation already timed out
}
}
protected def reply(what: Any): Unit = sender ! what
protected def kickOff(peers: AkkaPeers, value: Int): Unit = {
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 {
case PendingAkkaPeer(path, channel, client, token) => {
client ! Ok(token)
//println("connected to " + path)
AkkaPeers(AkkaPeer(path, channel) :: peers.connected, peers.pending filterNot (_.clientToken == token))
}
case AkkaPongDidNotRespond(pong, ping) => {
pendingPongs.find(x => x._1 == sender) match {
case Some((_, _, token)) => {
ping ! Error(pong + " did not respond", token)
become(recvLoop(pongs, pendingPongs.filterNot(x => x._1 == sender)))
}
case None => {
// operation already succeeded (received Olleh)
}
}
protected def newPending(peers: AkkaPeers, path: String, token: String) : AkkaPeers = {
val channel = system.actorFor(path)
channel ! Hello(token)
import akka.util.duration._
system.scheduler.scheduleOnce(5 seconds, self, AddPongTimeout(path, token))
//println("[" + peers + "]: sent 'Hello' to " + path)
AkkaPeers(peers.connected, PendingAkkaPeer(path, channel, sender, token) :: peers.pending)
}
protected override def handleAddPongTimeout(peers: AkkaPeers, path: String, token: String) = {
peers.pending find (x => x.path == path && x.clientToken == token) match {
case Some(PendingAkkaPeer(_, channel, client, _)) => {
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)))
}
case None => (false, peers)
}
case AddPong(path, token) => {
if (pongs.exists((x) => x.path == path)) {
sender ! Ok(token)
}
else {
import akka.util.duration._
println("try to add actor " + path)
val pong = system.actorFor(path)
// wait at most 5sec. for response
pong ! Hello("")
system.scheduler.scheduleOnce(5 seconds, self, AkkaPongDidNotRespond(pong, sender))
become(recvLoop(pongs, (pong, sender, token) :: pendingPongs))
//pong ! Hello
//pongs = pong :: pongs
//sender ! Ok
}
def bhvr(peers: AkkaPeers): Receive = {
case x => {
recvFun(peers)(x) match {
case (true, newPeers) => become(bhvr(newPeers))
case _ => Unit
}
}
case KickOff(value) => {
val client = sender
//println("KickOff(" + value + ") from " + client)
pongs.foreach((x) => context.actorOf(Props(new PingAkkaActor(client, x))) ! KickOff(value))
}
case Hello(token) => {
sender ! Olleh(token)
}
}
def receive = recvLoop(Nil, Nil)
def receive = bhvr(AkkaPeers(Nil, Nil))
}
class ClientAkkaActor extends AkkaActor {
case class TokenTimeout(token: String)
case class RunAkkaClient(paths: List[String], numPings: Int)
import context._
class AkkaClientActor(system: ActorSystem) extends AkkaActor {
import context.become
def collectDoneMessages(left: Int): Receive = {
case Done => {
......@@ -312,21 +268,23 @@ class ClientAkkaActor 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) => {
//println("Ok")
if (tokens.length + 1 == pongs.length) {
val left = pongs.length * (pongs.length - 1)
pongs.foreach(x => x ! KickOff(numPings))
become(collectDoneMessages(left))
if (left == 1) {
//println("collected all Ok messages (wait for Done messages)")
pongs foreach (_ ! KickOff(numPings))
become(collectDoneMessages(pongs.length * (pongs.length - 1)))
}
else {
become(collectOkMessages(pongs, token :: tokens, numPings))
become(collectOkMessages(pongs, left - 1, token :: receivedTokens, numPings))
}
}
case NoTokenReceived(token) => {
if (!tokens.contains(token)) {
println("Error: " + token + " did not reply within 10 seconds");
case TokenTimeout(token) => {
if (!receivedTokens.contains(token)) {
println("Error: " + token + " did not reply within 10 seconds")
global.latch.countDown
context.stop(self)
}
}
case Error(what, token) => {
......@@ -337,14 +295,20 @@ class ClientAkkaActor extends AkkaActor {
}
def receive = {
case RunAkkaClient(pongs, numPings) => {
case RunAkkaClient(paths, numPings) => {
//println("RunAkkaClient(" + paths.toString + ", " + numPings + ")")
import akka.util.duration._
pongs.foreach(x => pongs.foreach(y => if (x != y) {
val token = x.toString
x ! AddPong(y.path.toString, token)
system.scheduler.scheduleOnce(10 seconds, self, NoTokenReceived(token))
}))
become(collectOkMessages(pongs, Nil, numPings))
val pongs = paths map (x => {
val pong = system.actorFor(x)
paths foreach (y => if (x != y) {
val token = x + " -> " + y
pong ! AddPong(y, token)
//println(x + " ! AddPong(" + y + ", " + token + ")")
system.scheduler.scheduleOnce(10 seconds, self, TokenTimeout(token))
})
pong
})
become(collectOkMessages(pongs, pongs.length * (pongs.length - 1), Nil, numPings))
}
}
}
......@@ -354,83 +318,57 @@ object DistributedClientApp {
val NumPings = "num_pings=([0-9]+)".r
val SimpleUri = "([0-9a-zA-Z\\.]+):([0-9]+)".r
def runAkka(system: ActorSystem, args: List[String], pongs: List[AkkaActorRef], numPings: Option[Int]): Unit = args match {
@tailrec def run(args: List[String], paths: List[String], numPings: Option[Int], finalizer: (List[String], Int) => Unit): 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 => runAkka(system, tail, pongs, Some(num.toInt))
}
case path :: tail => {
runAkka(system, tail, system.actorFor(path) :: pongs, numPings)
case Some(x) => throw new IllegalArgumentException("\"num_pings\" already defined, first value = " + x + ", second value = " + num)
case None => run(tail, paths, Some(num.toInt), finalizer)
}
case arg :: tail => run(tail, arg :: paths, numPings, finalizer)
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
system.shutdown
System.exit(0)
}
case None => {
throw new RuntimeException("no \"num_pings\" found")
if (paths.length < 2) throw new RuntimeException("at least two hosts required")
finalizer(paths, x)
}
case None => throw new RuntimeException("no \"num_pings\" found")
}
}
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)
def main(args: Array[String]): Unit = {
try {
args(0) match {
case "remote_actors" => run(args.toList.drop(1), Nil, None, ((paths, x) => {
(new ClientActor(paths map (path => path match { case SimpleUri(host, port) => RemoteActorPath(path, host, port.toInt) }), x)).start
}))
case "akka" => run(args.toList.drop(1), Nil, None, ((paths, x) => {
val system = ActorSystem("benchmark", ConfigFactory.load.getConfig("benchmark"))
system.actorOf(Props(new AkkaClientActor(system))) ! RunAkkaClient(paths, x)
global.latch.await
system.shutdown
System.exit(0)
}))
}
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)
catch {
case e => {
println("usage: DistributedClientApp (remote_actors|akka) {nodes...} num_pings={num}\nexample: DistributedClientApp remote_actors localhost:1234 localhost:2468 num_pings=1000\n")
throw e
}
}
case Nil => numPings match {
case Some(x) => {
if (pongs.length < 2) throw new RuntimeException("at least two hosts required")
(new ClientActor(pongs, x)).start
}
case None => {
throw new RuntimeException("no \"num_pings\" found")
}
}
}
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: ...")
}
}
}
object DistributedServerApp {
def usage = println("usage: (akka [configName]) | (remote_actors {port})")
val IntStr = "([0-9]+)".r
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")
pong
}
case "remote_actors" :: port :: Nil => {
(new ServerActor(port.toInt)).start
def main(args: Array[String]): Unit = args match {
case Array("remote_actors", IntStr(istr)) => (new ServerActor(istr.toInt)).start
case Array("akka") => {
val system = ActorSystem("pongServer", ConfigFactory.load.getConfig("pongServer"))
val pong = system.actorOf(Props(new AkkaServerActor(system)), "pong")
Unit
}
case _ => usage
case _ => println("usage: DistributedServerApp remote_actors PORT\n" +
" or: DistributedServerApp akka")
}
}
benchmark {
akka {
loglevel = ERROR
actor.provider = "akka.remote.RemoteActorRefProvider"
remote {
transport = "akka.remote.netty.NettyRemoteTransport"
untrusted-mode = on
# remote-daemon-ack-timeout = 300s
# netty {
# connection-timeout = 1800s
# }
}
}
}
pongServer {
akka {
loglevel = ERROR
......@@ -5,12 +19,16 @@ pongServer {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
transport = "akka.remote.netty.NettyRemoteTransport"
untrusted-mode = on
remote-daemon-ack-timeout = 300s
# remote-daemon-ack-timeout = 300s
netty {
backoff-timeout = 0ms
connection-timeout = 300s
hostname = "mobi10"
# backoff-timeout = 0ms
connection-timeout = 1800s
# read-timeout = 1800s
# write-timeout = 10s
all-timeout = 1800s
#hostname = "mobi10"
port = 2244
}
}
......
......@@ -66,7 +66,7 @@ option<int> c_2i(char const* cstr) {
return result;
}
inline option<int> _2i(std::string const& str) {
inline option<int> _2i(const std::string& str) {
return c_2i(str.c_str());
}
......@@ -109,7 +109,7 @@ class actor_template {
actor_ptr spawn() const {
struct impl : fsm_actor<impl> {
behavior init_state;
impl(MatchExpr const& mx) : init_state(mx.as_partial_function()) {
impl(const MatchExpr& mx) : init_state(mx.as_partial_function()) {
}
};
return cppa::spawn(new impl{m_expr});
......@@ -118,7 +118,7 @@ class actor_template {
};
template<typename... Args>
auto actor_prototype(Args const&... args) -> actor_template<decltype(mexpr_concat(args...))> {
auto actor_prototype(const Args&... args) -> actor_template<decltype(mexpr_concat(args...))> {
return {mexpr_concat(args...)};
}
......@@ -169,7 +169,7 @@ struct server_actor : fsm_actor<server_actor> {
on(atom("ping"), arg_match) >> [=](uint32_t value) {
reply(atom("pong"), value);
},
on(atom("add_pong"), arg_match) >> [=](string const& host, uint16_t port) {
on(atom("add_pong"), arg_match) >> [=](const string& host, uint16_t port) {
auto key = std::make_pair(host, port);
auto i = m_pongs.find(key);
if (i == m_pongs.end()) {
......@@ -199,7 +199,7 @@ struct server_actor : fsm_actor<server_actor> {
on<atom("EXIT"), uint32_t>() >> [=]() {
actor_ptr who = last_sender();
auto i = std::find_if(m_pongs.begin(), m_pongs.end(),
[&](pong_map::value_type const& kvp) {
[&](const pong_map::value_type& kvp) {
return kvp.second == who;
});
if (i != m_pongs.end()) m_pongs.erase(i);
......@@ -233,7 +233,7 @@ template<typename Iterator>
void server_mode(Iterator first, Iterator last) {
string port_prefix = "--port=";
// extracts port from a key-value pair
auto kvp_port = [&](string const& str) -> option<int> {
auto kvp_port = [&](const string& str) -> option<int> {
if (std::equal(port_prefix.begin(), port_prefix.end(), str.begin())) {
return c_2i(str.c_str() + port_prefix.size());
}
......@@ -261,7 +261,7 @@ void client_mode(Iterator first, Iterator last) {
std::uint32_t init_value = 0;
std::vector<std::pair<string, uint16_t> > remotes;
string pings_prefix = "--num_pings=";
auto num_msgs = [&](string const& str) -> option<int> {
auto num_msgs = [&](const string& str) -> option<int> {
if (std::equal(pings_prefix.begin(), pings_prefix.end(), str.begin())) {
return c_2i(str.c_str() + pings_prefix.size());
}
......
......@@ -250,7 +250,7 @@ void usage() {
enum mode_type { event_based, fiber_based };
option<int> _2i(std::string const& str) {
option<int> _2i(const std::string& str) {
char* endptr = nullptr;
int result = static_cast<int>(strtol(str.c_str(), &endptr, 10));
if (endptr == nullptr || *endptr != '\0') {
......@@ -265,7 +265,7 @@ int main(int argc, char** argv) {
// skip argv[0] (app name)
std::vector<std::string> args{argv + 1, argv + argc};
match(args) (
on(val<std::string>, _2i, _2i, _2i, _2i) >> [](std::string const& mode,
on(val<std::string>, _2i, _2i, _2i, _2i) >> [](const std::string& mode,
int num_rings,
int ring_size,
int initial_token_value,
......
......@@ -37,7 +37,7 @@
#include <stdexcept>
#include <algorithm>
inline std::vector<std::string> split(std::string const& str, char delim) {
inline std::vector<std::string> split(const std::string& str, char delim) {
std::vector<std::string> result;
std::stringstream strs{str};
std::string tmp;
......@@ -45,8 +45,8 @@ inline std::vector<std::string> split(std::string const& str, char delim) {
return result;
}
inline std::string join(std::vector<std::string> const& vec,
std::string const& delim = "") {
inline std::string join(const std::vector<std::string>& vec,
const std::string& delim = "") {
if (vec.empty()) return "";
auto result = vec.front();
for (auto i = vec.begin() + 1; i != vec.end(); ++i) {
......
......@@ -53,16 +53,16 @@ class behavior {
friend behavior operator,(partial_function&& lhs, behavior&& rhs);
behavior(const behavior&) = delete;
behavior& operator=(const behavior&) = delete;
public:
behavior() = default;
behavior(behavior&&) = default;
behavior(const behavior&) = default;
behavior& operator=(behavior&&) = default;
behavior& operator=(const behavior&) = default;
inline behavior(partial_function&& fun) : m_fun(std::move(fun)) {
}
inline behavior(partial_function&& fun) : m_fun(std::move(fun)) { }
template<typename... Cases>
behavior(const match_expr<Cases...>& me) : m_fun(me) { }
......@@ -71,8 +71,6 @@ class behavior {
: m_timeout(tout), m_timeout_handler(std::move(handler)) {
}
behavior& operator=(behavior&&) = default;
inline void handle_timeout() const {
m_timeout_handler();
}
......@@ -89,7 +87,7 @@ class behavior {
return m_fun(value);
}
inline bool operator()(any_tuple const& value) {
inline bool operator()(const any_tuple& value) {
return m_fun(value);
}
......
......@@ -501,12 +501,12 @@ template<typename T>
struct spawn_fwd_ {
static inline T&& _(T&& arg) { return std::move(arg); }
static inline T& _(T& arg) { return arg; }
static inline T const& _(T const& arg) { return arg; }
static inline const T& _(const T& arg) { return arg; }
};
template<>
struct spawn_fwd_<self_type> {
static inline actor_ptr _(self_type const&) { return self; }
static inline actor_ptr _(const self_type&) { return self; }
};
template<typename F, typename Arg0, typename... Args>
......
......@@ -58,7 +58,7 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
std::atomic<int> m_state;
filter_result filter_msg(any_tuple const& msg) {
filter_result filter_msg(const any_tuple& msg) {
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
......
......@@ -56,7 +56,7 @@ class actor_proxy_cache {
bool erase(const actor_proxy_ptr& pptr);
template<typename Fun>
void erase_all(process_information::node_id_type const& nid,
void erase_all(const process_information::node_id_type& nid,
std::uint32_t process_id,
Fun fun) {
key_tuple lb{nid, process_id, std::numeric_limits<actor_id>::min()};
......@@ -82,7 +82,7 @@ class actor_proxy_cache {
key_tuple;
struct key_tuple_less {
bool operator()(key_tuple const& lhs, key_tuple const& rhs) const;
bool operator()(const key_tuple& lhs, const key_tuple& rhs) const;
};
util::shared_spinlock m_lock;
......
......@@ -83,7 +83,7 @@ class converted_thread_context
inline void pop_timeout() { }
filter_result filter_msg(any_tuple const& msg);
filter_result filter_msg(const any_tuple& msg);
inline decltype(m_mailbox)& mailbox() { return m_mailbox; }
......
......@@ -45,7 +45,7 @@ class network_manager {
virtual void stop() = 0;
virtual void send_to_post_office(po_message const& msg) = 0;
virtual void send_to_post_office(const po_message& msg) = 0;
virtual void send_to_post_office(any_tuple msg) = 0;
......
......@@ -55,7 +55,7 @@ struct recursive_queue_node {
recursive_queue_node(recursive_queue_node&&) = delete;
recursive_queue_node(recursive_queue_node const&) = delete;
recursive_queue_node& operator=(recursive_queue_node&&) = delete;
recursive_queue_node& operator=(recursive_queue_node const&) = delete;
recursive_queue_node& operator=(const recursive_queue_node&) = delete;
};
......
......@@ -44,7 +44,7 @@ struct scheduled_actor_dummy : abstract_scheduled_actor {
void unlink_from(intrusive_ptr<actor>&);
bool establish_backlink(intrusive_ptr<actor>&);
bool remove_backlink(intrusive_ptr<actor>&);
void detach(attachable::token const&);
void detach(const attachable::token&);
bool attach(attachable*);
};
......
......@@ -76,12 +76,12 @@ class value_guard {
}
template<typename T0, typename T1>
static inline bool cmp(T0 const& lhs, T1 const& rhs) {
static inline bool cmp(const T0& lhs, const T1& rhs) {
return lhs == rhs;
}
template<typename T0, typename T1>
static inline bool cmp(T0 const& lhs, std::reference_wrapper<T1> const& rhs) {
static inline bool cmp(const T0& lhs, const std::reference_wrapper<T1>& rhs) {
return lhs == rhs.get();
}
......
......@@ -36,6 +36,8 @@
#include <memory>
#include <utility>
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/intrusive/singly_linked_list.hpp"
......@@ -49,36 +51,34 @@ class behavior;
*/
class partial_function {
partial_function(const partial_function&) = delete;
partial_function& operator=(const partial_function&) = delete;
public:
struct impl {
virtual ~impl();
struct impl : ref_counted {
virtual bool invoke(any_tuple&) = 0;
virtual bool invoke(const any_tuple&) = 0;
virtual bool defined_at(const any_tuple&) = 0;
};
typedef std::unique_ptr<impl> impl_ptr;
typedef intrusive_ptr<impl> impl_ptr;
partial_function() = default;
partial_function(partial_function&&) = default;
partial_function(const partial_function&) = default;
partial_function& operator=(partial_function&&) = default;
partial_function& operator=(const partial_function&) = default;
partial_function(impl_ptr&& ptr);
inline bool defined_at(const any_tuple& value) {
return ((m_impl) && m_impl->defined_at(value));
return (m_impl) && m_impl->defined_at(value);
}
inline bool operator()(any_tuple& value) {
return ((m_impl) && m_impl->invoke(value));
return (m_impl) && m_impl->invoke(value);
}
inline bool operator()(const any_tuple& value) {
return ((m_impl) && m_impl->invoke(value));
return (m_impl) && m_impl->invoke(value);
}
inline bool operator()(any_tuple&& value) {
......
......@@ -39,12 +39,12 @@
namespace cppa {
actor_proxy::actor_proxy(std::uint32_t mid, process_information_ptr const& pptr)
actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
: super(mid, pptr) {
//attach(get_scheduler()->register_hidden_context());
}
void actor_proxy::forward_message(process_information_ptr const& piptr,
void actor_proxy::forward_message(const process_information_ptr& piptr,
actor* sender,
any_tuple&& msg) {
detail::addressed_message amsg{sender, this, std::move(msg)};
......
......@@ -66,12 +66,12 @@ actor_proxy_cache& get_actor_proxy_cache() {
actor_proxy_ptr actor_proxy_cache::get(actor_id aid,
std::uint32_t process_id,
process_information::node_id_type const& node_id) {
const process_information::node_id_type& node_id) {
key_tuple k{node_id, process_id, aid};
return get_impl(k);
}
actor_proxy_ptr actor_proxy_cache::get_impl(key_tuple const& key) {
actor_proxy_ptr actor_proxy_cache::get_impl(const key_tuple& key) {
{ // lifetime scope of shared guard
util::shared_lock_guard<util::shared_spinlock> guard{m_lock};
auto i = m_entries.find(key);
......@@ -95,7 +95,7 @@ actor_proxy_ptr actor_proxy_cache::get_impl(key_tuple const& key) {
return result;
}
bool actor_proxy_cache::erase(actor_proxy_ptr const& pptr) {
bool actor_proxy_cache::erase(const actor_proxy_ptr& pptr) {
auto pinfo = pptr->parent_process_ptr();
key_tuple key(pinfo->node_id(), pinfo->process_id(), pptr->id()); {
lock_guard<util::shared_spinlock> guard{m_lock};
......@@ -104,8 +104,8 @@ bool actor_proxy_cache::erase(actor_proxy_ptr const& pptr) {
return false;
}
bool actor_proxy_cache::key_tuple_less::operator()(key_tuple const& lhs,
key_tuple const& rhs) const {
bool actor_proxy_cache::key_tuple_less::operator()(const key_tuple& lhs,
const key_tuple& rhs) const {
int cmp_res = strncmp(reinterpret_cast<char const*>(std::get<0>(lhs).data()),
reinterpret_cast<char const*>(std::get<0>(rhs).data()),
process_information::node_id_size);
......
......@@ -90,7 +90,7 @@ void actor_registry::put(actor_id key, const actor_ptr& value) {
void actor_registry::erase(actor_id key) {
exclusive_guard guard(m_instances_mtx);
auto i = std::find_if(m_instances.begin(), m_instances.end(),
[=](std::pair<actor_id, actor_ptr> const& p) {
[=](const std::pair<actor_id, actor_ptr>& p) {
return p.first == key;
});
if (i != m_instances.end())
......
......@@ -99,7 +99,7 @@ void converted_thread_context::dequeue(behavior& bhvr) { // override
}
}
filter_result converted_thread_context::filter_msg(any_tuple const& msg) {
filter_result converted_thread_context::filter_msg(const any_tuple& msg) {
if (m_trap_exit == false && matches(msg, m_exit_msg_pattern)) {
auto reason = msg.get_as<std::uint32_t>(1);
if (reason != exit_reason::normal) {
......
......@@ -82,7 +82,7 @@ struct network_manager_impl : network_manager {
close(pipe_fd[0]);
}
void send_to_post_office(po_message const& msg) {
void send_to_post_office(const po_message& msg) {
if (write(pipe_fd[1], &msg, sizeof(po_message)) != sizeof(po_message)) {
CPPA_CRITICAL("cannot write to pipe");
}
......
......@@ -40,7 +40,4 @@ namespace cppa {
partial_function::partial_function(impl_ptr&& ptr) : m_impl(std::move(ptr)) {
}
partial_function::impl::~impl() {
}
} // namespace cppa
......@@ -109,7 +109,7 @@ inline void send2po_(network_manager* nm, Arg0&& arg0, Args&&... args) {
template<typename... Args>
inline void send2po(po_message const& msg, Args&&... args) {
inline void send2po(const po_message& msg, Args&&... args) {
auto nm = singleton_manager::get_network_manager();
nm->send_to_post_office(msg);
send2po_(nm, std::forward<Args>(args)...);
......@@ -440,7 +440,7 @@ void post_office_loop(int input_fd) {
case valof(atom("RM_PEER")): {
DEBUG("post_office: rm_peer");
auto i = std::find_if(handler.begin(), handler.end(),
[&](po_socket_handler_ptr const& hp) {
[&](const po_socket_handler_ptr& hp) {
return hp->get_socket() == msg.fd;
});
if (i != handler.end()) handler.erase(i);
......@@ -461,7 +461,7 @@ void post_office_loop(int input_fd) {
case valof(atom("UNPUBLISH")): {
DEBUG("post_office: unpublish_actor");
auto i = std::find_if(handler.begin(), handler.end(),
[&](po_socket_handler_ptr const& hp) {
[&](const po_socket_handler_ptr& hp) {
return hp->is_doorman_of(msg.aid);
});
if (i != handler.end()) handler.erase(i);
......@@ -491,13 +491,13 @@ void post_office_loop(int input_fd) {
******************************************************************************/
void post_office_add_peer(native_socket_type a0,
process_information_ptr const& a1) {
const process_information_ptr& a1) {
po_message msg{atom("ADD_PEER"), -1, 0};
send2po(msg, a0, a1);
}
void post_office_publish(native_socket_type server_socket,
actor_ptr const& published_actor) {
const actor_ptr& published_actor) {
po_message msg{atom("PUBLISH"), -1, 0};
send2po(msg, server_socket, published_actor);
}
......
......@@ -58,7 +58,7 @@ bool scheduled_actor_dummy::remove_backlink(intrusive_ptr<actor>&) {
return false;
}
void scheduled_actor_dummy::detach(attachable::token const&) {
void scheduled_actor_dummy::detach(const attachable::token&) {
}
bool scheduled_actor_dummy::attach(attachable*) {
......
......@@ -236,8 +236,8 @@ actor_ptr thread_pool_scheduler::spawn(std::function<void()> what,
}
#else
actor_ptr thread_pool_scheduler::spawn(std::function<void()> what,
scheduling_hint hint) {
return mock_scheduler::spawn(what, hint);
scheduling_hint) {
return mock_scheduler::spawn_impl(std::move(what));
}
#endif
......
......@@ -153,7 +153,7 @@ int main(int argc, char** argv) {
await_all_others_done();
exit(0);
},
on("run_ping", arg_match) >> [&](std::string const& num_pings) {
on("run_ping", arg_match) >> [&](const std::string& num_pings) {
auto ping_actor = spawn<detached>(ping, std::stoi(num_pings));
//auto ping_actor = spawn_event_based_ping(std::stoi(num_pings));
std::uint16_t port = 4242;
......
......@@ -27,12 +27,91 @@ bool ascending(int a, int b, int c) {
return a < b && b < c;
}
vector<uniform_type_info const*> to_vec(util::type_list<>, vector<uniform_type_info const*> vec = vector<uniform_type_info const*>{}) {
return vec;
}
template<typename Head, typename... Tail>
vector<uniform_type_info const*> to_vec(util::type_list<Head, Tail...>, vector<uniform_type_info const*> vec = vector<uniform_type_info const*>{}) {
vec.push_back(uniform_typeid<Head>());
return to_vec(util::type_list<Tail...>{}, std::move(vec));
}
template<typename Fun>
class __ {
typedef typename util::get_callable_trait<Fun>::type trait;
public:
__(Fun f, string annotation = "") : m_fun(f), m_annotation(annotation) { }
__ operator[](const string& str) const {
return {m_fun, str};
}
template<typename... Args>
void operator()(Args&&... args) const {
m_fun(std::forward<Args>(args)...);
}
void plot_signature() {
auto vec = to_vec(typename trait::arg_types{});
for (auto i : vec) {
cout << i->name() << " ";
}
cout << endl;
}
const string& annotation() const {
return m_annotation;
}
private:
Fun m_fun;
string m_annotation;
};
template<typename Fun>
__<Fun> get__(Fun f) {
return {f};
}
struct fobaz : fsm_actor<fobaz> {
behavior init_state;
void vfun() {
cout << "fobaz::mfun" << endl;
}
void ifun(int i) {
cout << "fobaz::ifun(" << i << ")" << endl;
}
fobaz() {
init_state = (
on<int>() >> [=](int i) { ifun(i); },
others() >> std::function<void()>{std::bind(&fobaz::vfun, this)}
);
}
};
size_t test__match() {
CPPA_TEST(test__match);
using namespace std::placeholders;
using namespace cppa::placeholders;
/*
auto x_ = get__([](int, float) { cout << "yeeeehaaaa!" << endl; })["prints 'yeeeehaaaa'"];
cout << "x_: "; x_(1, 1.f);
cout << "x_.annotation() = " << x_.annotation() << endl;
cout << "x_.plot_signature: "; x_.plot_signature();
auto fun = (
on<int>() >> [](int i) {
cout << "i = " << i << endl;
......@@ -41,6 +120,7 @@ size_t test__match() {
cout << "no int found in mailbox" << endl;
}
);
*/
auto expr0_a = gcall(ascending, _x1, _x2, _x3);
CPPA_CHECK(ge_invoke(expr0_a, 1, 2, 3));
......
......@@ -282,22 +282,16 @@ class actor_template {
actor_ptr spawn() const {
struct impl : fsm_actor<impl> {
behavior init_state;
impl(MatchExpr const& mx) : init_state(mx.as_partial_function()) {
impl(const MatchExpr& mx) : init_state(mx.as_partial_function()) {
}
};
return cppa::spawn(new impl{m_expr});
}
actor_ptr spawn_detached() const {
return cppa::spawn<detached>([m_expr]() {
receive_loop(m_expr);
});
}
};
template<typename... Args>
auto actor_prototype(Args const&... args) -> actor_template<decltype(mexpr_concat(args...))> {
auto actor_prototype(const Args&... args) -> actor_template<decltype(mexpr_concat(args...))> {
return {mexpr_concat(args...)};
}
......@@ -321,6 +315,29 @@ class str_wrapper {
};
struct some_integer {
void set(int value) { m_value = value; }
int get() const { return m_value; }
private:
int m_value;
};
template<class T, class MatchExpr = match_expr<>, class Parent = void>
class actor_facade_builder {
public:
private:
MatchExpr m_expr;
};
bool operator==(const str_wrapper& lhs, const std::string& rhs) {
return lhs.str() == rhs;
}
......@@ -336,10 +353,20 @@ void foobar(const str_wrapper& x, const std::string& y) {
);
}
size_t test__spawn() {
using std::string;
CPPA_TEST(test__spawn);
/*
actor_ptr tst = actor_facade<some_integer>()
.reply_to().when().with()
.reply_to().when().with()
.spawn();
send(tst, atom("EXIT"), exit_reason::user_defined);
*/
CPPA_IF_VERBOSE(cout << "test send() ... " << std::flush);
send(self, 1, 2, 3);
receive(on(1, 2, 3) >> []() { });
......@@ -420,9 +447,9 @@ size_t test__spawn() {
{
bool invoked = false;
str_wrapper x{"x"};
std::string y{"y"};
auto foo_actor = spawn(foobar, std::cref(x), y);
str_wrapper actor_facade_builder{"x"};
std::string actor_facade_builder_interim{"y"};
auto foo_actor = spawn(foobar, std::cref(actor_facade_builder), actor_facade_builder_interim);
send(foo_actor, atom("same"));
receive (
on(atom("yes")) >> [&]() {
......
......@@ -73,9 +73,9 @@ size_t test__yield_interface() {
++i;
}
while (ys != yield_state::done && i < 12);
CPPA_CHECK_EQUAL(ys, yield_state::done);
CPPA_CHECK_EQUAL(worker.m_count, 10);
CPPA_CHECK_EQUAL(i, 12);
CPPA_CHECK_EQUAL(yield_state::done, ys);
CPPA_CHECK_EQUAL(10, worker.m_count);
CPPA_CHECK_EQUAL(12, i);
# endif
return CPPA_TEST_RESULT;
}
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