Commit f03056e2 authored by Jakob Otto's avatar Jakob Otto

Use transport_base in stream_transport

parent 967793e2
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.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/net/transport_worker.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
...@@ -36,9 +37,13 @@ ...@@ -36,9 +37,13 @@
namespace caf { namespace caf {
namespace net { namespace net {
template <class Application>
using transport_base_type = transport_base<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 Application>
class stream_transport { class stream_transport : transport_base_type<Application> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -55,62 +60,21 @@ public: ...@@ -55,62 +60,21 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application) stream_transport(stream_socket handle, application_type application)
: worker_(std::move(application)), : transport_base_type<application_type>(handle, std::move(application)) {
handle_(handle),
// max_consecutive_reads_(0),
read_threshold_(1024),
collected_(0),
max_(1024),
rd_flag_(net::receive_policy_flag::exactly),
written_(0),
manager_(nullptr) {
// nop // nop
} }
// -- properties -------------------------------------------------------------
stream_socket handle() const noexcept {
return handle_;
}
actor_system& system() {
return manager().system();
}
application_type& application() {
return worker_.application();
}
transport_type& transport() {
return *this;
}
endpoint_manager& manager() {
return *manager_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> error init(endpoint_manager& parent) override {
error init(Parent& parent) { // call init function from base class
manager_ = &parent; transport_base_type<application_type>::init(parent);
auto& cfg = system().config();
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
defaults::middleman::max_header_buffers);
header_bufs_.reserve(max_header_bufs);
auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs);
if (auto err = worker_.init(*this))
return err;
return none;
} }
template <class Parent> bool handle_read_event(endpoint_manager&) override {
bool handle_read_event(Parent&) {
auto buf = read_buf_.data() + collected_; auto buf = read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_; size_t len = read_threshold_ - collected_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(this->handle_.id) << CAF_ARG(len));
auto ret = read(handle_, make_span(buf, len)); auto ret = read(handle_, make_span(buf, len));
// Update state. // Update state.
if (auto num_bytes = get_if<size_t>(&ret)) { if (auto num_bytes = get_if<size_t>(&ret)) {
...@@ -121,7 +85,9 @@ public: ...@@ -121,7 +85,9 @@ public:
CAF_LOG_WARNING("handle_data failed:" << CAF_ARG(err)); CAF_LOG_WARNING("handle_data failed:" << CAF_ARG(err));
return false; return false;
} }
prepare_next_read(); // TODO: Is this the proper way or would
// `transport_base_type<Application>::prepare_next_read()` be better?
this->prepare_next_read();
} }
} else { } else {
auto err = get<sec>(ret); auto err = get<sec>(ret);
...@@ -134,8 +100,7 @@ public: ...@@ -134,8 +100,7 @@ public:
return true; return true;
} }
template <class Parent> bool handle_write_event(endpoint_manager& parent) override {
bool handle_write_event(Parent& parent) {
// Try to write leftover data. // Try to write leftover data.
write_some(); write_some();
// Get new data from parent. // Get new data from parent.
...@@ -148,74 +113,10 @@ public: ...@@ -148,74 +113,10 @@ public:
return write_some(); return write_some();
} }
template <class Parent> void write_packet(unit_t, span<buffer_type*> buffers) override {
void resolve(Parent&, const uri& locator, const actor& listener) {
worker_.resolve(*this, locator.path(), listener);
}
template <class Parent>
void new_proxy(Parent&, const node_id& peer, actor_id id) {
worker_.new_proxy(*this, peer, id);
}
template <class Parent>
void local_actor_down(Parent&, const node_id& peer, actor_id id,
error reason) {
worker_.local_actor_down(*this, peer, id, std::move(reason));
}
template <class Parent>
void timeout(Parent&, atom_value value, uint64_t id) {
worker_.timeout(*this, value, id);
}
template <class... Ts>
void set_timeout(uint64_t, Ts&&...) {
// nop
}
void handle_error(sec code) {
worker_.handle_error(code);
}
void prepare_next_read() {
read_buf_.clear();
collected_ = 0;
// This cast does nothing, but prevents a weird compiler error on GCC
// <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch (static_cast<net::receive_policy_flag>(rd_flag_)) {
case net::receive_policy_flag::exactly:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = max_;
break;
case net::receive_policy_flag::at_most:
if (read_buf_.size() != max_)
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 (read_buf_.size() != max_size)
read_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
}
void configure_read(receive_policy::config cfg) {
rd_flag_ = cfg.first;
max_ = cfg.second;
prepare_next_read();
}
void write_packet(unit_t, span<buffer_type*> buffers) {
CAF_ASSERT(!buffers.empty()); CAF_ASSERT(!buffers.empty());
if (write_queue_.empty()) if (write_queue_.empty())
manager().register_writing(); this->manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is // By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer. // a payload buffer.
auto i = buffers.begin(); auto i = buffers.begin();
...@@ -224,28 +125,9 @@ public: ...@@ -224,28 +125,9 @@ public:
write_queue_.emplace_back(false, std::move(*(*i++))); write_queue_.emplace_back(false, std::move(*(*i++)));
} }
// -- buffer management ------------------------------------------------------
buffer_type next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
buffer_type next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
private: private:
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
if (cache.empty()) {
return {};
}
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
}
bool write_some() { bool write_some() {
CAF_LOG_TRACE(CAF_ARG(handle_.id)); CAF_LOG_TRACE(CAF_ARG(handle_.id));
// Helper function to sort empty buffers back into the right caches. // Helper function to sort empty buffers back into the right caches.
...@@ -269,7 +151,7 @@ private: ...@@ -269,7 +151,7 @@ private:
CAF_ASSERT(!buf.empty()); CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_; auto data = buf.data() + written_;
auto len = buf.size() - written_; auto len = buf.size() - written_;
auto write_ret = write(handle_, make_span(data, len)); auto write_ret = write(this->next_layer_, make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) { if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
if (*num_bytes + written_ >= buf.size()) { if (*num_bytes + written_ >= buf.size()) {
...@@ -283,7 +165,7 @@ private: ...@@ -283,7 +165,7 @@ private:
auto err = get<sec>(write_ret); auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) { if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err)); CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
worker_.handle_error(err); next_layer_.handle_error(err);
return false; return false;
} }
return true; return true;
...@@ -292,9 +174,6 @@ private: ...@@ -292,9 +174,6 @@ private:
return false; return false;
} }
worker_type worker_;
stream_socket handle_;
buffer_cache_type header_bufs_; buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_; buffer_cache_type payload_bufs_;
......
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