Merge pull request #1595 from pijyoi/signaler_failable

create signaler::recv_failable()

src/mailbox.cpp

@@ -77,14 +77,18 @@ int zmq::mailbox_t::recv (command_t *cmd_, int timeout_)
     }
 
     //  Wait for signal from the command sender.
-    const int rc = signaler.wait (timeout_);
+    int rc = signaler.wait (timeout_);
     if (rc == -1) {
         errno_assert (errno == EAGAIN || errno == EINTR);
         return -1;
     }
 
     //  Receive the signal.
-    signaler.recv ();
+    rc = signaler.recv_failable ();
+    if (rc == -1) {
+        errno_assert (errno == EAGAIN);
+        return -1;
+    }
 
     //  Switch into active state.
     active = true;

src/signaler.cpp

@@ -284,6 +284,38 @@ int zmq::signaler_t::wait (int timeout_)
 }
 
 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.
 #if defined ZMQ_HAVE_EVENTFD
@@ -291,6 +323,7 @@ void zmq::signaler_t::recv ()
     ssize_t sz = read (r, &dummy, sizeof (dummy));
     if (sz == -1) {
         errno_assert (errno == EAGAIN);
+        return -1;
     }
     else {
         errno_assert (sz == sizeof (dummy));
@@ -301,7 +334,7 @@ void zmq::signaler_t::recv ()
             const uint64_t inc = dummy - 1;
             ssize_t sz2 = write (w, &inc, sizeof (inc));
             errno_assert (sz2 == sizeof (inc));
-            return;
+            return 0;
         }
 
         zmq_assert (dummy == 1);
@@ -310,14 +343,28 @@ void zmq::signaler_t::recv ()
     unsigned char dummy;
 #if defined ZMQ_HAVE_WINDOWS
     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
     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
     zmq_assert (nbytes == sizeof (dummy));
     zmq_assert (dummy == 0);
 #endif
+    return 0;
 }
 
 #ifdef HAVE_FORK

src/signaler.hpp

@@ -55,6 +55,7 @@ namespace zmq
         void send ();
         int wait (int timeout_);
         void recv ();
+        int recv_failable ();
 
 #ifdef HAVE_FORK
         // close the file descriptors in a forked child process so that they