Commit 81bbe38b authored by Dominik Charousset's avatar Dominik Charousset

fixed several issues with asynchronous socket IO

parent 11852b66
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#include <ios> #include <ios>
#include <cstring> #include <cstring>
#include <errno.h> #include <errno.h>
#include <iostream>
#include "cppa/exception.hpp" #include "cppa/exception.hpp"
#include "cppa/detail/ipv4_io_stream.hpp" #include "cppa/detail/ipv4_io_stream.hpp"
...@@ -51,22 +52,20 @@ namespace cppa { namespace detail { ...@@ -51,22 +52,20 @@ namespace cppa { namespace detail {
namespace { namespace {
template<typename T> template<typename T>
void handle_syscall_result(T result, size_t num_bytes, bool nonblocking) { void handle_syscall_result(T result, bool is_recv_result) {
if (result < 0) { if (result < 0) {
if (!nonblocking || (errno != EAGAIN && errno != EWOULDBLOCK)) { if (errno != EAGAIN && errno != EWOULDBLOCK) {
char* cstr = strerror(errno); char* cstr = strerror(errno);
std::string errmsg = cstr; std::string errmsg = cstr;
errmsg += " [errno = "; errmsg += " [errno = ";
errmsg += std::to_string(errno); errmsg += std::to_string(errno);
errmsg += "]"; errmsg += "]";
std::cerr << errmsg << std::endl;
throw std::ios_base::failure(std::move(errmsg)); throw std::ios_base::failure(std::move(errmsg));
} }
} }
else if (result == 0) { else if (result == 0 && is_recv_result) {
throw std::ios_base::failure("cannot read/write from closed socket"); throw std::ios_base::failure("cannot read from closed socket");
}
else if (!nonblocking && static_cast<size_t>(result) != num_bytes) {
throw std::ios_base::failure("IO error on IPv4 socket");
} }
} }
...@@ -80,10 +79,8 @@ int rd_flags(native_socket_type fd) { ...@@ -80,10 +79,8 @@ int rd_flags(native_socket_type fd) {
void set_nonblocking(native_socket_type fd) { void set_nonblocking(native_socket_type fd) {
auto flags = rd_flags(fd); auto flags = rd_flags(fd);
if ((flags & O_NONBLOCK) != 0) { if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) {
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) { throw network_error("unable to set socket to nonblock");
throw network_error("unable to set socket to nonblock");
}
} }
} }
...@@ -99,13 +96,29 @@ native_socket_type ipv4_io_stream::write_file_handle() const { ...@@ -99,13 +96,29 @@ native_socket_type ipv4_io_stream::write_file_handle() const {
return m_fd; return m_fd;
} }
void ipv4_io_stream::read(void* buf, size_t len) { void ipv4_io_stream::read(void* vbuf, size_t len) {
handle_syscall_result(::recv(m_fd, buf, len, MSG_WAITALL), len, false); auto buf = reinterpret_cast<char*>(vbuf);
size_t rd = 0;
while (rd < len) {
auto recv_result = ::recv(m_fd, buf + rd, len - rd, 0);
handle_syscall_result(recv_result, true);
if (recv_result > 0) {
rd += static_cast<size_t>(recv_result);
}
if (rd < len) {
fd_set rdset;
FD_ZERO(&rdset);
FD_SET(m_fd, &rdset);
if (select(m_fd + 1, &rdset, nullptr, nullptr, nullptr) < 0) {
throw network_error("select() failed");
}
}
}
} }
size_t ipv4_io_stream::read_some(void* buf, size_t len) { size_t ipv4_io_stream::read_some(void* buf, size_t len) {
auto recv_result = ::recv(m_fd, buf, len, 0); auto recv_result = ::recv(m_fd, buf, len, 0);
handle_syscall_result(recv_result, len, true); handle_syscall_result(recv_result, true);
return (recv_result > 0) ? static_cast<size_t>(recv_result) : 0; return (recv_result > 0) ? static_cast<size_t>(recv_result) : 0;
} }
...@@ -114,14 +127,16 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) { ...@@ -114,14 +127,16 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) {
size_t written = 0; size_t written = 0;
while (written < len) { while (written < len) {
auto send_result = ::send(m_fd, buf + written, len - written, 0); auto send_result = ::send(m_fd, buf + written, len - written, 0);
handle_syscall_result(send_result, len - written, true); handle_syscall_result(send_result, false);
written += static_cast<size_t>(send_result); if (send_result > 0) {
written += static_cast<size_t>(send_result);
}
if (written < len) { if (written < len) {
// block until socked is writable again // block until socked is writable again
fd_set writeset; fd_set wrset;
FD_ZERO(&writeset); FD_ZERO(&wrset);
FD_SET(m_fd, &writeset); FD_SET(m_fd, &wrset);
if (select(m_fd + 1, nullptr, &writeset, nullptr, nullptr) < 0) { if (select(m_fd + 1, nullptr, &wrset, nullptr, nullptr) < 0) {
throw network_error("select() failed"); throw network_error("select() failed");
} }
} }
...@@ -130,7 +145,7 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) { ...@@ -130,7 +145,7 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) {
size_t ipv4_io_stream::write_some(const void* buf, size_t len) { size_t ipv4_io_stream::write_some(const void* buf, size_t len) {
auto send_result = ::send(m_fd, buf, len, 0); auto send_result = ::send(m_fd, buf, len, 0);
handle_syscall_result(send_result, len, true); handle_syscall_result(send_result, false);
return static_cast<size_t>(send_result); return static_cast<size_t>(send_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