Commit 27d75ca3 authored by Dominik Charousset's avatar Dominik Charousset

moved benchmarks to own repository

the benchmarks are now available at https://github.com/Neverlord/cppa-benchmarks
parent ace07fdb
cmake_minimum_required(VERSION 2.6)
project(cppa_benchmarks CXX)
add_custom_target(all_benchmarks)
include_directories(.)
macro(add_benchmark name)
add_executable(${name} cppa/${name}.cpp)
target_link_libraries(${name} ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES})
add_dependencies(${name} all_benchmarks)
endmacro()
add_benchmark(actor_creation)
add_benchmark(mailbox_performance)
add_benchmark(mixed_case)
add_benchmark(distributed)
add_benchmark(matching)
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
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
transport = "akka.remote.netty.NettyRemoteTransport"
untrusted-mode = on
# remote-daemon-ack-timeout = 300s
netty {
# backoff-timeout = 0ms
connection-timeout = 1800s
# read-timeout = 1800s
# write-timeout = 10s
all-timeout = 1800s
#hostname = "mobi10"
port = 2244
}
}
}
}
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <cassert>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include "utility.hpp"
#include "cppa/cppa.hpp"
using namespace std;
using namespace cppa;
struct testee : sb_actor<testee> {
behavior init_state;
testee(const actor_ptr& parent) {
init_state = (
on(atom("spread"), (uint32_t) 0) >> [=]() {
send(parent, atom("result"), (uint32_t) 1);
quit();
},
on(atom("spread"), arg_match) >> [=](uint32_t x) {
any_tuple msg = make_cow_tuple(atom("spread"), x - 1);
spawn<testee>(this) << msg;
spawn<testee>(this) << msg;
become (
on(atom("result"), arg_match) >> [=](uint32_t r1) {
become (
on(atom("result"), arg_match) >> [=](uint32_t r2) {
send(parent, atom("result"), r1 + r2);
quit();
}
);
}
);
}
);
}
};
void stacked_testee(actor_ptr parent) {
receive (
on(atom("spread"), (uint32_t) 0) >> [&]() {
send(parent, atom("result"), (uint32_t) 1);
},
on(atom("spread"), arg_match) >> [&](uint32_t x) {
any_tuple msg = make_cow_tuple(atom("spread"), x - 1);
spawn(stacked_testee, self) << msg;
spawn(stacked_testee, self) << msg;
receive (
on(atom("result"), arg_match) >> [&](uint32_t v1) {
receive (
on(atom("result"), arg_match) >> [&](uint32_t v2) {
send(parent, atom("result"), v1 + v2);
}
);
}
);
}
);
}
void usage() {
cout << "usage: actor_creation (stacked|event-based) POW" << endl
<< " creates 2^POW actors" << endl
<< endl;
}
int main(int argc, char** argv) {
vector<string> args(argv + 1, argv + argc);
match (args) (
on("stacked", spro<uint32_t>) >> [](uint32_t num) {
send(spawn(stacked_testee, self), atom("spread"), num);
},
on("event-based", spro<uint32_t>) >> [](uint32_t num) {
send(spawn<testee>(self), atom("spread"), num);
},
others() >> usage
);
await_all_others_done();
shutdown();
return 0;
}
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <chrono>
#include <utility>
#include <iostream>
#include <boost/timer.hpp>
#include <boost/progress.hpp>
#include "utility.hpp"
#include "cppa/cppa.hpp"
using namespace std;
using namespace cppa;
void usage() {
cout << "Running in server mode:" << endl
<< " mode=server " << endl
<< " --port=NUM publishes an actor at port NUM" << endl
<< " -p NUM alias for --port=NUM" << endl
<< endl
<< endl
<< "Running the benchmark:" << endl
<< " mode=benchmark run the benchmark, connect to any number" << endl
<< " of given servers, use HOST:PORT syntax" << endl
<< " num_pings=NUM run benchmark with NUM messages per node" << endl
<< endl
<< " example: mode=benchmark 192.168.9.1:1234 "
"192.168.9.2:1234 "
"--num_pings=100" << endl
<< endl
<< endl
<< "Shutdown servers:" << endl
<< " mode=shutdown shuts down any number of given servers" << endl
<< endl
<< endl
<< "Miscellaneous:" << endl
<< " -h, --help print this text and exit" << endl
<< endl;
exit(0);
}
template<class MatchExpr>
class actor_template {
MatchExpr m_expr;
public:
actor_template(MatchExpr me) : m_expr(move(me)) { }
actor_ptr spawn() const {
struct impl : sb_actor<impl> {
behavior init_state;
impl(const MatchExpr& mx) : init_state(mx.as_partial_function()) {
}
};
return cppa::spawn(new impl{m_expr});
}
};
template<typename... Args>
auto actor_prototype(const Args&... args) -> actor_template<decltype(mexpr_concat(args...))> {
return {mexpr_concat(args...)};
}
struct ping_actor : sb_actor<ping_actor> {
behavior init_state;
actor_ptr parent;
ping_actor(actor_ptr parent_ptr) : parent(move(parent_ptr)) {
init_state = (
on(atom("kickoff"), arg_match) >> [=](actor_ptr pong, uint32_t value) {
send(pong, atom("ping"), value);
become (
on(atom("pong"), uint32_t(0)) >> [=] {
send(parent, atom("done"));
quit();
},
on(atom("pong"), arg_match) >> [=](uint32_t value) {
reply(atom("ping"), value - 1);
},
others() >> [=] {
cout << "ping_actor: unexpected: "
<< to_string(last_dequeued())
<< endl;
}
);
},
others() >> [=] {
cout << "ping_actor: unexpected: "
<< to_string(last_dequeued())
<< endl;
}
);
}
};
struct server_actor : sb_actor<server_actor> {
typedef map<pair<string, uint16_t>, actor_ptr> pong_map;
behavior init_state;
pong_map m_pongs;
server_actor() {
trap_exit(true);
init_state = (
on(atom("ping"), arg_match) >> [=](uint32_t value) {
reply(atom("pong"), value);
},
on(atom("add_pong"), arg_match) >> [=](const string& host, uint16_t port) {
auto key = make_pair(host, port);
auto i = m_pongs.find(key);
if (i == m_pongs.end()) {
try {
auto p = remote_actor(host.c_str(), port);
link_to(p);
m_pongs.insert(make_pair(key, p));
reply(atom("ok"));
}
catch (exception& e) {
reply(atom("error"), e.what());
}
}
else {
reply(atom("ok"));
}
},
on(atom("kickoff"), arg_match) >> [=](uint32_t num_pings) {
for (auto& kvp : m_pongs) {
auto ping = spawn<ping_actor>(last_sender());
send(ping, atom("kickoff"), kvp.second, num_pings);
}
},
on(atom("purge")) >> [=] {
m_pongs.clear();
},
on<atom("EXIT"), uint32_t>() >> [=] {
actor_ptr who = last_sender();
auto i = find_if(m_pongs.begin(), m_pongs.end(),
[&](const pong_map::value_type& kvp) {
return kvp.second == who;
});
if (i != m_pongs.end()) m_pongs.erase(i);
},
on(atom("shutdown")) >> [=] {
m_pongs.clear();
quit();
},
others() >> [=] {
cout << "unexpected: " << to_string(last_dequeued()) << endl;
}
);
}
};
template<typename Arg0>
void usage(Arg0&& arg0) {
cout << forward<Arg0>(arg0) << endl << endl;
usage();
}
template<typename Arg0, typename Arg1, typename... Args>
void usage(Arg0&& arg0, Arg1&& arg1, Args&&... args) {
cout << forward<Arg0>(arg0);
usage(forward<Arg1>(arg1), forward<Args>(args)...);
}
template<typename Iterator>
void server_mode(Iterator first, Iterator last) {
string port_prefix = "--port=";
// extracts port from a key-value pair
auto kvp_port = [&](const string& str) -> option<int> {
if (equal(port_prefix.begin(), port_prefix.end(), str.begin())) {
return spro<int>(str.c_str() + port_prefix.size());
}
return {};
};
match(vector<string>{first, last}) ( (on(kvp_port) || on("-p", spro<int>)) >> [](int port) {
if (port > 1024 && port < 65536) {
publish(spawn<server_actor>(), port);
}
else {
usage("illegal port: ", port);
}
},
others() >> [=] {
if (first != last) usage("illegal argument: ", *first);
else usage();
}
);
}
template<typename Iterator>
void client_mode(Iterator first, Iterator last) {
if (first == last) usage("no server, no fun");
uint32_t init_value = 0;
vector<pair<string, uint16_t> > remotes;
string pings_prefix = "--num_pings=";
auto num_msgs = [&](const string& str) -> option<int> {
if (equal(pings_prefix.begin(), pings_prefix.end(), str.begin())) {
return spro<int>(str.c_str() + pings_prefix.size());
}
return {};
};
match_each(first, last, bind(split, std::placeholders::_1, ':')) (
on(val<string>, spro<int>) >> [&](string& host, int port) {
if (port <= 1024 || port >= 65536) {
throw invalid_argument("illegal port: " + to_string(port));
}
remotes.emplace_back(move(host), static_cast<uint16_t>(port));
},
on(num_msgs) >> [&](int num) {
if (num > 0) init_value = static_cast<uint32_t>(num);
}
);
if (init_value == 0) {
cout << "no non-zero, non-negative init value given" << endl;
exit(1);
}
if (remotes.size() < 2) {
cout << "less than two nodes given" << endl;
exit(1);
}
vector<actor_ptr> remote_actors;
for (auto& r : remotes) {
remote_actors.push_back(remote_actor(r.first.c_str(), r.second));
}
// setup phase
//cout << "tell server nodes to connect to each other" << endl;
for (size_t i = 0; i < remotes.size(); ++i) {
for (size_t j = 0; j < remotes.size(); ++j) {
if (i != j) {
auto& r = remotes[j];
send(remote_actors[i], atom("add_pong"), r.first, r.second);
}
}
}
{ // collect {ok} messages
size_t i = 0;
size_t end = remote_actors.size() * (remote_actors.size() - 1);
receive_for(i, end) (
on(atom("ok")) >> [] {
},
on(atom("error"), arg_match) >> [&](const string& str) {
cout << "error: " << str << endl;
for (auto& x : remote_actors) {
send(x, atom("purge"));
}
throw logic_error("");
},
others() >> [] {
cout << "expected {ok|error}, received: "
<< to_string(self->last_dequeued())
<< endl;
throw logic_error("");
},
after(chrono::seconds(10)) >> [&] {
cout << "remote didn't answer within 10sec." << endl;
for (auto& x : remote_actors) {
send(x, atom("purge"));
}
throw logic_error("");
}
);
}
// kickoff
//cout << "setup done" << endl;
//cout << "kickoff, init value = " << init_value << endl;
for (auto& r : remote_actors) {
send(r, atom("kickoff"), init_value);
}
{ // collect {done} messages
size_t i = 0;
size_t end = remote_actors.size() * (remote_actors.size() - 1);
receive_for(i, end) (
on(atom("done")) >> [] {
//cout << "...done..." << endl;
},
others() >> [] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
throw logic_error("");
}
);
}
}
template<typename Iterator>
void shutdown_mode(Iterator first, Iterator last) {
vector<pair<string, uint16_t> > remotes;
match_each(first, last, bind(split, std::placeholders::_1, ':')) (
on(val<string>, spro<int>) >> [&](string& host, int port) {
if (port <= 1024 || port >= 65536) {
throw invalid_argument("illegal port: " + to_string(port));
}
remotes.emplace_back(move(host), static_cast<uint16_t>(port));
}
);
for (auto& r : remotes) {
try {
actor_ptr x = remote_actor(r.first, r.second);
self->monitor(x);
send(x, atom("shutdown"));
receive (
on(atom("DOWN"), val<uint32_t>) >> [] {
// ok, done
},
after(chrono::seconds(10)) >> [&] {
cerr << r.first << ":" << r.second << " didn't shut down "
<< "within 10s"
<< endl;
}
);
}
catch (std::exception& e) {
cerr << "couldn't shutdown " << r.first << ":" << r.second
<< "; reason: " << e.what()
<< endl;
}
}
}
int main(int argc, char** argv) {
if (argc < 2) usage();
auto first = argv + 1;
auto last = argv + argc;
match(*first) (
on("mode=server") >> [=] {
server_mode(first + 1, last);
},
on("mode=benchmark") >> [=] {
client_mode(first + 1, last);
await_all_others_done();
},
on("mode=shutdown") >> [=] {
shutdown_mode(first + 1, last);
},
(on("-h") || on("--help")) >> [] {
usage();
},
others() >> [=] {
usage("unknown argument: ", *first);
}
);
await_all_others_done();
shutdown();
}
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <thread>
#include <cassert>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include "utility.hpp"
#include "cppa/cppa.hpp"
using namespace std;
using namespace cppa;
struct fsm_receiver : sb_actor<fsm_receiver> {
uint64_t m_value;
behavior init_state;
fsm_receiver(uint64_t max) : m_value(0) {
init_state = (
on(atom("msg")) >> [=] {
if (++m_value == max) {
quit();
}
}
);
}
};
void receiver(uint64_t max) {
uint64_t value;
receive_while (gref(value) < max) (
on(atom("msg")) >> [&] {
++value;
}
);
}
void sender(actor_ptr whom, uint64_t count) {
any_tuple msg = make_cow_tuple(atom("msg"));
for (uint64_t i = 0; i < count; ++i) {
whom->enqueue(nullptr, msg);
}
}
void usage() {
cout << "usage: mailbox_performance "
"(stacked|event-based) NUM_THREADS MSGS_PER_THREAD" << endl
<< endl;
}
enum impl_type { stacked, event_based };
option<impl_type> stoimpl(const std::string& str) {
if (str == "stacked") return stacked;
else if (str == "event-based") return event_based;
return {};
}
int main(int argc, char** argv) {
int result = 1;
vector<string> args(argv + 1, argv + argc);
match (args) (
on(stoimpl, spro<uint64_t>, spro<uint64_t>) >> [&](impl_type impl, uint64_t num_sender, uint64_t num_msgs) {
auto total = num_sender * num_msgs;
auto testee = (impl == stacked) ? spawn(receiver, total)
: spawn<fsm_receiver>(total);
vector<thread> senders;
for (uint64_t i = 0; i < num_sender; ++i) {
senders.emplace_back(sender, testee, num_msgs);
}
for (auto& s : senders) {
s.join();
}
result = 0; // no error occurred
},
others() >> usage
);
await_all_others_done();
shutdown();
return result;
}
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <list>
#include <string>
#include <cassert>
#include <cstdlib>
#include <cstring>
#include <cstdint>
#include <iostream>
#include "utility.hpp"
#include "cppa/cppa.hpp"
using namespace std;
using namespace cppa;
void usage() {
cout << "usage: matching (cow_tuple|object_array) NUM_LOOPS" << endl;
}
enum impl_type { cow_tuples, obj_arrays };
option<impl_type> implproj(const string& str) {
if (str == "cow_tuple") return cow_tuples;
else if (str == "object_array") return obj_arrays;
return {};
}
int main(int argc, char** argv) {
int result = 1;
announce<list<int>>();
vector<string> args(argv + 1, argv + argc);
match (args) (
on(implproj, spro<unsigned>) >> [&](impl_type impl, unsigned num_loops) {
result = 0;
any_tuple m1;
any_tuple m2;
any_tuple m3;
any_tuple m4;
any_tuple m5;
any_tuple m6;
if (impl == cow_tuples) {
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 {
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};
}
int64_t m1matched = 0;
int64_t m2matched = 0;
int64_t m3matched = 0;
int64_t m4matched = 0;
int64_t m5matched = 0;
int64_t m6matched = 0;
auto part_fun = (
on<atom("msg1"), int>() >> [&]() { ++m1matched; },
on<atom("msg2"), double>() >> [&]() { ++m2matched; },
on<atom("msg3"), list<int> >() >> [&]() { ++m3matched; },
on<atom("msg4"), int, string>() >> [&]() { ++m4matched; },
on<atom("msg5"), int, int, int>() >> [&]() { ++m5matched; },
on<atom("msg6"), int, double, string>() >> [&]() { ++m6matched; }
);
for (int64_t i = 0; i < num_loops; ++i) {
part_fun(m1);
part_fun(m2);
part_fun(m3);
part_fun(m4);
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);
result = 0;
},
others() >> usage
);
shutdown();
return result;
}
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <cassert>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include "utility.hpp"
#include "cppa/cppa.hpp"
using namespace std;
using namespace cppa;
typedef vector<uint64_t> factors;
constexpr uint64_t s_task_n = uint64_t(86028157)*329545133;
constexpr uint64_t s_factor1 = 86028157;
constexpr uint64_t s_factor2 = 329545133;
void check_factors(const factors& vec) {
assert(vec.size() == 2);
assert(vec[0] == s_factor1);
assert(vec[1] == s_factor2);
# ifdef NDEBUG
static_cast<void>(vec);
# endif
}
struct fsm_worker : sb_actor<fsm_worker> {
actor_ptr mc;
behavior init_state;
fsm_worker(const actor_ptr& msgcollector) : mc(msgcollector) {
init_state = (
on<atom("calc"), uint64_t>() >> [=](uint64_t what) {
send(mc, atom("result"), factorize(what));
},
on(atom("done")) >> [=]() {
quit();
}
);
}
};
struct fsm_chain_link : sb_actor<fsm_chain_link> {
actor_ptr next;
behavior init_state;
fsm_chain_link(const actor_ptr& n) : next(n) {
init_state = (
on<atom("token"), int>() >> [=](int v) {
next << std::move(last_dequeued());
if (v == 0) quit();
}
);
}
};
struct fsm_chain_master : sb_actor<fsm_chain_master> {
int iteration;
actor_ptr mc;
actor_ptr next;
actor_ptr worker;
behavior init_state;
void new_ring(int ring_size, int initial_token_value) {
send(worker, atom("calc"), s_task_n);
next = self;
for (int i = 1; i < ring_size; ++i) {
next = spawn<fsm_chain_link>(next);
}
send(next, atom("token"), initial_token_value);
}
fsm_chain_master(actor_ptr msgcollector) : iteration(0), mc(msgcollector) {
init_state = (
on(atom("init"), arg_match) >> [=](int rs, int itv, int n) {
worker = spawn<fsm_worker>(msgcollector);
iteration = 0;
new_ring(rs, itv);
become (
on(atom("token"), 0) >> [=]() {
if (++iteration < n) {
new_ring(rs, itv);
}
else {
send(worker, atom("done"));
send(mc, atom("masterdone"));
quit();
}
},
on<atom("token"), int>() >> [=](int v) {
send(next, atom("token"), v - 1);
}
);
}
);
}
};
struct fsm_supervisor : sb_actor<fsm_supervisor> {
int left;
behavior init_state;
fsm_supervisor(int num_msgs) : left(num_msgs) {
init_state = (
on(atom("masterdone")) >> [=]() {
if (--left == 0) quit();
},
on<atom("result"), factors>() >> [=](const factors& vec) {
check_factors(vec);
if (--left == 0) quit();
}
);
}
};
void chain_link(actor_ptr next) {
bool done = false;
do_receive (
on<atom("token"), int>() >> [&](int v) {
if (v == 0) {
done = true;
}
next << std::move(self->last_dequeued());
}
)
.until([&]() { return done == true; });
}
void worker_fun(actor_ptr msgcollector) {
bool done = false;
do_receive (
on<atom("calc"), uint64_t>() >> [&](uint64_t what) {
send(msgcollector, atom("result"), factorize(what));
},
on(atom("done")) >> [&]() {
done = true;
}
)
.until([&]() { return done == true; });
}
actor_ptr new_ring(actor_ptr next, int ring_size) {
for (int i = 1; i < ring_size; ++i) next = spawn(chain_link, next);
return next;
}
void chain_master(actor_ptr msgcollector) {
auto worker = spawn(worker_fun, msgcollector);
receive (
on(atom("init"), arg_match) >> [&](int rs, int itv, int n) {
int iteration = 0;
auto next = new_ring(self, rs);
send(next, atom("token"), itv);
send(worker, atom("calc"), s_task_n);
do_receive (
on<atom("token"), int>() >> [&](int v) {
if (v == 0) {
if (++iteration < n) {
next = new_ring(self, rs);
send(next, atom("token"), itv);
send(worker, atom("calc"), s_task_n);
}
}
else {
send(next, atom("token"), v - 1);
}
}
)
.until([&]() { return iteration == n; });
}
);
send(msgcollector, atom("masterdone"));
send(worker, atom("done"));
}
void supervisor(int num_msgs) {
do_receive (
on(atom("masterdone")) >> [&]() {
--num_msgs;
},
on<atom("result"), factors>() >> [&](const factors& vec) {
--num_msgs;
check_factors(vec);
}
)
.until([&]() { return num_msgs == 0; });
}
template<typename F>
void run_test(F spawn_impl,
int num_rings, int ring_size,
int initial_token_value, int repetitions) {
std::vector<actor_ptr> masters; // of the universe
// each master sends one masterdone message and one
// factorization is calculated per repetition
//auto supermaster = spawn(supervisor, num_rings+repetitions);
for (int i = 0; i < num_rings; ++i) {
masters.push_back(spawn_impl());
send(masters.back(), atom("init"),
ring_size,
initial_token_value,
repetitions);
}
await_all_others_done();
}
void usage() {
cout << "usage: mailbox_performance "
"(stacked|event-based) (num rings) (ring size) "
"(initial token value) (repetitions)"
<< endl
<< endl;
exit(1);
}
enum impl_type { stacked, event_based };
option<impl_type> stoimpl(const std::string& str) {
if (str == "stacked") return stacked;
else if (str == "event-based") return event_based;
return {};
}
int main(int argc, char** argv) {
announce<factors>();
int result = 1;
std::vector<std::string> args{argv + 1, argv + argc};
match(args) (
on(stoimpl, spro<int>, spro<int>, spro<int>, spro<int>)
>> [&](impl_type impl, int num_rings, int ring_size, int initial_token_value, int repetitions) {
int num_msgs = num_rings + (num_rings * repetitions);
auto sv = (impl == event_based) ? spawn<fsm_supervisor>(num_msgs)
: spawn(supervisor, num_msgs);
if (impl == event_based) {
run_test([sv] { return spawn<fsm_chain_master>(sv); },
num_rings, ring_size, initial_token_value, repetitions);
}
else {
run_test([sv] { return spawn(chain_master, sv); },
num_rings, ring_size, initial_token_value, repetitions);
}
result = 0; // no error occurred
},
others() >> usage
);
return result;
}
#!/bin/bash
echo "../build/bin/$@" | ./exec.sh
FILES=actor_creation.erl distributed.erl mailbox_performance.erl matching.erl mixed_case.erl
BEAMS=$(FILES:.erl=.beam)
ERLC=erlc
%.beam: %.erl
$(ERLC) $<
all: $(BEAMS)
clean:
rm -f $(BEAMS)
.PHONY: all clean
-module(actor_creation).
-export([start/1, testee/1]).
testee(Pid) ->
receive
{spread, 0} ->
Pid ! {result, 1};
{spread, X} ->
spawn(actor_creation, testee, [self()]) ! {spread, X-1},
spawn(actor_creation, testee, [self()]) ! {spread, X-1},
receive
{result, R1} ->
receive
{result, R2} ->
Pid ! {result, (R1+R2)}
end
end
end.
start(X) ->
[H|_] = X,
N = list_to_integer(atom_to_list(H)),
spawn(actor_creation, testee, [self()]) ! {spread, N},
receive
{result, R} ->
if
R == (1 bsl N) ->
R;
true ->
error("unexpected result!")
end
end.
-module(distributed).
-export([start/1, ping_loop/2]).
ping_loop(Parent, Pong) ->
receive
{pong, 0} -> Parent ! done;
{pong, X} ->
Pong ! {ping, self(), X - 1},
ping_loop(Parent, Pong);
{kickoff, X} ->
Pong ! {ping, self(), X},
ping_loop(Parent, Pong);
_ -> ping_loop(Parent, Pong)
end.
server_loop(Pongs) ->
receive
{ping, Pid, X} ->
Pid ! {pong, X},
server_loop(Pongs);
{add_pong, Pid, Node} ->
case lists:any(fun({N, _}) -> N == Node end, Pongs) of
true ->
Pid ! {ok, cached},
server_loop(Pongs);
false ->
case rpc:call(Node, erlang, whereis, [pong]) of
{badrpc, Reason} ->
Pid ! {error, Reason},
server_loop(Pongs);
undefined ->
Pid ! {error, 'pong is undefined'},
server_loop(Pongs);
Pong ->
Pid ! {ok, added},
server_loop(Pongs ++ [{Node, Pong}])
end
end;
{purge} -> server_loop([]);
{get_pongs, Pid} ->
Pid ! Pongs,
server_loop(Pongs);
{kickoff, Pid, NumPings} ->
lists:foreach(fun({_, P}) -> spawn(distributed, ping_loop, [Pid, P]) ! {kickoff, NumPings} end, Pongs),
server_loop(Pongs);
_ -> server_loop(Pongs)
end.
server_mode() ->
register(pong, self()),
server_loop([]).
add_pong_fun(_, _, []) -> true;
add_pong_fun(Pong, Node, [Node|T]) -> add_pong_fun(Pong, Node, T);
add_pong_fun(Pong, Node, [H|T]) ->
Pong ! {add_pong, self(), H},
receive
{ok, _} -> add_pong_fun(Pong, Node, T);
{error, Reason} -> error(Reason)
after 10000 -> error(timeout)
end.
client_mode_receive_done_msgs(0) -> true;
client_mode_receive_done_msgs(Left) ->
receive done -> client_mode_receive_done_msgs(Left - 1) end.
% receive a {done} message for each node
client_mode([], [], [], _) -> error("no node, no fun");
client_mode([], [], Nodes, _) ->
client_mode_receive_done_msgs(length(Nodes) * (length(Nodes) - 1));
% send kickoff messages
client_mode([Pong|Pongs], [], Nodes, NumPings) ->
Pong ! {kickoff, self(), NumPings},
client_mode(Pongs, [], Nodes, NumPings);
client_mode(Pongs, [H|T], Nodes, NumPings) ->
case rpc:call(H, erlang, whereis, [pong]) of
{badrpc, Reason} ->
io:format("cannot connect to ~s~n", [atom_to_list(H)]),
error(Reason);
undefined ->
io:format("no 'pong' defined on node ~s~n", [atom_to_list(H)]);
P ->
add_pong_fun(P, H, Nodes),
client_mode(Pongs ++ [P], T, Nodes, NumPings)
end.
run(_, undefined, []) -> error("NumPings is undefined");
run(Hosts, _, []) when length(Hosts) < 2 -> error("less than two nodes specified");
run(Hosts, NumPings, []) -> client_mode([], Hosts, Hosts, NumPings);
run(Hosts, NumPings, [H|T]) ->
Arg = atom_to_list(H),
case lists:prefix("num_pings=", Arg) of
true when NumPings /= undefined -> error("NumPings already set");
true -> run(Hosts, list_to_integer(lists:sublist(Arg, 11, length(Arg))), T);
false -> run(Hosts ++ [H], NumPings, T)
end.
start(X) ->
case X of
['mode=server'|[]] -> server_mode();
['mode=server'|_] -> io:format("too much arguments~n", []);
['mode=benchmark'|T] -> run([], undefined, T);
_ -> io:format("invalid arguments~n", [])
end.
-module(mailbox_performance).
-export([start/1,sender/2]).
sender(_, 0) ->
0;
sender(Pid, Num) ->
Pid ! {msg},
sender(Pid, Num-1).
start_senders(_, _, 0) ->
0;
start_senders(Pid, Arg, Num) ->
spawn(mailbox_performance, sender, [Pid, Arg]),
start_senders(Pid, Arg, Num-1).
receive_messages(0) ->
0;
receive_messages(Num) ->
receive
{msg} -> receive_messages(Num-1)
end.
start(Args) ->
[H1|T] = Args,
NumSender = list_to_integer(atom_to_list(H1)),
[H2|_] = T,
NumMsgs = list_to_integer(atom_to_list(H2)),
start_senders(self(), NumMsgs, NumSender),
receive_messages(NumMsgs * NumSender).
-module(matching).
-export([start/1]).
partFun(msg1, X) when X =:= 0 -> true;
partFun(msg2, 0.0) -> true;
partFun(msg3, [0|[]]) -> true.
partFun(msg4, 0, "0") -> true.
partFun(msg5, A, B, C) when A =:= 0; B =:= 0; C =:= 0 -> true;
partFun(msg6, A, 0.0, "0") when A =:= 0 -> true.
loop(0) -> true;
loop(N) ->
partFun(msg1, 0),
partFun(msg2, 0.0),
partFun(msg3, [0]),
partFun(msg4, 0, "0"),
partFun(msg5, 0, 0, 0),
partFun(msg6, 0, 0.0, "0"),
loop(N-1).
start(X) ->
[H|[]] = X,
N = list_to_integer(atom_to_list(H)),
loop(N).
-module(mixed_case).
-export([start/1, master/1, worker/1, chain_link/1]).
%next_factor(N, M) when N rem M =:= 0 -> M;
%next_factor(N, M) -> next_factor(N, M+1).
%factorize(1, _) -> [];
%factorize(N, M) ->
% X = next_factor(N,M),
% [ X | factorize(N div X, X) ].
%factorize(1) -> [];
%factorize(N) -> factorize(N,2).
factorize(N,M,R) when N =:= M -> R ++ [M];
factorize(N,M,R) when (N rem M) =:= 0 -> factorize(N div M, M, R ++ [M]);
factorize(N,2,R) -> factorize(N,3,R);
factorize(N,M,R) -> factorize(N,M+2,R).
factorize(1) -> [1];
factorize(2) -> [2];
factorize(N) -> factorize(N,2,[]).
worker(Supervisor) ->
receive
{calc, Val} ->
Supervisor ! {result, factorize(Val)},
worker(Supervisor)
end.
chain_link(Next) ->
receive
{token, 0} ->
Next ! {token, 0},
true;
{token, Val} ->
Next ! {token, Val},
chain_link(Next)
end.
cr_ring(Next, InitialTokenValue, 0) ->
Next ! {token, InitialTokenValue},
Next;
cr_ring(Next, InitialTokenValue, RingSize) ->
cr_ring(spawn(mixed_case, chain_link, [Next]), InitialTokenValue, RingSize-1).
master_recv_tokens(Next) ->
receive
{token, 0} -> true;
{token, Val} ->
Next ! {token, Val-1},
master_recv_tokens(Next)
end.
master_bhvr(Supervisor, _, _, _, 0) ->
Supervisor ! master_done;
master_bhvr(Supervisor, Worker, RingSize, InitialTokenValue, Repetitions) ->
Worker ! {calc, 28350160440309881},
master_recv_tokens(cr_ring(self(), InitialTokenValue, RingSize-1)),
master_bhvr(Supervisor, Worker, RingSize, InitialTokenValue, Repetitions-1).
master(Supervisor) ->
receive
{init, RingSize, InitialTokenValue, Repetitions} ->
master_bhvr(Supervisor, spawn(mixed_case, worker, [Supervisor]), RingSize, InitialTokenValue, Repetitions)
end.
check_result(List) ->
if
List == [329545133,86028157] -> true;
List == [86028157,329545133] -> true;
true -> error("wrong factorization result")
end.
wait4all(0) -> true;
wait4all(Remaining) ->
receive
master_done -> wait4all(Remaining-1);
{result, List} ->
check_result(List),
wait4all(Remaining-1)
end.
cr_masters(_, _, _, 0) ->
true;
cr_masters(RingSize, InitialTokenValue, Repetitions, NumRings) ->
spawn(mixed_case, master, [self()]) ! {init, RingSize, InitialTokenValue, Repetitions},
cr_masters(RingSize, InitialTokenValue, Repetitions, NumRings-1).
start(X) ->
[H0|T0] = X,
NumRings = list_to_integer(atom_to_list(H0)),
[H1|T1] = T0,
RingSize = list_to_integer(atom_to_list(H1)),
[H2|T2] = T1,
InitialTokenValue = list_to_integer(atom_to_list(H2)),
[H3|_] = T2,
Repetitions = list_to_integer(atom_to_list(H3)),
cr_masters(RingSize, InitialTokenValue, Repetitions, NumRings),
% each master sends one master_done message and one
% factorization is calculated per repetition
wait4all(NumRings+(NumRings*Repetitions)).
#!/bin/bash
echo "erl -noshell -noinput +P 20000000 -setcookie abc123 -sname benchmark -pa erlang -s $@ -s init stop" | ./exec.sh
#!/bin/bash
read -r cmd
export JAVA_OPTS="-Xmx4096M"
if [[ $(uname) == "Darwin" ]] ; then
/usr/bin/time -p $cmd 2>&1
else
/usr/bin/time -p -f "%e" $cmd 2>&1
fi
#!/bin/bash
$@
# &>/dev/null
#include <process/process.hpp>
#include <process/dispatch.hpp>
#include "utility.hpp"
typedef std::vector<uint64_t> factors;
constexpr uint64_t s_task_n = uint64_t(86028157)*329545133;
constexpr uint64_t s_factor1 = 86028157;
constexpr uint64_t s_factor2 = 329545133;
void check_factors(const factors& vec)
{
assert(vec.size() == 2);
assert(vec[0] == s_factor1);
assert(vec[1] == s_factor2);
}
class supervisor : public process::Process<supervisor>
{
public:
supervisor(int num_msgs)
: left_(num_msgs)
{
}
void subtract()
{
if (--left_ == 0)
process::terminate(self());
}
void check(const factors& vec)
{
check_factors(vec);
if (--left_ == 0)
process::terminate(self());
}
private:
int left_;
};
class chain_link : public process::Process<chain_link>
{
public:
chain_link()
{
}
chain_link(process::PID<chain_link> pid)
: next_(std::move(pid))
{
}
virtual ~chain_link() { }
virtual void token(int v)
{
assert(next_);
process::dispatch(next_, &chain_link::token, v);
if (v == 0)
process::terminate(self());
}
private:
process::PID<chain_link> next_;
};
class worker : public process::Process<worker>
{
public:
worker(process::PID<supervisor> collector)
: collector_(std::move(collector))
{
}
void calc(uint64_t what)
{
process::dispatch(collector_, &supervisor::check, factorize(what));
}
void done()
{
process::terminate(self());
}
private:
process::PID<supervisor> collector_;
};
class chain_master : public chain_link
{
public:
chain_master(process::PID<supervisor> collector, int rs, int itv, int n)
: chain_link()
, ring_size_(rs)
, initial_value_(itv)
, repetitions_(n)
, iteration_(0)
, collector_(collector)
{
}
void new_ring(int ring_size, int initial_token_value)
{
process::dispatch(worker_, &worker::calc, s_task_n);
next_ = self();
for (int i = 1; i < ring_size; ++i)
next_ = process::spawn(new chain_link(next_), true);
process::dispatch(next_, &chain_link::token, initial_token_value);
}
void init()
{
worker_ = process::spawn(new worker(collector_), true);
new_ring(ring_size_, initial_value_);
}
virtual void token(int t)
{
if (t == 0)
{
if (++iteration_ < repetitions_)
{
new_ring(ring_size_, initial_value_);
}
else
{
dispatch(worker_, &worker::done);
dispatch(collector_, &supervisor::subtract);
process::terminate(self());
}
}
else
{
process::dispatch(next_, &chain_link::token, t - 1);
}
}
private:
const int ring_size_;
const int initial_value_;
const int repetitions_;
int iteration_;
process::PID<supervisor> collector_;
process::PID<chain_link> next_;
process::PID<worker> worker_;
};
int main(int argc, char** argv)
{
if (argc != 5)
{
std::cout << "usage " << argv[0] << ": " <<
"(num rings) (ring size) (initial token value) (repetitions)"
<< std::endl;
return 1;
}
auto iter = argv;
++iter; // argv[0] (app name)
int num_rings = rd<int>(*iter++);
int ring_size = rd<int>(*iter++);
int initial_token_value = rd<int>(*iter++);
int repetitions = rd<int>(*iter++);
int num_msgs = num_rings + (num_rings * repetitions);
auto mc = process::spawn(new supervisor(num_msgs), true);
std::vector<process::PID<chain_master>> masters;
for (int i = 0; i < num_rings; ++i)
{
auto master = process::spawn(
new chain_master(mc, ring_size, initial_token_value, repetitions),
true);
process::dispatch(master, &chain_master::init);
masters.push_back(master);
}
for (auto& m : masters)
process::wait(m);
return 0;
}
#!/bin/bash
# quick & dirty: eval args to enable CXX=... and CXXFLAGS=...
for arg in "$@"; do
eval "$arg"
done
if test "" = "$CXX"; then
CXX=g++
fi
function verbose_exec {
echo $1
eval $1
if test "0" != "$?" ; then
echo ; echo "command failed!"
exit
fi
}
function check_var {
if test "" = "$1" ; then
echo "something went wrong ... $2 not found"
exit
fi
}
function fetch_lib {
find mesos/third_party/libprocess -name "$1"
if test "$2" = "true" ; then
echo "something went wrong ... $1 not found"
fi
}
if ! test -d mesos ; then
echo "fetch mesos (for libprocess)"
verbose_exec "git clone https://github.com/apache/mesos.git"
else
echo "found mesos repository"
fi
LIBFILE=$(fetch_lib libprocess.a)
LIBGLOGFILE=$(fetch_lib libglog.a)
LIBEVFILE=$(fetch_lib libev.a)
echo "LIBFILE=$LIBFILE"
echo "LIBGLOGFILE=$LIBGLOGFILE"
echo "LIBEVFILE=$LIBEVFILE"
if test "" = "$LIBFILE" -o "" = "$LIBGLOGFILE" -o "" = "$LIBEVFILE" ; then
CURR=$PWD
echo "build libprocess"
cd mesos/third_party/libprocess/
verbose_exec "autoreconf -Wnone -i && ./configure CXX=\"$CXX\" CXXFLAGS=\"$CXXFLAGS -D_XOPEN_SOURCE\" && make"
echo "build third_party libraries"
cd third_party
for dir in * ; do
if test -d $dir ; then
# skip boost in third_party directory
if [[ "$dir" == glog* ]] || [[ "$dir" == libev* ]] ; then
echo "cd $dir"
cd $dir
verbose_exec "autoreconf -i ; ./configure CXX=\"$CXX\" CXXFLAGS=\"$CXXFLAGS\" ; make"
echo "cd .."
cd ..
fi
fi
done
cd $CURR
LIBFILE=$(fetch_lib libprocess.a true)
LIBGLOGFILE=$(fetch_lib libglog.a true)
LIBEVFILE=$(fetch_lib libev.a true)
else
echo "found libprocess library: $LIBFILE"
fi
LIBDIRS="-L$(dirname "$LIBFILE") -L$(dirname "$LIBGLOGFILE") -L$(dirname "$LIBEVFILE")"
verbose_exec "$CXX --std=c++0x $CXXFLAGS -Imesos/third_party/libprocess/include/ $LIBDIRS mixed_case_libprocess.cpp -o mixed_case_libprocess -lprocess -lglog -lev"
#!/bin/bash
IFLAGS="-I../../Theron/Include/"
LFLAGS="-L../../Theron/Lib -ltheron -lboost_thread"
FLAGS="-O3 -fno-strict-aliasing -DNDEBUG -DTHERON_MAX_ACTORS=530000"
if [[ $# -eq 0 ]] ; then
for i in theron_*.cpp ; do
out_file=$(echo $i | sed 's/\(.*\)\..*/\1/')
echo "g++ -std=c++0x $i -o $out_file $FLAGS $LFLAGS $IFLAGS"
g++ -std=c++0x $i -o $out_file $FLAGS $LFLAGS $IFLAGS
done
elif [[ $# -eq 1 ]] ; then
echo "g++ -std=c++0x theron_$1.cpp -o theron_$1 $FLAGS $LFLAGS $IFLAGS"
g++ -std=c++0x theron_$1.cpp -o theron_$1 $FLAGS $LFLAGS $IFLAGS
fi
#!/bin/bash
usage="\
Usage: $0 INTERVAL CMD
example: $0 0.5 \"my_benchmark_program arg0 arg1 arg2 \"
"
if [ "$#" != "2" ]; then
echo "${usage}" 1>&2
exit
fi
runtime=""
line=$(cat /proc/1/status | grep "^Vm" | awk 'BEGIN {printf "%-10s", "elapsed"} {printf "%-20s", $1} END {print ""}')
exec $2 &
pid=$!
interval=$1
dtime=$(date "+%s%N")
time0=$(echo "$dtime / 1000000" | bc -q)
fields=$(echo "$line" | wc -w)
while [ "$fields" -gt "1" ]; do
echo "$line"
if [ -n "$runtime" ]; then
sleep $interval
runtime=$(echo $runtime + $interval | bc -q)
else
runtime=0
fi
dtime=$(date "+%s%N")
timeN=$(echo "( $dtime / 1000000 ) - $time0" | bc -q)
timeN_sec=$(echo "scale=2; $timeN / 1000" | bc)
line=$(cat /proc/${pid}/status 2>/dev/null | grep "^Vm" | awk -v rt=$timeN_sec 'BEGIN {printf "%-10s", rt} {printf "%-20s", ($2 " " $3)} END {print ""}')
fields=$(echo "$line" | wc -w)
done
#!/bin/bash
NumCores=$(cat /proc/cpuinfo | grep processor | wc -l)
if [ $NumCores -lt 10 ]; then
NumCores="0$NumCores"
fi
declare -A impls
impls["cppa"]="event-based stacked"
impls["scala"]="akka threaded threadless"
impls["erlang"]="start"
declare -A benchmarks
benchmarks["cppa.mixed_case"]="mixed_case"
benchmarks["cppa.actor_creation"]="actor_creation"
benchmarks["cppa.mailbox_performance"]="mailbox_performance"
benchmarks["scala.mixed_case"]="MixedCase"
benchmarks["scala.actor_creation"]="ActorCreation"
benchmarks["scala.mailbox_performance"]="MailboxPerformance"
benchmarks["erlang.mixed_case"]="mixed_case"
benchmarks["erlang.actor_creation"]="actor_creation"
benchmarks["erlang.mailbox_performance"]="mailbox_performance"
declare -A bench_args
# 20 rings, 50 actors in each ring, initial token value = 10000, 5 repetitions
bench_args["mixed_case"]="20 50 10000 5"
# 2^19 actors
bench_args["actor_creation"]="19"
# 20 threads, 1,000,000 each = 20,000,000 messages
bench_args["mailbox_performance"]="20 1000000"
for Lang in "cppa" "scala" "erlang" ; do
for Bench in "actor_creation" "mailbox_performance" "mixed_case" ; do
Args=${bench_args[$Bench]}
BenchName=${benchmarks["$Lang.$Bench"]}
for Impl in ${impls[$Lang]}; do
if [[ "$Lang.$Bench.$Impl" != "scala.actor_creation.threaded" ]] ; then
echo "$Lang: $Impl $Bench ..." >&2
FileName="$NumCores cores, $Lang $Impl $Bench.txt"
exec 1>>"$FileName"
for i in {1..5} ; do
./${Lang}_test.sh $BenchName $Impl $Args
done
fi
done
done
done
exit
for Impl in "event-based" "stacked" ; do
# actor creation performance
FileName="cppa 2^19 $Impl actors, $NumCores cores.txt"
echo "cppa performance: 2^19 $Impl actors..." >&2
exec 1>"$FileName"
for j in {1..10} ; do
./cppa_test.sh actor_creation $Impl 19
done
# mailbox performance
FileName="cppa 20*1,000,000 messages $Impl, $NumCores cores.txt"
echo "cppa mailbox performance: $Impl..." >&2
exec 1>"$FileName"
for j in {1..10} ; do
./cppa_test.sh mailbox_performance $Impl 20 1000000
done
done
for Impl in "threaded" "akka" "threadless" ; do
# actor creation performance
FileName="scala 2^19 $Impl actors, $NumCores cores.txt"
echo "scala performance: 2^19 $Impl actors" >&2
exec 1>"$FileName"
for j in {1..10} ; do
./scala_test.sh ActorCreation $Impl 19
done
# mailbox performance
FileName="scala 20*1,000,000 messages $Impl, $NumCores cores.txt"
echo "scala mailbox performance: $Impl..." >&2
exec 1>"$FileName"
for j in {1...10} ; do
./scala_test.sh MailboxPerformance $Impl 20 1000000
done
done
package org.libcppa.actor_creation
import org.libcppa.utility.IntStr
import scala.actors.Actor
import scala.actors.Actor._
import akka.actor.{ Props, Actor => AkkaActor, ActorRef => AkkaActorRef, ActorSystem }
case object GoAhead
case class Spread(value: Int)
case class Result(value: Int)
object global {
val latch = new java.util.concurrent.CountDownLatch(1)
}
class ThreadedTestee(parent: Actor) extends Actor {
def act() {
receive {
case Spread(s) =>
if (s > 0) {
val next = Spread(s-1)
(new ThreadedTestee(this)).start ! next
(new ThreadedTestee(this)).start ! next
receive {
case Result(v1) =>
receive {
case Result(v2) =>
parent ! Result(v1 + v2)
}
}
}
else {
parent ! Result(1)
}
}
}
}
class ThreadlessTestee(parent: Actor) extends Actor {
def act() {
react {
case Spread(s) =>
if (s > 0) {
val next = Spread(s-1)
(new ThreadlessTestee(this)).start ! next
(new ThreadlessTestee(this)).start ! next
react {
case Result(v1) =>
react {
case Result(v2) =>
parent ! Result(v1 + v2)
}
}
}
else {
parent ! Result(1)
}
}
}
}
class AkkaTestee(parent: AkkaActorRef) extends AkkaActor {
def receive = {
case Spread(0) =>
parent ! Result(1)
context.stop(self)
case Spread(s) =>
val msg = Spread(s-1)
context.actorOf(Props(new AkkaTestee(self))) ! msg
context.actorOf(Props(new AkkaTestee(self))) ! msg
context.become {
case Result(r1) =>
context.become {
case Result(r2) =>
parent ! Result(r1 + r2)
context.stop(self)
}
}
}
}
class AkkaRootTestee(n: Int) extends AkkaActor {
def receive = {
case GoAhead =>
context.actorOf(Props(new AkkaTestee(self))) ! Spread(n)
case Result(v) =>
if (v != (1 << n)) {
Console.println("Expected " + (1 << n) + ", received " + v)
System.exit(42)
}
global.latch.countDown
context.stop(self)
}
}
class ActorCreation(n: Int) {
def runThreaded() {
val newMax = (1 << n) + 100
System.setProperty("actors.maxPoolSize", newMax.toString)
(new ThreadedTestee(self)).start ! Spread(n)
receive {
case Result(v) =>
if (v != (1 << n))
Console.println("ERROR: expected " + (1 << n) + ", received " + v)
}
}
def runThreadless() {
actor {
(new ThreadlessTestee(self)).start ! Spread(n)
react {
case Result(v) =>
if (v != (1 << n))
Console.println("ERROR: expected " + (1 << n) + ", received " + v)
}
}
}
def runAkka() {
val system = ActorSystem()
system.actorOf(Props(new AkkaRootTestee(n))) ! GoAhead
global.latch.await
system.shutdown
System.exit(0)
}
}
object Main {
def usage() {
Console println "usage: (threaded|threadless|akka) POW\n creates 2^POW actors of given impl"
}
def main(args: Array[String]): Unit = args match {
case Array(impl, IntStr(n)) => {
val prog = new ActorCreation(n)
impl match {
case "threaded" => prog.runThreaded
case "threadless" => prog.runThreadless
case "akka" => prog.runAkka
case _ => usage
}
}
case _ => usage
}
}
package org.libcppa.distributed
import org.libcppa.utility._
import scala.actors._
import scala.actors.Actor.self
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 Console.println
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 Hello(token: String)
case class Olleh(token: String)
case class Error(msg: String, token: String)
case class AddPong(path: String, token: String)
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 {
private var left = pongs.length
private def recvLoop: Nothing = react {
case Pong(0) => {
parent ! Done
if (left > 1) {
left -= 1
recvLoop
}
}
case Pong(value) => sender ! Ping(value - 1); recvLoop
}
override def act() = react {
case KickOff(value) => pongs.foreach(_ ! Ping(value)); recvLoop
}
}
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 with ServerActorPrototype[Peers] {
//def reply(what: Any): Unit = sender ! what // inherited from ReplyReactor
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))
}
}
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)
}
}
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
alive(port)
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 => {
val pong = select(new Node(x.host, x.port), 'Pong)
pongPaths foreach (y => if (x != y) pong ! AddPong(y.uri, x.uri + " -> " + y.uri))
pong
})
@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)
// kickoff
pongs foreach (_ ! KickOff(numPings))
// collect done messages
for (_ <- 1 until (pongs.length * (pongs.length - 1))) {
receiveWithin(30*60*1000) {
case Done => Unit
case TIMEOUT => throw new RuntimeException("no Done within 30min")
case x => throw new RuntimeException("Unexpected message: " + x.toString)
}
}
}
}
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
//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)
}
}
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
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))
}
}
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)
}
}
def bhvr(peers: AkkaPeers): Receive = {
case x => {
recvFun(peers)(x) match {
case (true, newPeers) => become(bhvr(newPeers))
case _ => Unit
}
}
}
def receive = bhvr(AkkaPeers(Nil, Nil))
}
case class TokenTimeout(token: String)
case class RunAkkaClient(paths: List[String], numPings: Int)
class AkkaClientActor(system: ActorSystem) extends AkkaActor {
import context.become
def collectDoneMessages(left: Int): Receive = {
case Done => {
//println("Done")
if (left == 1) {
global.latch.countDown
context.stop(self)
} else {
become(collectDoneMessages(left - 1))
}
}
case _ => {
// ignore any other message
}
}
def collectOkMessages(pongs: List[AkkaActorRef], left: Int, receivedTokens: List[String], numPings: Int): Receive = {
case Ok(token) => {
//println("Ok")
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, left - 1, token :: receivedTokens, numPings))
}
}
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) => {
println("Error [from " + token+ "]: " + what)
global.latch.countDown
context.stop(self)
}
}
def receive = {
case RunAkkaClient(paths, numPings) => {
//println("RunAkkaClient(" + paths.toString + ", " + numPings + ")")
import akka.util.duration._
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))
}
}
}
class Distributed {
def runServer(port: Int) {
(new ServerActor(port)).start
}
def runAkkaServer() {
val system = ActorSystem("pongServer", ConfigFactory.load.getConfig("pongServer"))
system.actorOf(Props(new AkkaServerActor(system)), "pong")
}
//private val NumPings = "num_pings=([0-9]+)".r
private val SimpleUri = "([0-9a-zA-Z\\.]+):([0-9]+)".r
@tailrec private def run(args: List[String], paths: List[String], numPings: Option[Int], finalizer: (List[String], Int) => Unit): Unit = args match {
case KeyValuePair("num_pings", IntStr(num)) :: tail => numPings match {
case Some(x) => throw new IllegalArgumentException("\"num_pings\" already defined, first value = " + x + ", second value = " + num)
case None => run(tail, paths, Some(num), finalizer)
}
case arg :: tail => run(tail, arg :: paths, numPings, finalizer)
case Nil => numPings match {
case Some(x) => {
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 runBenchmark(args: List[String]) {
run(args, Nil, None, ((paths, x) => {
(new ClientActor(paths map (path => path match { case SimpleUri(host, port) => RemoteActorPath(path, host, port.toInt) }), x)).start
}))
}
def runAkkaBenchmark(args: List[String]) {
run(args, 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)
}))
}
}
object Main {
val prog = new Distributed
def main(args: Array[String]): Unit = args match {
// server mode
case Array("mode=server", "remote_actors", IntStr(port)) => prog.runServer(port)
case Array("mode=server", "akka") => prog.runAkkaServer
// client mode
case Array("mode=benchmark", "remote_actors", _*) => prog.runBenchmark(args.toList.drop(2))
case Array("mode=benchmark", "akka", _*) => prog.runAkkaBenchmark(args.toList.drop(2))
// error
case _ => {
println("Running in server mode:\n" +
" mode=server\n" +
" remote_actors PORT *or* akka\n" +
"\n" +
"Running the benchmark:\n" +
" mode=benchmark\n" +
" remote_actors ... *or* akka\n" );
}
}
}
package org.libcppa.mailbox_performance
import org.libcppa.utility._
import scala.actors.Actor
import scala.actors.Actor._
import akka.actor.{ Props, Actor => AkkaActor, ActorRef => AkkaActorRef, ActorSystem }
import scala.annotation.tailrec
case object Msg
object global {
val latch = new java.util.concurrent.CountDownLatch(1)
}
class ThreadedReceiver(n: Long) extends Actor {
override def act() {
var i: Long = 0
while (i < n)
receive {
case Msg => i += 1
}
}
}
class ThreadlessReceiver(n: Long) extends Actor {
var i: Long = 0
override def act() {
react {
case Msg =>
i += 1
if (i < n) act
}
}
}
class AkkaReceiver(n: Long) extends akka.actor.Actor {
var received: Long = 0
def receive = {
case Msg =>
received += 1
if (received == n) {
global.latch.countDown
context.stop(self)
}
}
}
class MailboxPerformance(threads: Int, msgs: Int) {
val total = threads * msgs;
def run[T](testee: T, fun: T => Unit) {
for (_ <- 0 until threads) {
(new Thread {
override def run() { for (_ <- 0 until msgs) fun(testee) }
}).start
}
}
def runThreaded() {
run((new ThreadedReceiver(total)).start, {(a: Actor) => a ! Msg})
}
def runThreadless() {
run((new ThreadlessReceiver(total)).start, {(a: Actor) => a ! Msg})
}
def runAkka() {
val system = ActorSystem()
run(system.actorOf(Props(new AkkaReceiver(total))), {(a: AkkaActorRef) => a ! Msg})
global.latch.await
system.shutdown
System.exit(0)
}
}
object Main {
def usage() {
Console println "usage: (threaded|threadless|akka) (num_threads) (msgs_per_thread)"
}
def main(args: Array[String]): Unit = args match {
case Array(impl, IntStr(threads), IntStr(msgs)) => {
val prog = new MailboxPerformance(threads, msgs)
impl match {
case "threaded" => prog.runThreaded
case "threadless" => prog.runThreadless
case "akka" => prog.runAkka
case _ => usage
}
}
case _ => usage
}
}
SCALAC=scalac
LIBPATH=/$(HOME)/akka-2.0.3/lib/akka
CLASSPATH=$(shell for i in $(LIBPATH)/*.jar ; do printf %s $$i: ; done)
#CLASSPATH=$(LIBPATH)/akka-kernel-2.0.3.jar:$(LIBPATH)/akka-actor-2.0.3.jar
FLAGS=-cp $(CLASSPATH)
#FILES=ActorCreation.scala Distributed.scala MailboxPerformance.scala MixedCase.scala Matching.scala
#CLASS_FILES=$(FILES:.scala=.class)
FILES=ActorCreation.scala Distributed.scala MailboxPerformance.scala Matching.scala MixedCase.scala
CLASS_FILES=$(foreach FILE,$(FILES),org/libcppa/$(shell echo $(FILE:.scala=) | sed 's/\(.\)\([A-Z]\)/\1_\2/g' | tr [:upper:] [:lower:])/$(FILE:.scala=.class))
UTILITY=org/libcppa/utility/Utility.class
all: $(CLASS_FILES)
org/libcppa/utility/Utility.class: Utility.scala
$(SCALAC) $(FLAGS) Utility.scala
#$(CLASS_FILES): $(FILES) org/libcppa/utility/Utility.class
org/libcppa/actor_creation/ActorCreation.class: ActorCreation.scala $(UTILITY)
$(SCALAC) $(FLAGS) $<
org/libcppa/distributed/Distributed.class: Distributed.scala $(UTILITY)
$(SCALAC) $(FLAGS) $<
org/libcppa/mailbox_performance/MailboxPerformance.class: MailboxPerformance.scala $(UTILITY)
$(SCALAC) $(FLAGS) $<
org/libcppa/matching/Matching.class: Matching.scala $(UTILITY)
$(SCALAC) $(FLAGS) $<
org/libcppa/mixed_case/MixedCase.class: MixedCase.scala $(UTILITY)
$(SCALAC) $(FLAGS) $<
clean:
rm -rf org/
.PHONY: all clean
package org.libcppa.matching
import org.libcppa.utility._
import Console.println
import scala.{PartialFunction => PF}
case class Msg1(val0: Int)
case class Msg2(val0: Double)
case class Msg3(val0: List[Int])
case class Msg4(val0: Int, val1: String)
case class Msg5(val0: Int, val1: Int, val2: Int)
case class Msg6(val0: Int, val1: Double, val2: String)
object Matching {
def apply(numLoops: Long) {
val zero: Long = 0
var msg1Matched: Long = 0;
var msg2Matched: Long = 0;
var msg3Matched: Long = 0;
var msg4Matched: Long = 0;
var msg5Matched: Long = 0;
var msg6Matched: Long = 0;
val partFun: PF[Any, Unit] = {
case Msg1(0) => msg1Matched += 1
case Msg2(0.0) => msg2Matched += 1
case Msg3(List(0)) => msg3Matched += 1
case Msg4(0, "0") => msg4Matched += 1
case Msg5(0, 0, 0) => msg5Matched += 1
case Msg6(0, 0.0, "0") => msg6Matched += 1
}
val m1: Any = Msg1(0)
val m2: Any = Msg2(0.0)
val m3: Any = Msg3(List(0))
val m4: Any = Msg4(0, "0")
val m5: Any = Msg5(0, 0, 0)
val m6: Any = Msg6(0, 0.0, "0")
for (_ <- zero until numLoops) {
partFun(m1)
partFun(m2)
partFun(m3)
partFun(m4)
partFun(m5)
partFun(m6)
}
assert(msg1Matched == numLoops)
assert(msg2Matched == numLoops)
assert(msg3Matched == numLoops)
assert(msg4Matched == numLoops)
assert(msg5Matched == numLoops)
assert(msg6Matched == numLoops)
println("msg1Matched = " + msg1Matched.toString)
}
}
object Main {
def main(args: Array[String]) = args match {
case Array(IntStr(numLoops)) => Matching(numLoops)
case _ => println("usage: Matching {NUM_LOOPS}")
}
}
package org.libcppa.mixed_case
import org.libcppa.utility.IntStr
import scala.actors.Actor
import scala.actors.Actor._
import akka.actor.{ Props, Actor => AkkaActor, ActorRef => AkkaActorRef, ActorSystem }
import scala.annotation.tailrec
case class Token(value: Int)
case class Init(ringSize: Int, initialTokenValue: Int, repetitions: Int)
case class Calc(value: Long)
case class Factors(values: List[Long])
case object Done
case object MasterExited
object global {
final val taskN: Long = 86028157l * 329545133
final val factor1: Long = 86028157
final val factor2: Long = 329545133
final val factors = List(factor2,factor1)
val latch = new java.util.concurrent.CountDownLatch(1)
def checkFactors(f: List[Long]) {
assert(f equals factors)
}
@tailrec final def fac(n: Long, m: Long, interim: List[Long]) : List[Long] = {
if (n == m) m :: interim
else if ((n % m) == 0) fac(n/m, m, m :: interim)
else fac(n, if (m == 2) 3 else m + 2, interim)
}
def factorize(arg: Long): List[Long] = {
if (arg <= 3) List(arg)
else fac(arg, 2, List())
}
}
class ThreadedWorker(supervisor: Actor) extends Actor {
override def act() {
var done = false
while (done == false) {
receive {
case Calc(value) => supervisor ! Factors(global.factorize(value))
case Done => done = true
}
}
}
}
class ThreadedChainLink(next: Actor) extends Actor {
override def act() {
var done = false
while (done == false)
receive {
case Token(value) => next ! Token(value); if (value == 0) done = true
}
}
}
class ThreadedChainMaster(supervisor: Actor) extends Actor {
val worker = (new ThreadedWorker(supervisor)).start
@tailrec final def newRing(next: Actor, rsize: Int): Actor = {
if (rsize == 0) next
else newRing((new ThreadedChainLink(next)).start, rsize-1)
}
override def act() = receive {
case Init(rsize, initialTokenValue, repetitions) =>
for (_ <- 0 until repetitions) {
worker ! Calc(global.taskN)
val next = newRing(this, rsize-1)
next ! Token(initialTokenValue)
var ringDone = false
while (ringDone == false) {
receive {
case Token(0) => ringDone = true
case Token(value) => next ! Token(value-1)
}
}
}
worker ! Done
supervisor ! MasterExited
}
}
class ThreadedSupervisor(numMessages: Int) extends Actor {
override def act() = for (_ <- 0 until numMessages) {
receive {
case Factors(f) => global.checkFactors(f)
case MasterExited =>
}
}
}
class ThreadlessWorker(supervisor: Actor) extends Actor {
override def act() = react {
case Calc(value) => supervisor ! Factors(global.factorize(value)); act
case Done => // recursion ends
}
}
class ThreadlessChainLink(next: Actor) extends Actor {
override def act() = react {
case Token(value) => next ! Token(value); if (value > 0) act
}
}
class ThreadlessChainMaster(supervisor: Actor) extends Actor {
val worker = (new ThreadlessWorker(supervisor)).start
@tailrec final def newRing(next: Actor, rsize: Int): Actor = {
if (rsize == 0) next
else newRing((new ThreadlessChainLink(next)).start, rsize-1)
}
var initialTokenValue = 0
var repetitions = 0
var iteration = 0
var rsize = 0
var next: Actor = null
def rloop(): Nothing = react {
case Token(0) =>
iteration += 1
if (iteration < repetitions) {
worker ! Calc(global.taskN)
next = newRing(this, rsize-1)
next ! Token(initialTokenValue)
rloop
}
else
{
worker ! Done
supervisor ! MasterExited
}
case Token(value) => next ! Token(value-1) ; rloop
}
override def act() = react {
case Init(rs, itv, rep) =>
rsize = rs ; initialTokenValue = itv ; repetitions = rep
worker ! Calc(global.taskN)
next = newRing(this, rsize-1)
next ! Token(initialTokenValue)
rloop
}
}
class ThreadlessSupervisor(numMessages: Int) extends Actor {
def rcv(remaining: Int): Nothing = react {
case Factors(f) => global.checkFactors(f); if (remaining > 1) rcv(remaining-1)
case MasterExited => if (remaining > 1) rcv(remaining-1)
}
override def act() = rcv(numMessages)
}
class AkkaWorker(supervisor: AkkaActorRef) extends AkkaActor {
def receive = {
case Calc(value) => supervisor ! Factors(global.factorize(value))
case Done => context.stop(self)
}
}
class AkkaChainLink(next: AkkaActorRef) extends AkkaActor {
def receive = {
case Token(value) => {
next ! Token(value)
if (value == 0) context.stop(self)
}
}
}
class AkkaChainMaster(supervisor: AkkaActorRef, worker: AkkaActorRef) extends AkkaActor {
@tailrec final def newRing(next: AkkaActorRef, rsize: Int): AkkaActorRef = {
if (rsize == 0) next
else newRing(context.actorOf(Props(new AkkaChainLink(next))), rsize-1)
}
def initialized(ringSize: Int, initialTokenValue: Int, repetitions: Int, next: AkkaActorRef, iteration: Int): Receive = {
case Token(0) =>
if (iteration + 1 < repetitions) {
worker ! Calc(global.taskN)
val next = newRing(self, ringSize - 1)
next ! Token(initialTokenValue)
context.become(initialized(ringSize, initialTokenValue, repetitions, next, iteration + 1))
}
else
{
worker ! Done
supervisor ! MasterExited
context.stop(self)
}
case Token(value) => next ! Token(value-1)
}
def receive = {
case Init(rs, itv, rep) =>
worker ! Calc(global.taskN)
val next = newRing(self, rs-1)
next ! Token(itv)
context.become(initialized(rs, itv, rep, next, 0))
}
}
class AkkaSupervisor(numMessages: Int) extends AkkaActor {
var i = 0
def inc() {
i = i + 1
if (i == numMessages) {
global.latch.countDown
context.stop(self)
}
}
def receive = {
case Factors(f) => global.checkFactors(f); inc
case MasterExited => inc
case Init(numRings, iterations, repetitions) =>
val initMsg = Init(numRings, iterations, repetitions)
for (_ <- 0 until numRings) {
val worker = context.actorOf(Props(new AkkaWorker(self)))
context.actorOf(Props(new AkkaChainMaster(self, worker))) ! initMsg
}
}
}
class MixedCase(numRings: Int, ringSize: Int, initToken: Int, reps: Int) {
final val numMessages = numRings + (numRings * reps)
final val initMsg = Init(ringSize, initToken, reps)
def runThreaded() {
val s = (new ThreadedSupervisor(numMessages)).start
for (_ <- 0 until numRings)
(new ThreadedChainMaster(s)).start ! initMsg
}
def runThreadless() {
val s = (new ThreadlessSupervisor(numMessages)).start
for (_ <- 0 until numRings)
(new ThreadlessChainMaster(s)).start ! initMsg
}
def runAkka() {
val system = ActorSystem();
val s = system.actorOf(Props(new AkkaSupervisor(numMessages)))
s ! initMsg
global.latch.await
system.shutdown
System.exit(0)
}
}
object Main {
def usage() = {
Console println "usage: ('threaded'|'threadless'|'akka') (num rings) (ring size) (initial token value) (repetitions)"
System.exit(1) // why doesn't exit return Nothing?
}
def main(args: Array[String]): Unit = args match {
case Array(impl, IntStr(numRings), IntStr(ringSize), IntStr(initToken), IntStr(reps)) => {
val mc = new MixedCase(numRings, ringSize, initToken, reps);
impl match {
case "threaded" => mc.runThreaded
case "threadless" => mc.runThreadless
case "akka" => mc.runAkka
case _ => usage
}
}
case _ => usage
}
}
package org.libcppa.utility
object IntStr {
val IntRegex = "([0-9]+)".r
def unapply(s: String): Option[Int] = s match {
case IntRegex(`s`) => Some(s.toInt)
case _ => None
}
}
object KeyValuePair {
val Rx = "([^=])+=([^=]*)".r
def unapply(s: String): Option[Pair[String, String]] = s match {
case Rx(key, value) => Some(Pair(key, value))
case _ => None
}
}
class Utility {
}
#!/bin/bash
#export JAVA_OPTS="-Xmx1024"
if [ ! -d $PWD/scala ] ; then
echo "$PWD/scala is not a valid directory!"
exit
fi
AKKA_LIBS=/$HOME/akka-2.0.3/lib/scala-library.jar
for JAR in /$HOME/akka-2.0.3/lib/akka/*.jar ; do
AKKA_LIBS=$JAR:$AKKA_LIBS
done
arg0=org.libcppa.$1.Main
shift
JARS="$AKKA_LIBS":$PWD/scala/
echo "java -cp $JARS $arg0 $@" | ./exec.sh
#define THERON_USE_BOOST_THREADS 1
#include <Theron/Framework.h>
#include <Theron/Actor.h>
#include <thread>
#include <cstdint>
#include <iostream>
#include <functional>
#define THERON_BENCHMARK
#include "utility.hpp"
using std::cout;
using std::endl;
using std::uint32_t;
struct spread { int value; };
struct result { uint32_t value; };
using namespace Theron;
struct testee : Actor {
Address m_parent;
bool m_first_result_received;
uint32_t m_first_result;
std::vector<ActorRef> m_children;
void spread_handler(const spread& arg, const Address) {
if (arg.value == 0) {
Send(result{1}, m_parent);
}
else {
spread msg = {arg.value-1};
Parameters params = {GetAddress()};
for (int i = 0; i < 2; ++i) {
m_children.push_back(GetFramework().CreateActor<testee>(params));
m_children.back().Push(msg, GetAddress());
}
}
}
void result_handler(const result& arg, const Address) {
if (!m_first_result_received) {
m_first_result_received = true;
m_first_result = arg.value;
}
else {
m_children.clear();
Send(result{m_first_result + arg.value}, m_parent);
}
}
typedef struct { Address arg0; } Parameters;
testee(const Parameters& p) : m_parent(p.arg0), m_first_result_received(false) {
RegisterHandler(this, &testee::spread_handler);
RegisterHandler(this, &testee::result_handler);
}
};
void usage() {
cout << "usage: theron_actor_creation _ POW" << endl
<< " creates 2^POW actors" << endl
<< endl;
}
int main(int argc, char** argv) {
if (argc != 3) {
usage();
return 1;
}
int num = rd<int>(argv[2]);
Receiver r;
Framework framework(num_cores());
ActorRef aref(framework.CreateActor<testee>(testee::Parameters{r.GetAddress()}));
aref.Push(spread{num}, r.GetAddress());
r.Wait();
}
#define THERON_USE_BOOST_THREADS 1
#include <Theron/Framework.h>
#include <Theron/Actor.h>
#include <thread>
#include <cstdint>
#include <iostream>
#include <functional>
#define THERON_BENCHMARK
#include "utility.hpp"
using std::cerr;
using std::cout;
using std::endl;
using std::int64_t;
using namespace Theron;
int64_t t_max = 0;
struct receiver : Actor {
int64_t m_num;
void handler(const int64_t&, const Address from) {
if (++m_num == t_max)
Send(t_max, from);
}
receiver() : m_num(0) {
RegisterHandler(this, &receiver::handler);
}
};
void send_sender(Framework& f, ActorRef ref, Address waiter, int64_t num) {
auto addr = ref.GetAddress();
int64_t msg;
for (int64_t i = 0; i < num; ++i) f.Send(msg, waiter, addr);
}
void push_sender(Framework& f, ActorRef ref, Address waiter, int64_t num) {
int64_t msg;
for (int64_t i = 0; i < num; ++i) ref.Push(msg, waiter);
}
void usage() {
cout << "usage ('push'|'send') (num_threads) (num_messages)" << endl;
exit(1);
}
int main(int argc, char** argv) {
if (argc != 4) {
usage();
}
enum { invalid_impl, push_impl, send_impl } impl = invalid_impl;
if (strcmp(argv[1], "push") == 0) impl = push_impl;
else if (strcmp(argv[1], "send") == 0) impl = send_impl;
else usage();
auto num_sender = rd<int64_t>(argv[2]);
auto num_msgs = rd<int64_t>(argv[3]);
Receiver r;
t_max = num_sender * num_msgs;
auto receiverAddr = r.GetAddress();
Framework framework(num_cores());
ActorRef aref(framework.CreateActor<receiver>());
std::list<std::thread> threads;
auto impl_fun = (impl == push_impl) ? send_sender : push_sender;
for (int64_t i = 0; i < num_sender; ++i) {
threads.push_back(std::thread(impl_fun, std::ref(framework), aref, receiverAddr, num_msgs));
}
r.Wait();
for (auto& t : threads) t.join();
return 0;
}
#define THERON_USE_BOOST_THREADS 1
#include <Theron/Framework.h>
#include <Theron/Actor.h>
#include <thread>
#include <cstdint>
#include <iostream>
#include <functional>
#define THERON_BENCHMARK
#include "utility.hpp"
using std::cerr;
using std::cout;
using std::endl;
using std::int64_t;
using namespace Theron;
constexpr uint64_t s_task_n = uint64_t(86028157)*329545133;
constexpr uint64_t s_factor1 = 86028157;
constexpr uint64_t s_factor2 = 329545133;
typedef std::vector<uint64_t> factors;
struct calc_msg { uint64_t value; };
struct result_msg { factors result; };
struct token_msg { int value; };
struct init_msg { int ring_size; int token_value; int iterations; };
struct master_done { };
struct worker_done { };
struct worker : Actor {
void handle_calc(const calc_msg& msg, Address from) {
factorize(msg.value);
}
void handle_master_done(const master_done&, Address from) {
Send(worker_done(), from);
}
worker() {
RegisterHandler(this, &worker::handle_calc);
RegisterHandler(this, &worker::handle_master_done);
}
};
struct chain_link : Actor {
Address next;
void handle_token(const token_msg& msg, Address) {
Send(msg, next);
}
typedef struct { Address next; } Parameters;
chain_link(const Parameters& p) : next(p.next) {
RegisterHandler(this, &chain_link::handle_token);
}
};
struct master : Actor {
Address mc;
int iteration;
int max_iterations;
Address next;
ActorRef w;
int ring_size;
int initial_token_value;
std::vector<ActorRef> m_children;
void new_ring() {
m_children.clear();
w.Push(calc_msg{s_task_n}, GetAddress());
next = GetAddress();
for (int i = 1; i < ring_size; ++i) {
m_children.push_back(GetFramework().CreateActor<chain_link>(chain_link::Parameters{next}));
next = m_children.back().GetAddress();
}
Send(token_msg{initial_token_value}, next);
}
void handle_init(const init_msg& msg, Address) {
w = GetFramework().CreateActor<worker>();
iteration = 0;
ring_size = msg.ring_size;
initial_token_value = msg.token_value;
max_iterations = msg.iterations;
new_ring();
}
void handle_token(const token_msg& msg, Address) {
if (msg.value == 0) {
if (++iteration < max_iterations) {
new_ring();
}
else {
w.Push(master_done(), GetAddress());
}
}
else {
Send(token_msg{msg.value - 1}, next);
}
}
void handle_worker_done(const worker_done&, Address) {
Send(master_done(), mc);
w = ActorRef::Null();
}
typedef struct { Address mc; } Parameters;
master(const Parameters& p) : mc(p.mc), iteration(0) {
RegisterHandler(this, &master::handle_init);
RegisterHandler(this, &master::handle_token);
RegisterHandler(this, &master::handle_worker_done);
}
};
void usage() {
cout << "usage: mailbox_performance "
"'send' (num rings) (ring size) "
"(initial token value) (repetitions)"
<< endl
<< endl;
exit(1);
}
int main(int argc, char** argv) {
if (argc != 6) usage();
if (strcmp("send", argv[1]) != 0) usage();
int num_rings = rd<int>(argv[2]);
int ring_size = rd<int>(argv[3]);
int inital_token_value = rd<int>(argv[4]);
int repetitions = rd<int>(argv[5]);
Receiver r;
Framework framework(num_cores());
std::vector<ActorRef> masters;
for (int i = 0; i < num_rings; ++i) {
masters.push_back(framework.CreateActor<master>(master::Parameters{r.GetAddress()}));
}
for (ActorRef& m : masters) {
m.Push(init_msg{ring_size, inital_token_value, repetitions}, r.GetAddress());
}
for (int i = 0; i < num_rings; ++i) r.Wait();
return 0;
}
#!/bin/bash
echo "./theron_$@" | ./exec.sh
#ifndef UTILITY_HPP
#define UTILITY_HPP
#include <vector>
#include <string>
#include <sstream>
#include <stdexcept>
#include <algorithm>
inline std::vector<std::string> split(const std::string& str, char delim) {
std::vector<std::string> result;
std::stringstream strs{str};
std::string tmp;
while (std::getline(strs, tmp, delim)) result.push_back(tmp);
return result;
}
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) {
result += delim;
result += *i;
}
return result;
}
template<typename T>
T rd(const char* cstr) {
char* endptr = nullptr;
T result = static_cast<T>(strtol(cstr, &endptr, 10));
if (endptr == nullptr || *endptr != '\0') {
std::string errstr;
errstr += "\"";
errstr += cstr;
errstr += "\" is not an integer";
throw std::invalid_argument(errstr);
}
return result;
}
int num_cores() {
char cbuf[100];
FILE* cmd = popen("/bin/cat /proc/cpuinfo | /bin/grep processor | /usr/bin/wc -l", "r");
if (fgets(cbuf, 100, cmd) == 0) {
throw std::runtime_error("cannot determine number of cores");
}
pclose(cmd);
// erase trailing newline
auto i = std::find(cbuf, cbuf + 100, '\n');
*i = '\0';
return rd<int>(cbuf);
}
std::vector<uint64_t> factorize(uint64_t n) {
std::vector<uint64_t> result;
if (n <= 3) {
result.push_back(n);
return std::move(result);
}
uint64_t d = 2;
while(d < n) {
if((n % d) == 0) {
result.push_back(d);
n /= d;
}
else {
d = (d == 2) ? 3 : (d + 2);
}
}
result.push_back(d);
return std::move(result);
}
// some utility for cppa benchmarks only
#ifndef THERON_BENCHMARK
#include "cppa/option.hpp"
// a string => T projection
template<typename T>
cppa::option<T> spro(const std::string& str) {
T value;
if (std::istringstream(str) >> value) {
return value;
}
return {};
}
#endif // THERON_BENCHMARK
#endif // UTILITY_HPP
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