Fixed IceDiscovery and collocation optimization issue which could cause indefinite hang if communicator was destroyed (ICE-6627)

3 files changed, 173 insertions, 69 deletions

diff --git a/java/src/Ice/src/main/java/IceInternal/CollocatedRequestHandler.java b/java/src/Ice/src/main/java/IceInternal/CollocatedRequestHandler.java
index 48b82937f11..55c2c69e7d3 100644
--- a/java/src/Ice/src/main/java/IceInternal/CollocatedRequestHandler.java
+++ b/java/src/Ice/src/main/java/IceInternal/CollocatedRequestHandler.java
@@ -78,6 +78,7 @@ public class CollocatedRequestHandler implements RequestHandler, ResponseHandler
             {
                 outAsync.invokeCompletedAsync();
             }
+            _adapter.decDirectCount(); // invokeAll won't be called, decrease the direct count.
             return;
         }
 
@@ -183,18 +184,32 @@ public class CollocatedRequestHandler implements RequestHandler, ResponseHandler
 
     int invokeAsyncRequest(OutgoingAsyncBase outAsync, int batchRequestNum, boolean synchronous)
     {
+        //
+        // Increase the direct count to prevent the thread pool from being destroyed before
+        // invokeAll is called. This will also throw if the object adapter has been deactivated.
+        //
+        _adapter.incDirectCount();
+
         int requestId = 0;
-        synchronized(this)
+        try
         {
-            outAsync.cancelable(this); // This will throw if the request is canceled
-
-            if(_response)
+            synchronized(this)
             {
-                requestId = ++_requestId;
-                _asyncRequests.put(requestId, outAsync);
-            }
+                outAsync.cancelable(this); // This will throw if the request is canceled
+
+                if(_response)
+                {
+                    requestId = ++_requestId;
+                    _asyncRequests.put(requestId, outAsync);
+                }
 
-            _sendAsyncRequests.put(outAsync, requestId);
+                _sendAsyncRequests.put(outAsync, requestId);
+            }
+        }
+        catch(Exception ex)
+        {
+            _adapter.decDirectCount();
+            throw ex;
         }
 
         outAsync.attachCollocatedObserver(_adapter, requestId);
@@ -288,6 +303,12 @@ public class CollocatedRequestHandler implements RequestHandler, ResponseHandler
         {
             while(invokeNum > 0)
             {
+                //
+                // Increase the direct count for the dispatch. We increase it again here for
+                // each dispatch. It's important for the direct count to be > 0 until the last
+                // collocated request response is sent to make sure the thread pool isn't
+                // destroyed before.
+                //
                 try
                 {
                     _adapter.incDirectCount();
@@ -295,7 +316,7 @@ public class CollocatedRequestHandler implements RequestHandler, ResponseHandler
                 catch(Ice.ObjectAdapterDeactivatedException ex)
                 {
                     handleException(requestId, ex, false);
-                    return;
+                    break;
                 }
 
                 Incoming in = new Incoming(_reference.getInstance(), this, null, _adapter, _response, (byte)0,
@@ -348,6 +369,8 @@ public class CollocatedRequestHandler implements RequestHandler, ResponseHandler
                 throw ex;
             }
         }
+
+        _adapter.decDirectCount();
     }
 
     private void
diff --git a/java/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java b/java/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java
index 386abe3c5f4..4d2aff7c370 100644
--- a/java/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java
+++ b/java/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java
@@ -213,7 +213,14 @@ class LookupI extends _LookupDisp
             //
             // Reply to the mulicast request using the given proxy.
             //
-            reply.begin_foundObjectById(id, proxy);
+            try
+            {
+               reply.begin_foundObjectById(id, proxy);
+            }
+            catch(Ice.LocalException ex)
+            {
+                // Ignore
+            }
         }
     }
 
@@ -233,7 +240,14 @@ class LookupI extends _LookupDisp
             //
             // Reply to the multicast request using the given proxy.
             //
-            reply.begin_foundAdapterById(adapterId, proxy, isReplicaGroup.value);
+            try
+            {
+                reply.begin_foundAdapterById(adapterId, proxy, isReplicaGroup.value);
+            }
+            catch(Ice.LocalException ex)
+            {
+                // Ignore
+            }
         }
     }
 
@@ -249,8 +263,16 @@ class LookupI extends _LookupDisp
 
         if(request.addCallback(cb))
         {
-            _lookup.begin_findObjectById(_domainId, id, _lookupReply);
-            request.scheduleTimer(_timeout);
+            try
+            {
+                _lookup.begin_findObjectById(_domainId, id, _lookupReply);
+                request.scheduleTimer(_timeout);
+            }
+            catch(Ice.LocalException ex)
+            {
+                request.finished(null);
+                _objectRequests.remove(id);
+            }
         }
     }
 
@@ -266,8 +288,16 @@ class LookupI extends _LookupDisp
 
         if(request.addCallback(cb))
         {
-            _lookup.begin_findAdapterById(_domainId, adapterId, _lookupReply);
-            request.scheduleTimer(_timeout);
+            try
+            {
+                _lookup.begin_findAdapterById(_domainId, adapterId, _lookupReply);
+                request.scheduleTimer(_timeout);
+            }
+            catch(Ice.LocalException ex)
+            {
+                request.finished(null);
+                _adapterRequests.remove(adapterId);
+            }
         }
     }
 
@@ -312,14 +342,19 @@ class LookupI extends _LookupDisp
 
         if(request.retry())
         {
-            _lookup.begin_findObjectById(_domainId, request.getId(), _lookupReply);
-            request.scheduleTimer(_timeout);
-        }
-        else
-        {
-            request.finished(null);
-            _objectRequests.remove(request.getId());
+            try
+            {
+                _lookup.begin_findObjectById(_domainId, request.getId(), _lookupReply);
+                request.scheduleTimer(_timeout);
+                return;
+            }
+            catch(Ice.LocalException ex)
+            {
+            }
         }
+
+        request.finished(null);
+        _objectRequests.remove(request.getId());
     }
 
     synchronized void
@@ -333,14 +368,19 @@ class LookupI extends _LookupDisp
 
         if(request.retry())
         {
-            _lookup.begin_findAdapterById(_domainId, request.getId(), _lookupReply);
-            request.scheduleTimer(_timeout);
-        }
-        else
-        {
-            request.finished(null);
-            _adapterRequests.remove(request.getId());
+            try
+            {
+                _lookup.begin_findAdapterById(_domainId, request.getId(), _lookupReply);
+                request.scheduleTimer(_timeout);
+                return;
+            }
+            catch(Ice.LocalException ex)
+            {
+            }
         }
+
+        request.finished(null);
+        _adapterRequests.remove(request.getId());
     }
 
     private LocatorRegistryI _registry;
@@ -355,6 +395,5 @@ class LookupI extends _LookupDisp
 
     private Map<Ice.Identity, ObjectRequest> _objectRequests = new HashMap<Ice.Identity, ObjectRequest>();
     private Map<String, AdapterRequest> _adapterRequests = new HashMap<String, AdapterRequest>();
-
 }
 
diff --git a/java/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java b/java/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java
index c93d4f19df3..b19bc6d59f8 100644
--- a/java/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java
+++ b/java/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java
@@ -36,38 +36,63 @@ class PluginI implements Ice.Plugin
         invoke(Ice.LocatorPrx l)
         {
             _locatorPrx = l;
-            l.begin_ice_invoke(_operation, _mode, _inParams, _context,
-                new Ice.Callback_Object_ice_invoke()
-                {
-                    @Override
-                    public void
-                    response(boolean ok, byte[] outParams)
-                    {
-                        _amdCB.ice_response(ok, outParams);
-                    }
-
-                    @Override
-                    public void
-                    exception(Ice.LocalException ex)
+            try
+            {
+                l.begin_ice_invoke(_operation, _mode, _inParams, _context,
+                    new Ice.Callback_Object_ice_invoke()
                     {
-                        try
-                        {
-                            throw ex;
-                        }
-                        catch(Ice.RequestFailedException exc)
-                        {
-                            _amdCB.ice_exception(ex);
-                        }
-                        catch(Ice.UnknownException exc)
+                        @Override
+                        public void
+                        response(boolean ok, byte[] outParams)
                         {
-                            _amdCB.ice_exception(ex);
+                            _amdCB.ice_response(ok, outParams);
                         }
-                        catch(Ice.LocalException exc)
+
+                        @Override
+                        public void
+                        exception(Ice.LocalException ex)
                         {
-                            _locator.invoke(_locatorPrx, Request.this); // Retry with new locator proxy
+                            Request.this.exception(ex);
                         }
-                    }
-                });
+                    });
+            }
+            catch(Ice.LocalException ex)
+            {
+                exception(ex);
+            }
+        }
+
+        private void
+        exception(Ice.LocalException ex)
+        {
+            try
+            {
+                throw ex;
+            }
+            catch(Ice.RequestFailedException exc)
+            {
+                _amdCB.ice_exception(ex);
+            }
+            catch(Ice.UnknownException exc)
+            {
+                _amdCB.ice_exception(ex);
+            }
+            catch(Ice.NoEndpointException exc)
+            {
+                _amdCB.ice_exception(new Ice.ObjectNotExistException());
+            }
+            catch(Ice.ObjectAdapterDeactivatedException exc)
+            {
+                _amdCB.ice_exception(new Ice.ObjectNotExistException());
+            }
+            catch(Ice.CommunicatorDestroyedException exc)
+            {
+                _amdCB.ice_exception(new Ice.ObjectNotExistException());
+            }
+            catch(Ice.LocalException exc)
+            {
+                _locator.invoke(_locatorPrx, Request.this); // Retry with new locator proxy
+            }
         }
 
         private final LocatorI _locator;
@@ -242,8 +267,20 @@ class PluginI implements Ice.Plugin
                 if(_pendingRetryCount == 0) // No request in progress
                 {
                     _pendingRetryCount = _retryCount;
-                    _lookup.begin_findLocator(_instanceName, _lookupReply); // Send multicast request.
-                    _future = _timer.schedule(_retryTask, _timeout, java.util.concurrent.TimeUnit.MILLISECONDS);
+                    try
+                    {
+                        _lookup.begin_findLocator(_instanceName, _lookupReply); // Send multicast request.
+                        _future = _timer.schedule(_retryTask, _timeout, java.util.concurrent.TimeUnit.MILLISECONDS);
+                    }
+                    catch(Ice.LocalException ex)
+                    {
+                        for(Request req : _pendingRequests)
+                        {
+                            req.invoke(_voidLocator);
+                        }
+                        _pendingRequests.clear();
+                        _pendingRetryCount = 0;
+                    }
                 }
             }
         }
@@ -257,19 +294,24 @@ class PluginI implements Ice.Plugin
                 {
                     if(--_pendingRetryCount > 0)
                     {
-                        _lookup.begin_findLocator(_instanceName, _lookupReply); // Send multicast request.
-                        _future = _timer.schedule(_retryTask, _timeout, java.util.concurrent.TimeUnit.MILLISECONDS);
-                    }
-                    else
-                    {
-                        assert !_pendingRequests.isEmpty();
-                        for(Request req : _pendingRequests)
+                        try
                         {
-                            req.invoke(_voidLocator);
+                            _lookup.begin_findLocator(_instanceName, _lookupReply); // Send multicast request.
+                            _future = _timer.schedule(_retryTask, _timeout, java.util.concurrent.TimeUnit.MILLISECONDS);
+                            return;
                         }
-                        _pendingRequests.clear();
-                        _nextRetry = IceInternal.Time.currentMonotonicTimeMillis() + _retryDelay;
+                        catch(Ice.LocalException ex)
+                        {
+                        }
+                        _pendingRetryCount = 0;
+                    }
+
+                    for(Request req : _pendingRequests)
+                    {
+                        req.invoke(_voidLocator);
                     }
+                    _pendingRequests.clear();
+                    _nextRetry = IceInternal.Time.currentMonotonicTimeMillis() + _retryDelay;
                 }
 
             }