Commit 172a53f4 authored by Dominik Charousset's avatar Dominik Charousset

fixed type bug in epoll implementation and flush outgoing messages before shutting down

parent 880732db
...@@ -512,6 +512,8 @@ class middleman_event_handler : public middleman_event_handler_base { ...@@ -512,6 +512,8 @@ class middleman_event_handler : public middleman_event_handler_base {
inline void init() { } inline void init() { }
size_t num_sockets() const { return m_pollset.size(); }
pair<event_iterator,event_iterator> poll() { pair<event_iterator,event_iterator> poll() {
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());
...@@ -622,6 +624,7 @@ struct epoll_iterator_access { ...@@ -622,6 +624,7 @@ struct epoll_iterator_access {
auto result = event::none; auto result = event::none;
if (events & EPOLLIN) result |= event::read; if (events & EPOLLIN) result |= event::read;
if (events & EPOLLOUT) result |= event::write; if (events & EPOLLOUT) result |= event::write;
CPPA_REQUIRE(result != 0);
return result; return result;
} }
...@@ -656,10 +659,14 @@ class middleman_event_handler : public middleman_event_handler_base { ...@@ -656,10 +659,14 @@ class middleman_event_handler : public middleman_event_handler_base {
m_events.resize(64); m_events.resize(64);
} }
size_t num_sockets() const { return m_epoll_data.size(); }
pair<event_iterator,event_iterator> poll() { pair<event_iterator,event_iterator> poll() {
CPPA_REQUIRE(m_epoll_data.empty() == false);
for (;;) { for (;;) {
DEBUG("epoll_wait on " << m_epoll_data.size() << " sockets"); DEBUG("epoll_wait on " << m_epoll_data.size() << " sockets");
auto presult = epoll_wait(m_epollfd, m_events.data(), (int) m_events.size(), -1); auto presult = epoll_wait(m_epollfd, m_events.data(),
(int) m_events.size(), -1);
DEBUG("epoll_wait returned " << presult); DEBUG("epoll_wait returned " << presult);
if (presult < 0) { if (presult < 0) {
// try again unless critical error occured // try again unless critical error occured
...@@ -694,9 +701,10 @@ class middleman_event_handler : public middleman_event_handler_base { ...@@ -694,9 +701,10 @@ class middleman_event_handler : public middleman_event_handler_base {
void handle_vec(vector<fd_meta_info>& vec, int eop) { void handle_vec(vector<fd_meta_info>& vec, int eop) {
for (auto& element : vec) { for (auto& element : vec) {
CPPA_REQUIRE((std::get<2>(element) & event::both) != event::none); auto mask = std::get<2>(element);
CPPA_REQUIRE(mask == event::write || mask == event::read || mask == event::both);
auto ptr = std::get<1>(element).get(); auto ptr = std::get<1>(element).get();
switch (std::get<2>(element)) { switch (mask) {
case event::read: case event::read:
epoll_op(eop, ptr->read_handle(), EPOLLIN, ptr); epoll_op(eop, ptr->read_handle(), EPOLLIN, ptr);
break; break;
...@@ -707,15 +715,17 @@ class middleman_event_handler : public middleman_event_handler_base { ...@@ -707,15 +715,17 @@ class middleman_event_handler : public middleman_event_handler_base {
break; break;
} }
case event::both: { case event::both: {
CPPA_REQUIRE(ptr->is_peer()); if (ptr->is_peer()) {
auto dptr = static_cast<peer*>(ptr); auto dptr = static_cast<peer*>(ptr);
auto rd = dptr->read_handle(); auto rd = dptr->read_handle();
auto wr = dptr->write_handle(); auto wr = dptr->write_handle();
if (rd == wr) epoll_op(eop, wr, EPOLLIN | EPOLLOUT, dptr); if (rd == wr) epoll_op(eop, wr, EPOLLIN|EPOLLOUT, dptr);
else { else {
epoll_op(eop, rd, EPOLLIN, ptr); epoll_op(eop, rd, EPOLLIN, ptr);
epoll_op(eop, wr, EPOLLOUT, dptr); epoll_op(eop, wr, EPOLLOUT, dptr);
}
} }
else epoll_op(eop, ptr->read_handle(), EPOLLIN, ptr);
break; break;
} }
default: CPPA_CRITICAL("invalid event mask found in handle_vec"); default: CPPA_CRITICAL("invalid event mask found in handle_vec");
...@@ -725,45 +735,38 @@ class middleman_event_handler : public middleman_event_handler_base { ...@@ -725,45 +735,38 @@ class middleman_event_handler : public middleman_event_handler_base {
} }
// operation: EPOLL_CTL_ADD or EPOLL_CTL_DEL // operation: EPOLL_CTL_ADD or EPOLL_CTL_DEL
// fd_op: EPOLLIN, EPOLLOUT or (EPOLLIN | EPOLLOUT) // mask: EPOLLIN, EPOLLOUT or (EPOLLIN | EPOLLOUT)
template<typename T = void> void epoll_op(int operation, int fd, uint32_t mask, continuable_reader* ptr) {
void epoll_op(int operation, int fd, int fd_op, T* ptr = nullptr) { CPPA_REQUIRE(ptr != nullptr);
CPPA_REQUIRE(operation == EPOLL_CTL_ADD || operation == EPOLL_CTL_DEL); CPPA_REQUIRE(operation == EPOLL_CTL_ADD || operation == EPOLL_CTL_DEL);
CPPA_REQUIRE(operation == EPOLL_CTL_DEL || ptr != nullptr); CPPA_REQUIRE(mask == EPOLLIN || mask == EPOLLOUT || mask == (EPOLLIN|EPOLLOUT));
CPPA_REQUIRE(fd_op == EPOLLIN || fd_op == EPOLLOUT || fd_op == (EPOLLIN | EPOLLOUT));
// make sure T has correct type
CPPA_REQUIRE( (operation == EPOLL_CTL_DEL && is_same<T,void>::value)
|| ((fd_op & EPOLLIN) && is_same<T,continuable_reader>::value)
|| (fd_op == EPOLLOUT && is_same<T,peer>::value));
epoll_event ee; epoll_event ee;
// also fire event on peer shutdown on input operations // also fire event on peer shutdown for input operations
ee.events = (fd_op & EPOLLIN) ? (fd_op | EPOLLRDHUP) : fd_op; ee.events = (mask & EPOLLIN) ? (mask | EPOLLRDHUP) : mask;
// always store peer_ptr, because we don't have full type information ee.data.ptr = ptr;
// in case of epoll_wait error otherwise
ee.data.ptr = static_cast<peer*>(ptr);
// check wheter fd is already registered to epoll // check wheter fd is already registered to epoll
auto iter = m_epoll_data.find(fd); auto iter = m_epoll_data.find(fd);
if (iter != end(m_epoll_data)) { if (iter != end(m_epoll_data)) {
CPPA_REQUIRE(ee.data.ptr == iter->second.data.ptr);
if (operation == EPOLL_CTL_ADD) { if (operation == EPOLL_CTL_ADD) {
if (mask == iter->second.events) {
// nothing to do here, fd is already registered with
// correct bitmask
return;
}
// modify instead // modify instead
operation = EPOLL_CTL_MOD; operation = EPOLL_CTL_MOD;
ee.events |= iter->second.events; ee.events |= iter->second.events; // new bitmask
iter->second.events = ee.events; iter->second.events = ee.events; // update bitmask in map
CPPA_REQUIRE(ee.data.ptr == iter->second.data.ptr);
} }
else if (operation == EPOLL_CTL_DEL) { else if (operation == EPOLL_CTL_DEL) {
// check wheter we have this fd registered for other fd_ops // check wheter we have fd registered for other events
ee.events = iter->second.events & ~(ee.events); ee.events = iter->second.events & ~(ee.events);
if (ee.events != 0) { if (ee.events != 0) {
// modify instead iter->second.events = ee.events; // update bitmask in map
ee.data.ptr = iter->second.data.ptr; operation = EPOLL_CTL_MOD; // modify instead
iter->second.events = ee.events;
operation = EPOLL_CTL_MOD;
}
else {
// erase from map as well
m_epoll_data.erase(iter);
} }
else m_epoll_data.erase(iter); // erase from map as well
} }
} }
else if (operation == EPOLL_CTL_DEL) { else if (operation == EPOLL_CTL_DEL) {
...@@ -776,12 +779,6 @@ class middleman_event_handler : public middleman_event_handler_base { ...@@ -776,12 +779,6 @@ class middleman_event_handler : public middleman_event_handler_base {
} }
if (epoll_ctl(m_epollfd, operation, fd, &ee) < 0) { if (epoll_ctl(m_epollfd, operation, fd, &ee) < 0) {
switch (errno) { switch (errno) {
// m_epollfd or read_handle() is not a valid file descriptor
case EBADF: {
// this is a critical bug, there's no plan B here
CPPA_CRITICAL("epoll_ctl returned EBADF");
break;
}
// supplied file descriptor is already registered // supplied file descriptor is already registered
case EEXIST: { case EEXIST: {
// shouldn't happen, but no big deal // shouldn't happen, but no big deal
...@@ -789,14 +786,6 @@ class middleman_event_handler : public middleman_event_handler_base { ...@@ -789,14 +786,6 @@ class middleman_event_handler : public middleman_event_handler_base {
<< flush; << flush;
break; break;
} }
// m_pollfd not an epoll file descriptor, or read_handle()
// is the same as m_pollfd, or read_handle() isn't supported
// by epoll
case EINVAL: {
// point of no return
CPPA_CRITICAL("epoll_ctl returned EINVAL");
break;
}
// op was EPOLL_CTL_MOD or EPOLL_CTL_DEL, // op was EPOLL_CTL_MOD or EPOLL_CTL_DEL,
// and fd is not registered with this epoll instance. // and fd is not registered with this epoll instance.
case ENOENT: { case ENOENT: {
...@@ -805,22 +794,9 @@ class middleman_event_handler : public middleman_event_handler_base { ...@@ -805,22 +794,9 @@ class middleman_event_handler : public middleman_event_handler_base {
<< flush; << flush;
break; break;
} }
// insufficient memory to handle requested default: {
case ENOMEM: { perror("epoll_ctl() failed");
// what the ... ? CPPA_CRITICAL("epoll_ctl() failed");
CPPA_CRITICAL("not enough memory for epoll operation");
break;
}
// The limit imposed by /proc/sys/fs/epoll/max_user_watches
// was encountered while trying to register a new file descriptor
case ENOSPC: {
CPPA_CRITICAL("reached max_user_watches limit");
break;
}
// The target file fd does not support epoll.
case EPERM: {
CPPA_CRITICAL("tried to add illegal file descriptor");
break;
} }
} }
} }
...@@ -899,6 +875,7 @@ void middleman_loop(middleman_impl* impl) { ...@@ -899,6 +875,7 @@ void middleman_loop(middleman_impl* impl) {
break; break;
case write_done: case write_done:
handler->erase_later(i->ptr(), event::write); handler->erase_later(i->ptr(), event::write);
break;
default: break; default: break;
} }
if (mask == event::write) break; if (mask == event::write) break;
...@@ -925,6 +902,30 @@ void middleman_loop(middleman_impl* impl) { ...@@ -925,6 +902,30 @@ void middleman_loop(middleman_impl* impl) {
} }
handler->update(); handler->update();
} }
DEBUG("flush outgoing messages");
// make sure to write everything before shutting down
for (auto ptr : impl->m_readers) { handler->erase_later(ptr, event::read); }
handler->update();
while (handler->num_sockets() > 0) {
auto iters = handler->poll();
for (auto i = iters.first; i != iters.second; ++i) {
switch (i->type()) {
case event::write:
switch (i->continue_writing()) {
case write_closed:
case write_failure:
case write_done:
handler->erase_later(i->ptr(), event::write);
break;
default: break;
}
break;
default: CPPA_CRITICAL("expected event::write only "
"during shutdown phase");
}
}
handler->update();
}
DEBUG("middleman loop done"); DEBUG("middleman loop done");
} }
......
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