Commit 03756ae4 authored by Jakob Otto's avatar Jakob Otto

Refactor stream_transport

parent c4e5343e
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/error.hpp" #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/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
...@@ -200,12 +201,17 @@ public: ...@@ -200,12 +201,17 @@ public:
prepare_next_read(); prepare_next_read();
} }
void write_packet(span<const byte> header, span<const byte> payload, void write_packet(unit_t, span<buffer_type*> buffers) {
typename worker_type::id_type) { // Sanity check
if (write_buf_.empty()) CAF_ASSERT(!buffers.empty());
auto it = buffers.begin();
if (write_queue_.empty())
manager().register_writing(); manager().register_writing();
write_buf_.insert(write_buf_.end(), header.begin(), header.end()); // move header by itself to keep things sorted.
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end()); write_queue_.emplace_back(true, std::move(**it++));
// payload buffers. just write them
for (; it != buffers.end(); ++it)
write_queue_.emplace_back(false, std::move(**it));
} }
// -- buffer recycling ------------------------------------------------------- // -- buffer recycling -------------------------------------------------------
...@@ -232,30 +238,50 @@ private: ...@@ -232,30 +238,50 @@ private:
// -- private member functions ----------------------------------------------- // -- private member functions -----------------------------------------------
bool write_some() { bool write_some() {
if (write_buf_.empty()) auto begin = [&]() { return write_queue_.begin(); };
// helper to sort empty buffers back into the right queues
auto recycle = [&]() {
auto is_header = [](std::pair<bool, buffer_type>& p) { return p.first; };
begin()->second.clear();
if (is_header(*begin()))
free_header_bufs_.emplace_back(std::move(begin()->second));
else
free_bufs_.emplace_back(std::move(begin()->second));
write_queue_.pop_front();
};
// nothing to write
if (write_queue_.empty())
return false; return false;
auto len = write_buf_.size() - written_; do {
auto buf = write_buf_.data() + written_; if (begin()->second.empty()) {
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len)); recycle();
auto ret = write(handle_, make_span(buf, len)); continue;
if (auto num_bytes = get_if<size_t>(&ret)) { }
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); // get size of send buffer
// Update state. auto ret = send_buffer_size(handle_);
written_ += *num_bytes; if (!ret) {
if (written_ >= write_buf_.size()) { CAF_LOG_ERROR("send_buffer_size returned an error" << CAF_ARG(ret));
written_ = 0;
write_buf_.clear();
return false; return false;
} }
// is send buffer of socket full?
if (begin()->second.size() > *ret)
return true;
CAF_LOG_TRACE(CAF_ARG(handle_.id));
auto write_ret = write(handle_, make_span(begin()->second));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
if (*num_bytes >= begin()->second.size())
recycle();
} else { } else {
auto err = get<sec>(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); worker_.handle_error(err);
return false; return false;
} }
} }
return true; } while (!write_queue_.empty());
return false;
} }
worker_type worker_; worker_type worker_;
...@@ -265,7 +291,7 @@ private: ...@@ -265,7 +291,7 @@ private:
std::deque<buffer_type> free_bufs_; std::deque<buffer_type> free_bufs_;
buffer_type read_buf_; buffer_type read_buf_;
buffer_type write_buf_; std::deque<std::pair<bool, buffer_type>> write_queue_;
// TODO implement retries using this member! // TODO implement retries using this member!
// size_t max_consecutive_reads_; // size_t max_consecutive_reads_;
......
...@@ -71,10 +71,11 @@ public: ...@@ -71,10 +71,11 @@ public:
return none; return none;
} }
template <class Transport> template <class Parent>
void write_message(Transport& transport, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> ptr) {
transport.write_packet(span<byte>{}, msg->payload); auto header_buf = parent.next_header_buffer();
parent.write_packet(header_buf, ptr->payload);
} }
template <class Parent> template <class Parent>
......
...@@ -43,6 +43,8 @@ using namespace caf::policy; ...@@ -43,6 +43,8 @@ using namespace caf::policy;
namespace { namespace {
using buffer_type = std::vector<byte>;
struct fixture : test_coordinator_fixture<>, host_fixture { struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() { fixture() {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
...@@ -98,13 +100,15 @@ public: ...@@ -98,13 +100,15 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> ptr) {
// Ignore proxy announcement messages. // Ignore proxy announcement messages.
if (msg->msg == nullptr) if (ptr->msg == nullptr)
return; return;
header_type header{static_cast<uint32_t>(msg->payload.size())}; auto header_buf = parent.next_header_buffer();
std::vector<byte> payload(msg->payload.begin(), msg->payload.end()); serializer_impl<buffer_type> sink{sys_, header_buf};
parent.write_packet(as_bytes(make_span(&header, 1)), make_span(payload)); header_type header{static_cast<uint32_t>(ptr->payload.size())};
sink(header);
parent.write_packet(header_buf, ptr->payload);
} }
static expected<std::vector<byte>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
...@@ -147,7 +151,8 @@ public: ...@@ -147,7 +151,8 @@ public:
} else { } else {
if (data.size() != sizeof(header_type)) if (data.size() != sizeof(header_type))
CAF_FAIL(""); CAF_FAIL("");
memcpy(&header_, data.data(), sizeof(header_type)); binary_deserializer source{nullptr, data};
source(header_);
if (header_.payload == 0) if (header_.payload == 0)
Base::handle_packet(parent, header_, span<const byte>{}); Base::handle_packet(parent, header_, span<const byte>{});
else else
...@@ -187,8 +192,8 @@ public: ...@@ -187,8 +192,8 @@ public:
// nop // nop
} }
void handle_error(sec) { void handle_error(sec sec) {
// nop CAF_FAIL("handle_error called: " << CAF_ARG(sec));
} }
private: private:
......
...@@ -37,6 +37,8 @@ using namespace caf::net; ...@@ -37,6 +37,8 @@ using namespace caf::net;
namespace { namespace {
using buffer_type = std::vector<byte>;
constexpr string_view hello_test = "hello test!"; constexpr string_view hello_test = "hello test!";
struct application_result { struct application_result {
...@@ -72,7 +74,8 @@ public: ...@@ -72,7 +74,8 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
parent.write_packet(span<const byte>{}, msg->payload); auto header_buffer = parent.next_header_buffer();
parent.write_packet(header_buffer, msg->payload);
} }
template <class Parent> template <class Parent>
...@@ -118,16 +121,17 @@ public: ...@@ -118,16 +121,17 @@ public:
using application_type = dummy_application; using application_type = dummy_application;
dummy_transport(std::shared_ptr<transport_result> res) : res_(res) { dummy_transport(std::shared_ptr<transport_result> res)
: res_(std::move(res)) {
// nop // nop
} }
void write_packet(span<const byte> header, span<const byte> payload, void write_packet(ip_endpoint ep, span<buffer_type*> buffers) {
ip_endpoint ep) {
auto& buf = res_->packet_buffer;
buf.insert(buf.begin(), header.begin(), header.end());
buf.insert(buf.begin(), payload.begin(), payload.end());
res_->ep = ep; res_->ep = ep;
auto& packet_buf = res_->packet_buffer;
packet_buf.clear();
for (auto buf : buffers)
packet_buf.insert(packet_buf.end(), buf->begin(), buf->end());
} }
transport_type& transport() { transport_type& transport() {
......
...@@ -33,6 +33,8 @@ using namespace caf::net; ...@@ -33,6 +33,8 @@ using namespace caf::net;
namespace { namespace {
using buffer_type = std::vector<byte>;
constexpr string_view hello_test = "hello_test"; constexpr string_view hello_test = "hello_test";
struct dummy_actor : public monitorable_actor { struct dummy_actor : public monitorable_actor {
...@@ -47,7 +49,7 @@ struct dummy_actor : public monitorable_actor { ...@@ -47,7 +49,7 @@ struct dummy_actor : public monitorable_actor {
class dummy_application { class dummy_application {
public: public:
dummy_application(std::shared_ptr<std::vector<byte>> rec_buf, uint8_t id) dummy_application(std::shared_ptr<buffer_type> rec_buf, uint8_t id)
: rec_buf_(std::move(rec_buf)), : rec_buf_(std::move(rec_buf)),
id_(id){ id_(id){
// nop // nop
...@@ -61,11 +63,12 @@ public: ...@@ -61,11 +63,12 @@ public:
return none; return none;
} }
template <class Transport> template <class Parent>
void write_message(Transport& transport, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
rec_buf_->push_back(static_cast<byte>(id_)); rec_buf_->push_back(static_cast<byte>(id_));
transport.write_packet(span<byte>{}, make_span(msg->payload)); auto header_buf = parent.next_header_buffer();
parent.write_packet(header_buf, msg->payload);
} }
template <class Parent> template <class Parent>
...@@ -88,13 +91,13 @@ public: ...@@ -88,13 +91,13 @@ public:
rec_buf_->push_back(static_cast<byte>(id_)); rec_buf_->push_back(static_cast<byte>(id_));
} }
static expected<std::vector<byte>> serialize(actor_system&, static expected<buffer_type> serialize(actor_system&,
const type_erased_tuple&) { const type_erased_tuple&) {
return std::vector<byte>{}; return buffer_type{};
} }
private: private:
std::shared_ptr<std::vector<byte>> rec_buf_; std::shared_ptr<buffer_type> rec_buf_;
uint8_t id_; uint8_t id_;
}; };
...@@ -102,8 +105,8 @@ struct dummy_application_factory { ...@@ -102,8 +105,8 @@ struct dummy_application_factory {
public: public:
using application_type = dummy_application; using application_type = dummy_application;
dummy_application_factory(std::shared_ptr<std::vector<byte>> buf) dummy_application_factory(std::shared_ptr<buffer_type> buf)
: buf_(buf), application_cnt_(0) { : buf_(std::move(buf)), application_cnt_(0) {
// nop // nop
} }
...@@ -112,7 +115,7 @@ public: ...@@ -112,7 +115,7 @@ public:
} }
private: private:
std::shared_ptr<std::vector<byte>> buf_; std::shared_ptr<buffer_type> buf_;
uint8_t application_cnt_; uint8_t application_cnt_;
}; };
...@@ -121,34 +124,35 @@ struct dummy_transport { ...@@ -121,34 +124,35 @@ struct dummy_transport {
using application_type = dummy_application; using application_type = dummy_application;
dummy_transport(std::shared_ptr<std::vector<byte>> buf) : buf_(buf) { dummy_transport(std::shared_ptr<buffer_type> buf) : buf_(std::move(buf)) {
// nop
} }
template <class IdType> template <class IdType>
void write_packet(span<const byte> header, span<const byte> payload, IdType) { void write_packet(IdType, span<buffer_type*> buffers) {
buf_->insert(buf_->end(), header.begin(), header.end()); for (auto buf : buffers)
buf_->insert(buf_->end(), payload.begin(), payload.end()); buf_->insert(buf_->end(), buf->begin(), buf->end());
} }
transport_type& transport() { transport_type& transport() {
return *this; return *this;
} }
std::vector<byte> next_buffer() { buffer_type next_buffer() {
return {}; return {};
} }
std::vector<byte> next_header_buffer() { buffer_type next_header_buffer() {
return {}; return {};
} }
private: private:
std::shared_ptr<std::vector<byte>> buf_; std::shared_ptr<buffer_type> buf_;
}; };
struct testdata { struct testdata {
testdata(uint8_t worker_id, node_id id, ip_endpoint ep) testdata(uint8_t worker_id, node_id id, ip_endpoint ep)
: worker_id(worker_id), nid(id), ep(ep) { : worker_id(worker_id), nid(std::move(id)), ep(ep) {
// nop // nop
} }
...@@ -180,7 +184,7 @@ struct fixture : host_fixture { ...@@ -180,7 +184,7 @@ struct fixture : host_fixture {
ip_endpoint>; ip_endpoint>;
fixture() fixture()
: buf{std::make_shared<std::vector<byte>>()}, : buf{std::make_shared<buffer_type>()},
dispatcher{dummy_application_factory{buf}}, dispatcher{dummy_application_factory{buf}},
dummy{buf} { dummy{buf} {
add_new_workers(); add_new_workers();
...@@ -192,7 +196,7 @@ struct fixture : host_fixture { ...@@ -192,7 +196,7 @@ struct fixture : host_fixture {
actor_config cfg; actor_config cfg;
auto p = make_actor<dummy_actor, strong_actor_ptr>(aid, nid, &sys, cfg); auto p = make_actor<dummy_actor, strong_actor_ptr>(aid, nid, &sys, cfg);
auto test_span = as_bytes(make_span(hello_test)); auto test_span = as_bytes(make_span(hello_test));
std::vector<byte> payload(test_span.begin(), test_span.end()); buffer_type payload(test_span.begin(), test_span.end());
auto strong_actor = actor_cast<strong_actor_ptr>(p); auto strong_actor = actor_cast<strong_actor_ptr>(p);
mailbox_element::forwarding_stack stack; mailbox_element::forwarding_stack stack;
auto elem = make_mailbox_element(std::move(strong_actor), auto elem = make_mailbox_element(std::move(strong_actor),
...@@ -225,7 +229,7 @@ struct fixture : host_fixture { ...@@ -225,7 +229,7 @@ struct fixture : host_fixture {
actor_system_config cfg{}; actor_system_config cfg{};
actor_system sys{cfg}; actor_system sys{cfg};
std::shared_ptr<std::vector<byte>> buf; std::shared_ptr<buffer_type> buf;
dispatcher_type dispatcher; dispatcher_type dispatcher;
dummy_transport dummy; dummy_transport dummy;
......
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