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

synchronous receive and unit test

parent 999369cd
...@@ -73,6 +73,7 @@ class context_switching_actor : public detail::stacked_actor_mixin< ...@@ -73,6 +73,7 @@ class context_switching_actor : public detail::stacked_actor_mixin<
context_switching_actor, context_switching_actor,
detail::abstract_scheduled_actor> { detail::abstract_scheduled_actor> {
friend class detail::behavior_stack;
friend class detail::receive_policy; friend class detail::receive_policy;
typedef detail::stacked_actor_mixin< typedef detail::stacked_actor_mixin<
......
...@@ -498,6 +498,8 @@ inline receive_response_helper receive_response(unspecified_handle handle) { ...@@ -498,6 +498,8 @@ inline receive_response_helper receive_response(unspecified_handle handle) {
return {handle}; return {handle};
} }
/** /**
* @brief Sends a message to the sender of the last received message. * @brief Sends a message to the sender of the last received message.
* @param what Message elements. * @param what Message elements.
......
...@@ -33,39 +33,87 @@ ...@@ -33,39 +33,87 @@
#include <vector> #include <vector>
#include <memory> #include <memory>
#include <utility>
#include "cppa/config.hpp"
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/message_id.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
namespace cppa { namespace detail { namespace cppa { namespace detail {
class behavior_stack_help_iterator;
class behavior_stack class behavior_stack
{ {
friend class behavior_stack_help_iterator;
behavior_stack(const behavior_stack&) = delete; behavior_stack(const behavior_stack&) = delete;
behavior_stack& operator=(const behavior_stack&) = delete; behavior_stack& operator=(const behavior_stack&) = delete;
typedef std::pair<behavior, message_id_t> element_type;
public: public:
behavior_stack() = default; behavior_stack() = default;
inline bool empty() const { return m_elements.empty(); } inline bool empty() const { return m_elements.empty(); }
inline behavior& back() { return m_elements.back(); } inline behavior& back() {
CPPA_REQUIRE(!empty());
return m_elements.back().first;
}
// executes the behavior stack // erases the last asynchronous message handler
void exec(); void pop_async_back();
void pop_back(); void push_back(behavior&& what,
message_id_t expected_response = message_id_t());
void push_back(behavior&& what);
void cleanup(); void cleanup();
void clear(); void clear();
template<class Policy, class Client>
bool invoke(Policy& policy, Client* client, recursive_queue_node* node) {
CPPA_REQUIRE(!m_elements.empty());
CPPA_REQUIRE(client != nullptr);
CPPA_REQUIRE(node != nullptr);
auto id = m_elements.back().second;
if (policy.invoke(client, node, back(), id)) {
// try to match cached messages
do {
// remove synchronous response handler if needed
if (id.valid()) {
auto last = m_elements.end();
auto i = std::find_if(m_elements.begin(), last,
[id](element_type& e) {
return id == e.second;
});
if (i != last) {
m_erased_elements.emplace_back(std::move(i->first));
m_elements.erase(i);
}
}
cleanup();
id = empty() ? message_id_t() : m_elements.back().second;
} while (!empty() && policy.invoke_from_cache(client, back(), id));
return true;
}
return false;
}
template<class Policy, class Client>
void exec(Policy& policy, Client* client) {
while (!empty()) {
invoke(policy, client, client->receive_node());
}
}
private: private:
std::vector<behavior> m_elements; std::vector<element_type> m_elements;
std::vector<behavior> m_erased_elements; std::vector<behavior> m_erased_elements;
}; };
......
...@@ -49,7 +49,7 @@ class stacked_actor_mixin : public Base { ...@@ -49,7 +49,7 @@ class stacked_actor_mixin : public Base {
virtual void unbecome() { virtual void unbecome() {
if (m_bhvr_stack_ptr) { if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->pop_back(); m_bhvr_stack_ptr->pop_async_back();
} }
} }
...@@ -62,12 +62,12 @@ class stacked_actor_mixin : public Base { ...@@ -62,12 +62,12 @@ class stacked_actor_mixin : public Base {
} }
virtual void dequeue_response(behavior& bhvr, message_id_t request_id) { virtual void dequeue_response(behavior& bhvr, message_id_t request_id) {
m_recv_policy.receive(dthis(), bhvr, request_id ); m_recv_policy.receive(dthis(), bhvr, request_id);
} }
virtual void run() { virtual void run() {
if (m_bhvr_stack_ptr) { if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->exec(); m_bhvr_stack_ptr->exec(m_recv_policy, dthis());
m_bhvr_stack_ptr.reset(); m_bhvr_stack_ptr.reset();
} }
if (m_behavior) { if (m_behavior) {
...@@ -83,14 +83,14 @@ class stacked_actor_mixin : public Base { ...@@ -83,14 +83,14 @@ class stacked_actor_mixin : public Base {
virtual void do_become(behavior&& bhvr, bool discard_old) { virtual void do_become(behavior&& bhvr, bool discard_old) {
if (m_bhvr_stack_ptr) { if (m_bhvr_stack_ptr) {
if (discard_old) m_bhvr_stack_ptr->pop_back(); if (discard_old) m_bhvr_stack_ptr->pop_async_back();
m_bhvr_stack_ptr->push_back(std::move(bhvr)); m_bhvr_stack_ptr->push_back(std::move(bhvr));
} }
else { else {
m_bhvr_stack_ptr.reset(new behavior_stack); m_bhvr_stack_ptr.reset(new behavior_stack);
m_bhvr_stack_ptr->push_back(std::move(bhvr)); m_bhvr_stack_ptr->push_back(std::move(bhvr));
if (this->initialized()) { if (this->initialized()) {
m_bhvr_stack_ptr->exec(); m_bhvr_stack_ptr->exec(m_recv_policy, dthis());
m_bhvr_stack_ptr.reset(); m_bhvr_stack_ptr.reset();
} }
} }
......
...@@ -342,7 +342,7 @@ class local_actor : public actor { ...@@ -342,7 +342,7 @@ class local_actor : public actor {
} }
else whom->sync_enqueue(this, id, std::move(what)); else whom->sync_enqueue(this, id, std::move(what));
auto awaited_response = id.response_id(); auto awaited_response = id.response_id();
m_awaited_responses.push_back(awaited_response); m_pending_responses.push_back(awaited_response);
return awaited_response; return awaited_response;
} }
...@@ -358,19 +358,19 @@ class local_actor : public actor { ...@@ -358,19 +358,19 @@ class local_actor : public actor {
return m_chained_actor; return m_chained_actor;
} }
bool awaits(message_id_t response_id) { inline bool awaits(message_id_t response_id) {
CPPA_REQUIRE(response_id.is_response()); CPPA_REQUIRE(response_id.is_response());
return std::any_of(m_awaited_responses.begin(), return std::any_of(m_pending_responses.begin(),
m_awaited_responses.end(), m_pending_responses.end(),
[=](message_id_t other) { [=](message_id_t other) {
return response_id == other; return response_id == other;
}); });
} }
void mark_arrived(message_id_t response_id) { inline void mark_arrived(message_id_t response_id) {
auto last = m_awaited_responses.end(); auto last = m_pending_responses.end();
auto i = std::find(m_awaited_responses.begin(), last, response_id); auto i = std::find(m_pending_responses.begin(), last, response_id);
if (i != last) m_awaited_responses.erase(i); if (i != last) m_pending_responses.erase(i);
} }
protected: protected:
...@@ -386,7 +386,7 @@ class local_actor : public actor { ...@@ -386,7 +386,7 @@ class local_actor : public actor {
// identifies the ID of the last sent synchronous request // identifies the ID of the last sent synchronous request
message_id_t m_last_request_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_awaited_responses; std::vector<message_id_t> m_pending_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,
......
...@@ -87,6 +87,8 @@ class thread_mapped_actor : public detail::stacked_actor_mixin< ...@@ -87,6 +87,8 @@ class thread_mapped_actor : public detail::stacked_actor_mixin<
detail::abstract_actor<local_actor> > { detail::abstract_actor<local_actor> > {
friend class self_type; // needs access to cleanup() friend class self_type; // needs access to cleanup()
friend class detail::behavior_stack;
friend class detail::receive_policy; friend class detail::receive_policy;
typedef detail::stacked_actor_mixin< typedef detail::stacked_actor_mixin<
......
...@@ -28,34 +28,74 @@ ...@@ -28,34 +28,74 @@
\******************************************************************************/ \******************************************************************************/
#include <iterator>
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/detail/behavior_stack.hpp" #include "cppa/detail/behavior_stack.hpp"
namespace cppa { namespace detail { namespace cppa { namespace detail {
// executes the behavior stack class behavior_stack_help_iterator
void behavior_stack::exec() { : public std::iterator<std::output_iterator_tag, void, void, void, void> {
while (!empty()) {
self->dequeue(back()); public:
cleanup();
typedef behavior_stack::element_type element_type;
explicit behavior_stack_help_iterator(behavior_stack& st) : m_stack(&st) { }
behavior_stack_help_iterator& operator=(element_type&& rval) {
m_stack->m_erased_elements.emplace_back(std::move(rval.first));
return *this;
}
behavior_stack_help_iterator& operator*() { return *this; }
behavior_stack_help_iterator& operator++() { return *this; }
behavior_stack_help_iterator operator++(int) { return *this; }
private:
behavior_stack* m_stack;
};
void behavior_stack::pop_async_back() {
if (m_elements.empty()) {
// nothing to do
}
else if (m_elements.back().second.valid() == false) {
m_erased_elements.emplace_back(std::move(m_elements.back().first));
m_elements.pop_back();
}
else {
auto rlast = m_elements.rend();
auto ri = std::find_if(m_elements.rbegin(), rlast,
[](element_type& e) {
return e.second.valid() == false;
});
if (ri != rlast) {
m_erased_elements.emplace_back(std::move(ri->first));
auto i = ri.base();
--i; // adjust 'normal' iterator to point to the correct element
m_elements.erase(i);
}
} }
}
void behavior_stack::pop_back() {
if (m_elements.empty() == false) { if (m_elements.empty() == false) {
m_erased_elements.emplace_back(std::move(m_elements.back())); m_erased_elements.emplace_back(std::move(m_elements.back().first));
m_elements.pop_back(); m_elements.pop_back();
} }
} }
void behavior_stack::push_back(behavior&& what) { void behavior_stack::push_back(behavior&& what, message_id_t response_id) {
m_elements.emplace_back(std::move(what)); m_elements.emplace_back(std::move(what), response_id);
} }
void behavior_stack::clear() { void behavior_stack::clear() {
if (m_elements.empty() == false) { if (m_elements.empty() == false) {
std::move(m_elements.begin(), m_elements.end(), std::move(m_elements.begin(), m_elements.end(),
std::back_inserter(m_erased_elements)); behavior_stack_help_iterator{*this});
m_elements.clear(); m_elements.clear();
} }
} }
......
...@@ -77,17 +77,11 @@ resume_result event_based_actor::resume(util::fiber*) { ...@@ -77,17 +77,11 @@ resume_result event_based_actor::resume(util::fiber*) {
}; };
} }
} }
else { else if (m_bhvr_stack.invoke(m_policy, this, e)) {
if (m_policy.invoke(this, e, get_behavior())) {
// try to match cached message before receiving new ones
do {
m_bhvr_stack.cleanup();
if (m_bhvr_stack.empty()) { if (m_bhvr_stack.empty()) {
done_cb(); done_cb();
return resume_result::actor_done; return resume_result::actor_done;
} }
} while (m_policy.invoke_from_cache(this, get_behavior()));
}
} }
} }
} }
...@@ -104,7 +98,7 @@ bool event_based_actor::has_behavior() { ...@@ -104,7 +98,7 @@ bool event_based_actor::has_behavior() {
void event_based_actor::do_become(behavior&& bhvr, bool discard_old) { void event_based_actor::do_become(behavior&& bhvr, bool discard_old) {
reset_timeout(); reset_timeout();
request_timeout(bhvr.timeout()); request_timeout(bhvr.timeout());
if (discard_old) m_bhvr_stack.pop_back(); if (discard_old) m_bhvr_stack.pop_async_back();
m_bhvr_stack.push_back(std::move(bhvr)); m_bhvr_stack.push_back(std::move(bhvr));
} }
...@@ -120,7 +114,7 @@ void event_based_actor::quit(std::uint32_t reason) { ...@@ -120,7 +114,7 @@ void event_based_actor::quit(std::uint32_t reason) {
} }
void event_based_actor::unbecome() { void event_based_actor::unbecome() {
m_bhvr_stack.pop_back(); m_bhvr_stack.pop_async_back();
} }
scheduled_actor_type event_based_actor::impl_type() { scheduled_actor_type event_based_actor::impl_type() {
......
...@@ -559,6 +559,51 @@ size_t test__spawn() { ...@@ -559,6 +559,51 @@ size_t test__spawn() {
await_all_others_done(); await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl); CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_IF_VERBOSE(cout << "test sync send/receive ... " << std::flush);
auto sync_testee1 = spawn([]() {
receive (
on(atom("get")) >> []() {
reply(42, 2);
}
);
});
send(self, 0, 0);
auto handle = sync_send(sync_testee1, atom("get"));
// wait for some time (until sync response arrived in mailbox)
receive (after(std::chrono::milliseconds(50)) >> []() { });
// enqueue async messages (must be skipped by receive_response)
send(self, 42, 1);
// must skip sync message
receive (
on(42, arg_match) >> [&](int i) {
CPPA_CHECK_EQUAL(1, i);
}
);
// must skip remaining async message
receive_response (handle) (
on_arg_match >> [&](int a, int b) {
CPPA_CHECK_EQUAL(42, a);
CPPA_CHECK_EQUAL(2, b);
},
others() >> [&]() {
CPPA_ERROR("unexpected message");
},
after(std::chrono::seconds(10)) >> [&]() {
CPPA_ERROR("timeout during receive_response");
}
);
// dequeue remaining async. message
receive (on(0, 0) >> []() { });
// make sure there's no other message in our mailbox
receive (
others() >> [&]() {
CPPA_ERROR("unexpected message");
},
after(std::chrono::seconds(0)) >> []() { }
);
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
int zombie_init_called = 0; int zombie_init_called = 0;
int zombie_on_exit_called = 0; int zombie_on_exit_called = 0;
factory::event_based([&]() { factory::event_based([&]() {
......
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