Commit a8156bf4 authored by Dominik Charousset's avatar Dominik Charousset

WIP

parent adcceeb9
...@@ -123,30 +123,31 @@ target_include_directories(caf-net-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/tes ...@@ -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) 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
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.message_queue
net.basp.ping_pong net.basp.ping_pong
net.basp.worker net.basp.worker
accept_socket net.length_prefix_framing
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
multiplexer
transport_worker_dispatcher transport_worker_dispatcher
udp_datagram_socket udp_datagram_socket
network_socket
net.backend.tcp
) )
...@@ -48,24 +48,52 @@ stack *up*. Outgoing data always travels the protocol stack *down*. ...@@ -48,24 +48,52 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
..code-block:: C++ ..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] { 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. /// Consumes bytes from the lower layer.
/// @param down Reference to the lower layer that received the data. /// @param down Reference to the lower layer that received the data.
/// @param buffer Available bytes to read. /// @param buffer Available bytes to read.
/// @param delta Bytes that arrived since last calling this function. /// @param delta Bytes that arrived since last calling this function.
/// @returns The number of consumed bytes (may be zero if waiting for more /// @returns The number of consumed bytes. May be zero if waiting for more
/// input or negative to signal an error) and a policy that /// input or negative to signal an error.
/// configures how many bytes to receive next (as well as /// @note When returning a negative value, clients should also call
/// thresholds for when to call this function again). /// `down.abort_reason(...)` with an appropriate error code.
/// @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.
template <class LowerLayer> template <class LowerLayer>
pair<ptrdiff_t, receive_policy> consume(LowerLayer& down, byte_span buffer, ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span delta);
byte_span delta);
} }
interface stream_oriented [role: lower layer] { 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 /// Prepares the layer for outgoing traffic, e.g., by allocating an output
/// buffer as necessary. /// buffer as necessary.
void begin_output(); void begin_output();
...@@ -78,6 +106,11 @@ stack *up*. Outgoing data always travels the protocol stack *down*. ...@@ -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 /// Prepares written data for transfer, e.g., by flushing buffers or
/// registering sockets for write events. /// registering sockets for write events.
void end_output(); 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] { interface datagram_oriented [role: upper layer] {
...@@ -137,6 +170,9 @@ stack *up*. Outgoing data always travels the protocol stack *down*. ...@@ -137,6 +170,9 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
byte_buffer& message_buffer(); byte_buffer& message_buffer();
/// Seals and prepares a message for transfer. /// 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 @@ ...@@ -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"
......
...@@ -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 -------------------------------------------------------
......
...@@ -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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 @@ ...@@ -18,45 +18,28 @@
#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 {
public:
receive_policy() = delete;
using config = std::pair<receive_policy_flag, size_t>;
static config at_least(size_t num_bytes) { /// @pre `min_size > 0`
CAF_ASSERT(num_bytes > 0); /// @pre `min_size <= max_size`
return {receive_policy_flag::at_least, num_bytes}; 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) { /// @pre `size > 0`
CAF_ASSERT(num_bytes > 0); static constexpr receive_policy exactly(uint32_t size) {
return {receive_policy_flag::at_most, num_bytes}; return {size, size};
} }
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};
} }
}; };
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#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/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"
...@@ -70,6 +71,21 @@ public: ...@@ -70,6 +71,21 @@ 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_;
}
// -- event loop management -------------------------------------------------- // -- event loop management --------------------------------------------------
void register_reading(); void register_reading();
...@@ -96,8 +112,65 @@ protected: ...@@ -96,8 +112,65 @@ protected:
operation mask_; operation mask_;
weak_multiplexer_ptr parent_; 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>; 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 } // namespace caf::net
...@@ -21,81 +21,172 @@ ...@@ -21,81 +21,172 @@
#include <deque> #include <deque>
#include "caf/byte_buffer.hpp" #include "caf/byte_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp" #include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.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/sec.hpp"
#include "caf/settings.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace caf::net { 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. /// Implements a stream_transport that manages a stream socket.
template <class Application> template <class UpperLayer>
class stream_transport : public stream_transport_base<Application> { class stream_transport {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using application_type = Application; using output_tag = tag::stream_oriented;
using worker_type = transport_worker<application_type>; // -- constructors, destructors, and assignment operators --------------------
using super = stream_transport_base<application_type>; template <class... Ts>
stream_transport(Ts&&... xs) : upper_layer_(std::forward<Ts>(xs)...) {
// nop
}
using id_type = typename super::id_type; virtual ~stream_transport() {
// nop
}
using write_queue_type = std::deque<std::pair<bool, byte_buffer>>; // -- interface for the upper layer ------------------------------------------
// -- constructors, destructors, and assignment operators -------------------- template <class Parent>
class access {
public:
access(Parent* parent, stream_transport* transport)
: parent_(parent), transport_(transport) {
// nop
}
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_;
}
auto& upper_layer() noexcept {
return upper_layer_;
}
const auto& upper_layer() const noexcept {
return upper_layer_;
}
stream_transport(stream_socket handle, application_type application) // -- initialization ---------------------------------------------------------
: super(handle, std::move(application)),
written_(0), template <class Parent>
read_threshold_(1024), error init(Parent& parent, const settings& config) {
collected_(0), namespace mm = defaults::middleman;
max_(1024), auto default_max_reads = static_cast<uint32_t>(mm::max_consecutive_reads);
rd_flag_(net::receive_policy_flag::exactly) { max_consecutive_reads_ = get_or(
CAF_ASSERT(handle != invalid_socket); config, "caf.middleman.max-consecutive-reads", default_max_reads);
if (auto err = nodelay(handle, true)) if (auto err = nodelay(parent.handle(), true)) {
CAF_LOG_ERROR("nodelay failed: " << err); 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 { template <class Parent>
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id)); bool handle_read_event(Parent& parent) {
auto fail = [this](sec err) { CAF_LOG_TRACE(CAF_ARG2("handle", parent.handle().id));
CAF_LOG_DEBUG("read failed" << CAF_ARG(err)); auto fail = [this, &parent](auto reason) {
this->next_layer_.handle_error(err); CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
parent.abort_reason(std::move(reason));
upper_layer_.abort(parent.abort_reason);
return false; return false;
}; };
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) { access<Parent> this_layer{&parent, this};
auto buf = this->read_buf_.data() + collected_; for (size_t i = 0; max_read_size_ > 0 && i < max_consecutive_reads_; ++i) {
size_t len = read_threshold_ - collected_; 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)); CAF_LOG_DEBUG(CAF_ARG2("missing", len));
auto num_bytes = read(this->handle_, make_span(buf, len)); auto num_bytes = read(parent.handle(), make_span(buf, len));
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG2("handle", this->handle().id) CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG2("handle", parent.handle().id)
<< CAF_ARG(num_bytes)); << CAF_ARG(num_bytes));
// Update state. // Update state.
if (num_bytes > 0) { if (num_bytes > 0) {
collected_ += num_bytes; read_size_ += num_bytes;
if (collected_ >= read_threshold_) { if (read_size_ < min_read_size_)
if (auto err = this->next_layer_.handle_data( continue;
*this, make_span(this->read_buf_))) { auto delta = make_span(read_buf_.data() + delta_offset_,
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err)); read_size_ - delta_offset_);
return false; auto consumed = upper_layer_.consume(this_layer, make_span(read_buf_),
} delta);
this->prepare_next_read(); 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;
} }
delta_offset_ = read_size_;
} else if (num_bytes < 0) { } else if (num_bytes < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and // Try again later on temporary errors such as EWOULDBLOCK and
// stop reading on the socket on hard errors. // stop reading on the socket on hard errors.
...@@ -108,127 +199,54 @@ public: ...@@ -108,127 +199,54 @@ public:
return fail(sec::socket_disconnected); return fail(sec::socket_disconnected);
} }
} }
return true; // Calling configure_read(read_policy::stop()) halts receive events.
return max_read_size_ > 0;
} }
bool handle_write_event(endpoint_manager& manager) override { template <class Parent>
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id) bool handle_write_event(Parent& parent) {
<< CAF_ARG2("queue-size", write_queue_.size())); CAF_LOG_TRACE(CAF_ARG2("handle", parent.handle().id));
auto drain_write_queue = [this]() -> error_code<sec> { auto fail = [this, &parent](sec reason) {
// Helper function to sort empty buffers back into the right caches. CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
auto recycle = [this]() { parent.abort_reason(std::move(reason));
auto& front = this->write_queue_.front(); upper_layer_.abort(parent.abort_reason);
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;
};
// 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;
}
} else if (num_bytes < 0) {
return last_socket_error_is_temporary()
? sec::unavailable_or_would_block
: fail(sec::socket_operation_failed);
} else {
// write() returns 0 iff the connection was closed.
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; return false;
}; };
do { // Allow the upper layer to add extra data to the write buffer.
if (auto err = drain_write_queue()) access<Parent> this_layer{&parent, this};
return err == sec::unavailable_or_would_block; if (!upper_layer_.prepare_send(this_layer)) {
} while (fetch_next_message()); upper_layer_.abort(parent.abort_reason_or(caf::sec::runtime_error));
CAF_ASSERT(write_queue_.empty()); return false;
return false; }
} if (write_buf_.empty())
return !upper_layer_.done_sending(this_layer);
void write_packet(id_type, span<byte_buffer*> buffers) override { auto written = write(parent.handle(), write_buf_);
CAF_LOG_TRACE(""); if (written > 0) {
CAF_ASSERT(!buffers.empty()); write_buf_.erase(write_buf_.begin(), write_buf_.begin() + written);
if (this->write_queue_.empty()) return !write_buf_.empty() || !upper_layer_.done_sending(this_layer);
this->manager().register_writing(); } else if (written < 0) {
// By convention, the first buffer is a header buffer. Every other buffer is // Try again later on temporary errors such as EWOULDBLOCK and
// a payload buffer. // stop writing to the socket on hard errors.
auto i = buffers.begin(); return last_socket_error_is_temporary()
this->write_queue_.emplace_back(true, std::move(*(*i++))); ? true
while (i != buffers.end()) : fail(sec::socket_operation_failed);
this->write_queue_.emplace_back(false, std::move(*(*i++)));
} } else {
// write() returns 0 iff the connection was closed.
void configure_read(receive_policy::config cfg) override { return fail(sec::socket_disconnected);
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_; private:
ptrdiff_t written_; uint32_t max_consecutive_reads_ = 0;
ptrdiff_t read_threshold_; uint32_t max_write_buf_size_ = 0;
ptrdiff_t collected_; uint32_t min_read_size_ = 0;
size_t max_; uint32_t max_read_size_ = 0;
receive_policy_flag rd_flag_; ptrdiff_t read_size_ = 0;
ptrdiff_t delta_offset_ = 0;
byte_buffer read_buf_;
byte_buffer write_buf_;
UpperLayer upper_layer_;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -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;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 @@ ...@@ -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;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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> { ...@@ -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);
} }
......
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