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

Merge pull request #846

Fix disconnect issue on Windows
parents 2f6ad935 eed5e9ca
...@@ -79,6 +79,9 @@ bool would_block_or_temporarily_unavailable(int errcode); ...@@ -79,6 +79,9 @@ bool would_block_or_temporarily_unavailable(int errcode);
/// Returns the last socket error as human-readable string. /// Returns the last socket error as human-readable string.
std::string last_socket_error_as_string(); std::string last_socket_error_as_string();
/// Returns a human-readable string for a given socket error.
std::string socket_error_as_string(int errcode);
/// Creates two connected sockets. The former is the read handle /// Creates two connected sockets. The former is the read handle
/// and the latter is the write handle. /// and the latter is the write handle.
std::pair<native_socket, native_socket> create_pipe(); std::pair<native_socket, native_socket> create_pipe();
......
...@@ -339,8 +339,6 @@ namespace network { ...@@ -339,8 +339,6 @@ namespace network {
presult = ::poll(pollset_.data(), presult = ::poll(pollset_.data(),
static_cast<nfds_t>(pollset_.size()), block ? -1 : 0); static_cast<nfds_t>(pollset_.size()), block ? -1 : 0);
# endif # endif
CAF_LOG_DEBUG("poll() on" << pollset_.size()
<< "sockets reported" << presult << "event(s)");
if (presult < 0) { if (presult < 0) {
switch (last_socket_error()) { switch (last_socket_error()) {
case EINTR: { case EINTR: {
...@@ -362,6 +360,8 @@ namespace network { ...@@ -362,6 +360,8 @@ namespace network {
} }
continue; // rinse and repeat continue; // rinse and repeat
} }
CAF_LOG_DEBUG("poll() on" << pollset_.size()
<< "sockets reported" << presult << "event(s)");
if (presult == 0) if (presult == 0)
return false; return false;
// scan pollset for events first, because we might alter pollset_ // scan pollset for events first, because we might alter pollset_
......
...@@ -125,6 +125,11 @@ namespace network { ...@@ -125,6 +125,11 @@ namespace network {
return strerror(errno); return strerror(errno);
} }
string socket_error_as_string(int errcode) {
return strerror(errcode);
}
expected<void> child_process_inherit(native_socket fd, bool new_value) { expected<void> child_process_inherit(native_socket fd, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(new_value)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(new_value));
// read flags for fd // read flags for fd
...@@ -187,9 +192,12 @@ namespace network { ...@@ -187,9 +192,12 @@ namespace network {
} }
string last_socket_error_as_string() { string last_socket_error_as_string() {
return socket_error_as_string(last_socket_error());
}
string socket_error_as_string(int errcode) {
LPTSTR errorText = NULL; LPTSTR errorText = NULL;
auto hresult = last_socket_error(); FormatMessage(// use system message tables to retrieve error text
FormatMessage( // use system message tables to retrieve error text
FORMAT_MESSAGE_FROM_SYSTEM FORMAT_MESSAGE_FROM_SYSTEM
// allocate buffer on local heap for error text // allocate buffer on local heap for error text
| FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_ALLOCATE_BUFFER
...@@ -197,7 +205,7 @@ namespace network { ...@@ -197,7 +205,7 @@ namespace network {
// (and CANNOT) pass insertion parameters // (and CANNOT) pass insertion parameters
| FORMAT_MESSAGE_IGNORE_INSERTS, | FORMAT_MESSAGE_IGNORE_INSERTS,
nullptr, // unused with FORMAT_MESSAGE_FROM_SYSTEM nullptr, // unused with FORMAT_MESSAGE_FROM_SYSTEM
hresult, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), errcode, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
(LPTSTR) & errorText, // output (LPTSTR) & errorText, // output
0, // minimum size for output buffer 0, // minimum size for output buffer
nullptr); // arguments - see note nullptr); // arguments - see note
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <cstring> #include <cstring>
#include "caf/io/network/native_socket.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
...@@ -33,7 +34,9 @@ using caf::io::network::is_error; ...@@ -33,7 +34,9 @@ using caf::io::network::is_error;
using caf::io::network::rw_state; using caf::io::network::rw_state;
using caf::io::network::native_socket; using caf::io::network::native_socket;
using caf::io::network::socket_size_type; using caf::io::network::socket_size_type;
using caf::io::network::last_socket_error;
using caf::io::network::no_sigpipe_io_flag; using caf::io::network::no_sigpipe_io_flag;
using caf::io::network::socket_error_as_string;
namespace caf { namespace caf {
namespace policy { namespace policy {
...@@ -43,11 +46,18 @@ rw_state tcp::read_some(size_t& result, native_socket fd, void* buf, ...@@ -43,11 +46,18 @@ rw_state tcp::read_some(size_t& result, native_socket fd, void* buf,
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::recv(fd, reinterpret_cast<io::network::socket_recv_ptr>(buf), auto sres = ::recv(fd, reinterpret_cast<io::network::socket_recv_ptr>(buf),
len, no_sigpipe_io_flag); len, no_sigpipe_io_flag);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres)); if (is_error(sres, true)) {
if (is_error(sres, true) || sres == 0) { // Make sure WSAGetLastError gets called immediately on Windows.
auto err = last_socket_error();
CAF_IGNORE_UNUSED(err);
CAF_LOG_ERROR("recv failed:" << socket_error_as_string(err));
return rw_state::failure;
} else if (sres == 0) {
// recv returns 0 when the peer has performed an orderly shutdown // recv returns 0 when the peer has performed an orderly shutdown
CAF_LOG_DEBUG("peer performed orderly shutdown" << CAF_ARG(fd));
return rw_state::failure; return rw_state::failure;
} }
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
result = (sres > 0) ? static_cast<size_t>(sres) : 0; result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return rw_state::success; return rw_state::success;
} }
...@@ -57,9 +67,14 @@ rw_state tcp::write_some(size_t& result, native_socket fd, const void* buf, ...@@ -57,9 +67,14 @@ rw_state tcp::write_some(size_t& result, native_socket fd, const void* buf,
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::send(fd, reinterpret_cast<io::network::socket_send_ptr>(buf), auto sres = ::send(fd, reinterpret_cast<io::network::socket_send_ptr>(buf),
len, no_sigpipe_io_flag); len, no_sigpipe_io_flag);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres)); if (is_error(sres, true)) {
if (is_error(sres, true)) // Make sure WSAGetLastError gets called immediately on Windows.
auto err = last_socket_error();
CAF_IGNORE_UNUSED(err);
CAF_LOG_ERROR("send failed:" << socket_error_as_string(err));
return rw_state::failure; return rw_state::failure;
}
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
result = (sres > 0) ? static_cast<size_t>(sres) : 0; result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return rw_state::success; return rw_state::success;
} }
...@@ -72,14 +87,15 @@ bool tcp::try_accept(native_socket& result, native_socket fd) { ...@@ -72,14 +87,15 @@ bool tcp::try_accept(native_socket& result, native_socket fd) {
socket_size_type addrlen = sizeof(addr); socket_size_type addrlen = sizeof(addr);
result = ::accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen); result = ::accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
// note accept4 is better to avoid races in setting CLOEXEC (but not posix) // note accept4 is better to avoid races in setting CLOEXEC (but not posix)
child_process_inherit(result, false);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
if (result == invalid_native_socket) { if (result == invalid_native_socket) {
auto err = last_socket_error(); auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) { if (!would_block_or_temporarily_unavailable(err)) {
CAF_LOG_ERROR("accept failed:" << socket_error_as_string(err));
return false; return false;
} }
} }
child_process_inherit(result, false);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
return true; return true;
} }
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include "caf/policy/udp.hpp" #include "caf/policy/udp.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
...@@ -32,6 +33,8 @@ using caf::io::network::ip_endpoint; ...@@ -32,6 +33,8 @@ using caf::io::network::ip_endpoint;
using caf::io::network::native_socket; using caf::io::network::native_socket;
using caf::io::network::signed_size_type; using caf::io::network::signed_size_type;
using caf::io::network::socket_size_type; using caf::io::network::socket_size_type;
using caf::io::network::last_socket_error;
using caf::io::network::socket_error_as_string;
namespace caf { namespace caf {
namespace policy { namespace policy {
...@@ -44,7 +47,10 @@ bool udp::read_datagram(size_t& result, native_socket fd, void* buf, ...@@ -44,7 +47,10 @@ bool udp::read_datagram(size_t& result, native_socket fd, void* buf,
auto sres = ::recvfrom(fd, static_cast<io::network::socket_recv_ptr>(buf), auto sres = ::recvfrom(fd, static_cast<io::network::socket_recv_ptr>(buf),
buf_len, 0, ep.address(), &len); buf_len, 0, ep.address(), &len);
if (is_error(sres, true)) { if (is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres)); // Make sure WSAGetLastError gets called immediately on Windows.
auto err = last_socket_error();
CAF_IGNORE_UNUSED(err);
CAF_LOG_ERROR("recvfrom failed:" << socket_error_as_string(err));
return false; return false;
} }
if (sres == 0) if (sres == 0)
...@@ -64,7 +70,10 @@ bool udp::write_datagram(size_t& result, native_socket fd, void* buf, ...@@ -64,7 +70,10 @@ bool udp::write_datagram(size_t& result, native_socket fd, void* buf,
auto sres = ::sendto(fd, reinterpret_cast<io::network::socket_send_ptr>(buf), auto sres = ::sendto(fd, reinterpret_cast<io::network::socket_send_ptr>(buf),
buf_len, 0, ep.caddress(), len); buf_len, 0, ep.caddress(), len);
if (is_error(sres, true)) { if (is_error(sres, true)) {
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres)); // Make sure WSAGetLastError gets called immediately on Windows.
auto err = last_socket_error();
CAF_IGNORE_UNUSED(err);
CAF_LOG_ERROR("sendto failed:" << socket_error_as_string(err));
return false; return false;
} }
result = (sres > 0) ? static_cast<size_t>(sres) : 0; result = (sres > 0) ? static_cast<size_t>(sres) : 0;
......
...@@ -81,13 +81,15 @@ struct ssl_policy { ...@@ -81,13 +81,15 @@ struct ssl_policy {
caf::io::network::socket_size_type addrlen = sizeof(addr); caf::io::network::socket_size_type addrlen = sizeof(addr);
result = accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen); result = accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
// note accept4 is better to avoid races in setting CLOEXEC (but not posix) // note accept4 is better to avoid races in setting CLOEXEC (but not posix)
io::network::child_process_inherit(result, false);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
if (result == io::network::invalid_native_socket) { if (result == io::network::invalid_native_socket) {
auto err = io::network::last_socket_error(); auto err = io::network::last_socket_error();
if (!io::network::would_block_or_temporarily_unavailable(err)) if (!io::network::would_block_or_temporarily_unavailable(err))
CAF_LOG_ERROR("accept failed:"
<< io::network::socket_error_as_string(err));
return false; return false;
} }
io::network::child_process_inherit(result, false);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
return session_->try_accept(result); return session_->try_accept(result);
} }
......
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