Commit c57e7e92 authored by Dominik Charousset's avatar Dominik Charousset

Implement non-null guarantee for handles

CAF now assumes handle types (actor_addr, actor, typed_actor<...>) to be
non-null. Actors can break this assumption by constructing handles with
`unsafe_actor_handle_init`, but any attempt to use such a handle causes
undefined behavior. This gives handles reference-like semantics, as opposed to
pointer-like semantics. Relates #461.
parent ce8b543f
......@@ -111,8 +111,7 @@ struct base_state {
return aout(self) << color << name << " (id = " << self->id() << "): ";
}
virtual void init(actor m_parent, std::string m_name, std::string m_color) {
parent = std::move(m_parent);
virtual void init(std::string m_name, std::string m_color) {
name = std::move(m_name);
color = std::move(m_color);
print() << "started" << color::reset_endl;
......@@ -123,15 +122,14 @@ struct base_state {
}
local_actor* self;
actor parent;
std::string name;
std::string color;
};
// encapsulates an HTTP request
behavior client_job(stateful_actor<base_state>* self, actor parent) {
self->state.init(std::move(parent), "client-job", color::blue);
self->send(self->state.parent, read_atom::value,
self->state.init("client-job", color::blue);
self->send(parent, read_atom::value,
"http://www.example.com/index.html",
uint64_t{0}, uint64_t{4095});
return {
......@@ -166,7 +164,7 @@ struct client_state : base_state {
behavior client(stateful_actor<client_state>* self, actor parent) {
using std::chrono::milliseconds;
self->link_to(parent);
self->state.init(std::move(parent), "client", color::green);
self->state.init("client", color::green);
self->send(self, next_atom::value);
return {
[=](next_atom) {
......@@ -175,7 +173,7 @@ behavior client(stateful_actor<client_state>* self, actor parent) {
<< color::reset_endl;
// client_job will use IO
// and should thus be spawned in a separate thread
self->spawn<detached+linked>(client_job, st.parent);
self->spawn<detached+linked>(client_job, parent);
// compute random delay until next job is launched
auto delay = st.dist(st.re);
self->delayed_send(self, milliseconds(delay), next_atom::value);
......@@ -202,14 +200,13 @@ struct curl_state : base_state {
return size;
}
void init(actor m_parent, std::string m_name, std::string m_color) override {
void init(std::string m_name, std::string m_color) override {
curl = curl_easy_init();
if (! curl)
throw std::runtime_error("Unable initialize CURL.");
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, &curl_state::callback);
curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
base_state::init(std::move(m_parent), std::move(m_name),
std::move(m_color));
base_state::init(std::move(m_name), std::move(m_color));
}
CURL* curl = nullptr;
......@@ -218,7 +215,7 @@ struct curl_state : base_state {
// manages a CURL session
behavior curl_worker(stateful_actor<curl_state>* self, actor parent) {
self->state.init(std::move(parent), "curl-worker", color::yellow);
self->state.init("curl-worker", color::yellow);
return {
[=](read_atom, const std::string& fname, uint64_t offset, uint64_t range)
-> message {
......@@ -259,7 +256,7 @@ behavior curl_worker(stateful_actor<curl_state>* self, actor parent) {
<< hc
<< color::reset_endl;
// tell parent that this worker is done
self->send(st.parent, finished_atom::value);
self->send(parent, finished_atom::value);
return make_message(reply_atom::value, std::move(st.buf));
case 404: // file does not exist
st.print() << "http error: download failed with "
......@@ -284,7 +281,7 @@ struct master_state : base_state {
};
behavior curl_master(stateful_actor<master_state>* self) {
self->state.init(invalid_actor, "curl-master", color::magenta);
self->state.init("curl-master", color::magenta);
// spawn workers
for(size_t i = 0; i < num_curl_workers; ++i)
self->state.idle.push_back(self->spawn<detached+linked>(curl_worker, self));
......
......@@ -34,7 +34,7 @@ using think_atom = atom_constant<atom("think")>;
using chopstick = typed_actor<replies_to<take_atom>::with<taken_atom, bool>,
reacts_to<put_atom>>;
chopstick::behavior_type taken_chopstick(chopstick::pointer self, actor_addr);
chopstick::behavior_type taken_chopstick(chopstick::pointer, strong_actor_ptr);
// either taken by a philosopher or available
chopstick::behavior_type available_chopstick(chopstick::pointer self) {
......@@ -50,7 +50,7 @@ chopstick::behavior_type available_chopstick(chopstick::pointer self) {
}
chopstick::behavior_type taken_chopstick(chopstick::pointer self,
actor_addr user) {
strong_actor_ptr user) {
return {
[](take_atom) -> std::tuple<taken_atom, bool> {
return std::make_tuple(taken_atom::value, false);
......
......@@ -53,6 +53,7 @@ class client_impl : public event_based_actor {
public:
client_impl(actor_config& cfg, string hostaddr, uint16_t port)
: event_based_actor(cfg),
server_(unsafe_actor_handle_init),
host_(std::move(hostaddr)),
port_(port) {
set_default_handler(skip);
......@@ -109,8 +110,8 @@ private:
auto mm = system().middleman().actor_handle();
send(mm, connect_atom::value, host_, port_);
return {
[=](ok_atom, node_id&, actor_addr& new_server, std::set<std::string>&) {
if (new_server == invalid_actor_addr) {
[=](ok_atom, node_id&, strong_actor_ptr& new_server, std::set<std::string>&) {
if (! new_server) {
aout(this) << "*** received invalid remote actor" << endl;
return;
}
......
......@@ -32,22 +32,13 @@
#include "caf/actor_marker.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/unsafe_actor_handle_init.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
struct invalid_actor_t {
constexpr invalid_actor_t() {
// nop
}
};
/// Identifies an invalid {@link actor}.
/// @relates actor
constexpr invalid_actor_t invalid_actor = invalid_actor_t{};
template <class T>
struct is_convertible_to_actor {
static constexpr bool value =
......@@ -65,12 +56,17 @@ struct is_convertible_to_actor<scoped_actor> : std::true_type {
/// of `event_based_actor`, `blocking_actor`, and `actor_proxy`.
class actor : detail::comparable<actor>,
detail::comparable<actor, actor_addr>,
detail::comparable<actor, invalid_actor_t>,
detail::comparable<actor, invalid_actor_addr_t> {
detail::comparable<actor, strong_actor_ptr> {
public:
// grant access to private ctor
// -- friend types that need access to private ctors
friend class local_actor;
template <class>
friend class data_processor;
template <class>
friend class type_erased_value_impl;
using signatures = none_t;
// allow conversion via actor_cast
......@@ -80,21 +76,24 @@ public:
// tell actor_cast which semantic this type uses
static constexpr bool has_weak_ptr_semantics = false;
actor() = default;
// tell actor_cast this is a non-null handle type
static constexpr bool has_non_null_guarantee = true;
actor(actor&&) = default;
actor(const actor&) = default;
actor& operator=(actor&&) = default;
actor& operator=(const actor&) = default;
actor(const scoped_actor&);
actor(const invalid_actor_t&);
explicit actor(const unsafe_actor_handle_init_t&);
template <class T,
class = typename std::enable_if<
std::is_base_of<dynamically_typed_actor_base, T>::value
>::type>
actor(T* ptr) : ptr_(ptr->ctrl()) {
// nop
CAF_ASSERT(ptr != nullptr);
}
template <class T>
......@@ -114,26 +113,19 @@ public:
}
actor& operator=(const scoped_actor& x);
actor& operator=(const invalid_actor_t&);
/// Returns the address of the stored actor.
actor_addr address() const noexcept;
/// Returns `*this != invalid_actor`.
inline operator bool() const noexcept {
return static_cast<bool>(ptr_);
}
/// Returns `*this == invalid_actor`.
inline bool operator!() const noexcept {
return ! ptr_;
}
/// Returns the origin node of this actor.
node_id node() const noexcept;
inline node_id node() const noexcept {
return ptr_->node();
}
/// Returns the ID of this actor.
actor_id id() const noexcept;
inline actor_id id() const noexcept {
return ptr_->id();
}
/// Exchange content of `*this` and `other`.
void swap(actor& other) noexcept;
......@@ -144,6 +136,12 @@ public:
return bind_impl(make_message(std::forward<Ts>(xs)...));
}
/// Queries whether this object was constructed using
/// `unsafe_actor_handle_init` or is in moved-from state.
bool unsafe() const {
return ! ptr_;
}
/// @cond PRIVATE
inline abstract_actor* operator->() const noexcept {
......@@ -155,13 +153,7 @@ public:
intptr_t compare(const actor_addr&) const noexcept;
inline intptr_t compare(const invalid_actor_t&) const noexcept {
return ptr_ ? 1 : 0;
}
inline intptr_t compare(const invalid_actor_addr_t&) const noexcept {
return compare(invalid_actor);
}
intptr_t compare(const strong_actor_ptr&) const noexcept;
static actor splice_impl(std::initializer_list<actor> xs);
......@@ -178,7 +170,15 @@ public:
/// @endcond
/// Releases the reference held by handle `x`. Using the
/// handle after invalidating it is undefined behavior.
friend void invalidate(actor& x) {
x.ptr_.reset();
}
private:
actor() = default;
actor bind_impl(message msg) const;
inline actor_control_block* get() const noexcept {
......@@ -222,9 +222,8 @@ namespace std {
template <>
struct hash<caf::actor> {
inline size_t operator()(const caf::actor& ref) const {
return ref ? static_cast<size_t>(ref->id()) : 0;
return static_cast<size_t>(ref->id());
}
};
} // namespace std
......
......@@ -28,20 +28,12 @@
#include "caf/fwd.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/unsafe_actor_handle_init.hpp"
#include "caf/detail/comparable.hpp"
namespace caf {
struct invalid_actor_addr_t {
constexpr invalid_actor_addr_t() {}
};
/// Identifies an invalid {@link actor_addr}.
/// @relates actor_addr
constexpr invalid_actor_addr_t invalid_actor_addr = invalid_actor_addr_t{};
/// Stores the address of typed as well as untyped actors.
class actor_addr : detail::comparable<actor_addr>,
detail::comparable<actor_addr, weak_actor_ptr>,
......@@ -49,9 +41,19 @@ class actor_addr : detail::comparable<actor_addr>,
detail::comparable<actor_addr, abstract_actor*>,
detail::comparable<actor_addr, actor_control_block*> {
public:
// -- friend types that need access to private ctors
template <class>
friend class type_erased_value_impl;
template <class>
friend class data_processor;
// grant access to private ctor
friend class actor;
friend class abstract_actor;
friend class down_msg;
friend class exit_msg;
// allow conversion via actor_cast
template <class, class, int>
......@@ -60,25 +62,15 @@ public:
// tell actor_cast which semantic this type uses
static constexpr bool has_weak_ptr_semantics = true;
actor_addr() = default;
// tell actor_cast this is a nullable handle type
static constexpr bool has_non_null_guarantee = true;
actor_addr(actor_addr&&) = default;
actor_addr(const actor_addr&) = default;
actor_addr& operator=(actor_addr&&) = default;
actor_addr& operator=(const actor_addr&) = default;
actor_addr(const invalid_actor_addr_t&);
actor_addr operator=(const invalid_actor_addr_t&);
/// Returns `*this != invalid_actor_addr`.
inline explicit operator bool() const noexcept {
return static_cast<bool>(ptr_);
}
/// Returns `*this == invalid_actor_addr`.
inline bool operator!() const noexcept {
return !ptr_;
}
actor_addr(const unsafe_actor_handle_init_t&);
/// Returns the ID of this actor.
actor_id id() const noexcept;
......@@ -120,11 +112,19 @@ public:
return to_string(x.ptr_);
}
/// Releases the reference held by handle `x`. Using the
/// handle after invalidating it is undefined behavior.
friend void invalidate(actor_addr& x) {
x.ptr_.reset();
}
actor_addr(actor_control_block*, bool);
/// @endcond
private:
actor_addr() = default;
inline actor_control_block* get() const noexcept {
return ptr_.get();
}
......
......@@ -30,23 +30,24 @@ namespace caf {
namespace {
// actor_cast computes the type of the cast for actor_cast_access via
// the following formula: x = 0 if To is a raw poiner
// = 1 if To is a strong pointer
// = 2 if To is a weak pointer
// y = 0 if From is a raw poiner
// = 6 if From is a strong pointer
// = 2 if From is a weak pointer
// the result of x * y then denotes which operation the cast is performing:
// raw <- raw = 0
// raw <- weak = 0
// raw <- strong = 0
// weak <- raw = 0
// weak <- weak = 6
// weak <- strong = 12
// strong <- raw = 0
// strong <- weak = 3
// strong <- strong = 6
// The function actor_cast<> computes the type of the cast for
// actor_cast_access via the following formula:
// x = 0 if To is a raw poiner
// = 1 if To is a strong pointer
// = 2 if To is a weak pointer
// y = 0 if From is a raw poiner
// = 6 if From is a strong pointer
// = 2 if From is a weak pointer
// the result of x * y + z then denotes which operation the cast is performing:
// raw <- raw = 0
// raw <- weak = 0
// raw <- strong = 0
// weak <- raw = 0
// weak <- weak = 6
// weak <- strong = 12
// strong <- raw = 0
// strong <- weak = 3
// strong <- strong = 6
// x * y is then interpreted as follows:
// - 0 is a conversion to or from a raw pointer
// - 6 is a conversion between pointers with same semantics
......@@ -68,6 +69,22 @@ struct is_weak_ptr<T*> {
static constexpr bool value = false;
};
template <class T>
struct is_non_null_handle {
static constexpr bool value = T::has_non_null_guarantee;
};
template <class T>
struct is_non_null_handle<T*> {
static constexpr bool value = false;
};
template <class T>
struct is_strong_non_null_handle {
static constexpr bool value = ! is_weak_ptr<T>::value
&& is_non_null_handle<T>::value;
};
} // namespace <anonymous>
template <class To, class From, int>
......@@ -81,7 +98,7 @@ public:
}
To operator()(abstract_actor* x) const {
return x ? x->ctrl() : nullptr;
return x->ctrl();
}
template <class T,
......@@ -91,24 +108,42 @@ public:
}
};
template <class From>
class actor_cast_access<abstract_actor*, From, raw_ptr_cast> {
template <class To, class From>
class actor_cast_access<To*, From, raw_ptr_cast> {
public:
abstract_actor* operator()(actor_control_block* x) const {
return x ? x->get() : nullptr;
To* operator()(actor_control_block* x) const {
return static_cast<To*>(x->get());
}
abstract_actor* operator()(abstract_actor* x) const {
return x;
To* operator()(abstract_actor* x) const {
return static_cast<To*>(x);
}
template <class T,
class = typename std::enable_if<! std::is_pointer<T>::value>::type>
abstract_actor* operator()(const T& x) const {
To* operator()(const T& x) const {
return (*this)(x.get());
}
};
template <class From>
class actor_cast_access<actor_control_block*, From, raw_ptr_cast> {
public:
actor_control_block* operator()(actor_control_block* x) const {
return x;
}
actor_control_block* operator()(abstract_actor* x) const {
return x->ctrl();
}
template <class T,
class = typename std::enable_if<! std::is_pointer<T>::value>::type>
actor_control_block* operator()(const T& x) const {
return x.get();
}
};
template <class To, class From>
class actor_cast_access<To, From, weak_ptr_downgrade_cast> {
public:
......@@ -141,12 +176,25 @@ public:
/// handle or raw pointer of type `T`.
template <class T, class U>
T actor_cast(U&& what) {
using from = typename std::remove_const<typename std::remove_reference<U>::type>::type;
constexpr int x = std::is_pointer<T>::value ? 0
: (is_weak_ptr<T>::value ? 2 : 1);
constexpr int y = std::is_pointer<from>::value ? 0
: (is_weak_ptr<from>::value ? 3 : 6);
actor_cast_access<T, from, x * y> f;
using from_type =
typename std::remove_const<
typename std::remove_reference<U>::type
>::type;
// query traits for T
constexpr bool to_raw = std::is_pointer<T>::value;
constexpr bool to_weak = is_weak_ptr<T>::value;
// query traits for U
constexpr bool from_raw = std::is_pointer<from_type>::value;
constexpr bool from_weak = is_weak_ptr<from_type>::value;
// check whether this cast is legal
static_assert(! from_weak || ! is_strong_non_null_handle<T>::value,
"casts from actor_addr to actor or typed_actor are prohibited");
// calculate x and y
constexpr int x = to_raw ? 0 : (to_weak ? 2 : 1);
constexpr int y = from_raw ? 0 : (from_weak ? 3 : 6);
// perform cast
actor_cast_access<T, from_type, x * y> f;
return f(std::forward<U>(what));
}
......
......@@ -106,6 +106,8 @@ public:
/// @cond PRIVATE
actor_addr address();
inline actor_id id() const noexcept {
return aid;
}
......@@ -145,6 +147,22 @@ void intrusive_ptr_release(actor_control_block* x);
/// @relates actor_control_block
using strong_actor_ptr = intrusive_ptr<actor_control_block>;
/// @relates strong_actor_ptr
bool operator==(const strong_actor_ptr&, const abstract_actor*);
/// @relates strong_actor_ptr
bool operator==(const abstract_actor*, const strong_actor_ptr&);
/// @relates strong_actor_ptr
inline bool operator!=(const strong_actor_ptr& x, const abstract_actor* y) {
return !(x == y);
}
/// @relates strong_actor_ptr
inline bool operator!=(const abstract_actor* x, const strong_actor_ptr& y) {
return !(x == y);
}
/// @relates abstract_actor
/// @relates actor_control_block
using weak_actor_ptr = weak_intrusive_ptr<actor_control_block>;
......@@ -169,4 +187,23 @@ std::string to_string(const weak_actor_ptr&);
} // namespace caf
// allow actor pointers to be used in hash maps
namespace std {
template <>
struct hash<caf::strong_actor_ptr> {
inline size_t operator()(const caf::strong_actor_ptr& ptr) const {
return ptr ? static_cast<size_t>(ptr->id()) : 0;
}
};
template <>
struct hash<caf::weak_actor_ptr> {
inline size_t operator()(const caf::weak_actor_ptr& ptr) const {
return ptr ? static_cast<size_t>(ptr->id()) : 0;
}
};
} // namespace std
#endif // CAF_ACTOR_CONTROL_BLOCK_HPP
......@@ -156,7 +156,7 @@ actor_factory make_actor_factory(F fun) {
};
handle hdl = cfg.host->system().spawn_class<impl, no_spawn_options>(cfg);
return {actor_cast<strong_actor_ptr>(std::move(hdl)),
cfg.host->system().message_types(hdl)};
cfg.host->system().message_types(detail::type_list<handle>{})};
};
}
......@@ -174,13 +174,12 @@ template <class T, class... Ts>
actor_factory_result dyn_spawn_class(actor_config& cfg, message& msg) {
CAF_ASSERT(cfg.host);
using handle = typename infer_handle_from_class<T>::type;
handle hdl;
handle hdl{unsafe_actor_handle_init};
dyn_spawn_class_helper<handle, T, Ts...> factory{hdl, cfg};
msg.apply(factory);
if (hdl == invalid_actor)
return {};
detail::type_list<handle> token;
return {actor_cast<strong_actor_ptr>(std::move(hdl)),
cfg.host->system().message_types(hdl)};
cfg.host->system().message_types(token)};
}
template <class T, class... Ts>
......
......@@ -171,19 +171,39 @@ public:
using message_types_set = std::set<std::string>;
inline message_types_set message_types(const actor&) {
inline message_types_set message_types(detail::type_list<scoped_actor>) {
return message_types_set{};
}
inline message_types_set message_types(detail::type_list<actor>) {
return message_types_set{};
}
/// Returns a string representation of the messaging
/// interface using portable names;
template <class... Ts>
message_types_set message_types(const typed_actor<Ts...>&) {
message_types_set message_types(detail::type_list<typed_actor<Ts...>>) {
static_assert(sizeof...(Ts) > 0, "empty typed actor handle given");
message_types_set result{get_rtti_from_mpi<Ts>(types())...};
return result;
}
inline message_types_set message_types(const actor&) {
return message_types_set{};
}
template <class... Ts>
message_types_set message_types(const typed_actor<Ts...>&) {
detail::type_list<typed_actor<Ts...>> token;
return message_types(token);
}
/// Returns a string representation of the messaging
/// interface using portable names;
template <class T>
message_types_set message_types() {
detail::type_list<T> token;
return message_types(token);
}
/// Returns the host-local identifier for this system.
const node_id& node() const;
......
......@@ -109,9 +109,10 @@ public:
execution_unit* host) {
if (! ptr->sender)
return;
actor_msg_vec xs(workers.size());
actor_msg_vec xs;
xs.reserve(workers.size());
for (size_t i = 0; i < workers.size(); ++i)
xs[i].first = workers[i];
xs.emplace_back(workers[i], message{});
ulock.unlock();
using collector_t = split_join_collector<T, Split, Join>;
auto hdl = sys.spawn<collector_t, lazy_init>(init_, sf_, jf_, std::move(xs));
......
......@@ -93,7 +93,7 @@ class function_view {
public:
using type = Actor;
function_view() {
function_view() : impl_(unsafe_actor_handle_init) {
// nop
}
......@@ -102,12 +102,12 @@ public:
}
~function_view() {
if (impl_)
if (! impl_.unsafe())
self_.~scoped_actor();
}
function_view(function_view&& x) : impl_(std::move(x.impl_)) {
if (impl_) {
if (! impl_.unsafe()) {
new (&self_) scoped_actor(std::move(x.self_));
x.self_.~scoped_actor();
}
......@@ -115,7 +115,7 @@ public:
function_view& operator=(function_view&& x) {
assign(x.impl_);
x.assign(invalid_actor);
x.assign(unsafe_actor_handle_init);
return *this;
}
......@@ -134,7 +134,7 @@ public:
>::tuple_type
>::type>
R operator()(Ts&&... xs) {
if (! impl_)
if (impl_.unsafe())
throw std::bad_function_call();
R result;
function_view_storage<R> h{result};
......@@ -142,23 +142,23 @@ public:
self_->request(impl_, infinite, std::forward<Ts>(xs)...).receive(h);
}
catch (std::exception&) {
assign(invalid_actor);
assign(unsafe_actor_handle_init);
throw;
}
return flatten(result);
}
void assign(type x) {
if (! impl_ && x)
if (impl_.unsafe() && ! x.unsafe())
new_self(x);
if (impl_ && ! x)
if (! impl_.unsafe() && x.unsafe())
self_.~scoped_actor();
impl_.swap(x);
}
/// Checks whether this function view has an actor assigned to it.
explicit operator bool() const {
return static_cast<bool>(impl_);
return ! impl_.unsafe();
}
private:
......@@ -173,7 +173,7 @@ private:
}
void new_self(const Actor& x) {
if (x)
if (! x.unsafe())
new (&self_) scoped_actor(x->home_system());
}
......
......@@ -88,8 +88,8 @@ class forwarding_actor_proxy;
// -- structs ------------------------------------------------------------------
struct unit_t;
struct exit_msg;
struct down_msg;
class exit_msg;
class down_msg;
struct timeout_msg;
struct group_down_msg;
struct invalid_actor_t;
......
......@@ -45,6 +45,9 @@ public:
// tell actor_cast which semantic this type uses
static constexpr bool has_weak_ptr_semantics = false;
// tell actor_cast this pointer can be null
static constexpr bool has_non_null_guarantee = false;
constexpr intrusive_ptr() : ptr_(nullptr) {
// nop
}
......
......@@ -214,10 +214,13 @@ public:
/// Sends an exit message to `dest`.
void send_exit(const actor_addr& dest, error reason);
void send_exit(const strong_actor_ptr& dest, error reason);
/// Sends an exit message to `dest`.
template <class ActorHandle>
void send_exit(const ActorHandle& dest, error reason) {
send_exit(dest.address(), std::move(reason));
dest->eq_impl(message_id::make(), nullptr, context(),
exit_msg{address(), std::move(reason)});
}
// -- miscellaneous actor operations -----------------------------------------
......@@ -308,10 +311,9 @@ public:
/// @endcond
/// Returns the address of the sender of the current message.
inline actor_addr current_sender() {
return current_element_ ? actor_cast<actor_addr>(current_element_->sender)
: invalid_actor_addr;
/// Returns a pointer to the sender of the current message.
inline strong_actor_ptr current_sender() {
return current_element_ ? current_element_->sender : nullptr;
}
/// Adds a unidirectional `monitor` to `whom`.
......@@ -478,8 +480,6 @@ public:
>::type...>;
static_assert(actor_accepts_message(signatures_of<Handle>(), token{}),
"receiver does not accept given message");
if (! dest)
return {};
auto mid = current_element_->mid;
current_element_->mid = P == message_priority::high
? mid.with_high_priority()
......
......@@ -68,9 +68,8 @@ public:
static_assert(actor_accepts_message(signatures_of<Handle>(), token{}),
"receiver does not accept given message");
auto req_id = dptr()->new_request_id(P);
if (dest)
dest->eq_impl(req_id, dptr()->ctrl(), dptr()->context(),
std::forward<Ts>(xs)...);
dest->eq_impl(req_id, dptr()->ctrl(), dptr()->context(),
std::forward<Ts>(xs)...);
dptr()->request_sync_timeout_msg(timeout, req_id);
return {req_id.response_id(), dptr()};
}
......
......@@ -70,8 +70,6 @@ public:
response_to(signatures_of<Dest>(),
token{})),
"this actor does not accept the response message");
if (! dest)
return;
dest->eq_impl(message_id::make(P), this->ctrl(),
this->context(), std::forward<Ts>(xs)...);
}
......@@ -97,8 +95,6 @@ public:
response_to(signatures_of<Dest>(),
token{})),
"this actor does not accept the response message");
if (! dest)
return;
this->system().scheduler().delayed_send(
rtime, this->ctrl(), actor_cast<strong_actor_ptr>(dest),
message_id::make(P), make_message(std::forward<Ts>(xs)...));
......
......@@ -44,9 +44,7 @@ public:
explicit abstract_coordinator(actor_system& sys);
/// Returns a handle to the central printing actor.
inline actor printer() const {
return printer_;
}
actor printer() const;
/// Puts `what` into the queue of a randomly chosen worker.
virtual void enqueue(resumable* what) = 0;
......@@ -92,8 +90,8 @@ protected:
// configured number of workers
size_t num_workers_;
actor timer_;
actor printer_;
strong_actor_ptr timer_;
strong_actor_ptr printer_;
actor_system& system_;
};
......
......@@ -42,6 +42,9 @@ public:
// tell actor_cast which semantic this type uses
static constexpr bool has_weak_ptr_semantics = false;
// tell actor_cast this is a non-null handle type
static constexpr bool has_non_null_guarantee = true;
scoped_actor(const scoped_actor&) = delete;
scoped_actor(actor_system& sys, bool hide_actor = false);
......@@ -65,10 +68,6 @@ public:
blocking_actor* ptr() const;
inline explicit operator bool() const {
return true;
}
private:
inline actor_control_block* get() const {
......
......@@ -39,8 +39,6 @@ enum class sec : uint8_t {
request_receiver_down,
/// Indicates that a request message timed out.
request_timeout,
/// Unpublishing failed because the actor is `invalid_actor`.
no_actor_to_unpublish,
/// Unpublishing failed because the actor is not bound to given port.
no_actor_published_at_port,
/// Migration failed because the state of an actor is not serializable.
......
......@@ -56,8 +56,6 @@ void send_as(const Source& src, const Dest& dest, Ts&&... xs) {
response_to(signatures_of<Dest>(),
token{})),
"this actor does not accept the response message");
if (! dest)
return;
dest->eq_impl(message_id::make(P), actor_cast<strong_actor_ptr>(src),
nullptr, std::forward<Ts>(xs)...);
}
......@@ -66,21 +64,29 @@ void send_as(const Source& src, const Dest& dest, Ts&&... xs) {
template <message_priority P = message_priority::normal,
class Dest = actor, class... Ts>
void anon_send(const Dest& dest, Ts&&... xs) {
send_as<P>(actor{}, dest, std::forward<Ts>(xs)...);
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
static_assert(actor_accepts_message(signatures_of<Dest>(), token{}),
"receiver does not accept given message");
dest->eq_impl(message_id::make(P), nullptr, nullptr, std::forward<Ts>(xs)...);
}
/// Anonymously sends `dest` an exit message.
template <class Dest>
void anon_send_exit(const Dest& dest, exit_reason reason) {
if (! dest)
return;
dest->enqueue(nullptr, message_id::make(),
make_message(exit_msg{invalid_actor_addr, reason}), nullptr);
make_message(exit_msg{dest->address(), reason}), nullptr);
}
/// Anonymously sends `to` an exit message.
inline void anon_send_exit(const actor_addr& to, exit_reason reason) {
anon_send_exit(actor_cast<actor>(to), reason);
auto ptr = actor_cast<strong_actor_ptr>(to);
if (ptr)
anon_send_exit(ptr, reason);
}
} // namespace caf
......
......@@ -24,6 +24,7 @@
#include <cstdint>
#include <type_traits>
#include "caf/fwd.hpp"
#include "caf/group.hpp"
#include "caf/actor_addr.hpp"
#include "caf/deep_to_string.hpp"
......@@ -36,11 +37,34 @@ namespace caf {
/// Sent to all links when an actor is terminated.
/// @note Actors can override the default handler by calling
/// `self->set_exit_handler(...)`.
struct exit_msg {
class exit_msg {
public:
// -- friend types that need access to private ctors -------------------------
template <class>
friend class data_processor;
template <class>
friend class type_erased_value_impl;
// -- constructors -----------------------------------------------------------
inline exit_msg(actor_addr x, error y)
: source(std::move(x)),
reason(std::move(y)) {
// nop
}
// -- data members -----------------------------------------------------------
/// The source of this message, i.e., the terminated actor.
actor_addr source;
/// The exit reason of the terminated actor.
error reason;
private:
exit_msg() = default;
};
inline std::string to_string(const exit_msg& x) {
......@@ -54,11 +78,34 @@ void serialize(Processor& proc, exit_msg& x, const unsigned int) {
}
/// Sent to all actors monitoring an actor when it is terminated.
struct down_msg {
class down_msg {
public:
// -- friend types that need access to private ctors -------------------------
template <class>
friend class data_processor;
template <class>
friend class type_erased_value_impl;
// -- constructors -----------------------------------------------------------
inline down_msg(actor_addr x, error y)
: source(std::move(x)),
reason(std::move(y)) {
// nop
}
// -- data members -----------------------------------------------------------
/// The source of this message, i.e., the terminated actor.
actor_addr source;
/// The exit reason of the terminated actor.
error reason;
private:
down_msg() = default;
};
inline std::string to_string(const down_msg& x) {
......
......@@ -22,7 +22,6 @@
#include "caf/intrusive_ptr.hpp"
#include "caf/actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/replies_to.hpp"
......@@ -31,6 +30,7 @@
#include "caf/stateful_actor.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/unsafe_actor_handle_init.hpp"
#include "caf/decorator/adapter.hpp"
#include "caf/decorator/splitter.hpp"
......@@ -58,10 +58,7 @@ class typed_broker;
template <class... Sigs>
class typed_actor : detail::comparable<typed_actor<Sigs...>>,
detail::comparable<typed_actor<Sigs...>, actor_addr>,
detail::comparable<typed_actor<Sigs...>, strong_actor_ptr>,
detail::comparable<typed_actor<Sigs...>, invalid_actor_t>,
detail::comparable<typed_actor<Sigs...>,
invalid_actor_addr_t> {
detail::comparable<typed_actor<Sigs...>, strong_actor_ptr> {
public:
static_assert(sizeof...(Sigs) > 0, "Empty typed actor handle");
......@@ -79,6 +76,9 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
// tell actor_cast which semantic this type uses
static constexpr bool has_weak_ptr_semantics = false;
// tell actor_cast this is a non-null handle type
static constexpr bool has_non_null_guarantee = true;
/// Creates a new `typed_actor` type by extending this one with `Es...`.
template <class... Es>
using extend = typed_actor<Sigs..., Es...>;
......@@ -126,7 +126,8 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
using stateful_broker_pointer =
stateful_actor<State, broker_base>*;
typed_actor() = default;
typed_actor() = delete;
typed_actor(typed_actor&&) = default;
typed_actor(const typed_actor&) = default;
typed_actor& operator=(typed_actor&&) = default;
......@@ -150,7 +151,7 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
typename TypedActor::signatures())
>::type>
typed_actor(TypedActor* ptr) : ptr_(ptr->ctrl()) {
// nop
CAF_ASSERT(ptr != nullptr);
}
template <class TypedActor,
......@@ -174,37 +175,23 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
// nop
}
typed_actor(const invalid_actor_t&) {
typed_actor(const unsafe_actor_handle_init_t&) {
// nop
}
typed_actor& operator=(const invalid_actor_t&) {
ptr_.reset();
}
/// Queries the address of the stored actor.
actor_addr address() const noexcept {
return {ptr_.get(), true};
}
/// Returns `*this != invalid_actor`.
explicit operator bool() const noexcept {
return static_cast<bool>(ptr_);
}
/// Returns `*this == invalid_actor`.
bool operator!() const noexcept {
return !ptr_;
}
/// Returns the origin node of this actor.
node_id node() const noexcept {
return ptr_ ? ptr_->node() : invalid_node_id;
return ptr_->node();
}
/// Returns the ID of this actor.
actor_id id() const noexcept {
return ptr_ ? ptr_->id() : invalid_actor_id;
return ptr_->id();
}
/// Exchange content of `*this` and `other`.
......@@ -219,14 +206,18 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
typename std::decay<Ts>::type...
>::type
bind(Ts&&... xs) const {
if (! ptr_)
return invalid_actor;
auto& sys = *(ptr_->home_system);
auto ptr = make_actor<decorator::adapter, strong_actor_ptr>(
sys.next_actor_id(), sys.node(), &sys, ptr_, make_message(xs...));
return {ptr.release(), false};
}
/// Queries whether this object was constructed using
/// `unsafe_actor_handle_init` or is in moved-from state.
bool unsafe() const {
return ! ptr_;
}
/// @cond PRIVATE
abstract_actor* operator->() const noexcept {
......@@ -249,14 +240,6 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
return actor_addr::compare(get(), actor_cast<actor_control_block*>(x));
}
intptr_t compare(const invalid_actor_t&) const noexcept {
return ptr_ ? 1 : 0;
}
intptr_t compare(const invalid_actor_addr_t&) const noexcept {
return ptr_ ? 1 : 0;
}
typed_actor(actor_control_block* ptr, bool add_ref) : ptr_(ptr, add_ref) {
// nop
}
......@@ -270,6 +253,12 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
return to_string(x.ptr_);
}
/// Releases the reference held by handle `x`. Using the
/// handle after invalidating it is undefined behavior.
friend void invalidate(typed_actor& x) {
x.ptr_.reset();
}
/// @endcond
private:
......@@ -318,10 +307,8 @@ operator*(typed_actor<Xs...> f, typed_actor<Ys...> g) {
detail::type_list<Xs...>,
Ys...
>::type;
if (! f || ! g)
return invalid_actor;
auto& sys = g->home_system();
auto mts = sys.message_types(result{});
auto mts = sys.message_types(detail::type_list<result>{});
return make_actor<decorator::sequencer, result>(
sys.next_actor_id(), sys.node(), &sys,
actor_cast<strong_actor_ptr>(std::move(f)),
......@@ -343,11 +330,8 @@ splice(const typed_actor<Xs...>& x, const Ts&... xs) {
>::type;
std::vector<strong_actor_ptr> tmp{actor_cast<strong_actor_ptr>(x),
actor_cast<strong_actor_ptr>(xs)...};
for (auto& addr : tmp)
if (! addr)
return invalid_actor;
auto& sys = x->home_system();
auto mts = sys.message_types(result{});
auto mts = sys.message_types(detail::type_list<result>{});
return make_actor<decorator::splitter, result>(sys.next_actor_id(),
sys.node(), &sys,
std::move(tmp),
......
......@@ -61,8 +61,8 @@ public:
using behavior_type = typed_behavior<Sigs...>;
std::set<std::string> message_types() const override {
typed_actor<Sigs...> hdl;
return this->system().message_types(hdl);
detail::type_list<typed_actor<Sigs...>> token;
return this->system().message_types(token);
}
void initialize() override {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_UNSAFE_ACTOR_HANDLE_INIT_HPP
#define CAF_UNSAFE_ACTOR_HANDLE_INIT_HPP
namespace caf {
/// Tag type to select the unsafe constructor of actor handles.
struct unsafe_actor_handle_init_t { };
static constexpr unsafe_actor_handle_init_t unsafe_actor_handle_init
= unsafe_actor_handle_init_t{};
} // namespace caf
#endif // CAF_UNSAFE_ACTOR_HANDLE_INIT_HPP
......@@ -47,6 +47,9 @@ public:
// tell actor_cast which semantic this type uses
static constexpr bool has_weak_ptr_semantics = true;
// tell actor_cast this pointer can be null
static constexpr bool has_non_null_guarantee = false;
constexpr weak_intrusive_ptr() : ptr_(nullptr) {
// nop
}
......
......@@ -308,11 +308,15 @@ void printer_loop(blocking_actor* self) {
* implementation of coordinator *
******************************************************************************/
actor abstract_coordinator::printer() const {
return actor_cast<actor>(printer_);
}
void abstract_coordinator::start() {
CAF_LOG_TRACE("");
// launch utility actors
timer_ = system_.spawn<timer_actor, hidden + detached>();
printer_ = system_.spawn<hidden + detached>(printer_loop);
timer_ = actor_cast<strong_actor_ptr>(system_.spawn<timer_actor, hidden + detached>());
printer_ = actor_cast<strong_actor_ptr>(system_.spawn<hidden + detached>(printer_loop));
}
void abstract_coordinator::init(actor_system_config& cfg) {
......
......@@ -41,7 +41,7 @@ actor::actor(const scoped_actor& x) : ptr_(actor_cast<strong_actor_ptr>(x)) {
// nop
}
actor::actor(const invalid_actor_t&) : ptr_(nullptr) {
actor::actor(const unsafe_actor_handle_init_t&) : ptr_(nullptr) {
// nop
}
......@@ -58,17 +58,16 @@ actor& actor::operator=(const scoped_actor& x) {
return *this;
}
actor& actor::operator=(const invalid_actor_t&) {
ptr_.reset();
return *this;
intptr_t actor::compare(const actor& x) const noexcept {
return actor_addr::compare(ptr_.get(), x.ptr_.get());
}
intptr_t actor::compare(const actor& other) const noexcept {
return actor_addr::compare(ptr_.get(), other.ptr_.get());
intptr_t actor::compare(const actor_addr& x) const noexcept {
return actor_addr::compare(ptr_.get(), x.ptr_.get());
}
intptr_t actor::compare(const actor_addr& other) const noexcept {
return actor_addr::compare(ptr_.get(), other.ptr_.get());
intptr_t actor::compare(const strong_actor_ptr& x) const noexcept {
return actor_addr::compare(ptr_.get(), x.get());
}
void actor::swap(actor& other) noexcept {
......@@ -79,25 +78,13 @@ actor_addr actor::address() const noexcept {
return actor_cast<actor_addr>(ptr_);
}
node_id actor::node() const noexcept {
return ptr_ ? ptr_->node() : invalid_node_id;
}
actor_id actor::id() const noexcept {
return ptr_ ? ptr_->id() : invalid_actor_id;
}
actor actor::bind_impl(message msg) const {
if (! ptr_)
return invalid_actor;
auto& sys = *(ptr_->home_system);
return make_actor<decorator::adapter, actor>(sys.next_actor_id(), sys.node(),
&sys, ptr_, std::move(msg));
}
actor operator*(actor f, actor g) {
if (! f || ! g)
return invalid_actor;
auto& sys = f->home_system();
return make_actor<decorator::sequencer, actor>(
sys.next_actor_id(), sys.node(), &sys,
......@@ -106,18 +93,14 @@ actor operator*(actor f, actor g) {
}
actor actor::splice_impl(std::initializer_list<actor> xs) {
if (xs.size() < 2)
return invalid_actor;
CAF_ASSERT(xs.size() >= 2);
actor_system* sys = nullptr;
std::vector<strong_actor_ptr> tmp;
for (auto& x : xs)
if (x) {
if (! sys)
sys = &(x->home_system());
tmp.push_back(actor_cast<strong_actor_ptr>(x));
} else {
return invalid_actor;
}
for (auto& x : xs) {
if (! sys)
sys = &(x->home_system());
tmp.push_back(actor_cast<strong_actor_ptr>(x));
}
return make_actor<decorator::splitter, actor>(sys->next_actor_id(),
sys->node(), sys,
std::move(tmp),
......
......@@ -28,7 +28,7 @@
namespace caf {
actor_addr::actor_addr(const invalid_actor_addr_t&) : ptr_(nullptr) {
actor_addr::actor_addr(const unsafe_actor_handle_init_t&) {
// nop
}
......@@ -41,11 +41,6 @@ actor_addr::actor_addr(actor_control_block* ptr, bool add_ref)
// nop
}
actor_addr actor_addr::operator=(const invalid_actor_addr_t&) {
ptr_.reset();
return *this;
}
intptr_t actor_addr::compare(const actor_control_block* lhs,
const actor_control_block* rhs) {
// invalid actors are always "less" than valid actors
......
......@@ -27,6 +27,10 @@
namespace caf {
actor_addr actor_control_block::address() {
return {this, true};
}
void actor_control_block::enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) {
get()->enqueue(std::move(sender), mid, std::move(content), host);
......@@ -62,6 +66,14 @@ void intrusive_ptr_release(actor_control_block* x) {
}
}
bool operator==(const strong_actor_ptr& x, const abstract_actor* y) {
return x.get() == actor_control_block::from(y);
}
bool operator==(const abstract_actor* x, const strong_actor_ptr& y) {
return actor_control_block::from(x) == y.get();
}
template <class T>
void safe_actor(serializer& sink, T& storage) {
CAF_LOG_TRACE(CAF_ARG(storage));
......
......@@ -176,8 +176,6 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
}
if (msg.match_elements<sys_atom, put_atom, actor>()) {
auto& worker = msg.get_as<actor>(2);
if (worker == invalid_actor)
return true;
worker->attach(default_attachable::make_monitor(worker.address(),
address()));
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
......
......@@ -148,10 +148,10 @@ namespace {
struct kvstate {
using key_type = std::string;
using mapped_type = message;
using subscriber_set = std::unordered_set<actor>;
using subscriber_set = std::unordered_set<strong_actor_ptr>;
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<actor,topic_set> subscribers;
std::unordered_map<strong_actor_ptr, topic_set> subscribers;
const char* name = "caf.config_server";
template <class Processor>
friend void serialize(Processor& proc, kvstate& x, const unsigned int) {
......@@ -166,19 +166,20 @@ void actor_registry::start() {
CAF_LOG_TRACE("");
std::string wildcard = "*";
auto unsubscribe_all = [=](actor subscriber) {
if (! subscriber)
return;
auto& subscribers = self->state.subscribers;
auto i = subscribers.find(subscriber);
auto ptr = actor_cast<strong_actor_ptr>(subscriber);
auto i = subscribers.find(ptr);
if (i == subscribers.end())
return;
for (auto& key : i->second)
self->state.data[key].second.erase(subscriber);
self->state.data[key].second.erase(ptr);
subscribers.erase(i);
};
self->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
unsubscribe_all(actor_cast<actor>(dm.source));
auto ptr = actor_cast<strong_actor_ptr>(dm.source);
if (ptr)
unsubscribe_all(actor_cast<actor>(std::move(ptr)));
});
return {
[=](put_atom, const std::string& key, message& msg) {
......@@ -187,13 +188,17 @@ void actor_registry::start() {
return;
auto& vp = self->state.data[key];
vp.first = std::move(msg);
for (auto& subscriber : vp.second)
for (auto& subscriber_ptr : vp.second) {
// we never put a nullptr in our map
auto subscriber = actor_cast<actor>(subscriber_ptr);
if (subscriber != self->current_sender())
self->send(subscriber, update_atom::value, key, vp.second);
}
// also iterate all subscribers for '*'
for (auto& subscriber : self->state.data[wildcard].second)
if (subscriber != self->current_sender())
self->send(subscriber, update_atom::value, key, vp.second);
self->send(actor_cast<actor>(subscriber), update_atom::value,
key, vp.second);
},
[=](get_atom, std::string& key) -> message {
CAF_LOG_TRACE(CAF_ARG(key));
......@@ -210,7 +215,7 @@ void actor_registry::start() {
: make_message());
},
[=](subscribe_atom, const std::string& key) {
auto subscriber = actor_cast<actor>(self->current_sender());
auto subscriber = actor_cast<strong_actor_ptr>(self->current_sender());
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(subscriber));
if (! subscriber)
return;
......@@ -225,12 +230,12 @@ void actor_registry::start() {
}
},
[=](unsubscribe_atom, const std::string& key) {
auto subscriber = actor_cast<actor>(self->current_sender());
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(subscriber));
auto subscriber = actor_cast<strong_actor_ptr>(self->current_sender());
if (! subscriber)
return;
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(subscriber));
if (key == wildcard) {
unsubscribe_all(actor_cast<actor>(subscriber));
unsubscribe_all(actor_cast<actor>(std::move(subscriber)));
return;
}
self->state.subscribers[subscriber].erase(key);
......@@ -277,8 +282,8 @@ void actor_registry::stop() {
// the scheduler is already stopped -> invoke exit messages manually
dropping_execution_unit dummy{&system_};
for (auto& kvp : named_entries_) {
auto mp = mailbox_element::make(nullptr,invalid_message_id, {},
exit_msg{invalid_actor_addr,
auto mp = mailbox_element::make(nullptr, invalid_message_id, {},
exit_msg{kvp.second->address(),
exit_reason::kill});
auto ptr = static_cast<local_actor*>(actor_cast<abstract_actor*>(kvp.second));
ptr->exec_single_event(&dummy, mp);
......
......@@ -112,7 +112,7 @@ size_t blocking_actor::attach_functor(const strong_actor_ptr& ptr) {
using wait_for_atom = atom_constant<atom("waitFor")>;
if (! ptr)
return 0;
auto self = actor_cast<actor>(this);
actor self{this};
ptr->get()->attach_functor([=](const error&) {
anon_send(self, wait_for_atom::value);
});
......
......@@ -29,20 +29,17 @@ namespace caf {
forwarding_actor_proxy::forwarding_actor_proxy(actor_config& cfg, actor mgr)
: actor_proxy(cfg),
manager_(mgr) {
CAF_ASSERT(mgr != invalid_actor);
// nop
}
forwarding_actor_proxy::~forwarding_actor_proxy() {
anon_send(manager_, make_message(delete_atom::value, node(), id()));
if (! manager_.unsafe())
anon_send(manager_, make_message(delete_atom::value, node(), id()));
}
actor forwarding_actor_proxy::manager() const {
actor result;
{
shared_lock<detail::shared_spinlock> guard_(manager_mtx_);
result = manager_;
}
return result;
shared_lock<detail::shared_spinlock> guard_(manager_mtx_);
return manager_;
}
void forwarding_actor_proxy::manager(actor new_manager) {
......@@ -57,11 +54,11 @@ void forwarding_actor_proxy::forward_msg(strong_actor_ptr sender,
<< CAF_ARG(mid) << CAF_ARG(msg));
forwarding_stack tmp;
shared_lock<detail::shared_spinlock> guard_(manager_mtx_);
if (manager_)
if (! manager_.unsafe())
manager_->enqueue(nullptr, invalid_message_id,
make_message(forward_atom::value, std::move(sender),
fwd ? *fwd : tmp,
actor_cast<strong_actor_ptr>(this),
strong_actor_ptr{ctrl()},
mid, std::move(msg)),
nullptr);
}
......@@ -125,7 +122,7 @@ void forwarding_actor_proxy::local_unlink_from(abstract_actor* other) {
void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) {
CAF_ASSERT(ctx != nullptr);
manager(invalid_actor);
actor tmp{std::move(manager_)}; // manually break cycle
cleanup(std::move(rsn), ctx);
}
......
......@@ -115,14 +115,14 @@ public:
void stop() override {
CAF_LOG_TRACE("");
await_all_locals_down(system(), {broker_});
broker_ = invalid_actor;
actor tmp{std::move(broker_)}; // manually break cycle
}
const actor& broker() const {
return broker_;
}
local_group(actor_system& sys, bool spawn_local_broker,
local_group(actor_system& sys, optional<actor> local_broker,
local_group_module* mod, std::string id, const node_id& nid);
~local_group();
......@@ -163,28 +163,26 @@ public:
set_default_handler(fwd);
set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
if (dm.source) {
auto first = acquaintances_.begin();
auto last = acquaintances_.end();
auto i = std::find_if(first, last, [&](const actor& a) {
return a == dm.source;
});
if (i != last)
acquaintances_.erase(i);
}
auto first = acquaintances_.begin();
auto last = acquaintances_.end();
auto i = std::find_if(first, last, [&](const actor& a) {
return a == dm.source;
});
if (i != last)
acquaintances_.erase(i);
});
// return behavior
return {
[=](join_atom, const actor& other) {
CAF_LOG_TRACE(CAF_ARG(other));
if (other && acquaintances_.insert(other).second) {
if (acquaintances_.insert(other).second) {
monitor(other);
}
},
[=](leave_atom, const actor& other) {
CAF_LOG_TRACE(CAF_ARG(other));
acquaintances_.erase(other);
if (other && acquaintances_.erase(other) > 0)
if (acquaintances_.erase(other) > 0)
demonitor(other);
},
[=](forward_atom, const message& what) {
......@@ -243,13 +241,10 @@ public:
template <class... Ts>
local_group_proxy(actor_system& sys, actor remote_broker, Ts&&... xs)
: super(sys, false, std::forward<Ts>(xs)...) {
CAF_ASSERT(broker_ == invalid_actor);
CAF_ASSERT(remote_broker != invalid_actor);
CAF_LOG_TRACE(CAF_ARG(remote_broker));
broker_ = std::move(remote_broker);
proxy_broker_ = system().spawn<proxy_broker, hidden>(this);
monitor_ = system().spawn<hidden>(broker_monitor_actor, this);
: super(sys, std::move(remote_broker), std::forward<Ts>(xs)...),
proxy_broker_{sys.spawn<proxy_broker, hidden>(this)},
monitor_{sys.spawn<hidden>(broker_monitor_actor, this)} {
// nop
}
bool subscribe(strong_actor_ptr who) override {
......@@ -286,9 +281,9 @@ public:
void stop() override {
CAF_LOG_TRACE("");
await_all_locals_down(system_, {monitor_, proxy_broker_, broker_});
monitor_ = invalid_actor;
proxy_broker_ = invalid_actor;
broker_ = invalid_actor;
invalidate(monitor_);
invalidate(proxy_broker_);
invalidate(broker_);
}
private:
......@@ -346,7 +341,7 @@ public:
auto i = instances_.find(identifier);
if (i != instances_.end())
return {i->second};
auto tmp = make_counted<local_group>(system(), true, this, identifier,
auto tmp = make_counted<local_group>(system(), none, this, identifier,
system().node());
upgrade_to_unique_guard uguard(guard);
auto p = instances_.emplace(identifier, tmp);
......@@ -362,18 +357,19 @@ public:
CAF_LOG_TRACE("");
// deserialize identifier and broker
std::string identifier;
actor broker;
source >> identifier >> broker;
CAF_LOG_DEBUG(CAF_ARG(identifier) << CAF_ARG(broker));
if (! broker)
strong_actor_ptr broker_ptr;
source >> identifier >> broker_ptr;
CAF_LOG_DEBUG(CAF_ARG(identifier) << CAF_ARG(broker_ptr));
if (! broker_ptr)
return invalid_group;
auto broker = actor_cast<actor>(broker_ptr);
if (broker->node() == system().node())
return this->get(identifier);
upgrade_guard guard(proxies_mtx_);
auto i = proxies_.find(broker);
if (i != proxies_.end())
return {i->second};
local_group_ptr tmp = make_counted<local_group_proxy>(system(),broker,
local_group_ptr tmp = make_counted<local_group_proxy>(system(), broker,
this, identifier,
broker->node());
upgrade_to_unique_guard uguard(guard);
......@@ -385,7 +381,7 @@ public:
void save(const local_group* ptr, serializer& sink) const {
CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE("");
sink << ptr->identifier() << ptr->broker();
sink << ptr->identifier() << actor_cast<strong_actor_ptr>(ptr->broker());
}
void stop() override {
......@@ -411,14 +407,12 @@ private:
std::map<actor, local_group_ptr> proxies_;
};
local_group::local_group(actor_system& sys, bool do_spawn,
local_group::local_group(actor_system& sys, optional<actor> lb,
local_group_module* mod, std::string id,
const node_id& nid)
: abstract_group(sys, mod, std::move(id), nid) {
CAF_LOG_TRACE(CAF_ARG(do_spawn) << CAF_ARG(id) << CAF_ARG(nid));
if (do_spawn)
broker_ = sys.spawn<local_broker, hidden>(this);
// else: derived class spawns broker_
: abstract_group(sys, mod, std::move(id), nid),
broker_(lb ? *lb : sys.spawn<local_broker, hidden>(this)) {
CAF_LOG_TRACE(CAF_ARG(id) << CAF_ARG(nid));
}
local_group::~local_group() {
......
......@@ -259,8 +259,7 @@ uint32_t local_actor::request_timeout(const duration& d) {
// immediately enqueue timeout message if duration == 0s
enqueue(ctrl(), invalid_message_id, std::move(msg), context());
else
system().scheduler().delayed_send(d, ctrl(),
actor_cast<strong_actor_ptr>(this),
system().scheduler().delayed_send(d, ctrl(), strong_actor_ptr(ctrl()),
message_id::make(), std::move(msg));
return result;
}
......@@ -985,11 +984,14 @@ void local_actor::do_become(behavior bhvr, bool discard_old) {
}
void local_actor::send_exit(const actor_addr& whom, error reason) {
auto ptr = actor_cast<actor>(whom);
if (! ptr)
send_exit(actor_cast<strong_actor_ptr>(whom), std::move(reason));
}
void local_actor::send_exit(const strong_actor_ptr& dest, error reason) {
if (! dest)
return;
ptr->eq_impl(message_id::make(), nullptr, context(),
exit_msg{address(), std::move(reason)});
dest->get()->eq_impl(message_id::make(), nullptr, context(),
exit_msg{address(), std::move(reason)});
}
const char* local_actor::name() const {
......
......@@ -270,7 +270,7 @@ bool monitorable_actor::handle_system_message(mailbox_element& node,
res = mailbox_element::make(ctrl(), node.mid.response_id(),
{}, std::move(err));
if (res) {
auto s = actor_cast<actor>(node.sender);
auto s = actor_cast<strong_actor_ptr>(node.sender);
if (s)
s->enqueue(std::move(res), context);
}
......
......@@ -30,7 +30,6 @@ const char* sec_strings[] = {
"unexpected_response",
"request_receiver_down",
"request_timeout",
"no_actor_to_unpublish",
"no_actor_published_at_port",
"state_not_serializable",
"unsupported_sys_key",
......
......@@ -53,7 +53,6 @@ struct fixture {
CAF_REQUIRE(res);
CAF_CHECK(ifs.empty());
auto aut = actor_cast<actor>(res);
CAF_REQUIRE(aut != invalid_actor);
self->wait_for(aut);
CAF_MESSAGE("aut done");
}
......
......@@ -87,17 +87,14 @@ CAF_TEST(round_robin_actor_pool) {
self->request(pool, infinite, i, i).receive(
[&](int res) {
CAF_CHECK_EQUAL(res, i + i);
auto sender = self->current_sender();
workers.push_back(actor_cast<actor>(sender));
auto sender = actor_cast<strong_actor_ptr>(self->current_sender());
CAF_REQUIRE(sender);
workers.push_back(actor_cast<actor>(std::move(sender)));
}
);
}
CAF_CHECK_EQUAL(workers.size(), 6u);
CAF_CHECK(std::unique(workers.begin(), workers.end()) == workers.end());
auto is_invalid = [](const actor& x) {
return x == invalid_actor;
};
CAF_CHECK(std::none_of(workers.begin(), workers.end(), is_invalid));
self->request(pool, infinite, sys_atom::value, get_atom::value).receive(
[&](std::vector<actor>& ws) {
std::sort(workers.begin(), workers.end());
......
......@@ -42,7 +42,10 @@ struct fixture {
actor mirror;
actor testee;
fixture() : self(system), mirror(system.spawn(mirror_impl)) {
fixture()
: self(system),
mirror(system.spawn(mirror_impl)),
testee(unsafe_actor_handle_init) {
self->set_down_handler([](local_actor*, down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
});
......
......@@ -52,7 +52,10 @@ CAF_TEST(constructor_attach) {
};
class spawner : public event_based_actor {
public:
spawner(actor_config& cfg) : event_based_actor(cfg), downs_(0) {
spawner(actor_config& cfg)
: event_based_actor(cfg),
downs_(0),
testee_(spawn<testee, monitored>(this)) {
set_down_handler([=](down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
CAF_CHECK_EQUAL(msg.source, testee_.address());
......@@ -62,7 +65,6 @@ CAF_TEST(constructor_attach) {
}
behavior make_behavior() {
testee_ = spawn<testee, monitored>(this);
return {
[=](done_atom, const error& reason) {
CAF_CHECK_EQUAL(reason, exit_reason::user_shutdown);
......@@ -77,7 +79,7 @@ CAF_TEST(constructor_attach) {
}
void on_exit() {
testee_ = invalid_actor;
invalidate(testee_);
}
private:
......
......@@ -514,7 +514,7 @@ CAF_TEST(constructor_attach) {
}
void on_exit() {
buddy_ = invalid_actor;
invalidate(buddy_);
}
private:
......@@ -522,7 +522,10 @@ CAF_TEST(constructor_attach) {
};
class spawner : public event_based_actor {
public:
spawner(actor_config& cfg) : event_based_actor(cfg), downs_(0) {
spawner(actor_config& cfg)
: event_based_actor(cfg),
downs_(0),
testee_(spawn<testee, monitored>(this)) {
set_down_handler([=](down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
if (++downs_ == 2)
......@@ -532,8 +535,8 @@ CAF_TEST(constructor_attach) {
send_exit(testee_, std::move(msg.reason));
});
}
behavior make_behavior() {
testee_ = spawn<testee, monitored>(this);
return {
[=](ok_atom, const error& reason) {
CAF_CHECK_EQUAL(reason, exit_reason::user_shutdown);
......@@ -545,7 +548,7 @@ CAF_TEST(constructor_attach) {
void on_exit() {
CAF_MESSAGE("spawner::on_exit()");
testee_ = invalid_actor;
invalidate(testee_);
}
private:
......
......@@ -93,7 +93,9 @@ using config_value = detail::parse_ini_t::value;
struct fixture {
actor_system system;
fixture() : system(test::engine::argc(), test::engine::argv()) {
fixture()
: system(test::engine::argc(), test::engine::argv()),
config_server(unsafe_actor_handle_init) {
// nop
}
......@@ -108,7 +110,6 @@ struct fixture {
void load_to_config_server(const char* str) {
config_server = actor_cast<actor>(system.registry().get(atom("ConfigServ")));
CAF_REQUIRE(config_server != invalid_actor);
// clear config
scoped_actor self{system};
self->request(config_server, infinite, get_atom::value, "*").receive(
......@@ -165,7 +166,7 @@ struct fixture {
template <class T>
bool value_is(const char* key, const T& what) {
if (config_server != invalid_actor)
if (! config_server.unsafe())
return config_server_has(key, what);
auto& cv = values[key];
using type =
......@@ -183,7 +184,7 @@ struct fixture {
}
size_t num_values() {
if (config_server != invalid_actor) {
if (! config_server.unsafe()) {
size_t result = 0;
scoped_actor self{system};
self->request(config_server, infinite, get_atom::value, "*").receive(
......
......@@ -390,7 +390,7 @@ CAF_TEST(client_server_worker_user_case) {
self->request(serv, infinite, request_atom::value).receive(
[&](response_atom) {
CAF_MESSAGE("received 'response'");
CAF_CHECK_EQUAL(self->current_sender(), work.address());
CAF_CHECK_EQUAL(self->current_sender(), work);
},
[&](const error& err) {
CAF_ERROR("error: " << self->system().render(err));
......
......@@ -53,15 +53,17 @@ behavior untyped_second_stage() {
struct fixture {
actor_system system;
scoped_actor self{system, true};
scoped_actor self;
actor first;
actor second;
actor first_and_second;
~fixture() {
anon_send_exit(first, exit_reason::kill);
anon_send_exit(second, exit_reason::kill);
fixture()
: self(system, true),
first(unsafe_actor_handle_init),
second(unsafe_actor_handle_init),
first_and_second(unsafe_actor_handle_init) {
// nop
}
void init_untyped() {
......
......@@ -54,8 +54,9 @@ public:
template <class Handle>
uint16_t publish(Handle&& whom, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) {
detail::type_list<typename std::decay<Handle>::type> tk;
return publish(actor_cast<strong_actor_ptr>(std::forward<Handle>(whom)),
system().message_types(whom), port, in, reuse_addr);
system().message_types(tk), port, in, reuse_addr);
}
/// Makes *all* local groups accessible via network
......@@ -69,8 +70,6 @@ public:
/// @param port TCP port.
template <class Handle>
void unpublish(const Handle& whom, uint16_t port = 0) {
if (! whom)
return;
unpublish(whom.address(), port);
}
......@@ -83,10 +82,11 @@ public:
/// to an untyped otherwise unexpected actor.
template <class ActorHandle>
ActorHandle typed_remote_actor(std::string host, uint16_t port) {
ActorHandle hdl;
auto res = remote_actor(system().message_types(hdl),
std::move(host), port);
return actor_cast<ActorHandle>(res);
detail::type_list<ActorHandle> tk;
auto x = remote_actor(system().message_types(tk), std::move(host), port);
if (! x)
throw std::runtime_error("received nullptr from middleman");
return actor_cast<ActorHandle>(std::move(x));
}
/// Establish a new connection to the actor at `host` on given `port`.
......@@ -179,11 +179,13 @@ public:
add_hook<impl>(std::move(fun));
}
/*
/// Returns the actor associated with `name` at `nid` or
/// `invalid_actor` if `nid` is not connected or has no actor
/// associated to this `name`.
/// @warning Blocks the caller until `nid` responded to the lookup.
actor remote_lookup(atom_value name, const node_id& nid);
*/
/// Smart pointer for `network::multiplexer`.
using backend_pointer = std::unique_ptr<network::multiplexer>;
......
......@@ -89,7 +89,7 @@ public:
/// @cond PRIVATE
std::set<std::string> message_types() const override {
typed_actor<Sigs...> hdl;
detail::type_list<typed_actor<Sigs...>> hdl;
return this->system().message_types(hdl);
}
......
......@@ -89,7 +89,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
actor_config cfg;
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>(
aid, nid, &(self->home_system()), cfg, self);
auto selfptr = actor_cast<strong_actor_ptr>(self);
strong_actor_ptr selfptr{self->ctrl()};
res->get()->attach_functor([=](const error& rsn) {
mm->backend().post([=] {
// using res->id() instead of aid keeps this actor instance alive
......@@ -129,15 +129,12 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
auto cleanup = detail::make_scope_guard([&] {
cb = none;
});
strong_actor_ptr ptr;
if (aid == invalid_actor_id) {
// can occur when connecting to the default port of a node
cb->deliver(make_message(ok_atom::value,
nid,
actor_addr{invalid_actor_addr},
std::move(sigs)));
cb->deliver(ok_atom::value, nid, ptr, std::move(sigs));
return;
}
strong_actor_ptr ptr;
if (nid == this_node()) {
// connected to self
ptr = actor_cast<strong_actor_ptr>(system().registry().get(aid));
......@@ -190,7 +187,7 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
// kill immediately if actor has already terminated
send_kill_proxy_instance(exit_reason::unknown);
} else {
auto tmp = actor_cast<strong_actor_ptr>(self); // keep broker alive ...
strong_actor_ptr tmp{self->ctrl()};
auto mm = &system().middleman();
ptr->get()->attach_functor([=](const error& fail_state) {
mm->backend().dispatch([=] {
......@@ -262,8 +259,11 @@ void basp_broker_state::deliver(const node_id& src_nid,
void basp_broker_state::learned_new_node(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(nid));
auto& tmp = spawn_servers[nid];
tmp = system().spawn<hidden>([=](event_based_actor* this_actor) -> behavior {
if (spawn_servers.count(nid) > 0) {
CAF_LOG_ERROR("learned_new_node called for known node " << CAF_ARG(nid));
return;
}
auto tmp = system().spawn<hidden>([=](event_based_actor* this_actor) -> behavior {
CAF_LOG_TRACE("");
// terminate when receiving a down message
this_actor->set_down_handler([=](down_msg& dm) {
......@@ -278,14 +278,17 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(config_serv_addr));
// drop unexpected messages from this point on
this_actor->set_default_handler(print_and_drop);
auto config_serv = actor_cast<actor>(config_serv_addr);
auto config_serv = actor_cast<strong_actor_ptr>(config_serv_addr);
if (! config_serv)
return;
this_actor->monitor(config_serv);
this_actor->become(
[=](spawn_atom, std::string& type, message& args)
-> delegated<ok_atom, strong_actor_ptr, std::set<std::string>> {
CAF_LOG_TRACE(CAF_ARG(type) << CAF_ARG(args));
this_actor->delegate(config_serv, get_atom::value,
std::move(type), std::move(args));
this_actor->delegate(actor_cast<actor>(std::move(config_serv)),
get_atom::value, std::move(type),
std::move(args));
return {};
}
);
......@@ -296,6 +299,7 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
}
};
});
spawn_servers.emplace(nid, tmp);
using namespace detail;
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([](serializer& sink) {
......@@ -471,10 +475,10 @@ behavior basp_broker::make_behavior() {
if (ctx.callback) {
CAF_LOG_WARNING("failed to handshake with remote node"
<< CAF_ARG(msg.handle));
ctx.callback->deliver(make_message(ok_atom::value,
ctx.callback->deliver(ok_atom::value,
node_id{invalid_node_id},
actor_addr{invalid_actor_addr},
std::set<std::string>{}));
strong_actor_ptr{nullptr},
std::set<std::string>{});
}
close(msg.handle);
state.ctx.erase(msg.handle);
......@@ -613,8 +617,6 @@ behavior basp_broker::make_behavior() {
},
[=](unpublish_atom, const actor_addr& whom, uint16_t port) -> result<void> {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
if (whom == invalid_actor_addr)
return sec::no_actor_to_unpublish;
auto cb = make_callback([&](const strong_actor_ptr&, uint16_t x) {
try { close(hdl_by_port(x)); }
catch (std::exception&) { }
......
......@@ -103,6 +103,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
middleman::middleman(actor_system& sys)
: system_(sys),
manager_(unsafe_actor_handle_init),
heartbeat_interval_(0),
enable_automatic_connections_(false) {
// nop
......@@ -243,19 +244,18 @@ group middleman::remote_group(const std::string& group_identifier,
return result;
}
/*
actor middleman::remote_lookup(atom_value name, const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(nid));
auto basp = named_broker<basp_broker>(atom("BASP"));
if (! basp)
return invalid_actor;
actor result;
scoped_actor self{system(), true};
try {
self->send(basp, forward_atom::value, self.address(), nid, name,
make_message(sys_atom::value, get_atom::value, "info"));
self->receive(
[&](ok_atom, const std::string& /* key == "info" */,
const actor_addr& addr, const std::string& /* name */) {
[&](ok_atom, const std::string&,
const actor_addr& addr, const std::string&) {
result = actor_cast<actor>(addr);
},
after(std::chrono::minutes(5)) >> [] {
......@@ -267,6 +267,7 @@ actor middleman::remote_lookup(atom_value name, const node_id& nid) {
}
return result;
}
*/
void middleman::start() {
CAF_LOG_TRACE("");
......
......@@ -61,7 +61,7 @@ public:
void on_exit() override {
CAF_LOG_TRACE("");
broker_ = invalid_actor;
invalidate(broker_);
}
const char* name() const override {
......
......@@ -332,10 +332,10 @@ public:
message msg;
source >> stages >> msg;
auto src = actor_cast<strong_actor_ptr>(registry_->get(hdr.source_actor));
auto dest = actor_cast<actor>(registry_->get(hdr.dest_actor));
auto dest = registry_->get(hdr.dest_actor);
CAF_REQUIRE(dest);
dest->enqueue(mailbox_element::make(src, message_id::make(),
std::move(stages), std::move(msg)),
std::move(stages), std::move(msg)),
nullptr);
}
......@@ -566,7 +566,7 @@ CAF_TEST(remote_actor_and_send) {
mpx()->provide_scribe(lo, 4242, remote_hdl(0));
CAF_REQUIRE(mpx()->pending_scribes().count(make_pair(lo, 4242)) == 1);
auto mm1 = system.middleman().actor_handle();
actor result;
actor result{unsafe_actor_handle_init};
auto f = self()->request(mm1, infinite,
connect_atom::value, lo, uint16_t{4242});
// wait until BASP broker has received and processed the connect message
......
......@@ -88,7 +88,6 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
CAF_MESSAGE("peer_fun called");
CAF_REQUIRE(self->subtype() == resumable::io_actor);
CAF_CHECK(self != nullptr);
CAF_CHECK(buddy != invalid_actor);
self->monitor(buddy);
// assume exactly one connection
CAF_REQUIRE(self->connections().size() == 1);
......
......@@ -172,7 +172,8 @@ behavior server(broker* self) {
class fixture {
public:
fixture() : system(actor_system_config{}
.load<io::middleman, network::test_multiplexer>()) {
.load<io::middleman, network::test_multiplexer>()),
aut_(unsafe_actor_handle_init) {
mpx_ = dynamic_cast<network::test_multiplexer*>(&system.middleman().backend());
CAF_REQUIRE(mpx_ != nullptr);
// spawn the actor-under-test
......
......@@ -123,15 +123,13 @@ CAF_TEST(identity_semantics) {
auto server = server_side.spawn(make_pong_behavior);
auto port1 = server_side_mm.publish(server, 0, local_host);
auto port2 = server_side_mm.publish(server, 0, local_host);
CAF_REQUIRE(port1 != port2);
CAF_REQUIRE_NOT_EQUAL(port1, port2);
auto same_server = server_side_mm.remote_actor(local_host, port2);
CAF_REQUIRE_EQUAL(same_server, server);
CAF_CHECK_EQUAL(same_server->node(), server_side.node());
// client side
auto server1 = client_side_mm.remote_actor(local_host, port1);
auto server2 = client_side_mm.remote_actor(local_host, port2);
CAF_REQUIRE(server1);
CAF_REQUIRE(server2);
CAF_CHECK_EQUAL(server1, client_side_mm.remote_actor(local_host, port1));
CAF_CHECK_EQUAL(server2, client_side_mm.remote_actor(local_host, port2));
// cleanup
......@@ -144,7 +142,6 @@ CAF_TEST(ping_pong) {
0, local_host);
// client side
auto pong = client_side_mm.remote_actor(local_host, port);
CAF_REQUIRE(pong);
client_side.spawn(make_ping_behavior, pong);
}
......@@ -154,7 +151,6 @@ CAF_TEST(custom_message_type) {
0, local_host);
// client side
auto sorter = client_side_mm.remote_actor(local_host, port);
CAF_REQUIRE(sorter);
client_side.spawn(make_sort_requester_behavior, sorter);
}
......
......@@ -133,7 +133,6 @@ CAF_TEST(server_side_group_comm) {
CAF_REQUIRE(port != 0);
// client side
auto server = client_side_mm.remote_actor(local_host, port);
CAF_REQUIRE(server);
scoped_actor group_resolver(client_side, true);
group grp;
group_resolver->request(server, infinite, get_group_atom::value).receive(
......@@ -151,7 +150,6 @@ CAF_TEST(client_side_group_comm) {
CAF_REQUIRE(port != 0);
// client side
auto server = client_side_mm.remote_actor(local_host, port);
CAF_REQUIRE(server);
client_side.spawn(make_client_behavior, server,
client_side.groups().get("local", "foobar"));
}
......
......@@ -57,6 +57,11 @@ behavior client(event_based_actor* self, actor serv) {
struct server_state {
actor client; // the spawn we connect to the server in our main
actor aut; // our mirror
server_state()
: client(unsafe_actor_handle_init),
aut(unsafe_actor_handle_init) {
// nop
}
};
behavior server(stateful_actor<server_state>* self) {
......@@ -65,12 +70,14 @@ behavior server(stateful_actor<server_state>* self) {
[=](ok_atom) {
CAF_LOG_TRACE("");
auto s = self->current_sender();
CAF_REQUIRE(s != invalid_actor_addr);
CAF_REQUIRE(self->node() != s.node());
self->state.client = actor_cast<actor>(s);
CAF_REQUIRE(s != nullptr);
CAF_REQUIRE(self->node() != s->node());
auto opt = actor_cast<actor>(s);
//CAF_REQUIRE(opt);
self->state.client = opt;
auto mm = self->system().middleman().actor_handle();
self->request(mm, infinite, spawn_atom::value,
s.node(), "mirror", make_message()).then(
s->node(), "mirror", make_message()).then(
[=](ok_atom, const strong_actor_ptr& ptr,
const std::set<std::string>& ifs) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(ifs));
......@@ -104,7 +111,6 @@ void run_client(int argc, char** argv, uint16_t port) {
.add_actor_type("mirror", mirror);
actor_system system{cfg};
auto serv = system.middleman().remote_actor("localhost", port);
CAF_REQUIRE(serv);
system.spawn(client, serv);
}
......
......@@ -97,8 +97,6 @@ behavior pong(event_based_actor* self) {
peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl,
const actor& buddy) {
CAF_MESSAGE("peer_fun called");
CAF_CHECK(self != nullptr);
CAF_CHECK(buddy != invalid_actor);
self->monitor(buddy);
// assume exactly one connection
CAF_REQUIRE_EQUAL(self->connections().size(), 1u);
......@@ -165,7 +163,6 @@ void run_client(int argc, char** argv, uint16_t port) {
auto p = system.spawn(ping, size_t{10});
CAF_MESSAGE("spawn_client_typed...");
auto cl = system.middleman().spawn_client(peer_fun, "localhost", port, p);
CAF_REQUIRE(cl);
CAF_MESSAGE("spawn_client_typed finished");
anon_send(p, kickoff_atom::value, cl);
CAF_MESSAGE("`kickoff_atom` has been send");
......
......@@ -92,7 +92,6 @@ void run_client(int argc, char** argv, uint16_t port) {
CAF_MESSAGE("connect to typed_remote_actor");
auto serv = system.middleman().typed_remote_actor<server_type>("127.0.0.1",
port);
CAF_REQUIRE(serv);
scoped_actor self{system};
self->request(serv, infinite, ping{42}).receive(
[](const pong& p) {
......
......@@ -55,7 +55,7 @@ public:
};
struct fixture {
fixture() {
fixture() : testee(unsafe_actor_handle_init) {
new (&system) actor_system(actor_system_config{test::engine::argc(),
test::engine::argv()}
.load<io::middleman>());
......@@ -64,29 +64,30 @@ struct fixture {
~fixture() {
anon_send_exit(testee, exit_reason::user_shutdown);
testee = invalid_actor;
invalidate(testee);
system.~actor_system();
CAF_CHECK_EQUAL(s_dtor_called.load(), 2);
}
actor remote_actor(const char* hostname, uint16_t port,
bool expect_fail = false) {
actor result;
actor result{unsafe_actor_handle_init};
scoped_actor self{system, true};
self->request(system.middleman().actor_handle(), infinite,
connect_atom::value, hostname, port).receive(
[&](ok_atom, node_id&, strong_actor_ptr& res, std::set<std::string>& xs) {
CAF_REQUIRE(xs.empty());
result = actor_cast<actor>(std::move(res));
if (res)
result = actor_cast<actor>(std::move(res));
},
[&](error&) {
// nop
}
);
if (! expect_fail)
CAF_REQUIRE(result != invalid_actor);
if (expect_fail)
CAF_REQUIRE(result.unsafe());
else
CAF_REQUIRE(result == invalid_actor);
CAF_REQUIRE(! result.unsafe());
return result;
}
......@@ -119,7 +120,7 @@ CAF_TEST(unpublishing) {
down_msg{testee.address(), exit_reason::normal});
// must fail now
auto x2 = remote_actor("127.0.0.1", port, true);
CAF_CHECK(! x2);
CAF_CHECK(x2.unsafe());
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -40,8 +40,8 @@ namespace caf {
namespace test {
template <class T, class U,
class Common = typename std::common_type<T, U>::type,
typename std::enable_if<std::is_floating_point<Common>::value,
typename std::enable_if<std::is_floating_point<T>::value
|| std::is_floating_point<U>::value,
int>::type = 0>
bool equal_to(const T& t, const U& u) {
auto x = static_cast<double>(t);
......@@ -52,8 +52,8 @@ bool equal_to(const T& t, const U& u) {
}
template <class T, class U,
class Common = typename std::common_type<T, U>::type,
typename std::enable_if<! std::is_floating_point<Common>::value,
typename std::enable_if<! std::is_floating_point<T>::value
&& !std::is_floating_point<U>::value,
int>::type = 0>
bool equal_to(const T& x, const U& y) {
return x == y;
......
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