Commit 0c3537cb authored by neverlord's avatar neverlord

distributed benchmark

parent 51cc731e
......@@ -86,7 +86,13 @@ void usage()
<< " 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:123 --num_pings=100"<< 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
......@@ -224,6 +230,11 @@ struct server_actor : fsm_actor<server_actor>
});
if (i != m_pongs.end()) m_pongs.erase(i);
},
on(atom("shutdown")) >> [=]()
{
m_pongs.clear();
become_void();
},
others() >> [=]()
{
cout << "unexpected: " << to_string(last_dequeued()) << endl;
......@@ -324,6 +335,11 @@ void client_mode(Iterator first, Iterator last)
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);
}
std::vector<actor_ptr> remote_actors;
for (auto& r : remotes)
{
......@@ -398,6 +414,48 @@ void client_mode(Iterator first, Iterator last)
await_all_others_done();
}
template<typename Iterator>
void shutdown_mode(Iterator first, Iterator last)
{
std::vector<std::pair<string, uint16_t> > remotes;
match_each(first, last, std::bind(split, std::placeholders::_1, ':'))
(
on(val<string>, _2i) >> [&](string& host, int port)
{
if (port <= 1024 || port >= 65536)
{
throw std::invalid_argument("illegal port: " + std::to_string(port));
}
remotes.emplace_back(std::move(host), static_cast<uint16_t>(port));
}
);
for (auto& r : remotes)
{
try
{
actor_ptr x = remote_actor(r.first.c_str(), r.second);
monitor(x);
send(x, atom("shutdown"));
receive
(
on(atom("DOWN"), x, val<std::uint32_t>) >> []()
{
// ok, done
},
after(std::chrono::seconds(10)) >> [&]()
{
cout << r.first << ":" << r.second << " didn't shut down "
<< "within 10s"
<< endl;
}
);
}
catch (...)
{
}
}
}
int main(int argc, char** argv)
{
if (argc < 2) usage();
......@@ -405,13 +463,23 @@ int main(int argc, char** argv)
auto last = argv + argc;
match(*first)
(
on<string>().when(_x1.in({"-h", "--help"})) >> []() { usage(); },
on("--mode=server") >> [=]() { server_mode(first + 1, last); },
on<string>().when(_x1.in({"-h", "--help"})) >> []()
{
usage();
},
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);
},
others() >> [=]()
{
usage("unknown argument: ", *first);
......
-module(distributed).
-export([start/1, ping_loop/2]).
ping_loop(Parent, Pong) ->
receive
{pong, 0} -> Parent ! done;
{pong, X} ->
Pong ! {self(), ping, X - 1},
ping_loop(Parent, Pong);
{kickoff, Value} ->
Pong ! {ping, self(), Value},
ping_loop(Parent, Pong)
end.
server_loop(Pongs) ->
receive
{ping, Pid, Value} -> Pid ! {pong, Value};
{add_pong, Pid, Node} ->
case lists:any(fun({N, _}) -> N == Node end, Pongs) of
true ->
Pid ! {ok},
server_loop(Pongs);
false ->
case rpc:call(Node, erlang, whereis, [pong]) of
{badrpc, Reason} -> Pid ! {error, Reason};
Pong ->
Pid ! {ok},
server_loop(Pongs ++ [{Node, Pong}])
end
end;
{purge} -> server_loop([]);
{kickoff, Pid, NumPings} ->
lists:foreach(fun({_, P}) -> spawn(distributed, ping_loop, [Pid, P]) ! {kickoff, NumPings} end, 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, H},
receive
{ok} -> add_pong_fun(Pong, Node, T);
{error, Reason} -> error(Reason)
end.
% receive a {done} message for each node
client_mode([], [], [], _) -> true;
client_mode([], [], [_|T], NumPings) ->
receive {done} ->
client_mode([], [], T, NumPings)
end;
% 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);
P ->
add_pong_fun(P, H, Nodes),
client_mode(Pongs ++ [P], T, Nodes, NumPings)
end.
run([], undefined, ['mode=server']) -> server_mode();
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),
SetNumPings = lists:prefix("num_pings=", Arg),
if
SetNumPings == true andalso NumPings /= undefined ->
error("NumPings already set");
SetNumPings == true ->
run(Hosts, list_to_integer(lists:sublist(Arg, 11, length(Arg))), T);
true ->
run(Hosts ++ [H], NumPings, T)
end.
start(X) ->
run([], undefined, X).
This diff is collapsed.
......@@ -31,29 +31,36 @@
#include "cppa/cppa.hpp"
#include "cppa/local_actor.hpp"
namespace cppa {
namespace {
class observer : public cppa::attachable
class down_observer : public attachable
{
cppa::actor_ptr m_client;
actor_ptr m_observer;
actor_ptr m_observed;
public:
observer(cppa::actor_ptr&& client) : m_client(std::move(client)) { }
down_observer(actor_ptr observer, actor_ptr observed)
: m_observer(std::move(observer))
, m_observed(std::move(observed))
{
}
void actor_exited(std::uint32_t reason)
{
using namespace cppa;
send(m_client, atom("DOWN"), actor_ptr(self), reason);
send(m_observer, atom("DOWN"), m_observed, reason);
}
bool matches(const cppa::attachable::token& match_token)
bool matches(const attachable::token& match_token)
{
if (match_token.subtype == typeid(observer))
if (match_token.subtype == typeid(down_observer))
{
auto ptr = reinterpret_cast<const cppa::local_actor*>(match_token.ptr);
return m_client == ptr;
auto ptr = reinterpret_cast<const local_actor*>(match_token.ptr);
return m_observer == ptr;
}
return false;
}
......@@ -62,8 +69,6 @@ class observer : public cppa::attachable
} // namespace <anonymous>
namespace cppa {
const self_type& operator<<(const self_type& s, const any_tuple& what)
{
local_actor* sptr = s;
......@@ -120,7 +125,7 @@ void unlink(actor_ptr& lhs, actor_ptr& rhs)
void monitor(actor_ptr& whom)
{
if (whom) whom->attach(new observer(actor_ptr(self)));
if (whom) whom->attach(new down_observer(self, whom));
}
void monitor(actor_ptr&& whom)
......@@ -131,7 +136,7 @@ void monitor(actor_ptr&& whom)
void demonitor(actor_ptr& whom)
{
attachable::token mtoken(typeid(observer), self);
attachable::token mtoken(typeid(down_observer), self);
if (whom) whom->detach(mtoken);
}
......
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