thread-safe shutdown

5 files changed, 98 insertions, 47 deletions

diff --git a/cpp/src/Ice/CommunicatorI.cpp b/cpp/src/Ice/CommunicatorI.cpp
index 05d1aaecc6f..85db7a08a7a 100644
--- a/cpp/src/Ice/CommunicatorI.cpp
+++ b/cpp/src/Ice/CommunicatorI.cpp
@@ -13,7 +13,6 @@
 #include <Ice/Instance.h>
 #include <Ice/ProxyFactory.h>
 #include <Ice/ThreadPool.h>
-#include <Ice/ObjectAdapter.h>
 #include <Ice/ServantFactoryManager.h>
 #include <Ice/ObjectAdapterFactory.h>
 #include <Ice/Logger.h>
@@ -28,37 +27,40 @@ void
 Ice::CommunicatorI::destroy()
 {
     JTCSyncT<JTCRecursiveMutex> sync(*this);
+
     shutdown();
-    _instance->destroy();
-    _instance = 0;
+
+    if (_instance)
+    {
+	_instance->destroy();
+	_instance = 0;
+    }
+
+    //
+    // Don't set _threadPool to null here! See the comments in the
+    // header file.
+    //
 }
 
 void
 Ice::CommunicatorI::shutdown()
 {
-    JTCSyncT<JTCRecursiveMutex> sync(*this);
-    if (!_instance)
-    {
-	throw CommunicatorDestroyedException(__FILE__, __LINE__);
-    }
-    _instance->objectAdapterFactory()->shutdown();
+    //
+    // No mutex locking here! This operation must be signal-safe.
+    //
+    _threadPool->initiateServerShutdown();
 }
 
 void
 Ice::CommunicatorI::waitForShutdown()
 {
-    ThreadPoolPtr threadPool;
-
+    JTCSyncT<JTCRecursiveMutex> sync(*this);
+    if (!_instance)
     {
-	JTCSyncT<JTCRecursiveMutex> sync(*this);
-	if (!_instance)
-	{
-	    throw CommunicatorDestroyedException(__FILE__, __LINE__);
-	}
-	threadPool = _instance->threadPool();
+	throw CommunicatorDestroyedException(__FILE__, __LINE__);
     }
 
-    threadPool->waitUntilServerFinished();
+    _threadPool->waitUntilServerFinished();
 }
 
 ObjectPrx
@@ -103,7 +105,7 @@ Ice::CommunicatorI::createObjectAdapterWithEndpoints(const string& name, const s
     {
 	throw CommunicatorDestroyedException(__FILE__, __LINE__);
     }
-    return _instance->objectAdapterFactory() -> createObjectAdapter(name, endpts);
+    return _instance->objectAdapterFactory()->createObjectAdapter(name, endpts);
 }
 
 void
@@ -161,9 +163,25 @@ Ice::CommunicatorI::getPickler()
     return _instance->pickler();
 }
 
-Ice::CommunicatorI::CommunicatorI(const PropertiesPtr& properties) :
-    _instance(new Instance(this, properties))
+Ice::CommunicatorI::CommunicatorI(const PropertiesPtr& properties)
 {
+    __setNoDelete(true);
+    try
+    {
+	_instance = new Instance(this, properties);
+    }
+    catch(...)
+    {
+	__setNoDelete(false);
+	throw;
+    }
+    __setNoDelete(false);
+
+    //
+    // Se the comments in the header file for an explanation of why we
+    // need _threadPool directly in CommunicatorI.
+    //
+    _threadPool = _instance->threadPool();
 }
 
 Ice::CommunicatorI::~CommunicatorI()
diff --git a/cpp/src/Ice/CommunicatorI.h b/cpp/src/Ice/CommunicatorI.h
index c9d402f469f..eaf797102f7 100644
--- a/cpp/src/Ice/CommunicatorI.h
+++ b/cpp/src/Ice/CommunicatorI.h
@@ -11,6 +11,7 @@
 #ifndef ICE_COMMUNICATOR_I_H
 #define ICE_COMMUNICATOR_I_H
 
+#include <Ice/ThreadPoolF.h>
 #include <Ice/Initialize.h>
 #include <Ice/Communicator.h>
 
@@ -50,6 +51,15 @@ private:
     friend ICE_API ::IceInternal::InstancePtr IceInternal::getInstance(const ::Ice::CommunicatorPtr&);
 
     ::IceInternal::InstancePtr _instance;
+
+    //
+    // We need _threadPool directly in CommunicatorI, and it must
+    // never be set to null. That's because the shutdown() operation
+    // is signal-safe, and thus must not access any mutex locks or
+    // _instance. It may only access _threadPool->initiateShutdown(),
+    // which is signal-safe as well.
+    //
+    ::IceInternal::ThreadPoolPtr _threadPool;
 };
 
 }
diff --git a/cpp/src/Ice/ObjectAdapterI.cpp b/cpp/src/Ice/ObjectAdapterI.cpp
index 6bca95c288d..66532312759 100644
--- a/cpp/src/Ice/ObjectAdapterI.cpp
+++ b/cpp/src/Ice/ObjectAdapterI.cpp
@@ -230,18 +230,7 @@ Ice::ObjectAdapterI::ObjectAdapterI(const InstancePtr& instance, const string& n
     string::size_type beg = 0;
     string::size_type end;
     
-    //
-    // Set the "no delete" flag to true, meaning that this object will
-    // not be deleted, even if the reference count drops to zero. This
-    // is needed because if the constructor of the CollectorFactory
-    // throws an exception, or if the CollectorFactories are destroyed
-    // with "deactivate" from within this constructor, all
-    // ObjectAdapterPtrs for this object will be destroyed, and thus
-    // this object would be deleted if the "no delete" flag is not
-    // set.
-    //
     __setNoDelete(true);
-
     try
     {
 	while (true)
@@ -258,10 +247,12 @@ Ice::ObjectAdapterI::ObjectAdapterI(const InstancePtr& instance, const string& n
 	    }
 	    
 	    string es = s.substr(beg, end - beg);
-	    
+	   
+	    //
 	    // Don't store the endpoint in the adapter. The Collector
 	    // might change it, for example, to fill in the real port
 	    // number if a zero port number is given.
+	    //
 	    EndpointPtr endp = Endpoint::endpointFromString(es);
 	    _collectorFactories.push_back(new CollectorFactory(instance, this, endp));
 	    
@@ -282,7 +273,6 @@ Ice::ObjectAdapterI::ObjectAdapterI(const InstancePtr& instance, const string& n
 	__setNoDelete(false);
 	throw;
     }
-
     __setNoDelete(false);
     
     if (_collectorFactories.empty())
diff --git a/cpp/src/Ice/ThreadPool.cpp b/cpp/src/Ice/ThreadPool.cpp
index 53bb5d0d956..e00cc92c324 100644
--- a/cpp/src/Ice/ThreadPool.cpp
+++ b/cpp/src/Ice/ThreadPool.cpp
@@ -13,11 +13,11 @@
 #include <Ice/Network.h>
 #include <Ice/LocalException.h>
 #include <Ice/Instance.h>
-#include <Ice/Communicator.h>
 #include <Ice/Properties.h>
 #include <Ice/Logger.h>
 #include <Ice/Functional.h>
 #include <Ice/Protocol.h>
+#include <Ice/ObjectAdapterFactory.h>
 #include <sstream>
 
 using namespace std;
@@ -54,6 +54,17 @@ IceInternal::ThreadPool::promoteFollower()
 }
 
 void
+IceInternal::ThreadPool::initiateServerShutdown()
+{
+    char c = 1;
+#ifdef WIN32
+    ::send(_fdIntrWrite, &c, 1, 0);
+#else
+    ::write(_fdIntrWrite, &c, 1);
+#endif
+}
+
+void
 IceInternal::ThreadPool::waitUntilServerFinished()
 {
     JTCSyncT<JTCMonitorT<JTCMutex> > sync(*this);
@@ -164,17 +175,24 @@ IceInternal::ThreadPool::destroy()
     setInterrupt();
 }
 
-void
+bool
 IceInternal::ThreadPool::clearInterrupt()
 {
-    char s[32]; // Clear up to 32 interrupts at once
+    bool shutdown = false;
+    char c;
 #ifdef WIN32
-    while (::recv(_fdIntrRead, s, 32, 0) == 32)
-	;
+    while (::recv(_fdIntrRead, &c, 1, 0) == 1)
 #else
-    while (::read(_fdIntrRead, s, 32) == 32)
-	;
+    while (::read(_fdIntrRead, &c, 1) == 1)
 #endif
+    {
+	if (c == 1) // Shutdown initiated?
+	{
+	    shutdown = true;
+	}
+    }
+
+    return shutdown;
 }
 
 void
@@ -191,6 +209,8 @@ IceInternal::ThreadPool::setInterrupt()
 void
 IceInternal::ThreadPool::run()
 {
+    bool shutdown = false;
+
     while (true)
     {
 	_threadMutex.lock();
@@ -199,6 +219,13 @@ IceInternal::ThreadPool::run()
 	InstancePtr instance;
 	
     repeatSelect:
+
+	if (shutdown) // Shutdown has been initiated
+	{
+	    shutdown = false;
+	    _instance->objectAdapterFactory()->shutdown();
+	}
+
 	fd_set fdSet;
 	memcpy(&fdSet, &_fdSet, sizeof(fd_set));
 	int ret;
@@ -218,7 +245,7 @@ IceInternal::ThreadPool::run()
 	{
 	    assert(_timeout);
 	    _timeout = 0;
-	    _instance->communicator()->shutdown();
+	    _instance->objectAdapterFactory()->shutdown();
 	    goto repeatSelect;
 	}
 	
@@ -248,9 +275,16 @@ IceInternal::ThreadPool::run()
 		return;
 	    }
 		
+	    bool again = false;
+	    
 	    if (FD_ISSET(_fdIntrRead, &fdSet))
 	    {
-		clearInterrupt();
+		shutdown = clearInterrupt();
+		if (shutdown)
+		{
+		    again = true;
+		}
+
 #ifdef WIN32
 		FD_CLR(static_cast<u_int>(_fdIntrRead), &fdSet);
 #else
@@ -258,12 +292,10 @@ IceInternal::ThreadPool::run()
 #endif
 	    }
 	    
-	    bool again = false;
-	    
 	    if (!_adds.empty())
 	    {
 		//
-		// New handlers have been addedf
+		// New handlers have been added
 		//
 		for (vector<pair<int, EventHandlerPtr> >::iterator p = _adds.begin(); p != _adds.end(); ++p)
 		{
diff --git a/cpp/src/Ice/ThreadPool.h b/cpp/src/Ice/ThreadPool.h
index c1374167d06..962bffe862d 100644
--- a/cpp/src/Ice/ThreadPool.h
+++ b/cpp/src/Ice/ThreadPool.h
@@ -28,6 +28,7 @@ public:
     void _register(int, const EventHandlerPtr&);
     void unregister(int);
     void promoteFollower();
+    void initiateServerShutdown(); // Signal-safe shutdown initiation.
     void waitUntilServerFinished();
     void waitUntilFinished();
     void joinWithAllThreads();
@@ -39,7 +40,7 @@ private:
     void destroy();
     friend class Instance;
 
-    void clearInterrupt();
+    bool clearInterrupt();
     void setInterrupt();
 
     void run();