Commit efa04d70 authored by Dominik Charousset's avatar Dominik Charousset

Rename {message_id::invalid => invalid_message_id}

parent 5d10f7c7
...@@ -215,7 +215,7 @@ class blocking_actor ...@@ -215,7 +215,7 @@ class blocking_actor
// required by receive() member function family // required by receive() member function family
inline void dequeue(behavior& bhvr) { inline void dequeue(behavior& bhvr) {
dequeue_response(bhvr, message_id::invalid); dequeue_response(bhvr, invalid_message_id);
} }
// implemented by detail::proper_actor // implemented by detail::proper_actor
......
...@@ -76,7 +76,7 @@ class behavior_stack { ...@@ -76,7 +76,7 @@ class behavior_stack {
} }
inline void push_back(behavior&& what, inline void push_back(behavior&& what,
message_id response_id = message_id::invalid) { message_id response_id = invalid_message_id) {
m_elements.emplace_back(std::move(what), response_id); m_elements.emplace_back(std::move(what), response_id);
} }
......
...@@ -296,7 +296,7 @@ class proper_actor<Base, Policies, true> ...@@ -296,7 +296,7 @@ class proper_actor<Base, Policies, true>
auto msg = make_message(timeout_msg{tid}); auto msg = make_message(timeout_msg{tid});
if (d.is_zero()) { if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s // immediately enqueue timeout message if duration == 0s
this->enqueue(this->address(), message_id::invalid, this->enqueue(this->address(), invalid_message_id,
std::move(msg), this->host()); std::move(msg), this->host());
// auto e = this->new_mailbox_element(this, std::move(msg)); // auto e = this->new_mailbox_element(this, std::move(msg));
// this->m_mailbox.enqueue(e); // this->m_mailbox.enqueue(e);
......
...@@ -27,25 +27,32 @@ ...@@ -27,25 +27,32 @@
namespace caf { namespace caf {
struct invalid_message_id { struct invalid_message_id_t {
constexpr invalid_message_id() {} constexpr invalid_message_id_t() {
// nop
}
}; };
constexpr invalid_message_id_t invalid_message_id = invalid_message_id_t{};
/** /**
* Denotes whether a message is asynchronous or synchronous * Denotes whether a message is asynchronous or synchronous
* @note Asynchronous messages always have an invalid message id. * @note Asynchronous messages always have an invalid message id.
*/ */
class message_id : detail::comparable<message_id> { class message_id : detail::comparable<message_id> {
public:
static constexpr uint64_t response_flag_mask = 0x8000000000000000; static constexpr uint64_t response_flag_mask = 0x8000000000000000;
static constexpr uint64_t answered_flag_mask = 0x4000000000000000; static constexpr uint64_t answered_flag_mask = 0x4000000000000000;
static constexpr uint64_t high_prioity_flag_mask = 0x2000000000000000; static constexpr uint64_t high_prioity_flag_mask = 0x2000000000000000;
static constexpr uint64_t request_id_mask = 0x1FFFFFFFFFFFFFFF; static constexpr uint64_t request_id_mask = 0x1FFFFFFFFFFFFFFF;
public: constexpr message_id() : m_value(0) {
// nop
}
constexpr message_id() : m_value(0) {} constexpr message_id(invalid_message_id_t) : m_value(0) {
// nop
}
message_id(message_id&&) = default; message_id(message_id&&) = default;
message_id(const message_id&) = default; message_id(const message_id&) = default;
...@@ -69,14 +76,16 @@ class message_id : detail::comparable<message_id> { ...@@ -69,14 +76,16 @@ class message_id : detail::comparable<message_id> {
return (m_value & high_prioity_flag_mask) != 0; return (m_value & high_prioity_flag_mask) != 0;
} }
inline bool valid() const { return (m_value & request_id_mask) != 0; } inline bool valid() const {
return (m_value & request_id_mask) != 0;
}
inline bool is_request() const { return valid() && !is_response(); } inline bool is_request() const {
return valid() && !is_response();
}
inline message_id response_id() const { inline message_id response_id() const {
// the response to a response is an asynchronous message return message_id{is_request() ? m_value | response_flag_mask : 0};
if (is_response()) return message_id{0};
return message_id{valid() ? m_value | response_flag_mask : 0};
} }
inline message_id request_id() const { inline message_id request_id() const {
...@@ -91,9 +100,13 @@ class message_id : detail::comparable<message_id> { ...@@ -91,9 +100,13 @@ class message_id : detail::comparable<message_id> {
return message_id(m_value & ~high_prioity_flag_mask); return message_id(m_value & ~high_prioity_flag_mask);
} }
inline void mark_as_answered() { m_value |= answered_flag_mask; } inline void mark_as_answered() {
m_value |= answered_flag_mask;
}
inline uint64_t integer_value() const { return m_value; } inline uint64_t integer_value() const {
return m_value;
}
static inline message_id from_integer_value(uint64_t value) { static inline message_id from_integer_value(uint64_t value) {
message_id result; message_id result;
...@@ -101,22 +114,16 @@ class message_id : detail::comparable<message_id> { ...@@ -101,22 +114,16 @@ class message_id : detail::comparable<message_id> {
return result; return result;
} }
static constexpr invalid_message_id invalid = invalid_message_id{};
constexpr message_id(invalid_message_id) : m_value(0) {}
long compare(const message_id& other) const { long compare(const message_id& other) const {
return (m_value == other.m_value) ? return (m_value == other.m_value) ? 0
0 : : (m_value < other.m_value ? -1 : 1);
((m_value < other.m_value) ? -1 : 1);
} }
private: private:
explicit inline message_id(uint64_t value) : m_value(value) {
explicit constexpr message_id(uint64_t value) : m_value(value) {} // nop
}
uint64_t m_value; uint64_t m_value;
}; };
} // namespace caf } // namespace caf
......
...@@ -51,7 +51,7 @@ class single_timeout : public Base { ...@@ -51,7 +51,7 @@ class single_timeout : public Base {
auto msg = make_message(timeout_msg{tid}); auto msg = make_message(timeout_msg{tid});
if (d.is_zero()) { if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s // immediately enqueue timeout message if duration == 0s
this->enqueue(this->address(), message_id::invalid, this->enqueue(this->address(), invalid_message_id,
std::move(msg), this->host()); std::move(msg), this->host());
} else } else
this->delayed_send_tuple(this, d, std::move(msg)); this->delayed_send_tuple(this, d, std::move(msg));
......
...@@ -64,7 +64,7 @@ class abstract_coordinator { ...@@ -64,7 +64,7 @@ class abstract_coordinator {
template <class Duration, class... Data> template <class Duration, class... Data>
void delayed_send(Duration rel_time, actor_addr from, channel to, void delayed_send(Duration rel_time, actor_addr from, channel to,
message_id mid, message data) { message_id mid, message data) {
m_timer->enqueue(invalid_actor_addr, message_id::invalid, m_timer->enqueue(invalid_actor_addr, invalid_message_id,
make_message(atom("_Send"), duration{rel_time}, make_message(atom("_Send"), duration{rel_time},
std::move(from), std::move(to), mid, std::move(from), std::move(to), mid,
std::move(data)), std::move(data)),
......
...@@ -37,7 +37,7 @@ namespace caf { ...@@ -37,7 +37,7 @@ namespace caf {
*/ */
inline void send_tuple_as(const actor& from, const channel& to, message msg) { inline void send_tuple_as(const actor& from, const channel& to, message msg) {
if (to) { if (to) {
to->enqueue(from.address(), message_id::invalid, std::move(msg), nullptr); to->enqueue(from.address(), invalid_message_id, std::move(msg), nullptr);
} }
} }
......
...@@ -129,7 +129,7 @@ bool abstract_actor::establish_link_impl(const actor_addr& other) { ...@@ -129,7 +129,7 @@ bool abstract_actor::establish_link_impl(const actor_addr& other) {
auto ptr = actor_cast<abstract_actor_ptr>(other); auto ptr = actor_cast<abstract_actor_ptr>(other);
// send exit message if already exited // send exit message if already exited
if (exited()) { if (exited()) {
ptr->enqueue(address(), message_id::invalid, ptr->enqueue(address(), invalid_message_id,
make_message(exit_msg{address(), exit_reason()}), m_host); make_message(exit_msg{address(), exit_reason()}), m_host);
} else if (ptr->establish_backlink(address())) { } else if (ptr->establish_backlink(address())) {
// add link if not already linked to other // add link if not already linked to other
...@@ -160,7 +160,7 @@ bool abstract_actor::establish_backlink_impl(const actor_addr& other) { ...@@ -160,7 +160,7 @@ bool abstract_actor::establish_backlink_impl(const actor_addr& other) {
// send exit message without lock // send exit message without lock
if (reason != exit_reason::not_exited) { if (reason != exit_reason::not_exited) {
auto ptr = actor_cast<abstract_actor_ptr>(other); auto ptr = actor_cast<abstract_actor_ptr>(other);
ptr->enqueue(address(), message_id::invalid, ptr->enqueue(address(), invalid_message_id,
make_message(exit_msg{address(), exit_reason()}), m_host); make_message(exit_msg{address(), exit_reason()}), m_host);
} }
return false; return false;
......
...@@ -56,7 +56,7 @@ class local_group : public abstract_group { ...@@ -56,7 +56,7 @@ class local_group : public abstract_group {
<< CAF_TARG(msg, to_string)); << CAF_TARG(msg, to_string));
shared_guard guard(m_mtx); shared_guard guard(m_mtx);
for (auto& s : m_subscribers) { for (auto& s : m_subscribers) {
actor_cast<abstract_actor_ptr>(s)->enqueue(sender, message_id::invalid, actor_cast<abstract_actor_ptr>(s)->enqueue(sender, invalid_message_id,
msg, host); msg, host);
} }
} }
...@@ -66,7 +66,7 @@ class local_group : public abstract_group { ...@@ -66,7 +66,7 @@ class local_group : public abstract_group {
CAF_LOG_TRACE(CAF_TARG(sender, to_string) << ", " CAF_LOG_TRACE(CAF_TARG(sender, to_string) << ", "
<< CAF_TARG(msg, to_string)); << CAF_TARG(msg, to_string));
send_all_subscribers(sender, msg, host); send_all_subscribers(sender, msg, host);
m_broker->enqueue(sender, message_id::invalid, msg, host); m_broker->enqueue(sender, invalid_message_id, msg, host);
} }
std::pair<bool, size_t> add_subscriber(const actor_addr& who) { std::pair<bool, size_t> add_subscriber(const actor_addr& who) {
...@@ -176,7 +176,7 @@ class local_broker : public event_based_actor { ...@@ -176,7 +176,7 @@ class local_broker : public event_based_actor {
<< " acquaintances; " << CAF_TSARG(sender) << ", " << " acquaintances; " << CAF_TSARG(sender) << ", "
<< CAF_TSARG(what)); << CAF_TSARG(what));
for (auto& acquaintance : m_acquaintances) { for (auto& acquaintance : m_acquaintances) {
acquaintance->enqueue(sender, message_id::invalid, what, host()); acquaintance->enqueue(sender, invalid_message_id, what, host());
} }
} }
......
...@@ -119,7 +119,7 @@ void local_actor::forward_message(const actor& dest, message_priority prio) { ...@@ -119,7 +119,7 @@ void local_actor::forward_message(const actor& dest, message_priority prio) {
: m_current_node->mid.with_normal_priority(); : m_current_node->mid.with_normal_priority();
dest->enqueue(m_current_node->sender, id, m_current_node->msg, host()); dest->enqueue(m_current_node->sender, id, m_current_node->msg, host());
// treat this message as asynchronous message from now on // treat this message as asynchronous message from now on
m_current_node->mid = message_id::invalid; m_current_node->mid = invalid_message_id;
} }
void local_actor::send_tuple(message_priority prio, const channel& dest, void local_actor::send_tuple(message_priority prio, const channel& dest,
......
...@@ -57,7 +57,7 @@ void broker::servant::disconnect(bool invoke_disconnect_message) { ...@@ -57,7 +57,7 @@ void broker::servant::disconnect(bool invoke_disconnect_message) {
remove_from_broker(); remove_from_broker();
if (invoke_disconnect_message) { if (invoke_disconnect_message) {
auto msg = disconnect_message(); auto msg = disconnect_message();
m_broker->invoke_message(m_broker->address(),message_id::invalid, msg); m_broker->invoke_message(m_broker->address(),invalid_message_id, msg);
} }
} }
} }
...@@ -87,7 +87,7 @@ void broker::scribe::consume(const void*, size_t num_bytes) { ...@@ -87,7 +87,7 @@ void broker::scribe::consume(const void*, size_t num_bytes) {
buf.resize(num_bytes); // make sure size is correct buf.resize(num_bytes); // make sure size is correct
read_msg().buf.swap(buf); // swap into message read_msg().buf.swap(buf); // swap into message
m_broker->invoke_message(invalid_actor_addr, // call client m_broker->invoke_message(invalid_actor_addr, // call client
message_id::invalid, m_read_msg); invalid_message_id, m_read_msg);
read_msg().buf.swap(buf); // swap buffer back to stream read_msg().buf.swap(buf); // swap buffer back to stream
flush(); // implicit flush of wr_buf() flush(); // implicit flush of wr_buf()
} }
...@@ -290,7 +290,7 @@ void broker::launch(bool is_hidden, execution_unit*) { ...@@ -290,7 +290,7 @@ void broker::launch(bool is_hidden, execution_unit*) {
} }
} }
); );
enqueue(invalid_actor_addr, message_id::invalid, enqueue(invalid_actor_addr, invalid_message_id,
make_message(atom("INITMSG")), nullptr); make_message(atom("INITMSG")), nullptr);
} }
......
...@@ -723,7 +723,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self, ...@@ -723,7 +723,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
accept_msg().handle = dm.add_tcp_scribe(parent(), accept_msg().handle = dm.add_tcp_scribe(parent(),
std::move(m_acceptor.accepted_socket())); std::move(m_acceptor.accepted_socket()));
parent()->invoke_message(invalid_actor_addr, parent()->invoke_message(invalid_actor_addr,
message_id::invalid, invalid_message_id,
m_accept_msg); m_accept_msg);
} }
void stop_reading() override { void stop_reading() override {
......
...@@ -62,7 +62,7 @@ void remote_actor_proxy::forward_msg(const actor_addr& sender, message_id mid, ...@@ -62,7 +62,7 @@ void remote_actor_proxy::forward_msg(const actor_addr& sender, message_id mid,
<< CAF_MARG(mid, integer_value) << ", " << CAF_MARG(mid, integer_value) << ", "
<< CAF_TSARG(msg)); << CAF_TSARG(msg));
m_parent->enqueue( m_parent->enqueue(
invalid_actor_addr, message_id::invalid, invalid_actor_addr, invalid_message_id,
make_message(atom("_Dispatch"), sender, address(), mid, std::move(msg)), make_message(atom("_Dispatch"), sender, address(), mid, std::move(msg)),
nullptr); nullptr);
} }
...@@ -79,7 +79,7 @@ bool remote_actor_proxy::link_impl(linking_operation op, ...@@ -79,7 +79,7 @@ bool remote_actor_proxy::link_impl(linking_operation op,
if (establish_link_impl(other)) { if (establish_link_impl(other)) {
// causes remote actor to link to (proxy of) other // causes remote actor to link to (proxy of) other
// receiving peer will call: this->local_link_to(other) // receiving peer will call: this->local_link_to(other)
forward_msg(address(), message_id::invalid, forward_msg(address(), invalid_message_id,
make_message(atom("_Link"), other)); make_message(atom("_Link"), other));
return true; return true;
} }
...@@ -87,7 +87,7 @@ bool remote_actor_proxy::link_impl(linking_operation op, ...@@ -87,7 +87,7 @@ bool remote_actor_proxy::link_impl(linking_operation op,
case remove_link_op: case remove_link_op:
if (remove_link_impl(other)) { if (remove_link_impl(other)) {
// causes remote actor to unlink from (proxy of) other // causes remote actor to unlink from (proxy of) other
forward_msg(address(), message_id::invalid, forward_msg(address(), invalid_message_id,
make_message(atom("_Unlink"), other)); make_message(atom("_Unlink"), other));
return true; return true;
} }
...@@ -95,7 +95,7 @@ bool remote_actor_proxy::link_impl(linking_operation op, ...@@ -95,7 +95,7 @@ bool remote_actor_proxy::link_impl(linking_operation op,
case establish_backlink_op: case establish_backlink_op:
if (establish_backlink_impl(other)) { if (establish_backlink_impl(other)) {
// causes remote actor to unlink from (proxy of) other // causes remote actor to unlink from (proxy of) other
forward_msg(address(), message_id::invalid, forward_msg(address(), invalid_message_id,
make_message(atom("_Link"), other)); make_message(atom("_Link"), other));
return true; return true;
} }
...@@ -103,7 +103,7 @@ bool remote_actor_proxy::link_impl(linking_operation op, ...@@ -103,7 +103,7 @@ bool remote_actor_proxy::link_impl(linking_operation op,
case remove_backlink_op: case remove_backlink_op:
if (remove_backlink_impl(other)) { if (remove_backlink_impl(other)) {
// causes remote actor to unlink from (proxy of) other // causes remote actor to unlink from (proxy of) other
forward_msg(address(), message_id::invalid, forward_msg(address(), invalid_message_id,
make_message(atom("_Unlink"), other)); make_message(atom("_Unlink"), other));
return true; return true;
} }
......
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