Commit ab27b354 authored by Dominik Charousset's avatar Dominik Charousset

used fd_util functions for file handle manipulations

parent 6118ec7a
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include <iostream> #include <iostream>
#include "cppa/exception.hpp" #include "cppa/exception.hpp"
#include "cppa/util/io_stream.hpp" #include "cppa/util/io_stream.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/detail/ipv4_acceptor.hpp" #include "cppa/detail/ipv4_acceptor.hpp"
#include "cppa/detail/ipv4_io_stream.hpp" #include "cppa/detail/ipv4_io_stream.hpp"
...@@ -50,6 +51,8 @@ ...@@ -50,6 +51,8 @@
namespace cppa { namespace detail { namespace cppa { namespace detail {
using namespace fd_util;
namespace { namespace {
struct socket_guard { struct socket_guard {
...@@ -71,14 +74,6 @@ struct socket_guard { ...@@ -71,14 +74,6 @@ struct socket_guard {
}; };
int rd_flags(native_socket_type fd) {
auto flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
throw network_error("unable to read socket flags");
}
return flags;
}
bool accept_impl(util::io_stream_ptr_pair& result, native_socket_type fd, bool nonblocking) { bool accept_impl(util::io_stream_ptr_pair& result, native_socket_type fd, bool nonblocking) {
sockaddr addr; sockaddr addr;
socklen_t addrlen; socklen_t addrlen;
...@@ -90,13 +85,8 @@ bool accept_impl(util::io_stream_ptr_pair& result, native_socket_type fd, bool n ...@@ -90,13 +85,8 @@ bool accept_impl(util::io_stream_ptr_pair& result, native_socket_type fd, bool n
// ok, try again // ok, try again
return false; return false;
} }
char* cstr = strerror(errno); throw_io_failure("accept failed");
std::string errmsg = cstr;
free(cstr);
throw std::ios_base::failure(std::move(errmsg));
} }
int flags = 1;
setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(int));
util::io_stream_ptr ptr(ipv4_io_stream::from_native_socket(sfd)); util::io_stream_ptr ptr(ipv4_io_stream::from_native_socket(sfd));
result.first = ptr; result.first = ptr;
result.second = ptr; result.second = ptr;
...@@ -115,7 +105,7 @@ std::unique_ptr<util::acceptor> ipv4_acceptor::create(std::uint16_t port) { ...@@ -115,7 +105,7 @@ std::unique_ptr<util::acceptor> ipv4_acceptor::create(std::uint16_t port) {
if (sockfd == invalid_socket) { if (sockfd == invalid_socket) {
throw network_error("could not create server socket"); throw network_error("could not create server socket");
} }
// sguard closes the socket in case of an exception // sguard closes the socket in case of exception
socket_guard sguard(sockfd); socket_guard sguard(sockfd);
memset((char*) &serv_addr, 0, sizeof(serv_addr)); memset((char*) &serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET; serv_addr.sin_family = AF_INET;
...@@ -127,14 +117,11 @@ std::unique_ptr<util::acceptor> ipv4_acceptor::create(std::uint16_t port) { ...@@ -127,14 +117,11 @@ std::unique_ptr<util::acceptor> ipv4_acceptor::create(std::uint16_t port) {
if (listen(sockfd, 10) != 0) { if (listen(sockfd, 10) != 0) {
throw network_error("listen() failed"); throw network_error("listen() failed");
} }
int flags = fcntl(sockfd, F_GETFL, 0); // default mode is nonblocking
if (flags == -1) { nonblocking(sockfd, true);
throw network_error("unable to get socket flags");
}
bool nonblocking = (flags & O_NONBLOCK) != 0;
// ok, no exceptions // ok, no exceptions
sguard.release(); sguard.release();
std::unique_ptr<ipv4_acceptor> result(new ipv4_acceptor(sockfd, nonblocking)); std::unique_ptr<ipv4_acceptor> result(new ipv4_acceptor(sockfd, true));
return std::move(result); return std::move(result);
} }
...@@ -148,10 +135,7 @@ native_socket_type ipv4_acceptor::acceptor_file_handle() const { ...@@ -148,10 +135,7 @@ native_socket_type ipv4_acceptor::acceptor_file_handle() const {
util::io_stream_ptr_pair ipv4_acceptor::accept_connection() { util::io_stream_ptr_pair ipv4_acceptor::accept_connection() {
if (m_is_nonblocking) { if (m_is_nonblocking) {
auto flags = rd_flags(m_fd); nonblocking(m_fd, false);
if (fcntl(m_fd, F_SETFL, flags & (~(O_NONBLOCK))) < 0) {
throw network_error("unable to set socket to nonblock");
}
m_is_nonblocking = false; m_is_nonblocking = false;
} }
util::io_stream_ptr_pair result; util::io_stream_ptr_pair result;
...@@ -161,10 +145,7 @@ util::io_stream_ptr_pair ipv4_acceptor::accept_connection() { ...@@ -161,10 +145,7 @@ util::io_stream_ptr_pair ipv4_acceptor::accept_connection() {
option<util::io_stream_ptr_pair> ipv4_acceptor::try_accept_connection() { option<util::io_stream_ptr_pair> ipv4_acceptor::try_accept_connection() {
if (!m_is_nonblocking) { if (!m_is_nonblocking) {
auto flags = rd_flags(m_fd); nonblocking(m_fd, true);
if (fcntl(m_fd, F_SETFL, flags | O_NONBLOCK) < 0) {
throw network_error("unable to set socket to nonblock");
}
m_is_nonblocking = true; m_is_nonblocking = true;
} }
util::io_stream_ptr_pair result; util::io_stream_ptr_pair result;
......
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include <iostream> #include <iostream>
#include "cppa/exception.hpp" #include "cppa/exception.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/detail/ipv4_io_stream.hpp" #include "cppa/detail/ipv4_io_stream.hpp"
#ifdef CPPA_WINDOWS #ifdef CPPA_WINDOWS
...@@ -49,41 +50,7 @@ ...@@ -49,41 +50,7 @@
namespace cppa { namespace detail { namespace cppa { namespace detail {
namespace { using namespace fd_util;
template<typename T>
void handle_syscall_result(T result, bool is_recv_result) {
if (result < 0) {
if (errno != EAGAIN && errno != EWOULDBLOCK) {
char* cstr = strerror(errno);
std::string errmsg = cstr;
errmsg += " [errno = ";
errmsg += std::to_string(errno);
errmsg += "]";
throw std::ios_base::failure(std::move(errmsg));
}
}
else if (result == 0 && is_recv_result) {
throw std::ios_base::failure("cannot read from closed socket");
}
}
int rd_flags(native_socket_type fd) {
auto flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
throw network_error("unable to read socket flags");
}
return flags;
}
void set_nonblocking(native_socket_type fd) {
auto flags = rd_flags(fd);
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) {
throw network_error("unable to set socket to nonblock");
}
}
} // namespace <anonymous>
ipv4_io_stream::ipv4_io_stream(native_socket_type fd) : m_fd(fd) { } ipv4_io_stream::ipv4_io_stream(native_socket_type fd) : m_fd(fd) { }
...@@ -100,7 +67,7 @@ void ipv4_io_stream::read(void* vbuf, size_t len) { ...@@ -100,7 +67,7 @@ void ipv4_io_stream::read(void* vbuf, size_t len) {
size_t rd = 0; size_t rd = 0;
while (rd < len) { while (rd < len) {
auto recv_result = ::recv(m_fd, buf + rd, len - rd, 0); auto recv_result = ::recv(m_fd, buf + rd, len - rd, 0);
handle_syscall_result(recv_result, true); handle_read_result(recv_result, true);
if (recv_result > 0) { if (recv_result > 0) {
rd += static_cast<size_t>(recv_result); rd += static_cast<size_t>(recv_result);
} }
...@@ -117,7 +84,7 @@ void ipv4_io_stream::read(void* vbuf, size_t len) { ...@@ -117,7 +84,7 @@ void ipv4_io_stream::read(void* vbuf, size_t len) {
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, true); handle_read_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;
} }
...@@ -126,7 +93,7 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) { ...@@ -126,7 +93,7 @@ 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, false); handle_write_result(send_result, true);
if (send_result > 0) { if (send_result > 0) {
written += static_cast<size_t>(send_result); written += static_cast<size_t>(send_result);
} }
...@@ -144,22 +111,22 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) { ...@@ -144,22 +111,22 @@ 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, false); handle_write_result(send_result, true);
return static_cast<size_t>(send_result); return static_cast<size_t>(send_result);
} }
util::io_stream_ptr ipv4_io_stream::from_native_socket(native_socket_type fd) { util::io_stream_ptr ipv4_io_stream::from_native_socket(native_socket_type fd) {
set_nonblocking(fd); tcp_nodelay(fd, true);
nonblocking(fd, true);
return new ipv4_io_stream(fd); return new ipv4_io_stream(fd);
} }
util::io_stream_ptr ipv4_io_stream::connect_to(const char* host, util::io_stream_ptr ipv4_io_stream::connect_to(const char* host,
std::uint16_t port) { std::uint16_t port) {
native_socket_type sockfd;
struct sockaddr_in serv_addr; struct sockaddr_in serv_addr;
struct hostent* server; struct hostent* server;
sockfd = socket(AF_INET, SOCK_STREAM, 0); native_socket_type fd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == invalid_socket) { if (fd == invalid_socket) {
throw network_error("socket creation failed"); throw network_error("socket creation failed");
} }
server = gethostbyname(host); server = gethostbyname(host);
...@@ -172,13 +139,12 @@ util::io_stream_ptr ipv4_io_stream::connect_to(const char* host, ...@@ -172,13 +139,12 @@ util::io_stream_ptr ipv4_io_stream::connect_to(const char* host,
serv_addr.sin_family = AF_INET; serv_addr.sin_family = AF_INET;
memmove(&serv_addr.sin_addr.s_addr, server->h_addr, server->h_length); memmove(&serv_addr.sin_addr.s_addr, server->h_addr, server->h_length);
serv_addr.sin_port = htons(port); serv_addr.sin_port = htons(port);
if (connect(sockfd, (const sockaddr*) &serv_addr, sizeof(serv_addr)) != 0) { if (connect(fd, (const sockaddr*) &serv_addr, sizeof(serv_addr)) != 0) {
throw network_error("could not connect to host"); throw network_error("could not connect to host");
} }
int flags = 1; tcp_nodelay(fd, true);
setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(int)); nonblocking(fd, true);
set_nonblocking(sockfd); return new ipv4_io_stream(fd);
return new ipv4_io_stream(sockfd);
} }
} } // namespace cppa::detail } } // namespace cppa::detail
...@@ -315,7 +315,7 @@ class middleman { ...@@ -315,7 +315,7 @@ class middleman {
} }
inline void add_channel_ptr(network_channel_ptr ptr) { inline void add_channel_ptr(network_channel_ptr ptr) {
m_new_channels.push_back(std::move(ptr)); m_new_channels.push_back(move(ptr));
} }
inline void add_peer(const process_information& pinf, peer_connection_ptr cptr) { inline void add_peer(const process_information& pinf, peer_connection_ptr cptr) {
...@@ -687,7 +687,7 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) { ...@@ -687,7 +687,7 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) {
if (FD_ISSET(channel->read_handle(), &rdset)) { if (FD_ISSET(channel->read_handle(), &rdset)) {
bool erase_channel = false; bool erase_channel = false;
try { erase_channel = !channel->continue_reading(); } try { erase_channel = !channel->continue_reading(); }
catch (std::exception& e) { catch (exception& e) {
DEBUG(demangle(typeid(e).name()) << ": " << e.what()); DEBUG(demangle(typeid(e).name()) << ": " << e.what());
erase_channel = true; erase_channel = true;
} }
...@@ -704,7 +704,7 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) { ...@@ -704,7 +704,7 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) {
if (FD_ISSET(peer->write_handle(), &wrset)) { if (FD_ISSET(peer->write_handle(), &wrset)) {
bool erase_channel = false; bool erase_channel = false;
try { erase_channel = !peer->continue_writing(); } try { erase_channel = !peer->continue_writing(); }
catch (std::exception& e) { catch (exception& e) {
DEBUG(demangle(typeid(e).name()) << ": " << e.what()); DEBUG(demangle(typeid(e).name()) << ": " << e.what());
erase_channel = true; erase_channel = true;
} }
......
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