Commit 433bb43d authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #86

Revise layering API, add actor_shell abstraction, add WebSocket layer
parents 916fd5e9 d6115e94
...@@ -34,8 +34,12 @@ endfunction() ...@@ -34,8 +34,12 @@ endfunction()
# -- add library targets ------------------------------------------------------- # -- add library targets -------------------------------------------------------
add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/actor_proxy_impl.cpp #src/actor_proxy_impl.cpp
src/basp/application.cpp #src/basp/application.cpp
#src/endpoint_manager.cpp
#src/net/backend/tcp.cpp
#src/net/backend/test.cpp
#src/net/endpoint_manager_queue.cpp
src/basp/connection_state_strings.cpp src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp src/basp/message_type_strings.cpp
...@@ -43,16 +47,13 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} ...@@ -43,16 +47,13 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/convert_ip_endpoint.cpp src/convert_ip_endpoint.cpp
src/datagram_socket.cpp src/datagram_socket.cpp
src/defaults.cpp src/defaults.cpp
src/defaults.cpp src/detail/rfc6455.cpp
src/endpoint_manager.cpp
src/header.cpp src/header.cpp
src/host.cpp src/host.cpp
src/ip.cpp src/ip.cpp
src/message_queue.cpp src/message_queue.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/test.cpp src/net/actor_shell.cpp
src/net/backend/tcp.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
src/net/packet_writer.cpp src/net/packet_writer.cpp
...@@ -80,7 +81,8 @@ set_property(TARGET libcaf_net_obj PROPERTY POSITION_INDEPENDENT_CODE ON) ...@@ -80,7 +81,8 @@ set_property(TARGET libcaf_net_obj PROPERTY POSITION_INDEPENDENT_CODE ON)
caf_net_set_default_properties(libcaf_net_obj libcaf_net) caf_net_set_default_properties(libcaf_net_obj libcaf_net)
target_include_directories(libcaf_net INTERFACE target_include_directories(libcaf_net INTERFACE
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>) $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>
$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}>)
add_library(CAF::net ALIAS libcaf_net) add_library(CAF::net ALIAS libcaf_net)
...@@ -123,30 +125,34 @@ target_include_directories(caf-net-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/tes ...@@ -123,30 +125,34 @@ target_include_directories(caf-net-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/tes
target_link_libraries(caf-net-test PRIVATE CAF::test) target_link_libraries(caf-net-test PRIVATE CAF::test)
caf_incubator_add_test_suites(caf-net-test caf_incubator_add_test_suites(caf-net-test
net.basp.message_queue
net.basp.ping_pong
net.basp.worker
accept_socket accept_socket
#application
convert_ip_endpoint
datagram_socket
detail.rfc6455
#datagram_transport
#doorman
#endpoint_manager
header
ip
multiplexer
net.actor_shell
#net.backend.tcp
#net.basp.message_queue
#net.basp.ping_pong
#net.basp.worker
net.length_prefix_framing
net.web_socket_server
network_socket
pipe_socket pipe_socket
application
socket socket
convert_ip_endpoint
socket_guard socket_guard
datagram_socket #stream_application
stream_application
datagram_transport
stream_socket stream_socket
doorman
stream_transport stream_transport
endpoint_manager #string_application
string_application
header
tcp_sockets tcp_sockets
ip #transport_worker
transport_worker #transport_worker_dispatcher
multiplexer
transport_worker_dispatcher
udp_datagram_socket udp_datagram_socket
network_socket
net.backend.tcp
) )
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/byte.hpp"
#include "caf/detail/net_export.hpp"
#include <cstdint>
#include <vector>
namespace caf::detail {
struct CAF_NET_EXPORT rfc6455 {
// -- member types -----------------------------------------------------------
using binary_buffer = std::vector<byte>;
struct header {
bool fin;
uint8_t opcode;
uint32_t mask_key;
uint64_t payload_len;
};
// -- constants --------------------------------------------------------------
static constexpr uint8_t continuation_frame = 0x00;
static constexpr uint8_t text_frame = 0x01;
static constexpr uint8_t binary_frame = 0x02;
static constexpr uint8_t connection_close = 0x08;
static constexpr uint8_t ping = 0x09;
static constexpr uint8_t pong = 0x0A;
// -- utility functions ------------------------------------------------------
static void mask_data(uint32_t key, span<char> data);
static void mask_data(uint32_t key, span<byte> data);
static void assemble_frame(uint32_t mask_key, span<const char> data,
binary_buffer& out);
static void assemble_frame(uint32_t mask_key, span<const byte> data,
binary_buffer& out);
static void assemble_frame(uint8_t opcode, uint32_t mask_key,
span<const byte> data, binary_buffer& out);
static ptrdiff_t decode_header(span<const byte> data, header& hdr);
};
} // 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 actor_shell
: public extend<local_actor, actor_shell>::with<mixin::sender,
mixin::requester>,
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, 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);
~actor_shell() override;
// -- 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:
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
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include "caf/net/datagram_transport.hpp" #include "caf/net/datagram_transport.hpp"
#include "caf/net/defaults.hpp" #include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
#include "caf/net/endpoint_manager_queue.hpp" #include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/host.hpp" #include "caf/net/host.hpp"
......
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/callback.hpp" #include "caf/callback.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
...@@ -57,8 +58,6 @@ class CAF_NET_EXPORT application { ...@@ -57,8 +58,6 @@ class CAF_NET_EXPORT application {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using byte_span = span<const byte>;
using hub_type = detail::worker_hub<worker>; using hub_type = detail::worker_hub<worker>;
struct test_tag {}; struct test_tag {};
......
...@@ -83,9 +83,10 @@ CAF_NET_EXPORT void to_bytes(header x, byte_buffer& buf); ...@@ -83,9 +83,10 @@ CAF_NET_EXPORT void to_bytes(header x, byte_buffer& buf);
/// @relates header /// @relates header
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& x) { bool inspect(Inspector& f, header& x) {
return f(meta::type_name("basp::header"), x.type, x.payload_len, return f.object(x).fields(f.field("type", x.type),
x.operation_data); f.field("payload_len", x.payload_len),
f.field("operation_data", x.operation_data));
} }
/// @} /// @}
......
...@@ -53,8 +53,9 @@ public: ...@@ -53,8 +53,9 @@ public:
std::vector<strong_actor_ptr> fwd_stack; std::vector<strong_actor_ptr> fwd_stack;
message content; message content;
binary_deserializer source{ctx, payload}; binary_deserializer source{ctx, payload};
if (auto err = source(src_node, src_id, dst_id, fwd_stack, content)) { if (!source.apply_objects(src_node, src_id, dst_id, fwd_stack, content)) {
CAF_LOG_ERROR("could not deserialize payload: " << CAF_ARG(err)); CAF_LOG_ERROR(
"failed to deserialize payload:" << CAF_ARG(source.get_error()));
return; return;
} }
// Sanity checks. // Sanity checks.
......
...@@ -29,102 +29,105 @@ ...@@ -29,102 +29,105 @@
namespace caf::net { namespace caf::net {
/// A doorman accepts TCP connections and creates stream_transports to handle /// A connection_acceptor accepts connections from an accept socket and creates
/// them. /// socket managers to handle them via its factory.
template <class Factory> template <class Socket, class Factory>
class doorman { class connection_acceptor {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using factory_type = Factory; using input_tag = tag::io_event_oriented;
using socket_type = Socket;
using application_type = typename Factory::application_type; using factory_type = Factory;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit doorman(net::tcp_accept_socket acceptor, factory_type factory) template <class... Ts>
: acceptor_(acceptor), factory_(std::move(factory)) { explicit connection_acceptor(Ts&&... xs) : factory_(std::forward<Ts>(xs)...) {
// nop // nop
} }
// -- properties -------------------------------------------------------------
net::tcp_accept_socket handle() {
return acceptor_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> template <class ParentPtr>
error init(Parent& parent) { error init(socket_manager* owner, ParentPtr parent, const settings& config) {
// TODO: is initializing application factory nessecary? CAF_LOG_TRACE("");
if (auto err = factory_.init(parent)) owner_ = owner;
cfg_ = config;
if (auto err = factory_.init(owner, config))
return err; return err;
parent->register_reading();
return none; return none;
} }
template <class Parent> template <class ParentPtr>
bool handle_read_event(Parent& parent) { bool handle_read_event(ParentPtr parent) {
auto x = net::accept(acceptor_); CAF_LOG_TRACE("");
if (!x) { if (auto x = accept(parent->handle())) {
CAF_LOG_ERROR("accept failed:" << x.error()); socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr());
CAF_ASSERT(child != nullptr);
if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
parent->abort_reason(std::move(err));
return false; return false;
} }
auto mpx = parent.multiplexer(); return true;
if (!mpx) { } else {
CAF_LOG_DEBUG("unable to get multiplexer from parent"); CAF_LOG_ERROR("accept failed:" << x.error());
return false; return false;
} }
auto child = make_endpoint_manager(
mpx, parent.system(),
stream_transport<application_type>{*x, factory_.make()});
if (auto err = child->init())
return false;
return true;
} }
template <class Parent> template <class ParentPtr>
bool handle_write_event(Parent&) { bool handle_write_event(ParentPtr) {
CAF_LOG_ERROR("doorman received write event"); CAF_LOG_ERROR("connection_acceptor received write event");
return false; return false;
} }
template <class Parent> template <class ParentPtr>
void void abort(ParentPtr, const error& reason) {
resolve(Parent&, [[maybe_unused]] const uri& locator, const actor& listener) { CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(locator)); factory_.abort(reason);
anon_send(listener, resolve_atom_v, "doormen cannot resolve paths");
} }
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) { private:
CAF_LOG_ERROR("doorman received new_proxy" << CAF_ARG(peer) << CAF_ARG(id)); factory_type factory_;
CAF_IGNORE_UNUSED(peer);
CAF_IGNORE_UNUSED(id); socket_manager* owner_;
settings cfg_;
};
/// Converts a function object into a factory object for a
/// @ref connection_acceptor.
template <class FunctionObject>
class connection_acceptor_factory_adapter {
public:
explicit connection_acceptor_factory_adapter(FunctionObject f)
: f_(std::move(f)) {
// nop
} }
void local_actor_down(endpoint_manager&, const node_id& peer, actor_id id, connection_acceptor_factory_adapter(connection_acceptor_factory_adapter&&)
error reason) { = default;
CAF_LOG_ERROR("doorman received local_actor_down"
<< CAF_ARG(peer) << CAF_ARG(id) << CAF_ARG(reason)); error init(socket_manager*, const settings&) {
CAF_IGNORE_UNUSED(peer); return none;
CAF_IGNORE_UNUSED(id);
CAF_IGNORE_UNUSED(reason);
} }
template <class Parent> template <class Socket>
void timeout(Parent&, [[maybe_unused]] const std::string& tag, socket_manager_ptr make(Socket connected_socket, multiplexer* mpx) {
[[maybe_unused]] uint64_t id) { return f_(std::move(connected_socket), mpx);
CAF_LOG_ERROR("doorman received timeout" << CAF_ARG(tag) << CAF_ARG(id));
} }
void handle_error([[maybe_unused]] sec err) { void abort(const error&) {
CAF_LOG_ERROR("doorman encounterd error: " << err); // nop
} }
private: private:
net::tcp_accept_socket acceptor_; FunctionObject f_;
factory_type factory_;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -36,4 +36,9 @@ CAF_NET_EXPORT extern const size_t max_header_buffers; ...@@ -36,4 +36,9 @@ CAF_NET_EXPORT extern const size_t max_header_buffers;
/// Port to listen on for tcp. /// Port to listen on for tcp.
CAF_NET_EXPORT extern const uint16_t tcp_port; CAF_NET_EXPORT extern const uint16_t tcp_port;
/// Caps how much Bytes a stream transport pushes to its write buffer before
/// stopping to read from its message queue. Default TCP send buffer is 16kB (at
/// least on Linux).
constexpr auto stream_output_buf_cap = size_t{32768};
} // namespace caf::defaults::middleman } // namespace caf::defaults::middleman
...@@ -52,10 +52,16 @@ public: ...@@ -52,10 +52,16 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
actor_system& system() { actor_system& system() noexcept {
return sys_; return sys_;
} }
const actor_system_config& config() const noexcept;
// -- queue access -----------------------------------------------------------
bool at_end_of_message_queue();
endpoint_manager_queue::message_ptr next_message(); endpoint_manager_queue::message_ptr next_message();
// -- event management ------------------------------------------------------- // -- event management -------------------------------------------------------
...@@ -75,7 +81,7 @@ public: ...@@ -75,7 +81,7 @@ public:
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop. /// Initializes the manager before adding it to the multiplexer's event loop.
virtual error init() = 0; // virtual error init() = 0;
protected: protected:
bool enqueue(endpoint_manager_queue::element* ptr); bool enqueue(endpoint_manager_queue::element* ptr);
......
...@@ -40,8 +40,8 @@ public: ...@@ -40,8 +40,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys, endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
Transport trans) socket handle, Transport trans)
: super(trans.handle(), parent, sys), transport_(std::move(trans)) { : super(handle, parent, sys), transport_(std::move(trans)) {
// nop // nop
} }
...@@ -73,7 +73,7 @@ public: ...@@ -73,7 +73,7 @@ public:
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
error init() override { error init() /*override*/ {
this->register_reading(); this->register_reading();
return transport_.init(*this); return transport_.init(*this);
} }
...@@ -124,6 +124,8 @@ private: ...@@ -124,6 +124,8 @@ private:
/// Stores the id for the next timeout. /// Stores the id for the next timeout.
uint64_t next_timeout_id_; uint64_t next_timeout_id_;
error err_;
}; };
} // namespace caf::net } // 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;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstdint>
#include <cstring>
#include <memory>
#include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/error.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/tag/message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace caf::net {
/// Length-prefixed message framing for discretizing a Byte stream into messages
/// of varying size. The framing uses 4 Bytes for the length prefix, but
/// messages (including the 4 Bytes for the length prefix) are limited to a
/// maximum size of INT32_MAX. This limitation comes from the POSIX API (recv)
/// on 32-bit platforms.
template <class UpperLayer>
class length_prefix_framing {
public:
using input_tag = tag::stream_oriented;
using output_tag = tag::message_oriented;
using length_prefix_type = uint32_t;
static constexpr size_t max_message_length = INT32_MAX;
// -- interface for the upper layer ------------------------------------------
template <class LowerLayer>
class access {
public:
access(LowerLayer* lower_layer, length_prefix_framing* this_layer)
: lower_layer_(lower_layer), this_layer_(this_layer) {
// nop
}
void begin_message() {
lower_layer_->begin_output();
auto& buf = message_buffer();
message_offset_ = buf.size();
buf.insert(buf.end(), 4, byte{0});
}
byte_buffer& message_buffer() {
return lower_layer_->output_buffer();
}
bool end_message() {
using detail::to_network_order;
auto& buf = message_buffer();
auto msg_begin = buf.begin() + message_offset_;
auto msg_size = std::distance(msg_begin + 4, buf.end());
if (msg_size > 0 && msg_size < max_message_length) {
auto u32_size = to_network_order(static_cast<uint32_t>(msg_size));
memcpy(std::addressof(*msg_begin), &u32_size, 4);
return true;
} else {
abort_reason(make_error(
sec::runtime_error, msg_size == 0 ? "logic error: message of size 0"
: "maximum message size exceeded"));
return false;
}
}
bool can_send_more() const noexcept {
return lower_layer_->can_send_more();
}
void abort_reason(error reason) {
return lower_layer_->abort_reason(std::move(reason));
}
void configure_read(receive_policy policy) {
lower_layer_->configure_read(policy);
}
private:
LowerLayer* lower_layer_;
length_prefix_framing* this_layer_;
size_t message_offset_ = 0;
};
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_;
}
const auto& upper_layer() const noexcept {
return upper_layer_;
}
// -- role: upper layer ------------------------------------------------------
template <class LowerLayer>
bool prepare_send(LowerLayer& down) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.prepare_send(this_layer);
}
template <class LowerLayer>
bool done_sending(LowerLayer& down) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.done_sending(this_layer);
}
template <class LowerLayer>
void abort(LowerLayer& down, const error& reason) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.abort(this_layer, reason);
}
template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span) {
using detail::from_network_order;
if (buffer.size() < 4)
return 0;
uint32_t u32_size = 0;
memcpy(&u32_size, buffer.data(), 4);
auto msg_size = static_cast<size_t>(from_network_order(u32_size));
if (buffer.size() < msg_size + 4)
return 0;
upper_layer_.consume(down, make_span(buffer.data() + 4, msg_size));
return msg_size + 4;
}
private:
UpperLayer upper_layer_;
};
} // namespace caf::net
...@@ -26,8 +26,8 @@ ...@@ -26,8 +26,8 @@
namespace caf::net { namespace caf::net {
template <class Transport> template <class Transport>
endpoint_manager_ptr CAF_NET_EXPORT make_endpoint_manager( endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx,
const multiplexer_ptr& mpx, actor_system& sys, Transport trans) { actor_system& sys, Transport trans) {
using impl = endpoint_manager_impl<Transport>; using impl = endpoint_manager_impl<Transport>;
return make_counted<impl>(mpx, sys, std::move(trans)); return make_counted<impl>(mpx, sys, std::move(trans));
} }
......
...@@ -27,8 +27,11 @@ ...@@ -27,8 +27,11 @@
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp" #include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
namespace caf::net { namespace caf::net {
...@@ -49,8 +52,32 @@ public: ...@@ -49,8 +52,32 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
~middleman() override; ~middleman() override;
// -- socket manager functions -----------------------------------------------
///
/// @param sock An accept socket in listening mode. For a TCP socket, this
/// socket must already listen to an address plus port.
/// @param factory
template <class Socket, class Factory>
auto make_acceptor(Socket sock, Factory factory) {
using connected_socket_type = typename Socket::connected_socket_type;
if constexpr (detail::is_callable_with<Factory, connected_socket_type,
multiplexer*>::value) {
connection_acceptor_factory_adapter<Factory> adapter{std::move(factory)};
return make_acceptor(std::move(sock), std::move(adapter));
} else {
using impl = connection_acceptor<Socket, Factory>;
auto ptr = make_socket_manager<impl>(std::move(sock), &mpx_,
std::move(factory));
mpx_.init(ptr);
return ptr;
}
}
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
void start() override; void start() override;
...@@ -128,7 +155,11 @@ public: ...@@ -128,7 +155,11 @@ public:
return sys_.config(); return sys_.config();
} }
const multiplexer_ptr& mpx() const noexcept { multiplexer& mpx() noexcept {
return mpx_;
}
const multiplexer& mpx() const noexcept {
return mpx_; return mpx_;
} }
...@@ -137,10 +168,6 @@ public: ...@@ -137,10 +168,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<>) {
...@@ -159,7 +186,7 @@ private: ...@@ -159,7 +186,7 @@ private:
actor_system& sys_; actor_system& sys_;
/// Stores the global socket I/O multiplexer. /// Stores the global socket I/O multiplexer.
multiplexer_ptr mpx_; multiplexer mpx_;
/// Stores all available backends for managing peers. /// Stores all available backends for managing peers.
middleman_backend_list backends_; middleman_backend_list backends_;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Wraps a pointer to a mixed-message-oriented layer with a pointer to its
/// lower layer. Both pointers are then used to implement the interface required
/// for a mixed-message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class mixed_message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_binary_message() {
lptr_->begin_binary_message(llptr_);
}
auto& binary_message_buffer() {
return lptr_->binary_message_buffer(llptr_);
}
void end_binary_message() {
lptr_->end_binary_message(llptr_);
}
void begin_text_message() {
lptr_->begin_text_message(llptr_);
}
auto& text_message_buffer() {
return lptr_->text_message_buffer(llptr_);
}
void end_text_message() {
lptr_->end_text_message(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
mixed_message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
mixed_message_oriented_layer_ptr(const mixed_message_oriented_layer_ptr&)
= default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_mixed_message_oriented_layer_ptr(Layer* this_layer,
LowerLayerPtr down) {
using result_t = mixed_message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
...@@ -38,8 +38,7 @@ struct pollfd; ...@@ -38,8 +38,7 @@ struct pollfd;
namespace caf::net { namespace caf::net {
/// Multiplexes any number of ::socket_manager objects with a ::socket. /// Multiplexes any number of ::socket_manager objects with a ::socket.
class CAF_NET_EXPORT multiplexer class CAF_NET_EXPORT multiplexer {
: public std::enable_shared_from_this<multiplexer> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -49,10 +48,16 @@ public: ...@@ -49,10 +48,16 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
multiplexer(); /// @param parent Points to the owning middleman instance. May be `nullptr`
/// only for the purpose of unit testing if no @ref
/// socket_manager requires access to the @ref middleman or the
/// @ref actor_system.
explicit multiplexer(middleman* parent);
~multiplexer(); ~multiplexer();
// -- initialization ---------------------------------------------------------
error init(); error init();
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
...@@ -63,6 +68,12 @@ public: ...@@ -63,6 +68,12 @@ public:
/// Returns the index of `mgr` in the pollset or `-1`. /// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr); ptrdiff_t index_of(const socket_manager_ptr& mgr);
/// Returns the owning @ref middleman instance.
middleman& owner();
/// Returns the enclosing @ref actor_system.
actor_system& system();
// -- thread-safe signaling -------------------------------------------------- // -- thread-safe signaling --------------------------------------------------
/// Registers `mgr` for read events. /// Registers `mgr` for read events.
...@@ -73,6 +84,10 @@ public: ...@@ -73,6 +84,10 @@ public:
/// @thread-safe /// @thread-safe
void register_writing(const socket_manager_ptr& mgr); void register_writing(const socket_manager_ptr& mgr);
/// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe
void init(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing /// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect. /// the pipe, calls to `update` no longer have any effect.
/// @thread-safe /// @thread-safe
...@@ -123,20 +138,17 @@ protected: ...@@ -123,20 +138,17 @@ protected:
/// calling `init()`. /// calling `init()`.
std::thread::id tid_; std::thread::id tid_;
/// Used for pushing updates to the multiplexer's thread.
pipe_socket write_handle_;
/// Guards `write_handle_`. /// Guards `write_handle_`.
std::mutex write_lock_; std::mutex write_lock_;
/// Signals shutdown has been requested. /// Used for pushing updates to the multiplexer's thread.
bool shutting_down_; pipe_socket write_handle_;
};
/// @relates multiplexer /// Points to the owning middleman.
using multiplexer_ptr = std::shared_ptr<multiplexer>; middleman* owner_;
/// @relates multiplexer /// Signals whether shutdown has been requested.
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>; bool shutting_down_ = false;
};
} // namespace caf::net } // namespace caf::net
...@@ -36,9 +36,19 @@ public: ...@@ -36,9 +36,19 @@ public:
using msg_buf = std::array<byte, sizeof(intptr_t) + 1>; using msg_buf = std::array<byte, sizeof(intptr_t) + 1>;
// -- constants --------------------------------------------------------------
static constexpr uint8_t register_reading_code = 0x00;
static constexpr uint8_t register_writing_code = 0x01;
static constexpr uint8_t init_manager_code = 0x02;
static constexpr uint8_t shutdown_code = 0x04;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, const multiplexer_ptr& parent); pollset_updater(pipe_socket read_handle, multiplexer* parent);
~pollset_updater() override; ~pollset_updater() override;
...@@ -51,6 +61,8 @@ public: ...@@ -51,6 +61,8 @@ public:
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
error init(const settings& config) override;
bool handle_read_event() override; bool handle_read_event() override;
bool handle_write_event() override; bool handle_write_event() override;
...@@ -59,7 +71,7 @@ public: ...@@ -59,7 +71,7 @@ public:
private: private:
msg_buf buf_; msg_buf buf_;
size_t buf_size_; size_t buf_size_ = 0;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -18,45 +18,32 @@ ...@@ -18,45 +18,32 @@
#pragma once #pragma once
#include <cstddef> #include <cstdint>
#include <string>
#include <utility>
#include "caf/config.hpp"
namespace caf::net { namespace caf::net {
enum class CAF_NET_EXPORT receive_policy_flag : unsigned { struct receive_policy {
at_least, uint32_t min_size;
at_most, uint32_t max_size;
exactly
};
inline std::string to_string(receive_policy_flag x) {
return x == receive_policy_flag::at_least
? "at_least"
: (x == receive_policy_flag::at_most ? "at_most" : "exactly");
}
class CAF_NET_EXPORT receive_policy { /// @pre `min_size > 0`
public: /// @pre `min_size <= max_size`
receive_policy() = delete; static constexpr receive_policy between(uint32_t min_size,
uint32_t max_size) {
using config = std::pair<receive_policy_flag, size_t>; return {min_size, max_size};
}
static config at_least(size_t num_bytes) { /// @pre `size > 0`
CAF_ASSERT(num_bytes > 0); static constexpr receive_policy exactly(uint32_t size) {
return {receive_policy_flag::at_least, num_bytes}; return {size, size};
} }
static config at_most(size_t num_bytes) { static constexpr receive_policy up_to(uint32_t max_size) {
CAF_ASSERT(num_bytes > 0); return {1, max_size};
return {receive_policy_flag::at_most, num_bytes};
} }
static config exactly(size_t num_bytes) { static constexpr receive_policy stop() {
CAF_ASSERT(num_bytes > 0); return {0, 0};
return {receive_policy_flag::exactly, num_bytes};
} }
}; };
......
...@@ -55,9 +55,8 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> { ...@@ -55,9 +55,8 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> {
/// @relates socket /// @relates socket
template <class Inspector> template <class Inspector>
typename Inspector::result_type CAF_NET_EXPORT inspect(Inspector& f, bool inspect(Inspector& f, socket& x) {
socket& x) { return f.object(x).fields(f.field("id", x.id));
return f(x.id);
} }
/// Denotes the invalid socket. /// Denotes the invalid socket.
...@@ -65,7 +64,7 @@ constexpr auto invalid_socket = socket{invalid_socket_id}; ...@@ -65,7 +64,7 @@ constexpr auto invalid_socket = socket{invalid_socket_id};
/// Converts between different socket types. /// Converts between different socket types.
template <class To, class From> template <class To, class From>
To CAF_NET_EXPORT socket_cast(From x) { To socket_cast(From x) {
return To{x.id}; return To{x.id};
} }
......
...@@ -18,24 +18,32 @@ ...@@ -18,24 +18,32 @@
#pragma once #pragma once
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/make_counted.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp" #include "caf/net/operation.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/tag/io_event_oriented.hpp"
namespace caf::net { namespace caf::net {
/// Manages the lifetime of a single socket and handles any I/O events on it. /// Manages the lifetime of a single socket and handles any I/O events on it.
class CAF_NET_EXPORT socket_manager : public ref_counted { class CAF_NET_EXPORT socket_manager : public ref_counted {
public: public:
// -- member types -----------------------------------------------------------
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
/// @pre `parent != nullptr`
/// @pre `handle != invalid_socket` /// @pre `handle != invalid_socket`
socket_manager(socket handle, const multiplexer_ptr& parent); /// @pre `parent != nullptr`
socket_manager(socket handle, multiplexer* parent);
~socket_manager() override; ~socket_manager() override;
...@@ -46,13 +54,28 @@ public: ...@@ -46,13 +54,28 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns the managed socket. /// Returns the managed socket.
socket handle() const noexcept { socket handle() const {
return handle_; return handle_;
} }
/// Returns a pointer to the multiplexer running this `socket_manager`. /// Returns the owning @ref multiplexer instance.
multiplexer_ptr multiplexer() const { multiplexer& mpx() noexcept {
return parent_.lock(); return *parent_;
}
/// Returns the owning @ref multiplexer instance.
const multiplexer& mpx() const noexcept {
return *parent_;
}
/// Returns a pointer to the owning @ref multiplexer instance.
multiplexer* mpx_ptr() noexcept {
return parent_;
}
/// Returns a pointer to the owning @ref multiplexer instance.
const multiplexer* mpx_ptr() const noexcept {
return parent_;
} }
/// Returns registered operations (read, write, or both). /// Returns registered operations (read, write, or both).
...@@ -70,6 +93,38 @@ public: ...@@ -70,6 +93,38 @@ public:
/// @pre `flag != operation::none` /// @pre `flag != operation::none`
bool mask_del(operation flag) noexcept; bool mask_del(operation flag) noexcept;
const error& abort_reason() const noexcept {
return abort_reason_;
}
void abort_reason(error reason) noexcept {
abort_reason_ = std::move(reason);
}
template <class... Ts>
const error& abort_reason_or(Ts&&... xs) {
if (!abort_reason_)
abort_reason_ = make_error(std::forward<Ts>(xs)...);
return abort_reason_;
}
// -- factories --------------------------------------------------------------
template <class LowerLayerPtr, class FallbackHandler>
actor_shell_ptr make_actor_shell(LowerLayerPtr, FallbackHandler f) {
auto ptr = make_actor_shell_impl();
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> {
down->abort_reason(make_error(sec::unexpected_message, std::move(msg)));
return make_error(sec::unexpected_message);
});
}
// -- event loop management -------------------------------------------------- // -- event loop management --------------------------------------------------
void register_reading(); void register_reading();
...@@ -78,6 +133,8 @@ public: ...@@ -78,6 +133,8 @@ public:
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
virtual error init(const settings& config) = 0;
/// Called whenever the socket received new data. /// Called whenever the socket received new data.
virtual bool handle_read_event() = 0; virtual bool handle_read_event() = 0;
...@@ -95,9 +152,140 @@ protected: ...@@ -95,9 +152,140 @@ protected:
operation mask_; operation mask_;
weak_multiplexer_ptr parent_; multiplexer* parent_;
error abort_reason_;
private:
actor_shell_ptr make_actor_shell_impl();
};
template <class Protocol>
class socket_manager_impl : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using output_tag = tag::io_event_oriented;
using socket_type = typename Protocol::socket_type;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
socket_manager_impl(socket_type handle, multiplexer* mpx, Ts&&... xs)
: socket_manager(handle, mpx), protocol_(std::forward<Ts>(xs)...) {
// nop
}
// -- initialization ---------------------------------------------------------
error init(const settings& config) override {
CAF_LOG_TRACE("");
if (auto err = nonblocking(handle(), true)) {
CAF_LOG_ERROR("failed to set nonblocking flag in socket:" << err);
return err;
}
return protocol_.init(static_cast<socket_manager*>(this), this, config);
}
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
socket_type handle() const {
return socket_cast<socket_type>(this->handle_);
}
// -- event callbacks --------------------------------------------------------
bool handle_read_event() override {
return protocol_.handle_read_event(this);
}
bool handle_write_event() override {
return protocol_.handle_write_event(this);
}
void handle_error(sec code) override {
abort_reason_ = code;
return protocol_.abort(this, abort_reason_);
}
auto& protocol() noexcept {
return protocol_;
}
const auto& protocol() const noexcept {
return protocol_;
}
auto& top_layer() noexcept {
return climb(protocol_);
}
const auto& top_layer() const noexcept {
return climb(protocol_);
}
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_;
}; };
/// @relates socket_manager
using socket_manager_ptr = intrusive_ptr<socket_manager>; using socket_manager_ptr = intrusive_ptr<socket_manager>;
template <class B, template <class> class... Layers>
struct make_socket_manager_helper;
template <class B>
struct make_socket_manager_helper<B> {
using type = B;
};
template <class B, template <class> class Layer,
template <class> class... Layers>
struct make_socket_manager_helper<B, Layer, Layers...>
: make_socket_manager_helper<Layer<B>, Layers...> {
using upper_input = typename B::input_tag;
using lower_output = typename Layer<B>::output_tag;
static_assert(std::is_same<upper_input, lower_output>::value);
};
template <class B, template <class> class... Layers>
using make_socket_manager_helper_t =
typename make_socket_manager_helper<B, Layers...>::type;
template <class B, template <class> class... Layers>
using socket_type_t =
typename make_socket_manager_helper_t<B, Layers...,
socket_manager_impl>::socket_type;
template <class App, template <class> class... Layers, class... Ts>
auto make_socket_manager(socket_type_t<App, Layers...> handle, multiplexer* mpx,
Ts&&... xs) {
using impl
= make_socket_manager_helper_t<App, Layers..., socket_manager_impl>;
static_assert(std::is_base_of<socket, typename impl::socket_type>::value);
return make_counted<impl>(std::move(handle), mpx, std::forward<Ts>(xs)...);
}
} // 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. *
******************************************************************************/
#pragma once
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Wraps a pointer to a stream-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// stream-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class stream_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_output() {
lptr_->begin_output(llptr_);
}
byte_buffer& output_buffer() {
return lptr_->output_buffer(llptr_);
}
void end_output() {
lptr_->end_output(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
void configure_read(receive_policy policy) {
lptr_->configure_read(llptr_, policy);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
stream_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
stream_oriented_layer_ptr(const stream_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_stream_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
return stream_oriented_layer_ptr<Layer, LowerLayerPtr>{this_layer, down};
}
} // namespace caf::net
This diff is collapsed.
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
...@@ -31,6 +32,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket { ...@@ -31,6 +32,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
using super = network_socket; using super = network_socket;
using super::super; using super::super;
using connected_socket_type = tcp_stream_socket;
}; };
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
......
...@@ -173,11 +173,6 @@ public: ...@@ -173,11 +173,6 @@ public:
// -- (pure) virtual functions ----------------------------------------------- // -- (pure) virtual functions -----------------------------------------------
/// Configures this transport for the next read event.
virtual void configure_read(receive_policy::config) {
// nop
}
/// Called by the endpoint manager when the transport can read data from its /// Called by the endpoint manager when the transport can read data from its
/// socket. /// socket.
virtual bool handle_read_event(endpoint_manager&) = 0; virtual bool handle_read_event(endpoint_manager&) = 0;
......
This diff is collapsed.
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
namespace caf::tag {
struct io_event_oriented {};
} // namespace caf::tag
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
namespace caf::tag {
struct message_oriented {};
} // namespace caf::tag
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
namespace caf::tag {
struct mixed_message_oriented {};
} // namespace caf::tag
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
namespace caf::tag {
struct stream_oriented {};
} // namespace caf::tag
...@@ -62,14 +62,15 @@ error application::write_message( ...@@ -62,14 +62,15 @@ error application::write_message(
if (src != nullptr) { if (src != nullptr) {
auto src_id = src->id(); auto src_id = src->id();
system().registry().put(src_id, src); system().registry().put(src_id, src);
if (auto err = sink(src->node(), src_id, dst->id(), ptr->msg->stages)) if (!sink.apply_objects(src->node(), src_id, dst->id(), ptr->msg->stages))
return err; return sink.get_error();
} else { } else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages)) if (!sink.apply_objects(node_id{}, actor_id{0}, dst->id(),
return err; ptr->msg->stages))
return sink.get_error();
} }
if (auto err = sink(ptr->msg->content())) if (!sink.apply_objects(ptr->msg->content()))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::actor_message, to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_buf.size()), static_cast<uint32_t>(payload_buf.size()),
...@@ -84,8 +85,8 @@ void application::resolve(packet_writer& writer, string_view path, ...@@ -84,8 +85,8 @@ void application::resolve(packet_writer& writer, string_view path,
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener)); CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(path)) { if (!sink.apply_objects(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err)); CAF_LOG_ERROR("unable to serialize path:" << sink.get_error());
return; return;
} }
auto req_id = next_request_id_++; auto req_id = next_request_id_++;
...@@ -108,7 +109,7 @@ void application::local_actor_down(packet_writer& writer, actor_id id, ...@@ -108,7 +109,7 @@ void application::local_actor_down(packet_writer& writer, actor_id id,
error reason) { error reason) {
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{system(), payload}; binary_serializer sink{system(), payload};
if (auto err = sink(reason)) if (!sink.apply_objects(reason))
CAF_RAISE_ERROR("unable to serialize an error"); CAF_RAISE_ERROR("unable to serialize an error");
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message, to_bytes(header{message_type::down_message,
...@@ -218,8 +219,8 @@ error application::handle_handshake(packet_writer&, header hdr, ...@@ -218,8 +219,8 @@ error application::handle_handshake(packet_writer&, header hdr,
node_id peer_id; node_id peer_id;
std::vector<std::string> app_ids; std::vector<std::string> app_ids;
binary_deserializer source{&executor_, payload}; binary_deserializer source{&executor_, payload};
if (auto err = source(peer_id, app_ids)) if (!source.apply_objects(peer_id, app_ids))
return err; return source.get_error();
if (!peer_id || app_ids.empty()) if (!peer_id || app_ids.empty())
return ec::invalid_handshake; return ec::invalid_handshake;
auto ids = get_or(system().config(), "caf.middleman.app-identifiers", auto ids = get_or(system().config(), "caf.middleman.app-identifiers",
...@@ -277,8 +278,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -277,8 +278,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
CAF_ASSERT(rec_hdr.type == message_type::resolve_request); CAF_ASSERT(rec_hdr.type == message_type::resolve_request);
size_t path_size = 0; size_t path_size = 0;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source.begin_sequence(path_size)) if (!source.begin_sequence(path_size))
return err; return source.get_error();
// We expect the received buffer to contain the path only. // We expect the received buffer to contain the path only.
if (path_size != source.remaining()) if (path_size != source.remaining())
return ec::invalid_payload; return ec::invalid_payload;
...@@ -298,8 +299,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -298,8 +299,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
// TODO: figure out how to obtain messaging interface. // TODO: figure out how to obtain messaging interface.
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(aid, ifs)) if (!sink.apply_objects(aid, ifs))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_response, to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
...@@ -323,9 +324,9 @@ error application::handle_resolve_response(packet_writer&, header received_hdr, ...@@ -323,9 +324,9 @@ error application::handle_resolve_response(packet_writer&, header received_hdr,
actor_id aid; actor_id aid;
std::set<std::string> ifs; std::set<std::string> ifs;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source(aid, ifs)) { if (!source.apply_objects(aid, ifs)) {
anon_send(i->second, sec::remote_lookup_failed); anon_send(i->second, sec::remote_lookup_failed);
return err; return source.get_error();
} }
if (aid == 0) { if (aid == 0) {
anon_send(i->second, strong_actor_ptr{nullptr}, std::move(ifs)); anon_send(i->second, strong_actor_ptr{nullptr}, std::move(ifs));
...@@ -354,8 +355,8 @@ error application::handle_monitor_message(packet_writer& writer, ...@@ -354,8 +355,8 @@ error application::handle_monitor_message(packet_writer& writer,
error reason = exit_reason::unknown; error reason = exit_reason::unknown;
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(reason)) if (!sink.apply_objects(reason))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message, to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
...@@ -372,17 +373,20 @@ error application::handle_down_message(packet_writer&, header received_hdr, ...@@ -372,17 +373,20 @@ error application::handle_down_message(packet_writer&, header received_hdr,
<< CAF_ARG2("received.size", received.size())); << CAF_ARG2("received.size", received.size()));
error reason; error reason;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source(reason)) if (!source.apply_objects(reason))
return err; return source.get_error();
proxies_.erase(peer_id_, received_hdr.operation_data, std::move(reason)); proxies_.erase(peer_id_, received_hdr.operation_data, std::move(reason));
return none; return none;
} }
error application::generate_handshake(byte_buffer& buf) { error application::generate_handshake(byte_buffer& buf) {
binary_serializer sink{&executor_, buf}; binary_serializer sink{&executor_, buf};
return sink(system().node(), if (!sink.apply_objects(system().node(),
get_or(system().config(), "caf.middleman.app-identifiers", get_or(system().config(),
application::default_app_ids())); "caf.middleman.app-identifiers",
application::default_app_ids())))
return sink.get_error();
return none;
} }
} // namespace caf::net::basp } // namespace caf::net::basp
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/rfc6455.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
#include <cstring>
namespace caf::detail {
void rfc6455::mask_data(uint32_t key, span<char> data) {
mask_data(key, as_writable_bytes(data));
}
void rfc6455::mask_data(uint32_t key, span<byte> data) {
auto no_key = to_network_order(key);
byte arr[4];
memcpy(arr, &no_key, 4);
size_t i = 0;
for (auto& x : data) {
x = x ^ arr[i];
i = (i + 1) % 4;
}
}
void rfc6455::assemble_frame(uint32_t mask_key, span<const char> data,
binary_buffer& out) {
assemble_frame(text_frame, mask_key, as_bytes(data), out);
}
void rfc6455::assemble_frame(uint32_t mask_key, span<const byte> data,
binary_buffer& out) {
assemble_frame(binary_frame, mask_key, data, out);
}
void rfc6455::assemble_frame(uint8_t opcode, uint32_t mask_key,
span<const byte> data, binary_buffer& out) {
// First 8 bits: FIN flag + opcode (we never fragment frames).
out.push_back(byte{static_cast<uint8_t>(0x80 | opcode)});
// Mask flag + payload length (7 bits, 7+16 bits, or 7+64 bits)
auto mask_bit = byte{static_cast<uint8_t>(mask_key == 0 ? 0x00 : 0x80)};
if (data.size() < 126) {
auto len = static_cast<uint8_t>(data.size());
out.push_back(mask_bit | byte{len});
} else if (data.size() < std::numeric_limits<uint16_t>::max()) {
auto len = static_cast<uint16_t>(data.size());
auto no_len = to_network_order(len);
byte len_data[2];
memcpy(len_data, &no_len, 2);
out.push_back(mask_bit | byte{126});
for (auto x : len_data)
out.push_back(x);
} else {
auto len = static_cast<uint64_t>(data.size());
auto no_len = to_network_order(len);
byte len_data[8];
memcpy(len_data, &no_len, 8);
out.push_back(mask_bit | byte{127});
out.insert(out.end(), len_data, len_data + 8);
}
// Masking key: 0 or 4 bytes.
if (mask_key != 0) {
auto no_key = to_network_order(mask_key);
byte key_data[4];
memcpy(key_data, &no_key, 4);
out.insert(out.end(), key_data, key_data + 4);
}
// Application data.
out.insert(out.end(), data.begin(), data.end());
}
ptrdiff_t rfc6455::decode_header(span<const byte> data, header& hdr) {
if (data.size() < 2)
return 0;
auto byte1 = to_integer<uint8_t>(data[0]);
auto byte2 = to_integer<uint8_t>(data[1]);
// Fetch FIN flag and opcode.
hdr.fin = (byte1 & 0x80) != 0;
hdr.opcode = byte1 & 0x0F;
// Decode mask bit and payload length field.
bool masked = (byte2 & 0x80) != 0;
auto len_field = byte2 & 0x7F;
size_t header_length;
if (len_field < 126) {
header_length = 2 + (masked ? 4 : 0);
hdr.payload_len = len_field;
} else if (len_field == 126) {
header_length = 4 + (masked ? 4 : 0);
} else {
header_length = 10 + (masked ? 4 : 0);
}
// Make sure we can read all the data we need.
if (data.size() < header_length)
return 0;
// Start decoding remaining header bytes.
const byte* p = data.data() + 2;
// Fetch payload size
if (len_field == 126) {
uint16_t no_len;
memcpy(&no_len, p, 2);
hdr.payload_len = from_network_order(no_len);
p += 2;
} else if (len_field == 127) {
uint64_t no_len;
memcpy(&no_len, p, 8);
hdr.payload_len = from_network_order(no_len);
p += 8;
}
// Fetch mask key.
if (masked) {
uint32_t no_key;
memcpy(&no_key, p, 4);
hdr.mask_key = from_network_order(no_key);
p += 4;
} else {
hdr.mask_key = 0;
}
// No extension bits allowed.
if (byte1 & 0x70)
return -1;
// Verify opcode and return number of consumed bytes.
switch (hdr.opcode) {
case continuation_frame:
case text_frame:
case binary_frame:
case connection_close:
case ping:
case pong:
return static_cast<ptrdiff_t>(header_length);
default:
return -1;
}
}
} // namespace caf::detail
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
namespace caf::net { namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent, endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys) actor_system& sys)
: super(handle, parent), sys_(sys), queue_(unit, unit, unit) { : super(handle, parent), sys_(sys), queue_(unit, unit, unit) {
...@@ -35,6 +37,18 @@ endpoint_manager::~endpoint_manager() { ...@@ -35,6 +37,18 @@ endpoint_manager::~endpoint_manager() {
// nop // nop
} }
// -- properties ---------------------------------------------------------------
const actor_system_config& endpoint_manager::config() const noexcept {
return sys_.config();
}
// -- queue access -------------------------------------------------------------
bool endpoint_manager::at_end_of_message_queue() {
return queue_.empty() && queue_.try_block();
}
endpoint_manager_queue::message_ptr endpoint_manager::next_message() { endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
if (queue_.blocked()) if (queue_.blocked())
return nullptr; return nullptr;
...@@ -50,6 +64,8 @@ endpoint_manager_queue::message_ptr endpoint_manager::next_message() { ...@@ -50,6 +64,8 @@ endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
return result; return result;
} }
// -- event management ---------------------------------------------------------
void endpoint_manager::resolve(uri locator, actor listener) { void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result; using intrusive::inbox_result;
using event_type = endpoint_manager_queue::event; using event_type = endpoint_manager_queue::event;
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/operation.hpp" #include "caf/net/operation.hpp"
#include "caf/net/pollset_updater.hpp" #include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
...@@ -78,7 +79,9 @@ short to_bitmask(operation op) { ...@@ -78,7 +79,9 @@ short to_bitmask(operation op) {
} // namespace } // namespace
multiplexer::multiplexer() : shutting_down_(false) { // -- constructors, destructors, and assignment operators ----------------------
multiplexer::multiplexer(middleman* owner) : owner_(owner) {
// nop // nop
} }
...@@ -86,15 +89,23 @@ multiplexer::~multiplexer() { ...@@ -86,15 +89,23 @@ multiplexer::~multiplexer() {
// nop // nop
} }
// -- initialization -----------------------------------------------------------
error multiplexer::init() { error multiplexer::init() {
auto pipe_handles = make_pipe(); auto pipe_handles = make_pipe();
if (!pipe_handles) if (!pipe_handles)
return std::move(pipe_handles.error()); return std::move(pipe_handles.error());
add(make_counted<pollset_updater>(pipe_handles->first, shared_from_this())); auto updater = make_counted<pollset_updater>(pipe_handles->first, this);
settings dummy;
if (auto err = updater->init(dummy))
return err;
add(std::move(updater));
write_handle_ = pipe_handles->second; write_handle_ = pipe_handles->second;
return none; return none;
} }
// -- properties ---------------------------------------------------------------
size_t multiplexer::num_socket_managers() const noexcept { size_t multiplexer::num_socket_managers() const noexcept {
return managers_.size(); return managers_.size();
} }
...@@ -106,7 +117,19 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) { ...@@ -106,7 +117,19 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
return i == last ? -1 : std::distance(first, i); return i == last ? -1 : std::distance(first, i);
} }
middleman& multiplexer::owner() {
CAF_ASSERT(owner_ != nullptr);
return *owner_;
}
actor_system& multiplexer::system() {
return owner().system();
}
// -- thread-safe signaling ----------------------------------------------------
void multiplexer::register_reading(const socket_manager_ptr& mgr) { void multiplexer::register_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (shutting_down_) { if (shutting_down_) {
// discard // discard
...@@ -120,11 +143,12 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) { ...@@ -120,11 +143,12 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) {
add(mgr); add(mgr);
} }
} else { } else {
write_to_pipe(0, mgr); write_to_pipe(pollset_updater::register_reading_code, mgr);
} }
} }
void multiplexer::register_writing(const socket_manager_ptr& mgr) { void multiplexer::register_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (shutting_down_) { if (shutting_down_) {
// discard // discard
...@@ -138,11 +162,30 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) { ...@@ -138,11 +162,30 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) {
add(mgr); add(mgr);
} }
} else { } else {
write_to_pipe(1, mgr); write_to_pipe(pollset_updater::register_writing_code, mgr);
}
}
void multiplexer::init(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
if (shutting_down_) {
// discard
} else {
if (auto err = mgr->init(content(system().config()))) {
CAF_LOG_ERROR("mgr->init failed: " << err);
// The socket manager should not register itself for any events if
// initialization fails. So there's probably nothing we could do
// here other than discarding the manager.
}
}
} else {
write_to_pipe(pollset_updater::init_manager_code, mgr);
} }
} }
void multiplexer::close_pipe() { void multiplexer::close_pipe() {
CAF_LOG_TRACE("");
std::lock_guard<std::mutex> guard{write_lock_}; std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) { if (write_handle_ != invalid_socket) {
close(write_handle_); close(write_handle_);
...@@ -150,7 +193,10 @@ void multiplexer::close_pipe() { ...@@ -150,7 +193,10 @@ void multiplexer::close_pipe() {
} }
} }
// -- control flow -------------------------------------------------------------
bool multiplexer::poll_once(bool blocking) { bool multiplexer::poll_once(bool blocking) {
CAF_LOG_TRACE(CAF_ARG(blocking));
if (pollset_.empty()) if (pollset_.empty())
return false; return false;
// We'll call poll() until poll() succeeds or fails. // We'll call poll() until poll() succeeds or fails.
...@@ -216,6 +262,7 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -216,6 +262,7 @@ bool multiplexer::poll_once(bool blocking) {
} }
void multiplexer::set_thread_id() { void multiplexer::set_thread_id() {
CAF_LOG_TRACE("");
tid_ = std::this_thread::get_id(); tid_ = std::this_thread::get_id();
} }
...@@ -226,7 +273,9 @@ void multiplexer::run() { ...@@ -226,7 +273,9 @@ void multiplexer::run() {
} }
void multiplexer::shutdown() { void multiplexer::shutdown() {
CAF_LOG_TRACE("");
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
CAF_LOG_DEBUG("initiate shutdown");
shutting_down_ = true; shutting_down_ = true;
// First manager is the pollset_updater. Skip it and delete later. // First manager is the pollset_updater. Skip it and delete later.
for (size_t i = 1; i < managers_.size();) { for (size_t i = 1; i < managers_.size();) {
...@@ -242,10 +291,13 @@ void multiplexer::shutdown() { ...@@ -242,10 +291,13 @@ void multiplexer::shutdown() {
} }
close_pipe(); close_pipe();
} else { } else {
CAF_LOG_DEBUG("push shutdown event to pipe");
write_to_pipe(4, nullptr); write_to_pipe(4, nullptr);
} }
} }
// -- utility functions --------------------------------------------------------
short multiplexer::handle(const socket_manager_ptr& mgr, short events, short multiplexer::handle(const socket_manager_ptr& mgr, short events,
short revents) { short revents) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle())); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
...@@ -279,6 +331,7 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events, ...@@ -279,6 +331,7 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events,
} }
void multiplexer::add(socket_manager_ptr mgr) { void multiplexer::add(socket_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
CAF_ASSERT(index_of(mgr) == -1); CAF_ASSERT(index_of(mgr) == -1);
pollfd new_entry{socket_cast<socket_id>(mgr->handle()), pollfd new_entry{socket_cast<socket_id>(mgr->handle()),
to_bitmask(mgr->mask()), 0}; to_bitmask(mgr->mask()), 0};
...@@ -293,10 +346,13 @@ void multiplexer::del(ptrdiff_t index) { ...@@ -293,10 +346,13 @@ void multiplexer::del(ptrdiff_t index) {
} }
void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
CAF_ASSERT(opcode == 0 || opcode == 1 || opcode == 4); CAF_ASSERT(opcode == pollset_updater::register_reading_code
CAF_ASSERT(mgr != nullptr || opcode == 4); || opcode == pollset_updater::register_writing_code
|| opcode == pollset_updater::init_manager_code
|| opcode == pollset_updater::shutdown_code);
CAF_ASSERT(mgr != nullptr || opcode == pollset_updater::shutdown_code);
pollset_updater::msg_buf buf; pollset_updater::msg_buf buf;
if (opcode != 4) if (opcode != pollset_updater::shutdown_code)
mgr->ref(); mgr->ref();
buf[0] = static_cast<byte>(opcode); buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(mgr.get()); auto value = reinterpret_cast<intptr_t>(mgr.get());
...@@ -307,7 +363,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { ...@@ -307,7 +363,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
if (write_handle_ != invalid_socket) if (write_handle_ != invalid_socket)
res = write(write_handle_, buf); res = write(write_handle_, buf);
} }
if (res <= 0 && opcode != 4) if (res <= 0 && opcode != pollset_updater::shutdown_code)
mgr->deref(); mgr->deref();
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 {
// -- constructors, destructors, and assignment operators ----------------------
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
}
// -- 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(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
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/net/basp/application.hpp" #include "caf/net/basp/application.hpp"
#include "caf/net/basp/application_factory.hpp" #include "caf/net/basp/application_factory.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/doorman.hpp" #include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp" #include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
......
...@@ -25,7 +25,6 @@ ...@@ -25,7 +25,6 @@
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/middleman_backend.hpp" #include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -37,8 +36,8 @@ void middleman::init_global_meta_objects() { ...@@ -37,8 +36,8 @@ void middleman::init_global_meta_objects() {
caf::init_global_meta_objects<id_block::net_module>(); caf::init_global_meta_objects<id_block::net_module>();
} }
middleman::middleman(actor_system& sys) : sys_(sys) { middleman::middleman(actor_system& sys) : sys_(sys), mpx_(this) {
mpx_ = std::make_shared<multiplexer>(); // nop
} }
middleman::~middleman() { middleman::~middleman() {
...@@ -47,15 +46,13 @@ middleman::~middleman() { ...@@ -47,15 +46,13 @@ middleman::~middleman() {
void middleman::start() { void middleman::start() {
if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) { if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) {
auto mpx = mpx_; mpx_thread_ = std::thread{[this] {
auto sys_ptr = &system(); CAF_SET_LOGGER_SYS(&sys_);
mpx_thread_ = std::thread{[mpx, sys_ptr] { detail::set_thread_name("caf.net.mpx");
CAF_SET_LOGGER_SYS(sys_ptr); sys_.thread_started();
detail::set_thread_name("caf.multiplexer"); mpx_.set_thread_id();
sys_ptr->thread_started(); mpx_.run();
mpx->set_thread_id(); sys_.thread_terminates();
mpx->run();
sys_ptr->thread_terminates();
}}; }};
} }
} }
...@@ -63,23 +60,23 @@ void middleman::start() { ...@@ -63,23 +60,23 @@ void middleman::start() {
void middleman::stop() { void middleman::stop() {
for (const auto& backend : backends_) for (const auto& backend : backends_)
backend->stop(); backend->stop();
mpx_->shutdown(); mpx_.shutdown();
if (mpx_thread_.joinable()) if (mpx_thread_.joinable())
mpx_thread_.join(); mpx_thread_.join();
else else
mpx_->run(); mpx_.run();
} }
void middleman::init(actor_system_config& cfg) { void middleman::init(actor_system_config& cfg) {
if (auto err = mpx_->init()) { if (auto err = mpx_.init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err); CAF_LOG_ERROR("mpx_.init() failed: " << err);
CAF_RAISE_ERROR("mpx->init() failed"); CAF_RAISE_ERROR("mpx_.init() failed");
} }
if (auto node_uri = get_if<uri>(&cfg, "caf.middleman.this-node")) { if (auto node_uri = get_if<uri>(&cfg, "caf.middleman.this-node")) {
auto this_node = make_node_id(std::move(*node_uri)); auto this_node = make_node_id(std::move(*node_uri));
sys_.node_.swap(this_node); sys_.node_.swap(this_node);
} else { } else {
CAF_RAISE_ERROR("no valid entry for caf.middleman.this-node found"); // CAF_RAISE_ERROR("no valid entry for caf.middleman.this-node found");
} }
for (auto& backend : backends_) for (auto& backend : backends_)
if (auto err = backend->init()) { if (auto err = backend->init()) {
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "caf/detail/socket_sys_aliases.hpp" #include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
...@@ -81,11 +82,13 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() { ...@@ -81,11 +82,13 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
} }
ptrdiff_t write(pipe_socket x, span<const byte> buf) { ptrdiff_t write(pipe_socket x, span<const byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), return ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size()); buf.size());
} }
ptrdiff_t read(pipe_socket x, span<byte> buf) { ptrdiff_t read(pipe_socket x, span<byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), return ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size()); buf.size());
} }
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <cstring> #include <cstring>
#include "caf/actor_system.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
...@@ -28,20 +29,24 @@ ...@@ -28,20 +29,24 @@
namespace caf::net { namespace caf::net {
pollset_updater::pollset_updater(pipe_socket read_handle, pollset_updater::pollset_updater(pipe_socket read_handle, multiplexer* parent)
const multiplexer_ptr& parent) : super(read_handle, parent) {
: super(read_handle, parent), buf_size_(0) {
mask_ = operation::read; mask_ = operation::read;
if (auto err = nonblocking(read_handle, true))
CAF_LOG_ERROR("nonblocking failed: " << err);
} }
pollset_updater::~pollset_updater() { pollset_updater::~pollset_updater() {
// nop // nop
} }
error pollset_updater::init(const settings&) {
CAF_LOG_TRACE("");
return nonblocking(handle(), true);
}
bool pollset_updater::handle_read_event() { bool pollset_updater::handle_read_event() {
CAF_LOG_TRACE("");
for (;;) { for (;;) {
CAF_ASSERT((buf_.size() - buf_size_) > 0);
auto num_bytes = read(handle(), make_span(buf_.data() + buf_size_, auto num_bytes = read(handle(), make_span(buf_.data() + buf_size_,
buf_.size() - buf_size_)); buf_.size() - buf_size_));
if (num_bytes > 0) { if (num_bytes > 0) {
...@@ -52,25 +57,29 @@ bool pollset_updater::handle_read_event() { ...@@ -52,25 +57,29 @@ bool pollset_updater::handle_read_event() {
intptr_t value; intptr_t value;
memcpy(&value, buf_.data() + 1, sizeof(intptr_t)); memcpy(&value, buf_.data() + 1, sizeof(intptr_t));
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false}; socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
if (auto ptr = parent_.lock()) {
switch (opcode) { switch (opcode) {
case 0: case register_reading_code:
ptr->register_reading(mgr); parent_->register_reading(mgr);
break;
case register_writing_code:
parent_->register_writing(mgr);
break; break;
case 1: case init_manager_code:
ptr->register_writing(mgr); parent_->init(mgr);
break; break;
case 4: case shutdown_code:
ptr->shutdown(); parent_->shutdown();
break; break;
default: default:
CAF_LOG_DEBUG("opcode not recognized: " << CAF_ARG(opcode)); CAF_LOG_ERROR("opcode not recognized: " << CAF_ARG(opcode));
break; break;
} }
} }
} } else if (num_bytes == 0) {
CAF_LOG_DEBUG("pipe closed, assume shutdown");
return false;
} else { } else {
return num_bytes < 0 && last_socket_error_is_temporary(); return last_socket_error_is_temporary();
} }
} }
} }
......
...@@ -19,14 +19,15 @@ ...@@ -19,14 +19,15 @@
#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 {
socket_manager::socket_manager(socket handle, const multiplexer_ptr& parent) socket_manager::socket_manager(socket handle, multiplexer* parent)
: handle_(handle), mask_(operation::none), parent_(parent) { : handle_(handle), mask_(operation::none), parent_(parent) {
CAF_ASSERT(parent != nullptr);
CAF_ASSERT(handle_ != invalid_socket); CAF_ASSERT(handle_ != invalid_socket);
CAF_ASSERT(parent != nullptr);
} }
socket_manager::~socket_manager() { socket_manager::~socket_manager() {
...@@ -54,17 +55,19 @@ bool socket_manager::mask_del(operation flag) noexcept { ...@@ -54,17 +55,19 @@ bool socket_manager::mask_del(operation flag) noexcept {
void socket_manager::register_reading() { void socket_manager::register_reading() {
if ((mask() & operation::read) == operation::read) if ((mask() & operation::read) == operation::read)
return; return;
auto ptr = parent_.lock(); parent_->register_reading(this);
if (ptr != nullptr)
ptr->register_reading(this);
} }
void socket_manager::register_writing() { void socket_manager::register_writing() {
if ((mask() & operation::write) == operation::write) if ((mask() & operation::write) == operation::write)
return; return;
auto ptr = parent_.lock(); parent_->register_writing(this);
if (ptr != nullptr) }
ptr->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 } // namespace caf::net
...@@ -169,11 +169,13 @@ error nodelay(stream_socket x, bool new_value) { ...@@ -169,11 +169,13 @@ error nodelay(stream_socket x, bool new_value) {
} }
ptrdiff_t read(stream_socket x, span<byte> buf) { ptrdiff_t read(stream_socket x, span<byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), buf.size(), return ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag); no_sigpipe_io_flag);
} }
ptrdiff_t write(stream_socket x, span<const byte> buf) { ptrdiff_t write(stream_socket x, span<const byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), buf.size(), return ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag); no_sigpipe_io_flag);
} }
......
...@@ -98,21 +98,23 @@ expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node, ...@@ -98,21 +98,23 @@ expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr) { bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(node)); CAF_LOG_TRACE(CAF_ARG(node));
auto addr = to_string(node.address()); auto addr = to_string(node.address());
auto make_acceptor = node.address().embeds_v4() bool is_v4 = node.address().embeds_v4();
? new_tcp_acceptor_impl<AF_INET> bool is_zero = is_v4 ? node.address().embedded_v4().bits() == 0
: node.address().zero();
auto make_acceptor = is_v4 ? new_tcp_acceptor_impl<AF_INET>
: new_tcp_acceptor_impl<AF_INET6>; : new_tcp_acceptor_impl<AF_INET6>;
auto p = make_acceptor(node.port(), addr.c_str(), reuse_addr, if (auto p = make_acceptor(node.port(), addr.c_str(), reuse_addr, is_zero)) {
node.address().zero());
if (!p) {
CAF_LOG_WARNING("could not create tcp socket for: " << to_string(node));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
to_string(node));
}
auto sock = socket_cast<tcp_accept_socket>(*p); auto sock = socket_cast<tcp_accept_socket>(*p);
auto sguard = make_socket_guard(sock); auto sguard = make_socket_guard(sock);
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(sock.id, SOMAXCONN)); CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(sock.id, SOMAXCONN));
CAF_LOG_DEBUG(CAF_ARG(sock.id)); CAF_LOG_DEBUG(CAF_ARG(sock.id));
return sguard.release(); return sguard.release();
} else {
CAF_LOG_WARNING("could not create tcp socket:" << CAF_ARG(node)
<< CAF_ARG(p.error()));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
to_string(node), std::move(p.error()));
}
} }
expected<tcp_accept_socket> expected<tcp_accept_socket>
......
...@@ -114,8 +114,8 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, ...@@ -114,8 +114,8 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) { ip_endpoint ep) {
sockaddr_storage addr = {}; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in) auto len = static_cast<socklen_t>(
: sizeof(sockaddr_in6); ep.address().embeds_v4() ? sizeof(sockaddr_in) : sizeof(sockaddr_in6));
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len); buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
auto ret = check_udp_datagram_socket_io_res(res); auto ret = check_udp_datagram_socket_io_res(res);
...@@ -172,7 +172,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs, ...@@ -172,7 +172,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in) message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6); : sizeof(sockaddr_in6);
message.msg_iov = buf_array; message.msg_iov = buf_array;
message.msg_iovlen = bufs.size(); message.msg_iovlen = static_cast<int>(bufs.size());
auto res = sendmsg(x.id, &message, 0); auto res = sendmsg(x.id, &message, 0);
return check_udp_datagram_socket_io_res(res); return check_udp_datagram_socket_io_res(res);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 detail.rfc6455
#include "caf/detail/rfc6455.hpp"
#include "caf/test/dsl.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
#include "caf/string_view.hpp"
#include <cstdint>
#include <initializer_list>
#include <vector>
using namespace caf;
namespace {
struct fixture {
using impl = detail::rfc6455;
auto bytes(std::initializer_list<uint8_t> xs) {
std::vector<byte> result;
for (auto x : xs)
result.emplace_back(static_cast<byte>(x));
return result;
}
template <class T>
auto take(const T& xs, size_t num_bytes) {
auto n = std::min(xs.size(), num_bytes);
return std::vector<typename T::value_type>{xs.begin(), xs.begin() + n};
}
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(rfc6455_tests, fixture)
CAF_TEST(masking) {
auto key = uint32_t{0xDEADC0DE};
auto data = bytes({0x12, 0x34, 0x45, 0x67, 0x89, 0x9A});
auto masked_data = data;
CAF_MESSAGE("masking XORs the repeated key to data");
impl::mask_data(key, masked_data);
CAF_CHECK_EQUAL(masked_data, bytes({
0x12 ^ 0xDE,
0x34 ^ 0xAD,
0x45 ^ 0xC0,
0x67 ^ 0xDE,
0x89 ^ 0xDE,
0x9A ^ 0xAD,
}));
CAF_MESSAGE("masking masked data again gives the original data");
impl::mask_data(key, masked_data);
CAF_CHECK_EQUAL(masked_data, data);
}
CAF_TEST(no mask key plus small data) {
std::vector<uint8_t> data{0x12, 0x34, 0x45, 0x67};
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(out, bytes({
0x82, // FIN + binary frame opcode
0x04, // data size = 4
0x12, 0x34, 0x45, 0x67, // masked data
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 2);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus small data) {
std::vector<uint8_t> data{0x12, 0x34, 0x45, 0x67};
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(out, bytes({
0x82, // FIN + binary frame opcode
0x84, // MASKED + data size = 4
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0x12, 0x34, 0x45, 0x67, // masked data
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 6);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(no mask key plus medium data) {
std::vector<uint8_t> data;
data.insert(data.end(), 126, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 8),
bytes({
0x82, // FIN + binary frame opcode
0x7E, // 126 -> uint16 size
0x00, 0x7E, // data size = 126
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 4);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus medium data) {
std::vector<uint8_t> data;
data.insert(data.end(), 126, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 12),
bytes({
0x82, // FIN + binary frame opcode
0xFE, // MASKED + 126 -> uint16 size
0x00, 0x7E, // data size = 126
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 8);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(no mask key plus large data) {
std::vector<uint8_t> data;
data.insert(data.end(), 65536, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 14),
bytes({
0x82, // FIN + binary frame opcode
0x7F, // 127 -> uint64 size
0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, // 65536
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 10);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus large data) {
std::vector<uint8_t> data;
data.insert(data.end(), 65536, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 18),
bytes({
0x82, // FIN + binary frame opcode
0xFF, // MASKED + 127 -> uint64 size
0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, // 65536
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 14);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp" #include "caf/byte_buffer.hpp"
#include "caf/deep_to_string.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -34,7 +35,7 @@ CAF_TEST(serialization) { ...@@ -34,7 +35,7 @@ CAF_TEST(serialization) {
byte_buffer buf; byte_buffer buf;
{ {
binary_serializer sink{nullptr, buf}; binary_serializer sink{nullptr, buf};
CAF_CHECK_EQUAL(sink(x), none); CAF_CHECK(sink.apply_object(x));
} }
CAF_CHECK_EQUAL(buf.size(), basp::header_size); CAF_CHECK_EQUAL(buf.size(), basp::header_size);
auto buf2 = to_bytes(x); auto buf2 = to_bytes(x);
...@@ -43,7 +44,7 @@ CAF_TEST(serialization) { ...@@ -43,7 +44,7 @@ CAF_TEST(serialization) {
basp::header y; basp::header y;
{ {
binary_deserializer source{nullptr, buf}; binary_deserializer source{nullptr, buf};
CAF_CHECK_EQUAL(source(y), none); CAF_CHECK(source.apply_object(y));
} }
CAF_CHECK_EQUAL(x, y); CAF_CHECK_EQUAL(x, y);
auto z = basp::header::from_bytes(buf); auto z = basp::header::from_bytes(buf);
...@@ -53,5 +54,5 @@ CAF_TEST(serialization) { ...@@ -53,5 +54,5 @@ CAF_TEST(serialization) {
CAF_TEST(to_string) { CAF_TEST(to_string) {
basp::header x{basp::message_type::handshake, 42, 4}; basp::header x{basp::message_type::handshake, 42, 4};
CAF_CHECK_EQUAL(to_string(x), "basp::header(handshake, 42, 4)"); CAF_CHECK_EQUAL(deep_to_string(x), "basp::header(handshake, 42, 4)");
} }
...@@ -40,18 +40,22 @@ namespace { ...@@ -40,18 +40,22 @@ namespace {
class dummy_manager : public socket_manager { class dummy_manager : public socket_manager {
public: public:
dummy_manager(size_t& manager_count, stream_socket handle, dummy_manager(size_t& manager_count, stream_socket handle,
multiplexer_ptr parent) multiplexer* parent)
: socket_manager(handle, std::move(parent)), : socket_manager(handle, parent), count_(manager_count) {
count_(manager_count), CAF_MESSAGE("created new dummy manager");
rd_buf_pos_(0) {
++count_; ++count_;
rd_buf_.resize(1024); rd_buf_.resize(1024);
} }
~dummy_manager() { ~dummy_manager() {
CAF_MESSAGE("destroyed dummy manager");
--count_; --count_;
} }
error init(const settings&) override {
return none;
}
stream_socket handle() const noexcept { stream_socket handle() const noexcept {
return socket_cast<stream_socket>(handle_); return socket_cast<stream_socket>(handle_);
} }
...@@ -110,7 +114,7 @@ private: ...@@ -110,7 +114,7 @@ private:
size_t& count_; size_t& count_;
size_t rd_buf_pos_; size_t rd_buf_pos_ = 0;
byte_buffer wr_buf_; byte_buffer wr_buf_;
...@@ -120,23 +124,22 @@ private: ...@@ -120,23 +124,22 @@ private:
using dummy_manager_ptr = intrusive_ptr<dummy_manager>; using dummy_manager_ptr = intrusive_ptr<dummy_manager>;
struct fixture : host_fixture { struct fixture : host_fixture {
fixture() : manager_count(0), mpx(std::make_shared<multiplexer>()) { fixture() : mpx(nullptr) {
mpx->set_thread_id(); mpx.set_thread_id();
} }
~fixture() { ~fixture() {
mpx.reset();
CAF_REQUIRE_EQUAL(manager_count, 0u); CAF_REQUIRE_EQUAL(manager_count, 0u);
} }
void exhaust() { void exhaust() {
while (mpx->poll_once(false)) while (mpx.poll_once(false))
; // Repeat. ; // Repeat.
} }
size_t manager_count; size_t manager_count = 0;
multiplexer_ptr mpx; multiplexer mpx;
}; };
} // namespace } // namespace
...@@ -144,32 +147,36 @@ struct fixture : host_fixture { ...@@ -144,32 +147,36 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(multiplexer_tests, fixture) CAF_TEST_FIXTURE_SCOPE(multiplexer_tests, fixture)
CAF_TEST(default construction) { CAF_TEST(default construction) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 0u);
} }
CAF_TEST(init) { CAF_TEST(init) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 0u);
CAF_REQUIRE_EQUAL(mpx->init(), none); CAF_REQUIRE_EQUAL(mpx.init(), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 1u);
mpx->close_pipe(); mpx.close_pipe();
exhaust(); exhaust();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 0u);
// Calling run must have no effect now. // Calling run must have no effect now.
mpx->run(); mpx.run();
} }
CAF_TEST(send and receive) { CAF_TEST(send and receive) {
CAF_REQUIRE_EQUAL(mpx->init(), none); CAF_REQUIRE_EQUAL(mpx.init(), none);
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
auto alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx); { // Lifetime scope of alice and bob.
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx); auto alice = make_counted<dummy_manager>(manager_count, sockets.first,
&mpx);
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, &mpx);
alice->register_reading(); alice->register_reading();
bob->register_reading(); bob->register_reading();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 3u);
alice->send("hello bob"); alice->send("hello bob");
alice->register_writing(); alice->register_writing();
exhaust(); exhaust();
CAF_CHECK_EQUAL(bob->receive(), "hello bob"); CAF_CHECK_EQUAL(bob->receive(), "hello bob");
}
mpx.shutdown();
} }
CAF_TEST(shutdown) { CAF_TEST(shutdown) {
...@@ -177,26 +184,27 @@ CAF_TEST(shutdown) { ...@@ -177,26 +184,27 @@ CAF_TEST(shutdown) {
std::condition_variable cv; std::condition_variable cv;
bool thread_id_set = false; bool thread_id_set = false;
auto run_mpx = [&] { auto run_mpx = [&] {
std::unique_lock<std::mutex> lk(m); {
mpx->set_thread_id(); std::unique_lock<std::mutex> guard(m);
mpx.set_thread_id();
thread_id_set = true; thread_id_set = true;
lk.unlock();
cv.notify_one(); cv.notify_one();
mpx->run(); }
mpx.run();
}; };
CAF_REQUIRE_EQUAL(mpx->init(), none); CAF_REQUIRE_EQUAL(mpx.init(), none);
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
auto alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx); auto alice = make_counted<dummy_manager>(manager_count, sockets.first, &mpx);
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx); auto bob = make_counted<dummy_manager>(manager_count, sockets.second, &mpx);
alice->register_reading(); alice->register_reading();
bob->register_reading(); bob->register_reading();
CAF_REQUIRE_EQUAL(mpx->num_socket_managers(), 3u); CAF_REQUIRE_EQUAL(mpx.num_socket_managers(), 3u);
std::thread mpx_thread{run_mpx}; std::thread mpx_thread{run_mpx};
std::unique_lock<std::mutex> lk(m); std::unique_lock<std::mutex> lk(m);
cv.wait(lk, [&] { return thread_id_set; }); cv.wait(lk, [&] { return thread_id_set; });
mpx->shutdown(); mpx.shutdown();
mpx_thread.join(); mpx_thread.join();
CAF_REQUIRE_EQUAL(mpx->num_socket_managers(), 0u); CAF_REQUIRE_EQUAL(mpx.num_socket_managers(), 0u);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
#pragma once
#include "caf/error.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/settings.hpp"
#include "caf/span.hpp"
#include "caf/string_view.hpp"
#include "caf/tag/stream_oriented.hpp"
#include "caf/test/dsl.hpp"
template <class UpperLayer>
class mock_stream_transport {
public:
// -- member types -----------------------------------------------------------
using output_tag = caf::tag::stream_oriented;
// -- interface for the upper layer ------------------------------------------
void begin_output() {
// nop
}
auto& output_buffer() {
return output;
}
constexpr void end_output() {
// nop
}
constexpr caf::net::socket handle() noexcept {
return caf::net::invalid_socket;
}
bool can_send_more() const noexcept {
return true;
}
const caf::error& abort_reason() {
return abort_reason_;
}
void abort_reason(caf::error reason) {
abort_reason_ = std::move(reason);
}
void configure_read(caf::net::receive_policy policy) {
min_read_size = policy.min_size;
max_read_size = policy.max_size;
}
// -- initialization ---------------------------------------------------------
caf::error init(const caf::settings& config) {
return upper_layer.init(nullptr, this, config);
}
caf::error init() {
caf::settings config;
return init(config);
}
// -- buffer management ------------------------------------------------------
void push(caf::span<const caf::byte> bytes) {
input.insert(input.begin(), bytes.begin(), bytes.end());
}
void push(caf::string_view str) {
push(caf::as_bytes(caf::make_span(str)));
}
size_t unconsumed() const noexcept {
return read_buf_.size();
}
caf::string_view output_as_str() const noexcept {
return {reinterpret_cast<const char*>(output.data()), output.size()};
}
// -- event callbacks --------------------------------------------------------
ptrdiff_t handle_input() {
ptrdiff_t result = 0;
while (max_read_size > 0) {
CAF_ASSERT(max_read_size > static_cast<size_t>(read_size_));
size_t len = max_read_size - static_cast<size_t>(read_size_);
CAF_LOG_DEBUG(CAF_ARG2("available capacity:", len));
auto num_bytes = std::min(input.size(), len);
if (num_bytes == 0)
return result;
auto delta_offset = static_cast<ptrdiff_t>(read_buf_.size());
read_buf_.insert(read_buf_.end(), input.begin(),
input.begin() + num_bytes);
input.erase(input.begin(), input.begin() + num_bytes);
read_size_ += static_cast<ptrdiff_t>(num_bytes);
if (static_cast<size_t>(read_size_) < min_read_size)
return result;
auto delta = make_span(read_buf_.data() + delta_offset,
read_size_ - delta_offset);
auto consumed = upper_layer.consume(this, caf::make_span(read_buf_),
delta);
if (consumed > 0) {
result += static_cast<ptrdiff_t>(consumed);
read_buf_.erase(read_buf_.begin(), read_buf_.begin() + consumed);
read_size_ -= consumed;
} else if (consumed < 0) {
if (!abort_reason_)
abort_reason_ = caf::sec::runtime_error;
upper_layer.abort(this, abort_reason_);
return -1;
}
}
return result;
}
// -- member variables -------------------------------------------------------
UpperLayer upper_layer;
std::vector<caf::byte> output;
std::vector<caf::byte> input;
uint32_t min_read_size = 0;
uint32_t max_read_size = 0;
private:
std::vector<caf::byte> read_buf_;
ptrdiff_t read_size_ = 0;
caf::error abort_reason_;
};
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/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"
using namespace caf;
using namespace std::string_literals;
namespace {
using svec = std::vector<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(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::actor_shell_ptr 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()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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.length_prefix_framing
#include "caf/net/length_prefix_framing.hpp"
#include "caf/test/dsl.hpp"
#include <deque>
#include <numeric>
#include <vector>
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
#include "caf/tag/message_oriented.hpp"
using namespace caf;
namespace {
/// upper layer: expect messages
/// Needs to be initilized by the layer two steps down.
struct ul_expect_messages {
using input_tag = tag::message_oriented;
void set_expected_messages(std::vector<byte_buffer> messages) {
expected_messages.clear();
for (auto& msg : messages)
expected_messages.emplace_back(std::move(msg));
}
void add_expected_messages(std::vector<byte_buffer> messages) {
for (auto& msg : messages)
expected_messages.emplace_back(std::move(msg));
}
template <class LowerLayer>
ptrdiff_t consume(LowerLayer&, byte_span buffer) {
CAF_REQUIRE(expected_messages.size() > 0);
auto& next = expected_messages.front();
CAF_CHECK_EQUAL(next.size(), buffer.size());
CAF_CHECK(std::equal(next.begin(), next.end(), buffer.begin()));
expected_messages.pop_front();
return buffer.size();
}
std::deque<byte_buffer> expected_messages;
};
/// lower layer: offer stream for message parsing
template <class UpperLayer>
struct ll_provide_stream_for_messages {
using output_tag = tag::stream_oriented;
void set_expectations(std::vector<byte> data,
std::vector<byte_buffer> messages) {
data_stream = std::move(data);
auto& checking_layer = upper_layer.upper_layer();
checking_layer.set_expected_messages(messages);
}
void add_expectations(const std::vector<byte>& data,
std::vector<byte_buffer> messages) {
data_stream.insert(data_stream.end(), data.begin(), data.end());
auto& checking_layer = upper_layer.upper_layer();
checking_layer.add_expect_messages(messages);
}
void run() {
CAF_CHECK(data_stream.size() != 0);
while (processed != data_stream.size()) {
auto all_data = make_span(data_stream.data() + processed,
data_stream.size() - processed);
auto new_data = make_span(data_stream.data() + offered,
data_stream.size() - offered);
auto newly_offered = new_data.size();
auto consumed = upper_layer.consume(*this, all_data, new_data);
CAF_CHECK(consumed >= 0);
CAF_CHECK(static_cast<size_t>(consumed) <= data_stream.size());
offered += newly_offered;
processed += consumed;
if (consumed > 0) {
data_stream.erase(data_stream.begin(), data_stream.begin() + consumed);
offered -= processed;
processed = 0;
}
if (consumed == 0 || data_stream.size() == 0)
return;
}
}
size_t processed = 0;
size_t offered = 0;
std::vector<byte> data_stream;
UpperLayer upper_layer;
};
template <class... Ts>
byte_buffer to_buf(const Ts&... xs) {
byte_buffer buf;
binary_serializer sink{nullptr, buf};
if (!sink.apply_objects(xs...))
CAF_FAIL("to_buf failed: " << sink.get_error());
return buf;
}
void encode_message(std::vector<byte>& data, const byte_buffer& message) {
using detail::to_network_order;
auto current_size = data.size();
data.insert(data.end(), 4, byte{0});
auto msg_begin = data.begin() + current_size;
auto msg_size = message.size();
auto u32_size = to_network_order(static_cast<uint32_t>(msg_size));
memcpy(std::addressof(*msg_begin), &u32_size, 4);
data.insert(data.end(), message.begin(), message.end());
}
struct fixture {
using test_layers = ll_provide_stream_for_messages<
net::length_prefix_framing<ul_expect_messages>>;
void generate_messages(size_t num, size_t factor = 10) {
for (size_t n = 1; n <= num; n += 1) {
std::vector<int> buf(n * factor);
std::iota(buf.begin(), buf.end(), n);
messages.emplace_back(to_buf(buf));
}
for (auto& msg : messages)
encode_message(data, msg);
}
void set_expectations() {
layers.set_expectations(data, messages);
}
void test_receive_data() {
layers.run();
}
void clear() {
data.clear();
messages.clear();
}
test_layers layers;
std::vector<byte> data;
std::vector<byte_buffer> messages;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(length_prefix_framing_tests, fixture)
CAF_TEST(process messages) {
// Single message.
generate_messages(1);
set_expectations();
test_receive_data();
clear();
// Multiple messages.
generate_messages(10);
set_expectations();
test_receive_data();
}
CAF_TEST(incomplete message) {
generate_messages(1, 1000);
CAF_MESSAGE("data.size() = " << data.size());
auto initial_size = data.size();
auto data_copy = data;
auto mid = data.size() / 2;
data.resize(mid);
CAF_MESSAGE("data.size() = " << data.size());
data_copy.erase(data_copy.begin(), data_copy.begin() + mid);
CAF_MESSAGE("data_copy.size() = " << data_copy.size());
CAF_REQUIRE(data.size() + data_copy.size() == initial_size);
// Don't set expectations because there shouldn't be a complete message
// in the bytes.
auto messages_copy = messages;
messages.clear();
CAF_REQUIRE(messages.empty());
set_expectations();
test_receive_data();
data.insert(data.end(), data_copy.begin(), data_copy.end());
messages = messages_copy;
set_expectations();
test_receive_data();
}
CAF_TEST_FIXTURE_SCOPE_END()
This diff is collapsed.
...@@ -61,7 +61,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -61,7 +61,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
mm.mpx()->set_thread_id(); mm.mpx()->set_thread_id();
auto backend = dynamic_cast<backend::test*>(mm.backend("test")); auto backend = dynamic_cast<backend::test*>(mm.backend("test"));
auto mgr = backend->peer(mars); auto mgr = backend->peer(mars);
auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr); auto& dref = dynamic_cast<socket_manager_impl<transport_type>&>(*mgr);
app = &dref.transport().application(); app = &dref.transport().application();
sock = backend->socket(mars); sock = backend->socket(mars);
} }
......
This diff is collapsed.
...@@ -65,9 +65,8 @@ struct string_application_header { ...@@ -65,9 +65,8 @@ struct string_application_header {
/// @relates header /// @relates header
template <class Inspector> template <class Inspector>
typename Inspector::result_type bool inspect(Inspector& f, string_application_header& x) {
inspect(Inspector& f, string_application_header& hdr) { return f.fields(f.field("payload", x.payload));
return f(meta::type_name("sa_header"), hdr.payload);
} }
class string_application { class string_application {
......
This diff is collapsed.
This diff is collapsed.
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