Unverified Commit 9ccb5dc6 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #9

Add scribe transport-policy
parents d6b50b06 8cb76f23
......@@ -18,6 +18,7 @@ set(LIBCAF_NET_SRCS
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
)
add_custom_target(libcaf_net)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
namespace caf {
namespace net {
/// Implements the interface for transport and application policies and
/// dispatches member functions either to `decorator` or `fallback`.
template <class Decorator, class Fallback>
class dispatcher {
public:
dispatcher(Decorator& decorator, Fallback& fallback)
: decorator_(decorator), fallback_(fallback) {
// nop
}
template <class Header>
void write_packet(const Header& header, span<const byte> payload) {
decltype(write_packet_delegate_test<Decorator>(nullptr, header, payload)) x;
write_packet_delegate(x, header, payload);
}
actor_system& system() {
fallback_.system();
}
void cancel_timeout(atom_value type, uint64_t id) {
fallback_.cancel_timeout(type, id);
}
void set_timeout(timestamp tout, atom_value type, uint64_t id) {
fallback_.set_timeout(tout, type, id);
}
typename Fallback::transport_type& transport() {
return fallback_.transport_;
}
typename Fallback::application_type& application() {
return fallback_.application_;
}
private:
template <class U, class Header>
static auto write_packet_delegate_test(U* x, const Header& hdr,
span<const byte> payload)
-> decltype(x->write_packet(hdr, payload), std::true_type());
template <class U>
static auto write_packet_delegate_test(...) -> std::false_type;
template <class Header>
void write_packet_delegate(std::true_type, const Header& header,
span<const byte> payload) {
decorator_.write_packet(header, payload);
}
template <class Header>
void write_packet_delegate(std::false_type, const Header& header,
span<const byte> payload) {
fallback_.write_packet(header, payload);
}
Object& decorator_;
Fallback& fallback_;
};
template <class Decorator, class Fallback>
dispatcher<Decorator, Fallback> make_dispatcher(Decorator& decorator,
Fallback& fallback) {
return {decorator, fallback};
}
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <cstddef>
#include <string>
#include <utility>
#include "caf/config.hpp"
namespace caf {
namespace net {
enum class receive_policy_flag : unsigned { at_least, at_most, exactly };
inline std::string to_string(receive_policy_flag x) {
return x == receive_policy_flag::at_least
? "at_least"
: (x == receive_policy_flag::at_most ? "at_most" : "exactly");
}
class receive_policy {
public:
receive_policy() = delete;
using config = std::pair<receive_policy_flag, size_t>;
static config at_least(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::at_least, num_bytes};
}
static config at_most(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::at_most, num_bytes};
}
static config exactly(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::exactly, num_bytes};
}
};
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/error.hpp>
#include <caf/fwd.hpp>
#include <caf/logger.hpp>
#include <caf/net/endpoint_manager.hpp>
#include <caf/net/receive_policy.hpp>
#include <caf/net/stream_socket.hpp>
#include <caf/sec.hpp>
#include <caf/span.hpp>
#include <caf/variant.hpp>
namespace caf {
namespace policy {
/// Implements a scribe policy that manages a stream socket.
class scribe {
public:
explicit scribe(net::stream_socket handle);
net::stream_socket handle() const noexcept {
return handle_;
}
template <class Parent>
error init(Parent& parent) {
parent.application().init(parent);
parent.mask_add(net::operation::read);
return none;
}
template <class Parent>
bool handle_read_event(Parent& parent) {
void* 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_, buf, len);
// Update state.
if (auto num_bytes = get_if<size_t>(&ret)) {
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_);
prepare_next_read();
}
return true;
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
parent.application().handle_error(err);
return false;
}
}
template <class Parent>
bool handle_write_event(Parent& parent) {
// Try to write leftover data.
write_some(parent);
// Get new data from parent.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
parent.application().write_message(*this, std::move(msg));
}
// Write prepared data.
return write_some(parent);
}
template <class Parent>
bool write_some(Parent& parent) {
if (write_buf_.empty())
return false;
auto len = write_buf_.size() - written_;
void* buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = net::write(handle_, 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.
written_ += *num_bytes;
if (written_ >= write_buf_.size()) {
written_ = 0;
write_buf_.clear();
return false;
}
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
parent.application().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);
}
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<char> buf);
private:
net::stream_socket handle_;
std::vector<char> read_buf_;
std::vector<char> write_buf_;
size_t max_consecutive_reads_;
size_t read_threshold_;
size_t collected_;
size_t max_;
net::receive_policy_flag rd_flag_;
size_t written_;
};
} // namespace policy
} // 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. *
******************************************************************************/
#include "caf/policy/scribe.hpp"
#include <system_error>
#include "caf/config.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<char> buf) {
write_buf_.insert(write_buf_.end(), buf.begin(), buf.end());
}
} // namespace policy
} // 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. *
******************************************************************************/
#define CAF_SUITE scribe_policy
#include "caf/policy/scribe.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
using namespace caf;
using namespace caf::net;
namespace {
constexpr string_view hello_manager = "hello manager!";
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
multiplexer_ptr mpx;
};
class dummy_application {
public:
dummy_application(std::shared_ptr<std::vector<char>> rec_buf)
: rec_buf_(std::move(rec_buf)){
// nop
};
~dummy_application() = default;
template <class Parent>
error init(Parent&) {
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(msg->payload);
}
template <class Parent>
void handle_data(Parent&, span<char> data) {
rec_buf_->clear();
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
}
template <class Manager>
void resolve(Manager& manager, const std::string& path, actor listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid));
actor_config cfg;
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&manager.system(),
cfg, &manager);
anon_send(listener, resolve_atom::value, std::move(path), p);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
static expected<std::vector<char>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
private:
std::shared_ptr<std::vector<char>> rec_buf_;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
std::vector<char> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<char>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
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});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_MESSAGE("sending data to scribe_policy");
CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(),
hello_manager.size()),
hello_manager.size());
run();
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), hello_manager);
}
CAF_TEST(resolve and proxy communication) {
std::vector<char> read_buf(1024);
auto buf = std::make_shared<std::vector<char>>();
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});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
mgr->resolve("/id/42", self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
CAF_MESSAGE("got a proxy, send a message to it");
self->send(actor_cast<actor>(p), "hello proxy!");
},
after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); });
run();
auto read_res = read(sockets.second, read_buf.data(), read_buf.size());
if (!holds_alternative<size_t>(read_res))
CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res)));
read_buf.resize(get<size_t>(read_res));
CAF_MESSAGE("receive buffer contains " << read_buf.size() << " bytes");
message msg;
binary_deserializer source{sys, read_buf};
CAF_CHECK_EQUAL(source(msg), none);
if (msg.match_elements<std::string>())
CAF_CHECK_EQUAL(msg.get_as<std::string>(0), "hello proxy!");
else
CAF_ERROR("expected a string, got: " << to_string(msg));
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE string_application
#include "caf/net/endpoint_manager.hpp"
#include <caf/policy/scribe.hpp>
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
using namespace caf;
using namespace caf::net;
using namespace caf::policy;
namespace {
string_view hello_manager{"hello manager!"};
string_view hello_test{"hello test!"};
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
multiplexer_ptr mpx;
};
struct string_application_header {
uint32_t payload;
};
/// @relates header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
string_application_header& hdr) {
return f(meta::type_name("sa_header"), hdr.payload);
}
class string_application {
public:
using header_type = string_application_header;
string_application(actor_system& sys, std::shared_ptr<std::vector<char>> buf)
: sys_(sys), buf_(std::move(buf)) {
// nop
}
template <class Parent>
error init(Parent&) {
return none;
}
template <class Parent>
void handle_packet(Parent&, header_type&, span<char> payload) {
binary_deserializer source{sys_, payload.data(), payload.size()};
message msg;
if (auto err = msg.load(source))
CAF_FAIL("unable to deserialize message: " << err);
if (!msg.match_elements<std::string>())
CAF_FAIL("unexpected message: " << msg);
auto& str = msg.get_as<std::string>(0);
buf_->insert(buf_->end(), str.begin(), str.end());
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
std::vector<char> 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);
}
static expected<std::vector<char>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
private:
actor_system& sys_;
std::shared_ptr<std::vector<char>> buf_;
};
class stream_string_application : public string_application {
public:
stream_string_application(actor_system& sys,
std::shared_ptr<std::vector<char>> buf)
: string_application(sys, std::move(buf)), await_payload_(false) {
// nop
}
template <class Parent>
error init(Parent& parent) {
parent.transport().configure_read(
net::receive_policy::exactly(sizeof(header_type)));
return string_application::init(parent);
}
template <class Parent>
void handle_data(Parent& parent, span<char> data) {
if (await_payload_) {
handle_packet(parent, header_, data);
await_payload_ = false;
} else {
if (data.size() != sizeof(header_type))
CAF_FAIL("");
memcpy(&header_, data.data(), sizeof(header_type));
if (header_.payload == 0)
handle_packet(parent, header_, span<char>{});
else
parent.configure_read(net::receive_policy::exactly(header_.payload));
await_payload_ = true;
}
}
template <class Manager>
void resolve(Manager& manager, const std::string& path, actor listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(aid, hid));
actor_config cfg;
auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid,
&manager
.system(),
cfg, &manager);
anon_send(listener, resolve_atom::value, std::move(path), p);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
private:
header_type header_;
bool await_payload_;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
std::vector<char> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<char>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first},
stream_string_application{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},
stream_string_application{sys, buf});
CAF_CHECK_EQUAL(mgr2->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
CAF_MESSAGE("resolve actor-proxy");
mgr1->resolve("/id/42", self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
CAF_MESSAGE("got a proxy, send a message to it");
self->send(actor_cast<actor>(p), "hello proxy!");
},
after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); });
run();
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), "hello proxy!");
}
CAF_TEST_FIXTURE_SCOPE_END()
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