Added Ice.TCP.SndSize/Ice.TCP.RcvSize properties.

6 files changed, 164 insertions, 121 deletions

diff --git a/cppe/src/IceE/Network.cpp b/cppe/src/IceE/Network.cpp
index 949f8b8b297..5338648cf1a 100644
--- a/cppe/src/IceE/Network.cpp
+++ b/cppe/src/IceE/Network.cpp
@@ -10,6 +10,8 @@
 #include <IceE/StaticMutex.h>
 #include <IceE/Network.h>
 #include <IceE/LocalException.h>
+#include <IceE/Properties.h> // For setTcpBufSize
+#include <IceE/LoggerUtil.h> // For setTcpBufSize
 #include <IceE/SafeStdio.h>
 
 #if defined(_WIN32)
@@ -35,7 +37,7 @@ using namespace IceInternal;
 static IceUtil::StaticMutex inetMutex = ICE_STATIC_MUTEX_INITIALIZER;
 
 static string
-inetAddrToString(struct in_addr in)
+inetAddrToString(const struct in_addr& in)
 {
     //
     // inet_ntoa uses static memory on some platforms so we protect
@@ -398,11 +400,53 @@ IceInternal::setSendBufferSize(SOCKET fd, int sz)
 {
     if(setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char*)&sz, int(sizeof(int))) == SOCKET_ERROR)
     {
-	closeSocketNoThrow(fd);
-	SocketException ex(__FILE__, __LINE__);
-	ex.error = getSocketErrno();
-	throw ex;
+        closeSocketNoThrow(fd);
+        SocketException ex(__FILE__, __LINE__);
+        ex.error = getSocketErrno();
+        throw ex;
+    }
+}
+
+int
+IceInternal::getSendBufferSize(SOCKET fd)
+{
+    int sz;
+    socklen_t len = sizeof(sz);
+    if(getsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char*)&sz, &len) == SOCKET_ERROR || len != sizeof(sz))
+    {
+        closeSocketNoThrow(fd);
+        SocketException ex(__FILE__, __LINE__);
+        ex.error = getSocketErrno();
+        throw ex;
     }
+    return sz;
+}
+
+void
+IceInternal::setRecvBufferSize(SOCKET fd, int sz)
+{
+    if(setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char*)&sz, int(sizeof(int))) == SOCKET_ERROR)
+    {
+        closeSocketNoThrow(fd);
+        SocketException ex(__FILE__, __LINE__);
+        ex.error = getSocketErrno();
+        throw ex;
+    }
+}
+
+int
+IceInternal::getRecvBufferSize(SOCKET fd)
+{
+    int sz;
+    socklen_t len = sizeof(sz);
+    if(getsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char*)&sz, &len) == SOCKET_ERROR || len != sizeof(sz))
+    {
+        closeSocketNoThrow(fd);
+        SocketException ex(__FILE__, __LINE__);
+        ex.error = getSocketErrno();
+        throw ex;
+    }
+    return sz;
 }
 
 void
@@ -442,12 +486,6 @@ IceInternal::doConnect(SOCKET fd, struct sockaddr_in& addr, int timeout)
 {
 #ifdef _WIN32
     //
-    // Set larger send buffer size to avoid performance problems on
-    // WIN32.
-    //
-    setSendBufferSize(fd, 64 * 1024);
-
-    //
     // Under WinCE its not possible to find out the connection failure
     // reason with SO_ERROR, so its necessary to use the WSAEVENT
     // mechanism. We use the same mechanism for any Winsock platform.
@@ -884,15 +922,6 @@ repeatAccept:
 
     setTcpNoDelay(ret);
     setKeepAlive(ret);
-
-#ifdef _WIN32
-    //
-    // Set larger send buffer size to avoid performance problems on
-    // WIN32.
-    //
-    setSendBufferSize(ret, 64 * 1024);
-#endif
-    
     return ret;
 }
 
@@ -922,10 +951,62 @@ IceInternal::getLocalHosts()
     vector<string> result;
 
 #if defined(_WIN32)
-    vector<struct sockaddr_in> addrs = getLocalAddresses();
-    for(unsigned int i = 0; i < addrs.size(); ++i)
+    try
     {
-        result.push_back(inetAddrToString(addrs[i].sin_addr));
+        SOCKET fd = createSocket();
+
+        vector<unsigned char> buffer;
+        buffer.resize(1024);
+        unsigned long len = 0;
+        DWORD rs = WSAIoctl(fd, SIO_ADDRESS_LIST_QUERY, 0, 0, &buffer[0], buffer.size(), &len, 0, 0);
+        if(rs == SOCKET_ERROR)
+        {
+            //
+            // If the buffer wasn't big enough, resize it to the
+            // required length and try again.
+            //
+            if(getSocketErrno() == WSAEFAULT)
+            {
+                buffer.resize(len);
+                rs = WSAIoctl(fd, SIO_ADDRESS_LIST_QUERY, 0, 0, &buffer[0], buffer.size(), &len, 0, 0);
+            }
+
+            if(rs == SOCKET_ERROR)
+            {
+                closeSocketNoThrow(fd);
+                SocketException ex(__FILE__, __LINE__);
+                ex.error = getSocketErrno();
+                throw ex;
+            }
+        }
+
+        //
+        // Add the local interface addresses.
+        //
+        SOCKET_ADDRESS_LIST* addrs = reinterpret_cast<SOCKET_ADDRESS_LIST*>(&buffer[0]);
+        for (int i = 0; i < addrs->iAddressCount; ++i)
+        {
+            result.push_back(
+		inetAddrToString(reinterpret_cast<struct sockaddr_in*>(addrs->Address[i].lpSockaddr)->sin_addr));
+        }
+
+        //
+        // Add the loopback interface address.
+        //
+        struct sockaddr_in addr;
+        memset(&addr, 0, sizeof(addr));
+        addr.sin_family = AF_INET;
+        addr.sin_port = htons(0);
+        addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+        result.push_back(inetAddrToString(addr.sin_addr));
+
+        closeSocket(fd);
+    }
+    catch(const Ice::LocalException&)
+    {
+        //
+        // TODO: Warning?
+        //
     }
 #elif defined(__APPLE__) || defined(__FreeBSD__)
     struct ifaddrs* ifap;
@@ -1021,106 +1102,54 @@ IceInternal::getLocalHosts()
     return result;
 }
 
-#ifdef _WIN32
 
-vector<struct sockaddr_in>
-IceInternal::getLocalAddresses()
+void
+IceInternal::setTcpBufSize(SOCKET fd, const Ice::PropertiesPtr& properties, const Ice::LoggerPtr& logger)
 {
-    vector<struct sockaddr_in> result;
-    try
-    {
-        SOCKET fd = createSocket();
+    assert(fd != INVALID_SOCKET);
 
-        vector<unsigned char> buffer;
-        buffer.resize(1024);
-        unsigned long len = 0;
-        DWORD rs = WSAIoctl(fd, SIO_ADDRESS_LIST_QUERY, 0, 0, &buffer[0], buffer.size(), &len, 0, 0);
-        if(rs == SOCKET_ERROR)
-        {
-            //
-            // If the buffer wasn't big enough, resize it to the
-            // required length and try again.
-            //
-            if(getSocketErrno() == WSAEFAULT)
-            {
-                buffer.resize(len);
-                rs = WSAIoctl(fd, SIO_ADDRESS_LIST_QUERY, 0, 0, &buffer[0], buffer.size(), &len, 0, 0);
-            }
-
-            if(rs == SOCKET_ERROR)
-            {
-                closeSocketNoThrow(fd);
-                SocketException ex(__FILE__, __LINE__);
-                ex.error = getSocketErrno();
-                throw ex;
-            }
-        }
-
-        //        // Add the local interface addresses.
-        //
-        SOCKET_ADDRESS_LIST* addrs = reinterpret_cast<SOCKET_ADDRESS_LIST*>(&buffer[0]);
-        for (int i = 0; i < addrs->iAddressCount; ++i)
-        {
-            result.push_back(*reinterpret_cast<struct sockaddr_in*>(addrs->Address[i].lpSockaddr));
-        }
-
-        //
-        // Add the loopback interface address.
-        //
-        struct sockaddr_in addr;
-        memset(&addr, 0, sizeof(addr));
-        addr.sin_family = AF_INET;
-        addr.sin_port = htons(0);
-        addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
-        result.push_back(addr);
+    //
+    // By default, on Windows we use a 128KB buffer size. On Unix
+    // platforms, we use the system defaults.
+    //
+#ifdef _WIN32
+    const int dfltBufSize = 128 * 1024;
+#else
+    const int dfltBufSize = 0;
+#endif
+    Int sizeRequested;
 
-        closeSocket(fd);
-    }
-    catch(const Ice::LocalException&)
+    sizeRequested = properties->getPropertyAsIntWithDefault("Ice.TCP.RcvSize", dfltBufSize);
+    if(sizeRequested > 0)
     {
         //
-        // TODO: Warning?
+        // Try to set the buffer size. The kernel will silently adjust
+        // the size to an acceptable value. Then read the size back to
+        // get the size that was actually set.
         //
-    }
-    return result;
-}
-
-bool
-IceInternal::isLocalAddress(const struct sockaddr_in& addr)
-{
-    struct sockaddr_in addr0 = addr;
-    addr0.sin_port = htons(0); // Local interface addresses have the port set to 0.
-    vector<struct sockaddr_in> localAddrs = getLocalAddresses();
-    for(vector<struct sockaddr_in>::const_iterator p = localAddrs.begin(); p != localAddrs.end(); ++p)
-    {
-        if(compareAddress(addr0, *p))
+        setRecvBufferSize(fd, sizeRequested);
+        int size = getRecvBufferSize(fd);
+        if(size < sizeRequested) // Warn if the size that was set is less than the requested size.
         {
-            return true;
+            Warning out(logger);
+            out << printfToString("TCP receive buffer size: requested size of %d adjusted to %d", sizeRequested, size);
         }
     }
-    return false;
-}
 
-bool
-IceInternal::isPeerLocal(SOCKET fd)
-{
-    socklen_t remoteLen = static_cast<socklen_t>(sizeof(struct sockaddr_in));
-    struct sockaddr_in remoteAddr;
-    if(getpeername(fd, reinterpret_cast<struct sockaddr*>(&remoteAddr), &remoteLen) == SOCKET_ERROR)
+    sizeRequested = properties->getPropertyAsIntWithDefault("Ice.TCP.SndSize", dfltBufSize);
+    if(sizeRequested > 0)
     {
-        if(notConnected())
-        {
-            return false;
-        }
-        else
+        //
+        // Try to set the buffer size. The kernel will silently adjust
+        // the size to an acceptable value. Then read the size back to
+        // get the size that was actually set.
+        //
+        setSendBufferSize(fd, sizeRequested);
+        int size = getSendBufferSize(fd);
+        if(size < sizeRequested) // Warn if the size that was set is less than the requested size.
         {
-            closeSocketNoThrow(fd);
-            SocketException ex(__FILE__, __LINE__);
-            ex.error = getSocketErrno();
-            throw ex;
+            Warning out(logger);
+            out << printfToString("TCP send buffer size: requested size of %d adjusted to %d", sizeRequested, size);
         }
     }
-    return isLocalAddress(remoteAddr);
 }
-
-#endif
diff --git a/cppe/src/IceE/Network.h b/cppe/src/IceE/Network.h
index d8c1233648c..d7f95cb634d 100644
--- a/cppe/src/IceE/Network.h
+++ b/cppe/src/IceE/Network.h
@@ -15,6 +15,8 @@
 #endif
 
 #include <IceE/Config.h>
+#include <IceE/PropertiesF.h> // For setTcpBufSize
+#include <IceE/LoggerF.h> // For setTcpBufSize
 
 #ifdef _WIN32
 #   include <winsock2.h>
@@ -92,6 +94,9 @@ void setTimeout(SOCKET, bool, int);
 void setTcpNoDelay(SOCKET);
 void setKeepAlive(SOCKET);
 void setSendBufferSize(SOCKET, int);
+int getSendBufferSize(SOCKET);
+void setRecvBufferSize(SOCKET, int);
+int getRecvBufferSize(SOCKET);
 
 void doBind(SOCKET, struct sockaddr_in&);
 void doListen(SOCKET, int);
@@ -109,11 +114,7 @@ std::string fdToString(SOCKET);
 std::string addrToString(const struct sockaddr_in&);
 
 std::vector<std::string> getLocalHosts();
-#ifdef _WIN32
-std::vector<struct sockaddr_in> getLocalAddresses();
-bool isLocalAddress(const struct sockaddr_in&);
-bool isPeerLocal(SOCKET);
-#endif
+ void setTcpBufSize(SOCKET, const Ice::PropertiesPtr&, const Ice::LoggerPtr&);
 
 int getSocketErrno();
 
diff --git a/cppe/src/IceE/Transceiver.h b/cppe/src/IceE/Transceiver.h
index b7e6f516058..404e125ec17 100644
--- a/cppe/src/IceE/Transceiver.h
+++ b/cppe/src/IceE/Transceiver.h
@@ -87,7 +87,7 @@ private:
 
     const std::string _desc;
 #ifdef _WIN32
-    const bool _isPeerLocal;
+    int _maxPacketSize;
 #endif
 };
 
diff --git a/cppe/src/TcpTransport/Acceptor.cpp b/cppe/src/TcpTransport/Acceptor.cpp
index 1631789effb..72e53c7ace9 100644
--- a/cppe/src/TcpTransport/Acceptor.cpp
+++ b/cppe/src/TcpTransport/Acceptor.cpp
@@ -68,6 +68,7 @@ IceInternal::Acceptor::accept()
 #ifndef ICEE_USE_SELECT_OR_POLL_FOR_TIMEOUTS
     setBlock(fd, true);
 #endif
+    setTcpBufSize(fd, _instance->initializationData().properties, _logger);
 
     if(_traceLevels->network >= 1)
     {
@@ -114,6 +115,7 @@ IceInternal::Acceptor::Acceptor(const InstancePtr& instance, const string& host,
     {
 	_fd = createSocket();
 	getAddress(host, port, _addr);
+        setTcpBufSize(_fd, _instance->initializationData().properties, _logger);
 	if(_traceLevels->network >= 2)
 	{
 	    Trace out(_logger, _traceLevels->networkCat);
diff --git a/cppe/src/TcpTransport/Connector.cpp b/cppe/src/TcpTransport/Connector.cpp
index c2c110f4dd3..04d44e49322 100644
--- a/cppe/src/TcpTransport/Connector.cpp
+++ b/cppe/src/TcpTransport/Connector.cpp
@@ -32,6 +32,7 @@ Connector::connect(int timeout)
 
     SOCKET fd = createSocket();
     setBlock(fd, false);
+    setTcpBufSize(fd, _instance->initializationData().properties, _logger);
     doConnect(fd, _addr, timeout);
 #ifndef ICEE_USE_SELECT_OR_POLL_FOR_TIMEOUTS
     setBlock(fd, true);
diff --git a/cppe/src/TcpTransport/Transceiver.cpp b/cppe/src/TcpTransport/Transceiver.cpp
index 9ee819d8057..8f0d18b5f4c 100644
--- a/cppe/src/TcpTransport/Transceiver.cpp
+++ b/cppe/src/TcpTransport/Transceiver.cpp
@@ -113,9 +113,9 @@ IceInternal::Transceiver::writeWithTimeout(Buffer& buf, int timeout)
     //
     // Limit packet size to avoid performance problems on WIN32
     //
-    if(_isPeerLocal && packetSize > 64 * 1024)
+    if(packetSize > _maxPacketSize)
     {
-	packetSize = 64 * 1024;
+	packetSize = _maxPacketSize;
     }
 #endif
 
@@ -378,9 +378,6 @@ IceInternal::Transceiver::Transceiver(const InstancePtr& instance, SOCKET fd) :
     _readTimeout(-1),
     _writeTimeout(-1),
     _desc(fdToString(fd))
-#ifdef _WIN32
-    , _isPeerLocal(isPeerLocal(fd))
-#endif
 {
 #ifdef ICEE_USE_SELECT_OR_POLL_FOR_TIMEOUTS
 #ifdef _WIN32
@@ -430,6 +427,19 @@ IceInternal::Transceiver::Transceiver(const InstancePtr& instance, SOCKET fd) :
     FD_ZERO(&_rFdSet);
 #endif
 #endif
+
+#ifdef _WIN32
+    //
+    // On Windows, limiting the buffer size is important to prevent
+    // poor throughput performances when transfering large amount of
+    // data. See Microsoft KB article KB823764.
+    //
+    _maxPacketSize = getSendBufferSize(_fd) / 2;
+    if(_maxPacketSize < 512)
+    {
+        _maxPacketSize = 512; // Make sure the packet size limiter isn't too small.
+    }
+#endif
 }
 
 IceInternal::Transceiver::~Transceiver()