Unverified Commit fdc00758 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #88

Implement statically typed actor_shell version
parents ac35f5a0 358c4436
......@@ -53,6 +53,7 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/ip.cpp
src/message_queue.cpp
src/multiplexer.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
......@@ -142,6 +143,7 @@ caf_incubator_add_test_suites(caf-net-test
#net.basp.ping_pong
#net.basp.worker
net.length_prefix_framing
net.typed_actor_shell
net.web_socket_server
network_socket
pipe_socket
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::detail {
template <class T>
struct actor_shell_ptr_type_oracle;
template <>
struct actor_shell_ptr_type_oracle<actor> {
using type = net::actor_shell_ptr;
};
template <class... Sigs>
struct actor_shell_ptr_type_oracle<typed_actor<Sigs...>> {
using type = net::typed_actor_shell_ptr<Sigs...>;
};
template <class T>
using infer_actor_shell_ptr_type =
typename actor_shell_ptr_type_oracle<T>::type;
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/callback.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.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/requester.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 CAF_NET_EXPORT abstract_actor_shell : public local_actor,
public non_blocking_actor_base {
public:
// -- member types -----------------------------------------------------------
using super = local_actor;
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>;
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators --------------------
abstract_actor_shell(actor_config& cfg, socket_manager* owner);
~abstract_actor_shell() override;
// -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox.
void quit(error reason);
/// Overrides the default handler for unexpected messages.
template <class F>
void set_fallback(F f) {
fallback_ = make_type_erased_callback(std::move(f));
}
// -- 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();
/// 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();
// -- message processing -----------------------------------------------------
/// Dequeues and processes the next message from the mailbox.
/// @returns `true` if a message was dequeued and processed, `false` if the
/// mailbox was empty.
bool consume_message();
/// Adds a callback for a multiplexed response.
void add_multiplexed_response_handler(message_id response_id, behavior bhvr);
// -- 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 ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override;
bool cleanup(error&& fail_state, execution_unit* host) override;
protected:
// 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_;
// Handler for consuming messages from the mailbox.
behavior bhvr_;
// Handler for unexpected messages.
fallback_handler fallback_;
// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
};
} // namespace caf::net
......@@ -19,28 +19,23 @@
#pragma once
#include "caf/actor_traits.hpp"
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.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/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/net/abstract_actor_shell.hpp"
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
#include "caf/policy/normal_messages.hpp"
namespace caf::net {
///
/// Enables socket managers to communicate with actors using dynamically typed
/// messaging.
class CAF_NET_EXPORT actor_shell
: public extend<local_actor, actor_shell>::with<mixin::sender,
mixin::requester>,
public dynamically_typed_actor_base,
public non_blocking_actor_base {
: public extend<abstract_actor_shell, actor_shell>::with<mixin::sender,
mixin::requester>,
public dynamically_typed_actor_base {
public:
// -- friends ----------------------------------------------------------------
......@@ -48,27 +43,13 @@ public:
// -- member types -----------------------------------------------------------
using super
= extend<local_actor, actor_shell>::with<mixin::sender, mixin::requester>;
using super = extend<abstract_actor_shell,
actor_shell>::with<mixin::sender, mixin::requester>;
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>;
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators --------------------
actor_shell(actor_config& cfg, socket_manager* owner);
......@@ -77,87 +58,33 @@ public:
// -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox.
void quit(error reason);
/// Overrides the callbacks for incoming messages.
template <class... Fs>
void set_behavior(Fs... fs) {
bhvr_ = behavior{std::move(fs)...};
}
/// Overrides the default handler for unexpected messages.
template <class F>
void set_fallback(F f) {
fallback_ = make_type_erased_callback(std::move(f));
}
// -- 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();
/// 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();
// -- message processing -----------------------------------------------------
/// Dequeues and processes the next message from the mailbox.
/// @returns `true` if a message was dequeued and process, `false` if the
/// mailbox was empty.
bool consume_message();
/// Adds a callback for a multiplexed response.
void add_multiplexed_response_handler(message_id response_id, behavior bhvr);
// -- 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_;
// Handler for consuming messages from the mailbox.
behavior bhvr_;
// Handler for unexpected messages.
fallback_handler fallback_;
// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
};
/// An "owning" pointer to an actor shell in the sense that it calls `quit()` on
/// the shell when going out of scope.
class CAF_NET_EXPORT actor_shell_ptr {
public:
// -- friends ----------------------------------------------------------------
friend class socket_manager;
// -- member types -----------------------------------------------------------
using handle_type = actor;
using element_type = actor_shell;
// -- constructors, destructors, and assignment operators --------------------
constexpr actor_shell_ptr() noexcept {
// nop
}
......@@ -176,18 +103,20 @@ public:
~actor_shell_ptr();
// -- smart pointer interface ------------------------------------------------
/// Returns an actor handle to the managed actor shell.
actor as_actor() const noexcept;
handle_type as_actor() const noexcept;
void detach(error reason);
actor_shell* get() const noexcept;
element_type* get() const noexcept;
actor_shell* operator->() const noexcept {
element_type* operator->() const noexcept {
return get();
}
actor_shell& operator*() const noexcept {
element_type& operator*() const noexcept {
return *get();
}
......
......@@ -38,6 +38,14 @@ class transport_worker;
template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher;
template <class... Sigs>
class typed_actor_shell;
template <class... Sigs>
class typed_actor_shell_ptr;
// -- enumerations -------------------------------------------------------------
enum class ec : uint8_t;
// -- classes ------------------------------------------------------------------
......@@ -61,7 +69,7 @@ struct tcp_stream_socket;
struct datagram_socket;
struct udp_datagram_socket;
// -- smart pointers -----------------------------------------------------------
// -- smart pointer aliases ----------------------------------------------------
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
......
......@@ -18,7 +18,10 @@
#pragma once
#include "caf/actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/callback.hpp"
#include "caf/detail/infer_actor_shell_ptr_type.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
......@@ -27,6 +30,7 @@
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/typed_actor_shell.hpp"
#include "caf/ref_counted.hpp"
#include "caf/tag/io_event_oriented.hpp"
......@@ -58,6 +62,9 @@ public:
return handle_;
}
/// Returns a reference to the hosting @ref actor_system instance.
actor_system& system() noexcept;
/// Returns the owning @ref multiplexer instance.
multiplexer& mpx() noexcept {
return *parent_;
......@@ -110,19 +117,23 @@ public:
// -- factories --------------------------------------------------------------
template <class LowerLayerPtr, class FallbackHandler>
actor_shell_ptr make_actor_shell(LowerLayerPtr, FallbackHandler f) {
auto ptr = make_actor_shell_impl();
template <class Handle = actor, class LowerLayerPtr, class FallbackHandler>
auto make_actor_shell(LowerLayerPtr, FallbackHandler f) {
using ptr_type = detail::infer_actor_shell_ptr_type<Handle>;
using impl_type = typename ptr_type::element_type;
auto hdl = system().spawn<impl_type>(this);
auto ptr = ptr_type{actor_cast<strong_actor_ptr>(std::move(hdl))};
ptr->set_fallback(std::move(f));
return ptr;
}
template <class LowerLayerPtr>
actor_shell_ptr make_actor_shell(LowerLayerPtr down) {
return make_actor_shell(down, [down](message& msg) -> result<message> {
template <class Handle = actor, class LowerLayerPtr>
auto make_actor_shell(LowerLayerPtr down) {
auto f = [down](message& msg) -> result<message> {
down->abort_reason(make_error(sec::unexpected_message, std::move(msg)));
return make_error(sec::unexpected_message);
});
};
return make_actor_shell<Handle>(down, std::move(f));
}
// -- event loop management --------------------------------------------------
......@@ -155,9 +166,6 @@ protected:
multiplexer* parent_;
error abort_reason_;
private:
actor_shell_ptr make_actor_shell_impl();
};
template <class 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. *
******************************************************************************/
#pragma once
#include "caf/actor_traits.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/net/abstract_actor_shell.hpp"
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
#include "caf/typed_actor.hpp"
namespace caf::net {
/// Enables socket managers to communicate with actors using statically typed
/// messaging.
template <class... Sigs>
class typed_actor_shell
// clang-format off
: public extend<abstract_actor_shell, typed_actor_shell<Sigs...>>::template
with<mixin::sender, mixin::requester>,
public statically_typed_actor_base {
// clang-format on
public:
// -- friends ----------------------------------------------------------------
template <class...>
friend class typed_actor_shell_ptr;
// -- member types -----------------------------------------------------------
// clang-format off
using super =
typename extend<abstract_actor_shell, typed_actor_shell<Sigs...>>::template
with<mixin::sender, mixin::requester>;
// clang-format on
using signatures = detail::type_list<Sigs...>;
using behavior_type = typed_behavior<Sigs...>;
// -- constructors, destructors, and assignment operators --------------------
using super::super;
// -- state modifiers --------------------------------------------------------
/// Overrides the callbacks for incoming messages.
template <class... Fs>
void set_behavior(Fs... fs) {
auto new_bhvr = behavior_type{std::move(fs)...};
this->bhvr_ = std::move(new_bhvr.unbox());
}
// -- overridden functions of local_actor ------------------------------------
const char* name() const override {
return "caf.net.typed-actor-shell";
}
};
/// An "owning" pointer to an actor shell in the sense that it calls `quit()` on
/// the shell when going out of scope.
template <class... Sigs>
class typed_actor_shell_ptr {
public:
// -- friends ----------------------------------------------------------------
friend class socket_manager;
// -- member types -----------------------------------------------------------
using handle_type = typed_actor<Sigs...>;
using element_type = typed_actor_shell<Sigs...>;
// -- constructors, destructors, and assignment operators --------------------
constexpr typed_actor_shell_ptr() noexcept {
// nop
}
constexpr typed_actor_shell_ptr(std::nullptr_t) noexcept {
// nop
}
typed_actor_shell_ptr(typed_actor_shell_ptr&& other) noexcept = default;
typed_actor_shell_ptr&
operator=(typed_actor_shell_ptr&& other) noexcept = default;
typed_actor_shell_ptr(const typed_actor_shell_ptr& other) = delete;
typed_actor_shell_ptr& operator=(const typed_actor_shell_ptr& other) = delete;
~typed_actor_shell_ptr() {
if (auto ptr = get())
ptr->quit(exit_reason::normal);
}
// -- smart pointer interface ------------------------------------------------
/// Returns an actor handle to the managed actor shell.
handle_type as_actor() const noexcept {
return actor_cast<handle_type>(ptr_);
}
void detach(error reason) {
if (auto ptr = get()) {
ptr->quit(std::move(reason));
ptr_.release();
}
}
element_type* get() const noexcept {
if (ptr_) {
auto ptr = actor_cast<abstract_actor*>(ptr_);
return static_cast<element_type*>(ptr);
} else {
return nullptr;
}
}
element_type* operator->() const noexcept {
return get();
}
element_type& 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 typed_actor_shell_ptr(strong_actor_ptr ptr) noexcept
: ptr_(std::move(ptr)) {
// nop
}
strong_actor_ptr ptr_;
};
} // namespace caf::net
namespace caf::detail {
template <class T>
struct typed_actor_shell_ptr_oracle;
template <class... Sigs>
struct typed_actor_shell_ptr_oracle<typed_actor<Sigs...>> {
using type = net::typed_actor_shell_ptr<Sigs...>;
};
} // namespace caf::detail
namespace caf::net {
template <class Handle>
using typed_actor_shell_ptr_t =
typename detail::typed_actor_shell_ptr_oracle<Handle>::type;
} // 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. *
******************************************************************************/
#include "caf/net/abstract_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 {
// -- constructors, destructors, and assignment operators ----------------------
abstract_actor_shell::abstract_actor_shell(actor_config& cfg,
socket_manager* owner)
: super(cfg), mailbox_(policy::normal_messages{}), owner_(owner) {
mailbox_.try_block();
}
abstract_actor_shell::~abstract_actor_shell() {
// nop
}
// -- state modifiers ----------------------------------------------------------
void abstract_actor_shell::quit(error reason) {
cleanup(std::move(reason), nullptr);
}
// -- mailbox access -----------------------------------------------------------
mailbox_element_ptr abstract_actor_shell::next_message() {
if (!mailbox_.blocked()) {
mailbox_.fetch_more();
auto& q = mailbox_.queue();
if (q.total_task_size() > 0) {
q.inc_deficit(1);
return q.next();
}
}
return nullptr;
}
bool abstract_actor_shell::try_block_mailbox() {
return mailbox_.try_block();
}
// -- message processing -------------------------------------------------------
bool abstract_actor_shell::consume_message() {
CAF_LOG_TRACE("");
if (auto msg = next_message()) {
current_element_ = msg.get();
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, *msg);
auto mid = msg->mid;
if (!mid.is_response()) {
detail::default_invoke_result_visitor<abstract_actor_shell> visitor{this};
if (auto result = bhvr_(msg->payload)) {
visitor(*result);
} else {
auto fallback_result = (*fallback_)(msg->payload);
visit(visitor, fallback_result);
}
} else if (auto i = multiplexed_responses_.find(mid);
i != multiplexed_responses_.end()) {
auto bhvr = std::move(i->second);
multiplexed_responses_.erase(i);
auto res = bhvr(msg->payload);
if (!res) {
CAF_LOG_DEBUG("got unexpected_response");
auto err_msg = make_message(
make_error(sec::unexpected_response, std::move(msg->payload)));
bhvr(err_msg);
}
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
return true;
}
return false;
}
void abstract_actor_shell::add_multiplexed_response_handler(
message_id response_id, behavior bhvr) {
if (bhvr.timeout() != infinite)
request_response_timeout(bhvr.timeout(), response_id);
multiplexed_responses_.emplace(response_id, std::move(bhvr));
}
// -- overridden functions of abstract_actor -----------------------------------
void abstract_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* abstract_actor_shell::peek_at_next_mailbox_element() {
return mailbox().closed() || mailbox().blocked() ? nullptr : mailbox().peek();
}
// -- overridden functions of local_actor --------------------------------------
void abstract_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 abstract_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);
}
} // namespace caf::net
......@@ -18,185 +18,36 @@
#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 {
// -- constructors, destructors, and assignment operators ----------------------
// -- actor_shell --------------------------------------------------------------
actor_shell::actor_shell(actor_config& cfg, socket_manager* owner)
: super(cfg), mailbox_(policy::normal_messages{}), owner_(owner) {
mailbox_.try_block();
: super(cfg, owner) {
// nop
}
actor_shell::~actor_shell() {
// nop
}
// -- state modifiers ----------------------------------------------------------
void actor_shell::quit(error reason) {
cleanup(std::move(reason), nullptr);
}
// -- mailbox access -----------------------------------------------------------
mailbox_element_ptr actor_shell::next_message() {
if (!mailbox_.blocked()) {
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::try_block_mailbox() {
return mailbox_.try_block();
}
// -- message processing -------------------------------------------------------
bool actor_shell::consume_message() {
CAF_LOG_TRACE("");
if (auto msg = next_message()) {
current_element_ = msg.get();
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, *msg);
auto mid = msg->mid;
if (!mid.is_response()) {
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);
}
} else if (auto i = multiplexed_responses_.find(mid);
i != multiplexed_responses_.end()) {
auto bhvr = std::move(i->second);
multiplexed_responses_.erase(i);
auto res = bhvr(msg->payload);
if (!res) {
CAF_LOG_DEBUG("got unexpected_response");
auto err_msg = make_message(
make_error(sec::unexpected_response, std::move(msg->payload)));
bhvr(err_msg);
}
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
return true;
}
return false;
}
void actor_shell::add_multiplexed_response_handler(message_id response_id,
behavior bhvr) {
if (bhvr.timeout() != infinite)
request_response_timeout(bhvr.timeout(), response_id);
multiplexed_responses_.emplace(response_id, std::move(bhvr));
}
// -- overridden functions of abstract_actor -----------------------------------
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::actor_shell_ptr(strong_actor_ptr ptr) noexcept
: ptr_(std::move(ptr)) {
// nop
}
actor actor_shell_ptr::as_actor() const noexcept {
actor_shell_ptr::~actor_shell_ptr() {
if (auto ptr = get())
ptr->quit(exit_reason::normal);
}
actor_shell_ptr::handle_type actor_shell_ptr::as_actor() const noexcept {
return actor_cast<actor>(ptr_);
}
......@@ -207,17 +58,13 @@ void actor_shell_ptr::detach(error reason) {
}
}
actor_shell_ptr::~actor_shell_ptr() {
if (auto ptr = get())
ptr->quit(exit_reason::normal);
}
actor_shell* actor_shell_ptr::get() const noexcept {
actor_shell_ptr::element_type* actor_shell_ptr::get() const noexcept {
if (ptr_) {
auto ptr = actor_cast<abstract_actor*>(ptr_);
return static_cast<actor_shell*>(ptr);
} else {
return nullptr;
}
return nullptr;
}
} // namespace caf::net
......@@ -34,6 +34,11 @@ socket_manager::~socket_manager() {
close(handle_);
}
actor_system& socket_manager::system() noexcept {
CAF_ASSERT(parent_ != nullptr);
return parent_->system();
}
bool socket_manager::mask_add(operation flag) noexcept {
CAF_ASSERT(flag != operation::none);
auto x = mask();
......@@ -64,10 +69,4 @@ void socket_manager::register_writing() {
parent_->register_writing(this);
}
actor_shell_ptr socket_manager::make_actor_shell_impl() {
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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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.typed_actor_shell
#include "caf/net/typed_actor_shell.hpp"
#include "net-test.hpp"
#include "caf/byte.hpp"
#include "caf/byte_span.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"
#include "caf/typed_actor.hpp"
using namespace caf;
using namespace std::string_literals;
namespace {
using svec = std::vector<std::string>;
using string_consumer = typed_actor<result<void>(std::string)>;
struct app_t {
using input_tag = tag::stream_oriented;
app_t() = default;
explicit app_t(actor hdl) : worker(std::move(hdl)) {
// nop
}
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr down, const settings&) {
self = mgr->make_actor_shell<string_consumer>(down);
self->set_behavior([this](std::string& line) {
CAF_MESSAGE("received an asynchronous message: " << line);
lines.emplace_back(std::move(line));
});
self->set_fallback([](message& msg) -> result<message> {
CAF_FAIL("unexpected message: " << msg);
return make_error(sec::unexpected_message);
});
down->configure_read(net::receive_policy::up_to(2048));
return none;
}
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
while (self->consume_message()) {
// We set abort_reason in our response handlers in case of an error.
if (down->abort_reason())
return false;
// else: repeat.
}
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
return self->try_block_mailbox();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
CAF_FAIL("app::abort called: " << reason);
}
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buf, byte_span) {
// Seek newline character.
constexpr auto nl = byte{'\n'};
if (auto i = std::find(buf.begin(), buf.end(), nl); i != buf.end()) {
// Skip empty lines.
if (i == buf.begin()) {
consumed_bytes += 1;
auto sub_res = consume(down, buf.subspan(1), {});
return sub_res >= 0 ? sub_res + 1 : sub_res;
}
// Deserialize config value from received line.
auto num_bytes = std::distance(buf.begin(), i) + 1;
string_view line{reinterpret_cast<const char*>(buf.data()),
static_cast<size_t>(num_bytes) - 1};
config_value val;
if (auto parsed_res = config_value::parse(line)) {
val = std::move(*parsed_res);
} else {
down->abort_reason(std::move(parsed_res.error()));
return -1;
}
if (!holds_alternative<settings>(val)) {
down->abort_reason(
make_error(pec::type_mismatch,
"expected a dictionary, got a "s + val.type_name()));
return -1;
}
// Deserialize message from received dictionary.
config_value_reader reader{&val};
caf::message msg;
if (!reader.apply_object(msg)) {
down->abort_reason(reader.get_error());
return -1;
}
// Dispatch message to worker.
CAF_MESSAGE("app received a message from its socket: " << msg);
self->request(worker, std::chrono::seconds{1}, std::move(msg))
.then(
[this, down](int32_t value) mutable {
++received_responses;
// Respond with the value as string.
auto str_response = std::to_string(value);
str_response += '\n';
down->begin_output();
auto& buf = down->output_buffer();
auto bytes = as_bytes(make_span(str_response));
buf.insert(buf.end(), bytes.begin(), bytes.end());
down->end_output();
},
[down](error& err) mutable { down->abort_reason(std::move(err)); });
// Try consuming more from the buffer.
consumed_bytes += static_cast<size_t>(num_bytes);
auto sub_buf = buf.subspan(num_bytes);
auto sub_res = consume(down, sub_buf, {});
return sub_res >= 0 ? num_bytes + sub_res : sub_res;
}
return 0;
}
// Handle to the worker-under-test.
actor worker;
// Lines received as asynchronous messages.
std::vector<std::string> lines;
// Actor shell representing this app.
net::typed_actor_shell_ptr_t<string_consumer> self;
// Counts how many bytes we've consumed in total.
size_t consumed_bytes = 0;
// Counts how many response messages we've received from the worker.
size_t received_responses = 0;
};
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(self_socket_guard.socket(), true))
CAF_FAIL("nonblocking returned an error: " << err);
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);
byte tmp[1024];
auto bytes = read(self_socket_guard.socket(), make_span(tmp, 1024));
if (bytes > 0)
recv_buf.insert(recv_buf.end(), tmp, tmp + bytes);
if (!predicate())
return;
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
CAF_FAIL("reached max repeat rate without meeting the predicate");
}
void send(string_view str) {
auto res = write(self_socket_guard.socket(), as_bytes(make_span(str)));
if (res != static_cast<ptrdiff_t>(str.size()))
CAF_FAIL("expected write() to return " << str.size() << ", got: " << res);
}
net::middleman mm;
net::multiplexer mpx;
net::socket_guard<net::stream_socket> self_socket_guard;
net::socket_guard<net::stream_socket> testee_socket_guard;
std::vector<byte> recv_buf;
};
constexpr std::string_view input = R"__(
{ values = [ { "@type" : "int32_t", value: 123 } ] }
)__";
} // 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(actor shells can send requests and receive responses) {
auto worker = sys.spawn([] {
return behavior{
[](int32_t value) { return value * 2; },
};
});
auto sck = testee_socket_guard.release();
auto mgr = net::make_socket_manager<app_t, net::stream_transport>(sck, &mpx,
worker);
if (auto err = mgr->init(content(cfg)))
CAF_FAIL("mgr->init() failed: " << err);
auto& app = mgr->top_layer();
send(input);
run_while([&] { return app.consumed_bytes != input.size(); });
expect((int32_t), to(worker).with(123));
string_view expected_response = "246\n";
run_while([&] { return recv_buf.size() < expected_response.size(); });
string_view received_response{reinterpret_cast<char*>(recv_buf.data()),
recv_buf.size()};
CAF_CHECK_EQUAL(received_response, expected_response);
}
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