Unverified Commit ab940731 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #41 from actor-framework/topic/add-transport-base

Add transport_base class
parents 82e36424 421aa67f
...@@ -19,29 +19,30 @@ ...@@ -19,29 +19,30 @@
#pragma once #pragma once
#include <deque> #include <deque>
#include <unordered_map> #include <vector>
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.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/transport_base.hpp"
#include "caf/net/transport_worker_dispatcher.hpp" #include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp" #include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf { namespace caf::net {
namespace net {
template <class Factory>
using datagram_transport_base = transport_base<
datagram_transport<Factory>,
transport_worker_dispatcher<Factory, ip_endpoint>, udp_datagram_socket,
Factory, ip_endpoint>;
/// Implements a udp_transport policy that manages a datagram socket. /// Implements a udp_transport policy that manages a datagram socket.
template <class Factory> template <class Factory>
class datagram_transport { class datagram_transport : public datagram_transport_base<Factory> {
public: public:
// Maximal UDP-packet size // Maximal UDP-packet size
static constexpr size_t max_datagram_size = std::numeric_limits< static constexpr size_t max_datagram_size = std::numeric_limits<
...@@ -49,163 +50,87 @@ public: ...@@ -49,163 +50,87 @@ public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using id_type = ip_endpoint; using factory_type = Factory;
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>; using id_type = ip_endpoint;
using factory_type = Factory; using application_type = typename factory_type::application_type;
using transport_type = datagram_transport<Factory>; using super = datagram_transport_base<factory_type>;
using dispatcher_type = transport_worker_dispatcher<transport_type, id_type>; using buffer_type = typename super::buffer_type;
using application_type = typename Factory::application_type; using buffer_cache_type = typename super::buffer_cache_type;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
datagram_transport(udp_datagram_socket handle, factory_type factory) datagram_transport(udp_datagram_socket handle, factory_type factory)
: dispatcher_(*this, std::move(factory)), : super(handle, std::move(factory)) {
handle_(handle),
read_buf_(max_datagram_size),
manager_(nullptr) {
// nop // nop
} }
// -- properties -------------------------------------------------------------
udp_datagram_socket handle() const noexcept {
return handle_;
}
actor_system& system() {
return manager().system();
}
transport_type& transport() {
return *this;
}
endpoint_manager& manager() {
return *manager_;
}
// -- public member functions ------------------------------------------------ // -- public member functions ------------------------------------------------
template <class Parent> error init(endpoint_manager& manager) override {
error init(Parent& parent) { CAF_LOG_TRACE("");
manager_ = &parent; if (auto err = super::init(manager))
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 = dispatcher_.init(*this))
return err; return err;
prepare_next_read();
return none; return none;
} }
template <class Parent> bool handle_read_event(endpoint_manager&) override {
bool handle_read_event(Parent&) { CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
CAF_LOG_TRACE(CAF_ARG(handle_.id)); auto ret = read(this->handle_, make_span(this->read_buf_));
auto ret = read(handle_, make_span(read_buf_));
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) { if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto num_bytes = res->first; auto num_bytes = res->first;
CAF_LOG_DEBUG("received " << num_bytes << " bytes");
auto ep = res->second; auto ep = res->second;
read_buf_.resize(num_bytes); this->read_buf_.resize(num_bytes);
dispatcher_.handle_data(*this, make_span(read_buf_), std::move(ep)); this->next_layer_.handle_data(*this, make_span(this->read_buf_),
std::move(ep));
prepare_next_read(); prepare_next_read();
} else { } else {
auto err = get<sec>(ret); auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err)); CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
dispatcher_.handle_error(err); this->next_layer_.handle_error(err);
return false; return false;
} }
return true; return true;
} }
template <class Parent> bool handle_write_event(endpoint_manager& manager) override {
bool handle_write_event(Parent& parent) { CAF_LOG_TRACE(CAF_ARG(this->handle_.id)
CAF_LOG_TRACE(CAF_ARG(handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size())); << CAF_ARG2("queue-size", packet_queue_.size()));
// Try to write leftover data. // Try to write leftover data.
write_some(); write_some();
// Get new data from parent. // Get new data from parent.
for (auto msg = parent.next_message(); msg != nullptr; for (auto msg = manager.next_message(); msg != nullptr;
msg = parent.next_message()) { msg = manager.next_message()) {
dispatcher_.write_message(*this, std::move(msg)); this->next_layer_.write_message(*this, std::move(msg));
} }
// Write prepared data. // Write prepared data.
return write_some(); return write_some();
} }
template <class Parent> // TODO: remove this function. `resolve` should add workers when needed.
void resolve(Parent&, const uri& locator, const actor& listener) {
dispatcher_.resolve(*this, locator, listener);
}
template <class Parent>
void new_proxy(Parent&, const node_id& peer, actor_id id) {
dispatcher_.new_proxy(*this, peer, id);
}
template <class Parent>
void local_actor_down(Parent&, const node_id& peer, actor_id id,
error reason) {
dispatcher_.local_actor_down(*this, peer, id, std::move(reason));
}
template <class Parent>
void timeout(Parent&, atom_value value, uint64_t id) {
dispatcher_.timeout(*this, value, id);
}
void set_timeout(uint64_t timeout_id, id_type id) {
dispatcher_.set_timeout(timeout_id, id);
}
void handle_error(sec code) {
dispatcher_.handle_error(code);
}
error add_new_worker(node_id node, id_type id) { error add_new_worker(node_id node, id_type id) {
auto worker = dispatcher_.add_new_worker(*this, node, id); auto worker = this->next_layer_.add_new_worker(*this, node, id);
if (!worker) if (!worker)
return worker.error(); return worker.error();
return none; return none;
} }
void prepare_next_read() { void write_packet(id_type id, span<buffer_type*> buffers) override {
read_buf_.clear(); CAF_LOG_TRACE("");
read_buf_.resize(max_datagram_size);
}
void configure_read(receive_policy::config) {
// nop
}
void write_packet(id_type id, span<buffer_type*> buffers) {
CAF_ASSERT(!buffers.empty()); CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty()) if (packet_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.
packet_queue_.emplace_back(id, buffers); packet_queue_.emplace_back(id, buffers);
} }
// -- buffer management ------------------------------------------------------
buffer_type next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
buffer_type next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
/// Helper struct for managing outgoing packets /// Helper struct for managing outgoing packets
struct packet { struct packet {
id_type id; id_type id;
...@@ -219,75 +144,66 @@ public: ...@@ -219,75 +144,66 @@ public:
bytes.emplace_back(std::move(*buf)); bytes.emplace_back(std::move(*buf));
} }
} }
std::vector<std::vector<byte>*> get_buffer_ptrs() {
std::vector<std::vector<byte>*> ptrs;
for (auto& buf : bytes)
ptrs.emplace_back(&buf);
return ptrs;
}
}; };
private: private:
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) { void prepare_next_read() {
if (cache.empty()) { this->read_buf_.resize(max_datagram_size);
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_ARG2("handle", this->handle_.id));
// Helper function to sort empty buffers back into the right caches. // Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() { auto recycle = [&]() {
auto& front = packet_queue_.front(); auto& front = packet_queue_.front();
auto& bufs = front.bytes; auto& bufs = front.bytes;
auto it = bufs.begin(); auto it = bufs.begin();
if (header_bufs_.size() < header_bufs_.capacity()) { if (this->header_bufs_.size() < this->header_bufs_.capacity()) {
it->clear(); it->clear();
header_bufs_.emplace_back(std::move(*it++)); this->header_bufs_.emplace_back(std::move(*it++));
} }
for (; for (; it != bufs.end()
it != bufs.end() && payload_bufs_.size() < payload_bufs_.capacity(); && this->payload_bufs_.size() < this->payload_bufs_.capacity();
++it) { ++it) {
it->clear(); it->clear();
payload_bufs_.emplace_back(std::move(*it)); this->payload_bufs_.emplace_back(std::move(*it));
} }
packet_queue_.pop_front(); packet_queue_.pop_front();
}; };
// Write as many bytes as possible. // Write as many bytes as possible.
while (!packet_queue_.empty()) { while (!packet_queue_.empty()) {
auto& packet = packet_queue_.front(); auto& packet = packet_queue_.front();
std::vector<std::vector<byte>*> ptrs; auto ptrs = packet.get_buffer_ptrs();
for (auto& buf : packet.bytes) auto write_ret = write(this->handle_, make_span(ptrs), packet.id);
ptrs.emplace_back(&buf);
auto write_ret = write(handle_, make_span(ptrs), packet.id);
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(this->handle_.id) << CAF_ARG(*num_bytes));
CAF_LOG_WARNING_IF(*num_bytes < packet.size, CAF_LOG_WARNING_IF(*num_bytes < packet.size,
"packet was not sent completely"); "packet was not sent completely");
recycle(); recycle();
} else { } else {
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_ERROR("write failed" << CAF_ARG(err));
dispatcher_.handle_error(err); this->next_layer_.handle_error(err);
return false; return false;
} }
CAF_LOG_DEBUG("write returned `unavailable_or_would_block`");
return true; return true;
} }
} }
return false; return false;
} }
dispatcher_type dispatcher_;
udp_datagram_socket handle_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
std::vector<byte> read_buf_;
std::deque<packet> packet_queue_; std::deque<packet> packet_queue_;
endpoint_manager* manager_;
}; };
} // namespace net } // namespace caf::net
} // namespace caf
...@@ -22,11 +22,16 @@ ...@@ -22,11 +22,16 @@
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
namespace caf { namespace caf::net {
namespace net {
// -- templates ---------------------------------------------------------------- // -- templates ----------------------------------------------------------------
template <class Application>
class stream_transport;
template <class Factory>
class datagram_transport;
template <class Application, class IdType = unit_t> template <class Application, class IdType = unit_t>
class transport_worker; class transport_worker;
...@@ -49,6 +54,8 @@ struct socket; ...@@ -49,6 +54,8 @@ struct socket;
struct stream_socket; struct stream_socket;
struct tcp_accept_socket; struct tcp_accept_socket;
struct tcp_stream_socket; struct tcp_stream_socket;
struct datagram_socket;
struct udp_datagram_socket;
// -- smart pointers ----------------------------------------------------------- // -- smart pointers -----------------------------------------------------------
...@@ -58,5 +65,4 @@ using multiplexer_ptr = std::shared_ptr<multiplexer>; ...@@ -58,5 +65,4 @@ using multiplexer_ptr = std::shared_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>; using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>; using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
} // namespace net } // namespace caf::net
} // namespace caf
...@@ -5,7 +5,7 @@ ...@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework * * | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2018 Dominik Charousset * * Copyright 2011-2019 Dominik Charousset *
* * * *
* Distributed under the terms and conditions of the BSD 3-Clause License or * * 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 * * (at your option) under the terms and conditions of the Boost Software *
...@@ -18,248 +18,163 @@ ...@@ -18,248 +18,163 @@
#pragma once #pragma once
#include "caf/actor_system_config.hpp" #include <deque>
#include "caf/byte.hpp" #include <vector>
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp" #include "caf/logger.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/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/net/transport_worker.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf { namespace caf::net {
namespace 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 Application>
class stream_transport { class stream_transport : public stream_transport_base<Application> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using application_type = Application; using application_type = Application;
using transport_type = stream_transport;
using worker_type = transport_worker<application_type>; using worker_type = transport_worker<application_type>;
using buffer_type = std::vector<byte>; using id_type = unit_t;
using super = stream_transport_base<application_type>;
using buffer_cache_type = std::vector<buffer_type>; using buffer_type = typename super::buffer_type;
using write_queue_type = std::deque<std::pair<bool, buffer_type>>;
// -- 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)), : super(handle, std::move(application)),
handle_(handle), written_(0),
// max_consecutive_reads_(0),
read_threshold_(1024), read_threshold_(1024),
collected_(0), collected_(0),
max_(1024), max_(1024),
rd_flag_(net::receive_policy_flag::exactly), 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> bool handle_read_event(endpoint_manager&) override {
error init(Parent& parent) { auto buf = this->read_buf_.data() + this->collected_;
manager_ = &parent; size_t len = this->read_threshold_ - this->collected_;
auto& cfg = system().config(); CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id)
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers", << CAF_ARG2("missing", len));
defaults::middleman::max_header_buffers); auto ret = read(this->handle_, make_span(buf, len));
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(Parent&) {
auto buf = read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(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)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(len)
collected_ += *num_bytes; << CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
if (collected_ >= read_threshold_) { this->collected_ += *num_bytes;
if (auto err = worker_.handle_data(*this, read_buf_)) { if (this->collected_ >= this->read_threshold_) {
CAF_LOG_WARNING("handle_data failed:" << CAF_ARG(err)); if (auto err = this->next_layer_.handle_data(*this, this->read_buf_)) {
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err));
return false; return false;
} }
prepare_next_read(); this->prepare_next_read();
} }
} else { } else {
auto err = get<sec>(ret); auto err = get<sec>(ret);
if (err != sec::unavailable_or_would_block) { if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err)); CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
worker_.handle_error(err); this->next_layer_.handle_error(err);
return false; return false;
} }
} }
return true; return true;
} }
template <class Parent> bool handle_write_event(endpoint_manager& parent) override {
bool handle_write_event(Parent& parent) { CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id));
// Try to write leftover data. // Try to write leftover data.
write_some(); write_some();
// Get new data from parent. // Get new data from parent.
// TODO: dont read all messages at once - get one by one. // TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr; for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) { msg = parent.next_message()) {
worker_.write_message(*this, std::move(msg)); this->next_layer_.write_message(*this, std::move(msg));
} }
// Write prepared data. // Write prepared data.
return write_some(); return write_some();
} }
template <class Parent> void write_packet(id_type, span<buffer_type*> buffers) override {
void resolve(Parent&, const uri& locator, const actor& listener) { CAF_LOG_TRACE("");
worker_.resolve(*this, locator.path(), listener); CAF_ASSERT(!buffers.empty());
} if (this->write_queue_.empty())
this->manager().register_writing();
template <class Parent> // By convention, the first buffer is a header buffer. Every other buffer is
void new_proxy(Parent&, const node_id& peer, actor_id id) { // a payload buffer.
worker_.new_proxy(*this, peer, id); auto i = buffers.begin();
} this->write_queue_.emplace_back(true, std::move(*(*i++)));
while (i != buffers.end())
template <class Parent> this->write_queue_.emplace_back(false, std::move(*(*i++)));
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 configure_read(receive_policy::config cfg) override {
void set_timeout(uint64_t, Ts&&...) { rd_flag_ = cfg.first;
// nop max_ = cfg.second;
prepare_next_read();
} }
void handle_error(sec code) { private:
worker_.handle_error(code); // -- utility functions ------------------------------------------------------
}
void prepare_next_read() { void prepare_next_read() {
read_buf_.clear();
collected_ = 0; collected_ = 0;
// This cast does nothing, but prevents a weird compiler error on GCC switch (rd_flag_) {
// <= 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: case net::receive_policy_flag::exactly:
if (read_buf_.size() != max_) if (this->read_buf_.size() != max_)
read_buf_.resize(max_); this->read_buf_.resize(max_);
read_threshold_ = max_; read_threshold_ = max_;
break; break;
case net::receive_policy_flag::at_most: case net::receive_policy_flag::at_most:
if (read_buf_.size() != max_) if (this->read_buf_.size() != max_)
read_buf_.resize(max_); this->read_buf_.resize(max_);
read_threshold_ = 1; read_threshold_ = 1;
break; break;
case net::receive_policy_flag::at_least: { case net::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more // read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10); auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (read_buf_.size() != max_size) if (this->read_buf_.size() != max_size)
read_buf_.resize(max_size); this->read_buf_.resize(max_size);
read_threshold_ = max_; read_threshold_ = max_;
break; 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());
if (write_queue_.empty())
manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
auto i = buffers.begin();
write_queue_.emplace_back(true, std::move(*(*i++)));
while (i != buffers.end())
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:
// -- 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_ARG2("handle", this->handle_.id));
// Helper function to sort empty buffers back into the right caches. // Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() { auto recycle = [&]() {
auto& front = write_queue_.front(); auto& front = this->write_queue_.front();
auto& is_header = front.first; auto& is_header = front.first;
auto& buf = front.second; auto& buf = front.second;
written_ = 0; written_ = 0;
buf.clear(); buf.clear();
if (is_header) { if (is_header) {
if (header_bufs_.size() < header_bufs_.capacity()) if (this->header_bufs_.size() < this->header_bufs_.capacity())
header_bufs_.emplace_back(std::move(buf)); this->header_bufs_.emplace_back(std::move(buf));
} else if (payload_bufs_.size() < payload_bufs_.capacity()) { } else if (this->payload_bufs_.size() < this->payload_bufs_.capacity()) {
payload_bufs_.emplace_back(std::move(buf)); this->payload_bufs_.emplace_back(std::move(buf));
} }
write_queue_.pop_front(); write_queue_.pop_front();
}; };
...@@ -269,9 +184,9 @@ private: ...@@ -269,9 +184,9 @@ 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->handle(), 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(this->handle_.id) << CAF_ARG(*num_bytes));
if (*num_bytes + written_ >= buf.size()) { if (*num_bytes + written_ >= buf.size()) {
recycle(); recycle();
written_ = 0; written_ = 0;
...@@ -283,7 +198,7 @@ private: ...@@ -283,7 +198,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); this->next_layer_.handle_error(err);
return false; return false;
} }
return true; return true;
...@@ -292,26 +207,12 @@ private: ...@@ -292,26 +207,12 @@ private:
return false; return false;
} }
worker_type worker_; write_queue_type write_queue_;
stream_socket handle_; size_t written_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
buffer_type read_buf_;
std::deque<std::pair<bool, buffer_type>> write_queue_;
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
size_t read_threshold_; size_t read_threshold_;
size_t collected_; size_t collected_;
size_t max_; size_t max_;
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
size_t written_;
endpoint_manager* manager_;
}; };
} // namespace net } // namespace caf::net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/detail/overload.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Implements base class for transports.
template <class Transport, class NextLayer, class Handle, class Application,
class IdType>
class transport_base {
public:
// -- member types -----------------------------------------------------------
using next_layer_type = NextLayer;
using handle_type = Handle;
using transport_type = Transport;
using application_type = Application;
using id_type = IdType;
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
// -- constructors, destructors, and assignment operators --------------------
transport_base(handle_type handle, application_type application)
: next_layer_(std::move(application)),
handle_(handle),
// max_consecutive_reads_(0),
manager_(nullptr) {
// nop
}
// -- properties -------------------------------------------------------------
handle_type handle() const noexcept {
return handle_;
}
actor_system& system() {
return manager().system();
}
application_type& application() {
return next_layer_.application();
}
transport_type& transport() {
return *reinterpret_cast<transport_type*>(this);
}
endpoint_manager& manager() {
return *manager_;
}
// -- transport member functions ---------------------------------------------
virtual error init(endpoint_manager& parent) {
CAF_LOG_TRACE("");
manager_ = &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 = next_layer_.init(*this))
return err;
return none;
}
auto resolve(endpoint_manager&, const uri& locator, const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(locator) << CAF_ARG(listener));
auto f = detail::make_overload(
[&](auto& layer) -> decltype(layer.resolve(*this, locator, listener)) {
return layer.resolve(*this, locator, listener);
},
[&](auto& layer) -> decltype(
layer.resolve(*this, locator.path(), listener)) {
return layer.resolve(*this, locator.path(), listener);
});
f(next_layer_);
}
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) {
next_layer_.new_proxy(*this, peer, id);
}
void local_actor_down(endpoint_manager&, const node_id& peer, actor_id id,
error reason) {
next_layer_.local_actor_down(*this, peer, id, std::move(reason));
}
void timeout(endpoint_manager&, atom_value value, uint64_t id) {
next_layer_.timeout(*this, value, id);
}
void set_timeout(uint64_t timeout_id, id_type id) {
next_layer_.set_timeout(timeout_id, id);
}
void handle_error(sec code) {
next_layer_.handle_error(code);
}
// -- (pure) virtual functions -----------------------------------------------
virtual void configure_read(receive_policy::config){
// nop
};
virtual bool handle_read_event(endpoint_manager&) = 0;
virtual bool handle_write_event(endpoint_manager& parent) = 0;
virtual void write_packet(id_type id, span<buffer_type*> buffers) = 0;
// -- 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:
// -- 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;
}
protected:
next_layer_type next_layer_;
handle_type handle_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
buffer_type read_buf_;
// TODO implement retries using this member! Should this go into stream_trans?
// size_t max_consecutive_reads_;
endpoint_manager* manager_;
};
} // namespace caf::net
...@@ -18,16 +18,12 @@ ...@@ -18,16 +18,12 @@
#pragma once #pragma once
#include "caf/byte.hpp" #include "caf/net/endpoint_manager_queue.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp" #include "caf/net/packet_writer_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
namespace caf { namespace caf::net {
namespace net {
/// Implements a worker for transport protocols. /// Implements a worker for transport protocols.
template <class Application, class IdType> template <class Application, class IdType>
...@@ -41,7 +37,8 @@ public: ...@@ -41,7 +37,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
transport_worker(application_type application, id_type id = id_type{}) explicit transport_worker(application_type application,
id_type id = id_type{})
: application_(std::move(application)), id_(std::move(id)) { : application_(std::move(application)), id_(std::move(id)) {
// nop // nop
} }
...@@ -119,5 +116,4 @@ template <class Application, class IdType = unit_t> ...@@ -119,5 +116,4 @@ template <class Application, class IdType = unit_t>
using transport_worker_ptr = std::shared_ptr< using transport_worker_ptr = std::shared_ptr<
transport_worker<Application, IdType>>; transport_worker<Application, IdType>>;
} // namespace net } // namespace caf::net
} // namespace caf
...@@ -18,34 +18,27 @@ ...@@ -18,34 +18,27 @@
#pragma once #pragma once
#include <caf/logger.hpp>
#include <caf/sec.hpp>
#include <unordered_map> #include <unordered_map>
#include "caf/byte.hpp" #include "caf/logger.hpp"
#include "caf/ip_endpoint.hpp" #include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp" #include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/transport_worker.hpp" #include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace caf { namespace caf::net {
namespace net {
/// Implements a dispatcher that dispatches between transport and workers. /// Implements a dispatcher that dispatches between transport and workers.
template <class Transport, class IdType> template <class Factory, class IdType>
class transport_worker_dispatcher { class transport_worker_dispatcher {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using id_type = IdType; using id_type = IdType;
using transport_type = Transport; using factory_type = Factory;
using factory_type = typename transport_type::factory_type;
using application_type = typename factory_type::application_type; using application_type = typename factory_type::application_type;
...@@ -55,8 +48,8 @@ public: ...@@ -55,8 +48,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
transport_worker_dispatcher(transport_type& transport, factory_type factory) explicit transport_worker_dispatcher(factory_type factory)
: factory_(std::move(factory)), transport_(&transport) { : factory_(std::move(factory)) {
// nop // nop
} }
...@@ -141,6 +134,7 @@ public: ...@@ -141,6 +134,7 @@ public:
template <class Parent> template <class Parent>
expected<worker_ptr> add_new_worker(Parent& parent, node_id node, expected<worker_ptr> add_new_worker(Parent& parent, node_id node,
id_type id) { id_type id) {
CAF_LOG_TRACE(CAF_ARG(node) << CAF_ARG(id));
auto application = factory_.make(); auto application = factory_.make();
auto worker = std::make_shared<worker_type>(std::move(application), id); auto worker = std::make_shared<worker_type>(std::move(application), id);
if (auto err = worker->init(parent)) if (auto err = worker->init(parent))
...@@ -176,8 +170,6 @@ private: ...@@ -176,8 +170,6 @@ private:
std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_; std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_;
factory_type factory_; factory_type factory_;
transport_type* transport_;
}; };
} // namespace net } // namespace caf::net
} // namespace caf
...@@ -25,9 +25,9 @@ ...@@ -25,9 +25,9 @@
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp" #include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/monitorable_actor.hpp" #include "caf/monitorable_actor.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
...@@ -182,12 +183,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) { ...@@ -182,12 +183,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
} }
struct fixture : host_fixture { struct fixture : host_fixture {
using dispatcher_type = transport_worker_dispatcher<dummy_transport, using dispatcher_type = transport_worker_dispatcher<dummy_application_factory,
ip_endpoint>; ip_endpoint>;
fixture() fixture()
: buf{std::make_shared<buffer_type>()}, : buf{std::make_shared<buffer_type>()},
dispatcher{dummy, dummy_application_factory{buf}}, dispatcher{dummy_application_factory{buf}},
dummy{buf} { dummy{buf} {
add_new_workers(); add_new_workers();
} }
...@@ -272,7 +273,7 @@ struct fixture : host_fixture { ...@@ -272,7 +273,7 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture) CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture)
CAF_TEST(init) { CAF_TEST(init) {
dispatcher_type dispatcher{dummy, dummy_application_factory{buf}}; dispatcher_type dispatcher{dummy_application_factory{buf}};
CAF_CHECK_EQUAL(dispatcher.init(dummy), none); CAF_CHECK_EQUAL(dispatcher.init(dummy), none);
} }
......
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