Commit 596d6e5b authored by KIU Shueng Chuan's avatar KIU Shueng Chuan

create signaler::recv_failable()

In real world usage, there have been reported signaler failures where the
eventfd read() or socket recv() system call in signaler::recv() fails,
despite having made a prior successful signaler::wait() call.

this patch creates a signaler::recv_failable() method that allows
unreadable eventfd / socket to return an error without asserting.
parent 52ee7241
...@@ -77,14 +77,18 @@ int zmq::mailbox_t::recv (command_t *cmd_, int timeout_) ...@@ -77,14 +77,18 @@ int zmq::mailbox_t::recv (command_t *cmd_, int timeout_)
} }
// Wait for signal from the command sender. // Wait for signal from the command sender.
const int rc = signaler.wait (timeout_); int rc = signaler.wait (timeout_);
if (rc == -1) { if (rc == -1) {
errno_assert (errno == EAGAIN || errno == EINTR); errno_assert (errno == EAGAIN || errno == EINTR);
return -1; return -1;
} }
// Receive the signal. // Receive the signal.
signaler.recv (); rc = signaler.recv_failable ();
if (rc == -1) {
errno_assert (errno == EAGAIN);
return -1;
}
// Switch into active state. // Switch into active state.
active = true; active = true;
......
...@@ -284,6 +284,38 @@ int zmq::signaler_t::wait (int timeout_) ...@@ -284,6 +284,38 @@ int zmq::signaler_t::wait (int timeout_)
} }
void zmq::signaler_t::recv () void zmq::signaler_t::recv ()
{
// Attempt to read a signal.
#if defined ZMQ_HAVE_EVENTFD
uint64_t dummy;
ssize_t sz = read (r, &dummy, sizeof (dummy));
errno_assert (sz == sizeof (dummy));
// If we accidentally grabbed the next signal(s) along with the current
// one, return it back to the eventfd object.
if (unlikely (dummy > 1)) {
const uint64_t inc = dummy - 1;
ssize_t sz2 = write (w, &inc, sizeof (inc));
errno_assert (sz2 == sizeof (inc));
return;
}
zmq_assert (dummy == 1);
#else
unsigned char dummy;
#if defined ZMQ_HAVE_WINDOWS
int nbytes = ::recv (r, (char*) &dummy, sizeof (dummy), 0);
wsa_assert (nbytes != SOCKET_ERROR);
#else
ssize_t nbytes = ::recv (r, &dummy, sizeof (dummy), 0);
errno_assert (nbytes >= 0);
#endif
zmq_assert (nbytes == sizeof (dummy));
zmq_assert (dummy == 0);
#endif
}
int zmq::signaler_t::recv_failable ()
{ {
// Attempt to read a signal. // Attempt to read a signal.
#if defined ZMQ_HAVE_EVENTFD #if defined ZMQ_HAVE_EVENTFD
...@@ -291,6 +323,7 @@ void zmq::signaler_t::recv () ...@@ -291,6 +323,7 @@ void zmq::signaler_t::recv ()
ssize_t sz = read (r, &dummy, sizeof (dummy)); ssize_t sz = read (r, &dummy, sizeof (dummy));
if (sz == -1) { if (sz == -1) {
errno_assert (errno == EAGAIN); errno_assert (errno == EAGAIN);
return -1;
} }
else { else {
errno_assert (sz == sizeof (dummy)); errno_assert (sz == sizeof (dummy));
...@@ -301,7 +334,7 @@ void zmq::signaler_t::recv () ...@@ -301,7 +334,7 @@ void zmq::signaler_t::recv ()
const uint64_t inc = dummy - 1; const uint64_t inc = dummy - 1;
ssize_t sz2 = write (w, &inc, sizeof (inc)); ssize_t sz2 = write (w, &inc, sizeof (inc));
errno_assert (sz2 == sizeof (inc)); errno_assert (sz2 == sizeof (inc));
return; return 0;
} }
zmq_assert (dummy == 1); zmq_assert (dummy == 1);
...@@ -310,14 +343,28 @@ void zmq::signaler_t::recv () ...@@ -310,14 +343,28 @@ void zmq::signaler_t::recv ()
unsigned char dummy; unsigned char dummy;
#if defined ZMQ_HAVE_WINDOWS #if defined ZMQ_HAVE_WINDOWS
int nbytes = ::recv (r, (char*) &dummy, sizeof (dummy), 0); int nbytes = ::recv (r, (char*) &dummy, sizeof (dummy), 0);
wsa_assert (nbytes != SOCKET_ERROR); if (nbytes == SOCKET_ERROR) {
const int last_error = WSAGetLastError();
if (last_error == WSAEWOULDBLOCK) {
errno = EAGAIN;
return -1;
}
wsa_assert (last_error == WSAEWOULDBLOCK);
}
#else #else
ssize_t nbytes = ::recv (r, &dummy, sizeof (dummy), 0); ssize_t nbytes = ::recv (r, &dummy, sizeof (dummy), 0);
errno_assert (nbytes >= 0); if (nbytes == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
errno = EAGAIN;
return -1;
}
errno_assert (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR);
}
#endif #endif
zmq_assert (nbytes == sizeof (dummy)); zmq_assert (nbytes == sizeof (dummy));
zmq_assert (dummy == 0); zmq_assert (dummy == 0);
#endif #endif
return 0;
} }
#ifdef HAVE_FORK #ifdef HAVE_FORK
......
...@@ -55,6 +55,7 @@ namespace zmq ...@@ -55,6 +55,7 @@ namespace zmq
void send (); void send ();
int wait (int timeout_); int wait (int timeout_);
void recv (); void recv ();
int recv_failable ();
#ifdef HAVE_FORK #ifdef HAVE_FORK
// close the file descriptors in a forked child process so that they // close the file descriptors in a forked child process so that they
......
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