Merge pull request #9670 from mckellyln/hpcc-17038c

HPCC-17038 Report and return POLLERR errors

Reviewed-By: Jake Smith <jake.smith@lexisnexis.com>
Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

system/jlib/jsocket.cpp

@@ -403,6 +403,7 @@ public:
 
     ISocket*    accept(bool allowcancel);
     int         wait_read(unsigned timeout);
+    void        logPollError(unsigned revents, const char *rwstr);
     int         wait_write(unsigned timeout);
     int         name(char *name,size32_t namemax);
     int         peer_name(char *name,size32_t namemax);
@@ -1508,20 +1509,53 @@ void CSocket::udpconnect()
 
 
 
+void CSocket::logPollError(unsigned revents, const char *rwstr)
+{
+    if (revents & POLLERR)
+    {
+        char lname[256];
+        int lport = name(lname, sizeof(lname));
+        char rname[256];
+        int rport = peer_name(rname, sizeof(rname));
+        StringBuffer errStr;
+        errStr.appendf("%s POLLERR %u l: %s:%d r: %s:%d", rwstr, sock, lname, lport, rname, rport);
+        int serror = 0;
+        socklen_t serrlen = sizeof(serror);
+        int srtn = getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&serror, &serrlen);
+        if (srtn != 0)
+            serror = ERRNO();
+        LOGERR2(serror,2,errStr.str());
+    }
+    else if (revents & POLLNVAL)
+    {
+        StringBuffer errStr;
+        errStr.appendf("%s POLLINVAL", rwstr);
+        LOGERR2(999,3,errStr.str());
+    }
+    else
+    {
+        StringBuffer errStr;
+        errStr.appendf("%s unknown poll() revents: 0x%x", rwstr, revents);
+        LOGERR2(999,4,errStr.str());
+    }
+}
 
 int CSocket::wait_read(unsigned timeout)
 {
     int ret = 0;
-    while (sock!=INVALID_SOCKET) {
+    while (sock!=INVALID_SOCKET)
+    {
 #ifdef _USE_SELECT
         T_FD_SET fds;
         CHECKSOCKRANGE(sock);
         XFD_ZERO(&fds);
         FD_SET((unsigned)sock, &fds);
-        if (timeout==WAIT_FOREVER) {
+        if (timeout==WAIT_FOREVER)
+        {
             ret = ::select( sock + 1, (fd_set *)&fds, NULL, NULL, NULL );
         }
-        else {
+        else
+        {
             struct timeval tv;
             tv.tv_sec = timeout / 1000;
             tv.tv_usec = (timeout % 1000)*1000;
@@ -1534,15 +1568,41 @@ int CSocket::wait_read(unsigned timeout)
         fds[0].revents = 0;
         ret = ::poll(fds, 1, timeout);
 #endif
-        if (ret==SOCKET_ERROR) {
+        if (ret == SOCKET_ERROR)
+        {   // error
             int err = ERRNO();
-            if (err!=JSE_INTR) {   // else retry (should adjust time but for our usage don't think it matters that much)
+            if (err!=JSE_INTR)
+            {   // else retry (should adjust time but for our usage don't think it matters that much)
                 LOGERR2(err,1,"wait_read");
                 break;
             }
         }
+        else if (ret == 0)
+        {   // timeout
+            break;
+        }
         else
+        {   // ret > 0 - ready or error
+#ifdef _USE_SELECT
+            if (!FD_ISSET(sock, &fds))
+            {
+                LOGERR2(998,7,"wait_read");
+                ret = -1;
+            }
+#else
+            if ( (fds[0].revents & (POLLERR | POLLNVAL)) || (!(fds[0].revents & (POLLIN | POLLHUP))) )
+            {
+                logPollError(fds[0].revents, "wait_read");
+                ret = -1;
+            }
+            else
+            {
+                // POLLIN | POLLHUP ok
+                break;
+            }
+#endif
             break;
+        }
     }
     return ret;
 }
@@ -1572,15 +1632,46 @@ int CSocket::wait_write(unsigned timeout)
         fds[0].revents = 0;
         ret = ::poll(fds, 1, timeout);
 #endif
-        if (ret==SOCKET_ERROR) {
+        if (ret==SOCKET_ERROR)
+        {
             int err = ERRNO();
-            if (err!=JSE_INTR) {   // else retry (should adjust time but for our usage don't think it matters that much)
+            if (err!=JSE_INTR)
+            {   // else retry (should adjust time but for our usage don't think it matters that much)
                 LOGERR2(err,1,"wait_write");
                 break;
             }
         }
+        else if (ret == 0)
+        {   // timeout
+            break;
+        }
         else
+        {   // ret > 0 - ready or error
+#ifdef _USE_SELECT
+            if (!FD_ISSET(sock, &fds))
+            {
+                LOGERR2(998,7,"wait_write");
+                ret = -1;
+            }
+#else
+            if ( (fds[0].revents & (POLLERR | POLLNVAL)) || (!(fds[0].revents & (POLLOUT | POLLHUP))) )
+            {
+                logPollError(fds[0].revents, "wait_write");
+                ret = -1;
+            }
+            else if (fds[0].revents & POLLHUP)
+            {
+                LOGERR2(998,5,"wait_write POLLHUP");
+                ret = -1;
+            }
+            else
+            {
+                // POLLOUT ok
+                break;
+            }
+#endif
             break;
+        }
     }
     return ret;
 }

system/jlib/jsocket.hpp

@@ -278,6 +278,10 @@ public:
     //
     virtual ISocket* accept(bool allowcancel=false) = 0; // not needed for UDP
 
+    //
+    // log poll() errors
+    //
+    virtual void logPollError(unsigned revents, const char *rwstr) = 0;
 
     //
     // This method is called to check whether a socket has data ready

system/security/securesocket/securesocket.cpp

@@ -148,6 +148,7 @@ public:
     virtual int secure_accept();
     virtual int secure_connect();
 
+    virtual void logPollError(unsigned revents, const char *rwstr);
     virtual int wait_read(unsigned timeoutms);
     virtual void read(void* buf, size32_t min_size, size32_t max_size, size32_t &size_read,unsigned timeoutsecs);
     virtual void readtms(void* buf, size32_t min_size, size32_t max_size, size32_t &size_read, unsigned timeoutms);
@@ -644,6 +645,14 @@ int CSecureSocket::secure_connect()
 }
 
 //
+// log poll() errors
+//
+void CSecureSocket::logPollError(unsigned revents, const char *rwstr)
+{
+    m_socket->logPollError(revents, rwstr);
+}
+
+//
 // This method is called to check whether a socket has data ready
 // 
 int CSecureSocket::wait_read(unsigned timeoutms)