Commit 35fedb87 authored by Dominik Charousset's avatar Dominik Charousset

Streamline distributed_calculator example

parent 3cb41c66
...@@ -11,6 +11,7 @@ ...@@ -11,6 +11,7 @@
* - ./build/bin/distributed_math_actor -c -p 4242 * * - ./build/bin/distributed_math_actor -c -p 4242 *
\ ******************************************************************************/ \ ******************************************************************************/
#include <array>
#include <vector> #include <vector>
#include <string> #include <string>
#include <sstream> #include <sstream>
...@@ -21,7 +22,11 @@ ...@@ -21,7 +22,11 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/all.hpp" #include "caf/io/all.hpp"
using namespace std; using std::cout;
using std::cerr;
using std::endl;
using std::string;
using namespace caf; using namespace caf;
using plus_atom = atom_constant<atom("plus")>; using plus_atom = atom_constant<atom("plus")>;
...@@ -32,18 +37,117 @@ using connect_atom = atom_constant<atom("connect")>; ...@@ -32,18 +37,117 @@ using connect_atom = atom_constant<atom("connect")>;
using reconnect_atom = atom_constant<atom("reconnect")>; using reconnect_atom = atom_constant<atom("reconnect")>;
// our "service" // our "service"
void calculator(event_based_actor* self) { behavior calculator() {
self->become ( return {
[](plus_atom, int a, int b) -> message { [](plus_atom, int a, int b) -> message {
return make_message(result_atom::value, a + b); return make_message(result_atom::value, a + b);
}, },
[](minus_atom, int a, int b) -> message { [](minus_atom, int a, int b) -> message {
return make_message(result_atom::value, a - b); return make_message(result_atom::value, a - b);
} }
); };
} }
inline string trim(std::string s) { class client_impl : public event_based_actor {
public:
client_impl(string hostaddr, uint16_t port)
: m_host(std::move(hostaddr)),
m_port(port) {
// nop
}
behavior make_behavior() override {
become(awaiting_task());
become(keep_behavior, reconnecting());
return {};
}
private:
void sync_send_task(atom_value op, int lhs, int rhs) {
on_sync_failure([=] {
aout(this) << "*** sync_failure!" << endl;
});
sync_send(m_server, op, lhs, rhs).then(
[=](result_atom, int result) {
aout(this) << lhs << (op == plus_atom::value ? " + " : " - ")
<< rhs << " = " << result << endl;
},
[=](const sync_exited_msg& msg) {
aout(this) << "*** server down [" << msg.reason << "], "
<< "try to reconnect ..." << endl;
// try sync_sending this again after successful reconnect
become(keep_behavior,
reconnecting([=] { sync_send_task(op, lhs, rhs); }));
}
);
}
behavior awaiting_task() {
return {
[=](atom_value op, int lhs, int rhs) {
if (op != plus_atom::value && op != minus_atom::value) {
return;
}
sync_send_task(op, lhs, rhs);
},
[=](rebind_atom, string& nhost, uint16_t nport) {
aout(this) << "*** rebind to " << nhost << ":" << nport << endl;
using std::swap;
swap(m_host, nhost);
swap(m_port, nport);
become(keep_behavior, reconnecting());
}
};
}
behavior reconnecting(std::function<void()> continuation = nullptr) {
using std::chrono::seconds;
auto mm = io::get_middleman_actor();
send(mm, get_atom::value, m_host, m_port);
return {
[=](ok_atom, const actor_addr& new_server) {
aout(this) << "*** connection succeeded, awaiting tasks" << endl;
m_server = actor_cast<actor>(new_server);
// return to previous behavior
if (continuation) {
continuation();
}
unbecome();
},
[=](error_atom, const string& errstr) {
aout(this) << "*** could not connect to " << m_host
<< " at port " << m_port
<< ": " << errstr
<< " [try again in 3s]"
<< endl;
delayed_send(mm, seconds(3), get_atom::value, m_host, m_port);
},
[=](rebind_atom, string& nhost, uint16_t nport) {
aout(this) << "*** rebind to " << nhost << ":" << nport << endl;
using std::swap;
swap(m_host, nhost);
swap(m_port, nport);
// await pending ok/error message first, then send new request to MM
become(
keep_behavior,
(on<ok_atom, actor_addr>() || on<error_atom, string>()) >> [=] {
unbecome();
send(mm, get_atom::value, m_host, m_port);
}
);
},
// simply ignore all requests until we have a connection
others >> skip_message
};
}
actor m_server;
string m_host;
uint16_t m_port;
};
// removes leading and trailing whitespaces
string trim(std::string s) {
auto not_space = [](char c) { return !isspace(c); }; auto not_space = [](char c) { return !isspace(c); };
// trim left // trim left
s.erase(s.begin(), find_if(s.begin(), s.end(), not_space)); s.erase(s.begin(), find_if(s.begin(), s.end(), not_space));
...@@ -52,132 +156,75 @@ inline string trim(std::string s) { ...@@ -52,132 +156,75 @@ inline string trim(std::string s) {
return s; return s;
} }
void client_bhvr(event_based_actor* self, const string& host, // tries to convert `str` to an int
uint16_t port, const actor& server) { optional<int> toint(const string& str) {
// recover from sync failures by trying to reconnect to server char* end;
if (!self->has_sync_failure_handler()) { auto result = static_cast<int>(strtol(str.c_str(), &end, 10));
self->on_sync_failure([=] { if (end == str.c_str() + str.size()) {
aout(self) << "*** lost connection to " << host return result;
<< ":" << port << endl;
client_bhvr(self, host, port, invalid_actor);
});
} }
// connect to server if needed return none;
if (!server) { }
aout(self) << "*** try to connect to " << host << ":" << port << endl;
try { // converts "+" to the atom '+' and "-" to the atom '-'
auto new_serv = io::remote_actor(host, port); optional<atom_value> plus_or_minus(const string& str) {
self->monitor(new_serv); if (str == "+") {
aout(self) << "reconnection succeeded" << endl; return {plus_atom::value};
client_bhvr(self, host, port, new_serv);
return;
}
catch (exception&) {
aout(self) << "connection failed, try again in 3s" << endl;
self->delayed_send(self, chrono::seconds(3), reconnect_atom::value);
}
} }
auto sync_send_request = [=](int lhs, const char* op, int rhs) { if (str == "-") {
self->sync_send(server, self->current_message()).then( return {minus_atom::value};
[=](result_atom, int result) { }
aout(self) << lhs << " " << op << " " << rhs << " = " << result << endl; return none;
}
);
};
self->become (
[=](plus_atom, int lhs, int rhs) {
sync_send_request(lhs, "+", rhs);
},
[=](minus_atom, int lhs, int rhs) {
sync_send_request(lhs, "-", rhs);
},
[=](const down_msg&) {
aout(self) << "*** server down, try to reconnect ..." << endl;
client_bhvr(self, host, port, invalid_actor);
},
[=](rebind_atom, const string& nhost, uint16_t nport) {
aout(self) << "*** rebind to new server: "
<< nhost << ":" << nport << endl;
client_bhvr(self, nhost, nport, invalid_actor);
},
[=](rebind_atom) {
client_bhvr(self, host, port, invalid_actor);
}
);
} }
void client_repl(const string& host, uint16_t port) { void client_repl(string host, uint16_t port) {
// keeps track of requests and tries to reconnect on server failures // keeps track of requests and tries to reconnect on server failures
cout << "Usage:\n" auto usage = [] {
"quit Quit the program\n" cout << "Usage:" << endl
"<x> + <y> Calculate <x>+<y> and print result\n" << " quit : terminates the program" << endl
"<x> - <y> Calculate <x>-<y> and print result\n" << " connect <host> <port> : connects to a remote actor" << endl
"connect <host> <port> Reconfigure server" << " <x> + <y> : adds two integers" << endl
<< endl << endl; << " <x> - <y> : subtracts two integers" << endl
string line; << endl;
auto client = spawn(client_bhvr, host, port, invalid_actor); };
const char connect[] = "connect "; usage();
while (getline(cin, line)) { bool done = false;
line = trim(std::move(line)); // ignore leading and trailing whitespaces auto client = spawn<client_impl>(std::move(host), port);
if (line == "quit") { // defining the handler outside the loop is more efficient as it avoids
// force client to quit // re-creating the same object over and over again
message_handler eval{
on("quit") >> [&] {
anon_send_exit(client, exit_reason::user_shutdown); anon_send_exit(client, exit_reason::user_shutdown);
return; done = true;
} else if (equal(begin(connect), end(connect) - 1, begin(line))) { },
vector<string> words; on("connect", arg_match) >> [&](string& nhost, string& sport) {
split(words, line, is_any_of(" ")); try {
message_builder(words.begin(), words.end()).apply({ auto lport = std::stoul(sport);
on("connect", arg_match) >> [&](string& nhost, string& sport) { if (lport < std::numeric_limits<uint16_t>::max()) {
try { anon_send(client, rebind_atom::value, move(nhost),
auto lport = std::stoul(sport); static_cast<uint16_t>(lport));
if (lport < std::numeric_limits<uint16_t>::max()) {
anon_send(client, rebind_atom::value, move(nhost),
static_cast<uint16_t>(lport));
}
else {
cout << lport << " is not a valid port" << endl;
}
}
catch (std::exception&) {
cout << "\"" << sport << "\" is not an unsigned integer"
<< endl;
}
},
others >> [] {
cout << "*** usage: connect <host> <port>" << endl;
}
});
} else {
auto toint = [](const string& str) -> optional<int> {
try { return {std::stoi(str)}; }
catch (std::exception&) {
cout << "\"" << str << "\" is not an integer" << endl;
return none;
} }
}; else {
bool success = false; cout << lport << " is not a valid port" << endl;
auto first = begin(line);
auto last = end(line);
auto pos = find_if(first, last, [](char c) { return c == '+' || c == '-'; });
if (pos != last) {
auto lsub = trim(string(first, pos));
auto rsub = trim(string(pos + 1, last));
auto lhs = toint(lsub);
auto rhs = toint(rsub);
if (lhs && rhs) {
atom_value op;
if (*pos == '+') {
op = plus_atom::value;
} else {
op = minus_atom::value;
}
anon_send(client, op, *lhs, *rhs);
} }
} }
else if (!success) { catch (std::exception&) {
cout << "*** invalid format; usage: <x> [+|-] <y>" << endl; cout << "\"" << sport << "\" is not an unsigned integer"
<< endl;
} }
} },
on(toint, plus_or_minus, toint) >> [&](int x, atom_value op, int y) {
anon_send(client, op, x, y);
},
others >> usage
};
// read next line, split it, and feed to the eval handler
string line;
while (!done && std::getline(std::cin, line)) {
line = trim(std::move(line)); // ignore leading and trailing whitespaces
std::vector<string> words;
split(words, line, is_any_of(" "), token_compress_on);
message_builder(words.begin(), words.end()).apply(eval);
} }
} }
...@@ -191,9 +238,11 @@ int main(int argc, char** argv) { ...@@ -191,9 +238,11 @@ int main(int argc, char** argv) {
{"client,c", "run in client mode"} {"client,c", "run in client mode"}
}); });
if (res.opts.count("help") > 0) { if (res.opts.count("help") > 0) {
// help text has already been printed
return 0; return 0;
} }
if (!res.remainder.empty()) { if (!res.remainder.empty()) {
// not all CLI arguments could be consumed
cerr << "*** invalid command line options" << endl << res.helptext << endl; cerr << "*** invalid command line options" << endl << res.helptext << endl;
return 1; return 1;
} }
...@@ -211,17 +260,23 @@ int main(int argc, char** argv) { ...@@ -211,17 +260,23 @@ int main(int argc, char** argv) {
return 1; return 1;
} }
if (is_server) { if (is_server) {
auto calc = spawn(calculator);
try { try {
// try to publish math actor at given port // try to publish math actor at given port
cout << "*** try publish at port " << port << endl; cout << "*** try publish at port " << port << endl;
auto p = io::publish(spawn(calculator), port); auto p = io::publish(calc, port);
cout << "*** server successfully published at port " << p << endl; cout << "*** server successfully published at port " << p << endl;
cout << "*** press [enter] to quit" << endl;
string dummy;
std::getline(std::cin, dummy);
cout << "... cya" << endl;
} }
catch (exception& e) { catch (std::exception& e) {
cerr << "*** unable to publish math actor at port " << port << "\n" cerr << "*** unable to publish math actor at port " << port << "\n"
<< to_verbose_string(e) // prints exception type and e.what() << to_verbose_string(e) // prints exception type and e.what()
<< endl; << endl;
} }
anon_send_exit(calc, exit_reason::user_shutdown);
} }
else { else {
client_repl(host, port); client_repl(host, port);
......
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