Commit 48cb551e authored by Dominik Charousset's avatar Dominik Charousset

WIP

parent 498c837b
......@@ -36,6 +36,7 @@ Convenience options:
--build-type=Debug
--sanitizers=address,undefined
--enable-utility-targets
--enable-export-compile-commands
Flags (use --enable-<name> to activate and --disable-<name> to deactivate):
......@@ -143,6 +144,7 @@ while [ $# -ne 0 ]; do
CMakeBuildType='Debug'
append_cache_entry CAF_INC_SANITIZERS STRING 'address,undefined'
set_build_flag utility-targets ON
set_build_flag export-compile-commands ON
;;
--enable-*)
set_build_flag $optarg ON
......
......@@ -123,30 +123,31 @@ target_include_directories(caf-net-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/tes
target_link_libraries(caf-net-test PRIVATE CAF::test)
caf_incubator_add_test_suites(caf-net-test
accept_socket
application
convert_ip_endpoint
datagram_socket
datagram_transport
doorman
endpoint_manager
header
ip
multiplexer
net.backend.tcp
net.basp.message_queue
net.basp.ping_pong
net.basp.worker
accept_socket
net.length_prefix_framing
network_socket
pipe_socket
application
socket
convert_ip_endpoint
socket_guard
datagram_socket
stream_application
datagram_transport
stream_socket
doorman
stream_transport
endpoint_manager
string_application
header
tcp_sockets
ip
transport_worker
multiplexer
transport_worker_dispatcher
udp_datagram_socket
network_socket
net.backend.tcp
)
......@@ -48,24 +48,52 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
..code-block:: C++
interface base [role: upper layer] {
/// Called whenever the underlying transport is ready to send, allowing the
/// upper layers to produce additional output data or use the event to read
/// from event queues, etc.
/// @returns `true` if the lower layers may proceed, `false` otherwise
/// (aborts execution).
template <class LowerLayer>
bool prepare_send(LowerLayer& down);
/// Called whenever the underlying transport finished writing all buffered
/// data for output to query whether an upper layer still has pending events
/// or may produce output data on the next call to `prepare_send`.
/// @returns `true` if the underlying socket may get removed from the I/O
/// event loop, `false` otherwise.
template <class LowerLayer>
bool done_sending(LowerLayer& down);
}
interface base [role: lower layer] {
/// Returns whether the layer has output slots available.
bool can_send_more() const noexcept;
}
interface stream_oriented [role: upper layer] {
/// Called by the lower layer for cleaning up any state in case of an error.
template <class LowerLayer>
void abort(LowerLayer& down, const error& reason);
/// Consumes bytes from the lower layer.
/// @param down Reference to the lower layer that received the data.
/// @param buffer Available bytes to read.
/// @param delta Bytes that arrived since last calling this function.
/// @returns The number of consumed bytes (may be zero if waiting for more
/// input or negative to signal an error) and a policy that
/// configures how many bytes to receive next (as well as
/// thresholds for when to call this function again).
/// @note When returning a negative value for the number of consumed bytes,
/// clients must also call `down.set_read_error(...)` with an
/// appropriate error code.
/// @returns The number of consumed bytes. May be zero if waiting for more
/// input or negative to signal an error.
/// @note When returning a negative value, clients should also call
/// `down.abort_reason(...)` with an appropriate error code.
template <class LowerLayer>
pair<ptrdiff_t, receive_policy> consume(LowerLayer& down, byte_span buffer,
byte_span delta);
ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span delta);
}
interface stream_oriented [role: lower layer] {
/// Configures threshold for the next receive operations. Policies remain
/// active until calling this function again.
/// @warning Calling this function in `consume` invalidates both byte spans.
void configure_read(read_policy policy);
/// Prepares the layer for outgoing traffic, e.g., by allocating an output
/// buffer as necessary.
void begin_output();
......@@ -78,6 +106,11 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
/// Prepares written data for transfer, e.g., by flushing buffers or
/// registering sockets for write events.
void end_output();
/// Propagates an abort reason to the lower layers. After processing the
/// current read or write event, the lowest layer will call `abort` on its
// upper layer.
void abort_reason(error reason);
}
interface datagram_oriented [role: upper layer] {
......@@ -137,6 +170,9 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
byte_buffer& message_buffer();
/// Seals and prepares a message for transfer.
void end_message();
/// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code.
template <class LowerLayer>
bool end_message();
}
......@@ -23,7 +23,6 @@
#include "caf/net/datagram_transport.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/host.hpp"
......
......@@ -36,4 +36,9 @@ CAF_NET_EXPORT extern const size_t max_header_buffers;
/// Port to listen on for tcp.
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
......@@ -52,10 +52,16 @@ public:
// -- properties -------------------------------------------------------------
actor_system& system() {
actor_system& system() noexcept {
return sys_;
}
const actor_system_config& config() const noexcept;
// -- queue access -----------------------------------------------------------
bool at_end_of_message_queue();
endpoint_manager_queue::message_ptr next_message();
// -- event management -------------------------------------------------------
......
......@@ -124,6 +124,8 @@ private:
/// Stores the id for the next timeout.
uint64_t next_timeout_id_;
error err_;
};
} // 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 <cstdint>
#include <cstring>
#include <memory>
#include "caf/byte.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/sec.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;
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.size();
}
void 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) {
if (policy.max_size > 0 && transport_->max_read_size_ == 0)
parent_->register_reading();
transport_->min_read_size_ = policy.min_size;
transport_->max_read_size_ = policy.max_size;
transport_->read_buf_.resize(policy.max_size);
}
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 delta) {
}
private:
UpperLayer upper_layer_;
};
} // namespace caf::net
......@@ -18,45 +18,28 @@
#pragma once
#include <cstddef>
#include <string>
#include <utility>
#include "caf/config.hpp"
#include <cstdint>
namespace caf::net {
enum class CAF_NET_EXPORT receive_policy_flag : unsigned {
at_least,
at_most,
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 {
public:
receive_policy() = delete;
using config = std::pair<receive_policy_flag, size_t>;
struct receive_policy {
uint32_t min_size;
uint32_t max_size;
static config at_least(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::at_least, num_bytes};
/// @pre `min_size > 0`
/// @pre `min_size <= max_size`
static constexpr receive_policy between(uint32_t min_size,
uint32_t max_size) {
return {min_size, max_size};
}
static config at_most(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::at_most, num_bytes};
/// @pre `size > 0`
static constexpr receive_policy exactly(uint32_t size) {
return {size, size};
}
static config exactly(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::exactly, num_bytes};
static constexpr receive_policy stop() {
return {0, 0};
}
};
......
......@@ -21,6 +21,7 @@
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/make_counted.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/socket.hpp"
......@@ -70,6 +71,21 @@ public:
/// @pre `flag != operation::none`
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_;
}
// -- event loop management --------------------------------------------------
void register_reading();
......@@ -96,8 +112,65 @@ protected:
operation mask_;
weak_multiplexer_ptr parent_;
error abort_reason_;
};
template <class Protocol>
class socket_manager_impl : public socket_manager {
public:
template <class... Ts>
socket_manager_impl(Ts&&... xs) : protocol_(std::forward<Ts>(xs)...) {
// nop
}
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_;
}
private:
Protocol protocol_;
};
/// @relates 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...> {
// no content
};
template <class App, template <class> class... Layers, class... Ts>
auto make_socket_manager(Ts&&... xs) {
using impl = make_socket_manager_helper<App, Layers..., socket_manager_impl>;
return make_counted<impl>(std::forward<Ts>(xs)...);
}
} // namespace caf::net
......@@ -21,81 +21,172 @@
#include <deque>
#include "caf/byte_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/settings.hpp"
#include "caf/span.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace caf::net {
template <class Application>
using stream_transport_base
= transport_base<stream_transport<Application>, transport_worker<Application>,
stream_socket, Application, unit_t>;
/// Implements a stream_transport that manages a stream socket.
template <class Application>
class stream_transport : public stream_transport_base<Application> {
template <class UpperLayer>
class stream_transport {
public:
// -- member types -----------------------------------------------------------
using application_type = Application;
using output_tag = tag::stream_oriented;
// -- constructors, destructors, and assignment operators --------------------
using worker_type = transport_worker<application_type>;
template <class... Ts>
stream_transport(Ts&&... xs) : upper_layer_(std::forward<Ts>(xs)...) {
// nop
}
using super = stream_transport_base<application_type>;
virtual ~stream_transport() {
// nop
}
using id_type = typename super::id_type;
// -- interface for the upper layer ------------------------------------------
using write_queue_type = std::deque<std::pair<bool, byte_buffer>>;
template <class Parent>
class access {
public:
access(Parent* parent, stream_transport* transport)
: parent_(parent), transport_(transport) {
// nop
}
// -- constructors, destructors, and assignment operators --------------------
void begin_output() {
if (transport_->write_buf_.empty())
parent_->register_writing();
}
byte_buffer& output_buffer() {
return transport_->write_buf_;
}
constexpr void end_output() {
// nop
}
bool can_send_more() const noexcept {
return output_buffer().size() < transport_->max_write_buf_size_;
}
void abort_reason(error reason) {
return parent_->abort_reason(std::move(reason));
}
void configure_read(receive_policy policy) {
if (policy.max_size > 0 && transport_->max_read_size_ == 0)
parent_->register_reading();
transport_->min_read_size_ = policy.min_size;
transport_->max_read_size_ = policy.max_size;
transport_->read_buf_.resize(policy.max_size);
}
private:
Parent* parent_;
stream_transport* transport_;
};
template <class Parent>
friend class access;
// -- properties -------------------------------------------------------------
auto& read_buffer() noexcept {
return read_buf_;
}
const auto& read_buffer() const noexcept {
return read_buf_;
}
auto& write_buffer() noexcept {
return write_buf_;
}
const auto& write_buffer() const noexcept {
return write_buf_;
}
stream_transport(stream_socket handle, application_type application)
: super(handle, std::move(application)),
written_(0),
read_threshold_(1024),
collected_(0),
max_(1024),
rd_flag_(net::receive_policy_flag::exactly) {
CAF_ASSERT(handle != invalid_socket);
if (auto err = nodelay(handle, true))
auto& upper_layer() noexcept {
return upper_layer_;
}
const auto& upper_layer() const noexcept {
return upper_layer_;
}
// -- initialization ---------------------------------------------------------
template <class Parent>
error init(Parent& parent, const settings& config) {
namespace mm = defaults::middleman;
auto default_max_reads = static_cast<uint32_t>(mm::max_consecutive_reads);
max_consecutive_reads_ = get_or(
config, "caf.middleman.max-consecutive-reads", default_max_reads);
if (auto err = nodelay(parent.handle(), true)) {
CAF_LOG_ERROR("nodelay failed: " << err);
return err;
}
if (auto socket_buf_size = send_buffer_size(parent.handle())) {
max_write_buf_size_ = *socket_buf_size;
CAF_ASSERT(max_write_buf_size_ > 0);
write_buf_.reserve(max_write_buf_size_ * 2);
} else {
CAF_LOG_ERROR("send_buffer_size: " << socket_buf_size.error());
return std::move(socket_buf_size.error());
}
access<Parent> this_layer{&parent, this};
return upper_layer_.init(this_layer, config);
}
// -- member functions -------------------------------------------------------
// -- event callbacks --------------------------------------------------------
bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id));
auto fail = [this](sec err) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
template <class Parent>
bool handle_read_event(Parent& parent) {
CAF_LOG_TRACE(CAF_ARG2("handle", parent.handle().id));
auto fail = [this, &parent](auto reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
parent.abort_reason(std::move(reason));
upper_layer_.abort(parent.abort_reason);
return false;
};
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) {
auto buf = this->read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_;
access<Parent> this_layer{&parent, this};
for (size_t i = 0; max_read_size_ > 0 && i < max_consecutive_reads_; ++i) {
CAF_ASSERT(min_read_size_ > read_size_);
auto buf = read_buf_.data() + read_size_;
size_t len = min_read_size_ - read_size_;
CAF_LOG_DEBUG(CAF_ARG2("missing", len));
auto num_bytes = read(this->handle_, make_span(buf, len));
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG2("handle", this->handle().id)
auto num_bytes = read(parent.handle(), make_span(buf, len));
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG2("handle", parent.handle().id)
<< CAF_ARG(num_bytes));
// Update state.
if (num_bytes > 0) {
collected_ += num_bytes;
if (collected_ >= read_threshold_) {
if (auto err = this->next_layer_.handle_data(
*this, make_span(this->read_buf_))) {
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err));
read_size_ += num_bytes;
if (read_size_ < min_read_size_)
continue;
auto delta = make_span(read_buf_.data() + delta_offset_,
read_size_ - delta_offset_);
auto consumed = upper_layer_.consume(this_layer, make_span(read_buf_),
delta);
if (consumed > 0) {
read_buf_.erase(read_buf_.begin(), read_buf_.begin() + consumed);
read_size_ -= consumed;
} else if (consumed < 0) {
upper_layer_.abort(parent.abort_reason_or(caf::sec::runtime_error));
return false;
}
this->prepare_next_read();
}
delta_offset_ = read_size_;
} else if (num_bytes < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and
// stop reading on the socket on hard errors.
......@@ -108,51 +199,36 @@ public:
return fail(sec::socket_disconnected);
}
}
return true;
}
bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id)
<< CAF_ARG2("queue-size", write_queue_.size()));
auto drain_write_queue = [this]() -> error_code<sec> {
// Helper function to sort empty buffers back into the right caches.
auto recycle = [this]() {
auto& front = this->write_queue_.front();
auto& is_header = front.first;
auto& buf = front.second;
written_ = 0;
buf.clear();
if (is_header) {
if (this->header_bufs_.size() < this->header_bufs_.capacity())
this->header_bufs_.emplace_back(std::move(buf));
} else if (this->payload_bufs_.size()
< this->payload_bufs_.capacity()) {
this->payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front();
};
auto fail = [this](sec err) {
CAF_LOG_DEBUG("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return err;
// Calling configure_read(read_policy::stop()) halts receive events.
return max_read_size_ > 0;
}
template <class Parent>
bool handle_write_event(Parent& parent) {
CAF_LOG_TRACE(CAF_ARG2("handle", parent.handle().id));
auto fail = [this, &parent](sec reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
parent.abort_reason(std::move(reason));
upper_layer_.abort(parent.abort_reason);
return false;
};
// Write buffers from the write_queue_ for as long as possible.
while (!write_queue_.empty()) {
auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_;
auto len = buf.size() - written_;
auto num_bytes = write(this->handle(), make_span(data, len));
if (num_bytes > 0) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(num_bytes));
written_ += num_bytes;
if (written_ >= static_cast<ptrdiff_t>(buf.size())) {
recycle();
written_ = 0;
// Allow the upper layer to add extra data to the write buffer.
access<Parent> this_layer{&parent, this};
if (!upper_layer_.prepare_send(this_layer)) {
upper_layer_.abort(parent.abort_reason_or(caf::sec::runtime_error));
return false;
}
} else if (num_bytes < 0) {
if (write_buf_.empty())
return !upper_layer_.done_sending(this_layer);
auto written = write(parent.handle(), write_buf_);
if (written > 0) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + written);
return !write_buf_.empty() || !upper_layer_.done_sending(this_layer);
} else if (written < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and
// stop writing to the socket on hard errors.
return last_socket_error_is_temporary()
? sec::unavailable_or_would_block
? true
: fail(sec::socket_operation_failed);
} else {
......@@ -160,75 +236,17 @@ public:
return fail(sec::socket_disconnected);
}
}
return none;
};
auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
return true;
}
return false;
};
do {
if (auto err = drain_write_queue())
return err == sec::unavailable_or_would_block;
} while (fetch_next_message());
CAF_ASSERT(write_queue_.empty());
return false;
}
void write_packet(id_type, span<byte_buffer*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (this->write_queue_.empty())
this->manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
auto i = buffers.begin();
this->write_queue_.emplace_back(true, std::move(*(*i++)));
while (i != buffers.end())
this->write_queue_.emplace_back(false, std::move(*(*i++)));
}
void configure_read(receive_policy::config cfg) override {
rd_flag_ = cfg.first;
max_ = cfg.second;
prepare_next_read();
}
private:
// -- utility functions ------------------------------------------------------
void prepare_next_read() {
collected_ = 0;
switch (rd_flag_) {
case net::receive_policy_flag::exactly:
if (this->read_buf_.size() != max_)
this->read_buf_.resize(max_);
read_threshold_ = max_;
break;
case net::receive_policy_flag::at_most:
if (this->read_buf_.size() != max_)
this->read_buf_.resize(max_);
read_threshold_ = 1;
break;
case net::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (this->read_buf_.size() != max_size)
this->read_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
}
write_queue_type write_queue_;
ptrdiff_t written_;
ptrdiff_t read_threshold_;
ptrdiff_t collected_;
size_t max_;
receive_policy_flag rd_flag_;
uint32_t max_consecutive_reads_ = 0;
uint32_t max_write_buf_size_ = 0;
uint32_t min_read_size_ = 0;
uint32_t max_read_size_ = 0;
ptrdiff_t read_size_ = 0;
ptrdiff_t delta_offset_ = 0;
byte_buffer read_buf_;
byte_buffer write_buf_;
UpperLayer upper_layer_;
};
} // namespace caf::net
......@@ -173,11 +173,6 @@ public:
// -- (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
/// socket.
virtual bool handle_read_event(endpoint_manager&) = 0;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 stream_oriented {};
} // namespace caf::tag
......@@ -25,6 +25,8 @@
namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys)
: super(handle, parent), sys_(sys), queue_(unit, unit, unit) {
......@@ -35,6 +37,18 @@ endpoint_manager::~endpoint_manager() {
// 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() {
if (queue_.blocked())
return nullptr;
......@@ -50,6 +64,8 @@ endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
return result;
}
// -- event management ---------------------------------------------------------
void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result;
using event_type = endpoint_manager_queue::event;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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"
using namespace caf;
namespace {
struct fixture {
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(length_prefix_framing_tests, fixture)
CAF_TEST(todo) {
// implement me
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -61,7 +61,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
mm.mpx()->set_thread_id();
auto backend = dynamic_cast<backend::test*>(mm.backend("test"));
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();
sock = backend->socket(mars);
}
......
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