Commit 3e0baf18 authored by Dominik Charousset's avatar Dominik Charousset

Implement new actor type for socket managers

Similarly to `scoped_actor`, the new class `actor_shell` exposes a
mailbox to another object. However, the new shell type grants
non-blocking access.
parent 2c905303
...@@ -51,6 +51,7 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} ...@@ -51,6 +51,7 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/ip.cpp src/ip.cpp
src/message_queue.cpp src/message_queue.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/actor_shell.cpp
#src/net/endpoint_manager_queue.cpp #src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/middleman_backend.cpp src/net/middleman_backend.cpp
...@@ -132,6 +133,7 @@ caf_incubator_add_test_suites(caf-net-test ...@@ -132,6 +133,7 @@ caf_incubator_add_test_suites(caf-net-test
header header
ip ip
multiplexer multiplexer
net.actor_shell
#net.backend.tcp #net.backend.tcp
#net.basp.message_queue #net.basp.message_queue
#net.basp.ping_pong #net.basp.ping_pong
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include "caf/actor_traits.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/local_actor.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
#include "caf/policy/normal_messages.hpp"
namespace caf::net {
///
class actor_shell
: public extend<local_actor, actor_shell>::with<mixin::sender>,
public dynamically_typed_actor_base,
public non_blocking_actor_base {
public:
// -- friends ----------------------------------------------------------------
friend class actor_shell_ptr;
// -- member types -----------------------------------------------------------
using super = extend<local_actor, actor_shell>::with<mixin::sender>;
using signatures = none_t;
using behavior_type = behavior;
struct mailbox_policy {
using queue_type = intrusive::drr_queue<policy::normal_messages>;
using deficit_type = policy::normal_messages::deficit_type;
using mapped_type = policy::normal_messages::mapped_type;
using unique_pointer = policy::normal_messages::unique_pointer;
};
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
// -- constructors, destructors, and assignment operators --------------------
explicit actor_shell(actor_config& cfg, socket_manager* owner);
~actor_shell() override;
// -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox.
void quit(error reason);
// -- mailbox access ---------------------------------------------------------
auto& mailbox() noexcept {
return mailbox_;
}
/// Dequeues and returns the next message from the mailbox or returns
/// `nullptr` if the mailbox is empty.
mailbox_element_ptr next_message();
/// Dequeues and processes the next message from the mailbox.
/// @param bhvr Available message handlers.
/// @param fallback Callback for processing message that failed to match
/// `bhvr`.
/// @returns `true` if a message was dequeued and process, `false` if the
/// mailbox was empty.
bool consume_message(behavior& bhvr,
callback<result<message>(message&)>& fallback);
/// Tries to put the mailbox into the `blocked` state, causing the next
/// enqueue to register the owning socket manager for write events.
bool try_block_mailbox();
// -- overridden functions of abstract_actor ---------------------------------
using abstract_actor::enqueue;
void enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
mailbox_element* peek_at_next_mailbox_element() override;
// -- overridden functions of local_actor ------------------------------------
const char* name() const override;
void launch(execution_unit* eu, bool lazy, bool hide) override;
bool cleanup(error&& fail_state, execution_unit* host) override;
private:
// Stores incoming actor messages.
mailbox_type mailbox_;
// Guards access to owner_.
std::mutex owner_mtx_;
// Points to the owning manager (nullptr after quit was called).
socket_manager* owner_;
};
/// An "owning" pointer to an actor shell in the sense that it calls `quit()` on
/// the shell when going out of scope.
class actor_shell_ptr {
public:
friend class socket_manager;
constexpr actor_shell_ptr() noexcept {
// nop
}
constexpr actor_shell_ptr(std::nullptr_t) noexcept {
// nop
}
actor_shell_ptr(actor_shell_ptr&& other) noexcept = default;
actor_shell_ptr& operator=(actor_shell_ptr&& other) noexcept = default;
actor_shell_ptr(const actor_shell_ptr& other) = delete;
actor_shell_ptr& operator=(const actor_shell_ptr& other) = delete;
~actor_shell_ptr();
/// Returns an actor handle to the managed actor shell.
actor as_actor() const noexcept;
void detach(error reason);
actor_shell* get() const noexcept;
actor_shell* operator->() const noexcept {
return get();
}
actor_shell& operator*() const noexcept {
return *get();
}
bool operator!() const noexcept {
return !ptr_;
}
explicit operator bool() const noexcept {
return static_cast<bool>(ptr_);
}
private:
/// @pre `ptr != nullptr`
explicit actor_shell_ptr(strong_actor_ptr ptr) noexcept;
strong_actor_ptr ptr_;
};
} // namespace caf::net
...@@ -42,6 +42,8 @@ enum class ec : uint8_t; ...@@ -42,6 +42,8 @@ enum class ec : uint8_t;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
class actor_shell;
class actor_shell_ptr;
class endpoint_manager; class endpoint_manager;
class middleman; class middleman;
class middleman_backend; class middleman_backend;
......
...@@ -50,6 +50,8 @@ public: ...@@ -50,6 +50,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
~middleman() override; ~middleman() override;
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
...@@ -142,10 +144,6 @@ public: ...@@ -142,10 +144,6 @@ public:
expected<uint16_t> port(string_view scheme) const; expected<uint16_t> port(string_view scheme) const;
private: private:
// -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
static void create_backends(middleman&, detail::type_list<>) { static void create_backends(middleman&, detail::type_list<>) {
......
...@@ -92,6 +92,10 @@ public: ...@@ -92,6 +92,10 @@ public:
return abort_reason_; return abort_reason_;
} }
// -- factories --------------------------------------------------------------
actor_shell_ptr make_actor_shell();
// -- event loop management -------------------------------------------------- // -- event loop management --------------------------------------------------
void register_reading(); void register_reading();
...@@ -163,7 +167,35 @@ public: ...@@ -163,7 +167,35 @@ public:
return protocol_; return protocol_;
} }
auto& top_layer() noexcept {
return climb(protocol_);
}
const auto& top_layer() const noexcept {
return climb(protocol_);
}
private: private:
template <class FinalLayer>
static FinalLayer& climb(FinalLayer& layer) {
return layer;
}
template <class FinalLayer>
static const FinalLayer& climb(const FinalLayer& layer) {
return layer;
}
template <template <class> class Layer, class Next>
static auto& climb(Layer<Next>& layer) {
return climb(layer.upper_layer());
}
template <template <class> class Layer, class Next>
static const auto& climb(const Layer<Next>& layer) {
return climb(layer.upper_layer());
}
Protocol protocol_; Protocol protocol_;
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/actor_shell.hpp"
#include "caf/callback.hpp"
#include "caf/config.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/logger.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf::net {
actor_shell::actor_shell(actor_config& cfg, socket_manager* owner)
: super(cfg), mailbox_(policy::normal_messages{}), owner_(owner) {
mailbox_.try_block();
}
actor_shell::~actor_shell() {
// nop
}
void actor_shell::quit(error reason) {
cleanup(std::move(reason), nullptr);
}
mailbox_element_ptr actor_shell::next_message() {
mailbox_.fetch_more();
auto& q = mailbox_.queue();
if (q.total_task_size() > 0) {
q.inc_deficit(1);
return q.next();
}
return nullptr;
}
bool actor_shell::consume_message(
behavior& bhvr, callback<result<message>(message&)>& fallback) {
CAF_LOG_TRACE("");
if (auto msg = next_message()) {
current_element_ = msg.get();
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, *msg);
detail::default_invoke_result_visitor<actor_shell> visitor{this};
if (auto result = bhvr(msg->payload)) {
visitor(*result);
} else {
auto fallback_result = fallback(msg->payload);
visit(visitor, fallback_result);
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
return true;
}
return false;
}
bool actor_shell::try_block_mailbox() {
return mailbox_.try_block();
}
void actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(!getf(is_blocking_flag));
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid;
auto sender = ptr->sender;
auto collects_metrics = getf(abstract_actor::collects_metrics_flag);
if (collects_metrics) {
ptr->set_enqueue_time();
metrics_.mailbox_size->inc();
}
switch (mailbox().push_back(std::move(ptr))) {
case intrusive::inbox_result::unblocked_reader: {
CAF_LOG_ACCEPT_EVENT(true);
std::unique_lock<std::mutex> guard{owner_mtx_};
// The owner can only be null if this enqueue succeeds, then we close the
// mailbox and reset owner_ in cleanup() before acquiring the mutex here.
// Hence, the mailbox element has already been disposed and we can simply
// skip any further processing.
if (owner_)
owner_->mpx().register_writing(owner_);
break;
}
case intrusive::inbox_result::queue_closed: {
CAF_LOG_REJECT_EVENT();
home_system().base_metrics().rejected_messages->inc();
if (collects_metrics)
metrics_.mailbox_size->dec();
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
}
break;
}
case intrusive::inbox_result::success:
// Enqueued to a running actors' mailbox: nothing to do.
CAF_LOG_ACCEPT_EVENT(false);
break;
}
}
mailbox_element* actor_shell::peek_at_next_mailbox_element() {
return mailbox().closed() || mailbox().blocked() ? nullptr : mailbox().peek();
}
const char* actor_shell::name() const {
return "caf.net.actor-shell";
}
void actor_shell::launch(execution_unit*, bool, bool hide) {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(CAF_ARG(hide));
CAF_ASSERT(!getf(is_blocking_flag));
if (!hide)
register_at_system();
}
bool actor_shell::cleanup(error&& fail_state, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(fail_state));
// Clear mailbox.
if (!mailbox_.closed()) {
mailbox_.close();
detail::sync_request_bouncer bounce{fail_state};
auto dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
while (dropped > 0) {
if (getf(abstract_actor::collects_metrics_flag)) {
auto val = static_cast<int64_t>(dropped);
metrics_.mailbox_size->dec(val);
}
dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
}
}
// Detach from owner.
{
std::unique_lock<std::mutex> guard{owner_mtx_};
owner_ = nullptr;
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
}
actor_shell_ptr::actor_shell_ptr(strong_actor_ptr ptr) noexcept
: ptr_(std::move(ptr)) {
// nop
}
actor actor_shell_ptr::as_actor() const noexcept {
return actor_cast<actor>(ptr_);
}
void actor_shell_ptr::detach(error reason) {
if (auto ptr = get()) {
ptr->quit(std::move(reason));
ptr_.release();
}
}
actor_shell_ptr::~actor_shell_ptr() {
if (auto ptr = get())
ptr->quit(exit_reason::normal);
}
actor_shell* actor_shell_ptr::get() const noexcept {
if (ptr_) {
auto ptr = actor_cast<abstract_actor*>(ptr_);
return static_cast<actor_shell*>(ptr);
}
return nullptr;
}
} // namespace caf::net
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
namespace caf::net { namespace caf::net {
...@@ -63,4 +64,10 @@ void socket_manager::register_writing() { ...@@ -63,4 +64,10 @@ void socket_manager::register_writing() {
parent_->register_writing(this); parent_->register_writing(this);
} }
actor_shell_ptr socket_manager::make_actor_shell() {
CAF_ASSERT(parent_ != nullptr);
auto hdl = parent_->system().spawn<actor_shell>(this);
return actor_shell_ptr{actor_cast<strong_actor_ptr>(std::move(hdl))};
}
} // namespace caf::net } // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#define CAF_SUITE net.actor_shell
#include "caf/net/actor_shell.hpp"
#include "net-test.hpp"
#include "caf/callback.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_transport.hpp"
using namespace caf;
namespace {
using byte_span = span<const byte>;
using svec = std::vector<std::string>;
struct app_t {
using input_tag = tag::stream_oriented;
template <class LowerLayer>
error init(net::socket_manager* mgr, LowerLayer&, const settings&) {
self = mgr->make_actor_shell();
bhvr = behavior{
[this](std::string& line) { lines.emplace_back(std::move(line)); },
};
fallback = make_type_erased_callback([](message& msg) -> result<message> {
CAF_FAIL("unexpected message: " << msg);
return make_error(sec::unexpected_message);
});
return none;
}
template <class LowerLayer>
bool prepare_send(LowerLayer&) {
while (self->consume_message(bhvr, *fallback))
; // Repeat.
return true;
}
template <class LowerLayer>
bool done_sending(LowerLayer&) {
return self->try_block_mailbox();
}
template <class LowerLayer>
void abort(LowerLayer&, const error& reason) {
CAF_FAIL("app::abort called: " << reason);
}
template <class LowerLayer>
ptrdiff_t consume(LowerLayer&, byte_span, byte_span) {
CAF_FAIL("received unexpected data");
return -1;
}
std::vector<std::string> lines;
net::actor_shell_ptr self;
behavior bhvr;
unique_callback_ptr<result<message>(message&)> fallback;
};
struct fixture : host_fixture, test_coordinator_fixture<> {
fixture() : mm(sys), mpx(&mm) {
mpx.set_thread_id();
if (auto err = mpx.init())
CAF_FAIL("mpx.init() failed: " << err);
auto sockets = unbox(net::make_stream_socket_pair());
self_socket_guard.reset(sockets.first);
testee_socket_guard.reset(sockets.second);
if (auto err = nonblocking(testee_socket_guard.socket(), true))
CAF_FAIL("nonblocking returned an error: " << err);
}
template <class Predicate>
void run_while(Predicate predicate) {
if (!predicate())
return;
for (size_t i = 0; i < 1000; ++i) {
mpx.poll_once(false);
if (!predicate())
return;
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
CAF_FAIL("reached max repeat rate without meeting the predicate");
}
net::middleman mm;
net::multiplexer mpx;
net::socket_guard<net::stream_socket> self_socket_guard;
net::socket_guard<net::stream_socket> testee_socket_guard;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(actor_shell_tests, fixture)
CAF_TEST(actor shells expose their mailbox to their owners) {
auto sck = testee_socket_guard.release();
auto mgr = net::make_socket_manager<app_t, net::stream_transport>(sck, &mpx);
if (auto err = mgr->init(content(cfg)))
CAF_FAIL("mgr->init() failed: " << err);
auto& app = mgr->top_layer();
auto hdl = app.self.as_actor();
anon_send(hdl, "line 1");
anon_send(hdl, "line 2");
anon_send(hdl, "line 3");
run_while([&] { return app.lines.size() != 3; });
CAF_CHECK_EQUAL(app.lines, svec({"line 1", "line 2", "line 3"}));
}
CAF_TEST_FIXTURE_SCOPE_END()
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