-fix RegSyncClient so that it detects breaks in the socket connection…

… to the server - by using keepalives -if we try to send on the socket periodically then we will detect connection errors -made EGAIN and EWOULDBLOCK processing the same in a few additional spots -add WinLeakCheck header to some new Websocket class files to aid in leak detection on windows git-svn-id: https://svn.resiprocate.org/rep/resiprocate/main@10968 ddefafc4-47db-0310-ae44-fa13212b10f2
TafThorne · Feb 14, 2014 · d2dcdaf · d2dcdaf
1 parent d5cd4c7
commit d2dcdaf
Show file tree

Hide file tree

Showing 9 changed files with 93 additions and 24 deletions.
diff --git a/reflow/ErrorCode.hxx b/reflow/ErrorCode.hxx
@@ -1,8 +1,6 @@
 #ifndef FM_ERRORCODE_HXX
 #define FM_ERRORCODE_HXX
 
-#include <asio/error_code.hpp>
-
 namespace flowmanager {
 
 typedef int ErrorType;

diff --git a/repro/HttpBase.cxx b/repro/HttpBase.cxx
@@ -174,7 +174,10 @@ HttpBase::process(FdSet& fdset)
          int e = getErrno();
          switch (e)
          {
-            case EWOULDBLOCK:
+            case EAGAIN:
+#if EAGAIN != EWOULDBLOCK
+            case EWOULDBLOCK:  // Treat EGAIN and EWOULDBLOCK as the same: http://stackoverflow.com/questions/7003234/which-systems-define-eagain-and-ewouldblock-as-different-values
+#endif
                // !jf! this can not be ready in some cases 
                return;
             default:

diff --git a/repro/RegSyncClient.cxx b/repro/RegSyncClient.cxx
@@ -93,7 +93,8 @@ RegSyncClient::thread()
       mSocketDesc = (int)socket(servAddr.ipVersion() == V6 ? PF_INET6 : PF_INET , SOCK_STREAM, 0);
       if(mSocketDesc < 0) 
       {
-         ErrLog(<< "RegSyncClient: cannot open socket");
+         int e = getErrno();
+         ErrLog(<< "RegSyncClient: cannot open socket, err=" << e);
          mSocketDesc = 0;
          return;
       }
@@ -102,7 +103,8 @@ RegSyncClient::thread()
       rc = ::bind(mSocketDesc, &localAddr.getMutableSockaddr(), localAddr.length());
       if(rc < 0) 
       {
-         ErrLog(<<"RegSyncClient: error binding locally");
+         int e = getErrno();
+         ErrLog(<<"RegSyncClient: error binding locally, err=" << e);
          closeSocket(mSocketDesc);
          mSocketDesc = 0;
          return;
@@ -112,7 +114,8 @@ RegSyncClient::thread()
       rc = ::connect(mSocketDesc, &servAddr.getMutableSockaddr(), servAddr.length());
       if(rc < 0) 
       {
-         if(!mShutdown) ErrLog(<< "RegSyncClient: error connecting to " << mAddress << ":" << mPort);
+         int e = getErrno();
+         if(!mShutdown) ErrLog(<< "RegSyncClient: error connecting to " << mAddress << ":" << mPort << ", err=" << e);
          closeSocket(mSocketDesc);
          mSocketDesc = 0;
          delaySeconds(30);
@@ -128,27 +131,78 @@ RegSyncClient::thread()
       rc = ::send(mSocketDesc, request.c_str(), (int)request.size(), 0);
       if(rc < 0) 
       {
-         if(!mShutdown) ErrLog(<< "RegSyncClient: error sending");
+         int e = getErrno();
+         if(!mShutdown) ErrLog(<< "RegSyncClient: error sending, err=" << e);
+         closeSocket(mSocketDesc);
+         mSocketDesc = 0;
+         continue;
+      }
+
+      // Make socket non blocking
+      bool ok = makeSocketNonBlocking(mSocketDesc);
+      if (!ok)
+      {
+         int e = getErrno();
+         ErrLog(<< "RegSyncClient: Could not make HTTP socket non-blocking, err=" << e);
          closeSocket(mSocketDesc);
          mSocketDesc = 0;
          continue;
       }
 
       while(rc > 0)
       {
-         rc = ::recv(mSocketDesc, (char*)&mRxBuffer, sizeof(mRxBuffer), 0);
-         if(rc < 0) 
+         FdSet fdset;
+         fdset.setRead(mSocketDesc);
+         fdset.setExcept(mSocketDesc);
+         timeval tm;
+         tm.tv_sec = 30; // TODO - make a setting?
+         tm.tv_usec = 0;
+         rc = fdset.select(tm);
+         if(rc > 0)
+         {
+            rc = ::recv(mSocketDesc, (char*)&mRxBuffer, sizeof(mRxBuffer), 0);
+            if(rc < 0) 
+            {
+                int e = getErrno();
+                if(!mShutdown) ErrLog(<< "RegSyncClient: error receiving, err=" << e);
+                closeSocket(mSocketDesc);
+                mSocketDesc = 0;
+                break;
+            }
+
+            if(rc > 0)
+            {
+                mRxDataBuffer += Data(Data::Borrow, (const char*)&mRxBuffer, rc);   
+                while(tryParse());
+            }
+         }
+         else if(rc == 0) // timeout - send keepalive
          {
-            if(!mShutdown) ErrLog(<< "RegSyncClient: error receiving");
-            closeSocket(mSocketDesc);
-            mSocketDesc = 0;
-            break;
+            rc = ::send(mSocketDesc, Symbols::CRLFCRLF, (int)request.size(), 0);
+            if(rc < 0) 
+            {
+               int e = getErrno();
+               // If send is blocking then we must have pending send data - so just ignore error - no need to keepalive
+               if ( e != EAGAIN && e != EWOULDBLOCK ) // Treat EGAIN and EWOULDBLOCK as the same: http://stackoverflow.com/questions/7003234/which-systems-define-eagain-and-ewouldblock-as-different-values
+               {
+                  if(!mShutdown) ErrLog(<< "RegSyncClient: error sending keepalive, err=" << e);
+                  closeSocket(mSocketDesc);
+                  mSocketDesc = 0;
+                  continue;
+               }
+            }
+            //else
+            //{
+            //    InfoLog( << "RegSyncClient: keepalive sent!");
+            //}
          }
-
-         if(rc > 0)
+         else
          {
-            mRxDataBuffer += Data(Data::Borrow, (const char*)&mRxBuffer, rc);   
-            while(tryParse());
+             int e = getErrno();
+             if(!mShutdown) ErrLog(<< "RegSyncClient: error calling select, err=" << e);
+             closeSocket(mSocketDesc);
+             mSocketDesc = 0;
+             break;
          }
       }
    } // end while
@@ -369,6 +423,8 @@ RegSyncClient::processModify(const resip::Uri& aor, ContactList& syncContacts)
    bool found;
    for(; itSync != syncContacts.end(); itSync++)
    {
+       InfoLog(<< "  RegSyncClient::processModify: contact=" << itSync->mContact << ", instance=" << itSync->mInstance << ", regid=" << itSync->mRegId);
+
        // See if contact already exists in currentContacts       
        found = false;
        for(itCurrent = currentContacts.begin(); itCurrent != currentContacts.end(); itCurrent++)

diff --git a/repro/RegSyncServerThread.cxx b/repro/RegSyncServerThread.cxx
@@ -2,7 +2,6 @@
 #include <rutil/Logger.hxx>
 
 #include "repro/RegSyncServer.hxx"
-#include "repro/RegSyncClient.hxx"
 #include "repro/RegSyncServerThread.hxx"
 
 #define RESIPROCATE_SUBSYSTEM Subsystem::REPRO

diff --git a/repro/XmlRpcServerBase.cxx b/repro/XmlRpcServerBase.cxx
@@ -203,7 +203,10 @@ XmlRpcServerBase::process(FdSet& fdset)
          int e = getErrno();
          switch (e)
          {
-            case EWOULDBLOCK:
+            case EAGAIN:
+#if EAGAIN != EWOULDBLOCK
+            case EWOULDBLOCK:  // Treat EGAIN and EWOULDBLOCK as the same: http://stackoverflow.com/questions/7003234/which-systems-define-eagain-and-ewouldblock-as-different-values
+#endif
                return;
             default:
                logSocketError(e);

diff --git a/resip/stack/TcpBaseTransport.cxx b/resip/stack/TcpBaseTransport.cxx
@@ -153,7 +153,10 @@ TcpBaseTransport::processListen()
          int e = getErrno();
          switch (e)
          {
-            case EWOULDBLOCK:
+            case EAGAIN:
+#if EAGAIN != EWOULDBLOCK
+            case EWOULDBLOCK:  // Treat EGAIN and EWOULDBLOCK as the same: http://stackoverflow.com/questions/7003234/which-systems-define-eagain-and-ewouldblock-as-different-values
+#endif
                // !jf! this can not be ready in some cases
                // !kw! this will happen every epoll cycle
                return 0;
@@ -245,7 +248,10 @@ TcpBaseTransport::makeOutgoingConnection(const Tuple &dest,
       switch (err)
       {
          case EINPROGRESS:
-         case EWOULDBLOCK:
+         case EAGAIN:
+#if EAGAIN != EWOULDBLOCK
+         case EWOULDBLOCK:  // Treat EGAIN and EWOULDBLOCK as the same: http://stackoverflow.com/questions/7003234/which-systems-define-eagain-and-ewouldblock-as-different-values
+#endif
             break;
          default:
          {

diff --git a/resip/stack/WsDecorator.cxx b/resip/stack/WsDecorator.cxx
@@ -3,7 +3,7 @@
 #include "SipMessage.hxx"
 #include "Tuple.hxx"
 #include "Transport.hxx"
-
+#include "rutil/WinLeakCheck.hxx"
 
 using namespace std;
 using namespace resip;

diff --git a/resip/stack/WsFrameExtractor.cxx b/resip/stack/WsFrameExtractor.cxx
@@ -1,6 +1,7 @@
 
 #include "rutil/Logger.hxx"
 #include "resip/stack/WsFrameExtractor.hxx"
+#include "rutil/WinLeakCheck.hxx"
 
 using namespace resip;
 

diff --git a/rutil/SelectInterruptor.cxx b/rutil/SelectInterruptor.cxx
@@ -118,13 +118,16 @@ SelectInterruptor::interrupt()
    assert(count == sizeof(wakeUp));
 #else
    ssize_t res = write(mPipe[1], wakeUp, sizeof(wakeUp));
-   if ( res == -1 && errno==EAGAIN )
+
+   if ( res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK) )   // Treat EGAIN and EWOULDBLOCK as the same: http://stackoverflow.com/questions/7003234/which-systems-define-eagain-and-ewouldblock-as-different-values
    {
       ; // this can happen when SipStack thread gets behind.
       // no need to block since our only purpose is to wake up the thread
       // also, this write can occur within the SipStack thread, in which
       // case we get dead-lock if this blocks
-   } else {
+   } 
+   else 
+   {
       assert(res == sizeof(wakeUp));
    }
 #endif