Commit 7f9d99c4 authored by Dominik Charousset's avatar Dominik Charousset

added `continue_with` to nonblocking API

this patch enables `sync_send(...).then(...).continue_with(...)`
(relates #58) and makes using the nonblocking API explicit to use
in context-switching and thread-based actors by requiring a call
to `self->exec_behavior_stack()`
parent 27ca7829
...@@ -152,8 +152,6 @@ class actor_companion_mixin : public Base { ...@@ -152,8 +152,6 @@ class actor_companion_mixin : public Base {
void dequeue(cppa::partial_function&) { throw_no_recv(); } void dequeue(cppa::partial_function&) { throw_no_recv(); }
void unbecome() { throw_no_become(); }
void dequeue_response(cppa::behavior&, cppa::message_id_t) { void dequeue_response(cppa::behavior&, cppa::message_id_t) {
throw_no_recv(); throw_no_recv();
} }
......
...@@ -104,6 +104,11 @@ class behavior { ...@@ -104,6 +104,11 @@ class behavior {
return (m_impl) && m_impl->invoke(std::forward<T>(arg)); return (m_impl) && m_impl->invoke(std::forward<T>(arg));
} }
template<typename F>
inline behavior add_continuation(F fun) {
return {new detail::continuation_decorator<F>(std::move(fun), m_impl)};
}
private: private:
impl_ptr m_impl; impl_ptr m_impl;
......
...@@ -148,6 +148,53 @@ default_behavior_impl<MatchExpr, F>* new_default_behavior_impl(const MatchExpr& ...@@ -148,6 +148,53 @@ default_behavior_impl<MatchExpr, F>* new_default_behavior_impl(const MatchExpr&
return new default_behavior_impl<MatchExpr, F>(mexpr, d, f); return new default_behavior_impl<MatchExpr, F>(mexpr, d, f);
} }
template<typename F>
class continuation_decorator : public behavior_impl {
public:
typedef typename behavior_impl::pointer pointer;
template<typename Fun>
continuation_decorator(Fun&& fun, pointer decorated)
: m_fun(std::forward<Fun>(fun)), m_decorated(std::move(decorated)) {
CPPA_REQUIRE(m_decorated != nullptr);
}
template<typename T>
inline bool invoke_impl(T& tup) {
if (m_decorated->invoke(tup)) {
m_fun();
return true;
}
return false;
}
bool invoke(any_tuple& tup) {
return invoke_impl(tup);
}
bool invoke(const any_tuple& tup) {
return invoke_impl(tup);
}
bool defined_at(const any_tuple& tup) {
return m_decorated->defined_at(tup);
}
pointer copy(const generic_timeout_definition& tdef) const {
return new continuation_decorator<F>(m_fun, m_decorated->copy(tdef));
}
void handle_timeout() { m_decorated->handle_timeout(); }
private:
F m_fun;
pointer m_decorated;
};
} } // namespace cppa::detail } } // namespace cppa::detail
#endif // BEHAVIOR_IMPL_HPP #endif // BEHAVIOR_IMPL_HPP
...@@ -35,6 +35,7 @@ ...@@ -35,6 +35,7 @@
#include <memory> #include <memory>
#include <utility> #include <utility>
#include "cppa/option.hpp"
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/message_id.hpp" #include "cppa/message_id.hpp"
...@@ -65,6 +66,9 @@ class behavior_stack ...@@ -65,6 +66,9 @@ class behavior_stack
return m_elements.back().first; return m_elements.back().first;
} }
// @pre expected_response.valid()
option<behavior&> sync_handler(message_id_t expected_response);
// erases the last asynchronous message handler // erases the last asynchronous message handler
void pop_async_back(); void pop_async_back();
......
...@@ -47,7 +47,6 @@ struct scheduled_actor_dummy : abstract_scheduled_actor { ...@@ -47,7 +47,6 @@ struct scheduled_actor_dummy : abstract_scheduled_actor {
bool remove_backlink(const intrusive_ptr<actor>&); bool remove_backlink(const intrusive_ptr<actor>&);
void detach(const attachable::token&); void detach(const attachable::token&);
bool attach(attachable*); bool attach(attachable*);
void unbecome();
void do_become(behavior&&, bool); void do_become(behavior&&, bool);
void become_waiting_for(behavior&&, message_id_t); void become_waiting_for(behavior&&, message_id_t);
bool has_behavior(); bool has_behavior();
......
...@@ -38,7 +38,6 @@ ...@@ -38,7 +38,6 @@
#include "cppa/local_actor.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/recursive_queue_node.hpp" #include "cppa/detail/recursive_queue_node.hpp"
namespace cppa { namespace detail { namespace cppa { namespace detail {
...@@ -50,12 +49,6 @@ class stacked_actor_mixin : public Base { ...@@ -50,12 +49,6 @@ class stacked_actor_mixin : public Base {
public: public:
virtual void unbecome() {
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->pop_async_back();
}
}
virtual void dequeue(partial_function& fun) { virtual void dequeue(partial_function& fun) {
m_recv_policy.receive(dthis(), fun); m_recv_policy.receive(dthis(), fun);
} }
...@@ -69,9 +62,8 @@ class stacked_actor_mixin : public Base { ...@@ -69,9 +62,8 @@ class stacked_actor_mixin : public Base {
} }
virtual void run() { virtual void run() {
if (m_bhvr_stack_ptr) { if (!this->m_bhvr_stack.empty()) {
m_bhvr_stack_ptr->exec(m_recv_policy, dthis()); this->m_bhvr_stack.exec(m_recv_policy, dthis());
m_bhvr_stack_ptr.reset();
} }
if (m_behavior) { if (m_behavior) {
m_behavior(); m_behavior();
...@@ -94,41 +86,31 @@ class stacked_actor_mixin : public Base { ...@@ -94,41 +86,31 @@ class stacked_actor_mixin : public Base {
virtual bool has_behavior() { virtual bool has_behavior() {
return static_cast<bool>(m_behavior) return static_cast<bool>(m_behavior)
|| ( static_cast<bool>(m_bhvr_stack_ptr) || !this->m_bhvr_stack.empty();
&& m_bhvr_stack_ptr->empty() == false);
} }
private: private:
std::function<void()> m_behavior; std::function<void()> m_behavior;
receive_policy m_recv_policy; receive_policy m_recv_policy;
std::unique_ptr<behavior_stack> m_bhvr_stack_ptr;
inline Derived* dthis() { inline Derived* dthis() {
return static_cast<Derived*>(this); return static_cast<Derived*>(this);
} }
void become_impl(behavior&& bhvr, bool discard_old, message_id_t mid) { void become_impl(behavior&& bhvr, bool discard_old, message_id_t mid) {
if (bhvr.timeout().valid()) {
dthis()->reset_timeout(); dthis()->reset_timeout();
dthis()->request_timeout(bhvr.timeout()); dthis()->request_timeout(bhvr.timeout());
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();
} }
if (!this->m_bhvr_stack.empty() && discard_old) {
this->m_bhvr_stack.pop_async_back();
} }
this->m_bhvr_stack.push_back(std::move(bhvr), mid);
} }
inline void remove_handler(message_id_t id) { virtual void exec_behavior_stack() {
if (m_bhvr_stack_ptr) { this->m_bhvr_stack.exec(m_recv_policy, dthis());
m_bhvr_stack_ptr->erase(id);
}
} }
}; };
......
...@@ -40,7 +40,6 @@ ...@@ -40,7 +40,6 @@
#include "cppa/either.hpp" #include "cppa/either.hpp"
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/detail/receive_policy.hpp" #include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp" #include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa { namespace cppa {
...@@ -84,8 +83,6 @@ class event_based_actor : public detail::abstract_scheduled_actor { ...@@ -84,8 +83,6 @@ class event_based_actor : public detail::abstract_scheduled_actor {
void quit(std::uint32_t reason = exit_reason::normal); void quit(std::uint32_t reason = exit_reason::normal);
void unbecome();
bool has_behavior(); bool has_behavior();
scheduled_actor_type impl_type(); scheduled_actor_type impl_type();
...@@ -156,14 +153,7 @@ class event_based_actor : public detail::abstract_scheduled_actor { ...@@ -156,14 +153,7 @@ class event_based_actor : public detail::abstract_scheduled_actor {
request_timeout(get_behavior().timeout()); request_timeout(get_behavior().timeout());
} }
} }
inline void remove_handler(message_id_t id) {
m_bhvr_stack.erase(id);
}
// stack elements are moved to m_erased_stack_elements and erased later
// to prevent possible segfaults that can occur if a currently executed
// lambda gets deleted
detail::behavior_stack m_bhvr_stack;
detail::receive_policy m_policy; detail::receive_policy m_policy;
}; };
......
...@@ -47,6 +47,7 @@ ...@@ -47,6 +47,7 @@
#include "cppa/util/duration.hpp" #include "cppa/util/duration.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/recursive_queue_node.hpp" #include "cppa/detail/recursive_queue_node.hpp"
namespace cppa { namespace cppa {
...@@ -56,6 +57,9 @@ class scheduler; ...@@ -56,6 +57,9 @@ class scheduler;
class message_future; class message_future;
class local_scheduler; class local_scheduler;
namespace detail { class receive_policy; }
template<bool DiscardOld> template<bool DiscardOld>
struct behavior_policy { static const bool discard_old = DiscardOld; }; struct behavior_policy { static const bool discard_old = DiscardOld; };
...@@ -119,6 +123,7 @@ class local_actor : public memory_cached_mixin<actor> { ...@@ -119,6 +123,7 @@ class local_actor : public memory_cached_mixin<actor> {
friend class scheduler; friend class scheduler;
friend class detail::memory; friend class detail::memory;
friend class detail::receive_policy;
template<typename> friend class detail::basic_memory_cache; template<typename> friend class detail::basic_memory_cache;
public: public:
...@@ -315,7 +320,9 @@ class local_actor : public memory_cached_mixin<actor> { ...@@ -315,7 +320,9 @@ class local_actor : public memory_cached_mixin<actor> {
/** /**
* @brief Returns to a previous behavior if available. * @brief Returns to a previous behavior if available.
*/ */
virtual void unbecome() = 0; inline void unbecome() {
m_bhvr_stack.pop_async_back();
}
/** /**
* @brief Can be overridden to initialize an actor before any * @brief Can be overridden to initialize an actor before any
...@@ -339,6 +346,15 @@ class local_actor : public memory_cached_mixin<actor> { ...@@ -339,6 +346,15 @@ class local_actor : public memory_cached_mixin<actor> {
*/ */
std::vector<group_ptr> joined_groups(); std::vector<group_ptr> joined_groups();
/**
* @brief Executes an actor's behavior stack until it is empty.
*
* This member function allows thread-based and context-switching actors
* to make use of the nonblocking behavior API of libcppa.
* Calling this member function in an event-based actor does nothing.
*/
virtual void exec_behavior_stack();
// library-internal members and member functions that shall // library-internal members and member functions that shall
// not appear in the documentation // not appear in the documentation
...@@ -441,6 +457,10 @@ class local_actor : public memory_cached_mixin<actor> { ...@@ -441,6 +457,10 @@ class local_actor : public memory_cached_mixin<actor> {
else quit(exit_reason::unhandled_sync_failure); else quit(exit_reason::unhandled_sync_failure);
} }
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
protected: protected:
// true if this actor uses the chained_send optimization // true if this actor uses the chained_send optimization
...@@ -462,6 +482,12 @@ class local_actor : public memory_cached_mixin<actor> { ...@@ -462,6 +482,12 @@ class local_actor : public memory_cached_mixin<actor> {
detail::recursive_queue_node* m_current_node; detail::recursive_queue_node* m_current_node;
// {group => subscription} map of all joined groups // {group => subscription} map of all joined groups
std::map<group_ptr, group::subscription> m_subscriptions; std::map<group_ptr, group::subscription> m_subscriptions;
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
inline void remove_handler(message_id_t id) {
m_bhvr_stack.erase(id);
}
private: private:
......
...@@ -48,15 +48,39 @@ class message_future { ...@@ -48,15 +48,39 @@ class message_future {
public: public:
class continue_helper {
public:
inline continue_helper(message_id_t mid) : m_mid(mid) { }
template<typename F>
void continue_with(F fun) {
auto ref_opt = self->bhvr_stack().sync_handler(m_mid);
if (ref_opt) {
auto& ref = *ref_opt;
// copy original behavior
behavior cpy = ref;
ref = cpy.add_continuation(std::move(fun));
}
}
private:
message_id_t m_mid;
};
message_future() = delete; message_future() = delete;
/** /**
* @brief Sets @p mexpr as event-handler for the response message. * @brief Sets @p mexpr as event-handler for the response message.
*/ */
template<typename... Cases, typename... Args> template<typename... Cases, typename... Args>
void then(const match_expr<Cases...>& arg0, const Args&... args) { continue_helper then(const match_expr<Cases...>& arg0, const Args&... args) {
consistency_check(); consistency_check();
self->become_waiting_for(match_expr_convert(arg0, args...), m_id); self->become_waiting_for(match_expr_convert(arg0, args...), m_mid);
return {m_mid};
} }
/** /**
...@@ -65,7 +89,7 @@ class message_future { ...@@ -65,7 +89,7 @@ class message_future {
template<typename... Cases, typename... Args> template<typename... Cases, typename... Args>
void await(const match_expr<Cases...>& arg0, const Args&... args) { void await(const match_expr<Cases...>& arg0, const Args&... args) {
consistency_check(); consistency_check();
self->dequeue_response(match_expr_convert(arg0, args...), m_id); self->dequeue_response(match_expr_convert(arg0, args...), m_mid);
} }
/** /**
...@@ -74,9 +98,11 @@ class message_future { ...@@ -74,9 +98,11 @@ class message_future {
* is an 'EXITED', 'TIMEOUT', or 'VOID' message. * is an 'EXITED', 'TIMEOUT', or 'VOID' message.
*/ */
template<typename F> template<typename F>
typename std::enable_if<util::is_callable<F>::value>::type then(F fun) { typename std::enable_if<util::is_callable<F>::value,continue_helper>::type
then(F fun) {
consistency_check(); consistency_check();
self->become_waiting_for(bhvr_from_fun(fun), m_id); self->become_waiting_for(bhvr_from_fun(fun), m_mid);
return {m_mid};
} }
/** /**
...@@ -87,26 +113,26 @@ class message_future { ...@@ -87,26 +113,26 @@ class message_future {
template<typename F> template<typename F>
typename std::enable_if<util::is_callable<F>::value>::type await(F fun) { typename std::enable_if<util::is_callable<F>::value>::type await(F fun) {
consistency_check(); consistency_check();
self->dequeue_response(bhvr_from_fun(fun), m_id); self->dequeue_response(bhvr_from_fun(fun), m_mid);
} }
/** /**
* @brief Returns the awaited response ID. * @brief Returns the awaited response ID.
*/ */
inline const message_id_t& id() const { return m_id; } inline const message_id_t& id() const { return m_mid; }
message_future(const message_future&) = default; message_future(const message_future&) = default;
message_future& operator=(const message_future&) = default; message_future& operator=(const message_future&) = default;
# ifndef CPPA_DOCUMENTATION # ifndef CPPA_DOCUMENTATION
inline message_future(const message_id_t& from) : m_id(from) { } inline message_future(const message_id_t& from) : m_mid(from) { }
# endif # endif
private: private:
message_id_t m_id; message_id_t m_mid;
template<typename F> template<typename F>
behavior bhvr_from_fun(F fun) { behavior bhvr_from_fun(F fun) {
...@@ -121,32 +147,13 @@ class message_future { ...@@ -121,32 +147,13 @@ class message_future {
} }
void consistency_check() { void consistency_check() {
if (!m_id.valid() || !m_id.is_response()) { if (!m_mid.valid() || !m_mid.is_response()) {
throw std::logic_error("handle does not point to a response");
}
else if (!self->awaits(m_id)) {
throw std::logic_error("response already received");
}
}
/*
template<typename Fun, typename... Args>
void apply(Fun& fun, Args&&... args) {
auto bhvr = match_expr_convert(std::forward<Args>(args)...);
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_id.valid() || !m_id.is_response()) {
throw std::logic_error("handle does not point to a response"); throw std::logic_error("handle does not point to a response");
} }
else if (!self->awaits(m_id)) { else if (!self->awaits(m_mid)) {
throw std::logic_error("response already received"); throw std::logic_error("response already received");
} }
fun(bhvr, m_id);
} }
*/
}; };
......
...@@ -50,7 +50,6 @@ ...@@ -50,7 +50,6 @@
#include "cppa/detail/abstract_actor.hpp" #include "cppa/detail/abstract_actor.hpp"
#include "cppa/detail/receive_policy.hpp" #include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/stacked_actor_mixin.hpp" #include "cppa/detail/stacked_actor_mixin.hpp"
#include "cppa/detail/recursive_queue_node.hpp" #include "cppa/detail/recursive_queue_node.hpp"
...@@ -98,7 +97,7 @@ class thread_mapped_actor : public detail::thread_mapped_actor_base { ...@@ -98,7 +97,7 @@ class thread_mapped_actor : public detail::thread_mapped_actor_base {
friend class self_type; // needs access to cleanup() friend class self_type; // needs access to cleanup()
friend class detail::behavior_stack; friend class detail::behavior_stack; // needs access to receive_node()
friend class detail::receive_policy; friend class detail::receive_policy;
typedef detail::thread_mapped_actor_base super; typedef detail::thread_mapped_actor_base super;
......
...@@ -60,6 +60,18 @@ class behavior_stack_help_iterator ...@@ -60,6 +60,18 @@ class behavior_stack_help_iterator
}; };
option<behavior&> behavior_stack::sync_handler(message_id_t expected_response) {
if (expected_response.valid()) {
auto e = m_elements.rend();
auto i = std::find_if(m_elements.rbegin(), e, [=](element_type& val) {
return val.second == expected_response;
});
if (i != e) return i->first;
}
return {};
}
void behavior_stack::pop_async_back() { void behavior_stack::pop_async_back() {
if (m_elements.empty()) { if (m_elements.empty()) {
// nothing to do // nothing to do
......
...@@ -158,10 +158,6 @@ void event_based_actor::quit(std::uint32_t reason) { ...@@ -158,10 +158,6 @@ void event_based_actor::quit(std::uint32_t reason) {
} }
} }
void event_based_actor::unbecome() {
m_bhvr_stack.pop_async_back();
}
scheduled_actor_type event_based_actor::impl_type() { scheduled_actor_type event_based_actor::impl_type() {
return event_based_impl; return event_based_impl;
} }
......
...@@ -167,4 +167,8 @@ message_id_t local_actor::send_timed_sync_message(actor* whom, ...@@ -167,4 +167,8 @@ message_id_t local_actor::send_timed_sync_message(actor* whom,
return mid; return mid;
} }
void local_actor::exec_behavior_stack() {
// default implementation does nothing
}
} // namespace cppa } // namespace cppa
...@@ -40,7 +40,6 @@ void scheduled_actor_dummy::link_to(const intrusive_ptr<actor>&) { } ...@@ -40,7 +40,6 @@ void scheduled_actor_dummy::link_to(const intrusive_ptr<actor>&) { }
void scheduled_actor_dummy::unlink_from(const intrusive_ptr<actor>&) { } void scheduled_actor_dummy::unlink_from(const intrusive_ptr<actor>&) { }
void scheduled_actor_dummy::detach(const attachable::token&) { } 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::do_become(behavior&&, bool) { } void scheduled_actor_dummy::do_become(behavior&&, bool) { }
void scheduled_actor_dummy::become_waiting_for(behavior&&, message_id_t) { } void scheduled_actor_dummy::become_waiting_for(behavior&&, message_id_t) { }
bool scheduled_actor_dummy::has_behavior() { return false; } bool scheduled_actor_dummy::has_behavior() { return false; }
......
...@@ -4,6 +4,12 @@ ...@@ -4,6 +4,12 @@
using namespace cppa; using namespace cppa;
using namespace cppa::placeholders; using namespace cppa::placeholders;
struct sync_mirror : sb_actor<sync_mirror> {
behavior init_state = (
others() >> [] { reply_tuple(self->last_dequeued()); }
);
};
struct popular_actor : event_based_actor { // popular actors have a buddy struct popular_actor : event_based_actor { // popular actors have a buddy
actor_ptr m_buddy; actor_ptr m_buddy;
popular_actor(const actor_ptr& buddy) : m_buddy(buddy) { } popular_actor(const actor_ptr& buddy) : m_buddy(buddy) { }
...@@ -111,6 +117,19 @@ struct D : popular_actor { ...@@ -111,6 +117,19 @@ struct D : popular_actor {
int main() { int main() {
CPPA_TEST(test__sync_send); CPPA_TEST(test__sync_send);
auto mirror = spawn<sync_mirror>();
bool continuation_called = false;
sync_send(mirror, 42).then(
[](int value) { CPPA_CHECK_EQUAL(42, value); }
)
.continue_with(
[&] { continuation_called = true; }
);
self->exec_behavior_stack();
CPPA_CHECK_EQUAL(true, continuation_called);
send(mirror, atom("EXIT"), exit_reason::user_defined);
await_all_others_done();
CPPA_CHECKPOINT();
auto await_success_message = [&] { auto await_success_message = [&] {
receive ( receive (
on(atom("success")) >> CPPA_CHECKPOINT_CB(), on(atom("success")) >> CPPA_CHECKPOINT_CB(),
......
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