Commit 94f186a5 authored by Dominik Charousset's avatar Dominik Charousset

Refactor blocking actors to not use exceptions

parent e4855afe
...@@ -40,11 +40,20 @@ void caf_main(actor_system& system) { ...@@ -40,11 +40,20 @@ void caf_main(actor_system& system) {
auto serv = system.spawn(server); auto serv = system.spawn(server);
auto worker = system.spawn(client, serv); auto worker = system.spawn(client, serv);
scoped_actor self{system}; scoped_actor self{system};
self->request(serv, std::chrono::seconds(10), self->request(serv, std::chrono::seconds(10), request_atom::value).receive(
request_atom::value).receive([&](response_atom) { [&](response_atom) {
aout(self) << "received response from " aout(self) << "received response from "
<< (self->current_sender() == worker ? "worker\n" : "server\n"); << (self->current_sender() == worker ? "worker\n"
}); : "server\n");
},
[&](error& err) {
aout(self) << "received error "
<< system.render(err)
<< " from "
<< (self->current_sender() == worker ? "worker\n"
: "server\n");
}
);
self->send_exit(serv, exit_reason::user_shutdown); self->send_exit(serv, exit_reason::user_shutdown);
} }
......
...@@ -42,12 +42,19 @@ behavior calculator_fun(event_based_actor*) { ...@@ -42,12 +42,19 @@ behavior calculator_fun(event_based_actor*) {
// function-based, dynamically typed, blocking API // function-based, dynamically typed, blocking API
void blocking_calculator_fun(blocking_actor* self) { void blocking_calculator_fun(blocking_actor* self) {
self->receive_loop ( bool running = true;
self->receive_while(running) (
[](add_atom, int a, int b) { [](add_atom, int a, int b) {
return a + b; return a + b;
}, },
[](sub_atom, int a, int b) { [](sub_atom, int a, int b) {
return a - b; return a - b;
},
[&](exit_msg& em) {
if (em.reason) {
self->fail_state(std::move(em.reason));
running = false;
}
} }
); );
} }
...@@ -107,6 +114,11 @@ void tester(scoped_actor&) { ...@@ -107,6 +114,11 @@ void tester(scoped_actor&) {
// tests a calculator instance // tests a calculator instance
template <class Handle, class... Ts> template <class Handle, class... Ts>
void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) { void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) {
auto handle_err = [&](const error& err) {
aout(self) << "AUT (actor under test) failed: "
<< self->system().render(err) << endl;
throw std::runtime_error("AUT responded with an error");
};
// first test: x + y = z // first test: x + y = z
self->request(hdl, infinite, add_atom::value, x, y).receive( self->request(hdl, infinite, add_atom::value, x, y).receive(
[&](int res1) { [&](int res1) {
...@@ -115,14 +127,11 @@ void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) { ...@@ -115,14 +127,11 @@ void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) {
self->request(hdl, infinite, sub_atom::value, x, y).receive( self->request(hdl, infinite, sub_atom::value, x, y).receive(
[&](int res2) { [&](int res2) {
aout(self) << x << " - " << y << " = " << res2 << endl; aout(self) << x << " - " << y << " = " << res2 << endl;
} },
handle_err
); );
}, },
[&](const error& err) { handle_err
aout(self) << "AUT (actor under test) failed: "
<< self->system().render(err) << endl;
self->quit(exit_reason::user_shutdown);
}
); );
tester(self, std::forward<Ts>(xs)...); tester(self, std::forward<Ts>(xs)...);
} }
......
...@@ -52,9 +52,15 @@ void multiplexed_testee(event_based_actor* self, vector<cell> cells) { ...@@ -52,9 +52,15 @@ void multiplexed_testee(event_based_actor* self, vector<cell> cells) {
void blocking_testee(blocking_actor* self, vector<cell> cells) { void blocking_testee(blocking_actor* self, vector<cell> cells) {
for (auto& x : cells) for (auto& x : cells)
self->request(x, seconds(1), get_atom::value).receive([&](int y) { self->request(x, seconds(1), get_atom::value).receive(
[&](int y) {
aout(self) << "cell #" << x.id() << " -> " << y << endl; aout(self) << "cell #" << x.id() << " -> " << y << endl;
}); },
[&](error& err) {
aout(self) << "cell #" << x.id()
<< " -> " << self->system().render(err) << endl;
}
);
} }
void caf_main(actor_system& system) { void caf_main(actor_system& system) {
......
...@@ -34,10 +34,12 @@ set (LIBCAF_CORE_SRCS ...@@ -34,10 +34,12 @@ set (LIBCAF_CORE_SRCS
src/behavior_stack.cpp src/behavior_stack.cpp
src/behavior_impl.cpp src/behavior_impl.cpp
src/blocking_actor.cpp src/blocking_actor.cpp
src/blocking_behavior.cpp
src/concatenated_tuple.cpp src/concatenated_tuple.cpp
src/config_option.cpp src/config_option.cpp
src/continue_helper.cpp src/continue_helper.cpp
src/decorated_tuple.cpp src/decorated_tuple.cpp
src/default_invoke_result_visitor.cpp
src/deep_to_string.cpp src/deep_to_string.cpp
src/default_attachable.cpp src/default_attachable.cpp
src/deserializer.cpp src/deserializer.cpp
...@@ -45,7 +47,6 @@ set (LIBCAF_CORE_SRCS ...@@ -45,7 +47,6 @@ set (LIBCAF_CORE_SRCS
src/dynamic_message_data.cpp src/dynamic_message_data.cpp
src/error.cpp src/error.cpp
src/event_based_actor.cpp src/event_based_actor.cpp
src/exception.cpp
src/execution_unit.cpp src/execution_unit.cpp
src/exit_reason.cpp src/exit_reason.cpp
src/forwarding_actor_proxy.cpp src/forwarding_actor_proxy.cpp
...@@ -69,12 +70,14 @@ set (LIBCAF_CORE_SRCS ...@@ -69,12 +70,14 @@ set (LIBCAF_CORE_SRCS
src/message_handler.cpp src/message_handler.cpp
src/node_id.cpp src/node_id.cpp
src/parse_ini.cpp src/parse_ini.cpp
src/private_thread.cpp
src/ref_counted.cpp src/ref_counted.cpp
src/proxy_registry.cpp src/proxy_registry.cpp
src/response_promise.cpp src/response_promise.cpp
src/replies_to.cpp src/replies_to.cpp
src/resumable.cpp src/resumable.cpp
src/ripemd_160.cpp src/ripemd_160.cpp
src/scheduled_actor.cpp
src/scoped_actor.cpp src/scoped_actor.cpp
src/scoped_execution_unit.cpp src/scoped_execution_unit.cpp
src/sec.cpp src/sec.cpp
......
...@@ -59,8 +59,6 @@ public: ...@@ -59,8 +59,6 @@ public:
void enqueue(strong_actor_ptr sender, message_id mid, message content, void enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host) override; execution_unit* host) override;
void initialize() override;
private: private:
// set by parent to define custom enqueue action // set by parent to define custom enqueue action
enqueue_handler on_enqueue_; enqueue_handler on_enqueue_;
......
...@@ -39,7 +39,6 @@ ...@@ -39,7 +39,6 @@
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/exception.hpp"
#include "caf/exec_main.hpp" #include "caf/exec_main.hpp"
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/streambuf.hpp" #include "caf/streambuf.hpp"
......
...@@ -34,8 +34,12 @@ ...@@ -34,8 +34,12 @@
#include "caf/actor_config.hpp" #include "caf/actor_config.hpp"
#include "caf/actor_marker.hpp" #include "caf/actor_marker.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/is_timeout_or_catch_all.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/blocking_behavior.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
...@@ -59,61 +63,106 @@ class blocking_actor ...@@ -59,61 +63,106 @@ class blocking_actor
with<mixin::requester, mixin::sender>, with<mixin::requester, mixin::sender>,
public dynamically_typed_actor_base { public dynamically_typed_actor_base {
public: public:
// -- member types -----------------------------------------------------------
using super = extend<local_actor, blocking_actor>:: using super = extend<local_actor, blocking_actor>::
with<mixin::requester, mixin::sender>; with<mixin::requester, mixin::sender>;
using timeout_type = std::chrono::high_resolution_clock::time_point;
using behavior_type = behavior; using behavior_type = behavior;
using signatures = none_t; using signatures = none_t;
blocking_actor(actor_config& sys); // -- nested classes ---------------------------------------------------------
~blocking_actor(); /// Represents pre- and postconditions for receive loops.
class receive_cond {
public:
virtual ~receive_cond();
void enqueue(mailbox_element_ptr, execution_unit*) override; /// Returns whether a precondition for receiving a message still holds.
virtual bool pre();
/************************************************************************** /// Returns whether a postcondition for receiving a message still holds.
* utility stuff and receive() member function family * virtual bool post();
**************************************************************************/ };
using timeout_type = std::chrono::high_resolution_clock::time_point; /// Pseudo receive condition modeling a single receive.
class accept_one_cond : public receive_cond {
public:
virtual ~accept_one_cond();
bool post() override;
};
/// Implementation helper for `blocking_actor::receive_while`.
struct receive_while_helper { struct receive_while_helper {
std::function<void(behavior&)> dq_; using fun_type = std::function<bool()>;
std::function<bool()> stmt_;
blocking_actor* self;
fun_type stmt_;
template <class... Ts> template <class... Ts>
void operator()(Ts&&... xs) { void operator()(Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, static_assert(sizeof...(Ts) > 0,
"operator() requires at least one argument"); "operator() requires at least one argument");
behavior bhvr{std::forward<Ts>(xs)...}; struct cond : receive_cond {
while (stmt_()) fun_type stmt;
dq_(bhvr); cond(fun_type x) : stmt(std::move(x)) {
// nop
}
bool pre() override {
return stmt();
}
};
cond rc{std::move(stmt_)};
self->varargs_receive(rc, message_id::make(), std::forward<Ts>(xs)...);
} }
}; };
/// Implementation helper for `blocking_actor::receive_for`.
template <class T> template <class T>
struct receive_for_helper { struct receive_for_helper {
std::function<void(behavior&)> dq_; blocking_actor* self;
T& begin; T& begin;
T end; T end;
template <class... Ts> template <class... Ts>
void operator()(Ts&&... xs) { void operator()(Ts&&... xs) {
behavior bhvr{std::forward<Ts>(xs)...}; struct cond : receive_cond {
for (; begin != end; ++begin) dq_(bhvr); receive_for_helper& outer;
cond(receive_for_helper& x) : outer(x) {
// nop
}
bool pre() override {
return outer.begin != outer.end;
}
bool post() override {
++outer.begin;
return true;
}
};
cond rc{*this};
self->varargs_receive(rc, message_id::make(), std::forward<Ts>(xs)...);
} }
}; };
/// Implementation helper for `blocking_actor::do_receive`.
struct do_receive_helper { struct do_receive_helper {
std::function<void(behavior&)> dq_; std::function<void (receive_cond& rc)> cb;
behavior bhvr_;
template <class Statement> template <class Statement>
void until(Statement stmt) { void until(Statement stmt) {
do { struct cond : receive_cond {
dq_(bhvr_); Statement f;
} while (stmt() == false); cond(Statement x) : f(std::move(x)) {
// nop
}
bool post() override {
return ! f();
}
};
cond rc{std::move(stmt)};
cb(rc);
} }
void until(const bool& bvalue) { void until(const bool& bvalue) {
...@@ -121,21 +170,33 @@ public: ...@@ -121,21 +170,33 @@ public:
} }
}; };
// -- constructors and destructors -------------------------------------------
blocking_actor(actor_config& sys);
~blocking_actor();
// -- overridden modifiers of abstract_actor ---------------------------------
void enqueue(mailbox_element_ptr, execution_unit*) override;
// -- overridden modifiers of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override;
// -- virtual modifiers ------------------------------------------------------
/// Implements the actor's behavior.
virtual void act();
// -- modifiers --------------------------------------------------------------
/// Dequeues the next message from the mailbox that is /// Dequeues the next message from the mailbox that is
/// matched by given behavior. /// matched by given behavior.
template <class... Ts> template <class... Ts>
void receive(Ts&&... xs) { void receive(Ts&&... xs) {
static_assert(sizeof...(Ts), "at least one argument required"); accept_one_cond rc;
behavior bhvr{std::forward<Ts>(xs)...}; varargs_receive(rc, message_id::make(), std::forward<Ts>(xs)...);
dequeue(bhvr);
}
/// Semantically equal to: `for (;;) { receive(...); }`, but does
/// not cause a temporary behavior object per iteration.
template <class... Ts>
void receive_loop(Ts&&... xs) {
behavior bhvr{std::forward<Ts>(xs)...};
for (;;) dequeue(bhvr);
} }
/// Receives messages for range `[begin, first)`. /// Receives messages for range `[begin, first)`.
...@@ -152,8 +213,8 @@ public: ...@@ -152,8 +213,8 @@ public:
/// ); /// );
/// ~~~ /// ~~~
template <class T> template <class T>
receive_for_helper<T> receive_for(T& begin, const T& end) { receive_for_helper<T> receive_for(T& begin, T end) {
return {make_dequeue_callback(), begin, end}; return {this, begin, std::move(end)};
} }
/// Receives messages as long as `stmt` returns true. /// Receives messages as long as `stmt` returns true.
...@@ -162,18 +223,24 @@ public: ...@@ -162,18 +223,24 @@ public:
/// **Usage example:** /// **Usage example:**
/// ~~~ /// ~~~
/// int i = 0; /// int i = 0;
/// receive_while([&]() { return (++i <= 10); }) /// receive_while([&]() { return (++i <= 10); }) (
/// ( /// ...
/// on<int>() >> int_fun,
/// on<float>() >> float_fun
/// ); /// );
/// ~~~ /// ~~~
template <class Statement> receive_while_helper receive_while(std::function<bool()> stmt);
receive_while_helper receive_while(Statement stmt) {
static_assert(std::is_same<bool, decltype(stmt())>::value, /// Receives messages as long as `ref` is true.
"functor or function does not return a boolean"); /// Semantically equal to: `while (ref) { receive(...); }`.
return {make_dequeue_callback(), stmt}; ///
} /// **Usage example:**
/// ~~~
/// bool running = true;
/// receive_while(running) (
/// ...
/// );
/// ~~~
receive_while_helper receive_while(const bool& ref);
/// Receives messages until `stmt` returns true. /// Receives messages until `stmt` returns true.
/// ///
...@@ -192,21 +259,20 @@ public: ...@@ -192,21 +259,20 @@ public:
/// ~~~ /// ~~~
template <class... Ts> template <class... Ts>
do_receive_helper do_receive(Ts&&... xs) { do_receive_helper do_receive(Ts&&... xs) {
return {make_dequeue_callback(), behavior{std::forward<Ts>(xs)...}}; auto tup = std::make_tuple(std::forward<Ts>(xs)...);
auto cb = [=](receive_cond& rc) mutable {
varargs_tup_receive(rc, message_id::make(), tup);
};
return {cb};
} }
/// Blocks this actor until all other actors are done. /// Blocks this actor until all other actors are done.
void await_all_other_actors_done(); void await_all_other_actors_done();
/// Implements the actor's behavior.
virtual void act();
/// Blocks this actor until all `xs...` have terminated. /// Blocks this actor until all `xs...` have terminated.
template <class... Ts> template <class... Ts>
void wait_for(Ts&&... xs) { void wait_for(Ts&&... xs) {
using wait_for_atom = atom_constant<atom("waitFor")>; using wait_for_atom = atom_constant<atom("waitFor")>;
auto old = default_handler_;
default_handler_ = skip;
size_t expected = 0; size_t expected = 0;
size_t i = 0; size_t i = 0;
size_t attach_results[] = {attach_functor(xs)...}; size_t attach_results[] = {attach_functor(xs)...};
...@@ -217,32 +283,64 @@ public: ...@@ -217,32 +283,64 @@ public:
// nop // nop
} }
); );
// restore custom default handler
default_handler_ = old;
} }
/// @cond PRIVATE /// Sets a user-defined exit reason `err`. This reason
/// is signalized to other actors after `act()` returns.
void fail_state(error err);
// -- observers --------------------------------------------------------------
/// Returns the current exit reason.
inline const error& fail_state() { inline const error& fail_state() {
return fail_state_; return fail_state_;
} }
void initialize() override; /// @cond PRIVATE
void dequeue(behavior& bhvr, message_id mid = invalid_message_id);
/// Blocks until at least one message is in the mailbox.
void await_data(); void await_data();
bool await_data(std::chrono::high_resolution_clock::time_point timeout); /// Blocks until at least one message is in the mailbox or
/// the absolute `timeout` was reached.
bool await_data(timeout_type timeout);
/// @endcond /// Receives messages until either a pre- or postcheck of `rcc` fails.
template <class... Ts>
void varargs_tup_receive(receive_cond& rcc, message_id mid,
std::tuple<Ts...>& tup) {
using namespace detail;
static_assert(sizeof...(Ts), "at least one argument required");
// extract how many arguments are actually the behavior part,
// i.e., neither `after(...) >> ...` nor `others >> ...`.
using filtered =
typename tl_filter_not<
type_list<typename std::decay<Ts>::type...>,
is_timeout_or_catch_all
>::type;
filtered tk;
behavior bhvr{apply_args(make_behavior_impl, get_indices(tk), tup)};
using tail_indices = typename il_range<
tl_size<filtered>::value, sizeof...(Ts)
>::type;
make_blocking_behavior_t factory;
auto fun = apply_args_prefixed(factory, tail_indices{}, tup, bhvr);
receive_impl(rcc, mid, fun);
}
protected: /// Receives messages until either a pre- or postcheck of `rcc` fails.
// helper function to implement receive_(for|while) and do_receive template <class... Ts>
std::function<void(behavior&)> make_dequeue_callback() { void varargs_receive(receive_cond& rcc, message_id mid, Ts&&... xs) {
return [=](behavior& bhvr) { dequeue(bhvr); }; auto tup = std::forward_as_tuple(std::forward<Ts>(xs)...);
varargs_tup_receive(rcc, mid, tup);
} }
/// Receives messages until either a pre- or postcheck of `rcc` fails.
void receive_impl(receive_cond& rcc, message_id mid,
detail::blocking_behavior& bhvr);
/// @endcond
private: private:
size_t attach_functor(const actor&); size_t attach_functor(const actor&);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_CATCH_ALL_HPP
#define CAF_CATCH_ALL_HPP
#include <functional>
#include <type_traits>
namespace caf {
template <class F>
struct catch_all {
using fun_type = std::function<result<message> (const type_erased_tuple*)>;
F handler;
catch_all(catch_all&&) = default;
template <class T>
catch_all(T&& x) : handler(std::forward<T>(x)) {
// nop
}
static_assert(std::is_convertible<F, fun_type>::value,
"catch-all handler must have signature "
"result<message> (const type_erased_tuple*)");
fun_type lift() const {
return handler;
}
};
template <class T>
struct is_catch_all : std::false_type {};
template <class T>
struct is_catch_all<catch_all<T>> : std::true_type {};
} // namespace caf
#endif // CAF_CATCH_ALL_HPP
...@@ -221,15 +221,37 @@ default_behavior_impl<std::tuple<Ts...>>::copy(const generic_timeout_definition& ...@@ -221,15 +221,37 @@ default_behavior_impl<std::tuple<Ts...>>::copy(const generic_timeout_definition&
return apply_args_suffxied(factory, indices, cases_, td); return apply_args_suffxied(factory, indices, cases_, td);
} }
template <class... Ts> struct make_behavior_t {
intrusive_ptr<default_behavior_impl<std::tuple<typename lift_behavior<Ts>::type...>>> constexpr make_behavior_t() {
make_behavior(Ts... xs) { // nop
}
template <class... Ts>
intrusive_ptr<default_behavior_impl<std::tuple<typename lift_behavior<Ts>::type...>>>
operator()(Ts... xs) const {
using type = default_behavior_impl<std::tuple<typename lift_behavior<Ts>::type...>>; using type = default_behavior_impl<std::tuple<typename lift_behavior<Ts>::type...>>;
return make_counted<type>(std::move(xs)...); return make_counted<type>(std::move(xs)...);
} }
};
constexpr make_behavior_t make_behavior = make_behavior_t{};
using behavior_impl_ptr = intrusive_ptr<behavior_impl>; using behavior_impl_ptr = intrusive_ptr<behavior_impl>;
// utility for getting a type-erased version of make_behavior
struct make_behavior_impl_t {
constexpr make_behavior_impl_t() {
// nop
}
template <class... Ts>
behavior_impl_ptr operator()(Ts&&... xs) const {
return make_behavior(std::forward<Ts>(xs)...);
}
};
constexpr make_behavior_impl_t make_behavior_impl = make_behavior_impl_t{};
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_BLOCKING_BEHAVIOR_HPP
#define CAF_BLOCKING_BEHAVIOR_HPP
#include "caf/behavior.hpp"
#include "caf/catch_all.hpp"
#include "caf/timeout_definition.hpp"
namespace caf {
namespace detail {
class blocking_behavior {
public:
behavior nested;
blocking_behavior(behavior nested);
blocking_behavior(blocking_behavior&&) = default;
virtual ~blocking_behavior();
virtual result<message> fallback(const type_erased_tuple*);
virtual duration timeout();
virtual void handle_timeout();
};
template <class F>
class blocking_behavior_v2 : public blocking_behavior {
public:
catch_all<F> f;
blocking_behavior_v2(behavior x, catch_all<F> y)
: blocking_behavior(std::move(x)),
f(std::move(y)) {
// nop
}
blocking_behavior_v2(blocking_behavior_v2&&) = default;
result<message> fallback(const type_erased_tuple* x) override {
return f.handler(x);
}
};
template <class F>
class blocking_behavior_v3 : public blocking_behavior {
public:
timeout_definition<F> f;
blocking_behavior_v3(behavior x, timeout_definition<F> y)
: blocking_behavior(std::move(x)),
f(std::move(y)) {
// nop
}
blocking_behavior_v3(blocking_behavior_v3&&) = default;
duration timeout() override {
return f.timeout;
}
void handle_timeout() override {
f.handler();
}
};
template <class F1, class F2>
class blocking_behavior_v4 : public blocking_behavior {
public:
catch_all<F1> f1;
timeout_definition<F2> f2;
blocking_behavior_v4(behavior x, catch_all<F1> y, timeout_definition<F2> z)
: blocking_behavior(std::move(x)),
f1(std::move(y)),
f2(std::move(z)) {
// nop
}
blocking_behavior_v4(blocking_behavior_v4&&) = default;
result<message> fallback(const type_erased_tuple* x) override {
return f1.handler(x);
}
duration timeout() override {
return f2.timeout;
}
void handle_timeout() override {
f2.handler();
}
};
struct make_blocking_behavior_t {
constexpr make_blocking_behavior_t() {
// nop
}
inline blocking_behavior operator()(behavior& x) const {
return {std::move(x)};
}
template <class F>
blocking_behavior_v2<F> operator()(behavior& x, catch_all<F>& y) const {
return {std::move(x), std::move(y)};
}
template <class F>
blocking_behavior_v3<F> operator()(behavior& x,
timeout_definition<F>& y) const {
return {std::move(x), std::move(y)};
}
template <class F1, class F2>
blocking_behavior_v4<F1, F2> operator()(behavior& x, catch_all<F1>& y,
timeout_definition<F2>& z) const {
return {std::move(x), std::move(y), std::move(z)};
}
};
} // namespace detail
} // namespace caf
#endif // CAF_BLOCKING_BEHAVIOR_HPP
...@@ -17,77 +17,76 @@ ...@@ -17,77 +17,76 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_EXCEPTION_HPP #ifndef CAF_DETAIL_DEFAULT_INVOKE_VISITOR_HPP
#define CAF_EXCEPTION_HPP #define CAF_DETAIL_DEFAULT_INVOKE_VISITOR_HPP
#include <string> #include "caf/local_actor.hpp"
#include <cstdint>
#include <exception>
#include <stdexcept>
#include "caf/exit_reason.hpp" #include "caf/detail/invoke_result_visitor.hpp"
namespace caf { namespace caf {
namespace detail {
/// Base class for exceptions. class default_invoke_result_visitor : public invoke_result_visitor {
class caf_exception : public std::exception {
public: public:
~caf_exception() noexcept; inline default_invoke_result_visitor(local_actor* ptr) : self_(ptr) {
// nop
}
caf_exception() = delete; ~default_invoke_result_visitor();
caf_exception(const caf_exception&) = default;
caf_exception& operator=(const caf_exception&) = default;
/// Returns the error message. void operator()() override {
const char* what() const noexcept; // nop
}
protected: void operator()(error& x) override {
/// Creates an exception with the error string `what_str`. CAF_LOG_TRACE(CAF_ARG(x));
explicit caf_exception(std::string what_str); delegate(x);
}
private: void operator()(message& x) override {
std::string what_; CAF_LOG_TRACE(CAF_ARG(x));
}; delegate(x);
}
/// Thrown if an actor finished execution. void operator()(const none_t& x) override {
class actor_exited : public caf_exception { CAF_LOG_TRACE(CAF_ARG(x));
public: delegate(x);
~actor_exited() noexcept;
explicit actor_exited(error exit_reason);
actor_exited(const actor_exited&) = default;
actor_exited& operator=(const actor_exited&) = default;
/// Returns the exit reason.
inline error reason() const noexcept {
return reason_;
} }
private: private:
error reason_; void deliver(response_promise& rp, error& x) {
}; CAF_LOG_DEBUG("report error back to requesting actor");
rp.deliver(std::move(x));
}
/// Thrown to indicate that either an actor publishing failed or void deliver(response_promise& rp, message& x) {
/// the middleman was unable to connect to a remote host. CAF_LOG_DEBUG("respond via response_promise");
class network_error : public caf_exception { // suppress empty messages for asynchronous messages
public: if (x.empty() && rp.async())
~network_error() noexcept; return;
explicit network_error(std::string&& what_str); rp.deliver(std::move(x));
explicit network_error(const std::string& what_str); }
network_error(const network_error&) = default;
network_error& operator=(const network_error&) = default;
};
/// Thrown to indicate that an actor publishing failed because void deliver(response_promise& rp, const none_t&) {
/// the requested port could not be used. error err = sec::unexpected_response;
class bind_failure : public network_error { deliver(rp, err);
public: }
~bind_failure() noexcept;
explicit bind_failure(std::string&& what_str); template <class T>
explicit bind_failure(const std::string& what_str); void delegate(T& x) {
bind_failure(const bind_failure&) = default; auto rp = self_->make_response_promise();
bind_failure& operator=(const bind_failure&) = default; if (! rp.pending()) {
CAF_LOG_DEBUG("suppress response message: invalid response promise");
return;
}
deliver(rp, x);
}
local_actor* self_;
}; };
} // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_EXCEPTION_HPP #endif // CAF_DETAIL_DEFAULT_INVOKE_VISITOR_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_PRIVATE_THREAD_HPP
#define CAF_DETAIL_PRIVATE_THREAD_HPP
#include <mutex>
#include <condition_variable>
#include "caf/fwd.hpp"
namespace caf {
namespace detail {
class private_thread {
public:
enum worker_state {
active,
shutdown_requested,
await_resume_or_shutdown
};
private_thread(scheduled_actor* self);
void run();
bool await_resume();
void resume();
void shutdown();
static void exec(private_thread* this_ptr);
void notify_self_destroyed();
void await_self_destroyed();
void start();
private:
std::mutex mtx_;
std::condition_variable cv_;
volatile bool self_destroyed_;
volatile scheduled_actor* self_;
volatile worker_state state_;
actor_system& system_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_PRIVATE_THREAD_HPP
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/actor_marker.hpp" #include "caf/actor_marker.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
...@@ -45,12 +46,13 @@ public: ...@@ -45,12 +46,13 @@ public:
/// A cooperatively scheduled, event-based actor implementation. This is the /// A cooperatively scheduled, event-based actor implementation. This is the
/// recommended base class for user-defined actors. /// recommended base class for user-defined actors.
/// @extends local_actor /// @extends local_actor
class event_based_actor : public extend<local_actor, event_based_actor>:: class event_based_actor : public extend<scheduled_actor,
event_based_actor>::
with<mixin::sender, mixin::requester, with<mixin::sender, mixin::requester,
mixin::behavior_changer>, mixin::behavior_changer>,
public dynamically_typed_actor_base { public dynamically_typed_actor_base {
public: public:
using super = extend<local_actor, event_based_actor>:: using super = extend<scheduled_actor, event_based_actor>::
with<mixin::sender, mixin::requester, mixin::behavior_changer>; with<mixin::sender, mixin::requester, mixin::behavior_changer>;
using signatures = none_t; using signatures = none_t;
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include <new> #include <new>
#include <memory> #include <memory>
#include <ostream>
#include <type_traits> #include <type_traits>
#include "caf/unit.hpp" #include "caf/unit.hpp"
...@@ -396,6 +397,13 @@ private: ...@@ -396,6 +397,13 @@ private:
caf::error error_; caf::error error_;
}; };
template <class T>
auto to_string(const expected<T>& x) -> decltype(to_string(*x)) {
if (x)
return to_string(*x);
return "!" + to_string(x.error());
}
/// @cond PRIVATE /// @cond PRIVATE
/// Assigns the value of `expr` (which must return an `expected`) /// Assigns the value of `expr` (which must return an `expected`)
/// to a new variable named `var` or throws a `std::runtime_error` on error. /// to a new variable named `var` or throws a `std::runtime_error` on error.
...@@ -410,4 +418,18 @@ private: ...@@ -410,4 +418,18 @@ private:
} // namespace caf } // namespace caf
namespace std {
template <class T>
auto operator<<(ostream& oss, const caf::expected<T>& x)
-> decltype(oss << *x) {
if (x)
oss << *x;
else
oss << "!" << to_string(x.error());
return oss;
}
} // namespace std
#endif #endif
...@@ -32,6 +32,8 @@ namespace caf { ...@@ -32,6 +32,8 @@ namespace caf {
template <class T> template <class T>
class function_view_storage { class function_view_storage {
public: public:
using type = function_view_storage;
function_view_storage(T& storage) : storage_(&storage) { function_view_storage(T& storage) : storage_(&storage) {
// nop // nop
} }
...@@ -47,6 +49,8 @@ private: ...@@ -47,6 +49,8 @@ private:
template <class... Ts> template <class... Ts>
class function_view_storage<std::tuple<Ts...>> { class function_view_storage<std::tuple<Ts...>> {
public: public:
using type = function_view_storage;
function_view_storage(std::tuple<Ts...>& storage) : storage_(&storage) { function_view_storage(std::tuple<Ts...>& storage) : storage_(&storage) {
// nop // nop
} }
...@@ -62,6 +66,8 @@ private: ...@@ -62,6 +66,8 @@ private:
template <> template <>
class function_view_storage<unit_t> { class function_view_storage<unit_t> {
public: public:
using type = function_view_storage;
function_view_storage(unit_t&) { function_view_storage(unit_t&) {
// nop // nop
} }
...@@ -71,6 +77,25 @@ public: ...@@ -71,6 +77,25 @@ public:
} }
}; };
struct function_view_storage_catch_all {
message* storage_;
function_view_storage_catch_all(message& ptr) : storage_(&ptr) {
// nop
}
result<message> operator()(const type_erased_tuple* x) {
*storage_ = message::from(x);
return message{};
}
};
template <>
class function_view_storage<message> {
public:
using type = catch_all<function_view_storage_catch_all>;
};
template <class T> template <class T>
struct function_view_flattened_result { struct function_view_flattened_result {
using type = T; using type = T;
...@@ -137,8 +162,6 @@ public: ...@@ -137,8 +162,6 @@ public:
} }
/// Sends a request message to the assigned actor and returns the result. /// Sends a request message to the assigned actor and returns the result.
/// @throws std::runtime_error if no valid actor is assigned or
/// if the request fails
template <class... Ts, template <class... Ts,
class R = class R =
typename function_view_flattened_result< typename function_view_flattened_result<
...@@ -150,17 +173,19 @@ public: ...@@ -150,17 +173,19 @@ public:
>::type...> >::type...>
>::tuple_type >::tuple_type
>::type> >::type>
R operator()(Ts&&... xs) { expected<R> operator()(Ts&&... xs) {
if (impl_.unsafe()) if (impl_.unsafe())
CAF_RAISE_ERROR("bad function call"); return sec::bad_function_call;
error err;
function_view_result<R> result; function_view_result<R> result;
function_view_storage<R> h{result.value};
self_->request(impl_, infinite, std::forward<Ts>(xs)...).receive( self_->request(impl_, infinite, std::forward<Ts>(xs)...).receive(
h, [&](error& x) {
[] (error& x) { err = std::move(x);
CAF_RAISE_ERROR(to_string(x)); },
} typename function_view_storage<R>::type{result.value}
); );
if (err)
return err;
return flatten(result.value); return flatten(result.value);
} }
......
...@@ -77,6 +77,7 @@ class proxy_registry; ...@@ -77,6 +77,7 @@ class proxy_registry;
class continue_helper; class continue_helper;
class mailbox_element; class mailbox_element;
class message_handler; class message_handler;
class scheduled_actor;
class sync_timeout_msg; class sync_timeout_msg;
class response_promise; class response_promise;
class event_based_actor; class event_based_actor;
...@@ -147,6 +148,7 @@ namespace detail { ...@@ -147,6 +148,7 @@ namespace detail {
class disposer; class disposer;
class message_data; class message_data;
class group_manager; class group_manager;
class private_thread;
class dynamic_message_data; class dynamic_message_data;
} // namespace detail } // namespace detail
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IS_TIMEOUT_OR_CATCH_ALL_HPP
#define CAF_IS_TIMEOUT_OR_CATCH_ALL_HPP
#include "caf/catch_all.hpp"
#include "caf/timeout_definition.hpp"
namespace caf {
template <class T>
struct is_timeout_or_catch_all : std::false_type {};
template <class T>
struct is_timeout_or_catch_all<catch_all<T>> : std::true_type {};
template <class T>
struct is_timeout_or_catch_all<timeout_definition<T>> : std::true_type {};
} // namespace caf
#endif // CAF_IS_TIMEOUT_OR_CATCH_ALL_HPP
...@@ -106,8 +106,16 @@ result<message> drop(local_actor*, const type_erased_tuple*); ...@@ -106,8 +106,16 @@ result<message> drop(local_actor*, const type_erased_tuple*);
/// Base class for actors running on this node, either /// Base class for actors running on this node, either
/// living in an own thread or cooperatively scheduled. /// living in an own thread or cooperatively scheduled.
class local_actor : public monitorable_actor, public resumable { class local_actor : public monitorable_actor {
public: public:
// -- static helper functions to implement default handlers ------------------
static void default_error_handler(local_actor* ptr, error& x);
static void default_down_handler(local_actor* ptr, down_msg& x);
static void default_exit_handler(local_actor* ptr, exit_msg& x);
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using mailbox_type = detail::single_reader_queue<mailbox_element, using mailbox_type = detail::single_reader_queue<mailbox_element,
...@@ -133,6 +141,10 @@ public: ...@@ -133,6 +141,10 @@ public:
void on_destroy() override; void on_destroy() override;
// -- pure virtual modifiers -------------------------------------------------
virtual void launch(execution_unit* eu, bool lazy, bool hide) = 0;
// -- spawn functions -------------------------------------------------------- // -- spawn functions --------------------------------------------------------
template <class T, spawn_options Os = no_spawn_options, class... Ts> template <class T, spawn_options Os = no_spawn_options, class... Ts>
...@@ -204,44 +216,6 @@ public: ...@@ -204,44 +216,6 @@ public:
// -- miscellaneous actor operations ----------------------------------------- // -- miscellaneous actor operations -----------------------------------------
/// Sets a custom handler for unexpected messages.
inline void set_default_handler(default_handler fun) {
default_handler_ = std::move(fun);
}
/// Sets a custom handler for error messages.
inline void set_error_handler(error_handler fun) {
error_handler_ = std::move(fun);
}
/// Sets a custom handler for error messages.
template <class T>
auto set_error_handler(T fun) -> decltype(fun(std::declval<error&>())) {
set_error_handler([fun](local_actor*, error& x) { fun(x); });
}
/// Sets a custom handler for down messages.
inline void set_down_handler(down_handler fun) {
down_handler_ = std::move(fun);
}
/// Sets a custom handler for down messages.
template <class T>
auto set_down_handler(T fun) -> decltype(fun(std::declval<down_msg&>())) {
set_down_handler([fun](local_actor*, down_msg& x) { fun(x); });
}
/// Sets a custom handler for error messages.
inline void set_exit_handler(exit_handler fun) {
exit_handler_ = std::move(fun);
}
/// Sets a custom handler for exit messages.
template <class T>
auto set_exit_handler(T fun) -> decltype(fun(std::declval<exit_msg&>())) {
set_exit_handler([fun](local_actor*, exit_msg& x) { fun(x); });
}
/// Returns the execution unit currently used by this actor. /// Returns the execution unit currently used by this actor.
inline execution_unit* context() const { inline execution_unit* context() const {
return context_; return context_;
...@@ -265,25 +239,6 @@ public: ...@@ -265,25 +239,6 @@ public:
/// Causes this actor to leave the group `what`. /// Causes this actor to leave the group `what`.
void leave(const group& what); void leave(const group& what);
/// Finishes execution of this actor after any currently running
/// message handler is done.
/// This member function clears the behavior stack of the running actor
/// and invokes `on_exit()`. The actors does not finish execution
/// if the implementation of `on_exit()` sets a new behavior.
/// When setting a new behavior in `on_exit()`, one has to make sure
/// to not produce an infinite recursion.
///
/// If `on_exit()` did not set a new behavior, the actor sends an
/// exit message to all of its linked actors, sets its state to exited
/// and finishes execution.
///
/// In case this actor uses the blocking API, this member function unwinds
/// the stack by throwing an `actor_exited` exception.
/// @warning This member function throws immediately in thread-based actors
/// that do not use the behavior stack, i.e., actors that use
/// blocking API calls such as {@link receive()}.
void quit(error reason = error{});
/// @cond PRIVATE /// @cond PRIVATE
void monitor(abstract_actor* whom); void monitor(abstract_actor* whom);
...@@ -379,11 +334,6 @@ public: ...@@ -379,11 +334,6 @@ public:
/// The default implementation throws a `std::logic_error`. /// The default implementation throws a `std::logic_error`.
virtual void load_state(deserializer& source, const unsigned int version); virtual void load_state(deserializer& source, const unsigned int version);
// -- overridden member functions of resumable -------------------------------
subtype_t subtype() const override;
resume_result resume(execution_unit*, size_t) override;
// -- here be dragons: end of public interface ------------------------------- // -- here be dragons: end of public interface -------------------------------
...@@ -393,15 +343,8 @@ public: ...@@ -393,15 +343,8 @@ public:
std::pair<resumable::resume_result, invoke_message_result> std::pair<resumable::resume_result, invoke_message_result>
exec_event(mailbox_element_ptr& ptr); exec_event(mailbox_element_ptr& ptr);
// handle `ptr` in an event-based actor, not suitable to be called in a loop
virtual void exec_single_event(execution_unit* ctx, mailbox_element_ptr& ptr);
local_actor(actor_config& sys); local_actor(actor_config& sys);
void intrusive_ptr_add_ref_impl() override;
void intrusive_ptr_release_impl() override;
template <class ActorHandle> template <class ActorHandle>
inline ActorHandle eval_opts(spawn_options opts, ActorHandle res) { inline ActorHandle eval_opts(spawn_options opts, ActorHandle res) {
if (has_monitor_flag(opts)) { if (has_monitor_flag(opts)) {
...@@ -471,7 +414,7 @@ public: ...@@ -471,7 +414,7 @@ public:
|| ! multiplexed_responses_.empty(); || ! multiplexed_responses_.empty();
} }
virtual void initialize() = 0; virtual void initialize();
// clear behavior stack and call cleanup if actor either has no // clear behavior stack and call cleanup if actor either has no
// valid behavior left or has set a planned exit reason // valid behavior left or has set a planned exit reason
...@@ -524,14 +467,6 @@ public: ...@@ -524,14 +467,6 @@ public:
void set_multiplexed_response_handler(message_id response_id, behavior bhvr); void set_multiplexed_response_handler(message_id response_id, behavior bhvr);
// these functions are dispatched via the actor policies table
void launch(execution_unit* eu, bool lazy, bool hide);
using abstract_actor::enqueue;
void enqueue(mailbox_element_ptr, execution_unit*) override;
mailbox_element_ptr next_message(); mailbox_element_ptr next_message();
bool has_next_message(); bool has_next_message();
...@@ -588,6 +523,9 @@ protected: ...@@ -588,6 +523,9 @@ protected:
// used for setting custom exit message handlers // used for setting custom exit message handlers
exit_handler exit_handler_; exit_handler exit_handler_;
// used when detaching actors
detail::private_thread* private_thread_;
/// @endcond /// @endcond
private: private:
...@@ -602,10 +540,6 @@ private: ...@@ -602,10 +540,6 @@ private:
msg_type filter_msg(mailbox_element& node); msg_type filter_msg(mailbox_element& node);
void handle_response(mailbox_element_ptr&, local_actor::pending_response&); void handle_response(mailbox_element_ptr&, local_actor::pending_response&);
class private_thread;
private_thread* private_thread_;
}; };
/// A smart pointer to a {@link local_actor} instance. /// A smart pointer to a {@link local_actor} instance.
......
...@@ -129,6 +129,7 @@ public: ...@@ -129,6 +129,7 @@ public:
param_decay param_decay
>::type; >::type;
/*
static_assert(! std::is_same<pattern, detail::type_list<exit_msg>>::value, static_assert(! std::is_same<pattern, detail::type_list<exit_msg>>::value,
"exit_msg not allowed in message handlers, " "exit_msg not allowed in message handlers, "
"did you mean to use set_exit_handler()?"); "did you mean to use set_exit_handler()?");
...@@ -136,6 +137,7 @@ public: ...@@ -136,6 +137,7 @@ public:
static_assert(! std::is_same<pattern, detail::type_list<down_msg>>::value, static_assert(! std::is_same<pattern, detail::type_list<down_msg>>::value,
"down_msg not allowed in message handlers, " "down_msg not allowed in message handlers, "
"did you mean to use set_down_handler()?"); "did you mean to use set_down_handler()?");
*/
using decayed_arg_types = using decayed_arg_types =
typename detail::tl_map< typename detail::tl_map<
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_OTHERS_HPP
#define CAF_OTHERS_HPP
#include <functional>
#include <type_traits>
#include "caf/catch_all.hpp"
namespace caf {
struct others_t {
constexpr others_t() {
// nop
}
template <class F>
catch_all<F> operator>>(F fun) const {
return {fun};
}
};
constexpr others_t others = others_t{};
} // namespace caf
#endif // CAF_OTHERS_HPP
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <type_traits> #include <type_traits>
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/catch_all.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/typed_behavior.hpp" #include "caf/typed_behavior.hpp"
#include "caf/continue_helper.hpp" #include "caf/continue_helper.hpp"
...@@ -148,21 +149,9 @@ public: ...@@ -148,21 +149,9 @@ public:
using error_handler = std::function<void (error&)>; using error_handler = std::function<void (error&)>;
template <class F, class E = detail::is_callable_t<F>>
void receive(F f) {
receive_impl(f);
}
template <class F, class OnError, template <class F, class OnError,
class E1 = detail::is_callable_t<F>, class E = detail::is_handler_for_ef<OnError, error>>
class E2 = detail::is_handler_for_ef<OnError, error>> detail::is_callable_t<F> receive(F f, OnError ef) {
void receive(F f, OnError ef) {
receive_impl(f, ef);
}
private:
template <class F>
void receive_impl(F& f) {
static_assert(std::is_same< static_assert(std::is_same<
void, void,
typename detail::get_callable_trait<F>::result_type typename detail::get_callable_trait<F>::result_type
...@@ -170,23 +159,25 @@ private: ...@@ -170,23 +159,25 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
behavior tmp{std::move(f)}; typename Self::accept_one_cond rc;
self_->dequeue(tmp, mid_); self_->varargs_receive(rc, mid_, std::move(f), std::move(ef));
} }
template <class F, class OnError> template <class OnError, class F,
void receive_impl(F& f, OnError& ef) { class E = detail::is_callable_t<F>>
static_assert(std::is_same< detail::is_handler_for_ef<OnError, error> receive(OnError ef, F f) {
void, receive(std::move(f), std::move(ef));
typename detail::get_callable_trait<F>::result_type }
>::value,
"response handlers are not allowed to have a return " template <class OnError, class F,
"type other than void"); class E = detail::is_handler_for_ef<OnError, error>>
detail::type_checker<Output, F>::check(); void receive(OnError ef, catch_all<F> ca) {
behavior tmp{std::move(f), std::move(ef)}; typename Self::accept_one_cond rc;
self_->dequeue(tmp, mid_); self_->varargs_receive(rc, mid_, std::move(ef), std::move(ca));
} }
private:
message_id mid_; message_id mid_;
Self* self_; Self* self_;
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP
#define CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP
#include <type_traits>
#include "caf/fwd.hpp"
#include "caf/extend.hpp"
#include "caf/local_actor.hpp"
#include "caf/actor_marker.hpp"
#include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/logger.hpp"
namespace caf {
/// A cooperatively scheduled, event-based actor implementation. This is the
/// recommended base class for user-defined actors.
/// @extends local_actor
class scheduled_actor : public local_actor, public resumable {
public:
// -- constructors and destructors -------------------------------------------
scheduled_actor(actor_config& cfg);
~scheduled_actor();
// -- overridden modifiers of abstract_actor ---------------------------------
void enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
// -- overridden modifiers of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override;
// -- overridden modifiers of resumable --------------------------------------
subtype_t subtype() const override;
void intrusive_ptr_add_ref_impl() override;
void intrusive_ptr_release_impl() override;
resume_result resume(execution_unit*, size_t) override;
// -- virtual modifiers ------------------------------------------------------
/// Invoke `ptr`, not suitable to be called in a loop.
virtual void exec_single_event(execution_unit* ctx, mailbox_element_ptr& ptr);
// -- modifiers --------------------------------------------------------------
/// Finishes execution of this actor after any currently running
/// message handler is done.
/// This member function clears the behavior stack of the running actor
/// and invokes `on_exit()`. The actors does not finish execution
/// if the implementation of `on_exit()` sets a new behavior.
/// When setting a new behavior in `on_exit()`, one has to make sure
/// to not produce an infinite recursion.
///
/// If `on_exit()` did not set a new behavior, the actor sends an
/// exit message to all of its linked actors, sets its state to exited
/// and finishes execution.
///
/// In case this actor uses the blocking API, this member function unwinds
/// the stack by throwing an `actor_exited` exception.
/// @warning This member function throws immediately in thread-based actors
/// that do not use the behavior stack, i.e., actors that use
/// blocking API calls such as {@link receive()}.
void quit(error reason = error{});
/// Sets a custom handler for unexpected messages.
inline void set_default_handler(default_handler fun) {
default_handler_ = std::move(fun);
}
/// Sets a custom handler for error messages.
inline void set_error_handler(error_handler fun) {
error_handler_ = std::move(fun);
}
/// Sets a custom handler for error messages.
template <class T>
auto set_error_handler(T fun) -> decltype(fun(std::declval<error&>())) {
set_error_handler([fun](local_actor*, error& x) { fun(x); });
}
/// Sets a custom handler for down messages.
inline void set_down_handler(down_handler fun) {
down_handler_ = std::move(fun);
}
/// Sets a custom handler for down messages.
template <class T>
auto set_down_handler(T fun) -> decltype(fun(std::declval<down_msg&>())) {
set_down_handler([fun](local_actor*, down_msg& x) { fun(x); });
}
/// Sets a custom handler for error messages.
inline void set_exit_handler(exit_handler fun) {
exit_handler_ = std::move(fun);
}
/// Sets a custom handler for exit messages.
template <class T>
auto set_exit_handler(T fun) -> decltype(fun(std::declval<exit_msg&>())) {
set_exit_handler([fun](local_actor*, exit_msg& x) { fun(x); });
}
};
} // namespace caf
#endif // CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP
...@@ -75,7 +75,9 @@ enum class sec : uint8_t { ...@@ -75,7 +75,9 @@ enum class sec : uint8_t {
/// Middleman could not publish an actor because it was invalid. /// Middleman could not publish an actor because it was invalid.
cannot_publish_invalid_actor, cannot_publish_invalid_actor,
/// A remote spawn failed because the provided types did not match. /// A remote spawn failed because the provided types did not match.
cannot_spawn_actor_from_arguments cannot_spawn_actor_from_arguments,
/// A function view was called without assigning an actor first.
bad_function_call
}; };
/// @relates sec /// @relates sec
......
...@@ -20,7 +20,7 @@ ...@@ -20,7 +20,7 @@
#ifndef CAF_TYPED_ACTOR_VIEW_HPP #ifndef CAF_TYPED_ACTOR_VIEW_HPP
#define CAF_TYPED_ACTOR_VIEW_HPP #define CAF_TYPED_ACTOR_VIEW_HPP
#include "caf/local_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
...@@ -34,7 +34,7 @@ class typed_actor_view : public extend<typed_actor_view_base, ...@@ -34,7 +34,7 @@ class typed_actor_view : public extend<typed_actor_view_base,
typed_actor_view<Sigs...>>::template typed_actor_view<Sigs...>>::template
with<mixin::sender, mixin::requester> { with<mixin::sender, mixin::requester> {
public: public:
typed_actor_view(local_actor* selfptr) : self_(selfptr) { typed_actor_view(scheduled_actor* selfptr) : self_(selfptr) {
// nop // nop
} }
...@@ -96,7 +96,7 @@ public: ...@@ -96,7 +96,7 @@ public:
} }
private: private:
local_actor* self_; scheduled_actor* self_;
}; };
} // namespace caf } // namespace caf
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/typed_behavior.hpp" #include "caf/typed_behavior.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
#include "caf/mixin/behavior_changer.hpp" #include "caf/mixin/behavior_changer.hpp"
...@@ -41,7 +42,7 @@ public: ...@@ -41,7 +42,7 @@ public:
/// implementation with static type-checking. /// implementation with static type-checking.
/// @extends local_actor /// @extends local_actor
template <class... Sigs> template <class... Sigs>
class typed_event_based_actor : public extend<local_actor, class typed_event_based_actor : public extend<scheduled_actor,
typed_event_based_actor<Sigs...> typed_event_based_actor<Sigs...>
>::template >::template
with<mixin::sender, mixin::requester, with<mixin::sender, mixin::requester,
...@@ -49,7 +50,8 @@ class typed_event_based_actor : public extend<local_actor, ...@@ -49,7 +50,8 @@ class typed_event_based_actor : public extend<local_actor,
public statically_typed_actor_base { public statically_typed_actor_base {
public: public:
using super = typename using super = typename
extend<local_actor, typed_event_based_actor<Sigs...>>::template extend<scheduled_actor,
typed_event_based_actor<Sigs...>>::template
with<mixin::sender, mixin::requester, mixin::behavior_changer>; with<mixin::sender, mixin::requester, mixin::behavior_changer>;
explicit typed_event_based_actor(actor_config& cfg) : super(cfg) { explicit typed_event_based_actor(actor_config& cfg) : super(cfg) {
......
...@@ -29,11 +29,11 @@ ...@@ -29,11 +29,11 @@
#include <condition_variable> #include <condition_variable>
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/local_actor.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/actor_ostream.hpp" #include "caf/actor_ostream.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/scheduler/coordinator.hpp" #include "caf/scheduler/coordinator.hpp"
...@@ -126,8 +126,10 @@ public: ...@@ -126,8 +126,10 @@ public:
} }
if (msg_ptr->msg.match_element<exit_msg>(0)) { if (msg_ptr->msg.match_element<exit_msg>(0)) {
auto& em = msg_ptr->msg.get_as<exit_msg>(0); auto& em = msg_ptr->msg.get_as<exit_msg>(0);
if (em.reason) if (em.reason) {
quit(em.reason); fail_state(em.reason);
return;
}
} }
mfun(msg_ptr->msg); mfun(msg_ptr->msg);
msg_ptr.reset(); msg_ptr.reset();
...@@ -264,14 +266,8 @@ void printer_loop(blocking_actor* self) { ...@@ -264,14 +266,8 @@ void printer_loop(blocking_actor* self) {
std::cout << line << std::flush; std::cout << line << std::flush;
line.clear(); line.clear();
}; };
/* bool done = false;
bool running = true; self->do_receive(
self->set_exit_handler([&](exit_msg&) {
running = false;
});
self->receive_while([&] { return running; })(
*/
self->receive_loop(
[&](add_atom, actor_id aid, std::string& str) { [&](add_atom, actor_id aid, std::string& str) {
if (str.empty() || aid == invalid_actor_id) if (str.empty() || aid == invalid_actor_id)
return; return;
...@@ -298,8 +294,12 @@ void printer_loop(blocking_actor* self) { ...@@ -298,8 +294,12 @@ void printer_loop(blocking_actor* self) {
auto d = get_data(aid, true); auto d = get_data(aid, true);
if (d) if (d)
d->redirect = get_sink_handle(self->system(), fcache, fn, flag); d->redirect = get_sink_handle(self->system(), fcache, fn, flag);
},
[&](exit_msg& em) {
self->fail_state(std::move(em.reason));
done = true;
} }
); ).until([&] { return done; });
} }
} // namespace <anonymous> } // namespace <anonymous>
...@@ -362,7 +362,7 @@ void abstract_coordinator::cleanup_and_release(resumable* ptr) { ...@@ -362,7 +362,7 @@ void abstract_coordinator::cleanup_and_release(resumable* ptr) {
switch (ptr->subtype()) { switch (ptr->subtype()) {
case resumable::scheduled_actor: case resumable::scheduled_actor:
case resumable::io_actor: { case resumable::io_actor: {
auto dptr = static_cast<local_actor*>(ptr); auto dptr = static_cast<scheduled_actor*>(ptr);
dummy_unit dummy{dptr}; dummy_unit dummy{dptr};
dptr->cleanup(make_error(exit_reason::user_shutdown), &dummy); dptr->cleanup(make_error(exit_reason::user_shutdown), &dummy);
while (! dummy.resumables.empty()) { while (! dummy.resumables.empty()) {
...@@ -371,7 +371,7 @@ void abstract_coordinator::cleanup_and_release(resumable* ptr) { ...@@ -371,7 +371,7 @@ void abstract_coordinator::cleanup_and_release(resumable* ptr) {
switch (sub->subtype()) { switch (sub->subtype()) {
case resumable::scheduled_actor: case resumable::scheduled_actor:
case resumable::io_actor: { case resumable::io_actor: {
auto dsub = static_cast<local_actor*>(sub); auto dsub = static_cast<scheduled_actor*>(sub);
dsub->cleanup(make_error(exit_reason::user_shutdown), &dummy); dsub->cleanup(make_error(exit_reason::user_shutdown), &dummy);
break; break;
} }
......
...@@ -50,8 +50,4 @@ void actor_companion::enqueue(strong_actor_ptr src, message_id mid, ...@@ -50,8 +50,4 @@ void actor_companion::enqueue(strong_actor_ptr src, message_id mid,
enqueue(std::move(ptr), eu); enqueue(std::move(ptr), eu);
} }
void actor_companion::initialize() {
// nop
}
} // namespace caf } // namespace caf
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/locks.hpp" #include "caf/locks.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/exception.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/attachable.hpp" #include "caf/attachable.hpp"
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
......
...@@ -314,8 +314,10 @@ actor_system_config& actor_system_config::set(const char* cn, config_value cv) { ...@@ -314,8 +314,10 @@ actor_system_config& actor_system_config::set(const char* cn, config_value cv) {
} }
std::string actor_system_config::render_sec(uint8_t x, atom_value, std::string actor_system_config::render_sec(uint8_t x, atom_value,
const message&) { const message& xs) {
return "system_error" + deep_to_string_as_tuple(static_cast<sec>(x)); return "system_error"
+ (xs.empty() ? deep_to_string_as_tuple(static_cast<sec>(x))
: deep_to_string_as_tuple(static_cast<sec>(x), xs));
} }
std::string actor_system_config::render_exit_reason(uint8_t x, atom_value, std::string actor_system_config::render_exit_reason(uint8_t x, atom_value,
......
...@@ -20,17 +20,38 @@ ...@@ -20,17 +20,38 @@
#include "caf/blocking_actor.hpp" #include "caf/blocking_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/exception.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_registry.hpp" #include "caf/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/detail/invoke_result_visitor.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
namespace caf { namespace caf {
blocking_actor::receive_cond::~receive_cond() {
// nop
}
bool blocking_actor::receive_cond::pre() {
return true;
}
bool blocking_actor::receive_cond::post() {
return true;
}
blocking_actor::accept_one_cond::~accept_one_cond() {
// nop
}
bool blocking_actor::accept_one_cond::post() {
return false;
}
blocking_actor::blocking_actor(actor_config& sys) blocking_actor::blocking_actor(actor_config& sys)
: super(sys.add_flag(local_actor::is_blocking_flag)) { : super(sys.add_flag(local_actor::is_blocking_flag)) {
set_default_handler(skip); // nop
} }
blocking_actor::~blocking_actor() { blocking_actor::~blocking_actor() {
...@@ -49,6 +70,52 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) { ...@@ -49,6 +70,52 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
} }
} }
void blocking_actor::launch(execution_unit*, bool, bool hide) {
CAF_LOG_TRACE(CAF_ARG(hide));
CAF_ASSERT(is_blocking());
is_registered(! hide);
home_system().inc_detached_threads();
std::thread([](strong_actor_ptr ptr) {
// actor lives in its own thread
auto this_ptr = ptr->get();
CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0);
auto self = static_cast<blocking_actor*>(this_ptr);
error rsn;
std::exception_ptr eptr = nullptr;
try {
self->act();
rsn = self->fail_state_;
}
catch (...) {
rsn = exit_reason::unhandled_exception;
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = self->handle(eptr);
rsn = opt_reason ? *opt_reason
: exit_reason::unhandled_exception;
}
try {
self->on_exit();
}
catch (...) {
// simply ignore exception
}
self->cleanup(std::move(rsn), self->context());
ptr->home_system->dec_detached_threads();
}, ctrl()).detach();
}
blocking_actor::receive_while_helper
blocking_actor::receive_while(std::function<bool()> stmt) {
return {this, stmt};
}
blocking_actor::receive_while_helper
blocking_actor::receive_while(const bool& ref) {
return receive_while([&] { return ref; });
}
void blocking_actor::await_all_other_actors_done() { void blocking_actor::await_all_other_actors_done() {
system().registry().await_running_count_equal(is_registered() ? 1 : 0); system().registry().await_running_count_equal(is_registered() ? 1 : 0);
} }
...@@ -59,40 +126,79 @@ void blocking_actor::act() { ...@@ -59,40 +126,79 @@ void blocking_actor::act() {
initial_behavior_fac_(this); initial_behavior_fac_(this);
} }
void blocking_actor::initialize() { void blocking_actor::fail_state(error err) {
// nop fail_state_ = std::move(err);
} }
void blocking_actor::dequeue(behavior& bhvr, message_id mid) { void blocking_actor::receive_impl(receive_cond& rcc,
message_id mid,
detail::blocking_behavior& bhvr) {
CAF_LOG_TRACE(CAF_ARG(mid)); CAF_LOG_TRACE(CAF_ARG(mid));
// calculate absolute timeout if requested by user
auto rel_tout = bhvr.timeout();
std::chrono::high_resolution_clock::time_point abs_tout;
if (rel_tout.valid()) {
abs_tout = std::chrono::high_resolution_clock::now();
abs_tout += rel_tout;
}
// try to dequeue from cache first // try to dequeue from cache first
if (invoke_from_cache(bhvr, mid)) if (invoke_from_cache(bhvr.nested, mid))
return; return;
uint32_t timeout_id = 0; // read incoming messages until we have a match or a timeout
if (mid != invalid_message_id) detail::default_invoke_result_visitor visitor{this};
awaited_responses_.emplace_front(mid, bhvr);
else
timeout_id = request_timeout(bhvr.timeout());
if (mid != invalid_message_id && ! find_awaited_response(mid))
awaited_responses_.emplace_front(mid, behavior{});
// read incoming messages
for (;;) { for (;;) {
await_data(); if (! rcc.pre())
auto msg = next_message(); return;
switch (invoke_message(msg, bhvr, mid)) { bool skipped;
case im_success: do {
if (mid == invalid_message_id) skipped = false;
reset_timeout(timeout_id); if (rel_tout.valid()) {
if (! await_data(abs_tout)) {
bhvr.handle_timeout();
return; return;
case im_skipped: }
if (msg) } else {
push_to_cache(std::move(msg)); await_data();
}
auto ptr = next_message();
CAF_ASSERT(ptr != nullptr);
// skip messages that don't match our message ID
if (ptr->mid != mid) {
push_to_cache(std::move(ptr));
continue;
}
ptr.swap(current_element_);
switch (bhvr.nested(visitor, current_element_->msg)) {
case match_case::skip:
skipped = true;
break; break;
default: default:
// delete msg
break; break;
case match_case::no_match: {
auto sres = bhvr.fallback(current_element_->msg.cvals().get());
// when dealing with response messages, there's either a match
// on the first handler or we produce an error to
// get a match on the second (error) handler
if (sres.flag != rt_skip) {
visitor.visit(sres);
} else if (mid.valid()) {
// make new message to replace current_element_->msg
auto x = make_message(make_error(sec::unexpected_response,
std::move(current_element_->msg)));
current_element_->msg = std::move(x);
bhvr.nested(current_element_->msg);
} else {
skipped = true;
}
} }
} }
ptr.swap(current_element_);
if (skipped)
push_to_cache(std::move(ptr));
} while (skipped);
if (! rcc.post())
return;
}
} }
void blocking_actor::await_data() { void blocking_actor::await_data() {
...@@ -100,7 +206,7 @@ void blocking_actor::await_data() { ...@@ -100,7 +206,7 @@ void blocking_actor::await_data() {
mailbox().synchronized_await(mtx_, cv_); mailbox().synchronized_await(mtx_, cv_);
} }
bool blocking_actor::await_data(std::chrono::high_resolution_clock::time_point timeout) { bool blocking_actor::await_data(timeout_type timeout) {
if (has_next_message()) if (has_next_message())
return true; return true;
return mailbox().synchronized_await(mtx_, cv_, timeout); return mailbox().synchronized_await(mtx_, cv_, timeout);
......
...@@ -17,74 +17,30 @@ ...@@ -17,74 +17,30 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include <sstream> #include "caf/detail/blocking_behavior.hpp"
#include <stdlib.h>
#include "caf/config.hpp"
#include "caf/exception.hpp"
#ifdef CAF_WINDOWS
#include <winerror.h>
#else
#include <errno.h>
#include <sys/socket.h>
#include <sys/un.h>
#endif
namespace {
std::string ae_what(caf::error reason) {
std::ostringstream oss;
oss << "actor exited with reason: " << to_string(reason);
return oss.str();
}
} // namespace <anonymous>
namespace caf { namespace caf {
namespace detail {
caf_exception::~caf_exception() noexcept { blocking_behavior::~blocking_behavior() {
// nop
}
caf_exception::caf_exception(std::string x) : what_(std::move(x)) {
// nop
}
const char* caf_exception::what() const noexcept {
return what_.c_str();
}
actor_exited::~actor_exited() noexcept {
// nop
}
actor_exited::actor_exited(error x) : caf_exception(ae_what(x)) {
reason_ = x;
}
network_error::network_error(const std::string& x) : caf_exception(x) {
// nop
}
network_error::network_error(std::string&& x) : caf_exception(std::move(x)) {
// nop // nop
} }
network_error::~network_error() noexcept { blocking_behavior::blocking_behavior(behavior x) : nested(std::move(x)) {
// nop // nop
} }
bind_failure::bind_failure(const std::string& x) : network_error(x) { result<message> blocking_behavior::fallback(const type_erased_tuple*) {
// nop return skip;
} }
bind_failure::bind_failure(std::string&& x) : network_error(std::move(x)) { duration blocking_behavior::timeout() {
// nop return {};
} }
bind_failure::~bind_failure() noexcept { void blocking_behavior::handle_timeout() {
// nop // nop
} }
} // namespace detail
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/default_invoke_result_visitor.hpp"
namespace caf {
namespace detail {
default_invoke_result_visitor::~default_invoke_result_visitor() {
// nop
}
} // namespace detail
} // namespace caf
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/exception.hpp"
#include "caf/scheduler.hpp" #include "caf/scheduler.hpp"
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
...@@ -36,119 +35,19 @@ ...@@ -36,119 +35,19 @@
#include "caf/default_attachable.hpp" #include "caf/default_attachable.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
namespace caf { namespace caf {
class local_actor::private_thread {
public:
enum worker_state {
active,
shutdown_requested,
await_resume_or_shutdown
};
private_thread(local_actor* self)
: self_destroyed_(false),
self_(self),
state_(active),
system_(self->system()) {
intrusive_ptr_add_ref(self->ctrl());
system_.inc_detached_threads();
}
void run() {
auto job = const_cast<local_actor*>(self_);
CAF_PUSH_AID(job->id());
CAF_LOG_TRACE("");
scoped_execution_unit ctx{&job->system()};
auto max_throughput = std::numeric_limits<size_t>::max();
bool resume_later;
for (;;) {
state_ = await_resume_or_shutdown;
do {
resume_later = false;
switch (job->resume(&ctx, max_throughput)) {
case resumable::resume_later:
resume_later = true;
break;
case resumable::done:
intrusive_ptr_release(job->ctrl());
return;
case resumable::awaiting_message:
intrusive_ptr_release(job->ctrl());
break;
case resumable::shutdown_execution_unit:
return;
}
} while (resume_later);
// wait until actor becomes ready again or was destroyed
if (! await_resume())
return;
}
}
bool await_resume() {
std::unique_lock<std::mutex> guard(mtx_);
while (state_ == await_resume_or_shutdown)
cv_.wait(guard);
return state_ == active;
}
void resume() {
std::unique_lock<std::mutex> guard(mtx_);
state_ = active;
cv_.notify_one();
}
void shutdown() {
std::unique_lock<std::mutex> guard(mtx_);
state_ = shutdown_requested;
cv_.notify_one();
}
static void exec(private_thread* this_ptr) {
this_ptr->run();
// make sure to not destroy the private thread object before the
// detached actor is destroyed and this object is unreachable
this_ptr->await_self_destroyed();
// signalize destruction of detached thread to registry
this_ptr->system_.dec_detached_threads();
// done
delete this_ptr;
}
void notify_self_destroyed() {
std::unique_lock<std::mutex> guard(mtx_);
self_destroyed_ = true;
cv_.notify_one();
}
void await_self_destroyed() {
std::unique_lock<std::mutex> guard(mtx_);
while (! self_destroyed_)
cv_.wait(guard);
}
void start() {
std::thread{exec, this}.detach();
}
private:
std::mutex mtx_;
std::condition_variable cv_;
volatile bool self_destroyed_;
volatile local_actor* self_;
volatile worker_state state_;
actor_system& system_;
};
result<message> reflect(local_actor*, const type_erased_tuple* x) { result<message> reflect(local_actor*, const type_erased_tuple* x) {
return message::from(x); return message::from(x);
} }
result<message> reflect_and_quit(local_actor* ptr, const type_erased_tuple* x) { result<message> reflect_and_quit(local_actor* ptr, const type_erased_tuple* x) {
ptr->quit(); error err = exit_reason::normal;
local_actor::default_error_handler(ptr, err);
return reflect(ptr, x); return reflect(ptr, x);
} }
...@@ -166,19 +65,20 @@ result<message> drop(local_actor*, const type_erased_tuple*) { ...@@ -166,19 +65,20 @@ result<message> drop(local_actor*, const type_erased_tuple*) {
return sec::unexpected_message; return sec::unexpected_message;
} }
void default_error_handler(local_actor* ptr, error& x) { void local_actor::default_error_handler(local_actor* ptr, error& x) {
ptr->quit(std::move(x)); ptr->fail_state_ = std::move(x);
ptr->is_terminated(true);
} }
void default_down_handler(local_actor* ptr, down_msg& x) { void local_actor::default_down_handler(local_actor* ptr, down_msg& x) {
aout(ptr) << "*** unhandled down message [id: " << ptr->id() aout(ptr) << "*** unhandled down message [id: " << ptr->id()
<< ", name: " << ptr->name() << "]: " << to_string(x) << ", name: " << ptr->name() << "]: " << to_string(x)
<< std::endl; << std::endl;
} }
void default_exit_handler(local_actor* ptr, exit_msg& x) { void local_actor::default_exit_handler(local_actor* ptr, exit_msg& x) {
if (x.reason) if (x.reason)
ptr->quit(std::move(x.reason)); default_error_handler(ptr, x.reason);
} }
// local actors are created with a reference count of one that is adjusted // local actors are created with a reference count of one that is adjusted
...@@ -220,14 +120,6 @@ void local_actor::on_destroy() { ...@@ -220,14 +120,6 @@ void local_actor::on_destroy() {
} }
} }
void local_actor::intrusive_ptr_add_ref_impl() {
intrusive_ptr_add_ref(ctrl());
}
void local_actor::intrusive_ptr_release_impl() {
intrusive_ptr_release(ctrl());
}
void local_actor::monitor(abstract_actor* ptr) { void local_actor::monitor(abstract_actor* ptr) {
if (! ptr) if (! ptr)
return; return;
...@@ -358,22 +250,30 @@ local_actor::msg_type local_actor::filter_msg(mailbox_element& x) { ...@@ -358,22 +250,30 @@ local_actor::msg_type local_actor::filter_msg(mailbox_element& x) {
: msg_type::expired_timeout; : msg_type::expired_timeout;
} }
case make_type_token<exit_msg>(): { case make_type_token<exit_msg>(): {
if (is_blocking())
return msg_type::ordinary;
auto& em = msg.get_as_mutable<exit_msg>(0); auto& em = msg.get_as_mutable<exit_msg>(0);
// make sure to get rid of attachables if they're no longer needed // make sure to get rid of attachables if they're no longer needed
unlink_from(em.source); unlink_from(em.source);
// exit_reason::kill is always fatal // exit_reason::kill is always fatal
if (em.reason == exit_reason::kill) if (em.reason == exit_reason::kill) {
quit(std::move(em.reason)); fail_state_ = std::move(em.reason);
else is_terminated(true);
} else {
exit_handler_(this, em); exit_handler_(this, em);
}
return msg_type::sys_message; return msg_type::sys_message;
} }
case make_type_token<down_msg>(): { case make_type_token<down_msg>(): {
if (is_blocking())
return msg_type::ordinary;
auto& dm = msg.get_as_mutable<down_msg>(0); auto& dm = msg.get_as_mutable<down_msg>(0);
down_handler_(this, dm); down_handler_(this, dm);
return msg_type::sys_message; return msg_type::sys_message;
} }
case make_type_token<error>(): { case make_type_token<error>(): {
if (is_blocking())
return msg_type::ordinary;
auto& err = msg.get_as_mutable<error>(0); auto& err = msg.get_as_mutable<error>(0);
error_handler_(this, err); error_handler_(this, err);
return msg_type::sys_message; return msg_type::sys_message;
...@@ -383,78 +283,6 @@ local_actor::msg_type local_actor::filter_msg(mailbox_element& x) { ...@@ -383,78 +283,6 @@ local_actor::msg_type local_actor::filter_msg(mailbox_element& x) {
} }
} }
namespace {
class invoke_result_visitor_helper {
public:
invoke_result_visitor_helper(response_promise x) : rp(x) {
// nop
}
void operator()(error& x) {
CAF_LOG_DEBUG("report error back to requesting actor");
rp.deliver(std::move(x));
}
void operator()(message& x) {
CAF_LOG_DEBUG("respond via response_promise");
// suppress empty messages for asynchronous messages
if (x.empty() && rp.async())
return;
rp.deliver(std::move(x));
}
void operator()(const none_t&) {
error err = sec::unexpected_response;
(*this)(err);
}
private:
response_promise rp;
};
class local_actor_invoke_result_visitor : public detail::invoke_result_visitor {
public:
local_actor_invoke_result_visitor(local_actor* ptr) : self_(ptr) {
// nop
}
void operator()() override {
// nop
}
template <class T>
void delegate(T& x) {
auto rp = self_->make_response_promise();
if (! rp.pending()) {
CAF_LOG_DEBUG("suppress response message: invalid response promise");
return;
}
invoke_result_visitor_helper f{std::move(rp)};
f(x);
}
void operator()(error& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
delegate(x);
}
void operator()(message& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
delegate(x);
}
void operator()(const none_t& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
delegate(x);
}
private:
local_actor* self_;
};
} // namespace <anonymous>
void local_actor::handle_response(mailbox_element_ptr& ptr, void local_actor::handle_response(mailbox_element_ptr& ptr,
local_actor::pending_response& pr) { local_actor::pending_response& pr) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
...@@ -464,7 +292,7 @@ void local_actor::handle_response(mailbox_element_ptr& ptr, ...@@ -464,7 +292,7 @@ void local_actor::handle_response(mailbox_element_ptr& ptr,
auto guard = detail::make_scope_guard([&] { auto guard = detail::make_scope_guard([&] {
ptr.swap(current_element_); ptr.swap(current_element_);
}); });
local_actor_invoke_result_visitor visitor{this}; detail::default_invoke_result_visitor visitor{this};
auto invoke_error = [&](error err) { auto invoke_error = [&](error err) {
auto tmp = make_type_erased_tuple_view(err); auto tmp = make_type_erased_tuple_view(err);
if (ref_fun(visitor, tmp) == match_case::no_match) { if (ref_fun(visitor, tmp) == match_case::no_match) {
...@@ -538,7 +366,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -538,7 +366,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
return im_dropped; return im_dropped;
} }
case msg_type::ordinary: { case msg_type::ordinary: {
local_actor_invoke_result_visitor visitor{this}; detail::default_invoke_result_visitor visitor{this};
if (awaited_id.valid()) { if (awaited_id.valid()) {
CAF_LOG_DEBUG("skipped asynchronous message:" << CAF_ARG(awaited_id)); CAF_LOG_DEBUG("skipped asynchronous message:" << CAF_ARG(awaited_id));
return im_skipped; return im_skipped;
...@@ -557,6 +385,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -557,6 +385,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
case match_case::no_match: { case match_case::no_match: {
if (had_timeout) if (had_timeout)
has_timeout(true); has_timeout(true);
if (is_blocking()) {
skipped = true;
} else {
auto sres = default_handler_(this, auto sres = default_handler_(this,
current_element_->msg.cvals().get()); current_element_->msg.cvals().get());
if (sres.flag != rt_skip) if (sres.flag != rt_skip)
...@@ -565,6 +396,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -565,6 +396,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
skipped = true; skipped = true;
} }
} }
}
ptr.swap(current_element_); ptr.swap(current_element_);
if (skipped) { if (skipped) {
if (had_timeout) if (had_timeout)
...@@ -672,186 +504,6 @@ void local_actor::set_multiplexed_response_handler(message_id response_id, behav ...@@ -672,186 +504,6 @@ void local_actor::set_multiplexed_response_handler(message_id response_id, behav
multiplexed_responses_.emplace(response_id, std::move(bhvr)); multiplexed_responses_.emplace(response_id, std::move(bhvr));
} }
void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
is_registered(! hide);
if (is_detached()) {
if (is_blocking()) {
home_system().inc_detached_threads();
std::thread([](strong_actor_ptr ptr) {
// actor lives in its own thread
auto this_ptr = ptr->get();
CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0);
auto self = static_cast<blocking_actor*>(this_ptr);
error rsn;
std::exception_ptr eptr = nullptr;
try {
self->act();
rsn = self->fail_state_;
}
catch (actor_exited& e) {
rsn = e.reason();
}
catch (...) {
rsn = exit_reason::unhandled_exception;
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = self->handle(eptr);
rsn = opt_reason ? *opt_reason
: exit_reason::unhandled_exception;
}
try {
self->on_exit();
}
catch (...) {
// simply ignore exception
}
self->cleanup(std::move(rsn), self->context());
ptr->home_system->dec_detached_threads();
}, ctrl()).detach();
return;
}
private_thread_ = new private_thread(this);
private_thread_->start();
return;
}
CAF_ASSERT(eu != nullptr);
// do not schedule immediately when spawned with `lazy_init`
// mailbox could be set to blocked
if (lazy && mailbox().try_block())
return;
// scheduler has a reference count to the actor as long as
// it is waiting to get scheduled
intrusive_ptr_add_ref(ctrl());
eu->exec_later(this);
}
void local_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(! is_blocking());
auto mid = ptr->mid;
auto sender = ptr->sender;
switch (mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: {
// add a reference count to this actor and re-schedule it
intrusive_ptr_add_ref(ctrl());
if (is_detached()) {
CAF_ASSERT(private_thread_ != nullptr);
private_thread_->resume();
} else {
if (eu)
eu->exec_later(this);
else
home_system().scheduler().enqueue(this);
}
break;
}
case detail::enqueue_result::queue_closed: {
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
}
break;
}
case detail::enqueue_result::success:
// enqueued to a running actors' mailbox; nothing to do
break;
}
}
resumable::subtype_t local_actor::subtype() const {
return scheduled_actor;
}
resumable::resume_result local_actor::resume(execution_unit* eu,
size_t max_throughput) {
CAF_PUSH_AID(id());
CAF_LOG_TRACE("");
CAF_ASSERT(eu != nullptr);
CAF_ASSERT(! is_blocking());
context(eu);
if (is_initialized() && (! has_behavior() || is_terminated())) {
CAF_LOG_DEBUG_IF(! has_behavior(),
"resume called on an actor without behavior");
CAF_LOG_DEBUG_IF(is_terminated(),
"resume called on a terminated actor");
return resumable::done;
}
std::exception_ptr eptr = nullptr;
try {
if (! is_initialized()) {
initialize();
if (finished()) {
CAF_LOG_DEBUG("actor_done() returned true right "
<< "after make_behavior()");
return resumable::resume_result::done;
} else {
CAF_LOG_DEBUG("initialized actor:" << CAF_ARG(name()));
}
}
int handled_msgs = 0;
auto reset_timeout_if_needed = [&] {
if (handled_msgs > 0 && ! bhvr_stack_.empty()) {
request_timeout(bhvr_stack_.back().timeout());
}
};
for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = next_message();
if (ptr) {
auto res = exec_event(ptr);
if (res.first == resumable::resume_result::done)
return resumable::resume_result::done;
if (res.second == im_success)
++handled_msgs;
} else {
CAF_LOG_DEBUG("no more element in mailbox; going to block");
reset_timeout_if_needed();
if (mailbox().try_block())
return resumable::awaiting_message;
CAF_LOG_DEBUG("try_block() interrupted by new message");
}
}
reset_timeout_if_needed();
if (! has_next_message() && mailbox().try_block())
return resumable::awaiting_message;
// time's up
return resumable::resume_later;
}
catch (actor_exited& what) {
CAF_LOG_INFO("actor died because of exception:" << CAF_ARG(what.reason()));
if (! is_terminated())
quit(what.reason());
}
catch (std::exception& e) {
CAF_LOG_INFO("actor died because of an exception, what: " << e.what());
static_cast<void>(e); // keep compiler happy when not logging
if (! is_terminated())
quit(exit_reason::unhandled_exception);
eptr = std::current_exception();
}
catch (...) {
CAF_LOG_INFO("actor died because of an unknown exception");
if (! is_terminated())
quit(exit_reason::unhandled_exception);
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = handle(eptr);
if (opt_reason) {
// use exit reason defined by custom handler
quit(*opt_reason);
}
}
if (! finished()) {
// actor has been "revived", try running it again later
return resumable::resume_later;
}
return resumable::done;
}
std::pair<resumable::resume_result, invoke_message_result> std::pair<resumable::resume_result, invoke_message_result>
local_actor::exec_event(mailbox_element_ptr& ptr) { local_actor::exec_event(mailbox_element_ptr& ptr) {
CAF_LOG_TRACE(CAF_ARG(*ptr)); CAF_LOG_TRACE(CAF_ARG(*ptr));
...@@ -897,38 +549,6 @@ local_actor::exec_event(mailbox_element_ptr& ptr) { ...@@ -897,38 +549,6 @@ local_actor::exec_event(mailbox_element_ptr& ptr) {
return {resumable::resume_result::resume_later, res}; return {resumable::resume_result::resume_later, res};
} }
void local_actor::exec_single_event(execution_unit* ctx,
mailbox_element_ptr& ptr) {
CAF_ASSERT(ctx != nullptr);
context(ctx);
if (! is_initialized()) {
CAF_LOG_DEBUG("initialize actor");
initialize();
if (finished()) {
CAF_LOG_DEBUG("actor_done() returned true right "
<< "after make_behavior()");
return;
}
}
if (! has_behavior() || is_terminated()) {
CAF_LOG_DEBUG_IF(! has_behavior(),
"resume called on an actor without behavior");
CAF_LOG_DEBUG_IF(is_terminated(),
"resume called on a terminated actor");
return;
}
try {
exec_event(ptr);
}
catch (...) {
CAF_LOG_INFO("broker died because of an exception");
auto eptr = std::current_exception();
auto opt_reason = this->handle(eptr);
if (opt_reason)
quit(*opt_reason);
}
}
mailbox_element_ptr local_actor::next_message() { mailbox_element_ptr local_actor::next_message() {
if (! is_priority_aware()) if (! is_priority_aware())
return mailbox_element_ptr{mailbox().try_pop()}; return mailbox_element_ptr{mailbox().try_pop()};
...@@ -1032,6 +652,10 @@ void local_actor::load_state(deserializer&, const unsigned int) { ...@@ -1032,6 +652,10 @@ void local_actor::load_state(deserializer&, const unsigned int) {
CAF_RAISE_ERROR("local_actor::deserialize called"); CAF_RAISE_ERROR("local_actor::deserialize called");
} }
void local_actor::initialize() {
// nop
}
bool local_actor::finished() { bool local_actor::finished() {
if (has_behavior() && ! is_terminated()) if (has_behavior() && ! is_terminated())
return false; return false;
...@@ -1066,12 +690,4 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -1066,12 +690,4 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
return true; return true;
} }
void local_actor::quit(error x) {
CAF_LOG_TRACE(CAF_ARG(x));
fail_state_ = std::move(x);
is_terminated(true);
if (is_blocking())
CAF_RAISE_ERROR(to_string(fail_state_));
}
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/private_thread.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf {
namespace detail {
private_thread::private_thread(scheduled_actor* self)
: self_destroyed_(false),
self_(self),
state_(active),
system_(self->system()) {
intrusive_ptr_add_ref(self->ctrl());
system_.inc_detached_threads();
}
void private_thread::run() {
auto job = const_cast<scheduled_actor*>(self_);
CAF_PUSH_AID(job->id());
CAF_LOG_TRACE("");
scoped_execution_unit ctx{&job->system()};
auto max_throughput = std::numeric_limits<size_t>::max();
bool resume_later;
for (;;) {
state_ = await_resume_or_shutdown;
do {
resume_later = false;
switch (job->resume(&ctx, max_throughput)) {
case resumable::resume_later:
resume_later = true;
break;
case resumable::done:
intrusive_ptr_release(job->ctrl());
return;
case resumable::awaiting_message:
intrusive_ptr_release(job->ctrl());
break;
case resumable::shutdown_execution_unit:
return;
}
} while (resume_later);
// wait until actor becomes ready again or was destroyed
if (! await_resume())
return;
}
}
bool private_thread::await_resume() {
std::unique_lock<std::mutex> guard(mtx_);
while (state_ == await_resume_or_shutdown)
cv_.wait(guard);
return state_ == active;
}
void private_thread::resume() {
std::unique_lock<std::mutex> guard(mtx_);
state_ = active;
cv_.notify_one();
}
void private_thread::shutdown() {
std::unique_lock<std::mutex> guard(mtx_);
state_ = shutdown_requested;
cv_.notify_one();
}
void private_thread::exec(private_thread* this_ptr) {
this_ptr->run();
// make sure to not destroy the private thread object before the
// detached actor is destroyed and this object is unreachable
this_ptr->await_self_destroyed();
// signalize destruction of detached thread to registry
this_ptr->system_.dec_detached_threads();
// done
delete this_ptr;
}
void private_thread::notify_self_destroyed() {
std::unique_lock<std::mutex> guard(mtx_);
self_destroyed_ = true;
cv_.notify_one();
}
void private_thread::await_self_destroyed() {
std::unique_lock<std::mutex> guard(mtx_);
while (! self_destroyed_)
cv_.wait(guard);
}
void private_thread::start() {
std::thread{exec, this}.detach();
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/scheduled_actor.hpp"
#include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf {
scheduled_actor::scheduled_actor(actor_config& cfg) : local_actor(cfg) {
// nop
}
scheduled_actor::~scheduled_actor() {
// nop
}
void scheduled_actor::enqueue(mailbox_element_ptr ptr,
execution_unit* eu) {
CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(! is_blocking());
auto mid = ptr->mid;
auto sender = ptr->sender;
switch (mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: {
// add a reference count to this actor and re-schedule it
intrusive_ptr_add_ref(ctrl());
if (is_detached()) {
CAF_ASSERT(private_thread_ != nullptr);
private_thread_->resume();
} else {
if (eu)
eu->exec_later(this);
else
home_system().scheduler().enqueue(this);
}
break;
}
case detail::enqueue_result::queue_closed: {
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
}
break;
}
case detail::enqueue_result::success:
// enqueued to a running actors' mailbox; nothing to do
break;
}
}
void scheduled_actor::launch(execution_unit* eu, bool lazy, bool hide) {
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
CAF_ASSERT(! is_blocking());
is_registered(! hide);
if (is_detached()) {
private_thread_ = new detail::private_thread(this);
private_thread_->start();
return;
}
CAF_ASSERT(eu != nullptr);
// do not schedule immediately when spawned with `lazy_init`
// mailbox could be set to blocked
if (lazy && mailbox().try_block())
return;
// scheduler has a reference count to the actor as long as
// it is waiting to get scheduled
intrusive_ptr_add_ref(ctrl());
eu->exec_later(this);
}
resumable::subtype_t scheduled_actor::subtype() const {
return resumable::scheduled_actor;
}
void scheduled_actor::intrusive_ptr_add_ref_impl() {
intrusive_ptr_add_ref(ctrl());
}
void scheduled_actor::intrusive_ptr_release_impl() {
intrusive_ptr_release(ctrl());
}
resumable::resume_result
scheduled_actor::resume(execution_unit* eu, size_t max_throughput) {
CAF_PUSH_AID(id());
CAF_LOG_TRACE("");
CAF_ASSERT(eu != nullptr);
CAF_ASSERT(! is_blocking());
context(eu);
if (is_initialized() && (! has_behavior() || is_terminated())) {
CAF_LOG_DEBUG_IF(! has_behavior(),
"resume called on an actor without behavior");
CAF_LOG_DEBUG_IF(is_terminated(),
"resume called on a terminated actor");
return resumable::done;
}
std::exception_ptr eptr = nullptr;
try {
if (! is_initialized()) {
initialize();
if (finished()) {
CAF_LOG_DEBUG("actor_done() returned true right "
<< "after make_behavior()");
return resumable::resume_result::done;
} else {
CAF_LOG_DEBUG("initialized actor:" << CAF_ARG(name()));
}
}
int handled_msgs = 0;
auto reset_timeout_if_needed = [&] {
if (handled_msgs > 0 && ! bhvr_stack_.empty()) {
request_timeout(bhvr_stack_.back().timeout());
}
};
for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = next_message();
if (ptr) {
auto res = exec_event(ptr);
if (res.first == resumable::resume_result::done)
return resumable::resume_result::done;
if (res.second == im_success)
++handled_msgs;
} else {
CAF_LOG_DEBUG("no more element in mailbox; going to block");
reset_timeout_if_needed();
if (mailbox().try_block())
return resumable::awaiting_message;
CAF_LOG_DEBUG("try_block() interrupted by new message");
}
}
reset_timeout_if_needed();
if (! has_next_message() && mailbox().try_block())
return resumable::awaiting_message;
// time's up
return resumable::resume_later;
}
catch (std::exception& e) {
CAF_LOG_INFO("actor died because of an exception, what: " << e.what());
static_cast<void>(e); // keep compiler happy when not logging
if (! is_terminated())
quit(exit_reason::unhandled_exception);
eptr = std::current_exception();
}
catch (...) {
CAF_LOG_INFO("actor died because of an unknown exception");
if (! is_terminated())
quit(exit_reason::unhandled_exception);
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = handle(eptr);
if (opt_reason) {
// use exit reason defined by custom handler
quit(*opt_reason);
}
}
if (! finished()) {
// actor has been "revived", try running it again later
return resumable::resume_later;
}
return resumable::done;
}
void scheduled_actor::quit(error x) {
CAF_LOG_TRACE(CAF_ARG(x));
fail_state_ = std::move(x);
is_terminated(true);
}
void scheduled_actor::exec_single_event(execution_unit* ctx,
mailbox_element_ptr& ptr) {
CAF_ASSERT(ctx != nullptr);
context(ctx);
if (! is_initialized()) {
CAF_LOG_DEBUG("initialize actor");
initialize();
if (finished()) {
CAF_LOG_DEBUG("actor_done() returned true right "
<< "after make_behavior()");
return;
}
}
if (! has_behavior() || is_terminated()) {
CAF_LOG_DEBUG_IF(! has_behavior(),
"resume called on an actor without behavior");
CAF_LOG_DEBUG_IF(is_terminated(),
"resume called on a terminated actor");
return;
}
try {
exec_event(ptr);
}
catch (...) {
CAF_LOG_INFO("broker died because of an exception");
auto eptr = std::current_exception();
auto opt_reason = this->handle(eptr);
if (opt_reason)
quit(*opt_reason);
}
}
} // namespace caf
...@@ -46,14 +46,15 @@ const char* sec_strings[] = { ...@@ -46,14 +46,15 @@ const char* sec_strings[] = {
"invalid_argument", "invalid_argument",
"invalid_protocol_family", "invalid_protocol_family",
"cannot_publish_invalid_actor", "cannot_publish_invalid_actor",
"cannot_spawn_actor_from_arguments" "cannot_spawn_actor_from_arguments",
"bad_function_call"
}; };
} // namespace <anonymous> } // namespace <anonymous>
const char* to_string(sec x) { const char* to_string(sec x) {
auto index = static_cast<size_t>(x); auto index = static_cast<size_t>(x);
if (index > static_cast<size_t>(sec::cannot_spawn_actor_from_arguments)) if (index > static_cast<size_t>(sec::bad_function_call))
return "<unknown>"; return "<unknown>";
return sec_strings[index]; return sec_strings[index];
} }
......
...@@ -78,6 +78,7 @@ CAF_TEST(fun_no_args) { ...@@ -78,6 +78,7 @@ CAF_TEST(fun_no_args) {
}; };
cfg.add_actor_type("test_actor", test_actor_one_arg); cfg.add_actor_type("test_actor", test_actor_one_arg);
test_spawn(make_message()); test_spawn(make_message());
CAF_MESSAGE("test_spawn done");
} }
CAF_TEST(fun_no_args_selfptr) { CAF_TEST(fun_no_args_selfptr) {
......
...@@ -73,6 +73,10 @@ struct fixture { ...@@ -73,6 +73,10 @@ struct fixture {
} }
}; };
void handle_err(const error& err) {
throw std::runtime_error("AUT responded with an error: " + to_string(err));
}
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(actor_pool_tests, fixture) CAF_TEST_FIXTURE_SCOPE(actor_pool_tests, fixture)
...@@ -90,7 +94,8 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -90,7 +94,8 @@ CAF_TEST(round_robin_actor_pool) {
auto sender = actor_cast<strong_actor_ptr>(self->current_sender()); auto sender = actor_cast<strong_actor_ptr>(self->current_sender());
CAF_REQUIRE(sender); CAF_REQUIRE(sender);
workers.push_back(actor_cast<actor>(std::move(sender))); workers.push_back(actor_cast<actor>(std::move(sender)));
} },
handle_err
); );
} }
CAF_CHECK_EQUAL(workers.size(), 6u); CAF_CHECK_EQUAL(workers.size(), 6u);
...@@ -101,7 +106,8 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -101,7 +106,8 @@ CAF_TEST(round_robin_actor_pool) {
std::sort(ws.begin(), ws.end()); std::sort(ws.begin(), ws.end());
CAF_REQUIRE_EQUAL(workers.size(), ws.size()); CAF_REQUIRE_EQUAL(workers.size(), ws.size());
CAF_CHECK(std::equal(workers.begin(), workers.end(), ws.begin())); CAF_CHECK(std::equal(workers.begin(), workers.end(), ws.begin()));
} },
handle_err
); );
anon_send_exit(workers.back(), exit_reason::user_shutdown); anon_send_exit(workers.back(), exit_reason::user_shutdown);
self->wait_for(workers.back()); self->wait_for(workers.back());
...@@ -120,7 +126,8 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -120,7 +126,8 @@ CAF_TEST(round_robin_actor_pool) {
// wait a bit until polling again // wait a bit until polling again
std::this_thread::sleep_for(std::chrono::milliseconds(5)); std::this_thread::sleep_for(std::chrono::milliseconds(5));
} }
} },
handle_err
); );
} }
CAF_REQUIRE(success); CAF_REQUIRE(success);
...@@ -160,7 +167,8 @@ CAF_TEST(random_actor_pool) { ...@@ -160,7 +167,8 @@ CAF_TEST(random_actor_pool) {
self->request(pool, std::chrono::milliseconds(250), 1, 2).receive( self->request(pool, std::chrono::milliseconds(250), 1, 2).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 3); CAF_CHECK_EQUAL(res, 3);
} },
handle_err
); );
} }
self->send_exit(pool, exit_reason::user_shutdown); self->send_exit(pool, exit_reason::user_shutdown);
...@@ -192,12 +200,14 @@ CAF_TEST(split_join_actor_pool) { ...@@ -192,12 +200,14 @@ CAF_TEST(split_join_actor_pool) {
self->request(pool, infinite, std::vector<int>{1, 2, 3, 4, 5}).receive( self->request(pool, infinite, std::vector<int>{1, 2, 3, 4, 5}).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 15); CAF_CHECK_EQUAL(res, 15);
} },
handle_err
); );
self->request(pool, infinite, std::vector<int>{6, 7, 8, 9, 10}).receive( self->request(pool, infinite, std::vector<int>{6, 7, 8, 9, 10}).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 40); CAF_CHECK_EQUAL(res, 40);
} },
handle_err
); );
self->send_exit(pool, exit_reason::user_shutdown); self->send_exit(pool, exit_reason::user_shutdown);
} }
......
...@@ -46,14 +46,12 @@ struct fixture { ...@@ -46,14 +46,12 @@ struct fixture {
: scoped_self(system), : scoped_self(system),
mirror(system.spawn(mirror_impl)), mirror(system.spawn(mirror_impl)),
testee(unsafe_actor_handle_init) { testee(unsafe_actor_handle_init) {
scoped_self->set_down_handler([](local_actor*, down_msg& dm) { // nop
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
});
} }
template <class... Ts> template <class... Ts>
void spawn(Ts&&... xs) { void spawn(Ts&&... xs) {
testee = scoped_self->spawn<monitored>(std::forward<Ts>(xs)...); testee = scoped_self->spawn(std::forward<Ts>(xs)...);
} }
~fixture() { ~fixture() {
......
...@@ -52,6 +52,10 @@ struct fixture { ...@@ -52,6 +52,10 @@ struct fixture {
scoped_actor self{system, true}; scoped_actor self{system, true};
}; };
void handle_err(const error& err) {
throw std::runtime_error("AUT responded with an error: " + to_string(err));
}
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(adapter_tests, fixture) CAF_TEST_FIXTURE_SCOPE(adapter_tests, fixture)
...@@ -130,12 +134,14 @@ CAF_TEST(partial_currying) { ...@@ -130,12 +134,14 @@ CAF_TEST(partial_currying) {
self->request(bound, infinite, 2.0).receive( self->request(bound, infinite, 2.0).receive(
[](double y) { [](double y) {
CAF_CHECK_EQUAL(y, 2.0); CAF_CHECK_EQUAL(y, 2.0);
} },
handle_err
); );
self->request(bound, infinite, 10).receive( self->request(bound, infinite, 10).receive(
[](int y) { [](int y) {
CAF_CHECK_EQUAL(y, 10); CAF_CHECK_EQUAL(y, 10);
} },
handle_err
); );
self->send_exit(aut, exit_reason::kill); self->send_exit(aut, exit_reason::kill);
} }
...@@ -146,7 +152,8 @@ CAF_TEST(full_currying) { ...@@ -146,7 +152,8 @@ CAF_TEST(full_currying) {
self->request(bound, infinite, message{}).receive( self->request(bound, infinite, message{}).receive(
[](int v) { [](int v) {
CAF_CHECK_EQUAL(v, 2); CAF_CHECK_EQUAL(v, 2);
} },
handle_err
); );
anon_send_exit(bound, exit_reason::kill); anon_send_exit(bound, exit_reason::kill);
anon_send_exit(dbl_actor, exit_reason::kill); anon_send_exit(dbl_actor, exit_reason::kill);
...@@ -178,12 +185,14 @@ CAF_TEST(type_safe_currying) { ...@@ -178,12 +185,14 @@ CAF_TEST(type_safe_currying) {
self->request(bound, infinite, 2.0).receive( self->request(bound, infinite, 2.0).receive(
[](double y) { [](double y) {
CAF_CHECK_EQUAL(y, 2.0); CAF_CHECK_EQUAL(y, 2.0);
} },
handle_err
); );
self->request(bound, infinite, 10).receive( self->request(bound, infinite, 10).receive(
[](int y) { [](int y) {
CAF_CHECK_EQUAL(y, 10); CAF_CHECK_EQUAL(y, 10);
} },
handle_err
); );
self->send_exit(aut, exit_reason::kill); self->send_exit(aut, exit_reason::kill);
} }
...@@ -205,7 +214,8 @@ CAF_TEST(reordering) { ...@@ -205,7 +214,8 @@ CAF_TEST(reordering) {
self->request(bound, infinite, 2.0, 10).receive( self->request(bound, infinite, 2.0, 10).receive(
[](double y) { [](double y) {
CAF_CHECK_EQUAL(y, 20.0); CAF_CHECK_EQUAL(y, 20.0);
} },
handle_err
); );
self->send_exit(aut, exit_reason::kill); self->send_exit(aut, exit_reason::kill);
} }
...@@ -231,7 +241,8 @@ CAF_TEST(type_safe_reordering) { ...@@ -231,7 +241,8 @@ CAF_TEST(type_safe_reordering) {
self->request(bound, infinite, 2.0, 10).receive( self->request(bound, infinite, 2.0, 10).receive(
[](double y) { [](double y) {
CAF_CHECK_EQUAL(y, 20.0); CAF_CHECK_EQUAL(y, 20.0);
} },
handle_err
); );
self->send_exit(aut, exit_reason::kill); self->send_exit(aut, exit_reason::kill);
} }
......
...@@ -92,7 +92,8 @@ CAF_TEST(receive_atoms) { ...@@ -92,7 +92,8 @@ CAF_TEST(receive_atoms) {
[&](a_atom, b_atom, c_atom, float value) { [&](a_atom, b_atom, c_atom, float value) {
matched_pattern[2] = true; matched_pattern[2] = true;
CAF_CHECK_EQUAL(value, 23.f); CAF_CHECK_EQUAL(value, 23.f);
}); }
);
} }
CAF_CHECK(matched_pattern[0] && matched_pattern[1] && matched_pattern[2]); CAF_CHECK(matched_pattern[0] && matched_pattern[1] && matched_pattern[2]);
self->receive( self->receive(
...@@ -129,6 +130,9 @@ CAF_TEST(request_atom_constants) { ...@@ -129,6 +130,9 @@ CAF_TEST(request_atom_constants) {
self->request(tst, infinite, abc_atom::value).receive( self->request(tst, infinite, abc_atom::value).receive(
[](int i) { [](int i) {
CAF_CHECK_EQUAL(i, 42); CAF_CHECK_EQUAL(i, 42);
},
[&](error& err) {
CAF_FAIL("err: " << system.render(err));
} }
); );
} }
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace std; using namespace std;
using namespace caf; using namespace caf;
...@@ -230,9 +233,14 @@ CAF_TEST(param_detaching) { ...@@ -230,9 +233,14 @@ CAF_TEST(param_detaching) {
// to a test before moving to the second test; otherwise, reference counts // to a test before moving to the second test; otherwise, reference counts
// can diverge from what we expect // can diverge from what we expect
auto ping_pong = [&] { auto ping_pong = [&] {
self->request(dict, infinite, ping_atom::value).receive([](pong_atom) { self->request(dict, infinite, ping_atom::value).receive(
[](pong_atom) {
// nop // nop
}); },
[&](error& err) {
CAF_FAIL("error: " << system.render(err));
}
);
}; };
// Using CAF is the key to success! // Using CAF is the key to success!
counting_string key = "CAF"; counting_string key = "CAF";
...@@ -255,7 +263,8 @@ CAF_TEST(param_detaching) { ...@@ -255,7 +263,8 @@ CAF_TEST(param_detaching) {
CAF_CHECK_EQUAL(counting_strings_created.load(), 9); CAF_CHECK_EQUAL(counting_strings_created.load(), 9);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 2); CAF_CHECK_EQUAL(counting_strings_moved.load(), 2);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2); CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2);
} },
ERROR_HANDLER
); );
// send put message to dictionary again // send put message to dictionary again
self->request(dict, infinite, put_msg).receive( self->request(dict, infinite, put_msg).receive(
...@@ -265,7 +274,8 @@ CAF_TEST(param_detaching) { ...@@ -265,7 +274,8 @@ CAF_TEST(param_detaching) {
CAF_CHECK_EQUAL(counting_strings_created.load(), 9); CAF_CHECK_EQUAL(counting_strings_created.load(), 9);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 2); CAF_CHECK_EQUAL(counting_strings_moved.load(), 2);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2); CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 2);
} },
ERROR_HANDLER
); );
// alter our initial put, this time moving it to the dictionary // alter our initial put, this time moving it to the dictionary
put_msg.get_as_mutable<counting_string>(1) = "neverlord"; put_msg.get_as_mutable<counting_string>(1) = "neverlord";
...@@ -279,7 +289,8 @@ CAF_TEST(param_detaching) { ...@@ -279,7 +289,8 @@ CAF_TEST(param_detaching) {
CAF_CHECK_EQUAL(counting_strings_created.load(), 11); CAF_CHECK_EQUAL(counting_strings_created.load(), 11);
CAF_CHECK_EQUAL(counting_strings_moved.load(), 4); CAF_CHECK_EQUAL(counting_strings_moved.load(), 4);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 4); CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 4);
} },
ERROR_HANDLER
); );
// finally, check for original key // finally, check for original key
self->request(dict, infinite, std::move(get_msg)).receive( self->request(dict, infinite, std::move(get_msg)).receive(
...@@ -292,7 +303,8 @@ CAF_TEST(param_detaching) { ...@@ -292,7 +303,8 @@ CAF_TEST(param_detaching) {
CAF_CHECK_EQUAL(counting_strings_moved.load(), 5); CAF_CHECK_EQUAL(counting_strings_moved.load(), 5);
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 6); CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 6);
CAF_CHECK_EQUAL(str, "success"); CAF_CHECK_EQUAL(str, "success");
} },
ERROR_HANDLER
); );
// temporary of our handler is destroyed // temporary of our handler is destroyed
CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 7); CAF_CHECK_EQUAL(counting_strings_destroyed.load(), 7);
......
...@@ -73,19 +73,5 @@ CAF_TEST(test_custom_exception_handler) { ...@@ -73,19 +73,5 @@ CAF_TEST(test_custom_exception_handler) {
auto testee3 = self->spawn<exception_testee, monitored>(); auto testee3 = self->spawn<exception_testee, monitored>();
self->send(testee3, "foo"); self->send(testee3, "foo");
// receive all down messages // receive all down messages
self->set_down_handler([&](down_msg& dm) {
if (dm.source == testee1) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
else if (dm.source == testee2) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception);
}
else if (dm.source == testee3) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::remote_link_unreachable);
}
else {
throw std::runtime_error("received message from unexpected source");
}
});
self->wait_for(testee1, testee2, testee3); self->wait_for(testee1, testee2, testee3);
} }
...@@ -141,12 +141,19 @@ public: ...@@ -141,12 +141,19 @@ public:
} }
void act() override { void act() override {
receive_loop ( bool running = true;
receive_while(running) (
[&](int) { [&](int) {
wait4float(); wait4float();
}, },
[&](get_atom) { [&](get_atom) {
return "wait4int"; return "wait4int";
},
[&](exit_msg& em) {
if (em.reason) {
fail_state(std::move(em.reason));
running = false;
}
} }
); );
} }
...@@ -484,13 +491,8 @@ typed_testee::behavior_type testee() { ...@@ -484,13 +491,8 @@ typed_testee::behavior_type testee() {
CAF_TEST(typed_await) { CAF_TEST(typed_await) {
scoped_actor self{system}; scoped_actor self{system};
auto x = system.spawn(testee); auto f = make_function_view(system.spawn(testee));
self->request(x, infinite, abc_atom::value).receive( CAF_CHECK_EQUAL(f(abc_atom::value), "abc");
[](const std::string& str) {
CAF_CHECK_EQUAL(str, "abc");
}
);
self->send_exit(x, exit_reason::user_shutdown);
} }
// tests attach_functor() inside of an actor's constructor // tests attach_functor() inside of an actor's constructor
...@@ -604,28 +606,18 @@ CAF_TEST(custom_exception_handler) { ...@@ -604,28 +606,18 @@ CAF_TEST(custom_exception_handler) {
self->send(testee3, "foo"); self->send(testee3, "foo");
// receive all down messages // receive all down messages
int downs_received = 0; int downs_received = 0;
self->set_down_handler([&](down_msg& dm) { self->receive_for(downs_received, 3) (
if (dm.source == testee1) { [&](down_msg& dm) {
++downs_received; if (dm.source == testee1)
CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception); CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception);
} else if (dm.source == testee2)
else if (dm.source == testee2) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unknown); CAF_CHECK_EQUAL(dm.reason, exit_reason::unknown);
++downs_received; else if (dm.source == testee3)
}
else if (dm.source == testee3) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception); CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception);
++downs_received; else
}
else {
throw std::runtime_error("received message from unexpected source"); throw std::runtime_error("received message from unexpected source");
} }
if (downs_received == 3) );
self->send(self, message{});
});
// this exploits the fact that down messages are counted by dequeue
behavior dummy{[] {}};
self->dequeue(dummy);
} }
CAF_TEST(kill_the_immortal) { CAF_TEST(kill_the_immortal) {
...@@ -647,7 +639,7 @@ CAF_TEST(kill_the_immortal) { ...@@ -647,7 +639,7 @@ CAF_TEST(kill_the_immortal) {
CAF_TEST(move_only_argument) { CAF_TEST(move_only_argument) {
using unique_int = std::unique_ptr<int>; using unique_int = std::unique_ptr<int>;
unique_int uptr{new int(42)}; unique_int uptr{new int(42)};
auto f = [](event_based_actor* self, unique_int ptr) -> behavior { auto impl = [](event_based_actor* self, unique_int ptr) -> behavior {
auto i = *ptr; auto i = *ptr;
return { return {
[=](float) { [=](float) {
...@@ -656,13 +648,18 @@ CAF_TEST(move_only_argument) { ...@@ -656,13 +648,18 @@ CAF_TEST(move_only_argument) {
} }
}; };
}; };
auto f = make_function_view(system.spawn(impl, std::move(uptr)));
CAF_CHECK_EQUAL(to_string(f(1.f)), "(42)");
/*
auto testee = system.spawn(f, std::move(uptr)); auto testee = system.spawn(f, std::move(uptr));
scoped_actor self{system}; scoped_actor self{system};
self->request(testee, infinite, 1.f).receive( self->request(testee, infinite, 1.f).receive(
[](int i) { [](int i) {
CAF_CHECK_EQUAL(i, 42); CAF_CHECK_EQUAL(i, 42);
} },
ERROR_HANDLER
); );
*/
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -98,13 +98,7 @@ CAF_TEST_FIXTURE_SCOPE(function_view_tests, fixture) ...@@ -98,13 +98,7 @@ CAF_TEST_FIXTURE_SCOPE(function_view_tests, fixture)
CAF_TEST(empty_function_fiew) { CAF_TEST(empty_function_fiew) {
function_view<calculator> f; function_view<calculator> f;
try { CAF_CHECK_EQUAL(f(10, 20), sec::bad_function_call);
f(10, 20);
CAF_ERROR("line must be unreachable");
}
catch (std::runtime_error&) {
// nop
}
} }
CAF_TEST(single_res_function_view) { CAF_TEST(single_res_function_view) {
...@@ -122,13 +116,7 @@ CAF_TEST(single_res_function_view) { ...@@ -122,13 +116,7 @@ CAF_TEST(single_res_function_view) {
g.assign(system.spawn(multiplier)); g.assign(system.spawn(multiplier));
CAF_CHECK_EQUAL(g(10, 20), 200); CAF_CHECK_EQUAL(g(10, 20), 200);
g.assign(system.spawn(divider)); g.assign(system.spawn(divider));
try { CAF_CHECK(! g(1, 0));
g(1, 0);
CAF_ERROR("expected exception");
}
catch (std::runtime_error& e) {
CAF_MESSAGE(e.what());
}
g.assign(system.spawn(divider)); g.assign(system.spawn(divider));
CAF_CHECK_EQUAL(g(4, 2), 2); CAF_CHECK_EQUAL(g(4, 2), 2);
} }
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace caf; using namespace caf;
namespace { namespace {
...@@ -49,15 +52,24 @@ struct fixture { ...@@ -49,15 +52,24 @@ struct fixture {
void run_testee(actor testee) { void run_testee(actor testee) {
scoped_actor self{system}; scoped_actor self{system};
self->request(testee, infinite, a_atom::value).receive([](int i) { self->request(testee, infinite, a_atom::value).receive(
[](int i) {
CAF_CHECK_EQUAL(i, 1); CAF_CHECK_EQUAL(i, 1);
}); },
self->request(testee, infinite, b_atom::value).receive([](int i) { ERROR_HANDLER
);
self->request(testee, infinite, b_atom::value).receive(
[](int i) {
CAF_CHECK_EQUAL(i, 2); CAF_CHECK_EQUAL(i, 2);
}); },
self->request(testee, infinite, c_atom::value).receive([](int i) { ERROR_HANDLER
);
self->request(testee, infinite, c_atom::value).receive(
[](int i) {
CAF_CHECK_EQUAL(i, 3); CAF_CHECK_EQUAL(i, 3);
}); },
ERROR_HANDLER
);
self->send_exit(testee, exit_reason::user_shutdown); self->send_exit(testee, exit_reason::user_shutdown);
self->await_all_other_actors_done(); self->await_all_other_actors_done();
} }
......
...@@ -32,6 +32,9 @@ ...@@ -32,6 +32,9 @@
#include "caf/detail/parse_ini.hpp" #include "caf/detail/parse_ini.hpp"
#include "caf/detail/safe_equal.hpp" #include "caf/detail/safe_equal.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace caf; using namespace caf;
namespace { namespace {
...@@ -116,7 +119,8 @@ struct fixture { ...@@ -116,7 +119,8 @@ struct fixture {
[&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) { [&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) {
for (auto& kvp : msgs) for (auto& kvp : msgs)
self->send(config_server, put_atom::value, kvp.first, message{}); self->send(config_server, put_atom::value, kvp.first, message{});
} },
ERROR_HANDLER
); );
auto consume = [&](size_t, std::string key, config_value& value) { auto consume = [&](size_t, std::string key, config_value& value) {
message_visitor mv; message_visitor mv;
...@@ -159,7 +163,8 @@ struct fixture { ...@@ -159,7 +163,8 @@ struct fixture {
result = detail::safe_equal(what, val); result = detail::safe_equal(what, val);
} }
); );
} },
ERROR_HANDLER
); );
return result; return result;
} }
...@@ -192,7 +197,8 @@ struct fixture { ...@@ -192,7 +197,8 @@ struct fixture {
for (auto& kvp : msgs) for (auto& kvp : msgs)
if (! kvp.second.empty()) if (! kvp.second.empty())
++result; ++result;
} },
ERROR_HANDLER
); );
return result; return result;
} }
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace std; using namespace std;
using namespace caf; using namespace caf;
...@@ -254,7 +257,8 @@ CAF_TEST(test_void_res) { ...@@ -254,7 +257,8 @@ CAF_TEST(test_void_res) {
self->request(buddy, infinite, 1, 2).receive( self->request(buddy, infinite, 1, 2).receive(
[] { [] {
CAF_MESSAGE("received void res"); CAF_MESSAGE("received void res");
} },
ERROR_HANDLER
); );
} }
...@@ -309,28 +313,25 @@ CAF_TEST(request_float_or_int) { ...@@ -309,28 +313,25 @@ CAF_TEST(request_float_or_int) {
); );
CAF_CHECK_EQUAL(invocations, 2); CAF_CHECK_EQUAL(invocations, 2);
CAF_MESSAGE("trigger sync failure"); CAF_MESSAGE("trigger sync failure");
bool error_handler_called = false;
bool int_handler_called = false;
self->request(foi, infinite, f_atom::value).receive( self->request(foi, infinite, f_atom::value).receive(
[&](int) { [&](int) {
CAF_ERROR("int handler called"); CAF_FAIL("int handler called");
int_handler_called = true;
}, },
[&](const error& err) { [&](error& err) {
CAF_MESSAGE("error received"); CAF_MESSAGE("error received");
CAF_CHECK_EQUAL(err, sec::unexpected_response); CAF_CHECK_EQUAL(err, sec::unexpected_response);
error_handler_called = true;
} }
); );
CAF_CHECK_EQUAL(error_handler_called, true);
CAF_CHECK_EQUAL(int_handler_called, false);
} }
CAF_TEST(request_to_mirror) { CAF_TEST(request_to_mirror) {
auto mirror = system.spawn<sync_mirror>(); auto mirror = system.spawn<sync_mirror>();
self->request(mirror, infinite, 42).receive([&](int value) { self->request(mirror, infinite, 42).receive(
[&](int value) {
CAF_CHECK_EQUAL(value, 42); CAF_CHECK_EQUAL(value, 42);
}); },
ERROR_HANDLER
);
} }
CAF_TEST(request_to_a_fwd2_b_fwd2_c) { CAF_TEST(request_to_a_fwd2_b_fwd2_c) {
...@@ -338,7 +339,8 @@ CAF_TEST(request_to_a_fwd2_b_fwd2_c) { ...@@ -338,7 +339,8 @@ CAF_TEST(request_to_a_fwd2_b_fwd2_c) {
go_atom::value, self->spawn<B>(self->spawn<C>())).receive( go_atom::value, self->spawn<B>(self->spawn<C>())).receive(
[](ok_atom) { [](ok_atom) {
CAF_MESSAGE("received 'ok'"); CAF_MESSAGE("received 'ok'");
} },
ERROR_HANDLER
); );
} }
...@@ -347,7 +349,8 @@ CAF_TEST(request_to_a_fwd2_d_fwd2_c) { ...@@ -347,7 +349,8 @@ CAF_TEST(request_to_a_fwd2_d_fwd2_c) {
go_atom::value, self->spawn<D>(self->spawn<C>())).receive( go_atom::value, self->spawn<D>(self->spawn<C>())).receive(
[](ok_atom) { [](ok_atom) {
CAF_MESSAGE("received 'ok'"); CAF_MESSAGE("received 'ok'");
} },
ERROR_HANDLER
); );
} }
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace caf; using namespace caf;
namespace { namespace {
...@@ -148,7 +151,8 @@ CAF_TEST(dot_composition_1) { ...@@ -148,7 +151,8 @@ CAF_TEST(dot_composition_1) {
self->request(first_then_second, infinite, 42).receive( self->request(first_then_second, infinite, 42).receive(
[](double res) { [](double res) {
CAF_CHECK_EQUAL(res, (42 * 2.0) * (42 * 4.0)); CAF_CHECK_EQUAL(res, (42 * 2.0) * (42 * 4.0));
} },
ERROR_HANDLER
); );
} }
...@@ -161,9 +165,7 @@ CAF_TEST(dot_composition_2) { ...@@ -161,9 +165,7 @@ CAF_TEST(dot_composition_2) {
[](int v) { [](int v) {
CAF_CHECK_EQUAL(v, 16); CAF_CHECK_EQUAL(v, 16);
}, },
[](error) { ERROR_HANDLER
CAF_CHECK(false);
}
); );
} }
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace caf; using namespace caf;
namespace { namespace {
...@@ -104,7 +107,8 @@ CAF_TEST(untyped_splicing) { ...@@ -104,7 +107,8 @@ CAF_TEST(untyped_splicing) {
CAF_CHECK_EQUAL(x, (42.0 * 2.0)); CAF_CHECK_EQUAL(x, (42.0 * 2.0));
CAF_CHECK_EQUAL(y, (42.0 * 4.0)); CAF_CHECK_EQUAL(y, (42.0 * 4.0));
CAF_CHECK_EQUAL(z, (23.0 * 42.0)); CAF_CHECK_EQUAL(z, (23.0 * 42.0));
} },
ERROR_HANDLER
); );
} }
...@@ -122,7 +126,8 @@ CAF_TEST(typed_splicing) { ...@@ -122,7 +126,8 @@ CAF_TEST(typed_splicing) {
CAF_CHECK_EQUAL(x, (42.0 * 2.0)); CAF_CHECK_EQUAL(x, (42.0 * 2.0));
CAF_CHECK_EQUAL(y, (42.0 * 4.0)); CAF_CHECK_EQUAL(y, (42.0 * 4.0));
CAF_CHECK_EQUAL(z, (23.0 * 42.0)); CAF_CHECK_EQUAL(z, (23.0 * 42.0));
} },
ERROR_HANDLER
); );
// stage0 and stage1 go out of scope, leaving only the references // stage0 and stage1 go out of scope, leaving only the references
// in stages, which will also go out of scope // in stages, which will also go out of scope
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace std; using namespace std;
using namespace caf; using namespace caf;
...@@ -96,7 +99,8 @@ struct fixture { ...@@ -96,7 +99,8 @@ struct fixture {
self->request(aut, infinite, get_atom::value).receive( self->request(aut, infinite, get_atom::value).receive(
[](int x) { [](int x) {
CAF_CHECK_EQUAL(x, 20); CAF_CHECK_EQUAL(x, 20);
} },
ERROR_HANDLER
); );
} }
...@@ -112,7 +116,8 @@ struct fixture { ...@@ -112,7 +116,8 @@ struct fixture {
self->request(aut, infinite, get_atom::value).receive( self->request(aut, infinite, get_atom::value).receive(
[&](const string& str) { [&](const string& str) {
CAF_CHECK_EQUAL(str, expected); CAF_CHECK_EQUAL(str, expected);
} },
ERROR_HANDLER
); );
} }
}; };
......
...@@ -26,6 +26,9 @@ ...@@ -26,6 +26,9 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace caf; using namespace caf;
namespace { namespace {
...@@ -146,13 +149,15 @@ CAF_TEST(typed_response_promise) { ...@@ -146,13 +149,15 @@ CAF_TEST(typed_response_promise) {
self->request(foo, infinite, get_atom::value, 42).receive( self->request(foo, infinite, get_atom::value, 42).receive(
[](int x) { [](int x) {
CAF_CHECK_EQUAL(x, 84); CAF_CHECK_EQUAL(x, 84);
} },
ERROR_HANDLER
); );
self->request(foo, infinite, get_atom::value, 42, 52).receive( self->request(foo, infinite, get_atom::value, 42, 52).receive(
[](int x, int y) { [](int x, int y) {
CAF_CHECK_EQUAL(x, 84); CAF_CHECK_EQUAL(x, 84);
CAF_CHECK_EQUAL(y, 104); CAF_CHECK_EQUAL(y, 104);
} },
ERROR_HANDLER
); );
self->request(foo, infinite, get_atom::value, 3.14, 3.14).receive( self->request(foo, infinite, get_atom::value, 3.14, 3.14).receive(
[](double x, double y) { [](double x, double y) {
...@@ -198,12 +203,13 @@ CAF_TEST(error_response_message) { ...@@ -198,12 +203,13 @@ CAF_TEST(error_response_message) {
CAF_ERROR("unexpected ordinary response message received: " << x); CAF_ERROR("unexpected ordinary response message received: " << x);
} }
); );
self->set_error_handler([&](error& err) { self->send(foo, get_atom::value, 3.14);
self->receive(
[&](error& err) {
CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::unexpected_message)); CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::unexpected_message));
self->send(self, message{}); self->send(self, message{});
}); }
self->send(foo, get_atom::value, 3.14); );
self->receive([] {});
} }
// verify that delivering to a satisfied promise has no effect // verify that delivering to a satisfied promise has no effect
......
...@@ -29,6 +29,9 @@ ...@@ -29,6 +29,9 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace std; using namespace std;
using namespace caf; using namespace caf;
...@@ -321,12 +324,14 @@ struct fixture { ...@@ -321,12 +324,14 @@ struct fixture {
self->request(ts, infinite, my_request{10, 20}).receive( self->request(ts, infinite, my_request{10, 20}).receive(
[](bool value) { [](bool value) {
CAF_CHECK_EQUAL(value, false); CAF_CHECK_EQUAL(value, false);
} },
ERROR_HANDLER
); );
self->request(ts, infinite, my_request{0, 0}).receive( self->request(ts, infinite, my_request{0, 0}).receive(
[](bool value) { [](bool value) {
CAF_CHECK_EQUAL(value, true); CAF_CHECK_EQUAL(value, true);
} },
ERROR_HANDLER
); );
CAF_CHECK_EQUAL(system.registry().running(), 2u); CAF_CHECK_EQUAL(system.registry().running(), 2u);
auto c1 = self->spawn(client, self, ts); auto c1 = self->spawn(client, self, ts);
...@@ -408,7 +413,8 @@ CAF_TEST(string_delegator_chain) { ...@@ -408,7 +413,8 @@ CAF_TEST(string_delegator_chain) {
self->request(aut, infinite, "Hello World!").receive( self->request(aut, infinite, "Hello World!").receive(
[](const string& answer) { [](const string& answer) {
CAF_CHECK_EQUAL(answer, "!dlroW olleH"); CAF_CHECK_EQUAL(answer, "!dlroW olleH");
} },
ERROR_HANDLER
); );
} }
...@@ -431,7 +437,8 @@ CAF_TEST(maybe_string_delegator_chain) { ...@@ -431,7 +437,8 @@ CAF_TEST(maybe_string_delegator_chain) {
self->request(aut, infinite, "abcd").receive( self->request(aut, infinite, "abcd").receive(
[](ok_atom, const string& str) { [](ok_atom, const string& str) {
CAF_CHECK_EQUAL(str, "dcba"); CAF_CHECK_EQUAL(str, "dcba");
} },
ERROR_HANDLER
); );
} }
......
...@@ -23,7 +23,7 @@ ...@@ -23,7 +23,7 @@
#include <vector> #include <vector>
#include <unordered_map> #include <unordered_map>
#include "caf/local_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/prohibit_top_level_spawn_marker.hpp" #include "caf/prohibit_top_level_spawn_marker.hpp"
#include "caf/detail/intrusive_partitioned_list.hpp" #include "caf/detail/intrusive_partitioned_list.hpp"
...@@ -74,7 +74,7 @@ class middleman; ...@@ -74,7 +74,7 @@ class middleman;
/// A broker mediates between actor systems and other components in the network. /// A broker mediates between actor systems and other components in the network.
/// @ingroup Broker /// @ingroup Broker
class abstract_broker : public local_actor, class abstract_broker : public scheduled_actor,
public prohibit_top_level_spawn_marker { public prohibit_top_level_spawn_marker {
public: public:
virtual ~abstract_broker(); virtual ~abstract_broker();
...@@ -83,15 +83,25 @@ public: ...@@ -83,15 +83,25 @@ public:
friend class scribe; friend class scribe;
friend class doorman; friend class doorman;
void enqueue(strong_actor_ptr, message_id, message, execution_unit*) override; // -- overridden modifiers of abstract_actor ---------------------------------
void enqueue(mailbox_element_ptr, execution_unit*) override; void enqueue(mailbox_element_ptr, execution_unit*) override;
/// Called after this broker has finished execution. void enqueue(strong_actor_ptr, message_id, message, execution_unit*) override;
// -- overridden modifiers of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override;
// -- overridden modifiers of abstract_broker --------------------------------
bool cleanup(error&& reason, execution_unit* host) override; bool cleanup(error&& reason, execution_unit* host) override;
/// Starts running this broker in the `middleman`. // -- overridden modifiers of resumable --------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide);
resume_result resume(execution_unit*, size_t) override;
// -- modifiers --------------------------------------------------------------
/// Modifies the receive policy for given connection. /// Modifies the receive policy for given connection.
/// @param hdl Identifies the affected connection. /// @param hdl Identifies the affected connection.
...@@ -110,14 +120,6 @@ public: ...@@ -110,14 +120,6 @@ public:
/// Sends the content of the buffer for given connection. /// Sends the content of the buffer for given connection.
void flush(connection_handle hdl); void flush(connection_handle hdl);
/// Returns the number of open connections.
inline size_t num_connections() const {
return scribes_.size();
}
/// Returns all handles of all `scribe` instances attached to this broker.
std::vector<connection_handle> connections() const;
/// Returns the middleman instance this broker belongs to. /// Returns the middleman instance this broker belongs to.
inline middleman& parent() { inline middleman& parent() {
return system().middleman(); return system().middleman();
...@@ -190,10 +192,6 @@ public: ...@@ -190,10 +192,6 @@ public:
return get_map(hdl).count(hdl) > 0; return get_map(hdl).count(hdl) > 0;
} }
subtype_t subtype() const override;
resume_result resume(execution_unit*, size_t) override;
/// @cond PRIVATE /// @cond PRIVATE
template <class Handle> template <class Handle>
void erase(Handle hdl) { void erase(Handle hdl) {
...@@ -204,8 +202,24 @@ public: ...@@ -204,8 +202,24 @@ public:
} }
/// @endcond /// @endcond
// -- overridden observers of abstract_actor ---------------------------------
const char* name() const override; const char* name() const override;
// -- overridden observers of resumable --------------------------------------
subtype_t subtype() const override;
// -- observers --------------------------------------------------------------
/// Returns the number of open connections.
inline size_t num_connections() const {
return scribes_.size();
}
/// Returns all handles of all `scribe` instances attached to this broker.
std::vector<connection_handle> connections() const;
protected: protected:
void init_broker(); void init_broker();
......
...@@ -18,8 +18,6 @@ ...@@ -18,8 +18,6 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/exception.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
......
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/exception.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
......
...@@ -43,7 +43,7 @@ void abstract_broker::enqueue(strong_actor_ptr src, message_id mid, ...@@ -43,7 +43,7 @@ void abstract_broker::enqueue(strong_actor_ptr src, message_id mid,
void abstract_broker::enqueue(mailbox_element_ptr ptr, execution_unit*) { void abstract_broker::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_PUSH_AID(id()); CAF_PUSH_AID(id());
local_actor::enqueue(std::move(ptr), &backend()); scheduled_actor::enqueue(std::move(ptr), &backend());
} }
void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) { void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) {
...@@ -198,7 +198,7 @@ resumable::resume_result ...@@ -198,7 +198,7 @@ resumable::resume_result
abstract_broker::resume(execution_unit* ctx, size_t mt) { abstract_broker::resume(execution_unit* ctx, size_t mt) {
CAF_ASSERT(ctx != nullptr); CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(ctx == &backend()); CAF_ASSERT(ctx == &backend());
return local_actor::resume(ctx, mt); return scheduled_actor::resume(ctx, mt);
} }
const char* abstract_broker::name() const { const char* abstract_broker::name() const {
...@@ -215,7 +215,7 @@ void abstract_broker::init_broker() { ...@@ -215,7 +215,7 @@ void abstract_broker::init_broker() {
} }
abstract_broker::abstract_broker(actor_config& cfg) : local_actor(cfg) { abstract_broker::abstract_broker(actor_config& cfg) : scheduled_actor(cfg) {
// nop // nop
} }
......
...@@ -25,7 +25,6 @@ ...@@ -25,7 +25,6 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/after.hpp" #include "caf/after.hpp"
#include "caf/exception.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
......
...@@ -21,7 +21,6 @@ ...@@ -21,7 +21,6 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/optional.hpp" #include "caf/optional.hpp"
#include "caf/exception.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
...@@ -93,6 +92,7 @@ bool cc_valid_socket(caf::io::network::native_socket fd) { ...@@ -93,6 +92,7 @@ bool cc_valid_socket(caf::io::network::native_socket fd) {
fun_name, last_socket_error_as_string()) fun_name, last_socket_error_as_string())
// calls a C functions and calls exit() if `predicate(var)` returns false // calls a C functions and calls exit() if `predicate(var)` returns false
#ifdef CAF_WINDOWS
#define CALL_CRITICAL_CFUN(var, predicate, funname, expr) \ #define CALL_CRITICAL_CFUN(var, predicate, funname, expr) \
auto var = expr; \ auto var = expr; \
if (! predicate(var)) { \ if (! predicate(var)) { \
...@@ -100,6 +100,7 @@ bool cc_valid_socket(caf::io::network::native_socket fd) { ...@@ -100,6 +100,7 @@ bool cc_valid_socket(caf::io::network::native_socket fd) {
__FILE__, __LINE__, funname, last_socket_error_as_string().c_str());\ __FILE__, __LINE__, funname, last_socket_error_as_string().c_str());\
abort(); \ abort(); \
} static_cast<void>(0) } static_cast<void>(0)
#endif // CAF_WINDOWS
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -31,7 +31,6 @@ ...@@ -31,7 +31,6 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/exception.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
...@@ -181,16 +180,17 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs, ...@@ -181,16 +180,17 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
std::move(host), port).receive( std::move(host), port).receive(
[&](ok_atom, const node_id&, strong_actor_ptr res, std::set<std::string>& xs) { [&](ok_atom, const node_id&, strong_actor_ptr res, std::set<std::string>& xs) {
CAF_LOG_TRACE(CAF_ARG(res) << CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(res) << CAF_ARG(xs));
if (!res) { if (! res) {
err = make_error(sec::no_actor_published_at_port, err = make_error(sec::no_actor_published_at_port,
"no actor published at port", port); "no actor published at port", port);
return; return;
} }
if (! (xs.empty() && ifs.empty()) if (ifs.empty() != xs.empty()
&& ! std::includes(xs.begin(), xs.end(), ifs.begin(), ifs.end())) { || ! std::includes(xs.begin(), xs.end(), ifs.begin(), ifs.end())) {
using kvpair = std::pair<std::string, std::set<std::string>>;
err = make_error(sec::unexpected_actor_messaging_interface, err = make_error(sec::unexpected_actor_messaging_interface,
"expected signature:", deep_to_string(ifs), kvpair("expected", ifs),
"found:", deep_to_string(xs)); kvpair("found", xs));
return; return;
} }
result.swap(res); result.swap(res);
...@@ -249,7 +249,6 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) { ...@@ -249,7 +249,6 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) {
auto basp = named_broker<basp_broker>(atom("BASP")); auto basp = named_broker<basp_broker>(atom("BASP"));
strong_actor_ptr result; strong_actor_ptr result;
scoped_actor self{system(), true}; scoped_actor self{system(), true};
self->set_default_handler(print_and_drop);
try { try {
self->send(basp, forward_atom::value, actor_cast<strong_actor_ptr>(self), self->send(basp, forward_atom::value, actor_cast<strong_actor_ptr>(self),
nid, atom("ConfigServ"), nid, atom("ConfigServ"),
......
...@@ -27,7 +27,6 @@ ...@@ -27,7 +27,6 @@
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/exception.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/typed_event_based_actor.hpp" #include "caf/typed_event_based_actor.hpp"
......
...@@ -623,6 +623,9 @@ CAF_TEST(remote_actor_and_send) { ...@@ -623,6 +623,9 @@ CAF_TEST(remote_actor_and_send) {
CAF_REQUIRE(proxy != nullptr); CAF_REQUIRE(proxy != nullptr);
CAF_REQUIRE(proxy == res); CAF_REQUIRE(proxy == res);
result = actor_cast<actor>(res); result = actor_cast<actor>(res);
},
[&](error& err) {
CAF_FAIL("error: " << system.render(err));
} }
); );
CAF_MESSAGE("send message to proxy"); CAF_MESSAGE("send message to proxy");
......
...@@ -148,6 +148,9 @@ CAF_TEST(server_side_group_comm) { ...@@ -148,6 +148,9 @@ CAF_TEST(server_side_group_comm) {
group_resolver->request(server, infinite, get_group_atom::value).receive( group_resolver->request(server, infinite, get_group_atom::value).receive(
[&](const group& x) { [&](const group& x) {
grp = x; grp = x;
},
[&](error& err) {
CAF_FAIL("error: " << client_side.render(err));
} }
); );
client_side.spawn(make_client_behavior, server, grp); client_side.spawn(make_client_behavior, server, grp);
......
...@@ -185,6 +185,9 @@ void run_server(int argc, char** argv) { ...@@ -185,6 +185,9 @@ void run_server(int argc, char** argv) {
child = std::thread([=] { child = std::thread([=] {
run_client(argc, argv, port); run_client(argc, argv, port);
}); });
},
[&](error& err) {
CAF_FAIL("error: " << system.render(err));
} }
); );
self->await_all_other_actors_done(); self->await_all_other_actors_done();
......
...@@ -84,24 +84,16 @@ void run_client(int argc, char** argv, uint16_t port) { ...@@ -84,24 +84,16 @@ void run_client(int argc, char** argv, uint16_t port) {
// check whether invalid_argument is thrown // check whether invalid_argument is thrown
// when trying to connect to get an untyped // when trying to connect to get an untyped
// handle to the server // handle to the server
try { auto res = system.middleman().remote_actor("127.0.0.1", port);
system.middleman().remote_actor("127.0.0.1", port); CAF_REQUIRE(! res);
} CAF_MESSAGE(system.render(res.error()));
catch (network_error& e) {
CAF_MESSAGE(e.what());
}
CAF_MESSAGE("connect to typed_remote_actor"); CAF_MESSAGE("connect to typed_remote_actor");
CAF_EXP_THROW(serv, CAF_EXP_THROW(serv,
system.middleman().typed_remote_actor<server_type>("127.0.0.1", system.middleman().typed_remote_actor<server_type>("127.0.0.1",
port)); port));
scoped_actor self{system}; auto f = make_function_view(serv);
self->request(serv, infinite, ping{42}).receive( CAF_CHECK_EQUAL(f(ping{42}), pong{42});
[](const pong& p) {
CAF_CHECK_EQUAL(p.value, 42);
}
);
anon_send_exit(serv, exit_reason::user_shutdown); anon_send_exit(serv, exit_reason::user_shutdown);
self->wait_for(serv);
} }
void run_server(int argc, char** argv) { void run_server(int argc, char** argv) {
......
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