Commit ef3dfadf authored by Dominik Charousset's avatar Dominik Charousset

minor send optimization

parent c9437270
......@@ -4,6 +4,7 @@ import Console.println
case object Ok
case object Done
case object OkTimeout
case class Error(msg: String)
case class KickOff(value: Int)
case class Ping(value: Int)
......@@ -12,6 +13,8 @@ case class AddPong(path: String)
case object Hello
case object Olleh
case class PongDidNotRespond(pong: ActorRef, ping: ActorRef)
case class RunClient(pongs: List[ActorRef], numPings: Int)
object global {
......@@ -38,20 +41,50 @@ class PingActor(parent: ActorRef, pong: ActorRef) extends Actor {
class ServerActor(system: ActorSystem) extends Actor {
var pongs = List[ActorRef]()
var pendingPongs = List[Pair[ActorRef, ActorRef]]()
def receive = {
case Ping(value) => {
sender ! Pong(value)
}
case Olleh => {
pendingPongs.find(x => x._1 == sender) match {
case Some((pong, ping)) => {
println("added actor " + pong.path)
ping ! Ok
pendingPongs = pendingPongs.filterNot(x => x._1 == sender)
pongs = pong :: pongs
}
case None => {
// operation already timed out
}
}
}
case PongDidNotRespond(pong, ping) => {
pendingPongs.find(x => x._1 == sender) match {
case Some(Pair(_, y)) => {
ping ! Error(pong + " did not respond")
pendingPongs = pendingPongs.filterNot(x => x._1 == sender)
}
case None => {
// operation already succeeded
}
}
}
case AddPong(path) => {
//println("add actor " + path)
if (pongs.exists((x) => x.path == path)) {
sender ! Ok
} else {
import akka.util.duration._
println("try to add actor " + path)
val pong = system.actorFor(path)
// wait at most 5sec. for response
pong ! Hello
pongs = pong :: pongs
sender ! Ok
pendingPongs = Pair(pong, sender) :: pendingPongs
system.scheduler.scheduleOnce(5 seconds, self, PongDidNotRespond(pong, sender))
//pong ! Hello
//pongs = pong :: pongs
//sender ! Ok
}
}
case KickOff(value) => {
......@@ -62,9 +95,6 @@ class ServerActor(system: ActorSystem) extends Actor {
case Hello => {
sender ! Olleh
}
case Olleh => {
//println("Olleh from " + sender)
}
}
}
......@@ -101,21 +131,31 @@ class ClientActor extends Actor {
left = left - 1
}
}
case Error(what) => {
println("Error from " + sender + " => " + what)
global.latch.countDown
context.stop(self)
}
case OkTimeout => {
println("At least one pong did not reply... left: " + left)
}
}
def receive = {
case RunClient(pongs, numPings) => {
import akka.util.duration._
pongs.foreach(x => pongs.foreach(y => if (x != y) {
x ! AddPong(y.path.toString)
}))
system.scheduler.scheduleOnce(10 seconds, self, OkTimeout)
left = pongs.length * (pongs.length - 1)
become(collectOkMessages(pongs, numPings))
}
}
}
object distributedClientApp {
def main(args: Array[String]) = {
object DistributedClientApp {
def main(args: Array[String]): Unit = {
val system = ActorSystem("benchmark", ConfigFactory.load.getConfig("benchmark"))
......@@ -137,12 +177,13 @@ object distributedClientApp {
system.actorOf(Props[ClientActor]) ! RunClient(pongs, numPings)
global.latch.await
system.shutdown
System.exit(0)
}
}
object distributedServerApp {
def main(args: Array[String]) = {
object DistributedServerApp {
def main(args: Array[String]): Unit = {
val sysName = if (args.length > 0) args(0) else "pongServer"
......
#benchmark {
# include "common"
#
# akka {
# loglevel = ERROR
# remote {
# netty.port = 0
# untrusted-mode = on
# transport = "akka.remote.netty.NettyRemoteTransport"
# }
# }
#}
benchmark {
akka {
loglevel = ERROR
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
netty {
hostname = "127.0.0.1"
port = 0
}
}
}
}
pongServer {
akka {
loglevel = ERROR
......@@ -33,8 +5,12 @@ pongServer {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
untrusted-mode = on
remote-daemon-ack-timeout = 300s
netty {
hostname = "127.0.0.1"
backoff-timeout = 0ms
connection-timeout = 300s
hostname = "mobi10"
port = 2244
}
}
......
#!/bin/bash
echo "erl -noshell -noinput +P 20000000 -sname benchmark -s $@ -s init stop" | ./exec.sh
echo "erl -noshell -noinput +P 20000000 -setcookie abc123 -sname benchmark -s $@ -s init stop" | ./exec.sh
......@@ -68,21 +68,69 @@ T rd(char const* cstr)
return result;
}
void usage()
{
cerr << "usage: matching (cow_tuple|object_array) {NUM_LOOPS}" << endl;
exit(1);
}
int main(int argc, char** argv)
{
announce<list<int>>();
if (argc != 2)
if (argc != 3) usage();
auto num_loops = rd<int64_t>(argv[2]);
any_tuple m1;
any_tuple m2;
any_tuple m3;
any_tuple m4;
any_tuple m5;
any_tuple m6;
if (strcmp(argv[1], "cow_tuple") == 0)
{
m1 = make_cow_tuple(atom("msg1"), 0);
m2 = make_cow_tuple(atom("msg2"), 0.0);
m3 = make_cow_tuple(atom("msg3"), list<int>{0});
m4 = make_cow_tuple(atom("msg4"), 0, "0");
m5 = make_cow_tuple(atom("msg5"), 0, 0, 0);
m6 = make_cow_tuple(atom("msg6"), 0, 0.0, "0");
}
else if (strcmp(argv[1], "object_array") == 0)
{
auto m1o = new detail::object_array;
m1o->push_back(object::from(atom("msg1")));
m1o->push_back(object::from(0));
m1 = any_tuple{m1o};
auto m2o = new detail::object_array;
m2o->push_back(object::from(atom("msg2")));
m2o->push_back(object::from(0.0));
m2 = any_tuple{m2o};
auto m3o = new detail::object_array;
m3o->push_back(object::from(atom("msg3")));
m3o->push_back(object::from(list<int>{0}));
m3 = any_tuple{m3o};
auto m4o = new detail::object_array;
m4o->push_back(object::from(atom("msg4")));
m4o->push_back(object::from(0));
m4o->push_back(object::from(std::string("0")));
m4 = any_tuple{m4o};
auto m5o = new detail::object_array;
m5o->push_back(object::from(atom("msg5")));
m5o->push_back(object::from(0));
m5o->push_back(object::from(0));
m5o->push_back(object::from(0));
m5 = any_tuple{m5o};
auto m6o = new detail::object_array;
m6o->push_back(object::from(atom("msg6")));
m6o->push_back(object::from(0));
m6o->push_back(object::from(0.0));
m6o->push_back(object::from(std::string("0")));
m6 = any_tuple{m6o};
}
else
{
cerr << "usage: matching {NUM_LOOPS}" << endl;
return 1;
usage();
}
auto num_loops = rd<int64_t>(argv[1]);
any_tuple m1 = make_cow_tuple(atom("msg1"), 0);
any_tuple m2 = make_cow_tuple(atom("msg2"), 0.0);
any_tuple m3 = cppa::make_cow_tuple(atom("msg3"), list<int>{0});
any_tuple m4 = make_cow_tuple(atom("msg4"), 0, "0");
any_tuple m5 = make_cow_tuple(atom("msg5"), 0, 0, 0);
any_tuple m6 = make_cow_tuple(atom("msg6"), 0, 0.0, "0");
int64_t m1matched = 0;
int64_t m2matched = 0;
int64_t m3matched = 0;
......@@ -107,4 +155,10 @@ int main(int argc, char** argv)
part_fun(m5);
part_fun(m6);
}
assert(m1matched == num_loops);
assert(m2matched == num_loops);
assert(m3matched == num_loops);
assert(m4matched == num_loops);
assert(m5matched == num_loops);
assert(m6matched == num_loops);
}
......@@ -41,7 +41,7 @@ using std::enable_if;
namespace {
constexpr size_t chunk_size = 512;
constepxr size_t ui32_size = sizeof(std::uint32_t);
constexpr size_t ui32_size = sizeof(std::uint32_t);
} // namespace <anonymous>
......@@ -224,12 +224,12 @@ char const* binary_serializer::sendable_data()
{
auto s = static_cast<std::uint32_t>(size());
memcpy(m_begin, &s, ui32_size);
return m_begin();
return m_begin;
}
void binary_serializer::reset()
{
m_wr_pos = m_begin;
m_wr_pos = m_begin + ui32_size;
}
} // namespace cppa
......@@ -71,10 +71,9 @@ void mailman_loop()
try
{
bs << msg;
auto size32 = static_cast<std::uint32_t>(bs.size());
DEBUG("--> " << to_string(msg));
auto sent = ::send(peer_fd, bs.sendable_data(), bs.sendable_size());
if (sent != bs.sendable_size())
auto sent = ::send(peer_fd, bs.sendable_data(), bs.sendable_size(), 0);
if (sent < 0 || static_cast<size_t>(sent) != bs.sendable_size())
{
disconnect_peer = true;
DEBUG("too few bytes written");
......
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