Commit 3ce5f985 authored by Dominik Charousset's avatar Dominik Charousset

ignore IO errors until all buffers are empty

parent aa0ff417
...@@ -53,18 +53,14 @@ constexpr event_bitmask error = 0x04; ...@@ -53,18 +53,14 @@ constexpr event_bitmask error = 0x04;
} } // namespace event } } // namespace event
/** template<unsigned InputEvent, unsigned OutputEvent, unsigned ErrorEvent>
* @brief Converts an event bitmask to a human-readable string. inline event_bitmask from_int_bitmask(unsigned mask) {
*/ event_bitmask result = event::none;
inline const char* eb2str(event_bitmask e) { // read/write as long as possible
switch (e) { if (mask & InputEvent) result = event::read;
default: return "INVALID"; if (mask & OutputEvent) result |= event::write;
case event::none: return "event::none"; if (result == event::none && mask & ErrorEvent) result = event::error;
case event::read: return "event::read"; return result;
case event::write: return "event::write";
case event::both: return "event::both";
case event::error: return "event::error";
}
} }
struct fd_meta_info { struct fd_meta_info {
......
...@@ -32,6 +32,17 @@ ...@@ -32,6 +32,17 @@
namespace cppa { namespace network { namespace cppa { namespace network {
inline std::string eb2str(event_bitmask e) {
switch (e) {
default: return "INVALID";
case event::none: return "event::none";
case event::read: return "event::read";
case event::write: return "event::write";
case event::both: return "event::both";
case event::error: return "event::error";
}
}
middleman_event_handler::middleman_event_handler() { } middleman_event_handler::middleman_event_handler() { }
middleman_event_handler::~middleman_event_handler() { } middleman_event_handler::~middleman_event_handler() { }
......
...@@ -40,6 +40,10 @@ namespace cppa { namespace network { ...@@ -40,6 +40,10 @@ namespace cppa { namespace network {
namespace { namespace {
static constexpr unsigned input_event = EPOLLIN;
static constexpr unsigned error_event = EPOLLRDHUP | EPOLLERR | EPOLLHUP;
static constexpr unsigned output_event = EPOLLOUT;
class middleman_event_handler_impl : public middleman_event_handler { class middleman_event_handler_impl : public middleman_event_handler {
public: public:
...@@ -55,47 +59,45 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -55,47 +59,45 @@ class middleman_event_handler_impl : public middleman_event_handler {
+ strerror(errno)); + strerror(errno));
} }
// handle at most 64 events at a time // handle at most 64 events at a time
m_epoll_events.resize(64); m_epollset.resize(64);
} }
protected: protected:
void poll_impl() { void poll_impl() {
CPPA_REQUIRE(m_meta.empty() == false); CPPA_REQUIRE(m_meta.empty() == false);
int presult; int presult = -1;
do { while (presult < 0) {
CPPA_LOGMF(CPPA_DEBUG, self, "epoll_wait on " << num_sockets() << " sockets"); presult = epoll_wait(m_epollfd,
presult = epoll_wait(m_epollfd, m_epoll_events.data(), (int) m_epoll_events.size(), -1); m_epollset.data(),
CPPA_LOGMF(CPPA_DEBUG, self, "epoll_wait returned " << presult); static_cast<int>(m_epollset.size()),
-1);
CPPA_LOG_DEBUG("epoll_wait on " << num_sockets()
<< " sockets returned " << presult);
if (presult < 0) { if (presult < 0) {
// try again unless critical error occured
presult = 0;
switch (errno) { switch (errno) {
// a signal was caught
case EINTR: { case EINTR: {
// a signal was caught
// just try again // just try again
break; break;
} }
default: { default: {
perror("epoll() failed"); perror("epoll_wait() failed");
CPPA_CRITICAL("epoll() failed"); CPPA_CRITICAL("epoll_wait() failed");
}
}
}
else {
for (size_t i = 0; i < static_cast<size_t>(presult); ++i) {
auto events = m_epoll_events[i].events;
auto mask = event::none;
if (events & (EPOLLRDHUP | EPOLLERR | EPOLLHUP)) mask = event::error;
else {
if (events & EPOLLIN) mask |= event::read;
if (events & EPOLLOUT) mask |= event::write;
} }
CPPA_REQUIRE(mask != 0);
m_events.emplace_back(mask, reinterpret_cast<continuable_io*>(m_epoll_events[i].data.ptr));
} }
} }
} while (presult < 0); }
auto iter = m_epollset.begin();
auto last = iter + static_cast<size_t>(presult);
for ( ; iter != last; ++iter) {
auto eb = from_int_bitmask<input_event,
output_event,
error_event>(iter->events);
auto ptr = reinterpret_cast<continuable_io*>(iter->data.ptr);
CPPA_REQUIRE(eb != event::none);
m_events.emplace_back(eb, ptr);
}
} }
void handle_event(fd_meta_event me, void handle_event(fd_meta_event me,
...@@ -159,7 +161,7 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -159,7 +161,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
private: private:
int m_epollfd; int m_epollfd;
std::vector<epoll_event> m_epoll_events; std::vector<epoll_event> m_epollset;
}; };
......
...@@ -39,6 +39,10 @@ namespace cppa { namespace network { ...@@ -39,6 +39,10 @@ namespace cppa { namespace network {
namespace { namespace {
static constexpr unsigned input_event = POLLIN | POLLPRI;
static constexpr unsigned error_event = POLLRDHUP | POLLERR | POLLHUP | POLLNVAL;
static constexpr unsigned output_event = POLLOUT;
bool pollfd_less(const pollfd& lhs, native_socket_type rhs) { bool pollfd_less(const pollfd& lhs, native_socket_type rhs) {
return lhs.fd < rhs; return lhs.fd < rhs;
} }
...@@ -63,23 +67,22 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -63,23 +67,22 @@ class middleman_event_handler_impl : public middleman_event_handler {
void poll_impl() { void poll_impl() {
CPPA_REQUIRE(m_pollset.empty() == false); CPPA_REQUIRE(m_pollset.empty() == false);
CPPA_REQUIRE(m_pollset.size() == m_meta.size()); CPPA_REQUIRE(m_pollset.size() == m_meta.size());
int presult; int presult = -1;
do { while (presult < 0) {
presult = ::poll(m_pollset.data(), m_pollset.size(), -1); presult = ::poll(m_pollset.data(), m_pollset.size(), -1);
CPPA_LOG_DEBUG("poll() on " << num_sockets() CPPA_LOG_DEBUG("poll() on " << num_sockets()
<< " sockets returned " << presult); << " sockets returned " << presult);
if (presult < 0) { if (presult < 0) {
switch (errno) { switch (errno) {
// a signal was caught
case EINTR: { case EINTR: {
// a signal was caught
// just try again // just try again
break; break;
} }
case ENOMEM: { case ENOMEM: {
CPPA_LOGMF(CPPA_ERROR, self, "poll() failed for reason ENOMEM"); CPPA_LOG_ERROR("poll() failed for reason ENOMEM");
// there's not much we can do other than try again // there's not much we can do other than try again
// in hope someone releases memory // in hope someone else releases memory
//this_thread::yield();
break; break;
} }
default: { default: {
...@@ -88,28 +91,15 @@ class middleman_event_handler_impl : public middleman_event_handler { ...@@ -88,28 +91,15 @@ class middleman_event_handler_impl : public middleman_event_handler {
} }
} }
} }
else { }
for (size_t i = 0; i < m_pollset.size(); ++i) { for (size_t i = 0; i < m_pollset.size(); ++i) {
event_bitmask eb = event::none; auto mask = static_cast<unsigned>(m_pollset[i].revents);
auto& revents = m_pollset[i].revents; auto eb = from_int_bitmask<input_event,
// read as long as possible, ignore POLLHUP as long as output_event,
// there is still data available error_event>(mask);
if (revents & (POLLIN | POLLPRI)) eb |= event::read; m_pollset[i].revents = 0;
else if (revents & (POLLRDHUP | POLLERR | POLLHUP | POLLNVAL)) { if (eb != event::none) m_events.emplace_back(eb, m_meta[i].ptr.get());
CPPA_LOG_DEBUG_IF(revents & POLLRDHUP, "POLLRDHUP"); }
CPPA_LOG_DEBUG_IF(revents & POLLERR, "POLLERR");
CPPA_LOG_DEBUG_IF(revents & POLLHUP, "POLLHUP");
CPPA_LOG_DEBUG_IF(revents & POLLNVAL, "POLLNVAL");
eb = event::error;
}
// POLLOUT and POLLHUP are mutually exclusive:
// no need to check wheter event::error has been set
if (revents & POLLOUT) eb |= event::write;
revents = 0;
m_events.emplace_back(eb, m_meta[i].ptr.get());
}
}
} while (presult < 0);
} }
void handle_event(fd_meta_event me, void handle_event(fd_meta_event me,
......
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