Commit 9c4a0cb4 authored by Dominik Charousset's avatar Dominik Charousset

evaluate errno whenever select() fails, relates #64

parent fb5ba429
...@@ -689,7 +689,7 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) { ...@@ -689,7 +689,7 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) {
fd_set wrset; fd_set wrset;
fd_set* wrset_ptr = nullptr; fd_set* wrset_ptr = nullptr;
m_channels.emplace_back(new middleman_overseer(this, pipe_fd, queue)); m_channels.emplace_back(new middleman_overseer(this, pipe_fd, queue));
do { auto update_fd_sets = [&] {
FD_ZERO(&rdset); FD_ZERO(&rdset);
maxfd = 0; maxfd = 0;
CPPA_REQUIRE(m_channels.size() > 0); CPPA_REQUIRE(m_channels.size() > 0);
...@@ -710,36 +710,20 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) { ...@@ -710,36 +710,20 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) {
wrset_ptr = &wrset; wrset_ptr = &wrset;
} }
CPPA_REQUIRE(maxfd > 0); CPPA_REQUIRE(maxfd > 0);
//DEBUG("select()"); };
int sresult; auto continue_reading = [&](const network_channel_ptr& ch) {
do {
sresult = select(maxfd + 1, &rdset, wrset_ptr, nullptr, nullptr);
if (sresult < 0) {
CPPA_CRITICAL("select() failed");
}
}
while (sresult == 0);
//DEBUG("continue reading ...");
{ // iterate over all channels and remove channels as needed
for (auto& channel : m_channels) {
if (FD_ISSET(channel->read_handle(), &rdset)) {
bool erase_channel = false; bool erase_channel = false;
try { erase_channel = !channel->continue_reading(); } try { erase_channel = !ch->continue_reading(); }
catch (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;
} }
if (erase_channel) { if (erase_channel) {
DEBUG("erase worker"); DEBUG("erase worker");
m_erased_channels.insert(channel); m_erased_channels.insert(ch);
}
}
}
} }
if (wrset_ptr) { // iterate over peers with unwritten data };
DEBUG("continue writing ..."); auto continue_writing = [&](const peer_connection_ptr& peer) {
for (auto& peer : m_peers_with_unwritten_data) {
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 (exception& e) { catch (exception& e) {
...@@ -750,16 +734,16 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) { ...@@ -750,16 +734,16 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) {
DEBUG("erase worker"); DEBUG("erase worker");
m_erased_channels.insert(peer); m_erased_channels.insert(peer);
} }
} };
} auto insert_new_handlers = [&] {
}
// insert new handlers
if (m_new_channels.empty() == false) { if (m_new_channels.empty() == false) {
DEBUG("insert new channel(s)"); DEBUG("insert new channel(s)");
move(m_new_channels.begin(), m_new_channels.end(), move(m_new_channels.begin(), m_new_channels.end(),
back_inserter(m_channels)); back_inserter(m_channels));
m_new_channels.clear(); m_new_channels.clear();
} }
};
auto erase_erroneous_channels = [&] {
if (!m_erased_channels.empty()) { if (!m_erased_channels.empty()) {
DEBUG("erase channel(s)"); DEBUG("erase channel(s)");
// erase all marked channels // erase all marked channels
...@@ -772,6 +756,76 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) { ...@@ -772,6 +756,76 @@ void middleman::operator()(int pipe_fd, middleman_queue& queue) {
} }
m_erased_channels.clear(); m_erased_channels.clear();
} }
};
do {
update_fd_sets();
//DEBUG("select()");
int sresult;
do {
sresult = select(maxfd + 1, &rdset, wrset_ptr, nullptr, nullptr);
if (sresult < 0) {
// try again or die hard
sresult = 0;
switch (errno) {
// a signal was caught
case EINTR: {
// just try again
break;
}
// nfds is negative or the value
// contained within timeout is invalid
case EINVAL: {
if ((maxfd + 1) < 0) {
CPPA_CRITICAL("overflow: maxfd + 1 > 0");
}
break;
}
case ENOMEM: {
// there's not much we can do other than try again
// sleep some time in hope someone releases memory
// while we are sleeping
this_thread::sleep_for(chrono::milliseconds(10));
break;
}
case EBADF: {
// this really shouldn't happen
// try IO on each single socket and rebuild rd_set
for (auto& ch: m_channels) {
continue_reading(ch);
}
for (auto& peer : m_peers_with_unwritten_data) {
continue_writing(peer);
}
insert_new_handlers();
erase_erroneous_channels();
update_fd_sets();
break;
}
default: {
CPPA_CRITICAL("select() failed for an unknown reason");
}
}
}
}
while (sresult == 0);
//DEBUG("continue reading ...");
{ // iterate over all channels and remove channels as needed
for (auto& ch : m_channels) {
if (FD_ISSET(ch->read_handle(), &rdset)) {
continue_reading(ch);
}
}
}
if (wrset_ptr) { // iterate over peers with unwritten data
DEBUG("continue writing ...");
for (auto& peer : m_peers_with_unwritten_data) {
if (FD_ISSET(peer->write_handle(), &wrset)) {
continue_writing(peer);
}
}
}
insert_new_handlers();
erase_erroneous_channels();
} }
while (m_done == false); while (m_done == false);
DEBUG("middleman done"); DEBUG("middleman 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