Commit 86a74805 authored by Dominik Charousset's avatar Dominik Charousset

changed semantic of `quit`, deprecated reply

this patch changes the semantics of `quit` slightly,
i.e., `quit` always guarantees that all currently running message
handlers are executed and the given exit reason is not applied
until `on_exit` ran an did not set a new behavior;
this patch also deprecates `reply`, because the recommended way
of replying to a message is to return the value from the
message handler
parent b94c456b
...@@ -35,8 +35,16 @@ ...@@ -35,8 +35,16 @@
// if boost.context is not available on your platform // if boost.context is not available on your platform
//#define CPPA_DISABLE_CONTEXT_SWITCHING //#define CPPA_DISABLE_CONTEXT_SWITCHING
#if defined(__GNUC__) #if defined(__clang__)
# define CPPA_CLANG
# define CPPA_DEPRECATED __attribute__((__deprecated__))
#elif defined(__GNUC__)
# define CPPA_GCC # define CPPA_GCC
# define CPPA_DEPRECATED __attribute__((__deprecated__))
#elif defined(_MSC_VER)
# define CPPA_DEPRECATED __declspec(deprecated)
#else
# define CPPA_DEPRECATED
#endif #endif
#if defined(__APPLE__) #if defined(__APPLE__)
......
...@@ -41,21 +41,26 @@ ...@@ -41,21 +41,26 @@
#include "cppa/util/duration.hpp" #include "cppa/util/duration.hpp"
#include "cppa/util/type_traits.hpp" #include "cppa/util/type_traits.hpp"
namespace cppa {
typedef optional<any_tuple> bhvr_invoke_result;
} // namespace cppa
namespace cppa { namespace detail { namespace cppa { namespace detail {
struct optional_any_tuple_visitor { struct optional_any_tuple_visitor {
typedef optional<any_tuple> result_type; inline bhvr_invoke_result operator()() const { return any_tuple{}; }
inline result_type operator()() const { return any_tuple{}; } inline bhvr_invoke_result operator()(const none_t&) const { return none; }
inline result_type operator()(const none_t&) const { return none; }
template<typename T> template<typename T>
inline result_type operator()(T& value) const { inline bhvr_invoke_result operator()(T& value) const {
return make_any_tuple(std::move(value)); return make_any_tuple(std::move(value));
} }
template<typename... Ts> template<typename... Ts>
inline result_type operator()(cow_tuple<Ts...>& value) const { inline bhvr_invoke_result operator()(cow_tuple<Ts...>& value) const {
return any_tuple{std::move(value)}; return any_tuple{std::move(value)};
} }
inline result_type operator()(any_tuple& value) const { inline bhvr_invoke_result operator()(any_tuple& value) const {
return std::move(value); return std::move(value);
} }
}; };
...@@ -75,9 +80,9 @@ class behavior_impl : public ref_counted { ...@@ -75,9 +80,9 @@ class behavior_impl : public ref_counted {
inline behavior_impl(util::duration tout) : m_timeout(tout) { } inline behavior_impl(util::duration tout) : m_timeout(tout) { }
virtual optional<any_tuple> invoke(any_tuple&) = 0; virtual bhvr_invoke_result invoke(any_tuple&) = 0;
virtual optional<any_tuple> invoke(const any_tuple&) = 0; virtual bhvr_invoke_result invoke(const any_tuple&) = 0;
inline optional<any_tuple> invoke(any_tuple&& arg) { inline bhvr_invoke_result invoke(any_tuple&& arg) {
any_tuple tmp(std::move(arg)); any_tuple tmp(std::move(arg));
return invoke(tmp); return invoke(tmp);
} }
...@@ -96,12 +101,12 @@ class behavior_impl : public ref_counted { ...@@ -96,12 +101,12 @@ class behavior_impl : public ref_counted {
struct combinator : behavior_impl { struct combinator : behavior_impl {
pointer first; pointer first;
pointer second; pointer second;
optional<any_tuple> invoke(any_tuple& arg) { bhvr_invoke_result invoke(any_tuple& arg) {
auto res = first->invoke(arg); auto res = first->invoke(arg);
if (!res) return second->invoke(arg); if (!res) return second->invoke(arg);
return res; return res;
} }
optional<any_tuple> invoke(const any_tuple& arg) { bhvr_invoke_result invoke(const any_tuple& arg) {
auto res = first->invoke(arg); auto res = first->invoke(arg);
if (!res) return second->invoke(arg); if (!res) return second->invoke(arg);
return res; return res;
...@@ -142,8 +147,6 @@ class default_behavior_impl : public behavior_impl { ...@@ -142,8 +147,6 @@ class default_behavior_impl : public behavior_impl {
public: public:
typedef optional<any_tuple> invoke_result;
template<typename Expr> template<typename Expr>
default_behavior_impl(Expr&& expr, const timeout_definition<F>& d) default_behavior_impl(Expr&& expr, const timeout_definition<F>& d)
: super(d.timeout), m_expr(std::forward<Expr>(expr)), m_fun(d.handler) { } : super(d.timeout), m_expr(std::forward<Expr>(expr)), m_fun(d.handler) { }
...@@ -152,11 +155,11 @@ class default_behavior_impl : public behavior_impl { ...@@ -152,11 +155,11 @@ class default_behavior_impl : public behavior_impl {
default_behavior_impl(Expr&& expr, util::duration tout, F f) default_behavior_impl(Expr&& expr, util::duration tout, F f)
: super(tout), m_expr(std::forward<Expr>(expr)), m_fun(f) { } : super(tout), m_expr(std::forward<Expr>(expr)), m_fun(f) { }
invoke_result invoke(any_tuple& tup) { bhvr_invoke_result invoke(any_tuple& tup) {
return eval_res(m_expr(tup)); return eval_res(m_expr(tup));
} }
invoke_result invoke(const any_tuple& tup) { bhvr_invoke_result invoke(const any_tuple& tup) {
return eval_res(m_expr(tup)); return eval_res(m_expr(tup));
} }
...@@ -173,7 +176,7 @@ class default_behavior_impl : public behavior_impl { ...@@ -173,7 +176,7 @@ class default_behavior_impl : public behavior_impl {
private: private:
template<typename... Ts> template<typename... Ts>
typename std::enable_if<has_match_hint<Ts...>::value, invoke_result>::type typename std::enable_if<has_match_hint<Ts...>::value, bhvr_invoke_result>::type
eval_res(optional_variant<Ts...>&& res) { eval_res(optional_variant<Ts...>&& res) {
if (res) { if (res) {
if (res.template is<match_hint>()) { if (res.template is<match_hint>()) {
...@@ -188,7 +191,7 @@ class default_behavior_impl : public behavior_impl { ...@@ -188,7 +191,7 @@ class default_behavior_impl : public behavior_impl {
} }
template<typename... Ts> template<typename... Ts>
typename std::enable_if<!has_match_hint<Ts...>::value, invoke_result>::type typename std::enable_if<!has_match_hint<Ts...>::value, bhvr_invoke_result>::type
eval_res(optional_variant<Ts...>&& res) { eval_res(optional_variant<Ts...>&& res) {
return apply_visitor(optional_any_tuple_visitor{}, res); return apply_visitor(optional_any_tuple_visitor{}, res);
} }
......
...@@ -54,7 +54,7 @@ class event_based_actor_impl : public event_based_actor { ...@@ -54,7 +54,7 @@ class event_based_actor_impl : public event_based_actor {
void init() { apply(m_init); } void init() { apply(m_init); }
void on_exit() { void on_exit() override {
typedef typename util::get_callable_trait<CleanupFun>::arg_types arg_types; typedef typename util::get_callable_trait<CleanupFun>::arg_types arg_types;
std::integral_constant<size_t, util::tl_size<arg_types>::value> token; std::integral_constant<size_t, util::tl_size<arg_types>::value> token;
on_exit_impl(m_on_exit, token); on_exit_impl(m_on_exit, token);
......
...@@ -47,6 +47,7 @@ namespace cppa { ...@@ -47,6 +47,7 @@ namespace cppa {
/** /**
* @brief Base class for all event-based actor implementations. * @brief Base class for all event-based actor implementations.
* @extends scheduled_actor
*/ */
class event_based_actor : public extend<scheduled_actor>::with<stackless> { class event_based_actor : public extend<scheduled_actor>::with<stackless> {
...@@ -63,8 +64,6 @@ class event_based_actor : public extend<scheduled_actor>::with<stackless> { ...@@ -63,8 +64,6 @@ class event_based_actor : public extend<scheduled_actor>::with<stackless> {
*/ */
virtual void init() = 0; virtual void init() = 0;
virtual void quit(std::uint32_t reason = exit_reason::normal);
scheduled_actor_type impl_type(); scheduled_actor_type impl_type();
static intrusive_ptr<event_based_actor> from(std::function<void()> fun); static intrusive_ptr<event_based_actor> from(std::function<void()> fun);
......
...@@ -62,6 +62,24 @@ class message_future; ...@@ -62,6 +62,24 @@ class message_future;
class local_scheduler; class local_scheduler;
class sync_handle_helper; class sync_handle_helper;
#ifdef CPPA_DOCUMENTATION
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* discard the current behavior.
* @relates local_actor
*/
constexpr auto discard_behavior;
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* keep the current behavior available.
* @relates local_actor
*/
constexpr auto keep_behavior;
#else
template<bool DiscardBehavior> template<bool DiscardBehavior>
struct behavior_policy { static constexpr bool discard_old = DiscardBehavior; }; struct behavior_policy { static constexpr bool discard_old = DiscardBehavior; };
...@@ -74,20 +92,12 @@ struct is_behavior_policy<behavior_policy<DiscardBehavior>> : std::true_type { } ...@@ -74,20 +92,12 @@ struct is_behavior_policy<behavior_policy<DiscardBehavior>> : std::true_type { }
typedef behavior_policy<false> keep_behavior_t; typedef behavior_policy<false> keep_behavior_t;
typedef behavior_policy<true > discard_behavior_t; typedef behavior_policy<true > discard_behavior_t;
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* discard the current behavior.
* @relates local_actor
*/
constexpr discard_behavior_t discard_behavior = discard_behavior_t{}; constexpr discard_behavior_t discard_behavior = discard_behavior_t{};
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* keep the current behavior available.
* @relates local_actor
*/
constexpr keep_behavior_t keep_behavior = keep_behavior_t{}; constexpr keep_behavior_t keep_behavior = keep_behavior_t{};
#endif
/** /**
* @brief Base class for local running Actors. * @brief Base class for local running Actors.
* @extends actor * @extends actor
...@@ -119,16 +129,30 @@ class local_actor : public extend<actor>::with<memory_cached> { ...@@ -119,16 +129,30 @@ class local_actor : public extend<actor>::with<memory_cached> {
void leave(const group_ptr& what); void leave(const group_ptr& what);
/** /**
* @brief Finishes execution of this actor. * @brief Finishes execution of this actor after any currently running
* message handler is done.
*
* This member function clear the behavior stack of the running actor
* and invokes {@link on_exit()}. The actors does not finish execution
* if the implementation of {@link on_exit()} sets a new behavior.
* When setting a new behavior in {@link on_exit()}, one has to make sure
* to not produce an infinite recursion.
*
* If {@link on_exit()} did not set a new behavior, the actor sends an
* exit message to all of its linked actors, sets its state to @c exited
* and finishes execution.
* *
* Causes this actor to send an exit message to all of its
* linked actors, sets its state to @c exited and finishes execution.
* @param reason Exit reason that will be send to * @param reason Exit reason that will be send to
* linked actors and monitors. * linked actors and monitors. Can be queried using
* {@link planned_exit_reason()}, e.g., from inside
* {@link on_exit()}.
* @note Throws {@link actor_exited} to unwind the stack * @note Throws {@link actor_exited} to unwind the stack
* when called in context-switching or thread-based actors. * when called in context-switching or thread-based actors.
* @warning This member function throws imeediately in thread-based actors
* that do not use the behavior stack, i.e., actors that use
* blocking API calls such as {@link receive()}.
*/ */
virtual void quit(std::uint32_t reason = exit_reason::normal) = 0; virtual void quit(std::uint32_t reason = exit_reason::normal);
/** /**
* @brief Checks whether this actor traps exit messages. * @brief Checks whether this actor traps exit messages.
...@@ -314,6 +338,10 @@ class local_actor : public extend<actor>::with<memory_cached> { ...@@ -314,6 +338,10 @@ class local_actor : public extend<actor>::with<memory_cached> {
void debug_name(std::string str); void debug_name(std::string str);
inline std::uint32_t planned_exit_reason() const;
inline void planned_exit_reason(std::uint32_t value);
protected: protected:
inline void remove_handler(message_id id); inline void remove_handler(message_id id);
...@@ -354,12 +382,16 @@ class local_actor : public extend<actor>::with<memory_cached> { ...@@ -354,12 +382,16 @@ class local_actor : public extend<actor>::with<memory_cached> {
// allows actors to keep previous behaviors and enables unbecome() // allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack; detail::behavior_stack m_bhvr_stack;
// set by quit
std::uint32_t m_planned_exit_reason;
/** @endcond */
private: private:
std::function<void()> m_sync_failure_handler; std::function<void()> m_sync_failure_handler;
std::function<void()> m_sync_timeout_handler; std::function<void()> m_sync_timeout_handler;
}; };
/** /**
...@@ -373,7 +405,9 @@ typedef intrusive_ptr<local_actor> local_actor_ptr; ...@@ -373,7 +405,9 @@ typedef intrusive_ptr<local_actor> local_actor_ptr;
* inline and template member function implementations * * inline and template member function implementations *
******************************************************************************/ ******************************************************************************/
void local_actor::dequeue(behavior&& bhvr) { /** @cond PRIVATE */
inline void local_actor::dequeue(behavior&& bhvr) {
behavior tmp{std::move(bhvr)}; behavior tmp{std::move(bhvr)};
dequeue(tmp); dequeue(tmp);
} }
...@@ -456,6 +490,16 @@ inline void local_actor::remove_handler(message_id id) { ...@@ -456,6 +490,16 @@ inline void local_actor::remove_handler(message_id id) {
m_bhvr_stack.erase(id); m_bhvr_stack.erase(id);
} }
inline std::uint32_t local_actor::planned_exit_reason() const {
return m_planned_exit_reason;
}
inline void local_actor::planned_exit_reason(std::uint32_t value) {
m_planned_exit_reason = value;
}
/** @endcond */
} // namespace cppa } // namespace cppa
#endif // CPPA_CONTEXT_HPP #endif // CPPA_CONTEXT_HPP
...@@ -81,6 +81,25 @@ inline void send_tuple(destination_header hdr, any_tuple what) { ...@@ -81,6 +81,25 @@ inline void send_tuple(destination_header hdr, any_tuple what) {
else fhdr.deliver(std::move(what)); else fhdr.deliver(std::move(what));
} }
/**
* @brief Sends @p what to the receiver specified in @p hdr.
*/
template<typename... Signatures, typename... Ts>
void send(const typed_actor_ptr<Signatures...>& whom, Ts&&... what) {
static constexpr int input_pos = util::tl_find_if<
util::type_list<Signatures...>,
detail::input_is<util::type_list<
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Ts
>::type
>::type...
>>::template eval
>::value;
static_assert(input_pos >= 0, "typed actor does not support given input");
send(whom.unbox(), std::forward<Ts>(what)...);
}
/** /**
* @brief Sends <tt>{what...}</tt> to the receiver specified in @p hdr. * @brief Sends <tt>{what...}</tt> to the receiver specified in @p hdr.
* @pre <tt>sizeof...(Ts) > 0</tt> * @pre <tt>sizeof...(Ts) > 0</tt>
...@@ -287,7 +306,7 @@ message_future timed_sync_send(actor_ptr whom, ...@@ -287,7 +306,7 @@ message_future timed_sync_send(actor_ptr whom,
* @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 content as a tuple. * @param what Message content as a tuple.
*/ */
inline void reply_tuple(any_tuple what) { CPPA_DEPRECATED inline void reply_tuple(any_tuple what) {
self->reply_message(std::move(what)); self->reply_message(std::move(what));
} }
...@@ -296,7 +315,7 @@ inline void reply_tuple(any_tuple what) { ...@@ -296,7 +315,7 @@ inline void reply_tuple(any_tuple what) {
* @param what Message elements. * @param what Message elements.
*/ */
template<typename... Ts> template<typename... Ts>
inline void reply(Ts&&... what) { CPPA_DEPRECATED inline void reply(Ts&&... what) {
self->reply_message(make_any_tuple(std::forward<Ts>(what)...)); self->reply_message(make_any_tuple(std::forward<Ts>(what)...));
} }
......
...@@ -64,43 +64,58 @@ class stacked : public Base { ...@@ -64,43 +64,58 @@ class stacked : public Base {
static constexpr auto receive_flag = detail::rp_nestable; static constexpr auto receive_flag = detail::rp_nestable;
virtual void dequeue(behavior& bhvr) override { void dequeue(behavior& bhvr) override {
++m_nested_count;
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr); m_recv_policy.receive(util::dptr<Subtype>(this), bhvr);
--m_nested_count;
check_quit();
} }
virtual void dequeue_response(behavior& bhvr, message_id request_id) override { void dequeue_response(behavior& bhvr, message_id request_id) override {
++m_nested_count;
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr, request_id); m_recv_policy.receive(util::dptr<Subtype>(this), bhvr, request_id);
--m_nested_count;
check_quit();
} }
virtual void run() { virtual void run() {
auto dthis = util::dptr<Subtype>(this); exec_behavior_stack();
if (!dthis->m_bhvr_stack.empty()) dthis->exec_behavior_stack();
if (m_behavior) m_behavior(); if (m_behavior) m_behavior();
auto dthis = util::dptr<Subtype>(this);
auto rsn = dthis->planned_exit_reason();
dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn);
} }
inline void set_behavior(std::function<void()> fun) { inline void set_behavior(std::function<void()> fun) {
m_behavior = std::move(fun); m_behavior = std::move(fun);
} }
virtual void quit(std::uint32_t reason) override { void exec_behavior_stack() override {
this->cleanup(reason); m_in_bhvr_loop = true;
throw actor_exited(reason); auto dthis = util::dptr<Subtype>(this);
if (!stack_empty() && !m_nested_count) {
while (!stack_empty()) {
++m_nested_count;
dthis->m_bhvr_stack.exec(m_recv_policy, dthis);
--m_nested_count;
check_quit();
} }
}
virtual void exec_behavior_stack() override { m_in_bhvr_loop = false;
this->m_bhvr_stack.exec(m_recv_policy, util::dptr<Subtype>(this));
} }
protected: protected:
template<typename... Ts> template<typename... Ts>
stacked(Ts&&... args) : Base(std::forward<Ts>(args)...) { } stacked(Ts&&... args) : Base(std::forward<Ts>(args)...)
, m_in_bhvr_loop(false)
, m_nested_count(0) { }
virtual void do_become(behavior&& bhvr, bool discard_old) override { void do_become(behavior&& bhvr, bool discard_old) override {
become_impl(std::move(bhvr), discard_old, message_id()); become_impl(std::move(bhvr), discard_old, message_id());
} }
virtual void become_waiting_for(behavior bhvr, message_id mid) override { void become_waiting_for(behavior bhvr, message_id mid) override {
become_impl(std::move(bhvr), false, mid); become_impl(std::move(bhvr), false, mid);
} }
...@@ -114,6 +129,10 @@ class stacked : public Base { ...@@ -114,6 +129,10 @@ class stacked : public Base {
private: private:
inline bool stack_empty() {
return util::dptr<Subtype>(this)->m_bhvr_stack.empty();
}
void become_impl(behavior&& bhvr, bool discard_old, message_id mid) { void become_impl(behavior&& bhvr, bool discard_old, message_id mid) {
auto dthis = util::dptr<Subtype>(this); auto dthis = util::dptr<Subtype>(this);
if (bhvr.timeout().valid()) { if (bhvr.timeout().valid()) {
...@@ -126,6 +145,27 @@ class stacked : public Base { ...@@ -126,6 +145,27 @@ class stacked : public Base {
dthis->m_bhvr_stack.push_back(std::move(bhvr), mid); dthis->m_bhvr_stack.push_back(std::move(bhvr), mid);
} }
void check_quit() {
// do nothing if other message handlers are still running
if (m_nested_count > 0) return;
auto dthis = util::dptr<Subtype>(this);
auto rsn = dthis->planned_exit_reason();
if (rsn != exit_reason::not_exited) {
dthis->on_exit();
if (stack_empty()) {
dthis->cleanup(rsn);
throw actor_exited(rsn);
}
else {
dthis->planned_exit_reason(exit_reason::not_exited);
if (!m_in_bhvr_loop) exec_behavior_stack();
}
}
}
bool m_in_bhvr_loop;
size_t m_nested_count;
}; };
} // namespace cppa } // namespace cppa
......
...@@ -89,9 +89,17 @@ class threaded : public Base { ...@@ -89,9 +89,17 @@ class threaded : public Base {
void run_detached() { void run_detached() {
auto dthis = util::dptr<Subtype>(this); auto dthis = util::dptr<Subtype>(this);
dthis->init(); dthis->init();
if (dthis->planned_exit_reason() != exit_reason::not_exited) {
// init() did indeed call quit() for some reason
dthis->on_exit();
}
while (!dthis->m_bhvr_stack.empty()) {
dthis->m_bhvr_stack.exec(dthis->m_recv_policy, dthis); dthis->m_bhvr_stack.exec(dthis->m_recv_policy, dthis);
dthis->on_exit(); dthis->on_exit();
} }
auto rsn = dthis->planned_exit_reason();
dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn);
}
inline void initialized(bool value) { inline void initialized(bool value) {
m_initialized = value; m_initialized = value;
......
...@@ -49,6 +49,9 @@ spawn_typed(const match_expr<Ts...>&); ...@@ -49,6 +49,9 @@ spawn_typed(const match_expr<Ts...>&);
template<typename... Ts> template<typename... Ts>
void send_exit(const typed_actor_ptr<Ts...>&, std::uint32_t); void send_exit(const typed_actor_ptr<Ts...>&, std::uint32_t);
template<typename... Signatures, typename... Ts>
void send(const typed_actor_ptr<Signatures...>&, Ts&&...);
template<typename... Ts, typename... Us> template<typename... Ts, typename... Us>
typed_message_future< typed_message_future<
typename detail::deduce_output_type< typename detail::deduce_output_type<
...@@ -77,6 +80,10 @@ class typed_actor_ptr { ...@@ -77,6 +80,10 @@ class typed_actor_ptr {
template<typename... Ts> template<typename... Ts>
friend void send_exit(const typed_actor_ptr<Ts...>&, std::uint32_t); friend void send_exit(const typed_actor_ptr<Ts...>&, std::uint32_t);
template<typename... Sigs, typename... Ts>
friend void send(const typed_actor_ptr<Sigs...>&, Ts&&...);
template<typename... Ts, typename... Us> template<typename... Ts, typename... Us>
friend typed_message_future< friend typed_message_future<
typename detail::deduce_output_type< typename detail::deduce_output_type<
...@@ -114,7 +121,9 @@ class typed_actor_ptr { ...@@ -114,7 +121,9 @@ class typed_actor_ptr {
private: private:
/** @cond PRIVATE */
const actor_ptr& unbox() const { return m_ptr; } const actor_ptr& unbox() const { return m_ptr; }
/** @endcond */
typed_actor_ptr(actor_ptr ptr) : m_ptr(std::move(ptr)) { } typed_actor_ptr(actor_ptr ptr) : m_ptr(std::move(ptr)) { }
......
...@@ -129,7 +129,7 @@ class base_actor : public event_based_actor { ...@@ -129,7 +129,7 @@ class base_actor : public event_based_actor {
return ::print(m_color.c_str(), m_name.c_str()); return ::print(m_color.c_str(), m_name.c_str());
} }
virtual void on_exit() override { void on_exit() override {
print() << "on_exit" << color::reset_endl; print() << "on_exit" << color::reset_endl;
} }
...@@ -230,15 +230,14 @@ class curl_worker : public base_actor { ...@@ -230,15 +230,14 @@ class curl_worker : public base_actor {
curl_easy_setopt(m_curl, CURLOPT_WRITEFUNCTION, &curl_worker::cb); curl_easy_setopt(m_curl, CURLOPT_WRITEFUNCTION, &curl_worker::cb);
curl_easy_setopt(m_curl, CURLOPT_NOSIGNAL, 1); curl_easy_setopt(m_curl, CURLOPT_NOSIGNAL, 1);
become ( become (
on(atom("read"), arg_match) >> [=](const std::string& file_name, on(atom("read"), arg_match)
uint64_t offset, >> [=](const std::string& fname, uint64_t offset, uint64_t range)
uint64_t range) { -> cow_tuple<atom_value, util::buffer> {
print() << "read" << color::reset_endl; print() << "read" << color::reset_endl;
bool request_done = false; for (;;) {
while (!request_done) {
m_buf.clear(); m_buf.clear();
// set URL // set URL
curl_easy_setopt(m_curl, CURLOPT_URL, file_name.c_str()); curl_easy_setopt(m_curl, CURLOPT_URL, fname.c_str());
// set range // set range
std::ostringstream oss; std::ostringstream oss;
oss << offset << "-" << range; oss << offset << "-" << range;
...@@ -270,11 +269,9 @@ class curl_worker : public base_actor { ...@@ -270,11 +269,9 @@ class curl_worker : public base_actor {
<< " bytes with 'HTTP RETURN CODE': " << " bytes with 'HTTP RETURN CODE': "
<< hc << hc
<< color::reset_endl; << color::reset_endl;
reply(atom("reply"), m_buf);
// tell parent that this worker is done // tell parent that this worker is done
send(m_parent, atom("finished")); send(m_parent, atom("finished"));
request_done = true; return make_cow_tuple(atom("reply"), m_buf);
break;
case 404: // file does not exist case 404: // file does not exist
print() << "http error: download failed with " print() << "http error: download failed with "
"'HTTP RETURN CODE': 404 (file does " "'HTTP RETURN CODE': 404 (file does "
...@@ -282,14 +279,14 @@ class curl_worker : public base_actor { ...@@ -282,14 +279,14 @@ class curl_worker : public base_actor {
<< color::reset_endl; << color::reset_endl;
} }
} }
}
// avoid 100% cpu utilization if remote side is not accessible // avoid 100% cpu utilization if remote side is not accessible
if (!request_done) usleep(100000); // 100000 == 100ms usleep(100000); // 100000 == 100ms
}
} }
); );
} }
virtual void on_exit() override { void on_exit() override {
curl_easy_cleanup(m_curl); curl_easy_cleanup(m_curl);
print() << "on_exit" << color::reset_endl; print() << "on_exit" << color::reset_endl;
} }
......
...@@ -9,13 +9,13 @@ void mirror() { ...@@ -9,13 +9,13 @@ void mirror() {
// wait for messages // wait for messages
become ( become (
// invoke this lambda expression if we receive a string // invoke this lambda expression if we receive a string
on_arg_match >> [](const string& what) { on_arg_match >> [](const string& what) -> string {
// prints "Hello World!" via aout (thread-safe cout wrapper) // prints "Hello World!" via aout (thread-safe cout wrapper)
aout << what << endl; aout << what << endl;
// replies "!dlroW olleH"
reply(string(what.rbegin(), what.rend()));
// terminates this actor ('become' otherwise loops forever) // terminates this actor ('become' otherwise loops forever)
self->quit(); self->quit();
// reply "!dlroW olleH"
return string(what.rbegin(), what.rend());
} }
); );
} }
......
...@@ -21,10 +21,10 @@ void blocking_math_fun() { ...@@ -21,10 +21,10 @@ void blocking_math_fun() {
// - on<atom("plus"), int, int>() // - on<atom("plus"), int, int>()
// - on(atom("plus"), val<int>, val<int>) // - on(atom("plus"), val<int>, val<int>)
on(atom("plus"), arg_match) >> [](int a, int b) { on(atom("plus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a + b); return make_cow_tuple(atom("result"), a + b);
}, },
on(atom("minus"), arg_match) >> [](int a, int b) { on(atom("minus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a - b); return make_cow_tuple(atom("result"), a - b);
}, },
on(atom("quit")) >> [&]() { on(atom("quit")) >> [&]() {
// note: this actor uses the blocking API, hence self->quit() // note: this actor uses the blocking API, hence self->quit()
...@@ -38,10 +38,10 @@ void calculator() { ...@@ -38,10 +38,10 @@ void calculator() {
// execute this behavior until actor terminates // execute this behavior until actor terminates
become ( become (
on(atom("plus"), arg_match) >> [](int a, int b) { on(atom("plus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a + b); return make_cow_tuple(atom("result"), a + b);
}, },
on(atom("minus"), arg_match) >> [](int a, int b) { on(atom("minus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a - b); return make_cow_tuple(atom("result"), a - b);
}, },
on(atom("quit")) >> [] { on(atom("quit")) >> [] {
// terminate actor with normal exit reason // terminate actor with normal exit reason
......
...@@ -63,7 +63,7 @@ void ping(size_t num_pings) { ...@@ -63,7 +63,7 @@ void ping(size_t num_pings) {
become ( become (
on(atom("pong"), arg_match) >> [=](int value) { on(atom("pong"), arg_match) >> [=](int value) {
aout << "pong: " << value << endl; aout << "pong: " << value << endl;
if (++*count >= num_pings) self->quit(); if (++*count >= num_pings) self->quit_later();
else reply(atom("ping"), value + 1); else reply(atom("ping"), value + 1);
} }
); );
......
...@@ -17,9 +17,7 @@ class foo { ...@@ -17,9 +17,7 @@ class foo {
public: public:
foo() : m_a(0), m_b(0) { } foo(int a0 = 0, int b0 = 0) : m_a(a0), m_b(b0) { }
foo(int a0, int b0) : m_a(a0), m_b(b0) { }
foo(const foo&) = default; foo(const foo&) = default;
...@@ -64,4 +62,3 @@ int main(int, char**) { ...@@ -64,4 +62,3 @@ int main(int, char**) {
shutdown(); shutdown();
return 0; return 0;
} }
...@@ -202,6 +202,4 @@ void actor::cleanup(std::uint32_t reason) { ...@@ -202,6 +202,4 @@ void actor::cleanup(std::uint32_t reason) {
} }
} }
} // namespace cppa } // namespace cppa
...@@ -47,17 +47,17 @@ class continuation_decorator : public detail::behavior_impl { ...@@ -47,17 +47,17 @@ class continuation_decorator : public detail::behavior_impl {
} }
template<typename T> template<typename T>
inline optional<any_tuple> invoke_impl(T& tup) { inline bhvr_invoke_result invoke_impl(T& tup) {
auto res = m_decorated->invoke(tup); auto res = m_decorated->invoke(tup);
if (res) return m_fun(*res); if (res) return m_fun(*res);
return none; return none;
} }
optional<any_tuple> invoke(any_tuple& tup) { bhvr_invoke_result invoke(any_tuple& tup) {
return invoke_impl(tup); return invoke_impl(tup);
} }
optional<any_tuple> invoke(const any_tuple& tup) { bhvr_invoke_result invoke(const any_tuple& tup) {
return invoke_impl(tup); return invoke_impl(tup);
} }
......
...@@ -33,7 +33,7 @@ ...@@ -33,7 +33,7 @@
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/detail/demangle.hpp" #include "cppa/detail/demangle.hpp"
#ifdef CPPA_GCC #if defined(CPPA_GCC) || defined(CPPA_CLANG)
#include <cxxabi.h> #include <cxxabi.h>
#endif #endif
...@@ -79,7 +79,7 @@ std::string demangle(const char* decorated) { ...@@ -79,7 +79,7 @@ std::string demangle(const char* decorated) {
} }
} }
free(undecorated); free(undecorated);
# ifdef __clang__ # ifdef CPPA_CLANG
// replace "std::__1::" with "std::" (fixes strange clang names) // replace "std::__1::" with "std::" (fixes strange clang names)
std::string needle = "std::__1::"; std::string needle = "std::__1::";
std::string fixed_string = "std::"; std::string fixed_string = "std::";
......
...@@ -59,13 +59,17 @@ class default_scheduled_actor : public event_based_actor { ...@@ -59,13 +59,17 @@ class default_scheduled_actor : public event_based_actor {
m_initialized = true; m_initialized = true;
m_fun(); m_fun();
if (m_bhvr_stack.empty()) { if (m_bhvr_stack.empty()) {
if (exit_reason() == exit_reason::not_exited) quit(exit_reason::normal); if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::normal);
}
else {
set_state(actor_state::done); set_state(actor_state::done);
m_bhvr_stack.clear(); m_bhvr_stack.clear();
m_bhvr_stack.cleanup(); m_bhvr_stack.cleanup();
on_exit(); on_exit();
next.swap(m_chained_actor); next.swap(m_chained_actor);
set_state(actor_state::done); set_state(actor_state::done);
}
return resume_result::actor_done; return resume_result::actor_done;
} }
} }
...@@ -94,15 +98,26 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) { ...@@ -94,15 +98,26 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
CPPA_REQUIRE( state() == actor_state::ready CPPA_REQUIRE( state() == actor_state::ready
|| state() == actor_state::pending); || state() == actor_state::pending);
scoped_self_setter sss{this}; scoped_self_setter sss{this};
auto done_cb = [&]() { auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
if (exit_reason() == exit_reason::not_exited) quit(exit_reason::normal); if (exit_reason() == exit_reason::not_exited) {
if (planned_exit_reason() == exit_reason::not_exited) {
planned_exit_reason(exit_reason::normal);
}
on_exit();
if (!m_bhvr_stack.empty()) {
planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior
}
cleanup(planned_exit_reason());
}
set_state(actor_state::done); set_state(actor_state::done);
m_bhvr_stack.clear(); m_bhvr_stack.clear();
m_bhvr_stack.cleanup(); m_bhvr_stack.cleanup();
on_exit(); on_exit();
CPPA_REQUIRE(next_job == nullptr); CPPA_REQUIRE(next_job == nullptr);
next_job.swap(m_chained_actor); next_job.swap(m_chained_actor);
return true;
}; };
CPPA_REQUIRE(next_job == nullptr); CPPA_REQUIRE(next_job == nullptr);
try { try {
...@@ -150,9 +165,8 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) { ...@@ -150,9 +165,8 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
"set actor with ID " "set actor with ID "
<< m_chained_actor->id() << m_chained_actor->id()
<< " as successor"); << " as successor");
if (m_bhvr_stack.empty()) { if (m_bhvr_stack.empty() && done_cb()) {
CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty"); CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty");
done_cb();
return resume_result::actor_done; return resume_result::actor_done;
} }
m_bhvr_stack.cleanup(); m_bhvr_stack.cleanup();
...@@ -185,27 +199,6 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) { ...@@ -185,27 +199,6 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
return resume_result::actor_done; return resume_result::actor_done;
} }
void event_based_actor::quit(std::uint32_t reason) {
CPPA_LOG_TRACE("reason = " << reason
<< ", class " << detail::demangle(typeid(*this)));
cleanup(reason);
m_bhvr_stack.clear();
if (reason == exit_reason::unallowed_function_call) {
CPPA_LOG_WARNING("actor tried to use a blocking function");
// when using receive(), the non-blocking nature of event-based
// actors breaks any assumption the user has about his code,
// in particular, receive_loop() is a deadlock when not throwing
// an exception here
aout << "*** warning: event-based actor killed because it tried to "
"use receive()\n";
throw actor_exited(reason);
}
}
scheduled_actor_type event_based_actor::impl_type() { scheduled_actor_type event_based_actor::impl_type() {
return event_based_impl; return event_based_impl;
} }
......
...@@ -91,7 +91,7 @@ struct group_nameserver : event_based_actor { ...@@ -91,7 +91,7 @@ struct group_nameserver : event_based_actor {
void init() { void init() {
become ( become (
on(atom("GET_GROUP"), arg_match) >> [](const std::string& name) { on(atom("GET_GROUP"), arg_match) >> [](const std::string& name) {
reply(atom("GROUP"), group::get("local", name)); return make_cow_tuple(atom("GROUP"), group::get("local", name));
}, },
on(atom("SHUTDOWN")) >> [=] { on(atom("SHUTDOWN")) >> [=] {
quit(); quit();
......
...@@ -76,7 +76,8 @@ constexpr const char* s_default_debug_name = "actor"; ...@@ -76,7 +76,8 @@ constexpr const char* s_default_debug_name = "actor";
local_actor::local_actor(bool sflag) local_actor::local_actor(bool sflag)
: m_chaining(sflag), m_trap_exit(false) : m_chaining(sflag), m_trap_exit(false)
, m_is_scheduled(sflag), m_dummy_node(), m_current_node(&m_dummy_node) { , m_is_scheduled(sflag), m_dummy_node(), m_current_node(&m_dummy_node)
, m_planned_exit_reason(exit_reason::not_exited) {
# ifdef CPPA_DEBUG_MODE # ifdef CPPA_DEBUG_MODE
new (&m_debug_name) std::string (std::to_string(m_id) + "@local"); new (&m_debug_name) std::string (std::to_string(m_id) + "@local");
# endif // CPPA_DEBUG_MODE # endif // CPPA_DEBUG_MODE
...@@ -189,4 +190,24 @@ void local_actor::dequeue_response(behavior&, message_id) { ...@@ -189,4 +190,24 @@ void local_actor::dequeue_response(behavior&, message_id) {
quit(exit_reason::unallowed_function_call); quit(exit_reason::unallowed_function_call);
} }
void local_actor::quit(std::uint32_t reason) {
CPPA_LOG_TRACE("reason = " << reason
<< ", class " << detail::demangle(typeid(*this)));
if (reason == exit_reason::unallowed_function_call) {
// this is the only reason that causes an exception
cleanup(reason);
m_bhvr_stack.clear();
CPPA_LOG_WARNING("actor tried to use a blocking function");
// when using receive(), the non-blocking nature of event-based
// actors breaks any assumption the user has about his code,
// in particular, receive_loop() is a deadlock when not throwing
// an exception here
aout << "*** warning: event-based actor killed because it tried to "
"use receive()\n";
throw actor_exited(reason);
}
m_bhvr_stack.clear();
planned_exit_reason(reason);
}
} // namespace cppa } // namespace cppa
...@@ -363,6 +363,14 @@ void test_typed_actors() { ...@@ -363,6 +363,14 @@ void test_typed_actors() {
); );
} }
); );
// check async messages
send(ptr0_float, 4.0f);
receive(
on_arg_match >> [](float f) {
CPPA_CHECK_EQUAL(f, 4.0f / 2.0f);
}
);
// check sync messages
sync_send(ptr0_float, 4.0f).await( sync_send(ptr0_float, 4.0f).await(
[](float f) { [](float f) {
CPPA_CHECK_EQUAL(f, 4.0f / 2.0f); CPPA_CHECK_EQUAL(f, 4.0f / 2.0f);
......
...@@ -109,6 +109,11 @@ struct D : popular_actor { ...@@ -109,6 +109,11 @@ struct D : popular_actor {
reply_to(handle, last_dequeued()); reply_to(handle, last_dequeued());
quit(); quit();
}); });
/*/
return sync_send_tuple(buddy(), last_dequeued()).then([=]() -> any_tuple {
quit();
return last_dequeued();
});//*/
} }
); );
} }
......
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