Commit 4f70f3a1 authored by Dominik Charousset's avatar Dominik Charousset

Integrate review feedback

parent 1f25e71b
...@@ -39,7 +39,7 @@ using namespace caf::io; ...@@ -39,7 +39,7 @@ using namespace caf::io;
namespace { namespace {
// 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& hdl, const std::string& name) { void print_on_exit(const actor& hdl, const std::string& name) {
hdl->attach_functor([=](const error& reason) { hdl->attach_functor([=](const error& reason) {
cout << name << " exited: " << to_string(reason) << endl; cout << name << " exited: " << to_string(reason) << endl;
...@@ -73,7 +73,7 @@ behavior pong() { ...@@ -73,7 +73,7 @@ behavior pong() {
}; };
} }
// utility function for sending an integer type // Utility function for sending an integer type.
template <class T> template <class T>
void write_int(broker* self, connection_handle hdl, T value) { void write_int(broker* self, connection_handle hdl, T value) {
using unsigned_type = typename std::make_unsigned<T>::type; using unsigned_type = typename std::make_unsigned<T>::type;
...@@ -82,7 +82,7 @@ void write_int(broker* self, connection_handle hdl, T value) { ...@@ -82,7 +82,7 @@ void write_int(broker* self, connection_handle hdl, T value) {
self->flush(hdl); self->flush(hdl);
} }
// utility function for reading an ingeger from incoming data // Utility function for reading an ingeger from incoming data.
template <class T> template <class T>
void read_int(const void* data, T& storage) { void read_int(const void* data, T& storage) {
using unsigned_type = typename std::make_unsigned<T>::type; using unsigned_type = typename std::make_unsigned<T>::type;
...@@ -90,30 +90,30 @@ void read_int(const void* data, T& storage) { ...@@ -90,30 +90,30 @@ void read_int(const void* data, T& storage) {
storage = static_cast<T>(ntohl(static_cast<unsigned_type>(storage))); storage = static_cast<T>(ntohl(static_cast<unsigned_type>(storage)));
} }
// implementation of our broker // Implementation of our broker.
behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) { behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
// we assume io_fsm manages a broker with exactly one connection, // We assume io_fsm manages a broker with exactly one connection,
// i.e., the connection ponted to by `hdl` // i.e., the connection ponted to by `hdl`.
assert(self->num_connections() == 1); assert(self->num_connections() == 1);
// monitor buddy to quit broker if buddy is done // Monitor buddy to quit broker if buddy is done.
self->monitor(buddy); self->monitor(buddy);
self->set_down_handler([=](down_msg& dm) { self->set_down_handler([=](down_msg& dm) {
if (dm.source == buddy) { if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl; aout(self) << "our buddy is down" << endl;
// quit for same reason // Quit for same reason.
self->quit(dm.reason); self->quit(dm.reason);
} }
}); });
// setup: we are exchanging only messages consisting of an atom // Setup: we are exchanging only messages consisting of an atom
// (as uint64_t) and an integer value (int32_t) // (as uint64_t) and an integer value (int32_t).
self->configure_read( self->configure_read(
hdl, receive_policy::exactly(sizeof(uint64_t) + sizeof(int32_t))); hdl, receive_policy::exactly(sizeof(uint64_t) + sizeof(int32_t)));
// our message handlers // Our message handlers.
return { return {
[=](const connection_closed_msg& msg) { [=](const connection_closed_msg& msg) {
// brokers can multiplex any number of connections, however // Brokers can multiplex any number of connections, however
// this example assumes io_fsm to manage a broker with // this example assumes io_fsm to manage a broker with
// exactly one connection // exactly one connection.
if (msg.handle == hdl) { if (msg.handle == hdl) {
aout(self) << "connection closed" << endl; aout(self) << "connection closed" << endl;
// force buddy to quit // force buddy to quit
...@@ -132,16 +132,16 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -132,16 +132,16 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
write_int(self, hdl, i); write_int(self, hdl, i);
}, },
[=](const new_data_msg& msg) { [=](const new_data_msg& msg) {
// read the operation value as uint8_t from buffer // Read the operation value as uint8_t from buffer.
uint8_t op_val; uint8_t op_val;
read_int(msg.buf.data(), op_val); read_int(msg.buf.data(), op_val);
// read integer value from buffer, jumping to the correct // Read integer value from buffer, jumping to the correct
// position via offset_data(...) // position via offset_data(...).
int32_t ival; int32_t ival;
read_int(msg.buf.data() + sizeof(uint8_t), ival); read_int(msg.buf.data() + sizeof(uint8_t), ival);
// show some output // Show some output.
aout(self) << "received {" << op_val << ", " << ival << "}" << endl; aout(self) << "received {" << op_val << ", " << ival << "}" << endl;
// send composed message to our buddy // Send composed message to our buddy.
switch (static_cast<op>(op_val)) { switch (static_cast<op>(op_val)) {
case op::ping: case op::ping:
self->send(buddy, ping_atom_v, ival); self->send(buddy, ping_atom_v, ival);
...@@ -162,8 +162,8 @@ behavior server(broker* self, const actor& buddy) { ...@@ -162,8 +162,8 @@ behavior server(broker* self, const actor& buddy) {
return { return {
[=](const new_connection_msg& msg) { [=](const new_connection_msg& msg) {
aout(self) << "server accepted new connection" << endl; aout(self) << "server accepted new connection" << endl;
// by forking into a new broker, we are no longer // By forking into a new broker, we are no longer
// responsible for the connection // responsible for the connection.
auto impl = self->fork(broker_impl, msg.handle, buddy); auto impl = self->fork(broker_impl, msg.handle, buddy);
print_on_exit(impl, "broker_impl"); print_on_exit(impl, "broker_impl");
aout(self) << "quit server (only accept 1 connection)" << endl; aout(self) << "quit server (only accept 1 connection)" << endl;
......
...@@ -13,8 +13,6 @@ using namespace caf::io; ...@@ -13,8 +13,6 @@ using namespace caf::io;
namespace { namespace {
CAF_MSG_TYPE_ADD_ATOM(tick_atom);
constexpr const char http_ok[] = R"__(HTTP/1.1 200 OK constexpr const char http_ok[] = R"__(HTTP/1.1 200 OK
Content-Type: text/plain Content-Type: text/plain
Connection: keep-alive Connection: keep-alive
......
...@@ -134,15 +134,17 @@ behavior client_job(stateful_actor<base_state>* self, const actor& parent) { ...@@ -134,15 +134,17 @@ behavior client_job(stateful_actor<base_state>* self, const actor& parent) {
return {}; // returning an empty behavior terminates the actor return {}; // returning an empty behavior terminates the actor
self->send(parent, read_atom_v, "http://www.example.com/index.html", self->send(parent, read_atom_v, "http://www.example.com/index.html",
uint64_t{0}, uint64_t{4095}); uint64_t{0}, uint64_t{4095});
return {[=](reply_atom, const buffer_type& buf) { return {
self->state.print() << "successfully received " << buf.size() [=](reply_atom, const buffer_type& buf) {
<< " bytes" << color::reset_endl; self->state.print() << "successfully received " << buf.size() << " bytes"
<< color::reset_endl;
self->quit(); self->quit();
}, },
[=](fail_atom) { [=](fail_atom) {
self->state.print() << "failure" << color::reset_endl; self->state.print() << "failure" << color::reset_endl;
self->quit(); self->quit();
}}; },
};
} }
struct client_state : base_state { struct client_state : base_state {
...@@ -291,7 +293,8 @@ behavior curl_master(stateful_actor<master_state>* self) { ...@@ -291,7 +293,8 @@ behavior curl_master(stateful_actor<master_state>* self) {
}; };
self->state.print() << "spawned " << self->state.idle.size() << " worker(s)" self->state.print() << "spawned " << self->state.idle.size() << " worker(s)"
<< color::reset_endl; << color::reset_endl;
return {[=](read_atom rd, std::string& str, uint64_t x, uint64_t y) { return {
[=](read_atom rd, std::string& str, uint64_t x, uint64_t y) {
auto& st = self->state; auto& st = self->state;
st.print() << "received {'read'}" << color::reset_endl; st.print() << "received {'read'}" << color::reset_endl;
// forward job to an idle worker // forward job to an idle worker
...@@ -308,7 +311,8 @@ behavior curl_master(stateful_actor<master_state>* self) { ...@@ -308,7 +311,8 @@ behavior curl_master(stateful_actor<master_state>* self) {
}); });
} }
}, },
[=](finished_atom) { worker_finished(); }}; [=](finished_atom) { worker_finished(); },
};
} }
// signal handling for ctrl+c // signal handling for ctrl+c
......
...@@ -3,7 +3,7 @@ ...@@ -3,7 +3,7 @@
* for both the blocking and the event-based API. * * for both the blocking and the event-based API. *
\******************************************************************************/ \******************************************************************************/
// Manual refs: lines 17-18, 28-56, 58-92, 123-128 (Actor) // Manual refs: lines 17-18, 21-26, 28-56, 58-92, 123-128 (Actor)
#include <iostream> #include <iostream>
......
/******************************************************************************\ /******************************************************************************\
* This example illustrates how to do time-triggered loops in libcaf. * * This example illustrates how to do time-triggered loops in libcaf. *
\ \******************************************************************************/
******************************************************************************/
#include "caf/all.hpp"
#include <algorithm> #include <algorithm>
#include <chrono> #include <chrono>
#include <iostream> #include <iostream>
#include "caf/all.hpp"
// This file is partially included in the manual, do not modify // This file is partially included in the manual, do not modify
// without updating the references in the *.tex files! // without updating the references in the *.tex files!
// Manual references: lines 58-75 (MessagePassing.tex) // Manual references: lines 58-75 (MessagePassing.tex)
......
#include "caf/all.hpp"
#include <iostream> #include <iostream>
#include "caf/all.hpp"
// This file is partially included in the manual, do not modify // This file is partially included in the manual, do not modify
// without updating the references in the *.tex files! // without updating the references in the *.tex files!
// Manual references: lines 15-36 (MessagePassing.tex) // Manual references: lines 15-36 (MessagePassing.tex)
using std::endl;
using namespace caf; using namespace caf;
// using add_atom = atom_constant<atom("add")>; (defined in atom.hpp) // using add_atom = atom_constant<atom("add")>; (defined in atom.hpp)
...@@ -14,7 +14,7 @@ using calc = typed_actor<replies_to<add_atom, int, int>::with<int>>; ...@@ -14,7 +14,7 @@ using calc = typed_actor<replies_to<add_atom, int, int>::with<int>>;
void actor_a(event_based_actor* self, const calc& worker) { void actor_a(event_based_actor* self, const calc& worker) {
self->request(worker, std::chrono::seconds(10), add_atom_v, 1, 2) self->request(worker, std::chrono::seconds(10), add_atom_v, 1, 2)
.then([=](int result) { aout(self) << "1 + 2 = " << result << endl; }); .then([=](int result) { aout(self) << "1 + 2 = " << result << std::endl; });
} }
calc::behavior_type actor_b(calc::pointer self, const calc& worker) { calc::behavior_type actor_b(calc::pointer self, const calc& worker) {
......
...@@ -33,15 +33,15 @@ class fixed_stack : public event_based_actor { ...@@ -33,15 +33,15 @@ class fixed_stack : public event_based_actor {
public: public:
fixed_stack(actor_config& cfg, size_t stack_size) fixed_stack(actor_config& cfg, size_t stack_size)
: event_based_actor(cfg), size_(stack_size) { : event_based_actor(cfg), size_(stack_size) {
full_.assign([=](push_atom, full_.assign( //
int) -> error { return fixed_stack_errc::push_to_full; }, [=](push_atom, int) -> error { return fixed_stack_errc::push_to_full; },
[=](pop_atom) -> int { [=](pop_atom) -> int {
auto result = data_.back(); auto result = data_.back();
data_.pop_back(); data_.pop_back();
become(filled_); become(filled_);
return result; return result;
}); });
filled_.assign( filled_.assign( //
[=](push_atom, int what) { [=](push_atom, int what) {
data_.push_back(what); data_.push_back(what);
if (data_.size() == size_) if (data_.size() == size_)
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
become(empty_); become(empty_);
return result; return result;
}); });
empty_.assign( empty_.assign( //
[=](push_atom, int what) { [=](push_atom, int what) {
data_.push_back(what); data_.push_back(what);
become(filled_); become(filled_);
......
...@@ -157,7 +157,7 @@ behavior running(stateful_actor<state>* self, const actor& calculator) { ...@@ -157,7 +157,7 @@ behavior running(stateful_actor<state>* self, const actor& calculator) {
.then( .then(
[=](int result) { [=](int result) {
const char* op_str; const char* op_str;
if constexpr (std::is_same<add_atom, decltype(x)>::value) if constexpr (std::is_same<add_atom, decltype(op)>::value)
op_str = " + "; op_str = " + ";
else else
op_str = " - "; op_str = " - ";
......
...@@ -6,8 +6,7 @@ ...@@ -6,8 +6,7 @@
* - ./build/bin/group_chat -s -p 4242 * * - ./build/bin/group_chat -s -p 4242 *
* - ./build/bin/group_chat -g remote:chatroom@localhost:4242 -n alice * * - ./build/bin/group_chat -g remote:chatroom@localhost:4242 -n alice *
* - ./build/bin/group_chat -g remote:chatroom@localhost:4242 -n bob * * - ./build/bin/group_chat -g remote:chatroom@localhost:4242 -n bob *
\ \******************************************************************************/
******************************************************************************/
#include <cstdlib> #include <cstdlib>
#include <iostream> #include <iostream>
......
...@@ -152,7 +152,7 @@ auto middleman_actor_impl::make_behavior() -> behavior_type { ...@@ -152,7 +152,7 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
[=](spawn_atom atm, node_id& nid, std::string& str, message& msg, [=](spawn_atom atm, node_id& nid, std::string& str, message& msg,
std::set<std::string>& ifs) -> delegated<strong_actor_ptr> { std::set<std::string>& ifs) -> delegated<strong_actor_ptr> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
uint64_t id = str == "SpawnServ" ? 1u : 2u; static constexpr uint64_t id = 1u; // SpawnServ
delegate(broker_, forward_atom_v, nid, id, delegate(broker_, forward_atom_v, nid, id,
make_message(atm, std::move(str), std::move(msg), make_message(atm, std::move(str), std::move(msg),
std::move(ifs))); std::move(ifs)));
......
...@@ -76,6 +76,10 @@ constexpr uint64_t default_operation_data = make_message_id().integer_value(); ...@@ -76,6 +76,10 @@ constexpr uint64_t default_operation_data = make_message_id().integer_value();
constexpr uint32_t num_remote_nodes = 2; constexpr uint32_t num_remote_nodes = 2;
constexpr uint64_t spawn_serv_id = 2;
constexpr uint64_t config_serv_id = 2;
std::string hexstr(const byte_buffer& buf) { std::string hexstr(const byte_buffer& buf) {
return deep_to_string(meta::hex_formatted(), buf); return deep_to_string(meta::hex_formatted(), buf);
} }
...@@ -270,7 +274,7 @@ public: ...@@ -270,7 +274,7 @@ public:
// whether there is a SpawnServ actor on this node // whether there is a SpawnServ actor on this node
.receive(hdl, basp::message_type::direct_message, .receive(hdl, basp::message_type::direct_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, static_cast<uint64_t>(2), default_operation_data, any_vals, config_serv_id,
std::vector<strong_actor_ptr>{}, std::vector<strong_actor_ptr>{},
make_message(sys_atom_v, get_atom_v, "info")); make_message(sys_atom_v, get_atom_v, "info"));
// test whether basp instance correctly updates the // test whether basp instance correctly updates the
...@@ -582,7 +586,7 @@ CAF_TEST(remote_actor_and_send) { ...@@ -582,7 +586,7 @@ CAF_TEST(remote_actor_and_send) {
invalid_actor_id, this_node()) invalid_actor_id, this_node())
.receive(jupiter().connection, basp::message_type::direct_message, .receive(jupiter().connection, basp::message_type::direct_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, static_cast<uint64_t>(2), default_operation_data, any_vals, config_serv_id,
std::vector<strong_actor_ptr>{}, std::vector<strong_actor_ptr>{},
make_message(sys_atom_v, get_atom_v, "info")) make_message(sys_atom_v, get_atom_v, "info"))
.receive(jupiter().connection, basp::message_type::monitor_message, .receive(jupiter().connection, basp::message_type::monitor_message,
...@@ -678,8 +682,8 @@ CAF_TEST(indirect_connections) { ...@@ -678,8 +682,8 @@ CAF_TEST(indirect_connections) {
// this asks Jupiter if it has a 'SpawnServ' // this asks Jupiter if it has a 'SpawnServ'
mx.receive(mars().connection, basp::message_type::routed_message, mx.receive(mars().connection, basp::message_type::routed_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, static_cast<uint64_t>(2), default_operation_data, any_vals, spawn_serv_id, this_node(),
this_node(), jupiter().id, std::vector<strong_actor_ptr>{}, jupiter().id, std::vector<strong_actor_ptr>{},
make_message(sys_atom_v, get_atom_v, "info")); make_message(sys_atom_v, get_atom_v, "info"));
CAF_MESSAGE("expect announce_proxy message at Mars from Earth to Jupiter"); CAF_MESSAGE("expect announce_proxy message at Mars from Earth to Jupiter");
mx.receive(mars().connection, basp::message_type::monitor_message, no_flags, mx.receive(mars().connection, basp::message_type::monitor_message, no_flags,
...@@ -729,14 +733,14 @@ CAF_TEST(automatic_connection) { ...@@ -729,14 +733,14 @@ CAF_TEST(automatic_connection) {
make_message("hello from jupiter!")) make_message("hello from jupiter!"))
.receive(mars().connection, basp::message_type::routed_message, .receive(mars().connection, basp::message_type::routed_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, static_cast<uint64_t>(2), default_operation_data, any_vals, spawn_serv_id, this_node(),
this_node(), jupiter().id, std::vector<strong_actor_ptr>{}, jupiter().id, std::vector<strong_actor_ptr>{},
make_message(sys_atom_v, get_atom_v, "info")) make_message(sys_atom_v, get_atom_v, "info"))
.receive(mars().connection, basp::message_type::routed_message, .receive(mars().connection, basp::message_type::routed_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
default_operation_data, default_operation_data,
any_vals, // actor ID of an actor spawned by the BASP broker any_vals, // actor ID of an actor spawned by the BASP broker
static_cast<uint64_t>(1), this_node(), jupiter().id, spawn_serv_id, this_node(), jupiter().id,
std::vector<strong_actor_ptr>{}, std::vector<strong_actor_ptr>{},
make_message(get_atom_v, "basp.default-connectivity-tcp")) make_message(get_atom_v, "basp.default-connectivity-tcp"))
.receive(mars().connection, basp::message_type::monitor_message, no_flags, .receive(mars().connection, basp::message_type::monitor_message, no_flags,
......
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