Fixed non-thread safe AMD dispatch, fixes #448 (#449)

The caching of the output stream which was added back to solve #414 made the
dispatch of AMD requests non thread-safe if a dispatch interceptor was installed
and if the interceptor retried the dispatch. Multiple continuation could run
concurrently and eventually re-use the cached output stream to marshal multiple
responses.

3 files changed, 47 insertions, 8 deletions

diff --git a/cpp/test/Ice/interceptor/AMDInterceptorI.cpp b/cpp/test/Ice/interceptor/AMDInterceptorI.cpp
index e2e2d95b001..158cf2736fa 100644
--- a/cpp/test/Ice/interceptor/AMDInterceptorI.cpp
+++ b/cpp/test/Ice/interceptor/AMDInterceptorI.cpp
@@ -103,6 +103,15 @@ AMDInterceptorI::dispatch(Ice::Request& request)
 
         current.ctx["retry"] = "no";
     }
+    else if(current.ctx.find("retry") != current.ctx.end() && current.ctx["retry"] == "yes")
+    {
+        //
+        // Retry the dispatch to ensure that abandoning the result of the dispatch
+        // works fine and is thread-safe
+        //
+        _servant->ice_dispatch(request);
+        _servant->ice_dispatch(request);
+    }
 
 #ifdef ICE_CPP11_MAPPING
     _lastStatus = _servant->ice_dispatch(request, []() { return true; }, [this](exception_ptr ex) {
diff --git a/cpp/test/Ice/interceptor/Client.cpp b/cpp/test/Ice/interceptor/Client.cpp
index 191215fd59d..02317c19d2c 100644
--- a/cpp/test/Ice/interceptor/Client.cpp
+++ b/cpp/test/Ice/interceptor/Client.cpp
@@ -189,6 +189,16 @@ Client::runAmdTest(const Test::MyObjectPrxPtr& prx, const AMDInterceptorIPtr& in
     test(prx->amdAddWithRetry(33, 12) == 45);
     test(interceptor->getLastOperation() == "amdAddWithRetry");
     test(!interceptor->getLastStatus());
+    {
+        Ice::Context ctx;
+        ctx["retry"] = "yes";
+        for(int i = 0; i < 10; ++i)
+        {
+            test(prx->amdAdd(33, 12, ctx) == 45);
+            test(interceptor->getLastOperation() == "amdAdd");
+            test(!interceptor->getLastStatus());
+        }
+    }
     cout << "ok" << endl;
     cout << "testing user exception... " << flush;
     try
diff --git a/cpp/test/Ice/interceptor/MyObjectI.cpp b/cpp/test/Ice/interceptor/MyObjectI.cpp
index 8604cd7007f..f3164ac4426 100644
--- a/cpp/test/Ice/interceptor/MyObjectI.cpp
+++ b/cpp/test/Ice/interceptor/MyObjectI.cpp
@@ -83,13 +83,22 @@ MyObjectI::amdAddAsync(int x,
                        int y,
                        function<void(int)> response,
                        function<void(exception_ptr)>,
-                       const Ice::Current&)
+                       const Ice::Current& current)
 {
+    Ice::Context::const_iterator p = current.ctx.find("retry");
+    bool retry = p != current.ctx.end();
     std::thread t(
-        [x, y, response]()
+        [x, y, response, retry]()
         {
             this_thread::sleep_for(chrono::milliseconds(10));
-            response(x + y);
+            try
+            {
+                response(x + y);
+            }
+            catch(const Ice::ResponseSentException&)
+            {
+                test(retry);
+            }
         });
     t.detach();
 }
@@ -200,31 +209,42 @@ MyObjectI::amdBadSystemAddAsync(int,
 }
 #else
 void
-MyObjectI::amdAdd_async(const Test::AMD_MyObject_amdAddPtr& cb, int x, int y, const Ice::Current&)
+MyObjectI::amdAdd_async(const Test::AMD_MyObject_amdAddPtr& cb, int x, int y, const Ice::Current& current)
 {
     class ThreadI : public Thread
     {
     public:
 
-        ThreadI(const Test::AMD_MyObject_amdAddPtr& cb, int x, int y) :
+        ThreadI(const Test::AMD_MyObject_amdAddPtr& cb, int x, int y, bool retry) :
             _cb(cb),
             _x(x),
-            _y(y)
+            _y(y),
+            _retry(retry)
         {
         }
 
         void run()
         {
             ThreadControl::sleep(Time::milliSeconds(10));
-            _cb->ice_response(_x + _y);
+            try
+            {
+                _cb->ice_response(_x + _y);
+            }
+            catch(const Ice::ResponseSentException&)
+            {
+                test(_retry);
+            }
         }
     private:
         Test::AMD_MyObject_amdAddPtr _cb;
         int _x;
         int _y;
+        bool _retry;
     };
 
-    ThreadPtr thread = new ThreadI(cb, x, y);
+    Ice::Context::const_iterator p = current.ctx.find("retry");
+    bool retry = p != current.ctx.end();
+    ThreadPtr thread = new ThreadI(cb, x, y, retry);
     thread->start().detach();
 }