Commit c84b6b6c authored by Jakob Otto's avatar Jakob Otto

Cleanup transports

parent c81705c6
......@@ -19,22 +19,17 @@
#pragma once
#include <deque>
#include <unordered_map>
#include <vector>
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.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/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf::net {
......@@ -76,6 +71,7 @@ public:
// -- public member functions ------------------------------------------------
error init(endpoint_manager& manager) override {
CAF_LOG_TRACE("");
if (auto err = super::init(manager))
return err;
prepare_next_read();
......@@ -87,6 +83,7 @@ public:
auto ret = read(this->handle_, make_span(this->read_buf_));
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto num_bytes = res->first;
CAF_LOG_DEBUG("received " << num_bytes << " bytes");
auto ep = res->second;
this->read_buf_.resize(num_bytes);
this->next_layer_.handle_data(*this, make_span(this->read_buf_),
......@@ -115,6 +112,7 @@ public:
return write_some();
}
// TODO: remove this function. `resolve` should add workers when needed.
error add_new_worker(node_id node, id_type id) {
auto worker = this->next_layer_.add_new_worker(*this, node, id);
if (!worker)
......@@ -123,6 +121,7 @@ public:
}
void write_packet(id_type id, span<buffer_type*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty())
this->manager().register_writing();
......@@ -144,6 +143,13 @@ public:
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:
......@@ -154,7 +160,7 @@ private:
}
bool write_some() {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id));
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = packet_queue_.front();
......@@ -175,9 +181,7 @@ private:
// Write as many bytes as possible.
while (!packet_queue_.empty()) {
auto& packet = packet_queue_.front();
std::vector<std::vector<byte>*> ptrs;
for (auto& buf : packet.bytes)
ptrs.emplace_back(&buf);
auto ptrs = packet.get_buffer_ptrs();
auto write_ret = write(this->handle_, make_span(ptrs), packet.id);
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
......@@ -187,10 +191,11 @@ private:
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
CAF_LOG_ERROR("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
CAF_LOG_DEBUG("write returned `unavailable_or_would_block`");
return true;
}
}
......
......@@ -18,10 +18,9 @@
#pragma once
#include "caf/actor_system_config.hpp"
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include <deque>
#include <vector>
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
......@@ -33,7 +32,6 @@
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf::net {
......@@ -50,18 +48,16 @@ public:
using application_type = Application;
using transport_type = stream_transport<application_type>;
using worker_type = transport_worker<application_type>;
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
using id_type = unit_t;
using super = stream_transport_base<application_type>;
using buffer_type = typename super::buffer_type;
using write_queue_type = std::deque<std::pair<bool, buffer_type>>;
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application)
......@@ -76,15 +72,11 @@ public:
// -- member functions -------------------------------------------------------
error init(endpoint_manager& parent) override {
// call init function from base class
return super::init(parent);
}
bool handle_read_event(endpoint_manager&) override {
auto buf = this->read_buf_.data() + this->collected_;
size_t len = this->read_threshold_ - this->collected_;
CAF_LOG_TRACE(CAF_ARG(this->handle().id) << CAF_ARG(len));
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id)
<< CAF_ARG2("missing", len));
auto ret = read(this->handle_, make_span(buf, len));
// Update state.
if (auto num_bytes = get_if<size_t>(&ret)) {
......@@ -93,7 +85,7 @@ public:
this->collected_ += *num_bytes;
if (this->collected_ >= this->read_threshold_) {
if (auto err = this->next_layer_.handle_data(*this, this->read_buf_)) {
CAF_LOG_WARNING("handle_data failed:" << CAF_ARG(err));
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err));
return false;
}
this->prepare_next_read();
......@@ -101,7 +93,7 @@ public:
} else {
auto err = get<sec>(ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
......@@ -110,6 +102,7 @@ public:
}
bool handle_write_event(endpoint_manager& parent) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id));
// Try to write leftover data.
write_some();
// Get new data from parent.
......@@ -123,6 +116,7 @@ public:
}
void write_packet(id_type, span<buffer_type*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (this->write_queue_.empty())
this->manager().register_writing();
......@@ -145,10 +139,7 @@ private:
void prepare_next_read() {
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_)) {
switch (rd_flag_) {
case net::receive_policy_flag::exactly:
if (this->read_buf_.size() != max_)
this->read_buf_.resize(max_);
......@@ -171,7 +162,7 @@ private:
}
bool write_some() {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id));
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = this->write_queue_.front();
......@@ -216,8 +207,7 @@ private:
return false;
}
std::deque<std::pair<bool, buffer_type>> write_queue_;
write_queue_type write_queue_;
size_t written_;
size_t read_threshold_;
size_t collected_;
......
......@@ -18,24 +18,15 @@
#pragma once
#include "caf/actor_system_config.hpp"
#include "caf/byte.hpp"
#include "caf/detail/overload.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf::net {
/// Implements base class for transports.
template <class Transport, class NextLayer, class Handle, class Application,
class IdType>
class transport_base {
......@@ -88,9 +79,10 @@ public:
return *manager_;
}
// -- member functions -------------------------------------------------------
// -- 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",
......@@ -104,11 +96,8 @@ public:
return none;
}
virtual bool handle_read_event(endpoint_manager&) = 0;
virtual bool handle_write_event(endpoint_manager& parent) = 0;
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);
......@@ -141,10 +130,16 @@ public:
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 ------------------------------------------------------
......@@ -170,14 +165,15 @@ private:
}
protected:
NextLayer next_layer_;
Handle handle_;
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!
// TODO implement retries using this member! Should this go into stream_trans?
// size_t max_consecutive_reads_;
endpoint_manager* manager_;
......
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