Commit b11f1ae2 authored by Dominik Charousset's avatar Dominik Charousset

Update to latest CAF master

parent 9d7a9c4b
......@@ -50,7 +50,7 @@ if(CAF_INC_ENABLE_STANDALONE_BUILD)
FetchContent_Declare(
actor_framework
GIT_REPOSITORY https://github.com/actor-framework/actor-framework.git
GIT_TAG b3c59ff
GIT_TAG d7f70e997
)
FetchContent_Populate(actor_framework)
set(CAF_ENABLE_EXAMPLES OFF CACHE BOOL "" FORCE)
......
......@@ -67,6 +67,13 @@ public:
explicit application(proxy_registry& proxies);
// -- static utility functions -----------------------------------------------
static auto default_app_ids() {
return std::vector<std::string>{
to_string(defaults::middleman::app_identifier)};
}
// -- interface functions ----------------------------------------------------
template <class Parent>
......@@ -79,8 +86,12 @@ public:
// Allow unit tests to run the application without endpoint manager.
if (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager();
auto workers = get_or(system_->config(), "middleman.workers",
defaults::middleman::workers);
size_t workers;
if (auto workers_cfg = get_if<size_t>(&system_->config(),
"middleman.workers"))
workers = *workers_cfg;
else
workers = std::min(3u, std::thread::hardware_concurrency() / 4u) + 1;
for (size_t i = 0; i < workers; ++i)
hub_->add_new_worker(*queue_, proxies_);
// Write handshake.
......
......@@ -22,7 +22,7 @@
#include <string>
#include "caf/detail/net_export.hpp"
#include "caf/error_category.hpp"
#include "caf/is_error_code_enum.hpp"
namespace caf::net::basp {
......@@ -49,11 +49,4 @@ CAF_NET_EXPORT std::string to_string(ec x);
} // namespace caf::net::basp
namespace caf {
template <>
struct error_category<net::basp::ec> {
static constexpr uint8_t value = 4;
};
} // namespace caf
CAF_ERROR_CODE_ENUM(caf::net::basp::ec)
......@@ -69,7 +69,7 @@ public:
bool handle_read_event(Parent& parent) {
auto x = net::accept(acceptor_);
if (!x) {
CAF_LOG_ERROR("accept failed:" << parent.system().render(x.error()));
CAF_LOG_ERROR("accept failed:" << x.error());
return false;
}
auto mpx = parent.multiplexer();
......
......@@ -38,6 +38,8 @@ class transport_worker;
template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher;
enum class ec : uint8_t;
// -- classes ------------------------------------------------------------------
class endpoint_manager;
......@@ -66,3 +68,17 @@ using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
} // namespace caf::net
namespace caf::net::basp {
enum class ec : uint8_t;
} // namespace caf::net::basp
CAF_BEGIN_TYPE_ID_BLOCK(net_module, detail::net_module_begin)
CAF_ADD_TYPE_ID(net_module, (caf::net::basp::ec))
CAF_END_TYPE_ID_BLOCK(net_module)
static_assert(caf::id_block::net_module::end == caf::detail::net_module_end);
......@@ -43,19 +43,17 @@ expected<std::pair<pipe_socket, pipe_socket>> CAF_NET_EXPORT make_pipe();
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @param buf Memory region for reading the message to send.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
variant<size_t, sec> CAF_NET_EXPORT write(pipe_socket x, span<const byte> buf);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @param buf Memory region for storing the received bytes.
/// @returns The number of received bytes on success, otherwise an error code.
/// @relates pipe_socket
variant<size_t, sec> CAF_NET_EXPORT read(pipe_socket x, span<byte>);
variant<size_t, sec> CAF_NET_EXPORT read(pipe_socket x, span<byte> buf);
/// Converts the result from I/O operation on a ::pipe_socket to either an
/// error code or a non-zero positive integer.
......
......@@ -84,7 +84,7 @@ public:
virtual bool handle_write_event() = 0;
/// Called when the remote side becomes unreachable due to an error.
/// @param reason The error code as reported by the operating system.
/// @param code The error code as reported by the operating system.
virtual void handle_error(sec code) = 0;
protected:
......
......@@ -33,9 +33,9 @@ struct CAF_NET_EXPORT udp_datagram_socket : network_socket {
};
/// Creates a `udp_datagram_socket` bound to given port.
/// @param node ip_endpoint that contains the port to bind to. Pass port '0' to
/// bind to any unused port - The endpoint will be updated with the specific
/// port that was bound.
/// @param ep ip_endpoint that contains the port to bind to. Pass port '0' to
/// bind to any unused port - The endpoint will be updated with the
/// specific port that was bound.
/// @returns The connected socket or an error.
/// @relates udp_datagram_socket
expected<std::pair<udp_datagram_socket, uint16_t>> CAF_NET_EXPORT
......
......@@ -41,8 +41,7 @@ void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
if (auto payload = sf_(home_system(), what->content()))
dst_->enqueue(std::move(what), ctrl(), std::move(*payload));
else
CAF_LOG_ERROR(
"unable to serialize payload: " << home_system().render(payload.error()));
CAF_LOG_ERROR("unable to serialize payload: " << payload.error());
}
void actor_proxy_impl::kill_proxy(execution_unit* ctx, error rsn) {
......
......@@ -232,7 +232,7 @@ error application::handle_handshake(packet_writer&, header hdr,
if (!peer_id || app_ids.empty())
return ec::invalid_handshake;
auto ids = get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers);
basp::application::default_app_ids());
auto predicate = [=](const std::string& x) {
return std::find(ids.begin(), ids.end(), x) != ids.end();
};
......@@ -391,7 +391,7 @@ error application::generate_handshake(std::vector<byte>& buf) {
binary_serializer sink{&executor_, buf};
return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers));
application::default_app_ids()));
}
} // namespace caf::net::basp
......@@ -83,7 +83,7 @@ test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
auto mgr = make_endpoint_manager(mpx, mm_.system(),
transport_type{second, std::move(app)});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << mm_.system().render(err));
CAF_LOG_ERROR("mgr->init() failed: " << err);
CAF_RAISE_ERROR("mgr->init() failed");
}
mpx->register_reading(mgr);
......@@ -98,8 +98,7 @@ test::peer_entry& test::get_peer(const node_id& id) {
return i->second;
auto sockets = make_stream_socket_pair();
if (!sockets) {
CAF_LOG_ERROR("make_stream_socket_pair failed: "
<< mm_.system().render(sockets.error()));
CAF_LOG_ERROR("make_stream_socket_pair failed: " << sockets.error());
CAF_RAISE_ERROR("make_stream_socket_pair failed");
}
return emplace(id, sockets->first, sockets->second);
......
......@@ -20,6 +20,7 @@
#include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/init_global_meta_objects.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
......@@ -30,7 +31,7 @@
namespace caf::net {
void middleman::init_global_meta_objects() {
// nop
caf::init_global_meta_objects<id_block::net_module>();
}
middleman::middleman(actor_system& sys) : sys_(sys) {
......@@ -68,7 +69,7 @@ void middleman::stop() {
void middleman::init(actor_system_config& cfg) {
if (auto err = mpx_->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << system().render(err));
CAF_LOG_ERROR("mgr->init() failed: " << err);
CAF_RAISE_ERROR("mpx->init() failed");
}
if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) {
......@@ -79,7 +80,7 @@ void middleman::init(actor_system_config& cfg) {
}
for (auto& backend : backends_)
if (auto err = backend->init()) {
CAF_LOG_ERROR("failed to initialize backend: " << system().render(err));
CAF_LOG_ERROR("failed to initialize backend: " << err);
CAF_RAISE_ERROR("failed to initialize backend");
}
}
......
......@@ -100,7 +100,7 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
<< " bytes");
ip_endpoint ep;
if (auto err = detail::convert(addr, ep)) {
CAF_ASSERT(err.category() == error_category<sec>::value);
CAF_ASSERT(err.category() == type_id_v<sec>);
return static_cast<sec>(err.code());
}
return std::pair<size_t, ip_endpoint>(*num_bytes, ep);
......
......@@ -38,7 +38,7 @@ using namespace caf::net;
#define REQUIRE_OK(statement) \
if (auto err = statement) \
CAF_FAIL("failed to serialize data: " << sys.render(err));
CAF_FAIL("failed to serialize data: " << err);
namespace {
......@@ -71,7 +71,7 @@ struct fixture : test_coordinator_fixture<>,
void handle_handshake() {
CAF_CHECK_EQUAL(app.state(),
basp::connection_state::await_handshake_header);
auto payload = to_buf(mars, defaults::middleman::app_identifiers);
auto payload = to_buf(mars, basp::application::default_app_ids());
set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()),
basp::version});
......@@ -93,7 +93,7 @@ struct fixture : test_coordinator_fixture<>,
binary_deserializer source{sys, output};
source.skip(basp::header_size);
if (auto err = source(nid, app_ids))
CAF_FAIL("unable to deserialize payload: " << sys.render(err));
CAF_FAIL("unable to deserialize payload: " << err);
if (source.remaining() > 0)
CAF_FAIL("trailing bytes after reading payload");
output.clear();
......@@ -159,11 +159,9 @@ protected:
auto payload = to_buf(__VA_ARGS__); \
set_input(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
if (auto err = app.handle_data(*this, input)) \
CAF_FAIL("application-under-test failed to process header: " \
<< sys.render(err)); \
CAF_FAIL("application-under-test failed to process header: " << err); \
if (auto err = app.handle_data(*this, payload)) \
CAF_FAIL("application-under-test failed to process payload: " \
<< sys.render(err)); \
CAF_FAIL("application-under-test failed to process payload: " << err); \
} while (false)
#define RECEIVE(msg_type, op_data, ...) \
......@@ -171,7 +169,7 @@ protected:
binary_deserializer source{sys, output}; \
basp::header hdr; \
if (auto err = source(hdr, __VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
CAF_FAIL("failed to receive data: " << err); \
if (source.remaining() != 0) \
CAF_FAIL("unable to read entire message, " << source.remaining() \
<< " bytes left in buffer"); \
......
......@@ -54,7 +54,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() : shared_buf(std::make_shared<buffer_type>(1024)) {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
CAF_FAIL("mpx->init failed: " << err);
mpx->set_thread_id();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto addresses = local_addresses("localhost");
......@@ -203,7 +203,7 @@ CAF_TEST_FIXTURE_SCOPE(datagram_transport_tests, fixture)
CAF_TEST(receive) {
using transport_type = datagram_transport<dummy_application_factory>;
if (auto err = nonblocking(recv_socket, true))
CAF_FAIL("nonblocking() returned an error: " << sys.render(err));
CAF_FAIL("nonblocking() returned an error: " << err);
auto mgr = make_endpoint_manager(mpx, sys,
transport_type{recv_socket,
dummy_application_factory{
......
......@@ -43,7 +43,7 @@ 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));
CAF_FAIL("mpx->init failed: " << err);
mpx->set_thread_id();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auth.port = 0;
......
......@@ -48,7 +48,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
mpx = std::make_shared<multiplexer>();
mpx->set_thread_id();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
CAF_FAIL("mpx->init failed: " << err);
if (mpx->num_socket_managers() != 1)
CAF_FAIL("mpx->num_socket_managers() != 1");
}
......@@ -214,7 +214,7 @@ CAF_TEST(resolve and proxy communication) {
run();
auto read_res = read(sockets.second, read_buf);
if (!holds_alternative<size_t>(read_res)) {
CAF_ERROR("read() returned an error: " << sys.render(get<sec>(read_res)));
CAF_ERROR("read() returned an error: " << get<sec>(read_res));
return;
}
read_buf.resize(get<size_t>(read_res));
......
......@@ -112,7 +112,7 @@ CAF_TEST(deliver serialized message) {
binary_serializer sink{sys, payload};
if (auto err = sink(node_id{}, self->id(), testee.id(), stages,
make_message(ok_atom_v)))
CAF_FAIL("unable to serialize message: " << sys.render(err));
CAF_FAIL("unable to serialize message: " << err);
net::basp::header hdr{net::basp::message_type::actor_message,
static_cast<uint32_t>(payload.size()),
make_message_id().integer_value()};
......
......@@ -42,7 +42,7 @@ using namespace caf::net;
#define REQUIRE_OK(statement) \
if (auto err = statement) \
CAF_FAIL(sys.render(err));
CAF_FAIL(err);
namespace {
......@@ -98,7 +98,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
void handle_handshake() {
CAF_CHECK_EQUAL(app->state(),
basp::connection_state::await_handshake_header);
auto payload = to_buf(mars, defaults::middleman::app_identifiers);
auto payload = to_buf(mars, basp::application::default_app_ids());
mock(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version});
CAF_CHECK_EQUAL(app->state(),
......@@ -122,7 +122,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
std::vector<std::string> app_ids;
binary_deserializer source{sys, buf};
if (auto err = source(nid, app_ids))
CAF_FAIL("unable to deserialize payload: " << sys.render(err));
CAF_FAIL("unable to deserialize payload: " << err);
if (source.remaining() > 0)
CAF_FAIL("trailing bytes after reading payload");
}
......@@ -172,7 +172,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); \
binary_deserializer source{sys, buf}; \
if (auto err = source(__VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
CAF_FAIL("failed to receive data: " << err); \
} else { \
if (hdr.payload_len != 0) \
CAF_FAIL("unexpected payload"); \
......
......@@ -51,7 +51,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() : recv_buf(1024), shared_buf{std::make_shared<buffer_type>()} {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
CAF_FAIL("mpx->init failed: " << err);
mpx->set_thread_id();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto sockets = unbox(make_stream_socket_pair());
......@@ -195,7 +195,7 @@ CAF_TEST(resolve and proxy communication) {
run();
auto read_res = read(recv_socket_guard.socket(), recv_buf);
if (!holds_alternative<size_t>(read_res))
CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res)));
CAF_FAIL("read() returned an error: " << get<sec>(read_res));
recv_buf.resize(get<size_t>(read_res));
CAF_MESSAGE("receive buffer contains " << recv_buf.size() << " bytes");
message msg;
......
......@@ -49,7 +49,7 @@ 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));
CAF_FAIL("mpx->init failed: " << err);
mpx->set_thread_id();
}
......
......@@ -160,7 +160,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
worker{dummy_application{application_results}} {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
CAF_FAIL("mpx->init failed: " << err);
if (auto err = parse("[::1]:12345", ep))
CAF_FAIL("parse returned an error: " << err);
worker = worker_type{dummy_application{application_results}, ep};
......
......@@ -121,7 +121,7 @@ CAF_TEST(read / write using span<std::vector<byte>*>) {
std::vector<byte> hdr_buf;
binary_serializer sink(sys, hdr_buf);
if (auto err = sink(hdr))
CAF_FAIL("serializing payload failed" << sys.render(err));
CAF_FAIL("serializing payload failed" << err);
auto bytes = as_bytes(make_span(hello_test));
std::vector<byte> payload_buf(bytes.begin(), bytes.end());
auto packet_size = hdr_buf.size() + payload_buf.size();
......
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