Commit e35c957a authored by Dominik Charousset's avatar Dominik Charousset

sync send & receive

parent efa308f7
...@@ -263,4 +263,3 @@ cppa/detail/event_based_actor_factory.hpp ...@@ -263,4 +263,3 @@ cppa/detail/event_based_actor_factory.hpp
cppa/factory.hpp cppa/factory.hpp
src/factory.cpp src/factory.cpp
cppa/message_id.hpp cppa/message_id.hpp
cppa/message_future.hpp
...@@ -72,7 +72,7 @@ class actor : public channel { ...@@ -72,7 +72,7 @@ class actor : public channel {
/** /**
* @brief Enqueues @p msg as a synchronous message to this actor's mailbox. * @brief Enqueues @p msg as a synchronous message to this actor's mailbox.
* @pre <tt>id.is_async() == false</tt> * @pre <tt>id.valid()</tt>
*/ */
virtual void sync_enqueue(actor* sender, message_id_t id, any_tuple msg) = 0; virtual void sync_enqueue(actor* sender, message_id_t id, any_tuple msg) = 0;
......
...@@ -471,8 +471,9 @@ template<typename... Args> ...@@ -471,8 +471,9 @@ template<typename... Args>
inline unspecified_handle sync_send(const actor_ptr& whom, Args&&... what) { inline unspecified_handle sync_send(const actor_ptr& whom, Args&&... what) {
static_assert(sizeof...(Args) > 0, "no message to send"); static_assert(sizeof...(Args) > 0, "no message to send");
if (whom) { if (whom) {
self->send_sync_message(whom.get(), return self->send_sync_message(
make_any_tuple(std::forward<Args>(what)...)); whom.get(),
make_any_tuple(std::forward<Args>(what)...));
} }
else throw std::invalid_argument("whom == nullptr"); else throw std::invalid_argument("whom == nullptr");
} }
......
...@@ -71,13 +71,13 @@ class receive_policy { ...@@ -71,13 +71,13 @@ class receive_policy {
template<class Client, class Fun> template<class Client, class Fun>
bool invoke_from_cache(Client* client, bool invoke_from_cache(Client* client,
Fun& fun, Fun& fun,
message_id_t active_request = message_id_t()) { message_id_t awaited_response = message_id_t()) {
std::integral_constant<receive_policy_flag,Client::receive_flag> policy; std::integral_constant<receive_policy_flag,Client::receive_flag> policy;
auto i = m_cache.begin(); auto i = m_cache.begin();
auto e = m_cache.end(); auto e = m_cache.end();
while (i != e) { while (i != e) {
switch (this->handle_message(client, i->get(), fun, switch (this->handle_message(client, i->get(), fun,
active_request, policy)) { awaited_response, policy)) {
case hm_msg_handled: { case hm_msg_handled: {
client->release_node(i->release()); client->release_node(i->release());
m_cache.erase(i); m_cache.erase(i);
...@@ -105,10 +105,10 @@ class receive_policy { ...@@ -105,10 +105,10 @@ class receive_policy {
bool invoke(Client* client, bool invoke(Client* client,
pointer node, pointer node,
Fun& fun, Fun& fun,
message_id_t active_request = message_id_t()) { message_id_t awaited_response = message_id_t()) {
std::integral_constant<receive_policy_flag,Client::receive_flag> policy; std::integral_constant<receive_policy_flag,Client::receive_flag> policy;
switch (this->handle_message(client, node, fun, switch (this->handle_message(client, node, fun,
active_request, policy)) { awaited_response, policy)) {
case hm_msg_handled: { case hm_msg_handled: {
client->release_node(node); client->release_node(node);
return true; return true;
...@@ -170,15 +170,15 @@ class receive_policy { ...@@ -170,15 +170,15 @@ class receive_policy {
} }
template<class Client> template<class Client>
void receive(Client* client, behavior& bhvr, message_id_t active_request) { void receive(Client* client, behavior& bhvr, message_id_t awaited_response) {
CPPA_REQUIRE(bhvr.timeout().valid()); CPPA_REQUIRE(bhvr.timeout().valid());
CPPA_REQUIRE(bhvr.timeout().is_zero() == false); CPPA_REQUIRE(bhvr.timeout().is_zero() == false);
if (invoke_from_cache(client, bhvr, active_request) == false) { if (invoke_from_cache(client, bhvr, awaited_response) == false) {
auto timeout = client->init_timeout(bhvr.timeout()); auto timeout = client->init_timeout(bhvr.timeout());
pointer 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, active_request)) { if (invoke(client, e, bhvr, awaited_response)) {
return; // done return; // done
} }
} }
...@@ -229,7 +229,7 @@ class receive_policy { ...@@ -229,7 +229,7 @@ class receive_policy {
auto v0 = msg.get_as<atom_value>(0); auto v0 = msg.get_as<atom_value>(0);
auto v1 = msg.get_as<std::uint32_t>(1); auto v1 = msg.get_as<std::uint32_t>(1);
if (v0 == atom("EXIT")) { if (v0 == atom("EXIT")) {
CPPA_REQUIRE(message_id.is_async()); CPPA_REQUIRE(!message_id.valid());
if (client->m_trap_exit == false) { if (client->m_trap_exit == false) {
if (v1 != exit_reason::normal) { if (v1 != exit_reason::normal) {
// TODO: check for possible memory leak here // TODO: check for possible memory leak here
...@@ -240,7 +240,7 @@ class receive_policy { ...@@ -240,7 +240,7 @@ class receive_policy {
} }
} }
else if (v0 == atom("TIMEOUT")) { else if (v0 == atom("TIMEOUT")) {
CPPA_REQUIRE(message_id.is_async()); CPPA_REQUIRE(!message_id.valid());
return client->waits_for_timeout(v1) ? timeout_message return client->waits_for_timeout(v1) ? timeout_message
: expired_timeout_message; : expired_timeout_message;
} }
...@@ -322,8 +322,8 @@ class receive_policy { ...@@ -322,8 +322,8 @@ class receive_policy {
handle_message_result handle_message(Client* client, handle_message_result handle_message(Client* client,
pointer node, pointer node,
Fun& fun, Fun& fun,
message_id_t active_request, message_id_t awaited_response,
Policy policy ) { Policy policy ) {
if (hm_should_skip(node, policy)) { if (hm_should_skip(node, policy)) {
return hm_skip_msg; return hm_skip_msg;
} }
...@@ -337,11 +337,13 @@ class receive_policy { ...@@ -337,11 +337,13 @@ class receive_policy {
} }
case timeout_message: { case timeout_message: {
handle_timeout(client, fun); handle_timeout(client, fun);
if (awaited_response.valid()) {
client->mark_arrived(awaited_response);
}
return hm_msg_handled; return hm_msg_handled;
} }
case sync_response: { case sync_response: {
if ( active_request.is_async() == false if (awaited_response.valid() && node->mid == awaited_response) {
&& node->mid.request_id() == active_request) {
hm_begin(client, node, policy); hm_begin(client, node, policy);
# ifdef CPPA_DEBUG # ifdef CPPA_DEBUG
if (!fun(node->msg)) { if (!fun(node->msg)) {
...@@ -352,19 +354,20 @@ class receive_policy { ...@@ -352,19 +354,20 @@ class receive_policy {
fun(node->msg); fun(node->msg);
# endif # endif
hm_cleanup(client, policy); hm_cleanup(client, policy);
client->mark_arrived(awaited_response);
return hm_msg_handled; return hm_msg_handled;
} }
return hm_cache_msg; return hm_cache_msg;
} }
case ordinary_message: { case ordinary_message: {
if (active_request.is_async()) { if (!awaited_response.valid()) {
auto previous_node = hm_begin(client, node, policy); auto previous_node = hm_begin(client, node, policy);
if (fun(node->msg)) { if (fun(node->msg)) {
// make sure synchronous request // make sure synchronous request
// always receive a response // always receive a response
auto id = node->mid; auto id = node->mid;
auto sender = node->sender; auto sender = node->sender;
if (!id.is_async() && !id.is_answered() && sender) { if (id.valid() && !id.is_answered() && sender) {
sender->sync_enqueue(client, sender->sync_enqueue(client,
id.response_id(), id.response_id(),
any_tuple()); any_tuple());
......
...@@ -333,7 +333,7 @@ class local_actor : public actor { ...@@ -333,7 +333,7 @@ class local_actor : public actor {
} }
inline message_id_t send_sync_message(actor* whom, any_tuple&& what) { inline message_id_t send_sync_message(actor* whom, any_tuple&& what) {
auto id = ++m_current_message_id; auto id = ++m_last_request_id;
CPPA_REQUIRE(id.is_request()); CPPA_REQUIRE(id.is_request());
if (chaining_enabled()) { if (chaining_enabled()) {
if (whom->chained_sync_enqueue(this, id, std::move(what))) { if (whom->chained_sync_enqueue(this, id, std::move(what))) {
...@@ -341,8 +341,9 @@ class local_actor : public actor { ...@@ -341,8 +341,9 @@ class local_actor : public actor {
} }
} }
else whom->sync_enqueue(this, id, std::move(what)); else whom->sync_enqueue(this, id, std::move(what));
m_requests.push_back(id); auto awaited_response = id.response_id();
return id; m_awaited_responses.push_back(awaited_response);
return awaited_response;
} }
// returns 0 if last_dequeued() is an asynchronous or sync request message, // returns 0 if last_dequeued() is an asynchronous or sync request message,
...@@ -357,10 +358,19 @@ class local_actor : public actor { ...@@ -357,10 +358,19 @@ class local_actor : public actor {
return m_chained_actor; return m_chained_actor;
} }
bool awaits(message_id_t id) { bool awaits(message_id_t response_id) {
auto req_id = id.request_id(); CPPA_REQUIRE(response_id.is_response());
return std::any_of(m_requests.begin(), m_requests.end(), return std::any_of(m_awaited_responses.begin(),
[req_id](message_id_t id) { return id == req_id; }); m_awaited_responses.end(),
[=](message_id_t other) {
return response_id == other;
});
}
void mark_arrived(message_id_t response_id) {
auto last = m_awaited_responses.end();
auto i = std::find(m_awaited_responses.begin(), last, response_id);
if (i != last) m_awaited_responses.erase(i);
} }
protected: protected:
...@@ -374,9 +384,9 @@ class local_actor : public actor { ...@@ -374,9 +384,9 @@ class local_actor : public actor {
// pointer to the actor that is marked as successor due to a chained send // pointer to the actor that is marked as successor due to a chained send
actor_ptr m_chained_actor; actor_ptr m_chained_actor;
// identifies the ID of the last sent synchronous request // identifies the ID of the last sent synchronous request
message_id_t m_current_message_id; message_id_t m_last_request_id;
// identifies all IDs of sync messages waiting for a response // identifies all IDs of sync messages waiting for a response
std::vector<message_id_t> m_requests; std::vector<message_id_t> m_awaited_responses;
// "default value" for m_current_node // "default value" for m_current_node
detail::recursive_queue_node m_dummy_node; detail::recursive_queue_node m_dummy_node;
// points to m_dummy_node if no callback is currently invoked, // points to m_dummy_node if no callback is currently invoked,
......
...@@ -35,6 +35,10 @@ ...@@ -35,6 +35,10 @@
namespace cppa { namespace cppa {
/**
* @brief
* @note Asynchronous messages always have an invalid message id.
*/
class message_id_t { class message_id_t {
static constexpr std::uint64_t response_flag_mask = 0x8000000000000000; static constexpr std::uint64_t response_flag_mask = 0x8000000000000000;
...@@ -65,16 +69,16 @@ class message_id_t { ...@@ -65,16 +69,16 @@ class message_id_t {
return (m_value & answered_flag_mask) != 0; return (m_value & answered_flag_mask) != 0;
} }
inline bool is_async() const { inline bool valid() const {
return m_value == 0; return m_value != 0;
} }
inline bool is_request() const { inline bool is_request() const {
return !is_async() && !is_response(); return valid() && !is_response();
} }
inline message_id_t response_id() const { inline message_id_t response_id() const {
return message_id_t(is_async() ? 0 : m_value | response_flag_mask); return message_id_t(valid() ? m_value | response_flag_mask : 0);
} }
inline message_id_t request_id() const { inline message_id_t request_id() const {
......
...@@ -132,11 +132,8 @@ class scheduler { ...@@ -132,11 +132,8 @@ class scheduler {
message_id_t id, message_id_t id,
Data&&... data ) { Data&&... data ) {
static_assert(sizeof...(Data) > 0, "no message to send"); static_assert(sizeof...(Data) > 0, "no message to send");
CPPA_REQUIRE(id.is_async() || id.is_response()); CPPA_REQUIRE(!id.valid() || id.is_response());
if (id.is_async()) { if (id.valid()) {
delayed_send(to, rel_time, id, std::forward<Data>(data)...);
}
else {
auto sub = make_any_tuple(std::forward<Data>(data)...); auto sub = make_any_tuple(std::forward<Data>(data)...);
auto tup = make_any_tuple(atom("REPLY"), auto tup = make_any_tuple(atom("REPLY"),
util::duration{rel_time}, util::duration{rel_time},
...@@ -145,6 +142,9 @@ class scheduler { ...@@ -145,6 +142,9 @@ class scheduler {
std::move(sub)); std::move(sub));
delayed_send_helper()->enqueue(self, std::move(tup)); delayed_send_helper()->enqueue(self, std::move(tup));
} }
else {
delayed_send(to, rel_time, id, std::forward<Data>(data)...);
}
} }
/** /**
......
...@@ -108,10 +108,10 @@ void local_actor::reply_message(any_tuple&& what) { ...@@ -108,10 +108,10 @@ void local_actor::reply_message(any_tuple&& what) {
return; return;
} }
auto response_id = get_response_id(); auto response_id = get_response_id();
if (response_id.is_async()) { if (!response_id.valid()) {
send_message(whom.get(), std::move(what)); send_message(whom.get(), std::move(what));
} }
else if (m_chaining && !m_chained_actor) { else if (chaining_enabled()) {
if (whom->chained_sync_enqueue(this, response_id, std::move(what))) { if (whom->chained_sync_enqueue(this, response_id, std::move(what))) {
m_chained_actor = whom; m_chained_actor = whom;
} }
......
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