Fixed again ICE-5687 - adapterDeactivation test warnings

6 files changed, 132 insertions, 48 deletions

diff --git a/cpp/src/Ice/RetryQueue.cpp b/cpp/src/Ice/RetryQueue.cpp
index 19fa86c61d5..6d58bc66fda 100644
--- a/cpp/src/Ice/RetryQueue.cpp
+++ b/cpp/src/Ice/RetryQueue.cpp
@@ -26,10 +26,15 @@ IceInternal::RetryTask::RetryTask(const RetryQueuePtr& queue, const OutgoingAsyn
 void
 IceInternal::RetryTask::runTimerTask()
 {
-    if(_queue->remove(this))
-    {
-        _outAsync->__processRetry(false);
-    }
+    _outAsync->__processRetry(false);
+
+    //
+    // NOTE: this must be called last, destroy() blocks until all task
+    // are removed to prevent the client thread pool to be destroyed
+    // (we still need the client thread pool at this point to call
+    // exception callbacks with CommunicatorDestroyedException).
+    //
+    _queue->remove(this);
 }
 
 void
@@ -72,19 +77,39 @@ void
 IceInternal::RetryQueue::destroy()
 {
     Lock sync(*this);
-    for(set<RetryTaskPtr>::const_iterator p = _requests.begin(); p != _requests.end(); ++p)
+    assert(_instance);
+
+    set<RetryTaskPtr>::const_iterator p = _requests.begin();
+    while(p != _requests.end())
     {
-        _instance->timer()->cancel(*p);
-        (*p)->destroy();
+        if(_instance->timer()->cancel(*p))
+        {
+            (*p)->destroy();
+            _requests.erase(p++);
+        }
+        else
+        {
+            ++p;
+        }
     }
-    _requests.clear();
+
     _instance = 0;
+    while(!_requests.empty())
+    {
+        wait();
+    }
 }
 
-bool
+void
 IceInternal::RetryQueue::remove(const RetryTaskPtr& task)
 {
     Lock sync(*this);
-    return _requests.erase(task) > 0;
+    assert(_requests.find(task) != _requests.end());
+    _requests.erase(task);
+    if(!_instance && _requests.empty())
+    {
+        notify(); // If we are destroying the queue, destroy is probably waiting on the queue to be empty.
+    }
 }
 
+
diff --git a/cpp/src/Ice/RetryQueue.h b/cpp/src/Ice/RetryQueue.h
index 07307310c8d..ecb1b5dd7c1 100644
--- a/cpp/src/Ice/RetryQueue.h
+++ b/cpp/src/Ice/RetryQueue.h
@@ -38,7 +38,7 @@ private:
 };
 typedef IceUtil::Handle<RetryTask> RetryTaskPtr;
 
-class RetryQueue : public IceUtil::Shared, public IceUtil::Mutex
+class RetryQueue : public IceUtil::Shared, public IceUtil::Monitor<IceUtil::Mutex>
 {
 public:
 
@@ -49,7 +49,7 @@ public:
 
 private:
 
-    bool remove(const RetryTaskPtr&);
+    void remove(const RetryTaskPtr&);
     friend class RetryTask;
 
     InstancePtr _instance;
diff --git a/cs/src/Ice/RetryQueue.cs b/cs/src/Ice/RetryQueue.cs
index 03ca716e542..80620c1fe0c 100644
--- a/cs/src/Ice/RetryQueue.cs
+++ b/cs/src/Ice/RetryQueue.cs
@@ -10,6 +10,7 @@
 namespace IceInternal
 {
     using System.Collections.Generic;
+    using System.Diagnostics;
 
     public class RetryTask : TimerTask
     {
@@ -21,10 +22,15 @@ namespace IceInternal
 
         public void runTimerTask()
         {
-            if(_retryQueue.remove(this))
-            {
-                _outAsync.processRetry(false);
-            }
+            _outAsync.processRetry(false);
+
+            //
+            // NOTE: this must be called last, destroy() blocks until all task
+            // are removed to prevent the client thread pool to be destroyed
+            // (we still need the client thread pool at this point to call
+            // exception callbacks with CommunicatorDestroyedException).
+            //
+            _retryQueue.remove(this);
         }
 
         public void destroy()
@@ -62,22 +68,39 @@ namespace IceInternal
         {
             lock(this)
             {
+                Dictionary<RetryTask, object> keep = new Dictionary<RetryTask, object>();
                 foreach(RetryTask task in _requests.Keys)
                 {
-                    _instance.timer().cancel(task);
-                    task.destroy();
+                    if(_instance.timer().cancel(task))
+                    {
+                        task.destroy();
+                    }
+                    else
+                    {
+                        keep.Add(task, null);
+                    }
                 }
-                _requests.Clear();
+                _requests = keep;
                 _instance = null;
+                while(_requests.Count > 0)
+                {
+                    System.Threading.Monitor.Wait(this);
+                }
             }
         }
 
-        public bool
+        public void
         remove(RetryTask task)
         {
             lock(this)
             {
-                return _requests.Remove(task);
+                Debug.Assert(_requests.ContainsKey(task));
+                _requests.Remove(task);
+                if(_instance == null && _requests.Count == 0)
+                {
+                    // If we are destroying the queue, destroy is probably waiting on the queue to be empty.
+                    System.Threading.Monitor.Pulse(this);
+                }
             }
         }
 
diff --git a/java/src/IceInternal/RetryQueue.java b/java/src/IceInternal/RetryQueue.java
index 4d6f8144410..311d53b4cc3 100644
--- a/java/src/IceInternal/RetryQueue.java
+++ b/java/src/IceInternal/RetryQueue.java
@@ -31,20 +31,50 @@ public class RetryQueue
     synchronized public void
     destroy()
     {
+         java.util.HashSet<RetryTask> keep = new java.util.HashSet<RetryTask>();
         for(RetryTask task : _requests)
         {
-            task.destroy();
+            if(!task.destroy())
+            {
+                keep.add(task);
+            }
         }
-        _requests.clear();
+        _requests = keep;
         _instance = null;
+
+        //
+        // Wait for the tasks to be executed, it shouldn't take long
+        // since they couldn't be canceled. If interrupted, we
+        // preserve the interrupt.
+        //
+        boolean interrupted = false;
+        while(!_requests.isEmpty())
+        {
+            try
+            {
+                wait();
+            }
+            catch(InterruptedException ex)
+            {
+                interrupted = true;
+            }
+        }
+        if(interrupted)
+        {
+            Thread.currentThread().interrupt();
+        }
     }
 
-    synchronized boolean
+    synchronized void
     remove(RetryTask task)
     {
-        return _requests.remove(task);
+        _requests.remove(task);
+        if(_instance == null && _requests.isEmpty())
+        {
+            notify();
+        }
     }
 
     private Instance _instance;
-    final private java.util.HashSet<RetryTask> _requests = new java.util.HashSet<RetryTask>();
+    private java.util.HashSet<RetryTask> _requests = new java.util.HashSet<RetryTask>();
 }
diff --git a/java/src/IceInternal/RetryTask.java b/java/src/IceInternal/RetryTask.java
index 2c991f65819..713c8bc856a 100644
--- a/java/src/IceInternal/RetryTask.java
+++ b/java/src/IceInternal/RetryTask.java
@@ -21,17 +21,26 @@ class RetryTask implements Runnable
     public void
     run()
     {
-        if(_queue.remove(this))
-        {
-            _outAsync.processRetry(false);
-        }
+        _outAsync.processRetry(false);
+
+        //
+        // NOTE: this must be called last, destroy() blocks until all task
+        // are removed to prevent the client thread pool to be destroyed
+        // (we still need the client thread pool at this point to call
+        // exception callbacks with CommunicatorDestroyedException).
+        //
+        _queue.remove(this);
     }
 
-    public void
+    public boolean
     destroy()
     {
-        _future.cancel(false);
-        _outAsync.processRetry(true);
+        if(_future.cancel(false))
+        {
+            _outAsync.processRetry(true);
+            return true;
+        }
+        return false;
     }
 
     public void setFuture(java.util.concurrent.Future<?> future)
diff --git a/js/src/Ice/RetryQueue.js b/js/src/Ice/RetryQueue.js
index c33a3d9146d..5101dbb87fc 100644
--- a/js/src/Ice/RetryQueue.js
+++ b/js/src/Ice/RetryQueue.js
@@ -25,16 +25,17 @@ var RetryQueue = Class({
             throw new Ice.CommunicatorDestroyedException();
         }
         var task = new RetryTask(this, outAsync);
-        this._instance.timer().schedule(function()
-            {
-                task.run();
-            }, interval);
+        task.token = this._instance.timer().schedule(function()
+                                                     {
+                                                         task.run();
+                                                     }, interval);
         this._requests.push(task);
     },
     destroy: function()
     {
         for(var i = 0; i < this._requests.length; ++i)
         {
+            this._instance.timer().cancel(this._requests[i].token);
             this._requests[i].destroy();
         }
         this._requests = [];
@@ -46,9 +47,7 @@ var RetryQueue = Class({
         if(idx >= 0)
         {
             this._requests.splice(idx, 1);
-            return true;
         }
-        return false;
     }
 });
 Ice.RetryQueue = RetryQueue;
@@ -61,17 +60,15 @@ var RetryTask = Class({
     },
     run: function()
     {
-        if(this.queue.remove(this))
+        try
         {
-            try
-            {
-                this.outAsync.__invoke();
-            }
-            catch(ex)
-            {
-                this.outAsync.__invokeException(ex);
-            }
+            this.outAsync.__invoke();
         }
+        catch(ex)
+        {
+            this.outAsync.__invokeException(ex);
+        }
+        this.queue.remove(this);
     },
     destroy: function()
     {