Commit 04c9dea2 authored by Dominik Charousset's avatar Dominik Charousset

improved distributed math example

parent 172190f0
...@@ -8,19 +8,30 @@ ...@@ -8,19 +8,30 @@
using namespace std; using namespace std;
using namespace cppa; using namespace cppa;
using namespace cppa::placeholders;
static const char* usage = static const char* s_usage =
"Usage: distributed_math_actor_example [OPTIONS] \n" "Usage: distributed_math_actor_example [OPTIONS] \n"
" \n" " \n"
" General options: \n" " General options: \n"
" -s | --server run in server mode \n" " -h | --help Print this text and quit \n"
" -c | --client run in client mode \n" " -v | --version Print the program version and quit \n"
" -p PORT | --port=PORT publish at PORT (server mode) \n" " -s | --server Run in server mode \n"
" connect to PORT (client mode) \n" " -c | --client Run in client mode \n"
" \n" " -p <arg> | --port=<arg> Publish actor at port <arg> (server) \n"
" Connect to remote actor at port <arg> (client) \n"
" Client options: \n" " Client options: \n"
" \n" " -t <arg> | --host=<arg> Connect to host <arg>, default: localhost \n";
" -h HOST | --host=HOST connect to HOST, default: localhost (client mode) \n";
static const char* s_interactive_usage =
"quit Quit the program \n"
"<x> + <y> Calculate <x>+<y> and print result \n"
"<x> - <y> Calculate <x>-<y> and print result \n";
static const char* s_version =
"libcppa distributed math actor example version 1.0";
actor_ptr s_printer;
struct math_actor : event_based_actor { struct math_actor : event_based_actor {
void init() { void init() {
...@@ -39,18 +50,8 @@ struct math_actor : event_based_actor { ...@@ -39,18 +50,8 @@ struct math_actor : event_based_actor {
} }
}; };
option<atom_value> get_op(const string& from) {
if (from.size() == 1) {
switch (from[0]) {
case '+': return atom("plus");
case '-': return atom("minus");
default: break;
}
}
return {};
}
option<int> toint(const string& str) { option<int> toint(const string& str) {
if (str.empty()) return {};
char* endptr = nullptr; char* endptr = nullptr;
int result = static_cast<int>(strtol(str.c_str(), &endptr, 10)); int result = static_cast<int>(strtol(str.c_str(), &endptr, 10));
if (endptr != nullptr && *endptr == '\0') { if (endptr != nullptr && *endptr == '\0') {
...@@ -59,69 +60,121 @@ option<int> toint(const string& str) { ...@@ -59,69 +60,121 @@ option<int> toint(const string& str) {
return {}; return {};
} }
inline std::vector<std::string> split(const std::string& str, char delim) { inline string& ltrim(string &s) {
vector<string> result; s.erase(s.begin(), find_if(s.begin(), s.end(), [](char c) { return !isspace(c); }));
stringstream strs{str}; return s;
string tmp; }
while (std::getline(strs, tmp, delim)) result.push_back(tmp);
return result; inline string& rtrim(std::string& s) {
s.erase(find_if(s.rbegin(), s.rend(), [](char c) { return !isspace(c); }).base(), s.end());
return s;
}
inline string& trim(std::string& s) {
return ltrim(rtrim(s));
} }
void client_repl(actor_ptr server, string host, uint16_t port) { void client_repl(actor_ptr server, string host, uint16_t port) {
self->monitor(server); typedef cow_tuple<atom_value, int, int> request;
string line; // keeps track of requests and tries to reconnect on server failures
while (getline(cin, line)) { auto client = factory::event_based([=](actor_ptr* serv, vector<request>* q) {
match (split(line, ' ')) ( self->monitor(*serv);
on(toint, get_op, toint) >> [&](int x, atom_value op, int y) { auto send_next_request = [=] {
// send request if (!q->empty()) {
send(server, op, x, y); *serv << q->front();
// wait for result }
bool done = false; };
receive_while (gref(done) == false) ( self->become (
on(atom("result"), arg_match) >> [&](int result) { on<atom_value, int, int>().when(_x1.in({atom("plus"), atom("minus")})) >> [=] {
cout << x << " " if (q->empty()) {
<< to_string(op) << " " *serv << self->last_dequeued();
<< y << " = " }
<< result q->push_back(*tuple_cast<atom_value, int, int>(self->last_dequeued()));
<< endl;
done = true;
}, },
on(atom("DOWN"), arg_match) >> [&](uint32_t reason) { on(atom("result"), arg_match) >> [=](int result) {
cerr << "*** server exited with reason = " << reason if (q->empty()) {
<< endl; send(s_printer, "received a result, "
"but didn't send a request");
return;
}
ostringstream oss;
auto& r = q->front();
oss << get<1>(r) << " "
<< to_string(get<0>(r)) << " "
<< get<2>(r)
<< " = " << result;
send(s_printer, oss.str());
q->erase(q->begin());
send_next_request();
},
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
ostringstream oss;
oss << "*** server exited with reason = " << reason;
send(s_printer, oss.str());
send(self, atom("reconnect")); send(self, atom("reconnect"));
}, },
on(atom("reconnect")) >> [&] { on(atom("reconnect")) >> [=] {
cout << "try reconnecting ... " << flush;
try { try {
server = remote_actor(host, port); *serv = remote_actor(host, port);
self->monitor(server); self->monitor(*serv);
send(server, op, x, y); send(s_printer, "reconnection succeeded");
cout << "success" << endl; send_next_request();
} }
catch (exception&) { catch (exception&) {
cout << "failed, try again in 3s" << endl; send(s_printer, "reconnection failed, try again in 3s");
delayed_send(self, chrono::seconds(3), atom("reconnect")); delayed_send(self, chrono::seconds(3), atom("reconnect"));
} }
}, },
others() >> [&] { on(atom("quit")) >> [=] {
cerr << "unexpected: " self->quit();
<< to_string(self->last_dequeued())
<< endl;
}
);
}, },
others() >> [&] { others() >> [] {
cerr << "*** invalid format, please use: X +/- Y" << endl; forward_to(s_printer);
} }
); );
}).spawn(server);
send(s_printer, s_interactive_usage);
string line;
while (getline(cin, line)) {
trim(line);
if (line == "quit") {
send(client, atom("quit"));
send(s_printer, atom("quit"));
return;
}
bool success = false;
auto pos = find_if(begin(line), end(line), [](char c) { return c == '+' || c == '-'; });
if (pos != end(line)) {
string lsub(begin(line), pos);
string rsub(pos + 1, end(line));
auto lhs = toint(trim(lsub));
auto rhs = toint(trim(rsub));
if (lhs && rhs) {
auto op = (*pos == '+') ? atom("plus") : atom("minus");
send(client, op, *lhs, *rhs);
}
else {
if (!lhs) {
send(s_printer, "\"" + lsub + "\" is not an integer");
}
if (!rhs) {
send(s_printer, "\"" + rsub + "\" is not an integer");
}
}
}
else if (!success) {
send(s_printer, "*** invalid format, please use: X +/- Y");
}
} }
} }
function<option<string> (const string&)> kvp(const string& key) { auto on_opt(char short_opt, const char* long_opt) -> decltype(on("", val<string>) || on(function<option<string> (const string&)>())) {
auto prefix = "--" + key; const char short_flag_arr[] = {'-', short_opt, '\0' };
const char* lhs_str = short_flag_arr;
string prefix = "--";
prefix += long_opt;
prefix += "="; prefix += "=";
return [prefix](const string& input) -> option<string> { function<option<string> (const string&)> kvp = [prefix](const string& input) -> option<string> {
if ( input.compare(0, prefix.size(), prefix) == 0 if ( input.compare(0, prefix.size(), prefix) == 0
// accept '-key=' as well // accept '-key=' as well
|| input.compare(1, prefix.size(), prefix) == 0) { || input.compare(1, prefix.size(), prefix) == 0) {
...@@ -129,15 +182,33 @@ function<option<string> (const string&)> kvp(const string& key) { ...@@ -129,15 +182,33 @@ function<option<string> (const string&)> kvp(const string& key) {
} }
return {}; return {};
}; };
return on(lhs_str, val<string>) || on(kvp);
} }
auto on_opt(char short_opt, const char* long_opt) -> decltype(on("", val<string>) || on(kvp(""))) { auto on_void_opt(char short_opt, const char* long_opt) -> decltype(on<string>().when(_x1.in(vector<string>()))) {
const char lhs[] = {'-', short_opt, '\0' }; const char short_flag_arr[] = {'-', short_opt, '\0' };
const char* lhs_str = lhs; vector<string> opt_strs = { short_flag_arr };
return on(lhs_str, val<string>) || on(kvp(long_opt)); opt_strs.push_back(string("-") + long_opt);
string str = "-";
str += opt_strs.back();
opt_strs.push_back(std::move(str));
return on<string>().when(_x1.in(opt_strs));
} }
int main(int argc, char** argv) { int main(int argc, char** argv) {
s_printer = factory::event_based([] {
self->become (
on_arg_match >> [](const string& str) {
cout << str << endl;
},
on(atom("quit")) >> [] {
self->quit();
},
others() >> [] {
cout << to_string(self->last_dequeued()) << endl;
}
);
}).spawn();
string mode; string mode;
uint16_t port = 0; uint16_t port = 0;
string host; string host;
...@@ -160,7 +231,7 @@ int main(int argc, char** argv) { ...@@ -160,7 +231,7 @@ int main(int argc, char** argv) {
cerr << port << " is not a valid port" << endl; cerr << port << " is not a valid port" << endl;
return false; // no match return false; // no match
}, },
on_opt('h', "host") >> [&](const string& arg) -> bool { on_opt('t', "host") >> [&](const string& arg) -> bool {
if (host.empty()) { if (host.empty()) {
host = arg; host = arg;
return true; return true;
...@@ -168,11 +239,19 @@ int main(int argc, char** argv) { ...@@ -168,11 +239,19 @@ int main(int argc, char** argv) {
cerr << "host previously set to \"" << arg << "\"" << endl; cerr << "host previously set to \"" << arg << "\"" << endl;
return false; return false;
}, },
(on("-s") || on("--server")) >> [&]() -> bool { on_void_opt('s', "server") >> [&]() -> bool {
return set_mode("server"); return set_mode("server");
}, },
(on("-c") || on("--client")) >> [&]() -> bool { on_void_opt('c', "client") >> [&]() -> bool {
return set_mode("client"); return set_mode("client");
},
on_void_opt('h', "help") >> [] {
cout << s_usage << endl;
exit(0);
},
on_void_opt('v', "version") >> [] {
cout << s_version << endl;
exit(0);
} }
); );
if (!args_valid || port == 0 || mode.empty() || (mode == "server" && !host.empty())) { if (!args_valid || port == 0 || mode.empty() || (mode == "server" && !host.empty())) {
...@@ -183,9 +262,9 @@ int main(int argc, char** argv) { ...@@ -183,9 +262,9 @@ int main(int argc, char** argv) {
cerr << "no mode given" << endl; cerr << "no mode given" << endl;
} }
if (mode == "server" && !host.empty()) { if (mode == "server" && !host.empty()) {
cerr << "host is a client-only option" << endl; cerr << "--host=<arg> is a client-only option" << endl;
} }
cout << endl << usage << endl; cout << endl << s_usage << endl;
return -1; return -1;
} }
if (mode == "server") { if (mode == "server") {
...@@ -214,4 +293,3 @@ int main(int argc, char** argv) { ...@@ -214,4 +293,3 @@ int main(int argc, char** argv) {
shutdown(); shutdown();
return 0; return 0;
} }
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