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

implemented event-based synchronous response handling

parent 18ac4661
...@@ -478,38 +478,21 @@ inline message_future sync_send(const actor_ptr& whom, Args&&... what) { ...@@ -478,38 +478,21 @@ inline message_future sync_send(const actor_ptr& whom, Args&&... what) {
else throw std::invalid_argument("whom == nullptr"); else throw std::invalid_argument("whom == nullptr");
} }
struct receive_response_helper {
message_future m_handle;
inline receive_response_helper(message_future handle)
: m_handle(handle) { }
template<typename... Expression>
inline void operator()(Expression&&... mexpr) const {
auto bhvr = match_expr_convert(std::forward<Expression>(mexpr)...);
static_assert(std::is_same<decltype(bhvr), behavior>::value,
"no timeout specified");
if (bhvr.timeout().valid() == false || bhvr.timeout().is_zero()) {
throw std::invalid_argument("specified timeout is invalid or zero");
}
else if (!m_handle.valid() || !m_handle.is_response()) {
throw std::logic_error("handle does not point to a response");
}
else if (!self->awaits(m_handle)) {
throw std::logic_error("response already received");
}
self->dequeue_response(bhvr, m_handle);
}
};
/** /**
* @brief Receives a synchronous response message. * @brief Handles a synchronous response message in an event-based way.
* @param handle A future for a synchronous response. * @param handle A future for a synchronous response.
* @throws std::invalid_argument if given behavior does not has a valid * @throws std::invalid_argument if given behavior does not has a valid
* timeout definition * timeout definition
* @throws std::logic_error if @p handle is not valid or if the actor * @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle * already received the response for @p handle
*/ */
inline receive_response_helper receive_response(message_future handle) { inline sync_recv_helper receive_response(message_future handle) {
return {handle}; return {handle, [](behavior& bhvr, message_future mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
self->dequeue_response(bhvr, mf);
}};
} }
/** /**
......
...@@ -200,7 +200,7 @@ class abstract_actor : public abstract_actor_base<Base> { ...@@ -200,7 +200,7 @@ class abstract_actor : public abstract_actor_base<Base> {
m_nodes.pop_back(); m_nodes.pop_back();
} }
} }
if (result) result->reset(sender, id, std::move(msg)); if (result) result->reset(sender, std::move(msg), id);
else result = new mailbox_element(sender, std::move(msg), id); else result = new mailbox_element(sender, std::move(msg), id);
return result; return result;
} }
......
...@@ -58,7 +58,7 @@ struct recursive_queue_node { ...@@ -58,7 +58,7 @@ struct recursive_queue_node {
if (reset_msg) msg.reset(); if (reset_msg) msg.reset();
} }
inline void reset(actor* sptr, message_id_t id, any_tuple&& data) { inline void reset(actor* sptr, any_tuple&& data, message_id_t id) {
reset(sptr, id, false); reset(sptr, id, false);
msg = std::move(data); msg = std::move(data);
} }
......
...@@ -49,6 +49,7 @@ struct scheduled_actor_dummy : abstract_scheduled_actor { ...@@ -49,6 +49,7 @@ struct scheduled_actor_dummy : abstract_scheduled_actor {
bool attach(attachable*); bool attach(attachable*);
void unbecome(); void unbecome();
void do_become(behavior&&, bool); void do_become(behavior&&, bool);
void become_waiting_for(behavior&&, message_future);
bool has_behavior(); bool has_behavior();
scheduled_actor_type impl_type(); scheduled_actor_type impl_type();
}; };
......
...@@ -35,6 +35,7 @@ ...@@ -35,6 +35,7 @@
#include <functional> #include <functional>
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/detail/receive_policy.hpp" #include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp" #include "cppa/detail/behavior_stack.hpp"
...@@ -82,18 +83,11 @@ class stacked_actor_mixin : public Base { ...@@ -82,18 +83,11 @@ class stacked_actor_mixin : public Base {
stacked_actor_mixin(std::function<void()> f) : m_behavior(std::move(f)) { } stacked_actor_mixin(std::function<void()> f) : m_behavior(std::move(f)) { }
virtual void do_become(behavior&& bhvr, bool discard_old) { virtual void do_become(behavior&& bhvr, bool discard_old) {
if (m_bhvr_stack_ptr) { become_impl(std::move(bhvr), discard_old, message_id_t());
if (discard_old) m_bhvr_stack_ptr->pop_async_back(); }
m_bhvr_stack_ptr->push_back(std::move(bhvr));
} virtual void become_waiting_for(behavior&& bhvr, message_future mid) {
else { become_impl(std::move(bhvr), false, mid);
m_bhvr_stack_ptr.reset(new behavior_stack);
m_bhvr_stack_ptr->push_back(std::move(bhvr));
if (this->initialized()) {
m_bhvr_stack_ptr->exec(m_recv_policy, dthis());
m_bhvr_stack_ptr.reset();
}
}
} }
virtual bool has_behavior() { virtual bool has_behavior() {
...@@ -112,6 +106,21 @@ class stacked_actor_mixin : public Base { ...@@ -112,6 +106,21 @@ class stacked_actor_mixin : public Base {
return static_cast<Derived*>(this); return static_cast<Derived*>(this);
} }
void become_impl(behavior&& bhvr, bool discard_old, message_id_t mid) {
if (m_bhvr_stack_ptr) {
if (discard_old) m_bhvr_stack_ptr->pop_async_back();
m_bhvr_stack_ptr->push_back(std::move(bhvr), mid);
}
else {
m_bhvr_stack_ptr.reset(new behavior_stack);
m_bhvr_stack_ptr->push_back(std::move(bhvr), mid);
if (this->initialized()) {
m_bhvr_stack_ptr->exec(m_recv_policy, dthis());
m_bhvr_stack_ptr.reset();
}
}
}
}; };
} } // namespace cppa::detail } } // namespace cppa::detail
......
...@@ -134,6 +134,8 @@ class event_based_actor : public detail::abstract_scheduled_actor { ...@@ -134,6 +134,8 @@ class event_based_actor : public detail::abstract_scheduled_actor {
void do_become(behavior&& bhvr, bool discard_old); void do_become(behavior&& bhvr, bool discard_old);
void become_waiting_for(behavior&& bhvr, message_future mf);
private: private:
inline behavior& get_behavior() { inline behavior& get_behavior() {
......
...@@ -46,6 +46,10 @@ ...@@ -46,6 +46,10 @@
namespace cppa { namespace cppa {
#ifndef CPPA_DOCUMENTATION
typedef message_id_t message_future;
#endif // CPPA_DOCUMENTATION
// forward declarations // forward declarations
class scheduler; class scheduler;
class local_scheduler; class local_scheduler;
...@@ -79,12 +83,36 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t(); ...@@ -79,12 +83,36 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t();
} // namespace <anonymous> } // namespace <anonymous>
#endif // CPPA_DOCUMENTATION #endif // CPPA_DOCUMENTATION
struct sync_recv_helper {
typedef void (*callback_type)(behavior&, message_future);
message_future m_handle;
callback_type m_fun;
inline sync_recv_helper(message_future handle, callback_type fun)
: m_handle(handle), m_fun(fun) { }
template<typename... Expression>
inline void operator()(Expression&&... mexpr) const {
auto bhvr = match_expr_convert(std::forward<Expression>(mexpr)...);
static_assert(std::is_same<decltype(bhvr), behavior>::value,
"no timeout specified");
if (bhvr.timeout().valid() == false || bhvr.timeout().is_zero()) {
throw std::invalid_argument("specified timeout is invalid or zero");
}
else if (!m_handle.valid() || !m_handle.is_response()) {
throw std::logic_error("handle does not point to a response");
}
m_fun(bhvr, m_handle);
}
};
inline sync_recv_helper receive_response(message_future);
/** /**
* @brief Base class for local running Actors. * @brief Base class for local running Actors.
*/ */
class local_actor : public actor { class local_actor : public actor {
friend class scheduler; friend class scheduler;
friend inline sync_recv_helper receive_response(message_future);
public: public:
...@@ -141,18 +169,6 @@ class local_actor : public actor { ...@@ -141,18 +169,6 @@ class local_actor : public actor {
*/ */
virtual void dequeue(behavior& bhvr) = 0; virtual void dequeue(behavior& bhvr) = 0;
/**
* @brief Waits for a response to @p request_id.
* Blocks until either a response message arrives or until a
* timeout occurs.
* @param bhvr A behavior with timeout denoting the
* actor's response to the response message.
* @warning You should not call this member function by hand.
* Use always the {@link cppa::receive_response receive_response}
* function.
*/
virtual void dequeue_response(behavior& bhvr, message_id_t request_id) = 0;
/** /**
* @brief Checks whether this actor traps exit messages. * @brief Checks whether this actor traps exit messages.
*/ */
...@@ -279,6 +295,23 @@ class local_actor : public actor { ...@@ -279,6 +295,23 @@ class local_actor : public actor {
true); true);
} }
/**
* @brief Receives a synchronous response message.
* @param handle A future for a synchronous response.
* @throws std::invalid_argument if given behavior does not has a valid
* timeout definition
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
*/
inline sync_recv_helper handle_response(message_future handle) {
return {handle, [](behavior& bhvr, message_future mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
self->become_waiting_for(std::move(bhvr), mf);
}};
}
/** /**
* @brief Returns to a previous behavior if available. * @brief Returns to a previous behavior if available.
*/ */
...@@ -349,7 +382,8 @@ class local_actor : public actor { ...@@ -349,7 +382,8 @@ class local_actor : public actor {
// returns 0 if last_dequeued() is an asynchronous or sync request message, // returns 0 if last_dequeued() is an asynchronous or sync request message,
// a response id generated from the request id otherwise // a response id generated from the request id otherwise
inline message_id_t get_response_id() { inline message_id_t get_response_id() {
return m_current_node->mid.response_id(); auto id = m_current_node->mid;
return (id.is_request()) ? id.response_id() : message_id_t();
} }
void reply_message(any_tuple&& what); void reply_message(any_tuple&& what);
...@@ -406,6 +440,20 @@ class local_actor : public actor { ...@@ -406,6 +440,20 @@ class local_actor : public actor {
do_become(std::move(copy), discard_old); do_become(std::move(copy), discard_old);
} }
/*
* @brief Waits for a response to @p request_id.
* Blocks until either a response message arrives or until a
* timeout occurs.
* @param bhvr A behavior with timeout denoting the
* actor's response to the response message.
* @warning You should not call this member function by hand.
* Use always the {@link cppa::receive_response receive_response}
* function.
*/
virtual void dequeue_response(behavior&, message_future) = 0;
virtual void become_waiting_for(behavior&&, message_future) = 0;
}; };
/** /**
......
...@@ -102,6 +102,12 @@ void event_based_actor::do_become(behavior&& bhvr, bool discard_old) { ...@@ -102,6 +102,12 @@ void event_based_actor::do_become(behavior&& bhvr, bool discard_old) {
m_bhvr_stack.push_back(std::move(bhvr)); m_bhvr_stack.push_back(std::move(bhvr));
} }
void event_based_actor::become_waiting_for(behavior&& bhvr, message_future mf) {
reset_timeout();
request_timeout(bhvr.timeout());
m_bhvr_stack.push_back(std::move(bhvr), mf);
}
void event_based_actor::quit(std::uint32_t reason) { void event_based_actor::quit(std::uint32_t reason) {
if (reason == exit_reason::normal) { if (reason == exit_reason::normal) {
cleanup(exit_reason::normal); cleanup(exit_reason::normal);
......
...@@ -107,16 +107,16 @@ void local_actor::reply_message(any_tuple&& what) { ...@@ -107,16 +107,16 @@ void local_actor::reply_message(any_tuple&& what) {
if (whom == nullptr) { if (whom == nullptr) {
return; return;
} }
auto response_id = get_response_id(); auto id = m_current_node->mid;
if (!response_id.valid()) { if (id.valid() == false || id.is_response()) {
send_message(whom.get(), std::move(what)); send_message(whom.get(), std::move(what));
} }
else if (chaining_enabled()) { else if (chaining_enabled()) {
if (whom->chained_sync_enqueue(this, response_id, std::move(what))) { if (whom->chained_sync_enqueue(this, id.response_id(), std::move(what))) {
m_chained_actor = whom; m_chained_actor = whom;
} }
} }
else whom->sync_enqueue(this, response_id, std::move(what)); else whom->sync_enqueue(this, id.response_id(), std::move(what));
} }
} // namespace cppa } // namespace cppa
...@@ -42,6 +42,7 @@ void scheduled_actor_dummy::detach(const attachable::token&) { } ...@@ -42,6 +42,7 @@ void scheduled_actor_dummy::detach(const attachable::token&) { }
bool scheduled_actor_dummy::attach(attachable*) { return false; } bool scheduled_actor_dummy::attach(attachable*) { return false; }
void scheduled_actor_dummy::unbecome() { } void scheduled_actor_dummy::unbecome() { }
void scheduled_actor_dummy::do_become(behavior&&, bool) { } void scheduled_actor_dummy::do_become(behavior&&, bool) { }
void scheduled_actor_dummy::become_waiting_for(behavior&&, message_future) { }
bool scheduled_actor_dummy::has_behavior() { return false; } bool scheduled_actor_dummy::has_behavior() { return false; }
resume_result scheduled_actor_dummy::resume(util::fiber*) { resume_result scheduled_actor_dummy::resume(util::fiber*) {
......
...@@ -422,23 +422,6 @@ size_t test__spawn() { ...@@ -422,23 +422,6 @@ size_t test__spawn() {
self << any_tuple{}; self << any_tuple{};
receive(on() >> []() { }); receive(on() >> []() { });
/*
// (2) sync_send returning a 'future'
auto future = sync_send(foo, atom("get_state"));
// * blocking
receive_response(future) (
// ...
);
// * event-based
become_waiting_for(future) (
// ...
);
*/
CPPA_IF_VERBOSE(cout << "test receive with zero timeout ... " << std::flush); CPPA_IF_VERBOSE(cout << "test receive with zero timeout ... " << std::flush);
receive ( receive (
others() >> []() { others() >> []() {
...@@ -604,6 +587,47 @@ size_t test__spawn() { ...@@ -604,6 +587,47 @@ size_t test__spawn() {
await_all_others_done(); await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl); CPPA_IF_VERBOSE(cout << "ok" << endl);
auto sync_testee_factory = factory::event_based(
[&]() {
self->become (
on("hi") >> [&]() {
auto handle = sync_send(self->last_sender(), "whassup?");
self->handle_response(handle) (
on_arg_match >> [&](const std::string& str) {
CPPA_CHECK(self->last_sender() != nullptr);
CPPA_CHECK_EQUAL("nothing", str);
reply("goodbye!");
self->quit();
},
after(std::chrono::minutes(1)) >> []() {
cerr << "PANIC!!!!" << endl;
abort();
}
);
},
others() >> []() {
cerr << "UNEXPECTED: " << to_string(self->last_dequeued())
<< endl;
}
);
}
);
auto sync_testee = sync_testee_factory.spawn();
send(sync_testee, "hi");
receive (
on("whassup?") >> []() {
// this is NOT a reply, it's just an asynchronous message
send(self->last_sender(), "a lot!");
reply("nothing");
}
);
receive (
on("goodbye!") >> []() { }
);
await_all_others_done();
auto inflater = factory::event_based( auto inflater = factory::event_based(
[](std::string*, actor_ptr* receiver) { [](std::string*, actor_ptr* receiver) {
self->become( self->become(
...@@ -645,8 +669,7 @@ size_t test__spawn() { ...@@ -645,8 +669,7 @@ size_t test__spawn() {
*pal = spawn_next("Bob", self); *pal = spawn_next("Bob", self);
} }
self->become ( self->become (
others() >> [pal,name]() { others() >> [pal]() {
cout << "hello & bye from " << *name << endl;
// forward message and die // forward message and die
*pal << self->last_dequeued(); *pal << self->last_dequeued();
self->quit(); self->quit();
......
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