Commit eaa811ef authored by Dominik Charousset's avatar Dominik Charousset

Implement new mailbox factory

parent 4cc28e04
...@@ -120,6 +120,8 @@ caf_add_component( ...@@ -120,6 +120,8 @@ caf_add_component(
caf/detail/json.cpp caf/detail/json.cpp
caf/detail/latch.cpp caf/detail/latch.cpp
caf/detail/local_group_module.cpp caf/detail/local_group_module.cpp
caf/detail/mailbox_factory.cpp
caf/detail/mailbox_factory.test.cpp
caf/detail/message_builder_element.cpp caf/detail/message_builder_element.cpp
caf/detail/message_data.cpp caf/detail/message_data.cpp
caf/detail/meta_object.cpp caf/detail/meta_object.cpp
......
...@@ -60,10 +60,20 @@ public: ...@@ -60,10 +60,20 @@ public:
/// @note Only the owning actor is allowed to call this function. /// @note Only the owning actor is allowed to call this function.
virtual size_t size() = 0; virtual size_t size() = 0;
/// Increases the reference count by one.
virtual void ref_mailbox() noexcept = 0;
/// Decreases the reference count by one and deletes this instance if the
/// reference count drops to zero.
virtual void deref_mailbox() noexcept = 0;
/// Checks whether the mailbox is empty. /// Checks whether the mailbox is empty.
bool empty() { bool empty() {
return size() == 0; return size() == 0;
} }
/// @private
virtual mailbox_element* peek(message_id id) = 0;
}; };
} // namespace caf } // namespace caf
...@@ -34,6 +34,7 @@ public: ...@@ -34,6 +34,7 @@ public:
int flags; int flags;
input_range<const group>* groups; input_range<const group>* groups;
detail::unique_function<behavior(local_actor*)> init_fun; detail::unique_function<behavior(local_actor*)> init_fun;
detail::mailbox_factory* mbox_factory = nullptr;
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
......
...@@ -107,4 +107,13 @@ bool default_mailbox::fetch_more() { ...@@ -107,4 +107,13 @@ bool default_mailbox::fetch_more() {
return true; return true;
} }
void default_mailbox::ref_mailbox() noexcept {
++ref_count_;
}
void default_mailbox::deref_mailbox() noexcept {
if (--ref_count_ == 0)
delete this;
}
} // namespace caf::detail } // namespace caf::detail
...@@ -9,6 +9,9 @@ ...@@ -9,6 +9,9 @@
#include "caf/intrusive/lifo_inbox.hpp" #include "caf/intrusive/lifo_inbox.hpp"
#include "caf/intrusive/linked_list.hpp" #include "caf/intrusive/linked_list.hpp"
#include <atomic>
#include <cstddef>
namespace caf::detail { namespace caf::detail {
/// Our default mailbox implementation. Uses a LIFO inbox for storing incoming /// Our default mailbox implementation. Uses a LIFO inbox for storing incoming
...@@ -30,13 +33,15 @@ public: ...@@ -30,13 +33,15 @@ public:
} }
}; };
default_mailbox() = default; default_mailbox() noexcept : ref_count_(1) {
// nop
}
default_mailbox(const default_mailbox&) = delete; default_mailbox(const default_mailbox&) = delete;
default_mailbox& operator=(const default_mailbox&) = delete; default_mailbox& operator=(const default_mailbox&) = delete;
mailbox_element* peek(message_id id); mailbox_element* peek(message_id id) override;
intrusive::inbox_result push_back(mailbox_element_ptr ptr) override; intrusive::inbox_result push_back(mailbox_element_ptr ptr) override;
...@@ -56,6 +61,14 @@ public: ...@@ -56,6 +61,14 @@ public:
size_t size() override; size_t size() override;
void ref_mailbox() noexcept override;
void deref_mailbox() noexcept override;
size_t ref_count() const noexcept {
return ref_count_.load();
}
private: private:
/// Returns the total number of elements stored in the queues. /// Returns the total number of elements stored in the queues.
size_t cached() const noexcept { size_t cached() const noexcept {
...@@ -65,14 +78,17 @@ private: ...@@ -65,14 +78,17 @@ private:
/// Tries to fetch more messages from the LIFO inbox. /// Tries to fetch more messages from the LIFO inbox.
bool fetch_more(); bool fetch_more();
/// Stores incoming messages in LIFO order.
alignas(CAF_CACHE_LINE_SIZE) intrusive::lifo_inbox<policy> inbox_;
/// Stores urgent messages in FIFO order. /// Stores urgent messages in FIFO order.
intrusive::linked_list<mailbox_element> urgent_queue_; intrusive::linked_list<mailbox_element> urgent_queue_;
/// Stores normal messages in FIFO order. /// Stores normal messages in FIFO order.
intrusive::linked_list<mailbox_element> normal_queue_; intrusive::linked_list<mailbox_element> normal_queue_;
/// Stores incoming messages in LIFO order.
alignas(CAF_CACHE_LINE_SIZE) intrusive::lifo_inbox<policy> inbox_;
/// The intrusive reference count.
alignas(CAF_CACHE_LINE_SIZE) std::atomic<size_t> ref_count_;
}; };
} // namespace caf::detail } // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/detail/mailbox_factory.hpp"
namespace caf::detail {
mailbox_factory::~mailbox_factory() {
// nop
}
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
namespace caf::detail {
/// The base class for all mailbox implementations.
class CAF_CORE_EXPORT mailbox_factory {
public:
virtual ~mailbox_factory();
/// Creates a new mailbox for `owner`.
virtual abstract_mailbox* make(scheduled_actor* owner) = 0;
/// Creates a new mailbox for `owner`.
virtual abstract_mailbox* make(blocking_actor* owner) = 0;
};
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/detail/mailbox_factory.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include "caf/blocking_actor.hpp"
#include "caf/detail/default_mailbox.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/local_actor.hpp"
#include "caf/scheduler/test_coordinator.hpp"
using namespace caf;
class dummy_mailbox_factory : public detail::mailbox_factory {
public:
~dummy_mailbox_factory() override {
for (auto& kvp : mailboxes)
kvp.second->deref_mailbox();
}
abstract_mailbox* make(local_actor* owner) {
auto ptr = new detail::default_mailbox;
ptr->ref_mailbox();
mailboxes.emplace(owner->id(), ptr);
return ptr;
}
abstract_mailbox* make(scheduled_actor* owner) override {
return make(static_cast<local_actor*>(owner));
}
abstract_mailbox* make(blocking_actor* owner) override {
return make(static_cast<local_actor*>(owner));
}
std::map<actor_id, detail::default_mailbox*> mailboxes;
};
TEST("a mailbox factory creates mailboxes for actors") {
dummy_mailbox_factory factory;
actor_system_config cfg;
cfg.set("caf.scheduler.policy", "testing");
actor_system sys{cfg};
auto& sched = static_cast<scheduler::test_coordinator&>(sys.scheduler());
auto spawn_dummy = [&sys, &factory](auto fn) {
using impl_t = event_based_actor;
actor_config cfg{sys.dummy_execution_unit()};
cfg.init_fun.emplace([fn](local_actor* self) {
fn(static_cast<impl_t*>(self));
return behavior{};
});
cfg.mbox_factory = &factory;
auto res = make_actor<impl_t>(sys.next_actor_id(), sys.node(), &sys, cfg);
auto ptr = static_cast<impl_t*>(actor_cast<abstract_actor*>(res));
ptr->launch(cfg.host, false, false);
return res;
};
SECTION("spawning dummies creates mailboxes") {
auto initialized = std::make_shared<size_t>(0u);
auto dummy_impl = [this, &factory, initialized](event_based_actor* self) {
++*initialized;
auto* mbox = factory.mailboxes[self->id()];
check_eq(std::addressof(self->mailbox()), mbox);
check_eq(mbox->ref_count(), 2u);
};
check_eq(factory.mailboxes.size(), 0u);
auto dummy1 = spawn_dummy(dummy_impl).id();
check_eq(factory.mailboxes.size(), 1u);
auto dummy2 = spawn_dummy(dummy_impl).id();
check_eq(factory.mailboxes.size(), 2u);
sched.run();
check_eq(*initialized, 2u);
check_eq(factory.mailboxes[dummy1]->ref_count(), 1u);
check_eq(factory.mailboxes[dummy2]->ref_count(), 1u);
}
}
CAF_TEST_MAIN()
...@@ -126,6 +126,18 @@ public: ...@@ -126,6 +126,18 @@ public:
*this = unique_function{ptr}; *this = unique_function{ptr};
} }
template <class Fn>
void emplace(Fn fn) {
destroy();
if constexpr (std::is_convertible<Fn, raw_pointer>::value) {
holds_wrapper_ = false;
fptr_ = fn;
} else {
holds_wrapper_ = true;
new (&wptr_) wrapper_pointer{make_wrapper(std::move(fn))};
}
}
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
bool is_nullptr() const noexcept { bool is_nullptr() const noexcept {
......
...@@ -74,6 +74,7 @@ class [[deprecated("use std::string_view instead")]] string_view; ...@@ -74,6 +74,7 @@ class [[deprecated("use std::string_view instead")]] string_view;
class [[nodiscard]] error; class [[nodiscard]] error;
class abstract_actor; class abstract_actor;
class abstract_group; class abstract_group;
class abstract_mailbox;
class action; class action;
class actor; class actor;
class actor_addr; class actor_addr;
...@@ -267,6 +268,8 @@ using int_gauge_family = metric_family_impl<int_gauge>; ...@@ -267,6 +268,8 @@ using int_gauge_family = metric_family_impl<int_gauge>;
namespace detail { namespace detail {
class mailbox_factory;
template <class> template <class>
struct gauge_oracle; struct gauge_oracle;
......
...@@ -52,6 +52,12 @@ TEST("a default default-constructed uut is empty") { ...@@ -52,6 +52,12 @@ TEST("a default default-constructed uut is empty") {
check_eq(uut.begin(), uut.end()); check_eq(uut.begin(), uut.end());
} }
TEST("uuts are convertible to strings") {
check_eq(deep_to_string(uut), "[]");
fill(uut, 1, 2, 3, 4);
check_eq(deep_to_string(uut), "[1, 2, 3, 4]");
}
TEST("push_back adds elements to the back of the uut") { TEST("push_back adds elements to the back of the uut") {
uut.emplace_back(1); uut.emplace_back(1);
uut.push_back(std::make_unique<inode>(2)); uut.push_back(std::make_unique<inode>(2));
...@@ -105,12 +111,6 @@ TEST("the size of the uut is the number of elements") { ...@@ -105,12 +111,6 @@ TEST("the size of the uut is the number of elements") {
check_eq(uut.size(), 5u); check_eq(uut.size(), 5u);
} }
TEST("uuts are convertible to strings") {
check_eq(deep_to_string(uut), "[]");
fill(uut, 1, 2, 3, 4);
check_eq(deep_to_string(uut), "[1, 2, 3, 4]");
}
TEST("calling clear removes all elements from a uut") { TEST("calling clear removes all elements from a uut") {
fill(uut, 1, 2, 3); fill(uut, 1, 2, 3);
check_eq(uut.size(), 3u); check_eq(uut.size(), 3u);
......
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp" #include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/mailbox_factory.hpp"
#include "caf/detail/meta_object.hpp" #include "caf/detail/meta_object.hpp"
#include "caf/detail/private_thread.hpp" #include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
...@@ -19,6 +20,8 @@ ...@@ -19,6 +20,8 @@
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/stream.hpp" #include "caf/stream.hpp"
#include <new>
using namespace std::string_literals; using namespace std::string_literals;
namespace caf { namespace caf {
...@@ -129,11 +132,18 @@ scheduled_actor::scheduled_actor(actor_config& cfg) ...@@ -129,11 +132,18 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
exception_handler_(default_exception_handler) exception_handler_(default_exception_handler)
#endif // CAF_ENABLE_EXCEPTIONS #endif // CAF_ENABLE_EXCEPTIONS
{ {
// nop if (cfg.mbox_factory == nullptr)
mailbox_ = new (&default_mailbox_) detail::default_mailbox();
else
mailbox_ = cfg.mbox_factory->make(this);
} }
scheduled_actor::~scheduled_actor() { scheduled_actor::~scheduled_actor() {
// nop unstash();
if (mailbox_ == &default_mailbox_)
default_mailbox_.~default_mailbox();
else
mailbox_->deref_mailbox();
} }
// -- overridden functions of abstract_actor ----------------------------------- // -- overridden functions of abstract_actor -----------------------------------
...@@ -181,9 +191,9 @@ bool scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) { ...@@ -181,9 +191,9 @@ bool scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
} }
mailbox_element* scheduled_actor::peek_at_next_mailbox_element() { mailbox_element* scheduled_actor::peek_at_next_mailbox_element() {
return mailbox_.peek(awaited_responses_.empty() return mailbox().peek(awaited_responses_.empty()
? make_message_id() ? make_message_id()
: awaited_responses_.begin()->first); : awaited_responses_.begin()->first);
} }
// -- overridden functions of local_actor -------------------------------------- // -- overridden functions of local_actor --------------------------------------
......
...@@ -238,7 +238,7 @@ public: ...@@ -238,7 +238,7 @@ public:
/// Returns the queue for storing incoming messages. /// Returns the queue for storing incoming messages.
abstract_mailbox& mailbox() noexcept { abstract_mailbox& mailbox() noexcept {
return mailbox_; return *mailbox_;
} }
// -- event handlers --------------------------------------------------------- // -- event handlers ---------------------------------------------------------
...@@ -684,7 +684,7 @@ protected: ...@@ -684,7 +684,7 @@ protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Stores incoming messages. /// Stores incoming messages.
detail::default_mailbox mailbox_; abstract_mailbox* mailbox_;
/// Stores user-defined callbacks for message handling. /// Stores user-defined callbacks for message handling.
detail::behavior_stack bhvr_stack_; detail::behavior_stack bhvr_stack_;
...@@ -820,6 +820,12 @@ private: ...@@ -820,6 +820,12 @@ private:
/// Stashes skipped messages until the actor processes the next message. /// Stashes skipped messages until the actor processes the next message.
intrusive::stack<mailbox_element> stash_; intrusive::stack<mailbox_element> stash_;
union {
/// The default mailbox instance that we use if the user does not configure
/// a mailbox via the ::actor_config.
detail::default_mailbox default_mailbox_;
};
}; };
} // namespace caf } // namespace caf
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