Commit e646ce1e authored by Dominik Charousset's avatar Dominik Charousset

code cleanup and documentation

parent fe55e3cb
/******************************************************************************\
* This program is a distributed version of the math_actor example. *
* Client and server use a stateless request/response protocol and the client *
* is failure resilient by using a FIFO request queue. *
* The client auto-reconnects and also allows for server reconfiguration. *
* *
* Run server at port 4242: *
* - ./build/bin/distributed_math_actor -s -p 4242 *
* *
* Run client at the same host: *
* - ./build/bin/distributed_math_actor -c -p 4242 *
\******************************************************************************/
#include <vector> #include <vector>
#include <string> #include <string>
#include <sstream> #include <sstream>
#include <cassert> #include <cassert>
#include <iostream> #include <iostream>
#include <functional> #include <functional>
#include "cppa/opt.hpp"
#include "cppa/cppa.hpp" #include "cppa/cppa.hpp"
using namespace std; using namespace std;
using namespace cppa; using namespace cppa;
using namespace cppa::placeholders; using namespace cppa::placeholders;
static const char* s_usage = R"___( // our service provider
Usage: distributed_math_actor_example [OPTIONS]
General options:
-h | --help Print this text and quit
-v | --version Print the program version and quit
-s | --server Run in server mode
-c | --client Run in client mode
-p <arg> | --port=<arg> Publish actor at port <arg> (server)
Connect to remote actor at port <arg> (client)
Client options:
-t <arg> | --host=<arg> Connect to host <arg>, default: localhost
)___";
static const char* s_interactive_usage = R"___(
quit Quit the program
<x> + <y> Calculate <x>+<y> and print result
<x> - <y> Calculate <x>-<y> and print result
)___";
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() {
// execute this behavior until actor terminates // execute this behavior until actor terminates
...@@ -53,16 +43,97 @@ struct math_actor : event_based_actor { ...@@ -53,16 +43,97 @@ struct math_actor : event_based_actor {
} }
}; };
option<int> toint(const string& str) { namespace {
if (str.empty()) return {};
char* endptr = nullptr; // this actor manages the output of the program and has a per-actor cache
int result = static_cast<int>(strtol(str.c_str(), &endptr, 10)); actor_ptr s_printer = factory::event_based([](map<actor_ptr,string>* out) {
if (endptr != nullptr && *endptr == '\0') { auto flush_output = [out](const actor_ptr& s) {
return result; auto i = out->find(s);
if (i != out->end()) {
auto& ref = i->second;
if (!ref.empty()) {
cout << move(ref) << flush;
ref = string();
}
}
};
self->become (
on(atom("add"), arg_match) >> [=](string& str) {
if (!str.empty()) {
auto s = self->last_sender();
auto i = out->find(s);
if (i == out->end()) {
i = out->insert(make_pair(s, move(str))).first;
// monitor actor to flush its output on exit
self->monitor(s);
}
auto& ref = i->second;
ref += move(str);
if (ref.back() == '\n') {
cout << move(ref) << flush;
ref = string();
}
}
},
on(atom("flush")) >> [=] {
flush_output(self->last_sender());
},
on(atom("DOWN"), any_vals) >> [=] {
auto s = self->last_sender();
flush_output(s);
out->erase(s);
},
on(atom("quit")) >> [] {
self->quit();
},
others() >> [] {
cout << "*** unexpected: "
<< to_string(self->last_dequeued())
<< endl;
}
);
}).spawn();
} // namespace <anonymous>
// an output stream sending messages to s_printer
struct actor_ostream {
typedef actor_ostream& (*fun_type)(actor_ostream&);
virtual ~actor_ostream() { }
virtual actor_ostream& write(string arg) {
send(s_printer, atom("add"), move(arg));
return *this;
} }
return {};
virtual actor_ostream& flush() {
send(s_printer, atom("flush"));
return *this;
}
};
inline actor_ostream& operator<<(actor_ostream& o, string arg) {
return o.write(move(arg));
}
template<typename T>
inline typename enable_if<is_convertible<T,string>::value == false, actor_ostream&>::type
operator<<(actor_ostream& o, T&& arg) {
return o.write(to_string(std::forward<T>(arg)));
}
inline actor_ostream& operator<<(actor_ostream& o, actor_ostream::fun_type f) {
return f(o);
} }
namespace { actor_ostream aout; }
inline actor_ostream& endl(actor_ostream& o) { return o.write("\n"); }
inline actor_ostream& flush(actor_ostream& o) { return o.flush(); }
inline string& ltrim(string &s) { inline string& ltrim(string &s) {
s.erase(s.begin(), find_if(s.begin(), s.end(), [](char c) { return !isspace(c); })); s.erase(s.begin(), find_if(s.begin(), s.end(), [](char c) { return !isspace(c); }));
return s; return s;
...@@ -77,6 +148,36 @@ inline string& trim(std::string& s) { ...@@ -77,6 +148,36 @@ inline string& trim(std::string& s) {
return ltrim(rtrim(s)); return ltrim(rtrim(s));
} }
option<int> toint(const string& str) {
if (str.empty()) return {};
char* endptr = nullptr;
int result = static_cast<int>(strtol(str.c_str(), &endptr, 10));
if (endptr != nullptr && *endptr == '\0') {
return result;
}
return {};
}
template<typename T>
struct project_helper;
template<>
struct project_helper<string> {
template<typename T>
inline option<T> convert(const string& from, typename enable_if<is_integral<T>::value>::type* = 0) {
char* endptr = nullptr;
auto result = static_cast<T>(strtol(from.c_str(), &endptr, 10));
if (endptr != nullptr && *endptr == '\0') return result;
return {};
}
};
template<typename From, typename To>
option<To> projection(const From& from) {
project_helper<From> f;
return f.template convert<To>(from);
}
void client_repl(actor_ptr server, string host, uint16_t port) { void client_repl(actor_ptr server, string host, uint16_t port) {
typedef cow_tuple<atom_value, int, int> request; typedef cow_tuple<atom_value, int, int> request;
// keeps track of requests and tries to reconnect on server failures // keeps track of requests and tries to reconnect on server failures
...@@ -96,48 +197,70 @@ void client_repl(actor_ptr server, string host, uint16_t port) { ...@@ -96,48 +197,70 @@ void client_repl(actor_ptr server, string host, uint16_t port) {
}, },
on(atom("result"), arg_match) >> [=](int result) { on(atom("result"), arg_match) >> [=](int result) {
if (q->empty()) { if (q->empty()) {
send(s_printer, "received a result, " aout << "received a result, but didn't send a request\n";
"but didn't send a request");
return; return;
} }
ostringstream oss; ostringstream oss;
auto& r = q->front(); auto& r = q->front();
oss << get<1>(r) << " " aout << get<1>(r) << " "
<< to_string(get<0>(r)) << " " << to_string(get<0>(r)) << " "
<< get<2>(r) << get<2>(r)
<< " = " << result; << " = " << result
send(s_printer, oss.str()); << endl;
q->erase(q->begin()); q->erase(q->begin());
send_next_request(); send_next_request();
}, },
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) { on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
ostringstream oss; if (*serv == self->last_sender()) {
oss << "*** server exited with reason = " << reason; serv->reset(); // sets *serv = nullptr
send(s_printer, oss.str()); ostringstream oss;
send(self, atom("reconnect")); aout << "*** server exited with reason = " << reason
<< ", try to reconnect"
<< endl;
send(self, atom("reconnect"));
}
}, },
on(atom("reconnect")) >> [=] { on(atom("reconnect")) >> [=] {
if (*serv != nullptr) return;
try { try {
*serv = remote_actor(host, port); *serv = remote_actor(host, port);
self->monitor(*serv); self->monitor(*serv);
send(s_printer, "reconnection succeeded"); aout << "reconnection succeeded" << endl;
send_next_request(); send_next_request();
} }
catch (exception&) { catch (exception&) {
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"));
} }
}, },
on(atom("rebind"), arg_match) >> [=](string& host, uint16_t port) {
actor_ptr new_serv;
try {
new_serv = remote_actor(move(host), port);
self->monitor(new_serv);
aout << "rebind succeeded" << endl;
*serv = new_serv;
send_next_request();
}
catch (exception& e) {
aout << "*** rebind failed: "
<< to_verbose_string(e) << endl;
}
},
on(atom("quit")) >> [=] { on(atom("quit")) >> [=] {
self->quit(); self->quit();
}, },
others() >> [] { others() >> [] {
forward_to(s_printer); aout << "unexpected message: " << self->last_dequeued() << endl;
} }
); );
}).spawn(server); }).spawn(server);
send(s_printer, s_interactive_usage); aout << "quit Quit the program\n"
"<x> + <y> Calculate <x>+<y> and print result\n"
"<x> - <y> Calculate <x>-<y> and print result\n"
"connect <host> <port> Reconfigure server"
<< endl;
string line; string line;
const char connect[] = "connect ";
while (getline(cin, line)) { while (getline(cin, line)) {
trim(line); trim(line);
if (line == "quit") { if (line == "quit") {
...@@ -145,151 +268,91 @@ void client_repl(actor_ptr server, string host, uint16_t port) { ...@@ -145,151 +268,91 @@ void client_repl(actor_ptr server, string host, uint16_t port) {
send(s_printer, atom("quit")); send(s_printer, atom("quit"));
return; return;
} }
bool success = false; // the STL way of line.starts_with("connect")
auto pos = find_if(begin(line), end(line), [](char c) { return c == '+' || c == '-'; }); else if (equal(begin(connect), end(connect) - 1, begin(line))) {
if (pos != end(line)) { match_split(line, ' ') (
string lsub(begin(line), pos); on("connect", val<string>, projection<string,uint16_t>) >> [&](string& host, uint16_t port) {
string rsub(pos + 1, end(line)); send(client, atom("rebind"), move(host), port);
auto lhs = toint(trim(lsub)); },
auto rhs = toint(trim(rsub)); others() >> [] {
if (lhs && rhs) { aout << "illegal host/port definition" << endl;
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 {
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) aout << "\"" + lsub + "\" is not an integer" << endl;
if (!rhs) aout << "\"" + rsub + "\" is not an integer" << endl;
} }
} }
} else if (!success) aout << "*** invalid format; use: X +/- Y" << endl;
else if (!success) {
send(s_printer, "*** invalid format, please use: X +/- Y");
} }
} }
} }
auto on_opt(char short_opt, const char* long_opt) -> decltype(on("", val<string>) || on(function<option<string> (const string&)>())) {
const char short_flag_arr[] = {'-', short_opt, '\0' };
const char* lhs_str = short_flag_arr;
string prefix = "--";
prefix += long_opt;
prefix += "=";
function<option<string> (const string&)> kvp = [prefix](const string& input) -> option<string> {
if ( input.compare(0, prefix.size(), prefix) == 0
// accept '-key=' as well
|| input.compare(1, prefix.size(), prefix) == 0) {
return input.substr(prefix.size());
}
return {};
};
return on(lhs_str, 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 short_flag_arr[] = {'-', short_opt, '\0' };
vector<string> opt_strs = { short_flag_arr };
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;
string host; string host;
vector<string> args(argv + 1, argv + argc); uint16_t port = 0;
auto set_mode = [&](string arg) -> bool { options_description desc;
if (!mode.empty()) { auto set_mode = [&](const string& arg) -> function<bool()> {
cerr << "mode already set to " << mode << endl; return [arg,&mode]() -> bool {
return false; if (!mode.empty()) {
} cerr << "mode already set to " << mode << endl;
mode = move(arg); return false;
return true;
};
bool args_valid = !args.empty() && match_stream<string>(begin(args), end(args)) (
on_opt('p', "port") >> [&](const string& arg) -> bool {
auto p = toint(arg);
if (p && *p > 1024 && *p < 0xFFFFF) {
port = static_cast<uint16_t>(*p);
return true;
}
cerr << port << " is not a valid port" << endl;
return false; // no match
},
on_opt('t', "host") >> [&](const string& arg) -> bool {
if (host.empty()) {
host = arg;
return true;
} }
cerr << "host previously set to \"" << arg << "\"" << endl; mode = move(arg);
return false; return true;
}, };
on_void_opt('s', "server") >> [&]() -> bool { };
return set_mode("server"); string copts = "client options";
}, string sopts = "server options";
on_void_opt('c', "client") >> [&]() -> bool { bool args_valid = match_stream<string> (argv + 1, argv + argc) (
return set_mode("client"); on_opt1('p', "port", &desc, "set port") >> rd_arg(port),
}, on_opt1('H', "host", &desc, "set host (default: localhost)", copts) >> rd_arg(host),
on_void_opt('h', "help") >> [] { on_opt0('s', "server", &desc, "run in server mode", sopts) >> set_mode("server"),
cout << s_usage << endl; on_opt0('c', "client", &desc, "run in client mode", copts) >> set_mode("client"),
exit(0); on_opt0('h', "help", &desc, "print help") >> print_desc_and_exit(&desc)
},
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()) {
if (port == 0) { if (port == 0) cerr << "*** no port specified" << endl;
cerr << "no port given" << endl; if (mode.empty()) cerr << "*** no mode specified" << endl;
} cerr << endl;
if (mode.empty()) { print_desc(&desc, cerr)();
cerr << "no mode given" << endl;
}
if (mode == "server" && !host.empty()) {
cerr << "--host=<arg> is a client-only option" << endl;
}
cout << endl << s_usage << endl;
return -1; return -1;
} }
if (mode == "server") { if (mode == "server") {
try { try {
// try to publish math actor at given port
publish(spawn<math_actor>(), port); publish(spawn<math_actor>(), port);
} }
catch (bind_failure& bf) { catch (exception& e) {
cerr << "unable to publish actor at port " cerr << "*** unable to publish math actor at port " << port << "\n"
<< port << ":" << bf.what() << endl; << to_verbose_string(e) // prints exception type and e.what()
<< endl;
} }
} }
else { else {
if (host.empty()) { if (host.empty()) host = "localhost";
host = "localhost";
}
try { try {
auto server = remote_actor(host, port); auto server = remote_actor(host, port);
client_repl(server, host, port); client_repl(server, host, port);
} }
catch (exception& e) { catch (exception& e) {
cerr << "unable to connect to remote actor at host \"" cerr << "unable to connect to remote actor at host \""
<< host << "\" on port " << port << endl; << host << "\" on port " << port << "\n"
<< to_verbose_string(e) << endl;
} }
} }
await_all_others_done(); await_all_others_done();
......
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