Commit 9e36f2e3 authored by Dominik Charousset's avatar Dominik Charousset

maintenance

parent e13e5d51
...@@ -57,6 +57,8 @@ class receive_policy { ...@@ -57,6 +57,8 @@ class receive_policy {
public: public:
typedef recursive_queue_node* pointer;
enum handle_message_result { enum handle_message_result {
hm_timeout_msg, hm_timeout_msg,
hm_skip_msg, hm_skip_msg,
...@@ -96,7 +98,7 @@ class receive_policy { ...@@ -96,7 +98,7 @@ class receive_policy {
} }
template<class Client, class FunOrBehavior> template<class Client, class FunOrBehavior>
bool invoke(Client* client, recursive_queue_node* node, FunOrBehavior& fun){ bool invoke(Client* client, pointer node, FunOrBehavior& fun){
std::integral_constant<receive_policy_flag, Client::receive_flag> token; std::integral_constant<receive_policy_flag, Client::receive_flag> token;
switch (this->handle_message(client, node, fun, token)) { switch (this->handle_message(client, node, fun, token)) {
case hm_msg_handled: { case hm_msg_handled: {
...@@ -136,7 +138,7 @@ class receive_policy { ...@@ -136,7 +138,7 @@ class receive_policy {
} }
else if (invoke_from_cache(client, fun) == false) { else if (invoke_from_cache(client, fun) == false) {
if (bhvr.timeout().is_zero()) { if (bhvr.timeout().is_zero()) {
recursive_queue_node* e = nullptr; pointer e = nullptr;
while ((e = client->try_receive_node()) != nullptr) { while ((e = client->try_receive_node()) != nullptr) {
CPPA_REQUIRE(e->marked == false); CPPA_REQUIRE(e->marked == false);
if (invoke(client, e, bhvr)) { if (invoke(client, e, bhvr)) {
...@@ -147,7 +149,7 @@ class receive_policy { ...@@ -147,7 +149,7 @@ class receive_policy {
} }
else { else {
auto timeout = client->init_timeout(bhvr.timeout()); auto timeout = client->init_timeout(bhvr.timeout());
recursive_queue_node* e = nullptr; pointer e = nullptr;
while ((e = client->try_receive_node(timeout)) != nullptr) { while ((e = client->try_receive_node(timeout)) != nullptr) {
CPPA_REQUIRE(e->marked == false); CPPA_REQUIRE(e->marked == false);
if (invoke(client, e, bhvr)) { if (invoke(client, e, bhvr)) {
...@@ -177,7 +179,7 @@ class receive_policy { ...@@ -177,7 +179,7 @@ class receive_policy {
} }
template<class Client> template<class Client>
filter_result filter_msg(Client* client, recursive_queue_node* node) { filter_result filter_msg(Client* client, pointer node) {
const any_tuple& msg = node->msg; const any_tuple& msg = node->msg;
bool is_sync_msg = node->seq_id != 0; bool is_sync_msg = node->seq_id != 0;
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr; auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr;
...@@ -214,12 +216,34 @@ class receive_policy { ...@@ -214,12 +216,34 @@ class receive_policy {
return ordinary_message; return ordinary_message;
} }
template<class Client, class FunOrBehavior> static inline bool hm_should_skip(pointer node, nestable) {
return node->marked;
}
template<class Client>
static inline pointer hm_begin(Client* client, pointer node, nestable) {
auto previous_node = client->m_current_node;
client->m_current_node = node;
client->push_timeout();
node->marked = true;
return previous_node;
}
template<class Client>
static inline void hm_cleanup(Client* client, pointer, pointer, nestable) {
client->m_current_node = &(client->m_dummy_node);
}
static inline bool hm_should_skip(pointer, sequential) {
return false;
}
template<class Client, class FunOrBehavior, class Policy>
handle_message_result handle_message(Client* client, handle_message_result handle_message(Client* client,
recursive_queue_node* node, pointer node,
FunOrBehavior& fun, FunOrBehavior& fun,
nestable) { Policy token) {
if (node->marked) { if (hm_should_skip(node, token)) {
return hm_skip_msg; return hm_skip_msg;
} }
switch (this->filter_msg(client, node)) { switch (this->filter_msg(client, node)) {
...@@ -253,7 +277,7 @@ class receive_policy { ...@@ -253,7 +277,7 @@ class receive_policy {
template<class Client, class FunOrBehavior> template<class Client, class FunOrBehavior>
handle_message_result handle_message(Client* client, handle_message_result handle_message(Client* client,
recursive_queue_node* node, pointer node,
FunOrBehavior& fun, FunOrBehavior& fun,
sequential) { sequential) {
CPPA_REQUIRE(node->marked == false); CPPA_REQUIRE(node->marked == false);
......
...@@ -314,37 +314,16 @@ class local_actor : public actor { ...@@ -314,37 +314,16 @@ class local_actor : public actor {
else whom->enqueue(this, std::move(what)); else whom->enqueue(this, std::move(what));
} }
// returns 0 if last_dequeued() is an asynchronous or sync request message,
// a response id generated from the request id otherwise
inline std::uint64_t get_response_id() { inline std::uint64_t get_response_id() {
constexpr std::uint64_t response_mask = 0x8000000000000000; // mask extracts the first bit that is only set on response messages
auto& whom = last_sender(); constexpr std::uint64_t mask = 0x8000000000000000;
if (whom) {
auto seq_id = m_current_node->seq_id; auto seq_id = m_current_node->seq_id;
if (seq_id != 0 && (seq_id & response_mask) == 0) { return (seq_id != 0 && (seq_id & mask) == 0) ? (seq_id | mask) : 0;
return seq_id | response_mask;
}
}
return 0;
} }
inline void reply_message(any_tuple&& what) { void reply_message(any_tuple&& what);
auto& whom = last_sender();
if (whom) {
auto response_id = get_response_id();
if (response_id == 0) {
send_message(whom.get(), std::move(what));
}
else {
if (m_chaining && !m_chained_actor) {
if (whom->chained_sync_enqueue(this,
response_id,
std::move(what))) {
m_chained_actor = whom;
}
}
else whom->sync_enqueue(this, response_id, std::move(what));
}
}
}
inline actor_ptr& chained_actor() { inline actor_ptr& chained_actor() {
return m_chained_actor; return m_chained_actor;
......
...@@ -102,4 +102,24 @@ std::vector<group_ptr> local_actor::joined_groups() { ...@@ -102,4 +102,24 @@ std::vector<group_ptr> local_actor::joined_groups() {
return result; return result;
} }
void local_actor::reply_message(any_tuple&& what) {
auto& whom = last_sender();
if (whom) {
auto response_id = get_response_id();
if (response_id == 0) {
send_message(whom.get(), std::move(what));
}
else {
if (m_chaining && !m_chained_actor) {
if (whom->chained_sync_enqueue(this,
response_id,
std::move(what))) {
m_chained_actor = whom;
}
}
else whom->sync_enqueue(this, response_id, std::move(what));
}
}
}
} // namespace cppa } // namespace cppa
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