Commit 6de99fdc authored by Dominik Charousset's avatar Dominik Charousset

new message types for brokers

this patch adds new message types for brokers replacing the old
atom-prefixed messages for less verbose broker implementation
and to pave the path for future type-safe brokers
parent c9092b45
...@@ -61,11 +61,13 @@ namespace cppa { namespace detail { ...@@ -61,11 +61,13 @@ namespace cppa { namespace detail {
// ordered according to demangled type name (see uniform_type_info_map.cpp) // ordered according to demangled type name (see uniform_type_info_map.cpp)
using mapped_type_list = util::type_list< using mapped_type_list = util::type_list<
bool, bool,
acceptor_closed_msg,
actor, actor,
actor_addr, actor_addr,
any_tuple, any_tuple,
atom_value, atom_value,
channel, channel,
connection_closed_msg,
down_msg, down_msg,
exit_msg, exit_msg,
group, group,
...@@ -74,6 +76,8 @@ using mapped_type_list = util::type_list< ...@@ -74,6 +76,8 @@ using mapped_type_list = util::type_list<
io::accept_handle, io::accept_handle,
io::connection_handle, io::connection_handle,
message_header, message_header,
new_connection_msg,
new_data_msg,
sync_exited_msg, sync_exited_msg,
sync_timeout_msg, sync_timeout_msg,
timeout_msg, timeout_msg,
......
...@@ -36,6 +36,11 @@ ...@@ -36,6 +36,11 @@
#include "cppa/group.hpp" #include "cppa/group.hpp"
#include "cppa/actor_addr.hpp" #include "cppa/actor_addr.hpp"
#include "cppa/util/buffer.hpp"
#include "cppa/io/accept_handle.hpp"
#include "cppa/io/connection_handle.hpp"
namespace cppa { namespace cppa {
/** /**
...@@ -112,6 +117,54 @@ struct timeout_msg { ...@@ -112,6 +117,54 @@ struct timeout_msg {
std::uint32_t timeout_id; std::uint32_t timeout_id;
}; };
/**
* @brief Signalizes a newly accepted connection from a {@link broker}.
*/
struct new_connection_msg {
/**
* @brief The handle that accepted the new connection.
*/
io::accept_handle source;
/**
* @brief The handle for the new connection.
*/
io::connection_handle handle;
};
/**
* @brief Signalizes newly arrived data for a {@link broker}.
*/
struct new_data_msg {
/**
* @brief Handle to the related connection
*/
io::connection_handle handle;
/**
* @brief Buffer containing the received data.
*/
util::buffer buf;
};
/**
* @brief Signalizes that a {@link broker} connection has been closed.
*/
struct connection_closed_msg {
/**
* @brief Handle to the closed connection.
*/
io::connection_handle handle;
};
/**
* @brief Signalizes that a {@link broker} acceptor has been closed.
*/
struct acceptor_closed_msg {
/**
* @brief Handle to the closed connection.
*/
io::accept_handle handle;
};
} // namespace cppa } // namespace cppa
#endif // CPPA_SYSTEM_MESSAGES_HPP #endif // CPPA_SYSTEM_MESSAGES_HPP
...@@ -169,9 +169,10 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> { ...@@ -169,9 +169,10 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
scribe(broker_ptr parent, input_stream_ptr in, output_stream_ptr out) scribe(broker_ptr parent, input_stream_ptr in, output_stream_ptr out)
: super{get_middleman(), out, move(parent), in->read_handle(), out->write_handle()} : super{get_middleman(), out, move(parent), in->read_handle(), out->write_handle()}
, m_is_continue_reading{false}, m_dirty{false} , m_is_continue_reading{false}, m_dirty{false}
, m_policy{broker::at_least}, m_policy_buffer_size{0}, m_in{in} , m_policy{broker::at_least}, m_policy_buffer_size{0}, m_in{in} {
, m_read_msg{atom("IO_read"), connection_handle::from_int(in->read_handle())} { auto& ndm = get_ref<0>(m_read_msg);
get_ref<2>(m_read_msg).final_size(default_max_buffer_size); ndm.handle = connection_handle::from_int(in->read_handle());
ndm.buf.final_size(default_max_buffer_size);
} }
void receive_policy(broker::policy_flag policy, size_t buffer_size) { void receive_policy(broker::policy_flag policy, size_t buffer_size) {
...@@ -192,9 +193,11 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> { ...@@ -192,9 +193,11 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
for (;;) { for (;;) {
// stop reading if actor finished execution // stop reading if actor finished execution
if (m_broker->exit_reason() != exit_reason::not_exited) { if (m_broker->exit_reason() != exit_reason::not_exited) {
CPPA_LOG_DEBUG("broker already done; exit reason: "
<< m_broker->exit_reason());
return continue_reading_result::closed; return continue_reading_result::closed;
} }
auto& buf = get_ref<2>(m_read_msg); auto& buf = get_ref<0>(m_read_msg).buf;
if (m_dirty) { if (m_dirty) {
m_dirty = false; m_dirty = false;
if (m_policy == broker::at_most || m_policy == broker::exactly) { if (m_policy == broker::at_most || m_policy == broker::exactly) {
...@@ -220,7 +223,7 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> { ...@@ -220,7 +223,7 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
CPPA_LOG_DEBUG("invoke io actor"); CPPA_LOG_DEBUG("invoke io actor");
m_broker->invoke_message({invalid_actor_addr, nullptr}, m_read_msg); m_broker->invoke_message({invalid_actor_addr, nullptr}, m_read_msg);
CPPA_LOG_INFO_IF(!m_read_msg.vals()->unique(), "detached buffer"); CPPA_LOG_INFO_IF(!m_read_msg.vals()->unique(), "detached buffer");
get_ref<2>(m_read_msg).clear(); get_ref<0>(m_read_msg).buf.clear();
} }
} }
} }
...@@ -232,8 +235,8 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> { ...@@ -232,8 +235,8 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
protected: protected:
any_tuple disconnect_message() override { any_tuple disconnect_message() override {
return make_any_tuple(atom("IO_closed"), auto hdl = connection_handle::from_int(m_in->read_handle());
connection_handle::from_int(m_in->read_handle())); return make_any_tuple(connection_closed_msg{hdl});
} }
private: private:
...@@ -243,7 +246,7 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> { ...@@ -243,7 +246,7 @@ class broker::scribe : public extend<broker::servant>::with<buffered_writing> {
broker::policy_flag m_policy; broker::policy_flag m_policy;
size_t m_policy_buffer_size; size_t m_policy_buffer_size;
input_stream_ptr m_in; input_stream_ptr m_in;
cow_tuple<atom_value, connection_handle, util::buffer> m_read_msg; cow_tuple<new_data_msg> m_read_msg;
}; };
...@@ -259,10 +262,9 @@ class broker::doorman : public broker::servant { ...@@ -259,10 +262,9 @@ class broker::doorman : public broker::servant {
~doorman(); ~doorman();
doorman(broker_ptr parent, acceptor_uptr ptr) doorman(broker_ptr parent, acceptor_uptr ptr)
: super{move(parent), ptr->file_handle()} : super{move(parent), ptr->file_handle()} {
, m_accept_msg{atom("IO_accept"), get_ref<0>(m_accept_msg).source = accept_handle::from_int(ptr->file_handle());
accept_handle::from_int(ptr->file_handle())} { m_ptr.swap(ptr);
m_ptr.reset(ptr.release());
} }
continue_reading_result continue_reading() override { continue_reading_result continue_reading() override {
...@@ -278,7 +280,7 @@ class broker::doorman : public broker::servant { ...@@ -278,7 +280,7 @@ class broker::doorman : public broker::servant {
if (opt) { if (opt) {
using namespace std; using namespace std;
auto& p = *opt; auto& p = *opt;
get_ref<2>(m_accept_msg) = m_broker->add_scribe(move(p.first), get_ref<0>(m_accept_msg).handle = m_broker->add_scribe(move(p.first),
move(p.second)); move(p.second));
m_broker->invoke_message({invalid_actor_addr, nullptr}, m_accept_msg); m_broker->invoke_message({invalid_actor_addr, nullptr}, m_accept_msg);
} }
...@@ -289,14 +291,14 @@ class broker::doorman : public broker::servant { ...@@ -289,14 +291,14 @@ class broker::doorman : public broker::servant {
protected: protected:
any_tuple disconnect_message() override { any_tuple disconnect_message() override {
return make_any_tuple(atom("IO_closed"), auto hdl = accept_handle::from_int(m_ptr->file_handle());
accept_handle::from_int(m_ptr->file_handle())); return make_any_tuple(acceptor_closed_msg{hdl});
} }
private: private:
acceptor_uptr m_ptr; acceptor_uptr m_ptr;
cow_tuple<atom_value, accept_handle, connection_handle> m_accept_msg; cow_tuple<new_connection_msg> m_accept_msg;
}; };
......
...@@ -62,11 +62,13 @@ namespace cppa { namespace detail { ...@@ -62,11 +62,13 @@ namespace cppa { namespace detail {
// WARNING: this map is sorted, insert new elements *in sorted order* as well! // WARNING: this map is sorted, insert new elements *in sorted order* as well!
/* extern */ const char* mapped_type_names[][2] = { /* extern */ const char* mapped_type_names[][2] = {
{ "bool", "bool" }, { "bool", "bool" },
{ "cppa::acceptor_closed_msg", "@acceptor_closed" },
{ "cppa::actor", "@actor" }, { "cppa::actor", "@actor" },
{ "cppa::actor_addr", "@addr" }, { "cppa::actor_addr", "@addr" },
{ "cppa::any_tuple", "@tuple" }, { "cppa::any_tuple", "@tuple" },
{ "cppa::atom_value", "@atom" }, { "cppa::atom_value", "@atom" },
{ "cppa::channel", "@channel" }, { "cppa::channel", "@channel" },
{ "cppa::connection_closed_msg", "@conn_closed" },
{ "cppa::down_msg", "@down" }, { "cppa::down_msg", "@down" },
{ "cppa::exit_msg", "@exit" }, { "cppa::exit_msg", "@exit" },
{ "cppa::group", "@group" }, { "cppa::group", "@group" },
...@@ -75,6 +77,8 @@ namespace cppa { namespace detail { ...@@ -75,6 +77,8 @@ namespace cppa { namespace detail {
{ "cppa::io::accept_handle", "@ac_hdl" }, { "cppa::io::accept_handle", "@ac_hdl" },
{ "cppa::io::connection_handle", "@cn_hdl" }, { "cppa::io::connection_handle", "@cn_hdl" },
{ "cppa::message_header", "@header" }, { "cppa::message_header", "@header" },
{ "cppa::new_connection_msg", "@new_conn" },
{ "cppa::new_data_msg", "@new_data" },
{ "cppa::sync_exited_msg", "@sync_exited" }, { "cppa::sync_exited_msg", "@sync_exited" },
{ "cppa::sync_timeout_msg", "@sync_timeout" }, { "cppa::sync_timeout_msg", "@sync_timeout" },
{ "cppa::timeout_msg", "@timeout" }, { "cppa::timeout_msg", "@timeout" },
...@@ -435,6 +439,40 @@ inline void serialize_impl(const sync_timeout_msg&, serializer*) { } ...@@ -435,6 +439,40 @@ inline void serialize_impl(const sync_timeout_msg&, serializer*) { }
inline void deserialize_impl(const sync_timeout_msg&, deserializer*) { } inline void deserialize_impl(const sync_timeout_msg&, deserializer*) { }
inline void serialize_impl(const new_connection_msg& ncm, serializer* sink) {
serialize_impl(ncm.source, sink);
serialize_impl(ncm.handle, sink);
}
inline void deserialize_impl(new_connection_msg& ncm, deserializer* source) {
deserialize_impl(ncm.source, source);
deserialize_impl(ncm.handle, source);
}
// serialize_impl + deserialize_impl for new_data_msg depend on
// buffer_type_info_impl and are thus implemented below that class
void serialize_impl(const new_data_msg& ndm, serializer* sink);
void deserialize_impl(new_data_msg& ndm, deserializer* source);
template<typename T>
typename std::enable_if<
std::is_same<T, connection_closed_msg>::value
|| std::is_same<T, acceptor_closed_msg>::value
>::type
serialize_impl(const T& cm, serializer* sink) {
serialize_impl(cm.handle, sink);
}
// exit_msg & down_msg have the same fields
template<typename T>
typename std::enable_if<
std::is_same<T, connection_closed_msg>::value
|| std::is_same<T, acceptor_closed_msg>::value
>::type
deserialize_impl(T& cm, deserializer* source) {
deserialize_impl(cm.handle, source);
}
bool types_equal(const std::type_info* lhs, const std::type_info* rhs) { bool types_equal(const std::type_info* lhs, const std::type_info* rhs) {
// in some cases (when dealing with dynamic libraries), // in some cases (when dealing with dynamic libraries),
// address can be different although types are equal // address can be different although types are equal
...@@ -738,6 +776,18 @@ class default_meta_tuple : public uniform_type_info { ...@@ -738,6 +776,18 @@ class default_meta_tuple : public uniform_type_info {
}; };
void serialize_impl(const new_data_msg& ndm, serializer* sink) {
buffer_type_info_impl bti;
serialize_impl(ndm.handle, sink);
bti.serialize(&(ndm.buf), sink);
}
void deserialize_impl(new_data_msg& ndm, deserializer* source) {
buffer_type_info_impl bti;
deserialize_impl(ndm.handle, source);
bti.deserialize(&(ndm.buf), source);
}
template<typename T> template<typename T>
void push_native_type(abstract_int_tinfo* m [][2]) { void push_native_type(abstract_int_tinfo* m [][2]) {
m[sizeof(T)][std::is_signed<T>::value ? 1 : 0]->add_native_type(typeid(T)); m[sizeof(T)][std::is_signed<T>::value ? 1 : 0]->add_native_type(typeid(T));
...@@ -793,12 +843,14 @@ class utim_impl : public uniform_type_info_map { ...@@ -793,12 +843,14 @@ class utim_impl : public uniform_type_info_map {
auto i = m_builtin_types.begin(); auto i = m_builtin_types.begin();
*i++ = &m_type_unit; // @0 *i++ = &m_type_unit; // @0
*i++ = &m_ac_hdl; // @ac_hdl *i++ = &m_ac_hdl; // @ac_hdl
*i++ = &m_acceptor_closed_msg; // @acceptor_closed
*i++ = &m_type_actor; // @actor *i++ = &m_type_actor; // @actor
*i++ = &m_type_actor_addr; // @actor_addr *i++ = &m_type_actor_addr; // @actor_addr
*i++ = &m_type_atom; // @atom *i++ = &m_type_atom; // @atom
*i++ = &m_type_buffer; // @buffer *i++ = &m_type_buffer; // @buffer
*i++ = &m_type_channel; // @channel *i++ = &m_type_channel; // @channel
*i++ = &m_cn_hdl; // @cn_hdl *i++ = &m_cn_hdl; // @cn_hdl
*i++ = &m_connection_closed_msg; // @conn_closed
*i++ = &m_type_down_msg; // @down *i++ = &m_type_down_msg; // @down
*i++ = &m_type_duration; // @duration *i++ = &m_type_duration; // @duration
*i++ = &m_type_exit_msg; // @exit *i++ = &m_type_exit_msg; // @exit
...@@ -810,6 +862,8 @@ class utim_impl : public uniform_type_info_map { ...@@ -810,6 +862,8 @@ class utim_impl : public uniform_type_info_map {
*i++ = &m_type_i64; // @i64 *i++ = &m_type_i64; // @i64
*i++ = &m_type_i8; // @i8 *i++ = &m_type_i8; // @i8
*i++ = &m_type_long_double; // @ldouble *i++ = &m_type_long_double; // @ldouble
*i++ = &m_new_connection_msg; // @new_conn
*i++ = &m_new_data_msg; // @new_data
*i++ = &m_type_proc; // @proc *i++ = &m_type_proc; // @proc
*i++ = &m_type_str; // @str *i++ = &m_type_str; // @str
*i++ = &m_type_strmap; // @strmap *i++ = &m_type_strmap; // @strmap
...@@ -956,15 +1010,19 @@ class utim_impl : public uniform_type_info_map { ...@@ -956,15 +1010,19 @@ class utim_impl : public uniform_type_info_map {
int_tinfo<std::uint8_t> m_type_u8; int_tinfo<std::uint8_t> m_type_u8;
int_tinfo<std::int16_t> m_type_i16; int_tinfo<std::int16_t> m_type_i16;
// 30-34 // 30-38
int_tinfo<std::uint16_t> m_type_u16; int_tinfo<std::uint16_t> m_type_u16;
int_tinfo<std::int32_t> m_type_i32; int_tinfo<std::int32_t> m_type_i32;
int_tinfo<std::uint32_t> m_type_u32; int_tinfo<std::uint32_t> m_type_u32;
int_tinfo<std::int64_t> m_type_i64; int_tinfo<std::int64_t> m_type_i64;
int_tinfo<std::uint64_t> m_type_u64; int_tinfo<std::uint64_t> m_type_u64;
uti_impl<new_connection_msg> m_new_connection_msg;
uti_impl<new_data_msg> m_new_data_msg;
uti_impl<connection_closed_msg> m_connection_closed_msg;
uti_impl<acceptor_closed_msg> m_acceptor_closed_msg;
// both containers are sorted by uniform name // both containers are sorted by uniform name
std::array<pointer, 35> m_builtin_types; std::array<pointer, 39> m_builtin_types;
std::vector<uniform_type_info*> m_user_types; std::vector<uniform_type_info*> m_user_types;
mutable util::shared_spinlock m_lock; mutable util::shared_spinlock m_lock;
......
...@@ -96,15 +96,15 @@ void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) { ...@@ -96,15 +96,15 @@ void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) {
self->write(hdl, sizeof(value), &value); self->write(hdl, sizeof(value), &value);
}; };
self->become ( self->become (
on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) { [=](const connection_closed_msg&) {
CPPA_PRINT("received IO_closed"); CPPA_PRINT("received connection_closed_msg");
self->quit(); self->quit();
}, },
on(atom("IO_read"), arg_match) >> [=](io::connection_handle, const util::buffer& buf) { [=](const new_data_msg& msg) {
atom_value type; atom_value type;
int value; int value;
memcpy(&type, buf.data(), sizeof(atom_value)); memcpy(&type, msg.buf.data(), sizeof(atom_value));
memcpy(&value, buf.offset_data(sizeof(atom_value)), sizeof(int)); memcpy(&value, msg.buf.offset_data(sizeof(atom_value)), sizeof(int));
self->send(buddy, type, value); self->send(buddy, type, value);
}, },
on(atom("ping"), arg_match) >> [=](int value) { on(atom("ping"), arg_match) >> [=](int value) {
...@@ -113,7 +113,7 @@ void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) { ...@@ -113,7 +113,7 @@ void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) {
on(atom("pong"), arg_match) >> [=](int value) { on(atom("pong"), arg_match) >> [=](int value) {
write(atom("pong"), value); write(atom("pong"), value);
}, },
on_arg_match >> [=](const down_msg& dm) { [=](const down_msg& dm) {
if (dm.source == buddy) self->quit(dm.reason); if (dm.source == buddy) self->quit(dm.reason);
}, },
others() >> CPPA_UNEXPECTED_MSG_CB(self) others() >> CPPA_UNEXPECTED_MSG_CB(self)
...@@ -123,10 +123,10 @@ void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) { ...@@ -123,10 +123,10 @@ void peer(io::broker* self, io::connection_handle hdl, const actor& buddy) {
void peer_acceptor(io::broker* self, const actor& buddy) { void peer_acceptor(io::broker* self, const actor& buddy) {
CPPA_CHECKPOINT(); CPPA_CHECKPOINT();
self->become ( self->become (
on(atom("IO_accept"), arg_match) >> [=](io::accept_handle, io::connection_handle hdl) { [=](const new_connection_msg& msg) {
CPPA_CHECKPOINT(); CPPA_CHECKPOINT();
CPPA_PRINT("received IO_accept"); CPPA_PRINT("received new_connection_msg");
self->fork(peer, hdl, buddy); self->fork(peer, msg.handle, buddy);
self->quit(); self->quit();
}, },
others() >> CPPA_UNEXPECTED_MSG_CB(self) others() >> CPPA_UNEXPECTED_MSG_CB(self)
......
...@@ -72,6 +72,7 @@ int main() { ...@@ -72,6 +72,7 @@ int main() {
// these types (and only those) are present if // these types (and only those) are present if
// the uniform_type_info implementation is correct // the uniform_type_info implementation is correct
std::set<std::string> expected = { std::set<std::string> expected = {
// basic types
"bool", "bool",
"$::foo", // <anonymous namespace>::foo "$::foo", // <anonymous namespace>::foo
"@i8", "@i16", "@i32", "@i64", // signed integer names "@i8", "@i16", "@i32", "@i64", // signed integer names
...@@ -99,6 +100,10 @@ int main() { ...@@ -99,6 +100,10 @@ int main() {
"@timeout", // timeout_msg "@timeout", // timeout_msg
"@sync_exited", // sync_exited_msg "@sync_exited", // sync_exited_msg
"@sync_timeout", // sync_timeout_msg "@sync_timeout", // sync_timeout_msg
"@acceptor_closed", // acceptor_closed_msg
"@conn_closed", // connection_closed_msg
"@new_conn", // new_connection_msg
"@new_data", // new_data_msg
// default announced cppa tuples // default announced cppa tuples
"@<>+@atom", // {atom_value} "@<>+@atom", // {atom_value}
"@<>+@atom+@actor", // {atom_value, actor_ptr} "@<>+@atom+@actor", // {atom_value, actor_ptr}
......
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