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

3 files changed, 169 insertions, 84 deletions

diff --git a/csharp/src/Ice/CollocatedRequestHandler.cs b/csharp/src/Ice/CollocatedRequestHandler.cs
index 4fcd670003c..21d4a7dc198 100644
--- a/csharp/src/Ice/CollocatedRequestHandler.cs
+++ b/csharp/src/Ice/CollocatedRequestHandler.cs
@@ -64,6 +64,7 @@ namespace IceInternal
                     {
                         outAsync.invokeCompletedAsync(cb);
                     }
+                    _adapter.decDirectCount(); // invokeAll won't be called, decrease the direct count.
                     return;
                 }
                 if(outAsync is OutgoingAsync)
@@ -161,7 +162,14 @@ namespace IceInternal
         public bool invokeAsyncRequest(OutgoingAsyncBase outAsync, int batchRequestNum, bool synchronous,
                                        out Ice.AsyncCallback sentCallback)
         {
+            //
+            // 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;
+            try
             {
                 lock(this)
                 {
@@ -176,6 +184,11 @@ namespace IceInternal
                     _sendAsyncRequests.Add(outAsync, requestId);
                 }
             }
+            catch(System.Exception ex)
+            {
+                _adapter.decDirectCount();
+                throw ex;
+            }
 
             outAsync.attachCollocatedObserver(_adapter, requestId);
 
@@ -277,6 +290,12 @@ namespace IceInternal
             {
                 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();
@@ -284,7 +303,7 @@ namespace IceInternal
                     catch(Ice.ObjectAdapterDeactivatedException ex)
                     {
                         handleException(requestId, ex, false);
-                        return;
+                        break;
                     }
 
                     Incoming @in = new Incoming(_reference.getInstance(), this, null, _adapter, _response, (byte)0,
@@ -297,6 +316,8 @@ namespace IceInternal
             {
                 invokeException(requestId, ex, invokeNum, false); // Fatal invocation exception
             }
+
+            _adapter.decDirectCount();
         }
 
         void
diff --git a/csharp/src/IceDiscovery/LookupI.cs b/csharp/src/IceDiscovery/LookupI.cs
index 27a143f3d9d..d516eb8c16a 100644
--- a/csharp/src/IceDiscovery/LookupI.cs
+++ b/csharp/src/IceDiscovery/LookupI.cs
@@ -25,7 +25,7 @@ namespace IceDiscovery
         {
             return _id;
         }
-        
+
         public bool addCallback(AmdCB cb)
         {
             callbacks_.Add(cb);
@@ -55,7 +55,7 @@ namespace IceDiscovery
         {
             return _proxies.Count == 0 && --nRetry_ >= 0;
         }
-        
+
         public bool response(Ice.ObjectPrx proxy, bool isReplicaGroup)
         {
             if(isReplicaGroup)
@@ -102,7 +102,7 @@ namespace IceDiscovery
             }
             sendResponse(result.ice_endpoints(endpoints.ToArray()));
         }
-        
+
         public void runTimerTask()
         {
             lookup_.adapterRequestTimedOut(this);
@@ -116,7 +116,7 @@ namespace IceDiscovery
             }
             callbacks_.Clear();
         }
-        
+
         private List<Ice.ObjectPrx> _proxies = new List<Ice.ObjectPrx>();
         private long _start;
         private long _latency;
@@ -132,7 +132,7 @@ namespace IceDiscovery
         {
             finished(proxy);
         }
-        
+
         public void finished(Ice.ObjectPrx proxy)
         {
             foreach(Ice.AMD_Locator_findObjectById cb in callbacks_)
@@ -160,12 +160,12 @@ namespace IceDiscovery
             _domainId = properties.getProperty("IceDiscovery.DomainId");
             _timer = IceInternal.Util.getInstance(lookup.ice_getCommunicator()).timer();
         }
-        
+
         public void setLookupReply(LookupReplyPrx lookupReply)
         {
             _lookupReply = lookupReply;
         }
-        
+
         public override void findObjectById(string domainId, Ice.Identity id, IceDiscovery.LookupReplyPrx reply,
                                             Ice.Current c)
         {
@@ -180,11 +180,18 @@ namespace IceDiscovery
                 //
                 // Reply to the mulicast request using the given proxy.
                 //
-                getLookupReply(reply, c).begin_foundObjectById(id, proxy);
+                try
+                {
+                    reply.begin_foundObjectById(id, proxy);
+                }
+                catch(Ice.LocalException)
+                {
+                    // Ignore.
+                }
             }
         }
 
-        public override void findAdapterById(string domainId, string adapterId, IceDiscovery.LookupReplyPrx reply, 
+        public override void findAdapterById(string domainId, string adapterId, IceDiscovery.LookupReplyPrx reply,
                                              Ice.Current c)
         {
             if(!domainId.Equals(_domainId))
@@ -199,7 +206,14 @@ namespace IceDiscovery
                 //
                 // Reply to the multicast request using the given proxy.
                 //
-                getLookupReply(reply, c).begin_foundAdapterById(adapterId, proxy, isReplicaGroup);
+                try
+                {
+                    reply.begin_foundAdapterById(adapterId, proxy, isReplicaGroup);
+                }
+                catch(Ice.LocalException)
+                {
+                    // Ignore.
+                }
             }
         }
 
@@ -215,8 +229,16 @@ namespace IceDiscovery
                 }
                 if(request.addCallback(cb))
                 {
-                    _lookup.begin_findObjectById(_domainId, id, _lookupReply);
-                    _timer.schedule(request, _timeout);
+                    try
+                    {
+                        _lookup.begin_findObjectById(_domainId, id, _lookupReply);
+                        _timer.schedule(request, _timeout);
+                    }
+                    catch(Ice.LocalException)
+                    {
+                        request.finished(null);
+                        _objectRequests.Remove(id);
+                    }
                 }
             }
         }
@@ -233,8 +255,16 @@ namespace IceDiscovery
                 }
                 if(request.addCallback(cb))
                 {
-                    _lookup.begin_findAdapterById(_domainId, adapterId, _lookupReply);
-                    _timer.schedule(request, _timeout);
+                    try
+                    {
+                        _lookup.begin_findAdapterById(_domainId, adapterId, _lookupReply);
+                        _timer.schedule(request, _timeout);
+                    }
+                    catch(Ice.LocalException)
+                    {
+                        request.finished(null);
+                        _adapterRequests.Remove(adapterId);
+                    }
                 }
             }
         }
@@ -284,15 +314,20 @@ namespace IceDiscovery
 
                 if(request.retry())
                 {
-                    _lookup.begin_findObjectById(_domainId, request.getId(), _lookupReply);
-                    _timer.schedule(request, _timeout);
-                }
-                else
-                {
-                    request.finished(null);
-                    _objectRequests.Remove(request.getId());
-                    _timer.cancel(request);
+                    try
+                    {
+                        _lookup.begin_findObjectById(_domainId, request.getId(), _lookupReply);
+                        _timer.schedule(request, _timeout);
+                        return;
+                    }
+                    catch(Ice.LocalException)
+                    {
+                    }
                 }
+
+                request.finished(null);
+                _objectRequests.Remove(request.getId());
+                _timer.cancel(request);
             }
         }
 
@@ -308,15 +343,20 @@ namespace IceDiscovery
 
                 if(request.retry())
                 {
-                    _lookup.begin_findAdapterById(_domainId, request.getId(), _lookupReply);
-                    _timer.schedule(request, _timeout);
-                }
-                else
-                {
-                    request.finished(null);
-                    _adapterRequests.Remove(request.getId());
-                    _timer.cancel(request);
+                    try
+                    {
+                        _lookup.begin_findAdapterById(_domainId, request.getId(), _lookupReply);
+                        _timer.schedule(request, _timeout);
+                        return;
+                    }
+                    catch(Ice.LocalException)
+                    {
+                    }
                 }
+
+                request.finished(null);
+                _adapterRequests.Remove(request.getId());
+                _timer.cancel(request);
             }
         }
 
@@ -330,23 +370,6 @@ namespace IceDiscovery
             return _latencyMultiplier;
         }
 
-        private LookupReplyPrx getLookupReply(LookupReplyPrx reply, Ice.Current current)
-        {
-            // Ice.UDPConnectionInfo info = Ice.UDPConnectionInfoPtr.dynamicCast(current.con.getInfo());
-            // if(info)
-            // {
-            //     Ice.Communicator com = current.adapter.getCommunicator();
-            //     ostringstream os;
-            //     os << "\"" << com.identityToString(reply.ice_getIdentity()) << "\"";
-            //     os << ":udp -h " << info.remoteAddress << " -p " << info.remotePort;
-            //     return LookupReplyPrx.uncheckedCast(com.stringToProxy(os.str()).ice_datagram());
-            // }
-            // else
-            {
-                return reply;
-            }
-        }
-
         private LocatorRegistryI _registry;
         private readonly LookupPrx _lookup;
         private LookupReplyPrx _lookupReply;
@@ -367,7 +390,7 @@ namespace IceDiscovery
         {
             _lookup = lookup;
         }
-        
+
         public override void foundObjectById(Ice.Identity id, Ice.ObjectPrx proxy, Ice.Current c)
         {
             _lookup.foundObject(id, proxy);
diff --git a/csharp/src/IceLocatorDiscovery/PluginI.cs b/csharp/src/IceLocatorDiscovery/PluginI.cs
index f349eaa560c..ccf9b266850 100644
--- a/csharp/src/IceLocatorDiscovery/PluginI.cs
+++ b/csharp/src/IceLocatorDiscovery/PluginI.cs
@@ -44,31 +44,55 @@ namespace IceLocatorDiscovery
         invoke(Ice.LocatorPrx l)
         {
             _locatorPrx = l;
-            Request self = this;
-            l.begin_ice_invoke(_operation, _mode, _inParams, _context).whenCompleted(
-                (bool ok, byte[] outParams) =>
-                {
-                    _amdCB.ice_response(ok, outParams);
-                },
-                (Ice.Exception ex) =>
-                {
-                    try
-                    {
-                        throw ex;
-                    }
-                    catch(Ice.RequestFailedException exc)
-                    {
-                        _amdCB.ice_exception(exc);
-                    }
-                    catch(Ice.UnknownException exc)
+            try
+            {
+                l.begin_ice_invoke(_operation, _mode, _inParams, _context).whenCompleted(
+                    (bool ok, byte[] outParams) =>
                     {
-                        _amdCB.ice_exception(exc);
-                    }
-                    catch(Ice.Exception)
+                        _amdCB.ice_response(ok, outParams);
+                    },
+                    (Ice.Exception ex) =>
                     {
-                        _locator.invoke(_locatorPrx, self); // Retry with new locator proxy
-                    }
-                });
+                        exception(ex);
+                    });
+            }
+            catch(Ice.LocalException ex)
+            {
+                exception(ex);
+            }
+        }
+
+        private void
+        exception(Ice.Exception ex)
+        {
+            try
+            {
+                throw ex;
+            }
+            catch(Ice.RequestFailedException exc)
+            {
+                _amdCB.ice_exception(exc);
+            }
+            catch(Ice.UnknownException exc)
+            {
+                _amdCB.ice_exception(exc);
+            }
+            catch(Ice.NoEndpointException)
+            {
+                _amdCB.ice_exception(new Ice.ObjectNotExistException());
+            }
+            catch(Ice.ObjectAdapterDeactivatedException)
+            {
+                _amdCB.ice_exception(new Ice.ObjectNotExistException());
+            }
+            catch(Ice.CommunicatorDestroyedException)
+            {
+                _amdCB.ice_exception(new Ice.ObjectNotExistException());
+            }
+            catch(Ice.Exception)
+            {
+                _locator.invoke(_locatorPrx, this); // Retry with new locator proxy
+            }
         }
 
         private readonly LocatorI _locator;
@@ -243,8 +267,20 @@ namespace IceLocatorDiscovery
                     if(_pendingRetryCount == 0) // No request in progress
                     {
                         _pendingRetryCount = _retryCount;
-                        _lookup.begin_findLocator(_instanceName, _lookupReply); // Send multicast request.
-                        _timer.schedule(this, _timeout);
+                        try
+                        {
+                            _lookup.begin_findLocator(_instanceName, _lookupReply); // Send multicast request.
+                            _timer.schedule(this, _timeout);
+                        }
+                        catch(Ice.LocalException)
+                        {
+                            foreach(Request req in _pendingRequests)
+                            {
+                                req.invoke(_voidLocator);
+                            }
+                            _pendingRequests.Clear();
+                            _pendingRetryCount = 0;
+                        }
                     }
                 }
             }
@@ -257,19 +293,24 @@ namespace IceLocatorDiscovery
             {
                 if(--_pendingRetryCount > 0)
                 {
-                    _lookup.begin_findLocator(_instanceName, _lookupReply); // Send multicast request
-                    _timer.schedule(this, _timeout);
-                }
-                else
-                {
-                    Debug.Assert(_pendingRequests.Count > 0, "Pending requests is not empty");
-                    foreach(Request req in _pendingRequests)
+                    try
+                    {
+                        _lookup.begin_findLocator(_instanceName, _lookupReply); // Send multicast request
+                        _timer.schedule(this, _timeout);
+                        return;
+                    }
+                    catch(Ice.LocalException)
                     {
-                        req.invoke(_voidLocator);
                     }
-                    _pendingRequests.Clear();
-                    _nextRetry = IceInternal.Time.currentMonotonicTimeMillis() + _retryDelay;
+                    _pendingRetryCount = 0;
+                }
+
+                foreach(Request req in _pendingRequests)
+                {
+                    req.invoke(_voidLocator);
                 }
+                _pendingRequests.Clear();
+                _nextRetry = IceInternal.Time.currentMonotonicTimeMillis() + _retryDelay;
             }
         }