Commit e53ff78b authored by Jakob Otto's avatar Jakob Otto

Refactor scribe to stream-transport

parent 23cf2b9c
......@@ -18,7 +18,6 @@ set(LIBCAF_NET_SRCS
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
)
add_custom_target(libcaf_net)
......
......@@ -23,7 +23,7 @@
namespace caf {
namespace net {
template <class Transport, class Application>
template <class Transport>
class endpoint_manager_impl : public endpoint_manager {
public:
// -- member types -----------------------------------------------------------
......@@ -32,15 +32,13 @@ public:
using transport_type = Transport;
using application_type = Application;
using application_type = typename transport_type::application_type;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
Transport trans, Application app)
: super(trans.handle(), parent, sys),
transport_(std::move(trans)),
application_(std::move(app)) {
Transport trans)
: super(trans.handle(), parent, sys), transport_(std::move(trans)) {
// nop
}
......@@ -54,10 +52,6 @@ public:
return transport_;
}
application_type& application() {
return application_;
}
// -- interface functions ----------------------------------------------------
error init() override {
......@@ -91,7 +85,7 @@ public:
}
void handle_error(sec code) override {
transport_.handle_error(application_, code);
transport_.handle_error(code);
}
serialize_fun_type serialize_fun() const noexcept override {
......@@ -100,7 +94,6 @@ public:
private:
transport_type transport_;
application_type application_;
};
} // namespace net
......
......@@ -25,12 +25,11 @@
namespace caf {
namespace net {
template <class Transport, class Application>
template <class Transport>
endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx,
actor_system& sys, Transport trans,
Application app) {
using impl = endpoint_manager_impl<Transport, Application>;
return make_counted<impl>(mpx, sys, std::move(trans), std::move(app));
actor_system& sys, Transport trans) {
using impl = endpoint_manager_impl<Transport>;
return make_counted<impl>(mpx, sys, std::move(trans));
}
} // namespace net
......
......@@ -25,26 +25,49 @@
#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 {
namespace policy {
namespace net {
/// Implements a scribe policy that manages a stream socket.
class scribe {
template <class Application>
class stream_transport {
public:
explicit scribe(net::stream_socket handle);
// -- member types -----------------------------------------------------------
using application_type = Application;
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application)
: worker_(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) {
// nop
}
// -- properties -------------------------------------------------------------
net::stream_socket handle() const noexcept {
stream_socket handle() const noexcept {
return handle_;
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent& parent) {
parent.application().init(parent);
parent.mask_add(net::operation::read);
auto decorator = make_write_packet_decorator(*this, parent);
worker_.init(decorator);
parent.mask_add(operation::read);
return none;
}
......@@ -59,14 +82,15 @@ public:
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes;
if (collected_ >= read_threshold_) {
parent.application().handle_data(*this, read_buf_);
auto decorator = make_write_packet_decorator(*this, parent);
worker_.handle_data(decorator, read_buf_);
prepare_next_read();
}
return true;
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
parent.application().handle_error(err);
worker_.handle_error(err);
return false;
}
}
......@@ -74,25 +98,82 @@ public:
template <class Parent>
bool handle_write_event(Parent& parent) {
// Try to write leftover data.
write_some(parent);
write_some();
// Get new data from parent.
// TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
parent.application().write_message(*this, std::move(msg));
auto decorator = make_write_packet_decorator(*this, parent);
worker_.write_message(decorator, std::move(msg));
}
// Write prepared data.
return write_some(parent);
return write_some();
}
template <class Parent>
bool write_some(Parent& parent) {
void resolve(Parent& parent, const std::string& path, actor listener) {
worker_.resolve(parent, path, listener);
}
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
auto decorator = make_write_packet_decorator(*this, parent);
worker_.timeout(decorator, value, id);
}
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();
}
template <class Parent>
void write_packet(Parent&, span<const byte> header, span<const byte> payload,
unit_t) {
write_buf_.insert(write_buf_.end(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
}
private:
bool write_some() {
if (write_buf_.empty())
return false;
auto len = write_buf_.size() - written_;
auto buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = net::write(handle_, make_span(buf, len));
auto ret = write(handle_, make_span(buf, len));
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
// Update state.
......@@ -105,47 +186,28 @@ public:
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
parent.application().handle_error(err);
worker_.handle_error(err);
return false;
}
return true;
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
parent.application().resolve(parent, path, listener);
}
transport_worker<application_type> worker_;
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
parent.application().timeout(*this, value, id);
}
template <class Application>
void handle_error(Application& application, sec code) {
application.handle_error(code);
}
void prepare_next_read();
void configure_read(net::receive_policy::config cfg);
void write_packet(span<const byte> buf);
private:
net::stream_socket handle_;
stream_socket handle_;
std::vector<byte> read_buf_;
std::vector<byte> write_buf_;
size_t max_consecutive_reads_; // TODO use this field!
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
size_t read_threshold_;
size_t collected_;
size_t max_;
net::receive_policy_flag rd_flag_;
receive_policy_flag rd_flag_;
size_t written_;
};
} // namespace policy
} // namespace net
} // namespace caf
......@@ -18,6 +18,9 @@
#pragma once
#include "caf/byte.hpp"
#include "caf/span.hpp"
namespace caf {
namespace net {
......@@ -26,34 +29,36 @@ namespace net {
template <class Object, class Parent>
class write_packet_decorator {
public:
dispatcher(Object& object, Parent& parent)
using transport_type = typename Parent::transport_type;
using application_type = typename Parent::application_type;
write_packet_decorator(Object& object, Parent& parent)
: object_(object), parent_(parent) {
// nop
}
template <class Header>
void write_packet(const Header& header, span<const byte> payload) {
object_.write_packet(parent_, header, payload);
template <class Header, class... Ts>
void write_packet(const Header& header, span<const byte> payload,
Ts&&... xs) {
object_.write_packet(parent_, header, payload, std::forward<Ts>(xs)...);
}
actor_system& system() {
parent_.system();
return parent_.system();
}
void cancel_timeout(atom_value type, uint64_t id) {
parent_.cancel_timeout(type, id);
}
void set_timeout(timestamp tout, atom_value type, uint64_t id) {
parent_.set_timeout(tout, type, id);
}
typename parent::transport_type& transport() {
return parent_.transport_;
template <class... Ts>
uint64_t set_timeout(timestamp tout, atom_value type, Ts&&... xs) {
return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...);
}
typename parent::application_type& application() {
return parent_.application_;
transport_type& transport() {
return parent_.transport();
}
private:
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/policy/scribe.hpp"
#include <system_error>
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/span.hpp"
namespace caf {
namespace policy {
scribe::scribe(caf::net::stream_socket handle)
: handle_(handle),
max_consecutive_reads_(0),
read_threshold_(1024),
collected_(0),
max_(1024),
rd_flag_(net::receive_policy_flag::exactly),
written_(0) {
// nop
}
void scribe::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 scribe::configure_read(net::receive_policy::config cfg) {
rd_flag_ = cfg.first;
max_ = cfg.second;
prepare_next_read();
}
void scribe::write_packet(span<const byte> buf) {
write_buf_.insert(write_buf_.end(), buf.begin(), buf.end());
}
} // namespace policy
} // namespace caf
......@@ -72,6 +72,8 @@ public:
class dummy_transport {
public:
using application_type = dummy_application;
dummy_transport(stream_socket handle, std::shared_ptr<std::vector<byte>> data)
: handle_(handle), data_(data), read_buf_(1024) {
// nop
......@@ -114,8 +116,7 @@ public:
return get<sec>(res) == sec::unavailable_or_would_block;
}
template <class Manager>
void handle_error(Manager&, sec) {
void handle_error(sec) {
// nop
}
......@@ -160,8 +161,7 @@ CAF_TEST(send and receive) {
sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf},
dummy_application{});
dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
......@@ -184,8 +184,7 @@ CAF_TEST(resolve and proxy communication) {
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf},
dummy_application{});
dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
......
......@@ -18,7 +18,7 @@
#define CAF_SUITE scribe_policy
#include "caf/policy/scribe.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/test/dsl.hpp"
......@@ -75,7 +75,7 @@ public:
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(msg->payload);
transport.write_packet(span<byte>{}, msg->payload);
}
template <class Parent>
......@@ -123,6 +123,7 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
using transport_type = stream_transport<dummy_application>;
std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>();
......@@ -132,9 +133,9 @@ CAF_TEST(receive) {
sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
CAF_MESSAGE("configure scribe_policy");
policy::scribe scribe{sockets.first};
scribe.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, scribe, dummy_application{buf});
transport_type transport{sockets.first, dummy_application{buf}};
transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, transport);
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
......@@ -148,13 +149,15 @@ CAF_TEST(receive) {
}
CAF_TEST(resolve and proxy communication) {
using transport_type = stream_transport<dummy_application>;
std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first},
dummy_application{buf});
auto mgr = make_endpoint_manager(mpx, sys,
transport_type{sockets.first,
dummy_application{buf}});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
......
......@@ -18,7 +18,7 @@
#define CAF_SUITE string_application
#include "caf/policy/scribe.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/test/dsl.hpp"
......@@ -100,12 +100,11 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
std::vector<byte> buf;
header_type header{static_cast<uint32_t>(msg->payload.size())};
buf.resize(sizeof(header_type));
memcpy(buf.data(), &header, buf.size());
buf.insert(buf.end(), msg->payload.begin(), msg->payload.end());
parent.write_packet(buf);
std::vector<byte> header_buf(sizeof(header_type));
std::vector<byte> payload(msg->payload.begin(), msg->payload.end());
memcpy(header_buf.data(), &header, header_buf.size());
parent.write_packet(make_span(header_buf), make_span(payload));
}
static expected<std::vector<byte>> serialize(actor_system& sys,
......@@ -152,7 +151,8 @@ public:
if (header_.payload == 0)
Base::handle_packet(parent, header_, span<const byte>{});
else
parent.configure_read(net::receive_policy::exactly(header_.payload));
parent.transport().configure_read(
net::receive_policy::exactly(header_.payload));
await_payload_ = true;
}
}
......@@ -191,6 +191,7 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
using application_type = extend<string_application>::with<
stream_string_application>;
using transport_type = stream_transport<application_type>;
std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>();
......@@ -199,13 +200,15 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first},
application_type{sys, buf});
auto mgr1 = make_endpoint_manager(mpx, sys,
transport_type{sockets.first,
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr1->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto mgr2 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.second},
application_type{sys, buf});
auto mgr2 = make_endpoint_manager(mpx, sys,
transport_type{sockets.second,
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr2->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
......
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