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

Refactor and extend actor_control_block

Add a new member for storing runtime-type information and improve
encapsulation by making all member variables private.
parent e35d7273
......@@ -110,12 +110,12 @@ public:
actor& operator=(const scoped_actor& x);
/// Queries whether this actor handle is valid.
inline explicit operator bool() const {
explicit operator bool() const {
return static_cast<bool>(ptr_);
}
/// Queries whether this actor handle is invalid.
inline bool operator!() const {
bool operator!() const {
return !ptr_;
}
......@@ -123,18 +123,27 @@ public:
actor_addr address() const noexcept;
/// Returns the ID of this actor.
inline actor_id id() const noexcept {
/// @pre `*this != nullptr`
actor_id id() const noexcept {
return ptr_->id();
}
/// Returns the origin node of this actor.
inline node_id node() const noexcept {
/// @pre `*this != nullptr`
node_id node() const noexcept {
return ptr_->node();
}
/// Returns the hosting actor system.
inline actor_system& home_system() const noexcept {
return *ptr_->home_system;
/// @pre `*this != nullptr`
actor_system& system() const noexcept {
return ptr_->system();
}
/// Returns the hosting actor system.
/// @pre `*this != nullptr`
actor_system& home_system() const noexcept {
return ptr_->system();
}
/// Exchange content of `*this` and `other`.
......@@ -142,7 +151,7 @@ public:
/// @cond PRIVATE
inline abstract_actor* operator->() const noexcept {
abstract_actor* operator->() const noexcept {
CAF_ASSERT(ptr_);
return ptr_->get();
}
......@@ -179,11 +188,11 @@ public:
}
private:
inline actor_control_block* get() const noexcept {
actor_control_block* get() const noexcept {
return ptr_.get();
}
inline actor_control_block* release() noexcept {
actor_control_block* release() noexcept {
return ptr_.release();
}
......@@ -219,7 +228,7 @@ bool operator!=(abstract_actor* lhs, const actor& rhs);
namespace std {
template <>
struct hash<caf::actor> {
inline size_t operator()(const caf::actor& ref) const {
size_t operator()(const caf::actor& ref) const {
return static_cast<size_t>(ref ? ref->id() : 0);
}
};
......
......@@ -60,24 +60,33 @@ public:
actor_addr& operator=(std::nullptr_t);
/// Returns the ID of this actor.
inline actor_id id() const noexcept {
/// @pre `*this != nullptr`
actor_id id() const noexcept {
return ptr_->id();
}
/// Returns the origin node of this actor.
inline node_id node() const noexcept {
/// @pre `*this != nullptr`
node_id node() const noexcept {
return ptr_->node();
}
/// Returns the hosting actor system.
inline actor_system& home_system() const noexcept {
return *ptr_->home_system;
/// @pre `*this != nullptr`
actor_system& system() const noexcept {
return ptr_->system();
}
/// Returns the hosting actor system.
/// @pre `*this != nullptr`
actor_system& home_system() const noexcept {
return ptr_->system();
}
/// Exchange content of `*this` and `other`.
void swap(actor_addr& other) noexcept;
inline explicit operator bool() const {
explicit operator bool() const {
return static_cast<bool>(ptr_);
}
......@@ -92,11 +101,11 @@ public:
intptr_t compare(const actor_control_block* other) const noexcept;
inline intptr_t compare(const weak_actor_ptr& other) const noexcept {
intptr_t compare(const weak_actor_ptr& other) const noexcept {
return compare(other.get());
}
inline intptr_t compare(const strong_actor_ptr& other) const noexcept {
intptr_t compare(const strong_actor_ptr& other) const noexcept {
return compare(other.get());
}
......@@ -121,18 +130,18 @@ public:
actor_addr(actor_control_block*, bool);
inline actor_control_block* get() const noexcept {
actor_control_block* get() const noexcept {
return ptr_.get();
}
/// @endcond
private:
inline actor_control_block* release() noexcept {
actor_control_block* release() noexcept {
return ptr_.release();
}
inline actor_control_block* get_locked() const noexcept {
actor_control_block* get_locked() const noexcept {
return ptr_.get_locked();
}
......@@ -163,7 +172,7 @@ inline bool operator!=(std::nullptr_t, const actor_addr& x) {
namespace std {
template <>
struct hash<caf::actor_addr> {
inline size_t operator()(const caf::actor_addr& ref) const {
size_t operator()(const caf::actor_addr& ref) const {
return static_cast<size_t>(ref.id());
}
};
......
......@@ -19,6 +19,7 @@
#pragma once
#include <atomic>
#include <set>
#include "caf/config.hpp"
#include "caf/error.hpp"
......@@ -67,49 +68,62 @@ namespace caf {
/// the last weak reference expires.
class actor_control_block {
public:
using data_destructor = void (*)(abstract_actor*);
using block_destructor = void (*)(actor_control_block*);
// -- member types -----------------------------------------------------------
actor_control_block(actor_id x, node_id& y, actor_system* sys,
data_destructor ddtor, block_destructor bdtor)
: strong_refs(1),
weak_refs(1),
aid(x),
nid(std::move(y)),
home_system(sys),
data_dtor(ddtor),
block_dtor(bdtor) {
// nop
/// Stores runtime-type information of the message passing interface.
using mpi_type = std::set<std::string>;
/// Destroys the pointed-to actor.
using data_deleter = void (*)(abstract_actor*);
/// Destroys the pointed-to actor.
using block_deleter = void (*)(actor_control_block*);
// -- friend functions -------------------------------------------------------
friend bool intrusive_ptr_upgrade_weak(actor_control_block* x);
friend inline void intrusive_ptr_add_weak_ref(actor_control_block* x) {
x->weak_refs_.fetch_add(1, std::memory_order_relaxed);
}
friend void intrusive_ptr_release_weak(actor_control_block* x);
friend inline void intrusive_ptr_add_ref(actor_control_block* x) {
x->strong_refs_.fetch_add(1, std::memory_order_relaxed);
}
friend void intrusive_ptr_release(actor_control_block* x);
// -- constructors, destructors, and assignment operators --------------------
actor_control_block(actor_id x, node_id& y, actor_system* sys,
data_deleter delete_data, block_deleter delete_block,
mpi_type* mpi = nullptr);
actor_control_block(const actor_control_block&) = delete;
actor_control_block& operator=(const actor_control_block&) = delete;
std::atomic<size_t> strong_refs;
std::atomic<size_t> weak_refs;
const actor_id aid;
const node_id nid;
actor_system* const home_system;
const data_destructor data_dtor;
const block_destructor block_dtor;
~actor_control_block();
static_assert(sizeof(std::atomic<size_t>) == sizeof(void*),
"std::atomic not lockfree on this platform");
"std::atomic<size_t> not lockfree on this platform");
static_assert(sizeof(intrusive_ptr<node_id::data>) == sizeof(void*),
"intrusive_ptr<T> and T* have different size");
static_assert(sizeof(node_id) == sizeof(void*),
"sizeof(node_id) != sizeof(size_t)");
"sizeof(node_id) != sizeof(void*)");
static_assert(sizeof(data_destructor) == sizeof(void*),
static_assert(sizeof(data_deleter) == sizeof(void*),
"functiion pointer and regular pointers have different size");
/// Returns a pointer to the actual actor instance.
inline abstract_actor* get() {
abstract_actor* get() {
// this pointer arithmetic is compile-time checked in actor_storage's ctor
return reinterpret_cast<abstract_actor*>(
reinterpret_cast<intptr_t>(this) + CAF_CACHE_LINE_SIZE);
return reinterpret_cast<abstract_actor*>(reinterpret_cast<intptr_t>(this)
+ CAF_CACHE_LINE_SIZE);
}
/// Returns a pointer to the control block that stores
......@@ -124,40 +138,43 @@ public:
actor_addr address();
inline actor_id id() const noexcept {
return aid;
actor_id id() const noexcept {
return aid_;
}
inline const node_id& node() const noexcept {
return nid;
const node_id& node() const noexcept {
return nid_;
}
void enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host);
void enqueue(mailbox_element_ptr what, execution_unit* host);
actor_system& system() const noexcept {
return *system_;
}
/// @endcond
};
bool has_mpi() const noexcept {
return mpi_ != nullptr;
}
/// @relates actor_control_block
bool intrusive_ptr_upgrade_weak(actor_control_block* x);
const mpi_type& mpi() const noexcept {
return *mpi_;
}
/// @relates actor_control_block
inline void intrusive_ptr_add_weak_ref(actor_control_block* x) {
x->weak_refs.fetch_add(1, std::memory_order_relaxed);
}
void enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host);
/// @relates actor_control_block
void intrusive_ptr_release_weak(actor_control_block* x);
void enqueue(mailbox_element_ptr what, execution_unit* host);
/// @relates actor_control_block
inline void intrusive_ptr_add_ref(actor_control_block* x) {
x->strong_refs.fetch_add(1, std::memory_order_relaxed);
}
/// @endcond
/// @relates actor_control_block
void intrusive_ptr_release(actor_control_block* x);
private:
std::atomic<size_t> strong_refs_;
std::atomic<size_t> weak_refs_;
actor_id aid_;
node_id nid_;
actor_system* system_;
data_deleter delete_data_;
block_deleter delete_block_;
mpi_type* mpi_;
};
/// @relates abstract_actor
/// @relates actor_control_block
......@@ -183,11 +200,11 @@ inline bool operator!=(const abstract_actor* x, const strong_actor_ptr& y) {
/// @relates actor_control_block
using weak_actor_ptr = weak_intrusive_ptr<actor_control_block>;
error load_actor(strong_actor_ptr& storage, execution_unit*,
actor_id aid, const node_id& nid);
error load_actor(strong_actor_ptr& storage, execution_unit*, actor_id aid,
const node_id& nid);
error save_actor(strong_actor_ptr& storage, execution_unit*,
actor_id aid, const node_id& nid);
error save_actor(strong_actor_ptr& storage, execution_unit*, actor_id aid,
const node_id& nid);
template <class Inspector>
auto context_of(Inspector* f) -> decltype(f->context()) {
......@@ -211,13 +228,12 @@ typename Inspector::result_type inspect(Inspector& f, strong_actor_ptr& x) {
actor_id aid = 0;
node_id nid;
if (x) {
aid = x->aid;
nid = x->nid;
aid = x->id();
nid = x->node();
}
auto load = [&] { return load_actor(x, context_of(&f), aid, nid); };
auto save = [&] { return save_actor(x, context_of(&f), aid, nid); };
return f(meta::type_name("actor"), aid,
meta::omittable_if_none(), nid,
return f(meta::type_name("actor"), aid, meta::omittable_if_none(), nid,
meta::load_callback(load), meta::save_callback(save));
}
......@@ -225,7 +241,10 @@ template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, weak_actor_ptr& x) {
// inspect as strong pointer, then write back to weak pointer on save
auto tmp = x.lock();
auto load = [&]() -> error { x.reset(tmp.get()); return none; };
auto load = [&]() -> error {
x.reset(tmp.get());
return none;
};
return f(tmp, meta::load_callback(load));
}
......@@ -236,7 +255,7 @@ namespace std {
template <>
struct hash<caf::strong_actor_ptr> {
inline size_t operator()(const caf::strong_actor_ptr& ptr) const {
size_t operator()(const caf::strong_actor_ptr& ptr) const {
return ptr ? static_cast<size_t>(ptr->id()) : 0;
}
};
......@@ -249,4 +268,3 @@ struct hash<caf::weak_actor_ptr> {
};
} // namespace std
......@@ -18,18 +18,18 @@
#pragma once
#include <new>
#include <atomic>
#include <cstddef>
#include <new>
#include <type_traits>
#include "caf/config.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/config.hpp"
#ifdef CAF_GCC
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Winvalid-offsetof"
# pragma GCC diagnostic push
# pragma GCC diagnostic ignored "-Winvalid-offsetof"
#endif
namespace caf {
......@@ -38,27 +38,26 @@ template <class T>
class actor_storage {
public:
template <class... Us>
actor_storage(actor_id x, node_id y, actor_system* sys, Us&&... zs)
: ctrl(x, y, sys, data_dtor, block_dtor) {
// construct data member
actor_storage(actor_id x, node_id y, actor_system* sys, Us&&... zs) {
new (&ctrl) actor_control_block(x, y, sys, data_dtor, block_dtor);
new (&data) T(std::forward<Us>(zs)...);
}
~actor_storage() {
// 1) make sure control block fits into a single cache line
static_assert(sizeof(actor_control_block) < CAF_CACHE_LINE_SIZE,
static_assert(sizeof(actor_control_block) <= CAF_CACHE_LINE_SIZE,
"actor_control_block exceeds a single cache line");
// Clang in combination with libc++ on Linux complains about offsetof:
// error: 'actor_storage' does not refer to a value
// Until we have found a reliable solution, we disable this safety check.
#if !(defined(CAF_CLANG) && defined(CAF_LINUX))
// Clang in combination with libc++ on Linux complains about offsetof:
// error: 'actor_storage' does not refer to a value
// Until we have found a reliable solution, we disable this safety check.
#if !(defined(CAF_CLANG) && defined(CAF_LINUX))
// 2) make sure reinterpret cast of the control block to the storage works
static_assert(offsetof(actor_storage, ctrl) == 0,
"control block is not at the start of the storage");
// 3) make sure we can obtain a data pointer by jumping one cache line
static_assert(offsetof(actor_storage, data) == CAF_CACHE_LINE_SIZE,
"data is not at cache line size boundary");
#else
#else
// 4) make sure static_cast and reinterpret_cast
// between T* and abstract_actor* are identical
constexpr abstract_actor* dummy = nullptr;
......@@ -66,18 +65,23 @@ public:
static_assert(derived_dummy == nullptr,
"actor subtype has illegal memory alignment "
"(probably due to virtual inheritance)");
#endif
#endif
ctrl.~actor_control_block();
}
actor_storage(const actor_storage&) = delete;
actor_storage& operator=(const actor_storage&) = delete;
static_assert(sizeof(actor_control_block) < CAF_CACHE_LINE_SIZE,
"actor_control_block exceeds 64 bytes");
// This union makes sure `ctrl` is padded to cache line size.
union {
actor_control_block ctrl;
char pad[CAF_CACHE_LINE_SIZE - sizeof(actor_control_block)];
union { T data; };
char pad[CAF_CACHE_LINE_SIZE];
};
// This union allows us to destroy `data` before we destroy `ctrl`.
union {
T data;
};
private:
static void data_dtor(abstract_actor* ptr) {
......@@ -139,6 +143,6 @@ void intrusive_ptr_release(actor_storage<T>* x) {
} // namespace caf
#ifdef CAF_GCC
#pragma GCC diagnostic pop
# pragma GCC diagnostic pop
#endif
......@@ -50,31 +50,34 @@ public:
~scoped_actor();
inline explicit operator bool() const {
explicit operator bool() const noexcept {
return static_cast<bool>(self_);
}
inline actor_system& home_system() const {
return *self_->home_system;
actor_system& system() const noexcept {
return self_->system();
}
inline blocking_actor* operator->() const {
actor_system& home_system() const noexcept {
return self_->system();
}
blocking_actor* operator->() const noexcept {
return ptr();
}
inline blocking_actor& operator*() const {
blocking_actor& operator*() const noexcept {
return *ptr();
}
inline actor_addr address() const {
actor_addr address() const noexcept {
return ptr()->address();
}
blocking_actor* ptr() const;
blocking_actor* ptr() const noexcept;
private:
inline actor_control_block* get() const {
actor_control_block* get() const noexcept {
return self_.get();
}
......
......@@ -166,18 +166,18 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
return *this;
}
inline typed_actor& operator=(std::nullptr_t) {
typed_actor& operator=(std::nullptr_t) {
ptr_.reset();
return *this;
}
/// Queries whether this actor handle is valid.
inline explicit operator bool() const {
explicit operator bool() const {
return static_cast<bool>(ptr_);
}
/// Queries whether this actor handle is invalid.
inline bool operator!() const {
bool operator!() const {
return !ptr_;
}
......@@ -197,8 +197,13 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
}
/// Returns the hosting actor system.
inline actor_system& home_system() const noexcept {
return *ptr_->home_system;
actor_system& system() const noexcept {
return ptr_->system();
}
/// Returns the hosting actor system.
actor_system& home_system() const noexcept {
return ptr_->system();
}
/// Exchange content of `*this` and `other`.
......
......@@ -87,7 +87,7 @@ node_id abstract_actor::node() const noexcept {
}
actor_system& abstract_actor::home_system() const noexcept {
return *(actor_control_block::from(this)->home_system);
return actor_control_block::from(this)->system();
}
mailbox_element* abstract_actor::peek_at_next_mailbox_element() {
......
......@@ -18,21 +18,35 @@
#include "caf/actor_control_block.hpp"
#include "caf/to_string.hpp"
#include "caf/message.hpp"
#include "caf/actor_system.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/message.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/to_string.hpp"
namespace caf {
actor_control_block::actor_control_block(actor_id x, node_id& y,
actor_system* sys,
data_deleter delete_data,
block_deleter delete_block,
mpi_type* mpi)
: strong_refs_(1),
weak_refs_(1),
aid_(x),
nid_(std::move(y)),
system_(sys),
delete_data_(delete_data),
delete_block_(delete_block),
mpi_(mpi) {
// nop
}
actor_control_block::~actor_control_block() {
// nop
}
actor_addr actor_control_block::address() {
return {this, true};
}
......@@ -48,9 +62,9 @@ void actor_control_block::enqueue(mailbox_element_ptr what,
}
bool intrusive_ptr_upgrade_weak(actor_control_block* x) {
auto count = x->strong_refs.load();
auto count = x->strong_refs_.load();
while (count != 0)
if (x->strong_refs.compare_exchange_weak(count, count + 1,
if (x->strong_refs_.compare_exchange_weak(count, count + 1,
std::memory_order_relaxed))
return true;
return false;
......@@ -58,16 +72,18 @@ bool intrusive_ptr_upgrade_weak(actor_control_block* x) {
void intrusive_ptr_release_weak(actor_control_block* x) {
// destroy object if last weak pointer expires
if (x->weak_refs == 1
|| x->weak_refs.fetch_sub(1, std::memory_order_acq_rel) == 1)
x->block_dtor(x);
if (x->weak_refs_ == 1
|| x->weak_refs_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
auto deleter = x->delete_block_;
deleter(x);
}
}
void intrusive_ptr_release(actor_control_block* x) {
// release implicit weak pointer if the last strong ref expires
// and destroy the data block
if (x->strong_refs.fetch_sub(1, std::memory_order_acq_rel) == 1) {
x->data_dtor(x->get());
if (x->strong_refs_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
x->delete_data_(x->get());
intrusive_ptr_release_weak(x);
}
}
......@@ -114,9 +130,9 @@ namespace {
void append_to_string_impl(std::string& x, const actor_control_block* y) {
if (y != nullptr) {
x += std::to_string(y->aid);
x += std::to_string(y->id());
x += '@';
append_to_string(x, y->nid);
append_to_string(x, y->node());
} else {
x += "0@invalid-node";
}
......
......@@ -97,11 +97,11 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
std::thread([](strong_actor_ptr ptr) {
// actor lives in its own thread
detail::set_thread_name("caf.actor");
ptr->home_system->thread_started();
ptr->system().thread_started();
auto this_ptr = ptr->get();
CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != nullptr);
auto self = static_cast<blocking_actor*>(this_ptr);
CAF_SET_LOGGER_SYS(ptr->home_system);
CAF_SET_LOGGER_SYS(&ptr->system());
CAF_PUSH_AID_FROM_PTR(self);
self->initialize();
error rsn;
......@@ -125,8 +125,8 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
self->on_exit();
# endif
self->cleanup(std::move(rsn), self->context());
ptr->home_system->thread_terminates();
ptr->home_system->dec_detached_threads();
ptr->system().thread_terminates();
ptr->system().dec_detached_threads();
}, strong_actor_ptr{ctrl()}).detach();
}
......
......@@ -72,7 +72,7 @@ scoped_actor::~scoped_actor() {
CAF_SET_AID(prev_);
}
blocking_actor* scoped_actor::ptr() const {
blocking_actor* scoped_actor::ptr() const noexcept {
return static_cast<blocking_actor*>(actor_cast<abstract_actor*>(self_));
}
......
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