Commit 916fd5e9 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #85

Switch to more lightweight socket API wrapper
parents 0c6234d1 6de665b4
...@@ -82,9 +82,8 @@ public: ...@@ -82,9 +82,8 @@ public:
system_ = &parent.system(); system_ = &parent.system();
executor_.system_ptr(system_); executor_.system_ptr(system_);
executor_.proxy_registry_ptr(&proxies_); executor_.proxy_registry_ptr(&proxies_);
// TODO: use `if constexpr` when switching to C++17.
// Allow unit tests to run the application without endpoint manager. // Allow unit tests to run the application without endpoint manager.
if (!std::is_base_of<test_tag, Parent>::value) if constexpr (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager(); manager_ = &parent.manager();
size_t workers; size_t workers;
if (auto workers_cfg = get_if<size_t>(&system_->config(), if (auto workers_cfg = get_if<size_t>(&system_->config(),
......
...@@ -27,6 +27,10 @@ ...@@ -27,6 +27,10 @@
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp" #include "caf/net/socket_id.hpp"
// Note: This API mostly wraps platform-specific functions that return ssize_t.
// We return ptrdiff_t instead, since only POSIX defines ssize_t and the two
// types are functionally equivalent.
namespace caf::net { namespace caf::net {
/// A unidirectional communication endpoint for inter-process communication. /// A unidirectional communication endpoint for inter-process communication.
...@@ -46,19 +50,13 @@ expected<std::pair<pipe_socket, pipe_socket>> CAF_NET_EXPORT make_pipe(); ...@@ -46,19 +50,13 @@ expected<std::pair<pipe_socket, pipe_socket>> CAF_NET_EXPORT make_pipe();
/// @param buf Memory region for reading the message to send. /// @param buf Memory region for reading the message to send.
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket /// @relates pipe_socket
variant<size_t, sec> CAF_NET_EXPORT write(pipe_socket x, span<const byte> buf); ptrdiff_t CAF_NET_EXPORT write(pipe_socket x, span<const byte> buf);
/// Receives data from `x`. /// Receives data from `x`.
/// @param x Connected endpoint. /// @param x Connected endpoint.
/// @param buf Memory region for storing the received bytes. /// @param buf Memory region for storing the received bytes.
/// @returns The number of received bytes on success, otherwise an error code. /// @returns The number of received bytes on success, otherwise an error code.
/// @relates pipe_socket /// @relates pipe_socket
variant<size_t, sec> CAF_NET_EXPORT read(pipe_socket x, span<byte> buf); ptrdiff_t 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.
/// @relates pipe_socket
variant<size_t, sec>
CAF_NET_EXPORT check_pipe_socket_io_res(std::make_signed<size_t>::type res);
} // namespace caf::net } // namespace caf::net
...@@ -69,7 +69,7 @@ To CAF_NET_EXPORT socket_cast(From x) { ...@@ -69,7 +69,7 @@ To CAF_NET_EXPORT socket_cast(From x) {
return To{x.id}; return To{x.id};
} }
/// Close socket `x`. /// Closes socket `x`.
/// @relates socket /// @relates socket
void CAF_NET_EXPORT close(socket x); void CAF_NET_EXPORT close(socket x);
...@@ -77,6 +77,14 @@ void CAF_NET_EXPORT close(socket x); ...@@ -77,6 +77,14 @@ void CAF_NET_EXPORT close(socket x);
/// @relates socket /// @relates socket
std::errc CAF_NET_EXPORT last_socket_error(); std::errc CAF_NET_EXPORT last_socket_error();
/// Checks whether `last_socket_error()` would return an error code that
/// indicates a temporary error.
/// @returns `true` if `last_socket_error()` returned either
/// `std::errc::operation_would_block` or
/// `std::errc::resource_unavailable_try_again`, `false` otherwise.
/// @relates socket
bool CAF_NET_EXPORT last_socket_error_is_temporary();
/// Returns the last socket error as human-readable string. /// Returns the last socket error as human-readable string.
/// @relates socket /// @relates socket
std::string CAF_NET_EXPORT last_socket_error_as_string(); std::string CAF_NET_EXPORT last_socket_error_as_string();
......
...@@ -18,10 +18,16 @@ ...@@ -18,10 +18,16 @@
#pragma once #pragma once
#include <cstddef>
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
// Note: This API mostly wraps platform-specific functions that return ssize_t.
// We return ptrdiff_t instead, since only POSIX defines ssize_t and the two
// types are functionally equivalent.
namespace caf::net { namespace caf::net {
/// A connection-oriented network communication endpoint for bidirectional byte /// A connection-oriented network communication endpoint for bidirectional byte
...@@ -47,37 +53,33 @@ error CAF_NET_EXPORT keepalive(stream_socket x, bool new_value); ...@@ -47,37 +53,33 @@ error CAF_NET_EXPORT keepalive(stream_socket x, bool new_value);
error CAF_NET_EXPORT nodelay(stream_socket x, bool new_value); error CAF_NET_EXPORT nodelay(stream_socket x, bool new_value);
/// Receives data from `x`. /// Receives data from `x`.
/// @param x Connected endpoint. /// @param x A connected endpoint.
/// @param buf Points to destination buffer. /// @param buf Points to destination buffer.
/// @returns The number of received bytes on success, an error code otherwise. /// @returns The number of received bytes on success, 0 if the socket is closed,
/// or -1 in case of an error.
/// @relates stream_socket /// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer /// @post Either the functions returned a non-negative integer or the caller can
variant<size_t, sec> CAF_NET_EXPORT read(stream_socket x, span<byte> buf); /// retrieve the error code by calling `last_socket_error()`.
ptrdiff_t CAF_NET_EXPORT read(stream_socket x, span<byte> buf);
/// Transmits data from `x` to its peer. /// Sends data to `x`.
/// @param x Connected endpoint. /// @param x A connected endpoint.
/// @param buf Points to the message to send. /// @param buf Points to the message to send.
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of received bytes on success, 0 if the socket is closed,
/// or -1 in case of an error.
/// @relates stream_socket /// @relates stream_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
variant<size_t, sec> CAF_NET_EXPORT write(stream_socket x, ptrdiff_t CAF_NET_EXPORT write(stream_socket x, span<const byte> buf);
span<const byte> buf);
/// Transmits data from `x` to its peer. /// Transmits data from `x` to its peer.
/// @param x Connected endpoint. /// @param x A connected endpoint.
/// @param bufs Points to the message to send, scattered across up to 10 /// @param bufs Points to the message to send, scattered across up to 10
/// buffers. /// buffers.
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of written bytes on success, otherwise an error code.
/// @relates stream_socket /// @relates stream_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> CAF_NET_EXPORT ptrdiff_t CAF_NET_EXPORT write(stream_socket x,
write(stream_socket x, std::initializer_list<span<const byte>> bufs); std::initializer_list<span<const byte>> bufs);
/// Converts the result from I/O operation on a ::stream_socket to either an
/// error code or a non-zero positive integer.
/// @relates stream_socket
variant<size_t, sec>
CAF_NET_EXPORT check_stream_socket_io_res(std::make_signed<size_t>::type res);
} // namespace caf::net } // namespace caf::net
...@@ -73,17 +73,22 @@ public: ...@@ -73,17 +73,22 @@ public:
bool handle_read_event(endpoint_manager&) override { bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id)); CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id));
auto fail = [this](sec err) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
};
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) { for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) {
auto buf = this->read_buf_.data() + this->collected_; auto buf = this->read_buf_.data() + collected_;
size_t len = this->read_threshold_ - this->collected_; size_t len = read_threshold_ - collected_;
CAF_LOG_DEBUG(CAF_ARG2("missing", len)); CAF_LOG_DEBUG(CAF_ARG2("missing", len));
auto ret = read(this->handle_, make_span(buf, len)); auto num_bytes = read(this->handle_, make_span(buf, len));
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG2("handle", this->handle().id)
<< CAF_ARG(num_bytes));
// Update state. // Update state.
if (auto num_bytes = get_if<size_t>(&ret)) { if (num_bytes > 0) {
CAF_LOG_DEBUG(CAF_ARG(len) collected_ += num_bytes;
<< CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes)); if (collected_ >= read_threshold_) {
this->collected_ += *num_bytes;
if (this->collected_ >= this->read_threshold_) {
if (auto err = this->next_layer_.handle_data( if (auto err = this->next_layer_.handle_data(
*this, make_span(this->read_buf_))) { *this, make_span(this->read_buf_))) {
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err)); CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err));
...@@ -91,15 +96,16 @@ public: ...@@ -91,15 +96,16 @@ public:
} }
this->prepare_next_read(); this->prepare_next_read();
} }
} else if (num_bytes < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and
// stop reading on the socket on hard errors.
return last_socket_error_is_temporary()
? true
: fail(sec::socket_operation_failed);
} else { } else {
auto err = get<sec>(ret); // read() returns 0 iff the connection was closed.
if (err == sec::unavailable_or_would_block) { return fail(sec::socket_disconnected);
break;
} else {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
} }
} }
return true; return true;
...@@ -125,27 +131,33 @@ public: ...@@ -125,27 +131,33 @@ public:
} }
write_queue_.pop_front(); write_queue_.pop_front();
}; };
auto fail = [this](sec err) {
CAF_LOG_DEBUG("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return err;
};
// Write buffers from the write_queue_ for as long as possible. // Write buffers from the write_queue_ for as long as possible.
while (!write_queue_.empty()) { 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_;
auto len = buf.size() - written_; auto len = buf.size() - written_;
auto write_ret = write(this->handle(), make_span(data, len)); auto num_bytes = write(this->handle(), make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) { if (num_bytes > 0) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(num_bytes));
written_ += *num_bytes; written_ += num_bytes;
if (written_ >= buf.size()) { if (written_ >= static_cast<ptrdiff_t>(buf.size())) {
recycle(); recycle();
written_ = 0; written_ = 0;
} }
} else if (num_bytes < 0) {
return last_socket_error_is_temporary()
? sec::unavailable_or_would_block
: fail(sec::socket_operation_failed);
} else { } else {
auto err = get<sec>(write_ret); // write() returns 0 iff the connection was closed.
if (err != sec::unavailable_or_would_block) { return fail(sec::socket_disconnected);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
}
return err;
} }
} }
return none; return none;
...@@ -212,9 +224,9 @@ private: ...@@ -212,9 +224,9 @@ private:
} }
write_queue_type write_queue_; write_queue_type write_queue_;
size_t written_; ptrdiff_t written_;
size_t read_threshold_; ptrdiff_t read_threshold_;
size_t collected_; ptrdiff_t collected_;
size_t max_; size_t max_;
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
}; };
......
...@@ -155,15 +155,40 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -155,15 +155,40 @@ bool multiplexer::poll_once(bool blocking) {
return false; return false;
// We'll call poll() until poll() succeeds or fails. // We'll call poll() until poll() succeeds or fails.
for (;;) { for (;;) {
int presult; int presult =
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
presult = ::WSAPoll(pollset_.data(), static_cast<ULONG>(pollset_.size()), ::WSAPoll(pollset_.data(), static_cast<ULONG>(pollset_.size()),
blocking ? -1 : 0); blocking ? -1 : 0);
#else #else
presult = ::poll(pollset_.data(), static_cast<nfds_t>(pollset_.size()), ::poll(pollset_.data(), static_cast<nfds_t>(pollset_.size()),
blocking ? -1 : 0); blocking ? -1 : 0);
#endif #endif
if (presult < 0) { if (presult > 0) {
CAF_LOG_DEBUG("poll() on" << pollset_.size() << "sockets reported"
<< presult << "event(s)");
// Scan pollset for events.
CAF_LOG_DEBUG("scan pollset for socket events");
for (size_t i = 0; i < pollset_.size() && presult > 0;) {
auto revents = pollset_[i].revents;
if (revents != 0) {
auto events = pollset_[i].events;
auto mgr = managers_[i];
auto new_events = handle(mgr, events, revents);
--presult;
if (new_events == 0) {
del(i);
continue;
} else if (new_events != events) {
pollset_[i].events = new_events;
}
}
++i;
}
return true;
} else if (presult == 0) {
// No activity.
return false;
} else {
auto code = last_socket_error(); auto code = last_socket_error();
switch (code) { switch (code) {
case std::errc::interrupted: { case std::errc::interrupted: {
...@@ -186,33 +211,7 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -186,33 +211,7 @@ bool multiplexer::poll_once(bool blocking) {
CAF_CRITICAL(msg.c_str()); CAF_CRITICAL(msg.c_str());
} }
} }
// Rinse and repeat.
continue;
} }
CAF_LOG_DEBUG("poll() on" << pollset_.size() << "sockets reported"
<< presult << "event(s)");
// No activity.
if (presult == 0)
return false;
// Scan pollset for events.
CAF_LOG_DEBUG("scan pollset for socket events");
for (size_t i = 0; i < pollset_.size() && presult > 0;) {
auto revents = pollset_[i].revents;
if (revents != 0) {
auto events = pollset_[i].events;
auto mgr = managers_[i];
auto new_events = handle(mgr, events, revents);
--presult;
if (new_events == 0) {
del(i);
continue;
} else if (new_events != events) {
pollset_[i].events = new_events;
}
}
++i;
}
return true;
} }
} }
...@@ -302,15 +301,13 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { ...@@ -302,15 +301,13 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
buf[0] = static_cast<byte>(opcode); buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(mgr.get()); auto value = reinterpret_cast<intptr_t>(mgr.get());
memcpy(buf.data() + 1, &value, sizeof(intptr_t)); memcpy(buf.data() + 1, &value, sizeof(intptr_t));
variant<size_t, sec> res; ptrdiff_t res = -1;
{ // Lifetime scope of guard. { // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_}; std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) if (write_handle_ != invalid_socket)
res = write(write_handle_, buf); res = write(write_handle_, buf);
else
res = sec::socket_invalid;
} }
if (holds_alternative<sec>(res) && opcode != 4) if (res <= 0 && opcode != 4)
mgr->deref(); mgr->deref();
} }
......
...@@ -105,7 +105,8 @@ void middleman::add_module_options(actor_system_config& cfg) { ...@@ -105,7 +105,8 @@ void middleman::add_module_options(actor_system_config& cfg) {
"max. number of consecutive reads per broker") "max. number of consecutive reads per broker")
.add<bool>("manual-multiplexing", .add<bool>("manual-multiplexing",
"disables background activity of the multiplexer") "disables background activity of the multiplexer")
.add<size_t>("workers", "number of deserialization workers"); .add<size_t>("workers", "number of deserialization workers")
.add<std::string>("network-backend", "legacy option");
} }
expected<endpoint_manager_ptr> middleman::connect(const uri& locator) { expected<endpoint_manager_ptr> middleman::connect(const uri& locator) {
......
...@@ -49,12 +49,12 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() { ...@@ -49,12 +49,12 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
} }
} }
variant<size_t, sec> write(pipe_socket x, span<const byte> buf) { ptrdiff_t write(pipe_socket x, span<const byte> buf) {
// On Windows, a pipe consists of two stream sockets. // On Windows, a pipe consists of two stream sockets.
return write(socket_cast<stream_socket>(x), buf); return write(socket_cast<stream_socket>(x), buf);
} }
variant<size_t, sec> read(pipe_socket x, span<byte> buf) { ptrdiff_t read(pipe_socket x, span<byte> buf) {
// On Windows, a pipe consists of two stream sockets. // On Windows, a pipe consists of two stream sockets.
return read(socket_cast<stream_socket>(x), buf); return read(socket_cast<stream_socket>(x), buf);
} }
...@@ -80,23 +80,16 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() { ...@@ -80,23 +80,16 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
return std::make_pair(pipe_socket{pipefds[0]}, pipe_socket{pipefds[1]}); return std::make_pair(pipe_socket{pipefds[0]}, pipe_socket{pipefds[1]});
} }
variant<size_t, sec> write(pipe_socket x, span<const byte> buf) { ptrdiff_t write(pipe_socket x, span<const byte> buf) {
auto res = ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), return ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size()); buf.size());
return check_pipe_socket_io_res(res);
} }
variant<size_t, sec> read(pipe_socket x, span<byte> buf) { ptrdiff_t read(pipe_socket x, span<byte> buf) {
auto res = ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), return ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size()); buf.size());
return check_pipe_socket_io_res(res);
} }
#endif // CAF_WINDOWS #endif // CAF_WINDOWS
variant<size_t, sec>
check_pipe_socket_io_res(std::make_signed<size_t>::type res) {
return check_stream_socket_io_res(res);
}
} // namespace caf::net } // namespace caf::net
...@@ -42,10 +42,10 @@ pollset_updater::~pollset_updater() { ...@@ -42,10 +42,10 @@ pollset_updater::~pollset_updater() {
bool pollset_updater::handle_read_event() { bool pollset_updater::handle_read_event() {
for (;;) { for (;;) {
auto res = read(handle(), make_span(buf_.data() + buf_size_, auto num_bytes = read(handle(), make_span(buf_.data() + buf_size_,
buf_.size() - buf_size_)); buf_.size() - buf_size_));
if (auto num_bytes = get_if<size_t>(&res)) { if (num_bytes > 0) {
buf_size_ += *num_bytes; buf_size_ += static_cast<size_t>(num_bytes);
if (buf_.size() == buf_size_) { if (buf_.size() == buf_size_) {
buf_size_ = 0; buf_size_ = 0;
auto opcode = static_cast<uint8_t>(buf_[0]); auto opcode = static_cast<uint8_t>(buf_[0]);
...@@ -70,7 +70,7 @@ bool pollset_updater::handle_read_event() { ...@@ -70,7 +70,7 @@ bool pollset_updater::handle_read_event() {
} }
} }
} else { } else {
return get<sec>(res) == sec::unavailable_or_would_block; return num_bytes < 0 && last_socket_error_is_temporary();
} }
} }
} }
......
...@@ -103,6 +103,11 @@ std::errc last_socket_error() { ...@@ -103,6 +103,11 @@ std::errc last_socket_error() {
abort(); abort();
} }
bool last_socket_error_is_temporary() {
int wsa_code = WSAGetLastError();
return wsa_code == WSAEWOULDBLOCK || wsa_code == WSATRY_AGAIN;
}
std::string last_socket_error_as_string() { std::string last_socket_error_as_string() {
int wsa_code = WSAGetLastError(); int wsa_code = WSAGetLastError();
LPTSTR errorText = NULL; LPTSTR errorText = NULL;
...@@ -149,6 +154,15 @@ std::errc last_socket_error() { ...@@ -149,6 +154,15 @@ std::errc last_socket_error() {
return static_cast<std::errc>(errno); return static_cast<std::errc>(errno);
} }
bool last_socket_error_is_temporary() {
auto code = errno;
# if EAGAIN == EWOULDBLOCK
return code == EAGAIN;
# else
return code == EAGAIN || code == EWOULDBLOCK;
# endif
}
std::string last_socket_error_as_string() { std::string last_socket_error_as_string() {
return strerror(errno); return strerror(errno);
} }
......
...@@ -168,22 +168,19 @@ error nodelay(stream_socket x, bool new_value) { ...@@ -168,22 +168,19 @@ error nodelay(stream_socket x, bool new_value) {
return none; return none;
} }
variant<size_t, sec> read(stream_socket x, span<byte> buf) { ptrdiff_t read(stream_socket x, span<byte> buf) {
auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), return ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), buf.size(),
buf.size(), no_sigpipe_io_flag); no_sigpipe_io_flag);
return check_stream_socket_io_res(res);
} }
variant<size_t, sec> write(stream_socket x, span<const byte> buf) { ptrdiff_t write(stream_socket x, span<const byte> buf) {
auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), return ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), buf.size(),
buf.size(), no_sigpipe_io_flag); no_sigpipe_io_flag);
return check_stream_socket_io_res(res);
} }
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
variant<size_t, sec> write(stream_socket x, ptrdiff_t write(stream_socket x, std::initializer_list<span<const byte>> bufs) {
std::initializer_list<span<const byte>> bufs) {
CAF_ASSERT(bufs.size() < 10); CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10]; WSABUF buf_array[10];
auto convert = [](span<const byte> buf) { auto convert = [](span<const byte> buf) {
...@@ -195,44 +192,21 @@ variant<size_t, sec> write(stream_socket x, ...@@ -195,44 +192,21 @@ variant<size_t, sec> write(stream_socket x,
DWORD bytes_sent = 0; DWORD bytes_sent = 0;
auto res = WSASend(x.id, buf_array, static_cast<DWORD>(bufs.size()), auto res = WSASend(x.id, buf_array, static_cast<DWORD>(bufs.size()),
&bytes_sent, 0, nullptr, nullptr); &bytes_sent, 0, nullptr, nullptr);
if (res != 0) { return (res == 0) ? bytes_sent : -1;
auto code = last_socket_error();
if (code == std::errc::operation_would_block
|| code == std::errc::resource_unavailable_try_again)
return sec::unavailable_or_would_block;
return sec::socket_operation_failed;
}
return static_cast<size_t>(bytes_sent);
} }
#else // CAF_WINDOWS #else // CAF_WINDOWS
variant<size_t, sec> write(stream_socket x, ptrdiff_t write(stream_socket x, std::initializer_list<span<const byte>> bufs) {
std::initializer_list<span<const byte>> bufs) {
CAF_ASSERT(bufs.size() < 10); CAF_ASSERT(bufs.size() < 10);
iovec buf_array[10]; iovec buf_array[10];
auto convert = [](span<const byte> buf) { auto convert = [](span<const byte> buf) {
return iovec{const_cast<byte*>(buf.data()), buf.size()}; return iovec{const_cast<byte*>(buf.data()), buf.size()};
}; };
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert); std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
auto res = writev(x.id, buf_array, static_cast<int>(bufs.size())); return writev(x.id, buf_array, static_cast<int>(bufs.size()));
return check_stream_socket_io_res(res);
} }
#endif // CAF_WINDOWS #endif // CAF_WINDOWS
variant<size_t, sec>
check_stream_socket_io_res(std::make_signed<size_t>::type res) {
if (res == 0)
return sec::socket_disconnected;
if (res < 0) {
auto code = last_socket_error();
if (code == std::errc::operation_would_block
|| code == std::errc::resource_unavailable_try_again)
return sec::unavailable_or_would_block;
return sec::socket_operation_failed;
}
return static_cast<size_t>(res);
}
} // namespace caf::net } // namespace caf::net
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/net/basp/connection_state.hpp" #include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp" #include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/packet_writer.hpp" #include "caf/net/packet_writer.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
...@@ -42,7 +43,13 @@ using namespace caf::net; ...@@ -42,7 +43,13 @@ using namespace caf::net;
namespace { namespace {
struct fixture : test_coordinator_fixture<>, struct config : actor_system_config {
config() {
net::middleman::add_module_options(*this);
}
};
struct fixture : test_coordinator_fixture<config>,
proxy_registry::backend, proxy_registry::backend,
basp::application::test_tag, basp::application::test_tag,
public packet_writer { public packet_writer {
...@@ -242,8 +249,7 @@ CAF_TEST(actor message) { ...@@ -242,8 +249,7 @@ CAF_TEST(actor message) {
MOCK(basp::message_type::actor_message, make_message_id().integer_value(), MOCK(basp::message_type::actor_message, make_message_id().integer_value(),
mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{}, mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{},
make_message("hello world!")); make_message("hello world!"));
allow((monitor_atom, strong_actor_ptr), expect((monitor_atom, strong_actor_ptr), from(_).to(self));
from(_).to(self).with(monitor_atom_v, _));
expect((std::string), from(_).to(self).with("hello world!")); expect((std::string), from(_).to(self).with("hello world!"));
} }
......
...@@ -90,13 +90,13 @@ public: ...@@ -90,13 +90,13 @@ public:
template <class Manager> template <class Manager>
bool handle_read_event(Manager&) { bool handle_read_event(Manager&) {
auto res = read(handle_, read_buf_); auto num_bytes = read(handle_, read_buf_);
if (auto num_bytes = get_if<size_t>(&res)) { if (num_bytes > 0) {
data_->insert(data_->end(), read_buf_.begin(), data_->insert(data_->end(), read_buf_.begin(),
read_buf_.begin() + *num_bytes); read_buf_.begin() + num_bytes);
return true; return true;
} }
return get<sec>(res) == sec::unavailable_or_would_block; return num_bytes < 0 && last_socket_error_is_temporary();
} }
template <class Manager> template <class Manager>
...@@ -106,12 +106,12 @@ public: ...@@ -106,12 +106,12 @@ public:
if (auto err = sink(x->msg->payload)) if (auto err = sink(x->msg->payload))
CAF_FAIL("serializing failed: " << err); CAF_FAIL("serializing failed: " << err);
} }
auto res = write(handle_, buf_); auto num_bytes = write(handle_, buf_);
if (auto num_bytes = get_if<size_t>(&res)) { if (num_bytes > 0) {
buf_.erase(buf_.begin(), buf_.begin() + *num_bytes); buf_.erase(buf_.begin(), buf_.begin() + num_bytes);
return buf_.size() > 0; return buf_.size() > 0;
} }
return get<sec>(res) == sec::unavailable_or_would_block; return num_bytes < 0 && last_socket_error_is_temporary();
} }
void handle_error(sec) { void handle_error(sec) {
...@@ -164,8 +164,8 @@ CAF_TEST(send and receive) { ...@@ -164,8 +164,8 @@ CAF_TEST(send and receive) {
auto buf = std::make_shared<byte_buffer>(); auto buf = std::make_shared<byte_buffer>();
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
CAF_CHECK_EQUAL(nonblocking(sockets.second, true), none); CAF_CHECK_EQUAL(nonblocking(sockets.second, true), none);
CAF_CHECK_EQUAL(read(sockets.second, read_buf), CAF_CHECK_LESS(read(sockets.second, read_buf), 0);
sec::unavailable_or_would_block); CAF_CHECK(last_socket_error_is_temporary());
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf}); dummy_transport{sockets.first, buf});
...@@ -208,11 +208,14 @@ CAF_TEST(resolve and proxy communication) { ...@@ -208,11 +208,14 @@ CAF_TEST(resolve and proxy communication) {
[&] { CAF_FAIL("manager did not respond with a proxy."); }); [&] { CAF_FAIL("manager did not respond with a proxy."); });
run(); run();
auto read_res = read(sockets.second, read_buf); auto read_res = read(sockets.second, read_buf);
if (!holds_alternative<size_t>(read_res)) { if (read_res <= 0) {
CAF_ERROR("read() returned an error: " << get<sec>(read_res)); std::string msg = "socket closed";
if (read_res < 0)
msg = last_socket_error_as_string();
CAF_ERROR("read() failed: " << msg);
return; return;
} }
read_buf.resize(get<size_t>(read_res)); read_buf.resize(static_cast<size_t>(read_res));
CAF_MESSAGE("receive buffer contains " << read_buf.size() << " bytes"); CAF_MESSAGE("receive buffer contains " << read_buf.size() << " bytes");
message msg; message msg;
binary_deserializer source{sys, read_buf}; binary_deserializer source{sys, read_buf};
......
...@@ -59,26 +59,25 @@ public: ...@@ -59,26 +59,25 @@ public:
bool handle_read_event() override { bool handle_read_event() override {
if (read_capacity() < 1024) if (read_capacity() < 1024)
rd_buf_.resize(rd_buf_.size() + 2048); rd_buf_.resize(rd_buf_.size() + 2048);
auto res = read(handle(), auto num_bytes = read(handle(),
make_span(read_position_begin(), read_capacity())); make_span(read_position_begin(), read_capacity()));
if (auto num_bytes = get_if<size_t>(&res)) { if (num_bytes > 0) {
CAF_ASSERT(*num_bytes > 0); CAF_ASSERT(num_bytes > 0);
rd_buf_pos_ += *num_bytes; rd_buf_pos_ += num_bytes;
return true; return true;
} }
return get<sec>(res) == sec::unavailable_or_would_block; return num_bytes < 0 && last_socket_error_is_temporary();
} }
bool handle_write_event() override { bool handle_write_event() override {
if (wr_buf_.size() == 0) if (wr_buf_.size() == 0)
return false; return false;
auto res = write(handle(), wr_buf_); auto num_bytes = write(handle(), wr_buf_);
if (auto num_bytes = get_if<size_t>(&res)) { if (num_bytes > 0) {
CAF_ASSERT(*num_bytes > 0); wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + num_bytes);
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + *num_bytes);
return wr_buf_.size() > 0; return wr_buf_.size() > 0;
} }
return get<sec>(res) == sec::unavailable_or_would_block; return num_bytes < 0 && last_socket_error_is_temporary();
} }
void handle_error(sec code) override { void handle_error(sec code) override {
......
...@@ -48,12 +48,6 @@ namespace { ...@@ -48,12 +48,6 @@ namespace {
using transport_type = stream_transport<basp::application>; using transport_type = stream_transport<basp::application>;
size_t fetch_size(variant<size_t, sec> x) {
if (holds_alternative<sec>(x))
CAF_FAIL("read/write failed: " << to_string(get<sec>(x)));
return get<size_t>(x);
}
struct config : actor_system_config { struct config : actor_system_config {
config() { config() {
put(content, "caf.middleman.this-node", unbox(make_uri("test:earth"))); put(content, "caf.middleman.this-node", unbox(make_uri("test:earth")));
...@@ -88,7 +82,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -88,7 +82,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
template <class... Ts> template <class... Ts>
void mock(const Ts&... xs) { void mock(const Ts&... xs) {
auto buf = to_buf(xs...); auto buf = to_buf(xs...);
if (fetch_size(write(sock, buf)) != buf.size()) if (write(sock, buf) != static_cast<ptrdiff_t>(buf.size()))
CAF_FAIL("unable to write " << buf.size() << " bytes"); CAF_FAIL("unable to write " << buf.size() << " bytes");
run(); run();
} }
...@@ -107,14 +101,14 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -107,14 +101,14 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
void consume_handshake() { void consume_handshake() {
byte_buffer buf(basp::header_size); byte_buffer buf(basp::header_size);
if (fetch_size(read(sock, buf)) != basp::header_size) if (read(sock, buf) != basp::header_size)
CAF_FAIL("unable to read " << basp::header_size << " bytes"); CAF_FAIL("unable to read " << basp::header_size << " bytes");
auto hdr = basp::header::from_bytes(buf); auto hdr = basp::header::from_bytes(buf);
if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0 if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0
|| hdr.operation_data != basp::version) || hdr.operation_data != basp::version)
CAF_FAIL("invalid handshake header"); CAF_FAIL("invalid handshake header");
buf.resize(hdr.payload_len); buf.resize(hdr.payload_len);
if (fetch_size(read(sock, buf)) != hdr.payload_len) if (read(sock, buf) != static_cast<ptrdiff_t>(hdr.payload_len))
CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); CAF_FAIL("unable to read " << hdr.payload_len << " bytes");
node_id nid; node_id nid;
std::vector<std::string> app_ids; std::vector<std::string> app_ids;
...@@ -158,7 +152,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -158,7 +152,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
do { \ do { \
CAF_MESSAGE("receive " << msg_type); \ CAF_MESSAGE("receive " << msg_type); \
byte_buffer buf(basp::header_size); \ byte_buffer buf(basp::header_size); \
if (fetch_size(read(sock, buf)) != basp::header_size) \ if (read(sock, buf) != static_cast<ptrdiff_t>(basp::header_size)) \
CAF_FAIL("unable to read " << basp::header_size << " bytes"); \ CAF_FAIL("unable to read " << basp::header_size << " bytes"); \
auto hdr = basp::header::from_bytes(buf); \ auto hdr = basp::header::from_bytes(buf); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \ CAF_CHECK_EQUAL(hdr.type, msg_type); \
...@@ -166,7 +160,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -166,7 +160,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \ if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \
std::tuple<unit_t>>::value) { \ std::tuple<unit_t>>::value) { \
buf.resize(hdr.payload_len); \ buf.resize(hdr.payload_len); \
if (fetch_size(read(sock, buf)) != size_t{hdr.payload_len}) \ if (read(sock, buf) != static_cast<ptrdiff_t>(hdr.payload_len)) \
CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); \ CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); \
binary_deserializer source{sys, buf}; \ binary_deserializer source{sys, buf}; \
if (auto err = source(__VA_ARGS__)) \ if (auto err = source(__VA_ARGS__)) \
......
...@@ -75,8 +75,10 @@ struct fixture : host_fixture { ...@@ -75,8 +75,10 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(network_socket_tests, fixture) CAF_TEST_FIXTURE_SCOPE(network_socket_tests, fixture)
CAF_TEST(read on empty sockets) { CAF_TEST(read on empty sockets) {
CAF_CHECK_EQUAL(read(first, rd_buf), sec::unavailable_or_would_block); CAF_CHECK_LESS(read(first, rd_buf), 0);
CAF_CHECK_EQUAL(read(second, rd_buf), sec::unavailable_or_would_block); CAF_CHECK(last_socket_error_is_temporary());
CAF_CHECK_LESS(read(second, rd_buf), 0);
CAF_CHECK(last_socket_error_is_temporary());
} }
CAF_TEST(transfer data from first to second socket) { CAF_TEST(transfer data from first to second socket) {
...@@ -97,7 +99,7 @@ CAF_TEST(transfer data from second to first socket) { ...@@ -97,7 +99,7 @@ CAF_TEST(transfer data from second to first socket) {
CAF_TEST(shut down first socket and observe shutdown on the second one) { CAF_TEST(shut down first socket and observe shutdown on the second one) {
close(first); close(first);
CAF_CHECK_EQUAL(read(second, rd_buf), sec::socket_disconnected); CAF_CHECK_EQUAL(read(second, rd_buf), 0);
first.id = invalid_socket_id; first.id = invalid_socket_id;
} }
......
...@@ -184,9 +184,11 @@ CAF_TEST(resolve and proxy communication) { ...@@ -184,9 +184,11 @@ CAF_TEST(resolve and proxy communication) {
[&] { CAF_FAIL("manager did not respond with a proxy."); }); [&] { CAF_FAIL("manager did not respond with a proxy."); });
run(); run();
auto read_res = read(recv_socket_guard.socket(), recv_buf); auto read_res = read(recv_socket_guard.socket(), recv_buf);
if (!holds_alternative<size_t>(read_res)) if (read_res < 0)
CAF_FAIL("read() returned an error: " << get<sec>(read_res)); CAF_FAIL("read() returned an error: " << last_socket_error_as_string);
recv_buf.resize(get<size_t>(read_res)); else if (read_res == 0)
CAF_FAIL("read() returned 0 (socket closed)");
recv_buf.resize(static_cast<size_t>(read_res));
CAF_MESSAGE("receive buffer contains " << recv_buf.size() << " bytes"); CAF_MESSAGE("receive buffer contains " << recv_buf.size() << " bytes");
message msg; message msg;
binary_deserializer source{sys, recv_buf}; binary_deserializer source{sys, recv_buf};
......
...@@ -208,8 +208,8 @@ CAF_TEST(receive) { ...@@ -208,8 +208,8 @@ CAF_TEST(receive) {
auto buf = std::make_shared<byte_buffer>(); auto buf = std::make_shared<byte_buffer>();
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
CAF_CHECK_EQUAL(nonblocking(sockets.second, true), none); CAF_CHECK_EQUAL(nonblocking(sockets.second, true), none);
CAF_CHECK_EQUAL(read(sockets.second, read_buf), CAF_CHECK_LESS(read(sockets.second, read_buf), 0);
sec::unavailable_or_would_block); CAF_CHECK(last_socket_error_is_temporary());
CAF_MESSAGE("adding both endpoint managers"); CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager( auto mgr1 = make_endpoint_manager(
mpx, sys, transport_type{sockets.first, application_type{buf}}); mpx, sys, transport_type{sockets.first, application_type{buf}});
......
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