Commit a6cb56a0 authored by Dominik Charousset's avatar Dominik Charousset

Port examples to new atom constant syntax

parent 18bcf49c
/******************************************************************************\ /******************************************************************************\
* This example illustrates how to use aout. * * This example illustrates how to use aout. *
\ ******************************************************************************/ \******************************************************************************/
#include <random> #include <random>
#include <chrono> #include <chrono>
...@@ -20,16 +20,17 @@ int main() { ...@@ -20,16 +20,17 @@ int main() {
std::random_device rd; std::random_device rd;
std::default_random_engine re(rd()); std::default_random_engine re(rd());
std::chrono::milliseconds tout{re() % 10}; std::chrono::milliseconds tout{re() % 10};
self->delayed_send(self, tout, atom("done")); self->delayed_send(self, tout, 42);
self->receive(others() >> [i, self] { self->receive(
aout(self) << "Actor nr. " [i, self](int) {
<< i << " says goodbye!" << endl; aout(self) << "Actor nr. "
}); << i << " says goodbye!" << endl;
}
);
}); });
} }
// wait until all other actors we've spawned are done // wait until all other actors we've spawned are done
await_all_actors_done(); await_all_actors_done();
// done // done
shutdown(); shutdown();
return 0;
} }
...@@ -29,6 +29,10 @@ using namespace std; ...@@ -29,6 +29,10 @@ using namespace std;
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>;
using kickoff_atom = atom_constant<atom("kickoff")>;
// utility function to print an exit message with custom name // utility function to print an exit message with custom name
void print_on_exit(const actor& ptr, const std::string& name) { void print_on_exit(const actor& ptr, const std::string& name) {
ptr->attach_functor([=](std::uint32_t reason) { ptr->attach_functor([=](std::uint32_t reason) {
...@@ -39,12 +43,12 @@ void print_on_exit(const actor& ptr, const std::string& name) { ...@@ -39,12 +43,12 @@ void print_on_exit(const actor& ptr, const std::string& name) {
behavior ping(event_based_actor* self, size_t num_pings) { behavior ping(event_based_actor* self, size_t num_pings) {
auto count = make_shared<size_t>(0); auto count = make_shared<size_t>(0);
return { return {
on(atom("kickoff"), arg_match) >> [=](const actor& pong) { [=](kickoff_atom, const actor& pong) {
self->send(pong, atom("ping"), int32_t(1)); self->send(pong, ping_atom::value, int32_t(1));
self->become ( self->become (
on(atom("pong"), arg_match) >> [=](int32_t value) -> message { [=](pong_atom, int32_t value) -> message {
if (++*count >= num_pings) self->quit(); if (++*count >= num_pings) self->quit();
return make_message(atom("ping"), value + 1); return make_message(ping_atom::value, value + 1);
} }
); );
} }
...@@ -53,8 +57,8 @@ behavior ping(event_based_actor* self, size_t num_pings) { ...@@ -53,8 +57,8 @@ behavior ping(event_based_actor* self, size_t num_pings) {
behavior pong() { behavior pong() {
return { return {
on(atom("ping"), arg_match) >> [](int32_t value) { [](ping_atom, int32_t value) {
return make_message(atom("pong"), value); return make_message(pong_atom::value, value);
} }
}; };
} }
...@@ -113,9 +117,8 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -113,9 +117,8 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
} }
}, },
[=](atom_value av, int32_t i) { [=](atom_value av, int32_t i) {
assert(av == atom("ping") || av == atom("pong")); assert(av == ping_atom::value || av == pong_atom::value);
aout(self) << "send {" << to_string(av) << ", " << i << "}" aout(self) << "send {" << to_string(av) << ", " << i << "}" << endl;
<< endl;
// cast atom to its underlying type, i.e., uint64_t // cast atom to its underlying type, i.e., uint64_t
write_int(self, hdl, static_cast<uint64_t>(av)); write_int(self, hdl, static_cast<uint64_t>(av));
write_int(self, hdl, i); write_int(self, hdl, i);
...@@ -185,7 +188,7 @@ int main(int argc, char** argv) { ...@@ -185,7 +188,7 @@ int main(int argc, char** argv) {
auto io_actor = spawn_io_client(broker_impl, host, port, ping_actor); auto io_actor = spawn_io_client(broker_impl, host, port, ping_actor);
print_on_exit(io_actor, "protobuf_io"); print_on_exit(io_actor, "protobuf_io");
print_on_exit(ping_actor, "ping"); print_on_exit(ping_actor, "ping");
send_as(io_actor, ping_actor, atom("kickoff"), io_actor); send_as(io_actor, ping_actor, kickoff_atom::value, io_actor);
}, },
others() >> [] { others() >> [] {
cerr << "use with eihter '-s PORT' as server or '-c HOST PORT' as client" cerr << "use with eihter '-s PORT' as server or '-c HOST PORT' as client"
......
...@@ -11,6 +11,8 @@ using std::endl; ...@@ -11,6 +11,8 @@ using std::endl;
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
using tick_atom = atom_constant<atom("tick")>;
const char http_ok[] = R"__(HTTP/1.1 200 OK const char http_ok[] = R"__(HTTP/1.1 200 OK
Content-Type: text/plain Content-Type: text/plain
Connection: keep-alive Connection: keep-alive
...@@ -44,7 +46,7 @@ behavior connection_worker(broker* self, connection_handle hdl) { ...@@ -44,7 +46,7 @@ behavior connection_worker(broker* self, connection_handle hdl) {
behavior server(broker* self) { behavior server(broker* self) {
auto counter = std::make_shared<int>(0); auto counter = std::make_shared<int>(0);
self->delayed_send(self, std::chrono::seconds(1), atom("tick")); self->delayed_send(self, std::chrono::seconds(1), tick_atom::value);
return { return {
[=](const new_connection_msg& ncm) { [=](const new_connection_msg& ncm) {
auto worker = self->fork(connection_worker, ncm.handle); auto worker = self->fork(connection_worker, ncm.handle);
...@@ -54,10 +56,10 @@ behavior server(broker* self) { ...@@ -54,10 +56,10 @@ behavior server(broker* self) {
[=](const down_msg&) { [=](const down_msg&) {
++*counter; ++*counter;
}, },
on(atom("tick")) >> [=] { [=](tick_atom) {
aout(self) << "Finished " << *counter << " requests per second." << endl; aout(self) << "Finished " << *counter << " requests per second." << endl;
*counter = 0; *counter = 0;
self->delayed_send(self, std::chrono::seconds(1), atom("tick")); self->delayed_send(self, std::chrono::seconds(1), tick_atom::value);
}, },
others() >> [=] { others() >> [=] {
aout(self) << "unexpected: " << to_string(self->last_dequeued()) << endl; aout(self) << "unexpected: " << to_string(self->last_dequeued()) << endl;
......
...@@ -7,25 +7,25 @@ ...@@ -7,25 +7,25 @@
* * * *
* Schematic view: * * Schematic view: *
* * * *
* client | client_job | curl_master | curl_worker * * client | client_job | curl_master | curl_worker *
* /--------------|*|-------------\ /-------------|*| * * /--------------|*|-------------\ /-------------|*| *
* /---------------|*|--------------\ / * * /---------------|*|--------------\ / *
* /----------------|*|---------------\ / * * /----------------|*|---------------\ / *
* |*| ----------------|*|----------------|*|----------------|*| * * |*| ----------------|*|----------------|*|----------------|*| *
* \________________|*|_______________/ \ * * \________________|*|_______________/ \ *
* \_______________|*|______________/ \ * * \_______________|*|______________/ \ *
* \______________|*|_____________/ \-------------|*| * * \______________|*|_____________/ \-------------|*| *
* * * *
* * * *
* Communication pattern: * * Communication pattern: *
* * * *
* client_job curl_master curl_worker * * client_job curl_master curl_worker *
* | | | * * | | | *
* | ----(read)-----> | | * * | ----(read)-----> | | *
* | | --(forward)----> | * * | | --(forward)----> | *
* | |---\ * * | |---\ *
* | | | * * | | | *
* | |<--/ * * | |<--/ *
* | <-------------(reply)-------------- | * * | <-------------(reply)-------------- | *
* X * * X *
\ ******************************************************************************/ \ ******************************************************************************/
...@@ -49,15 +49,21 @@ ...@@ -49,15 +49,21 @@
// disable some clang warnings here caused by CURL // disable some clang warnings here caused by CURL
#ifdef __clang__ #ifdef __clang__
# pragma clang diagnostic ignored "-Wshorten-64-to-32" # pragma clang diagnostic ignored "-Wshorten-64-to-32"
# pragma clang diagnostic ignored "-Wdisabled-macro-expansion" # pragma clang diagnostic ignored "-Wdisabled-macro-expansion"
# pragma clang diagnostic ignored "-Wunused-const-variable" # pragma clang diagnostic ignored "-Wunused-const-variable"
#endif // __clang__ #endif // __clang__
using namespace caf; using namespace caf;
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
using read_atom = atom_constant<atom("read")>;
using fail_atom = atom_constant<atom("fail")>;
using next_atom = atom_constant<atom("next")>;
using reply_atom = atom_constant<atom("reply")>;
using finished_atom = atom_constant<atom("finished")>;
namespace color { namespace color {
// UNIX terminal color codes // UNIX terminal color codes
...@@ -137,24 +143,19 @@ class client_job : public base_actor { ...@@ -137,24 +143,19 @@ class client_job : public base_actor {
protected: protected:
behavior make_behavior() override { behavior make_behavior() override {
print() << "init" << color::reset_endl; print() << "init" << color::reset_endl;
send(m_parent, send(m_parent, read_atom::value, "http://www.example.com/index.html",
atom("read"), uint64_t{0}, uint64_t{4095});
"http://www.example.com/index.html", return {
static_cast<uint64_t>(0), [=](reply_atom, const buffer_type& buf) {
static_cast<uint64_t>(4095)); print() << "successfully received " << buf.size() << " bytes"
return ( << color::reset_endl;
on(atom("reply"), arg_match) >> [=](const buffer_type& buf) {
print() << "successfully received "
<< buf.size()
<< " bytes"
<< color::reset_endl;
quit(); quit();
}, },
on(atom("fail")) >> [=] { [=](fail_atom) {
print() << "failure" << color::reset_endl; print() << "failure" << color::reset_endl;
quit(); quit();
} }
); };
} }
}; };
...@@ -182,9 +183,9 @@ class client : public base_actor { ...@@ -182,9 +183,9 @@ class client : public base_actor {
link_to(m_parent); link_to(m_parent);
print() << "init" << color::reset_endl; print() << "init" << color::reset_endl;
// start 'loop' // start 'loop'
send(this, atom("next")); send(this, next_atom::value);
return ( return (
on(atom("next")) >> [=] { [=](next_atom) {
print() << "spawn new client_job (nr. " print() << "spawn new client_job (nr. "
<< ++m_count << ++m_count
<< ")" << ")"
...@@ -194,7 +195,7 @@ class client : public base_actor { ...@@ -194,7 +195,7 @@ class client : public base_actor {
spawn<client_job, detached+linked>(m_parent); spawn<client_job, detached+linked>(m_parent);
// compute random delay until next job is launched // compute random delay until next job is launched
auto delay = m_dist(m_re); auto delay = m_dist(m_re);
delayed_send(this, milliseconds(delay), atom("next")); delayed_send(this, milliseconds(delay), next_atom::value);
} }
); );
} }
...@@ -227,9 +228,8 @@ class curl_worker : public base_actor { ...@@ -227,9 +228,8 @@ class curl_worker : public base_actor {
curl_easy_setopt(m_curl, CURLOPT_WRITEFUNCTION, &curl_worker::cb); curl_easy_setopt(m_curl, CURLOPT_WRITEFUNCTION, &curl_worker::cb);
curl_easy_setopt(m_curl, CURLOPT_NOSIGNAL, 1); curl_easy_setopt(m_curl, CURLOPT_NOSIGNAL, 1);
return ( return (
on(atom("read"), arg_match) [=](read_atom, const std::string& fname, uint64_t offset, uint64_t range)
>> [=](const std::string& fname, uint64_t offset, uint64_t range) -> message {
-> message {
print() << "read" << color::reset_endl; print() << "read" << color::reset_endl;
for (;;) { for (;;) {
m_buf.clear(); m_buf.clear();
...@@ -266,8 +266,8 @@ class curl_worker : public base_actor { ...@@ -266,8 +266,8 @@ class curl_worker : public base_actor {
<< hc << hc
<< color::reset_endl; << color::reset_endl;
// tell parent that this worker is done // tell parent that this worker is done
send(m_parent, atom("finished")); send(m_parent, finished_atom::value);
return make_message(atom("reply"), m_buf); return make_message(reply_atom::value, m_buf);
case 404: // file does not exist case 404: // file does not exist
print() << "http error: download failed with " print() << "http error: download failed with "
<< "'HTTP RETURN CODE': 404 (file does " << "'HTTP RETURN CODE': 404 (file does "
...@@ -305,7 +305,6 @@ curl_worker::~curl_worker() { ...@@ -305,7 +305,6 @@ curl_worker::~curl_worker() {
// avoid weak-vtables warning // avoid weak-vtables warning
} }
// manages {num_curl_workers} workers with a round-robin protocol // manages {num_curl_workers} workers with a round-robin protocol
class curl_master : public base_actor { class curl_master : public base_actor {
public: public:
...@@ -324,45 +323,37 @@ class curl_master : public base_actor { ...@@ -324,45 +323,37 @@ class curl_master : public base_actor {
} }
auto worker_finished = [=] { auto worker_finished = [=] {
auto sender = last_sender(); auto sender = last_sender();
auto i = std::find(m_busy_worker.begin(), auto i = std::find(m_busy_worker.begin(), m_busy_worker.end(), sender);
m_busy_worker.end(),
sender);
m_idle_worker.push_back(*i); m_idle_worker.push_back(*i);
m_busy_worker.erase(i); m_busy_worker.erase(i);
print() << "worker is done" << color::reset_endl; print() << "worker is done" << color::reset_endl;
}; };
print() << "spawned " print() << "spawned " << m_idle_worker.size()
<< m_idle_worker.size() << " worker(s)" << color::reset_endl;
<< " worker" return {
<< color::reset_endl; [=](read_atom, const std::string&, uint64_t, uint64_t) {
return (
on(atom("read"), arg_match) >> [=](const std::string&,
uint64_t,
uint64_t) {
print() << "received {'read'}" << color::reset_endl; print() << "received {'read'}" << color::reset_endl;
// forward job to first idle worker // forward job to an idle worker
actor worker = m_idle_worker.front(); actor worker = m_idle_worker.back();
m_idle_worker.erase(m_idle_worker.begin()); m_idle_worker.pop_back();
m_busy_worker.push_back(worker); m_busy_worker.push_back(worker);
forward_to(worker); forward_to(worker);
print() << m_busy_worker.size() print() << m_busy_worker.size() << " active jobs" << color::reset_endl;
<< " active jobs"
<< color::reset_endl;
if (m_idle_worker.empty()) { if (m_idle_worker.empty()) {
// wait until at least one worker finished its job // wait until at least one worker finished its job
become ( become (
keep_behavior, keep_behavior,
on(atom("finished")) >> [=] { [=](finished_atom) {
worker_finished(); worker_finished();
unbecome(); unbecome();
} }
); );
} }
}, },
on(atom("finished")) >> [=] { [=](finished_atom) {
worker_finished(); worker_finished();
} }
); };
} }
private: private:
...@@ -409,9 +400,9 @@ int main() { ...@@ -409,9 +400,9 @@ int main() {
act.sa_handler = [](int) { abort(); }; act.sa_handler = [](int) { abort(); };
set_sighandler(); set_sighandler();
aout(self) << color::cyan aout(self) << color::cyan
<< "await CURL; this may take a while " << "await CURL; this may take a while "
"(press CTRL+C again to abort)" "(press CTRL+C again to abort)"
<< color::reset_endl; << color::reset_endl;
self->await_all_other_actors_done(); self->await_all_other_actors_done();
} }
// shutdown libcaf // shutdown libcaf
......
...@@ -13,42 +13,37 @@ using std::cout; ...@@ -13,42 +13,37 @@ using std::cout;
using std::endl; using std::endl;
using namespace caf; using namespace caf;
using plus_atom = atom_constant<atom("plus")>;
using minus_atom = atom_constant<atom("minus")>;
// implementation using the blocking API // implementation using the blocking API
void blocking_math_fun(blocking_actor* self) { void blocking_calculator(blocking_actor* self) {
bool done = false; self->receive_loop (
self->do_receive ( [](plus_atom, int a, int b) {
// "arg_match" matches the parameter types of given lambda expression return a + b;
// thus, it's equal to
// - on<atom("plus"), int, int>()
// - on(atom("plus"), val<int>, val<int>)
on(atom("plus"), arg_match) >> [](int a, int b) {
return std::make_tuple(atom("result"), a + b);
}, },
on(atom("minus"), arg_match) >> [](int a, int b) { [](minus_atom, int a, int b) {
return std::make_tuple(atom("result"), a - b); return a - b;
}, },
on(atom("quit")) >> [&]() { others() >> [=] {
// note: this actor uses the blocking API, hence self->quit() cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
// would force stack unwinding by throwing an exception
done = true;
} }
).until(done); );
} }
void calculator(event_based_actor* self) { // execute this behavior until actor terminates
// execute this behavior until actor terminates behavior calculator(event_based_actor* self) {
self->become ( return behavior{
on(atom("plus"), arg_match) >> [](int a, int b) { [](plus_atom, int a, int b) {
return std::make_tuple(atom("result"), a + b); return a + b;
}, },
on(atom("minus"), arg_match) >> [](int a, int b) { [](minus_atom, int a, int b) {
return std::make_tuple(atom("result"), a - b); return a - b;
}, },
on(atom("quit")) >> [=] { others() >> [=] {
// terminate actor with normal exit reason cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
self->quit();
} }
); };
} }
void tester(event_based_actor* self, const actor& testee) { void tester(event_based_actor* self, const actor& testee) {
...@@ -59,15 +54,14 @@ void tester(event_based_actor* self, const actor& testee) { ...@@ -59,15 +54,14 @@ void tester(event_based_actor* self, const actor& testee) {
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
}); });
// first test: 2 + 1 = 3 // first test: 2 + 1 = 3
self->sync_send(testee, atom("plus"), 2, 1).then( self->sync_send(testee, plus_atom::value, 2, 1).then(
on(atom("result"), 3) >> [=] { on(3) >> [=] {
// second test: 2 - 1 = 1 // second test: 2 - 1 = 1
self->sync_send(testee, atom("minus"), 2, 1).then( self->sync_send(testee, minus_atom::value, 2, 1).then(
on(atom("result"), 1) >> [=] { on(1) >> [=] {
// both tests succeeded // both tests succeeded
aout(self) << "AUT (actor under test) seems to be ok" aout(self) << "AUT (actor under test) seems to be ok" << endl;
<< endl; self->quit(exit_reason::user_shutdown);
self->send(testee, atom("quit"));
} }
); );
} }
...@@ -75,8 +69,11 @@ void tester(event_based_actor* self, const actor& testee) { ...@@ -75,8 +69,11 @@ void tester(event_based_actor* self, const actor& testee) {
} }
int main() { int main() {
cout << "test blocking actor" << endl;
spawn(tester, spawn<blocking_api>(blocking_calculator));
await_all_actors_done();
cout << "test event-based actor" << endl;
spawn(tester, spawn(calculator)); spawn(tester, spawn(calculator));
await_all_actors_done(); await_all_actors_done();
shutdown(); shutdown();
return 0;
} }
...@@ -13,6 +13,8 @@ using std::pair; ...@@ -13,6 +13,8 @@ using std::pair;
using namespace caf; using namespace caf;
using step_atom = atom_constant<atom("step")>;
// ASCII art figures // ASCII art figures
constexpr const char* figures[] = { constexpr const char* figures[] = {
"<(^.^<)", "<(^.^<)",
...@@ -30,11 +32,6 @@ constexpr animation_step animation_steps[] = { ...@@ -30,11 +32,6 @@ constexpr animation_step animation_steps[] = {
{0, 12}, {0, 11}, {0, 10}, {0, 9}, {0, 8}, {0, 7}, {1, 7} {0, 12}, {0, 11}, {0, 10}, {0, 9}, {0, 8}, {0, 7}, {1, 7}
}; };
template <class T, size_t S>
constexpr size_t array_size(const T (&) [S]) {
return S;
}
constexpr size_t animation_width = 20; constexpr size_t animation_width = 20;
// "draws" an animation step by printing "{offset_whitespaces}{figure}{padding}" // "draws" an animation step by printing "{offset_whitespaces}{figure}{padding}"
...@@ -55,19 +52,20 @@ void draw_kirby(const animation_step& animation) { ...@@ -55,19 +52,20 @@ void draw_kirby(const animation_step& animation) {
// uses a message-based loop to iterate over all animation steps // uses a message-based loop to iterate over all animation steps
void dancing_kirby(event_based_actor* self) { void dancing_kirby(event_based_actor* self) {
// let's get it started // let's get it started
self->send(self, atom("Step"), size_t{0}); self->send(self, step_atom::value, size_t{0});
self->become ( self->become (
on(atom("Step"), array_size(animation_steps)) >> [=] { [=](step_atom, size_t step) {
// we've printed all animation steps (done) if (step == sizeof(animation_step)) {
cout << endl; // we've printed all animation steps (done)
self->quit(); cout << endl;
}, self->quit();
on(atom("Step"), arg_match) >> [=](size_t step) { return;
}
// print given step // print given step
draw_kirby(animation_steps[step]); draw_kirby(animation_steps[step]);
// animate next step in 150ms // animate next step in 150ms
self->delayed_send(self, std::chrono::milliseconds(150), self->delayed_send(self, std::chrono::milliseconds(150),
atom("Step"), step + 1); step_atom::value, step + 1);
} }
); );
} }
...@@ -76,5 +74,4 @@ int main() { ...@@ -76,5 +74,4 @@ int main() {
spawn(dancing_kirby); spawn(dancing_kirby);
await_all_actors_done(); await_all_actors_done();
shutdown(); shutdown();
return 0;
} }
...@@ -18,160 +18,176 @@ using namespace caf; ...@@ -18,160 +18,176 @@ using namespace caf;
namespace { namespace {
// atoms for chopstick interface
using put_atom = atom_constant<atom("put")>;
using take_atom = atom_constant<atom("take")>;
using busy_atom = atom_constant<atom("busy")>;
using taken_atom = atom_constant<atom("taken")>;
// atoms for philosopher interface
using eat_atom = atom_constant<atom("eat")>;
using think_atom = atom_constant<atom("think")>;
// a chopstick
using chopstick = typed_actor<replies_to<take_atom>
::with_either<taken_atom>
::or_else<busy_atom>,
reacts_to<put_atom>>;
chopstick::behavior_type taken_chopstick(chopstick::pointer self, actor_addr);
// either taken by a philosopher or available // either taken by a philosopher or available
void chopstick(event_based_actor* self) { chopstick::behavior_type available_chopstick(chopstick::pointer self) {
self->become( return {
on(atom("take"), arg_match) >> [=](const actor& philos) { [=](take_atom) {
// tell philosopher it took this chopstick self->become(taken_chopstick(self, self->last_sender()));
self->send(philos, atom("taken"), self); return taken_atom::value;
// await 'put' message and reject other 'take' messages },
self->become( [](put_atom) {
// allows us to return to the previous behavior cerr << "chopstick received unexpected 'put'" << endl;
keep_behavior,
on(atom("take"), arg_match) >> [=](const actor& other) {
self->send(other, atom("busy"), self);
},
on(atom("put"), philos) >> [=] {
// return to previous behaivor, i.e., await next 'take'
self->unbecome();
}
);
} }
); };
} }
/* See: http://www.dalnefre.com/wp/2010/08/dining-philosophers-in-humus/ chopstick::behavior_type taken_chopstick(chopstick::pointer self,
actor_addr user) {
return {
[](take_atom) {
return busy_atom::value;
},
[=](put_atom) {
if (self->last_sender() == user) {
self->become(available_chopstick(self));
}
}
};
}
/* Based on: http://www.dalnefre.com/wp/2010/08/dining-philosophers-in-humus/
* *
* *
* +-------------+ {(busy|taken), Y} * +-------------+ {busy|taken}
* /-------->| thinking |<------------------\ * /-------->| thinking |<------------------\
* | +-------------+ | * | +-------------+ |
* | | | * | | |
* | | {eat} | * | | {eat} |
* | | | * | | |
* | V | * | V |
* | +-------------+ {busy, X} +-------------+ * | +-------------+ {busy} +-------------+
* | | hungry |----------->| denied | * | | hungry |----------->| denied |
* | +-------------+ +-------------+ * | +-------------+ +-------------+
* | | * | |
* | | {taken, X} * | | {taken}
* | | * | |
* | V * | V
* | +-------------+ * | +-------------+
* | | wait_for(Y) | * | | granted |
* | +-------------+ * | +-------------+
* | | | * | | |
* | {busy, Y} | | {taken, Y} * | {busy} | | {taken}
* \-----------/ | * \-----------/ |
* | V * | V
* | {think} +-------------+ * | {think} +-------------+
* \---------| eating | * \---------| eating |
* +-------------+ * +-------------+
*
*
* [ X = left => Y = right ]
* [ X = right => Y = left ]
*/ */
class philosopher : public event_based_actor { class philosopher : public event_based_actor {
public: public:
philosopher(const std::string& n, const actor& l, const actor& r) philosopher(const std::string& n, const chopstick& l, const chopstick& r)
: name(n), left(l), right(r) { : name(n),
left(l),
right(r) {
// a philosopher that receives {eat} stops thinking and becomes hungry // a philosopher that receives {eat} stops thinking and becomes hungry
thinking = ( thinking = behavior{
on(atom("eat")) >> [=] { [=](eat_atom) {
become(hungry); become(hungry);
send(left, atom("take"), this); send(left, take_atom::value);
send(right, atom("take"), this); send(right, take_atom::value);
} }
); };
// wait for the first answer of a chopstick // wait for the first answer of a chopstick
hungry = ( hungry = behavior{
on(atom("taken"), left) >> [=] { [=](taken_atom) {
become(waiting_for(right)); become(granted);
},
on(atom("taken"), right) >> [=] {
become(waiting_for(left));
}, },
on<atom("busy"), actor>() >> [=] { [=](busy_atom) {
become(denied); become(denied);
} }
); };
// philosopher was not able to obtain the first chopstick // philosopher was able to obtain the first chopstick
denied = ( granted = behavior{
on(atom("taken"), arg_match) >> [=](const actor& ptr) { [=](taken_atom) {
send(ptr, atom("put"), this); aout(this) << name
send(this, atom("eat")); << " has picked up chopsticks with IDs "
<< left->id() << " and " << right->id()
<< " and starts to eat\n";
// eat some time
delayed_send(this, seconds(5), think_atom::value);
become(eating);
},
[=](busy_atom) {
send(last_sender() == left ? right : left, put_atom::value);
send(this, eat_atom::value);
become(thinking);
}
};
// philosopher was *not* able to obtain the first chopstick
denied = behavior{
[=](taken_atom) {
send(last_sender() == left ? left : right, put_atom::value);
send(this, eat_atom::value);
become(thinking); become(thinking);
}, },
on<atom("busy"), actor>() >> [=] { [=](busy_atom) {
send(this, atom("eat")); send(this, eat_atom::value);
become(thinking); become(thinking);
} }
); };
// philosopher obtained both chopstick and eats (for five seconds) // philosopher obtained both chopstick and eats (for five seconds)
eating = ( eating = behavior{
on(atom("think")) >> [=] { [=](think_atom) {
send(left, atom("put"), this); send(left, put_atom::value);
send(right, atom("put"), this); send(right, put_atom::value);
delayed_send(this, seconds(5), atom("eat")); delayed_send(this, seconds(5), eat_atom::value);
aout(this) << name << " puts down his chopsticks and starts to think\n"; aout(this) << name << " puts down his chopsticks and starts to think\n";
become(thinking); become(thinking);
} }
); };
} }
protected: protected:
behavior make_behavior() override { behavior make_behavior() override {
// start thinking // start thinking
send(this, atom("think")); send(this, think_atom::value);
// philosophers start to think after receiving {think} // philosophers start to think after receiving {think}
return ( return (
on(atom("think")) >> [=] { [=](think_atom) {
aout(this) << name << " starts to think\n"; aout(this) << name << " starts to think\n";
delayed_send(this, seconds(5), atom("eat")); delayed_send(this, seconds(5), eat_atom::value);
become(thinking); become(thinking);
} }
); );
} }
private: private:
// wait for second chopstick std::string name; // the name of this philosopher
behavior waiting_for(const actor& what) { chopstick left; // left chopstick
return { chopstick right; // right chopstick
on(atom("taken"), what) >> [=] { behavior thinking; // initial behavior
aout(this) << name behavior hungry; // tries to take chopsticks
<< " has picked up chopsticks with IDs " behavior granted; // has one chopstick and waits for the second one
<< left->id() << " and " << right->id() behavior denied; // could not get first chopsticks
<< " and starts to eat\n"; behavior eating; // waits for some time, then go thinking again
// eat some time
delayed_send(this, seconds(5), atom("think"));
become(eating);
},
on(atom("busy"), what) >> [=] {
send((what == left) ? right : left, atom("put"), this);
send(this, atom("eat"));
become(thinking);
}
};
}
std::string name; // the name of this philosopher
actor left; // left chopstick
actor right; // right chopstick
behavior thinking;
behavior hungry; // tries to take chopsticks
behavior denied; // could not get chopsticks
behavior eating; // wait for some time, then go thinking again
}; };
void dining_philosophers() { void dining_philosophers() {
scoped_actor self; scoped_actor self;
// create five chopsticks // create five chopsticks
aout(self) << "chopstick ids are:"; aout(self) << "chopstick ids are:";
std::vector<actor> chopsticks; std::vector<chopstick> chopsticks;
for (size_t i = 0; i < 5; ++i) { for (size_t i = 0; i < 5; ++i) {
chopsticks.push_back(spawn(chopstick)); chopsticks.push_back(spawn_typed(available_chopstick));
aout(self) << " " << chopsticks.back()->id(); aout(self) << " " << chopsticks.back()->id();
} }
aout(self) << endl; aout(self) << endl;
...@@ -190,5 +206,4 @@ int main(int, char**) { ...@@ -190,5 +206,4 @@ int main(int, char**) {
// real philosophers are never done // real philosophers are never done
await_all_actors_done(); await_all_actors_done();
shutdown(); shutdown();
return 0;
} }
...@@ -27,17 +27,21 @@ ...@@ -27,17 +27,21 @@
using namespace std; using namespace std;
using namespace caf; using namespace caf;
using plus_atom = atom_constant<atom("plus")>;
using minus_atom = atom_constant<atom("minus")>;
using result_atom = atom_constant<atom("result")>;
using rebind_atom = atom_constant<atom("rebind")>;
using connect_atom = atom_constant<atom("connect")>;
using reconnect_atom = atom_constant<atom("reconnect")>;
// our "service" // our "service"
void calculator(event_based_actor* self) { void calculator(event_based_actor* self) {
self->become ( self->become (
on(atom("plus"), arg_match) >> [](int a, int b) -> message { [](plus_atom, int a, int b) -> message {
return make_message(atom("result"), a + b); return make_message(result_atom::value, a + b);
},
on(atom("minus"), arg_match) >> [](int a, int b) -> message {
return make_message(atom("result"), a - b);
}, },
on(atom("quit")) >> [=] { [](minus_atom, int a, int b) -> message {
self->quit(); return make_message(result_atom::value, a - b);
} }
); );
} }
...@@ -57,7 +61,7 @@ void client_bhvr(event_based_actor* self, const string& host, ...@@ -57,7 +61,7 @@ void client_bhvr(event_based_actor* self, const string& host,
if (!self->has_sync_failure_handler()) { if (!self->has_sync_failure_handler()) {
self->on_sync_failure([=] { self->on_sync_failure([=] {
aout(self) << "*** lost connection to " << host aout(self) << "*** lost connection to " << host
<< ":" << port << endl; << ":" << port << endl;
client_bhvr(self, host, port, invalid_actor); client_bhvr(self, host, port, invalid_actor);
}); });
} }
...@@ -73,38 +77,33 @@ void client_bhvr(event_based_actor* self, const string& host, ...@@ -73,38 +77,33 @@ void client_bhvr(event_based_actor* self, const string& host,
} }
catch (exception&) { catch (exception&) {
aout(self) << "connection failed, try again in 3s" << endl; aout(self) << "connection failed, try again in 3s" << endl;
self->delayed_send(self, chrono::seconds(3), atom("reconnect")); self->delayed_send(self, chrono::seconds(3), reconnect_atom::value);
} }
} }
// our predicate guarding the first callback auto sync_send_request = [=](int lhs, const char* op, int rhs) {
auto pred = [=](atom_value val) -> optional<atom_value> { self->sync_send_tuple(server, self->last_dequeued()).then(
if (server != invalid_actor [=](result_atom, int result) {
&& (val == atom("plus") || val == atom("minus"))) { aout(self) << lhs << " " << op << " " << rhs << " = " << result << endl;
return val; }
} );
return none;
}; };
self->become ( self->become (
on(pred, val<int>, val<int>) >> [=](atom_value op, int lhs, int rhs) { [=](plus_atom, int lhs, int rhs) {
self->sync_send_tuple(server, self->last_dequeued()).then( sync_send_request(lhs, "+", rhs);
on(atom("result"), arg_match) >> [=](int result) { },
aout(self) << lhs << " " [=](minus_atom, int lhs, int rhs) {
<< to_string(op) << " " sync_send_request(lhs, "-", rhs);
<< rhs << " = "
<< result << endl;
}
);
}, },
[=](const down_msg&) { [=](const down_msg&) {
aout(self) << "*** server down, try to reconnect ..." << endl; aout(self) << "*** server down, try to reconnect ..." << endl;
client_bhvr(self, host, port, invalid_actor); client_bhvr(self, host, port, invalid_actor);
}, },
on(atom("rebind"), arg_match) >> [=](const string& nhost, uint16_t nport) { [=](rebind_atom, const string& nhost, uint16_t nport) {
aout(self) << "*** rebind to new server: " aout(self) << "*** rebind to new server: "
<< nhost << ":" << nport << endl; << nhost << ":" << nport << endl;
client_bhvr(self, nhost, nport, invalid_actor); client_bhvr(self, nhost, nport, invalid_actor);
}, },
on(atom("reconnect")) >> [=] { [=](rebind_atom) {
client_bhvr(self, host, port, invalid_actor); client_bhvr(self, host, port, invalid_actor);
} }
); );
...@@ -133,8 +132,8 @@ void client_repl(const string& host, uint16_t port) { ...@@ -133,8 +132,8 @@ void client_repl(const string& host, uint16_t port) {
try { try {
auto lport = std::stoul(sport); auto lport = std::stoul(sport);
if (lport < std::numeric_limits<uint16_t>::max()) { if (lport < std::numeric_limits<uint16_t>::max()) {
anon_send(client, atom("rebind"), move(nhost), anon_send(client, rebind_atom::value, move(nhost),
static_cast<uint16_t>(lport)); static_cast<uint16_t>(lport));
} }
else { else {
cout << lport << " is not a valid port" << endl; cout << lport << " is not a valid port" << endl;
...@@ -167,7 +166,7 @@ void client_repl(const string& host, uint16_t port) { ...@@ -167,7 +166,7 @@ void client_repl(const string& host, uint16_t port) {
auto lhs = toint(lsub); auto lhs = toint(lsub);
auto rhs = toint(rsub); auto rhs = toint(rsub);
if (lhs && rhs) { if (lhs && rhs) {
auto op = (*pos == '+') ? atom("plus") : atom("minus"); auto op = (*pos == '+') ? plus_atom::value : minus_atom::value;
anon_send(client, op, *lhs, *rhs); anon_send(client, op, *lhs, *rhs);
} }
} }
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
using namespace std; using namespace std;
using namespace caf; using namespace caf;
using join_atom = atom_constant<atom("join")>;
using broadcast_atom = atom_constant<atom("broadcast")>;
struct line { string str; }; struct line { string str; };
istream& operator>>(istream& is, line& l) { istream& operator>>(istream& is, line& l) {
...@@ -46,12 +49,12 @@ message split_line(const line& l) { ...@@ -46,12 +49,12 @@ message split_line(const line& l) {
void client(event_based_actor* self, const string& name) { void client(event_based_actor* self, const string& name) {
self->become ( self->become (
on(atom("broadcast"), arg_match) >> [=](const string& message) { [=](broadcast_atom, const string& message) {
for(auto& dest : self->joined_groups()) { for(auto& dest : self->joined_groups()) {
self->send(dest, name + ": " + message); self->send(dest, name + ": " + message);
} }
}, },
on(atom("join"), arg_match) >> [=](const group& what) { [=](join_atom, const group& what) {
for (auto g : self->joined_groups()) { for (auto g : self->joined_groups()) {
cout << "*** leave " << to_string(g) << endl; cout << "*** leave " << to_string(g) << endl;
self->send(self, g, name + " has left the chatroom"); self->send(self, g, name + " has left the chatroom");
...@@ -106,7 +109,7 @@ int main(int argc, char** argv) { ...@@ -106,7 +109,7 @@ int main(int argc, char** argv) {
auto group_uri = group_id.substr(p + 1); auto group_uri = group_id.substr(p + 1);
auto g = (module == "remote") ? io::remote_group(group_uri) auto g = (module == "remote") ? io::remote_group(group_uri)
: group::get(module, group_uri); : group::get(module, group_uri);
anon_send(client_actor, atom("join"), g); anon_send(client_actor, join_atom::value, g);
} }
catch (exception& e) { catch (exception& e) {
ostringstream err; ostringstream err;
...@@ -132,7 +135,7 @@ int main(int argc, char** argv) { ...@@ -132,7 +135,7 @@ int main(int argc, char** argv) {
try { try {
group grp = (mod == "remote") ? io::remote_group(id) group grp = (mod == "remote") ? io::remote_group(id)
: group::get(mod, id); : group::get(mod, id);
anon_send(client_actor, atom("join"), grp); anon_send(client_actor, join_atom::value, grp);
} }
catch (exception& e) { catch (exception& e) {
cerr << "*** exception: " << to_verbose_string(e) << endl; cerr << "*** exception: " << to_verbose_string(e) << endl;
...@@ -150,7 +153,7 @@ int main(int argc, char** argv) { ...@@ -150,7 +153,7 @@ int main(int argc, char** argv) {
}, },
others() >> [&] { others() >> [&] {
if (!s_last_line.empty()) { if (!s_last_line.empty()) {
anon_send(client_actor, atom("broadcast"), s_last_line); anon_send(client_actor, broadcast_atom::value, s_last_line);
} }
} }
); );
...@@ -158,5 +161,4 @@ int main(int argc, char** argv) { ...@@ -158,5 +161,4 @@ int main(int argc, char** argv) {
anon_send_exit(client_actor, exit_reason::user_shutdown); anon_send_exit(client_actor, exit_reason::user_shutdown);
await_all_actors_done(); await_all_actors_done();
shutdown(); shutdown();
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