Commit f756161e authored by Jakob Otto's avatar Jakob Otto

Fix build errors

parent 1c9be4bf
...@@ -53,17 +53,17 @@ public: ...@@ -53,17 +53,17 @@ public:
using factory_type = Factory; using factory_type = Factory;
using transport_type = datagram_transport; using transport_type = datagram_transport<Factory>;
using application_type = typename Factory::application_type; using dispatcher_type = transport_worker_dispatcher<transport_type,
ip_endpoint>;
using dispatcher_type = transport_worker_dispatcher< using application_type = typename Factory::application_type;
datagram_transport, factory_type, ip_endpoint>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
datagram_transport(udp_datagram_socket handle, factory_type factory) datagram_transport(udp_datagram_socket handle, factory_type factory)
: dispatcher_(std::move(factory)), : dispatcher_(*this, std::move(factory)),
handle_(handle), handle_(handle),
max_consecutive_reads_(0), max_consecutive_reads_(0),
read_threshold_(1024), read_threshold_(1024),
...@@ -100,6 +100,7 @@ public: ...@@ -100,6 +100,7 @@ public:
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
manager_ = &parent;
auto& cfg = system().config(); auto& cfg = system().config();
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers", auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
defaults::middleman::max_header_buffers); defaults::middleman::max_header_buffers);
...@@ -107,21 +108,21 @@ public: ...@@ -107,21 +108,21 @@ public:
auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers", auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers); defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs); payload_bufs_.reserve(max_payload_bufs);
if (auto err = dispatcher_.init(parent)) if (auto err = dispatcher_.init(*this))
return err; return err;
parent.mask_add(operation::read); // parent.mask_add(operation::read);
return none; return none;
} }
template <class Parent> template <class Parent>
bool handle_read_event(Parent& parent) { bool handle_read_event(Parent&) {
CAF_LOG_TRACE(CAF_ARG(handle_.id)); CAF_LOG_TRACE(CAF_ARG(handle_.id));
auto ret = read(handle_, make_span(read_buf_)); auto ret = read(handle_, make_span(read_buf_));
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) { if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto num_bytes = res->first; auto num_bytes = res->first;
auto ep = res->second; auto ep = res->second;
read_buf_.resize(num_bytes); read_buf_.resize(num_bytes);
dispatcher_.handle_data(parent, make_span(read_buf_), std::move(ep)); dispatcher_.handle_data(*this, make_span(read_buf_), std::move(ep));
prepare_next_read(); prepare_next_read();
} else { } else {
auto err = get<sec>(ret); auto err = get<sec>(ret);
...@@ -231,14 +232,16 @@ public: ...@@ -231,14 +232,16 @@ public:
/// Helper struct for managing outgoing packets /// Helper struct for managing outgoing packets
struct packet { struct packet {
ip_endpoint destination; ip_endpoint destination;
size_t payload_buf_num;
buffer_cache_type bytes; buffer_cache_type bytes;
size_t size;
packet(ip_endpoint destination, span<buffer_type*> bufs) packet(ip_endpoint destination, span<buffer_type*> bufs)
: destination(destination) { : destination(destination) {
payload_buf_num = bufs.size() - 1; size = 0;
for (auto buf : bufs) for (auto buf : bufs) {
bytes.emplace_back(false, std::move(*buf)); size += buf->size();
bytes.emplace_back(std::move(*buf));
}
} }
}; };
...@@ -255,18 +258,43 @@ private: ...@@ -255,18 +258,43 @@ private:
} }
bool write_some() { bool write_some() {
CAF_LOG_TRACE(CAF_ARG(handle_.id));
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = packet_queue_.front();
auto& bufs = front.bytes;
auto it = bufs.begin();
if (header_bufs_.size() < header_bufs_.capacity()) {
it->clear();
header_bufs_.emplace_back(std::move(*it++));
}
for (;
it != bufs.end() && payload_bufs_.size() < payload_bufs_.capacity();
++it) {
it->clear();
payload_bufs_.emplace_back(std::move(*it));
}
packet_queue_.pop_front();
};
// Write buffers from the write_queue_ for as long as possible.
while (!packet_queue_.empty()) { while (!packet_queue_.empty()) {
auto& next_packet = packet_queue_.front(); auto& packet = packet_queue_.front();
auto send_res = write(handle_, next_packet.bytes, auto write_ret = write(handle_, make_span(packet.bytes),
next_packet.destination); packet.destination);
if (auto num_bytes = get_if<size_t>(&send_res)) { if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
packet_queue_.pop_front(); CAF_LOG_WARNING_IF(*num_bytes < packet.size,
return true; "packet was not sent completely");
} recycle();
auto err = get<sec>(send_res); } else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err)); CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
dispatcher_.handle_error(err); dispatcher_.handle_error(err);
return false;
}
return true;
}
} }
return false; return false;
} }
......
...@@ -30,7 +30,7 @@ namespace net { ...@@ -30,7 +30,7 @@ namespace net {
template <class Application, class IdType = unit_t> template <class Application, class IdType = unit_t>
class transport_worker; class transport_worker;
template <class Transport, class Application, class IdType = unit_t> template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher; class transport_worker_dispatcher;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
......
...@@ -248,8 +248,6 @@ private: ...@@ -248,8 +248,6 @@ private:
bool write_some() { bool write_some() {
CAF_LOG_TRACE(CAF_ARG(handle_.id)); CAF_LOG_TRACE(CAF_ARG(handle_.id));
if (write_queue_.empty())
return false;
// Helper function to sort empty buffers back into the right caches. // Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() { auto recycle = [&]() {
auto& front = write_queue_.front(); auto& front = write_queue_.front();
...@@ -266,7 +264,7 @@ private: ...@@ -266,7 +264,7 @@ private:
write_queue_.pop_front(); write_queue_.pop_front();
}; };
// Write buffers from the write_queue_ for as long as possible. // Write buffers from the write_queue_ for as long as possible.
do { while (!write_queue_.empty()) {
auto& buf = write_queue_.front().second; auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty()); CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_; auto data = buf.data() + written_;
...@@ -290,7 +288,7 @@ private: ...@@ -290,7 +288,7 @@ private:
} }
return true; return true;
} }
} while (!write_queue_.empty()); }
return false; return false;
} }
......
...@@ -34,18 +34,18 @@ namespace caf { ...@@ -34,18 +34,18 @@ namespace caf {
namespace net { namespace net {
/// implements a dispatcher that dispatches between transport and workers. /// implements a dispatcher that dispatches between transport and workers.
template <class Transport, class ApplicationFactory, class IdType> template <class Transport, class IdType>
class transport_worker_dispatcher { class transport_worker_dispatcher {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using id_type = IdType; using id_type = IdType;
using factory_type = ApplicationFactory;
using transport_type = Transport; using transport_type = Transport;
using application_type = typename ApplicationFactory::application_type; using factory_type = typename transport_type::factory_type;
using application_type = typename factory_type::application_type;
using worker_type = transport_worker<application_type, id_type>; using worker_type = transport_worker<application_type, id_type>;
...@@ -53,8 +53,8 @@ public: ...@@ -53,8 +53,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit transport_worker_dispatcher(factory_type factory) transport_worker_dispatcher(transport_type& transport, factory_type factory)
: factory_(std::move(factory)) { : factory_(std::move(factory)), transport_(&transport) {
// nop // nop
} }
...@@ -100,16 +100,16 @@ public: ...@@ -100,16 +100,16 @@ public:
} }
template <class Parent> template <class Parent>
void new_proxy(Parent&, const node_id& nid, actor_id id) { void new_proxy(Parent& parent, const node_id& nid, actor_id id) {
if (auto worker = find_by_node(nid)) if (auto worker = find_by_node(nid))
worker->new_proxy(*this, nid, id); worker->new_proxy(parent, nid, id);
} }
template <class Parent> template <class Parent>
void local_actor_down(Parent&, const node_id& nid, actor_id id, void local_actor_down(Parent& parent, const node_id& nid, actor_id id,
error reason) { error reason) {
if (auto worker = find_by_node(nid)) if (auto worker = find_by_node(nid))
worker->local_actor_down(*this, nid, id, std::move(reason)); worker->local_actor_down(parent, nid, id, std::move(reason));
} }
template <class... Ts> template <class... Ts>
...@@ -172,6 +172,7 @@ private: ...@@ -172,6 +172,7 @@ private:
std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_; std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_;
factory_type factory_; factory_type factory_;
transport_type* transport_;
}; };
} // namespace net } // namespace net
......
...@@ -66,8 +66,7 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -66,8 +66,7 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
/// @relates udp_datagram_socket /// @relates udp_datagram_socket
/// @post either the result is a `sec` or a positive (non-zero) integer /// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10` /// @pre `bufs.size() < 10`
variant<size_t, sec> write(udp_datagram_socket x, variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>> bufs,
std::initializer_list<span<const byte>> bufs,
ip_endpoint ep); ip_endpoint ep);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`. /// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
......
...@@ -17,6 +17,7 @@ ...@@ -17,6 +17,7 @@
******************************************************************************/ ******************************************************************************/
#include "caf/net/udp_datagram_socket.hpp" #include "caf/net/udp_datagram_socket.hpp"
#include <sys/uio.h>
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/convert_ip_endpoint.hpp" #include "caf/detail/convert_ip_endpoint.hpp"
...@@ -58,7 +59,7 @@ error allow_connreset(udp_datagram_socket x, bool) { ...@@ -58,7 +59,7 @@ error allow_connreset(udp_datagram_socket x, bool) {
expected<std::pair<udp_datagram_socket, uint16_t>> expected<std::pair<udp_datagram_socket, uint16_t>>
make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr) { make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(ep)); CAF_LOG_TRACE(CAF_ARG(ep));
sockaddr_storage addr; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id, CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id,
::socket(addr.ss_family, SOCK_DGRAM, 0)); ::socket(addr.ss_family, SOCK_DGRAM, 0));
...@@ -110,14 +111,35 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -110,14 +111,35 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
} }
} }
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>> bufs,
ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10);
auto convert = [](std::vector<byte> buf) {
return iovec{const_cast<byte*>(buf.data()), buf.size()};
};
sockaddr_storage addr = {};
detail::convert(ep, addr);
iovec buf_array[10];
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
msghdr message = {};
memset(&message, 0, sizeof(msghdr));
message.msg_name = &addr;
message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
message.msg_iov = buf_array;
message.msg_iovlen = bufs.size();
auto res = sendmsg(x.id, &message, 0);
return check_udp_datagram_socket_io_res(res);
}
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) { ip_endpoint ep) {
sockaddr_storage addr; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6));
auto ret = check_udp_datagram_socket_io_res(res); auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = get_if<size_t>(&ret)) if (auto num_bytes = get_if<size_t>(&ret))
return *num_bytes; return *num_bytes;
......
...@@ -59,8 +59,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -59,8 +59,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application { class dummy_application {
public: public:
using transport_type = datagram_transport<dummy_application_factory>;
dummy_application(std::shared_ptr<std::vector<byte>> rec_buf) dummy_application(std::shared_ptr<std::vector<byte>> rec_buf)
: rec_buf_(std::move(rec_buf)){ : rec_buf_(std::move(rec_buf)){
// nop // nop
...@@ -76,7 +74,7 @@ public: ...@@ -76,7 +74,7 @@ public:
template <class Transport> template <class Transport>
void write_message(Transport& transport, void write_message(Transport& transport,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
transport.write_packet(span<byte>{}, msg->payload); transport.write_packet(msg->payload);
} }
template <class Parent> template <class Parent>
...@@ -110,8 +108,8 @@ public: ...@@ -110,8 +108,8 @@ public:
// nop // nop
} }
template <class Transport> template <class Parent>
void timeout(Transport&, atom_value, uint64_t) { void timeout(Parent&, atom_value, uint64_t) {
// nop // nop
} }
...@@ -159,7 +157,7 @@ CAF_TEST(receive) { ...@@ -159,7 +157,7 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
ip_endpoint ep; ip_endpoint ep;
if (auto err = parse("127.0.0.1:0", ep)) if (auto err = parse("127.0.0.1:0", ep))
CAF_FAIL("parse returned an error: " << err); CAF_FAIL("parse returned an error: " << sys.render(err));
auto send_pair = unbox(make_udp_datagram_socket(ep)); auto send_pair = unbox(make_udp_datagram_socket(ep));
auto sender = send_pair.first; auto sender = send_pair.first;
auto receive_pair = unbox(make_udp_datagram_socket(ep)); auto receive_pair = unbox(make_udp_datagram_socket(ep));
...@@ -168,7 +166,7 @@ CAF_TEST(receive) { ...@@ -168,7 +166,7 @@ CAF_TEST(receive) {
auto send_guard = make_socket_guard(sender); auto send_guard = make_socket_guard(sender);
auto receive_guard = make_socket_guard(receiver); auto receive_guard = make_socket_guard(receiver);
if (auto err = nonblocking(receiver, true)) if (auto err = nonblocking(receiver, true))
CAF_FAIL("nonblocking() returned an error: " << err); CAF_FAIL("nonblocking() returned an error: " << sys.render(err));
transport_type transport{receiver, dummy_application_factory{buf}}; transport_type transport{receiver, dummy_application_factory{buf}};
transport.configure_read(net::receive_policy::exactly(hello_manager.size())); transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, transport); auto mgr = make_endpoint_manager(mpx, sys, transport);
......
...@@ -122,6 +122,8 @@ private: ...@@ -122,6 +122,8 @@ private:
struct dummy_transport { struct dummy_transport {
using transport_type = dummy_transport; using transport_type = dummy_transport;
using factory_type = dummy_application_factory;
using application_type = dummy_application; using application_type = dummy_application;
dummy_transport(std::shared_ptr<buffer_type> buf) : buf_(std::move(buf)) { dummy_transport(std::shared_ptr<buffer_type> buf) : buf_(std::move(buf)) {
...@@ -180,12 +182,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) { ...@@ -180,12 +182,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
} }
struct fixture : host_fixture { struct fixture : host_fixture {
using dispatcher_type = transport_worker_dispatcher< using dispatcher_type = transport_worker_dispatcher<dummy_transport,
dummy_transport, dummy_application_factory, ip_endpoint>; ip_endpoint>;
fixture() fixture()
: buf{std::make_shared<buffer_type>()}, : buf{std::make_shared<buffer_type>()},
dispatcher{dummy_application_factory{buf}}, dispatcher{dummy, dummy_application_factory{buf}},
dummy{buf} { dummy{buf} {
add_new_workers(); add_new_workers();
} }
...@@ -268,7 +270,7 @@ struct fixture : host_fixture { ...@@ -268,7 +270,7 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture) CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture)
CAF_TEST(init) { CAF_TEST(init) {
dispatcher_type dispatcher{dummy_application_factory{buf}}; dispatcher_type dispatcher{dummy, dummy_application_factory{buf}};
if (auto err = dispatcher.init(dummy)) if (auto err = dispatcher.init(dummy))
CAF_FAIL("init failed with error: " << err); CAF_FAIL("init failed with error: " << err);
} }
......
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