Unverified Commit dd16e3ec authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #85

Switch to more lightweight socket API wrapper
parents b20f0ad8 4e522052
......@@ -82,9 +82,8 @@ public:
system_ = &parent.system();
executor_.system_ptr(system_);
executor_.proxy_registry_ptr(&proxies_);
// TODO: use `if constexpr` when switching to C++17.
// 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();
size_t workers;
if (auto workers_cfg = get_if<size_t>(&system_->config(),
......
......@@ -27,6 +27,10 @@
#include "caf/net/socket.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 {
/// A unidirectional communication endpoint for inter-process communication.
......@@ -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.
/// @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);
ptrdiff_t CAF_NET_EXPORT write(pipe_socket x, span<const byte> buf);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @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> 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);
ptrdiff_t CAF_NET_EXPORT read(pipe_socket x, span<byte> buf);
} // namespace caf::net
......@@ -69,7 +69,7 @@ To CAF_NET_EXPORT socket_cast(From x) {
return To{x.id};
}
/// Close socket `x`.
/// Closes socket `x`.
/// @relates socket
void CAF_NET_EXPORT close(socket x);
......@@ -77,6 +77,14 @@ void CAF_NET_EXPORT close(socket x);
/// @relates socket
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.
/// @relates socket
std::string CAF_NET_EXPORT last_socket_error_as_string();
......
......@@ -18,10 +18,16 @@
#pragma once
#include <cstddef>
#include "caf/detail/net_export.hpp"
#include "caf/fwd.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 {
/// A connection-oriented network communication endpoint for bidirectional byte
......@@ -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);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param x A connected endpoint.
/// @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
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> CAF_NET_EXPORT read(stream_socket x, span<byte> buf);
/// @post Either the functions returned a non-negative integer or the caller can
/// 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.
/// @param x Connected endpoint.
/// Sends data to `x`.
/// @param x A connected endpoint.
/// @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
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> CAF_NET_EXPORT write(stream_socket x,
span<const byte> buf);
ptrdiff_t CAF_NET_EXPORT write(stream_socket x, span<const byte> buf);
/// 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
/// buffers.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10`
variant<size_t, sec> CAF_NET_EXPORT
write(stream_socket x, 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);
ptrdiff_t CAF_NET_EXPORT write(stream_socket x,
std::initializer_list<span<const byte>> bufs);
} // namespace caf::net
......@@ -73,17 +73,22 @@ public:
bool handle_read_event(endpoint_manager&) override {
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) {
auto buf = this->read_buf_.data() + this->collected_;
size_t len = this->read_threshold_ - this->collected_;
auto buf = this->read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_;
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.
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len)
<< CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
this->collected_ += *num_bytes;
if (this->collected_ >= this->read_threshold_) {
if (num_bytes > 0) {
collected_ += num_bytes;
if (collected_ >= read_threshold_) {
if (auto err = this->next_layer_.handle_data(
*this, make_span(this->read_buf_))) {
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err));
......@@ -91,15 +96,16 @@ public:
}
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 {
auto err = get<sec>(ret);
if (err == sec::unavailable_or_would_block) {
break;
} else {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
// read() returns 0 iff the connection was closed.
return fail(sec::socket_disconnected);
}
}
return true;
......@@ -125,27 +131,33 @@ public:
}
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.
while (!write_queue_.empty()) {
auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_;
auto len = buf.size() - written_;
auto write_ret = write(this->handle(), make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
written_ += *num_bytes;
if (written_ >= buf.size()) {
auto num_bytes = write(this->handle(), make_span(data, len));
if (num_bytes > 0) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(num_bytes));
written_ += num_bytes;
if (written_ >= static_cast<ptrdiff_t>(buf.size())) {
recycle();
written_ = 0;
}
} else if (num_bytes < 0) {
return last_socket_error_is_temporary()
? sec::unavailable_or_would_block
: fail(sec::socket_operation_failed);
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
}
return err;
// write() returns 0 iff the connection was closed.
return fail(sec::socket_disconnected);
}
}
return none;
......@@ -212,9 +224,9 @@ private:
}
write_queue_type write_queue_;
size_t written_;
size_t read_threshold_;
size_t collected_;
ptrdiff_t written_;
ptrdiff_t read_threshold_;
ptrdiff_t collected_;
size_t max_;
receive_policy_flag rd_flag_;
};
......
......@@ -155,15 +155,40 @@ bool multiplexer::poll_once(bool blocking) {
return false;
// We'll call poll() until poll() succeeds or fails.
for (;;) {
int presult;
int presult =
#ifdef CAF_WINDOWS
presult = ::WSAPoll(pollset_.data(), static_cast<ULONG>(pollset_.size()),
blocking ? -1 : 0);
::WSAPoll(pollset_.data(), static_cast<ULONG>(pollset_.size()),
blocking ? -1 : 0);
#else
presult = ::poll(pollset_.data(), static_cast<nfds_t>(pollset_.size()),
blocking ? -1 : 0);
::poll(pollset_.data(), static_cast<nfds_t>(pollset_.size()),
blocking ? -1 : 0);
#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();
switch (code) {
case std::errc::interrupted: {
......@@ -186,33 +211,7 @@ bool multiplexer::poll_once(bool blocking) {
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) {
buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(mgr.get());
memcpy(buf.data() + 1, &value, sizeof(intptr_t));
variant<size_t, sec> res;
ptrdiff_t res = -1;
{ // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket)
res = write(write_handle_, buf);
else
res = sec::socket_invalid;
}
if (holds_alternative<sec>(res) && opcode != 4)
if (res <= 0 && opcode != 4)
mgr->deref();
}
......
......@@ -105,7 +105,8 @@ void middleman::add_module_options(actor_system_config& cfg) {
"max. number of consecutive reads per broker")
.add<bool>("manual-multiplexing",
"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) {
......
......@@ -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.
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.
return read(socket_cast<stream_socket>(x), buf);
}
......@@ -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]});
}
variant<size_t, sec> write(pipe_socket x, span<const byte> buf) {
auto res = ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size());
return check_pipe_socket_io_res(res);
ptrdiff_t write(pipe_socket x, span<const byte> buf) {
return ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size());
}
variant<size_t, sec> read(pipe_socket x, span<byte> buf) {
auto res = ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size());
return check_pipe_socket_io_res(res);
ptrdiff_t read(pipe_socket x, span<byte> buf) {
return ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size());
}
#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
......@@ -42,10 +42,10 @@ pollset_updater::~pollset_updater() {
bool pollset_updater::handle_read_event() {
for (;;) {
auto res = read(handle(), make_span(buf_.data() + buf_size_,
buf_.size() - buf_size_));
if (auto num_bytes = get_if<size_t>(&res)) {
buf_size_ += *num_bytes;
auto num_bytes = read(handle(), make_span(buf_.data() + buf_size_,
buf_.size() - buf_size_));
if (num_bytes > 0) {
buf_size_ += static_cast<size_t>(num_bytes);
if (buf_.size() == buf_size_) {
buf_size_ = 0;
auto opcode = static_cast<uint8_t>(buf_[0]);
......@@ -70,7 +70,7 @@ bool pollset_updater::handle_read_event() {
}
}
} 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() {
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() {
int wsa_code = WSAGetLastError();
LPTSTR errorText = NULL;
......@@ -149,6 +154,15 @@ std::errc last_socket_error() {
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() {
return strerror(errno);
}
......
......@@ -168,22 +168,19 @@ error nodelay(stream_socket x, bool new_value) {
return none;
}
variant<size_t, sec> read(stream_socket x, span<byte> buf) {
auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size(), no_sigpipe_io_flag);
return check_stream_socket_io_res(res);
ptrdiff_t read(stream_socket x, span<byte> buf) {
return ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag);
}
variant<size_t, sec> write(stream_socket x, span<const byte> buf) {
auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), no_sigpipe_io_flag);
return check_stream_socket_io_res(res);
ptrdiff_t write(stream_socket x, span<const byte> buf) {
return ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag);
}
#ifdef CAF_WINDOWS
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs) {
ptrdiff_t write(stream_socket x, std::initializer_list<span<const byte>> bufs) {
CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10];
auto convert = [](span<const byte> buf) {
......@@ -195,44 +192,21 @@ variant<size_t, sec> write(stream_socket x,
DWORD bytes_sent = 0;
auto res = WSASend(x.id, buf_array, static_cast<DWORD>(bufs.size()),
&bytes_sent, 0, nullptr, nullptr);
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>(bytes_sent);
return (res == 0) ? bytes_sent : -1;
}
#else // CAF_WINDOWS
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs) {
ptrdiff_t write(stream_socket x, std::initializer_list<span<const byte>> bufs) {
CAF_ASSERT(bufs.size() < 10);
iovec buf_array[10];
auto convert = [](span<const byte> buf) {
return iovec{const_cast<byte*>(buf.data()), buf.size()};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
auto res = writev(x.id, buf_array, static_cast<int>(bufs.size()));
return check_stream_socket_io_res(res);
return writev(x.id, buf_array, static_cast<int>(bufs.size()));
}
#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
......@@ -29,6 +29,7 @@
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/none.hpp"
#include "caf/uri.hpp"
......@@ -42,7 +43,13 @@ using namespace caf::net;
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,
basp::application::test_tag,
public packet_writer {
......@@ -242,8 +249,7 @@ CAF_TEST(actor message) {
MOCK(basp::message_type::actor_message, make_message_id().integer_value(),
mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{},
make_message("hello world!"));
allow((monitor_atom, strong_actor_ptr),
from(_).to(self).with(monitor_atom_v, _));
expect((monitor_atom, strong_actor_ptr), from(_).to(self));
expect((std::string), from(_).to(self).with("hello world!"));
}
......
......@@ -90,13 +90,13 @@ public:
template <class Manager>
bool handle_read_event(Manager&) {
auto res = read(handle_, read_buf_);
if (auto num_bytes = get_if<size_t>(&res)) {
auto num_bytes = read(handle_, read_buf_);
if (num_bytes > 0) {
data_->insert(data_->end(), read_buf_.begin(),
read_buf_.begin() + *num_bytes);
read_buf_.begin() + num_bytes);
return true;
}
return get<sec>(res) == sec::unavailable_or_would_block;
return num_bytes < 0 && last_socket_error_is_temporary();
}
template <class Manager>
......@@ -106,12 +106,12 @@ public:
if (auto err = sink(x->msg->payload))
CAF_FAIL("serializing failed: " << err);
}
auto res = write(handle_, buf_);
if (auto num_bytes = get_if<size_t>(&res)) {
buf_.erase(buf_.begin(), buf_.begin() + *num_bytes);
auto num_bytes = write(handle_, buf_);
if (num_bytes > 0) {
buf_.erase(buf_.begin(), buf_.begin() + num_bytes);
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) {
......@@ -164,8 +164,8 @@ CAF_TEST(send and receive) {
auto buf = std::make_shared<byte_buffer>();
auto sockets = unbox(make_stream_socket_pair());
CAF_CHECK_EQUAL(nonblocking(sockets.second, true), none);
CAF_CHECK_EQUAL(read(sockets.second, read_buf),
sec::unavailable_or_would_block);
CAF_CHECK_LESS(read(sockets.second, read_buf), 0);
CAF_CHECK(last_socket_error_is_temporary());
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf});
......@@ -208,11 +208,14 @@ CAF_TEST(resolve and proxy communication) {
[&] { CAF_FAIL("manager did not respond with a proxy."); });
run();
auto read_res = read(sockets.second, read_buf);
if (!holds_alternative<size_t>(read_res)) {
CAF_ERROR("read() returned an error: " << get<sec>(read_res));
if (read_res <= 0) {
std::string msg = "socket closed";
if (read_res < 0)
msg = last_socket_error_as_string();
CAF_ERROR("read() failed: " << msg);
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");
message msg;
binary_deserializer source{sys, read_buf};
......
......@@ -59,26 +59,25 @@ public:
bool handle_read_event() override {
if (read_capacity() < 1024)
rd_buf_.resize(rd_buf_.size() + 2048);
auto res = read(handle(),
make_span(read_position_begin(), read_capacity()));
if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes > 0);
rd_buf_pos_ += *num_bytes;
auto num_bytes = read(handle(),
make_span(read_position_begin(), read_capacity()));
if (num_bytes > 0) {
CAF_ASSERT(num_bytes > 0);
rd_buf_pos_ += num_bytes;
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 {
if (wr_buf_.size() == 0)
return false;
auto res = write(handle(), wr_buf_);
if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes > 0);
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + *num_bytes);
auto num_bytes = write(handle(), wr_buf_);
if (num_bytes > 0) {
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + num_bytes);
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 {
......
......@@ -48,12 +48,6 @@ namespace {
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 {
config() {
put(content, "caf.middleman.this-node", unbox(make_uri("test:earth")));
......@@ -88,7 +82,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
template <class... Ts>
void mock(const Ts&... 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");
run();
}
......@@ -107,14 +101,14 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
void consume_handshake() {
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");
auto hdr = basp::header::from_bytes(buf);
if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0
|| hdr.operation_data != basp::version)
CAF_FAIL("invalid handshake header");
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");
node_id nid;
std::vector<std::string> app_ids;
......@@ -158,7 +152,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
do { \
CAF_MESSAGE("receive " << msg_type); \
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"); \
auto hdr = basp::header::from_bytes(buf); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \
......@@ -166,7 +160,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \
std::tuple<unit_t>>::value) { \
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"); \
binary_deserializer source{sys, buf}; \
if (auto err = source(__VA_ARGS__)) \
......
......@@ -75,8 +75,10 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(network_socket_tests, fixture)
CAF_TEST(read on empty sockets) {
CAF_CHECK_EQUAL(read(first, rd_buf), sec::unavailable_or_would_block);
CAF_CHECK_EQUAL(read(second, rd_buf), sec::unavailable_or_would_block);
CAF_CHECK_LESS(read(first, rd_buf), 0);
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) {
......@@ -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) {
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;
}
......
......@@ -184,9 +184,11 @@ CAF_TEST(resolve and proxy communication) {
[&] { CAF_FAIL("manager did not respond with a proxy."); });
run();
auto read_res = read(recv_socket_guard.socket(), recv_buf);
if (!holds_alternative<size_t>(read_res))
CAF_FAIL("read() returned an error: " << get<sec>(read_res));
recv_buf.resize(get<size_t>(read_res));
if (read_res < 0)
CAF_FAIL("read() returned an error: " << last_socket_error_as_string);
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");
message msg;
binary_deserializer source{sys, recv_buf};
......
......@@ -208,8 +208,8 @@ CAF_TEST(receive) {
auto buf = std::make_shared<byte_buffer>();
auto sockets = unbox(make_stream_socket_pair());
CAF_CHECK_EQUAL(nonblocking(sockets.second, true), none);
CAF_CHECK_EQUAL(read(sockets.second, read_buf),
sec::unavailable_or_would_block);
CAF_CHECK_LESS(read(sockets.second, read_buf), 0);
CAF_CHECK(last_socket_error_is_temporary());
CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(
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