thread-per-connection fixes

3 files changed, 36 insertions, 7 deletions

diff --git a/cpp/src/Ice/ConnectionFactory.cpp b/cpp/src/Ice/ConnectionFactory.cpp
index 4533032af78..34012751e39 100644
--- a/cpp/src/Ice/ConnectionFactory.cpp
+++ b/cpp/src/Ice/ConnectionFactory.cpp
@@ -540,13 +540,11 @@ IceInternal::IncomingConnectionFactory::waitUntilFinished()
 	//
 	// First we wait until the factory is destroyed.
 	//
-	while(_acceptor)
+	while(_state != StateClosed)
 	{
 	    wait();
 	}
 	
-	assert(_state == StateClosed);
-
 	threadPerIncomingConnectionFactory = _threadPerIncomingConnectionFactory;
 	_threadPerIncomingConnectionFactory = 0;
 
diff --git a/cpp/src/Ice/UdpTransceiver.cpp b/cpp/src/Ice/UdpTransceiver.cpp
index 5b2dbc72c17..482ae72543f 100644
--- a/cpp/src/Ice/UdpTransceiver.cpp
+++ b/cpp/src/Ice/UdpTransceiver.cpp
@@ -50,6 +50,21 @@ IceInternal::UdpTransceiver::shutdownWrite()
 void
 IceInternal::UdpTransceiver::shutdownReadWrite()
 {
+    if(_traceLevels->network >= 2)
+    {
+	Trace out(_logger, _traceLevels->networkCat);
+	out << "shutting down udp connection for reading and writing\n" << toString();
+    }
+
+    //
+    // Set a flag and then shutdown the socket in order to wake a thread that is
+    // blocked in read().
+    //
+    IceUtil::Mutex::Lock sync(_shutdownReadWriteMutex);
+    _shutdownReadWrite = true;
+
+    assert(_fd != INVALID_SOCKET);
+    shutdownSocketReadWrite(_fd);
 }
 
 void
@@ -144,6 +159,17 @@ IceInternal::UdpTransceiver::read(Buffer& buf, int)
 
 repeat:
 
+    //
+    // Check the shutdown flag.
+    //
+    {
+	IceUtil::Mutex::Lock sync(_shutdownReadWriteMutex);
+	if(_shutdownReadWrite)
+	{
+	    throw ConnectionLostException(__FILE__, __LINE__);
+	}
+    }
+
     ssize_t ret;
     if(_connect)
     {
@@ -174,7 +200,7 @@ repeat:
 	assert(_fd != INVALID_SOCKET);
 	ret = ::recv(_fd, reinterpret_cast<char*>(&buf.b[0]), packetSize, 0);
     }
-    
+
     if(ret == SOCKET_ERROR)
     {
 	if(interrupted())
@@ -189,7 +215,7 @@ repeat:
 	    assert(_fd != INVALID_SOCKET);
 	    FD_SET(_fd, &_rFdSet);
 	    int rs = ::select(_fd + 1, &_rFdSet, 0, 0, 0);
-	    
+
 	    if(rs == SOCKET_ERROR)
 	    {
 		if(interrupted())
@@ -268,7 +294,8 @@ IceInternal::UdpTransceiver::UdpTransceiver(const InstancePtr& instance, const s
     _logger(instance->logger()),
     _incoming(false),
     _connect(true),
-    _warn(instance->properties()->getPropertyAsInt("Ice.Warn.Datagrams") > 0)
+    _warn(instance->properties()->getPropertyAsInt("Ice.Warn.Datagrams") > 0),
+    _shutdownReadWrite(false)
 {
     try
     {
@@ -301,7 +328,8 @@ IceInternal::UdpTransceiver::UdpTransceiver(const InstancePtr& instance, const s
     _stats(instance->stats()),
     _incoming(true),
     _connect(connect),
-    _warn(instance->properties()->getPropertyAsInt("Ice.Warn.Datagrams") > 0)
+    _warn(instance->properties()->getPropertyAsInt("Ice.Warn.Datagrams") > 0),
+    _shutdownReadWrite(false)
 {
     try
     {
diff --git a/cpp/src/Ice/UdpTransceiver.h b/cpp/src/Ice/UdpTransceiver.h
index 2300cadef29..79ba34a8b81 100644
--- a/cpp/src/Ice/UdpTransceiver.h
+++ b/cpp/src/Ice/UdpTransceiver.h
@@ -15,6 +15,7 @@
 #include <Ice/LoggerF.h>
 #include <Ice/StatsF.h>
 #include <Ice/Transceiver.h>
+#include <IceUtil/Mutex.h>
 
 #ifndef _WIN32
 #   include <netinet/in.h> // For struct sockaddr_in
@@ -68,6 +69,8 @@ private:
     const bool _warn;
     static const int _udpOverhead;
     static const int _maxPacketSize;
+    bool _shutdownReadWrite;
+    IceUtil::Mutex _shutdownReadWriteMutex;
 };
 
 }