Commit 85bee49f authored by Dominik Charousset's avatar Dominik Charousset

Drop net::operation and obsolete header

parent e013cfd0
......@@ -19,7 +19,6 @@ caf_add_component(
ENUM_TYPES
net.http.method
net.http.status
net.operation
net.ssl.dtls
net.ssl.errc
net.ssl.format
......@@ -103,7 +102,6 @@ caf_add_component(
net.length_prefix_framing
net.multiplexer
net.network_socket
net.operation
net.pipe_socket
net.prometheus.server
net.socket
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/disposable.hpp"
#include "caf/net/binary/default_trait.hpp"
#include "caf/net/binary/flow_bridge.hpp"
#include "caf/net/binary/frame.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_oriented.hpp"
#include <cstdint>
#include <cstring>
#include <memory>
#include <type_traits>
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.
class CAF_NET_EXPORT length_prefix_framing
: public stream_oriented::upper_layer,
public binary::lower_layer {
public:
// -- member types -----------------------------------------------------------
using upper_layer_ptr = std::unique_ptr<binary::upper_layer>;
// -- constants --------------------------------------------------------------
static constexpr size_t hdr_size = sizeof(uint32_t);
static constexpr size_t max_message_length = INT32_MAX - sizeof(uint32_t);
// -- constructors, destructors, and assignment operators --------------------
explicit length_prefix_framing(upper_layer_ptr up);
// -- factories --------------------------------------------------------------
static std::unique_ptr<length_prefix_framing> make(upper_layer_ptr up);
// -- high-level factory functions -------------------------------------------
/// Binds a trait class to the framing protocol to enable a high-level API for
/// operating on flows.
template <class Trait>
struct bind {
/// Describes the one-time connection event.
using connect_event_t
= cow_tuple<async::consumer_resource<typename Trait::input_type>,
async::producer_resource<typename Trait::output_type>>;
/// Runs length-prefix framing on given connection.
/// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on
/// the
/// `Transport`.
/// @param pull Source for pulling data to send.
/// @param push Source for pushing received data.
template <class Connection>
static disposable
run(actor_system& sys, Connection conn,
async::consumer_resource<typename Trait::output_type> pull,
async::producer_resource<typename Trait::input_type> push) {
using transport_t = typename Connection::transport_type;
auto mpx = sys.network_manager().mpx_ptr();
auto fc = flow_connector<Trait>::make_trivial(std::move(pull),
std::move(push));
auto bridge = binary::flow_bridge<Trait>::make(mpx, std::move(fc));
auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge));
auto transport = transport_t::make(std::move(conn), std::move(impl));
auto ptr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable());
mpx->start(ptr);
return disposable{std::move(ptr)};
}
/// Runs length-prefix framing on given connection.
/// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on
/// the
/// `Transport`.
/// @param init Function object for setting up the created flows.
template <class Connection, class Init>
static disposable run(actor_system& sys, Connection conn, Init init) {
using app_in_t = typename Trait::input_type;
using app_out_t = typename Trait::output_type;
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
auto [app_pull, fd_push] = async::make_spsc_buffer_resource<app_in_t>();
auto [fd_pull, app_push] = async::make_spsc_buffer_resource<app_out_t>();
auto result = run(sys, std::move(conn), std::move(fd_pull),
std::move(fd_push));
init(connect_event_t{std::move(app_pull), std::move(app_push)});
return result;
}
/// The default number of concurrently open connections when using `accept`.
static constexpr size_t default_max_connections = 128;
/// A producer resource for the acceptor. Any accepted connection is
/// represented by two buffers: one for input and one for output.
using acceptor_resource_t = async::producer_resource<connect_event_t>;
/// Describes the per-connection event.
using accept_event_t
= cow_tuple<async::consumer_resource<typename Trait::input_type>,
async::producer_resource<typename Trait::output_type>>;
/// Convenience function for creating an event listener resource and an
/// @ref acceptor_resource_t via @ref async::make_spsc_buffer_resource.
static auto make_accept_event_resources() {
return async::make_spsc_buffer_resource<accept_event_t>();
}
/// Listens for incoming connection on @p fd.
/// @param sys The host system.
/// @param acc A connection acceptor such as @ref tcp_accept_socket or
/// @ref ssl::tcp_acceptor.
/// @param cfg Configures the acceptor. Currently, the only supported
/// configuration parameter is `max-connections`.
template <class Acceptor>
static disposable accept(actor_system& sys, Acceptor acc,
acceptor_resource_t out,
const settings& cfg = {}) {
using transport_t = typename Acceptor::transport_type;
using trait_t = binary::default_trait;
using factory_t = cf_impl<transport_t>;
using impl_t = detail::accept_handler<Acceptor>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto conn = flow_connector<trait_t>::make_basic_server(std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(conn));
auto impl = impl_t::make(std::move(acc), std::move(factory),
max_connections);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(&mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
mpx.start(ptr);
return disposable{std::move(ptr)};
} else {
return {};
}
}
};
// -- implementation of stream_oriented::upper_layer -------------------------
error start(stream_oriented::lower_layer* down) override;
void abort(const error& reason) override;
ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void prepare_send() override;
bool done_sending() override;
// -- implementation of binary::lower_layer ----------------------------------
bool can_send_more() const noexcept override;
void request_messages() override;
void suspend_reading() override;
bool is_reading() const noexcept override;
void write_later() override;
void begin_message() override;
byte_buffer& message_buffer() override;
bool end_message() override;
void shutdown() override;
// -- utility functions ------------------------------------------------------
static std::pair<size_t, byte_span> split(byte_span buffer) noexcept;
private:
// -- helper classes ---------------------------------------------------------
template <class Transport>
class cf_impl
: public detail::connection_factory<typename Transport::connection_handle> {
public:
using connection_handle = typename Transport::connection_handle;
using connector_ptr = flow_connector_ptr<binary::default_trait>;
explicit cf_impl(connector_ptr connector) : conn_(std::move(connector)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
using trait_t = binary::default_trait;
auto bridge = binary::flow_bridge<trait_t>::make(mpx, conn_);
auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge));
auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(impl));
transport->active_policy().accept(fd);
auto mgr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(mgr->as_disposable());
return mgr;
}
private:
connector_ptr conn_;
};
// -- member variables -------------------------------------------------------
stream_oriented::lower_layer* down_;
upper_layer_ptr up_;
size_t message_offset_ = 0;
};
} // namespace caf::net
......@@ -9,7 +9,6 @@
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
......@@ -96,9 +95,6 @@ public:
/// Returns the enclosing @ref actor_system.
actor_system& system();
/// Computes the current mask for the manager. Mostly useful for testing.
operation mask_of(const socket_manager_ptr& mgr);
// -- implementation of execution_context ------------------------------------
void ref_execution_context() const noexcept override;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/is_error_code_enum.hpp"
#include <cstdint>
#include <string>
#include <string_view>
#include <type_traits>
namespace caf::net {
/// Values for representing bitmask of I/O operations.
enum class operation {
none = 0b0000,
read = 0b0001,
write = 0b0010,
block_read = 0b0100,
block_write = 0b1000,
read_write = 0b0011,
read_only = 0b1001,
write_only = 0b0110,
shutdown = 0b1100,
};
/// @relates operation
[[nodiscard]] constexpr int to_integer(operation x) noexcept {
return static_cast<int>(x);
}
/// Adds the `read` flag to `x` unless the `block_read` bit is set.
/// @relates operation
[[nodiscard]] constexpr operation add_read_flag(operation x) noexcept {
if (auto bits = to_integer(x); !(bits & 0b0100))
return static_cast<operation>(bits | 0b0001);
else
return x;
}
/// Adds the `write` flag to `x` unless the `block_write` bit is set.
/// @relates operation
[[nodiscard]] constexpr operation add_write_flag(operation x) noexcept {
if (auto bits = to_integer(x); !(bits & 0b1000))
return static_cast<operation>(bits | 0b0010);
else
return x;
}
/// Removes the `read` flag from `x`.
/// @relates operation
[[nodiscard]] constexpr operation remove_read_flag(operation x) noexcept {
return static_cast<operation>(to_integer(x) & 0b1110);
}
/// Removes the `write` flag from `x`.
/// @relates operation
[[nodiscard]] constexpr operation remove_write_flag(operation x) noexcept {
return static_cast<operation>(to_integer(x) & 0b1101);
}
/// Adds the `block_read` flag to `x` and removes the `read` flag if present.
/// @relates operation
[[nodiscard]] constexpr operation block_reads(operation x) noexcept {
auto bits = to_integer(x) | 0b0100;
return static_cast<operation>(bits & 0b1110);
}
/// Adds the `block_write` flag to `x` and removes the `write` flag if present.
/// @relates operation
[[nodiscard]] constexpr operation block_writes(operation x) noexcept {
auto bits = to_integer(x) | 0b1000;
return static_cast<operation>(bits & 0b1101);
}
/// Returns whether the read flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_reading(operation x) noexcept {
return (to_integer(x) & 0b0001) == 0b0001;
}
/// Returns whether the write flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_writing(operation x) noexcept {
return (to_integer(x) & 0b0010) == 0b0010;
}
/// Returns `!is_reading(x) && !is_writing(x)`.
/// @relates operation
[[nodiscard]] constexpr bool is_idle(operation x) noexcept {
return (to_integer(x) & 0b0011) == 0b0000;
}
/// Returns whether the block read flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_read_blocked(operation x) noexcept {
return (to_integer(x) & 0b0100) == 0b0100;
}
/// Returns whether the block write flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_write_blocked(operation x) noexcept {
return (to_integer(x) & 0b1000) == 0b1000;
}
/// @relates operation
CAF_NET_EXPORT std::string to_string(operation x);
/// @relates operation
CAF_NET_EXPORT bool from_string(std::string_view, operation&);
/// @relates operation
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<operation>, operation&);
/// @relates operation
template <class Inspector>
bool inspect(Inspector& f, operation& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net
......@@ -13,7 +13,6 @@
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
......@@ -53,22 +52,6 @@ const short error_mask = POLLRDHUP | POLLERR | POLLHUP | POLLNVAL;
const short output_mask = POLLOUT;
// short to_bitmask(operation mask) {
// return static_cast<short>((is_reading(mask) ? input_mask : 0)
// | (is_writing(mask) ? output_mask : 0));
// }
operation to_operation(const socket_manager_ptr&, std::optional<short> mask) {
operation res = operation::none;
if (mask) {
if ((*mask & input_mask) != 0)
res = add_read_flag(res);
if ((*mask & output_mask) != 0)
res = add_write_flag(res);
}
return res;
}
} // namespace
// -- static utility functions -------------------------------------------------
......@@ -154,16 +137,6 @@ actor_system& multiplexer::system() {
return owner().system();
}
operation multiplexer::mask_of(const socket_manager_ptr& mgr) {
auto fd = mgr->handle();
if (auto i = updates_.find(fd); i != updates_.end())
return to_operation(mgr, i->second.events);
else if (auto index = index_of(mgr); index != -1)
return to_operation(mgr, pollset_[index].events);
else
return to_operation(mgr, std::nullopt);
}
// -- implementation of execution_context --------------------------------------
void multiplexer::ref_execution_context() const noexcept {
......
......@@ -200,11 +200,11 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
CHECK_EQ(mgr->start(), none);
mpx->apply_updates();
REQUIRE_EQ(mpx->num_socket_managers(), 2u);
CHECK_EQ(mpx->mask_of(mgr), net::operation::read);
CHECK(mpx->is_reading(mgr.get()));
WHEN("the app calls suspend_reading") {
while (mpx->num_socket_managers() > 1u)
mpx->poll_once(true);
CHECK_EQ(mpx->mask_of(mgr), net::operation::none);
CHECK(!mpx->is_reading(mgr.get()));
if (CHECK_EQ(buf->size(), 3u)) {
CHECK_EQ(buf->at(0), "first");
CHECK_EQ(buf->at(1), "second");
......
#define CAF_SUITE net.operation
#include "caf/net/operation.hpp"
#include "net-test.hpp"
using namespace caf;
using namespace caf::net;
SCENARIO("add_read_flag adds the read bit unless block_read prevents it") {
GIVEN("a valid operation enum value") {
WHEN("calling add_read_flag") {
THEN("the read bit is set unless the block_read flag is present") {
auto add_rd = [](auto x) { return add_read_flag(x); };
CHECK_EQ(add_rd(operation::none), operation::read);
CHECK_EQ(add_rd(operation::read), operation::read);
CHECK_EQ(add_rd(operation::write), operation::read_write);
CHECK_EQ(add_rd(operation::block_read), operation::block_read);
CHECK_EQ(add_rd(operation::block_write), operation::read_only);
CHECK_EQ(add_rd(operation::read_write), operation::read_write);
CHECK_EQ(add_rd(operation::read_only), operation::read_only);
CHECK_EQ(add_rd(operation::write_only), operation::write_only);
CHECK_EQ(add_rd(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("add_write_flag adds the write bit unless block_write prevents it") {
GIVEN("a valid operation enum value") {
WHEN("calling add_write_flag") {
THEN("the write bit is set unless the block_write flag is present") {
auto add_wr = [](auto x) { return add_write_flag(x); };
CHECK_EQ(add_wr(operation::none), operation::write);
CHECK_EQ(add_wr(operation::read), operation::read_write);
CHECK_EQ(add_wr(operation::write), operation::write);
CHECK_EQ(add_wr(operation::block_read), operation::write_only);
CHECK_EQ(add_wr(operation::block_write), operation::block_write);
CHECK_EQ(add_wr(operation::read_write), operation::read_write);
CHECK_EQ(add_wr(operation::read_only), operation::read_only);
CHECK_EQ(add_wr(operation::write_only), operation::write_only);
CHECK_EQ(add_wr(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("remove_read_flag erases the read flag") {
GIVEN("a valid operation enum value") {
WHEN("calling remove_read_flag") {
THEN("the read bit is removed if present") {
auto del_rd = [](auto x) { return remove_read_flag(x); };
CHECK_EQ(del_rd(operation::none), operation::none);
CHECK_EQ(del_rd(operation::read), operation::none);
CHECK_EQ(del_rd(operation::write), operation::write);
CHECK_EQ(del_rd(operation::block_read), operation::block_read);
CHECK_EQ(del_rd(operation::block_write), operation::block_write);
CHECK_EQ(del_rd(operation::read_write), operation::write);
CHECK_EQ(del_rd(operation::read_only), operation::block_write);
CHECK_EQ(del_rd(operation::write_only), operation::write_only);
CHECK_EQ(del_rd(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("remove_write_flag erases the write flag") {
GIVEN("a valid operation enum value") {
WHEN("calling remove_write_flag") {
THEN("the write bit is removed if present") {
auto del_wr = [](auto x) { return remove_write_flag(x); };
CHECK_EQ(del_wr(operation::none), operation::none);
CHECK_EQ(del_wr(operation::read), operation::read);
CHECK_EQ(del_wr(operation::write), operation::none);
CHECK_EQ(del_wr(operation::block_read), operation::block_read);
CHECK_EQ(del_wr(operation::block_write), operation::block_write);
CHECK_EQ(del_wr(operation::read_write), operation::read);
CHECK_EQ(del_wr(operation::read_only), operation::read_only);
CHECK_EQ(del_wr(operation::write_only), operation::block_read);
CHECK_EQ(del_wr(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("block_reads removes the read flag and sets the block read flag") {
GIVEN("a valid operation enum value") {
WHEN("calling block_reads") {
THEN("the read bit is removed if present and block_read is set") {
auto block_rd = [](auto x) { return block_reads(x); };
CHECK_EQ(block_rd(operation::none), operation::block_read);
CHECK_EQ(block_rd(operation::read), operation::block_read);
CHECK_EQ(block_rd(operation::write), operation::write_only);
CHECK_EQ(block_rd(operation::block_read), operation::block_read);
CHECK_EQ(block_rd(operation::block_write), operation::shutdown);
CHECK_EQ(block_rd(operation::read_write), operation::write_only);
CHECK_EQ(block_rd(operation::read_only), operation::shutdown);
CHECK_EQ(block_rd(operation::write_only), operation::write_only);
CHECK_EQ(block_rd(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("block_writes removes the write flag and sets the block write flag") {
GIVEN("a valid operation enum value") {
WHEN("calling block_writes") {
THEN("the write bit is removed if present and block_write is set") {
auto block_wr = [](auto x) { return block_writes(x); };
CHECK_EQ(block_wr(operation::none), operation::block_write);
CHECK_EQ(block_wr(operation::read), operation::read_only);
CHECK_EQ(block_wr(operation::write), operation::block_write);
CHECK_EQ(block_wr(operation::block_read), operation::shutdown);
CHECK_EQ(block_wr(operation::block_write), operation::block_write);
CHECK_EQ(block_wr(operation::read_write), operation::read_only);
CHECK_EQ(block_wr(operation::read_only), operation::read_only);
CHECK_EQ(block_wr(operation::write_only), operation::shutdown);
CHECK_EQ(block_wr(operation::shutdown), operation::shutdown);
}
}
}
}
SCENARIO("is-predicates check whether certain flags are present") {
GIVEN("a valid operation enum value") {
WHEN("calling is_reading") {
THEN("the predicate returns true if the read bit is present") {
CHECK_EQ(is_reading(operation::none), false);
CHECK_EQ(is_reading(operation::read), true);
CHECK_EQ(is_reading(operation::write), false);
CHECK_EQ(is_reading(operation::block_read), false);
CHECK_EQ(is_reading(operation::block_write), false);
CHECK_EQ(is_reading(operation::read_write), true);
CHECK_EQ(is_reading(operation::read_only), true);
CHECK_EQ(is_reading(operation::write_only), false);
CHECK_EQ(is_reading(operation::shutdown), false);
}
}
WHEN("calling is_writing") {
THEN("the predicate returns true if the write bit is present") {
CHECK_EQ(is_writing(operation::none), false);
CHECK_EQ(is_writing(operation::read), false);
CHECK_EQ(is_writing(operation::write), true);
CHECK_EQ(is_writing(operation::block_read), false);
CHECK_EQ(is_writing(operation::block_write), false);
CHECK_EQ(is_writing(operation::read_write), true);
CHECK_EQ(is_writing(operation::read_only), false);
CHECK_EQ(is_writing(operation::write_only), true);
CHECK_EQ(is_writing(operation::shutdown), false);
}
}
WHEN("calling is_idle") {
THEN("the predicate returns true when neither reading nor writing") {
CHECK_EQ(is_idle(operation::none), true);
CHECK_EQ(is_idle(operation::read), false);
CHECK_EQ(is_idle(operation::write), false);
CHECK_EQ(is_idle(operation::block_read), true);
CHECK_EQ(is_idle(operation::block_write), true);
CHECK_EQ(is_idle(operation::read_write), false);
CHECK_EQ(is_idle(operation::read_only), false);
CHECK_EQ(is_idle(operation::write_only), false);
CHECK_EQ(is_idle(operation::shutdown), true);
}
}
WHEN("calling is_read_blocked") {
THEN("the predicate returns when blocking reads") {
CHECK_EQ(is_read_blocked(operation::none), false);
CHECK_EQ(is_read_blocked(operation::read), false);
CHECK_EQ(is_read_blocked(operation::write), false);
CHECK_EQ(is_read_blocked(operation::block_read), true);
CHECK_EQ(is_read_blocked(operation::block_write), false);
CHECK_EQ(is_read_blocked(operation::read_write), false);
CHECK_EQ(is_read_blocked(operation::read_only), false);
CHECK_EQ(is_read_blocked(operation::write_only), true);
CHECK_EQ(is_read_blocked(operation::shutdown), true);
}
}
WHEN("calling is_write_blocked") {
THEN("the predicate returns when blocking writes") {
CHECK_EQ(is_write_blocked(operation::none), false);
CHECK_EQ(is_write_blocked(operation::read), false);
CHECK_EQ(is_write_blocked(operation::write), false);
CHECK_EQ(is_write_blocked(operation::block_read), false);
CHECK_EQ(is_write_blocked(operation::block_write), true);
CHECK_EQ(is_write_blocked(operation::read_write), false);
CHECK_EQ(is_write_blocked(operation::read_only), true);
CHECK_EQ(is_write_blocked(operation::write_only), false);
CHECK_EQ(is_write_blocked(operation::shutdown), true);
}
}
}
}
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