Fixed ICE-8157 - ignore responses from old requests and removed duplicated proxies for replica groups

3 files changed, 45 insertions, 22 deletions

diff --git a/cpp/src/IceDiscovery/LookupI.cpp b/cpp/src/IceDiscovery/LookupI.cpp
index ae560cbacfd..6ffb3d94764 100644
--- a/cpp/src/IceDiscovery/LookupI.cpp
+++ b/cpp/src/IceDiscovery/LookupI.cpp
@@ -13,6 +13,7 @@
 #include <Ice/LocalException.h>
 #include <Ice/Initialize.h>
 #include <Ice/LoggerUtil.h>
+#include <Ice/UUID.h>
 
 #include <IceDiscovery/LookupI.h>
 #include <iterator>
@@ -83,7 +84,7 @@ private:
 #endif
 
 IceDiscovery::Request::Request(const LookupIPtr& lookup, int retryCount) :
-    _lookup(lookup), _retryCount(retryCount), _lookupCount(0), _failureCount(0)
+    _lookup(lookup), _requestId(Ice::generateUUID()), _retryCount(retryCount), _lookupCount(0), _failureCount(0)
 {
 }
 
@@ -98,9 +99,11 @@ IceDiscovery::Request::invoke(const string& domainId, const vector<pair<LookupPr
 {
     _lookupCount = lookups.size();
     _failureCount = 0;
+    Ice::Identity id;
+    id.name = _requestId;
     for(vector<pair<LookupPrxPtr, LookupReplyPrxPtr> >::const_iterator p = lookups.begin(); p != lookups.end(); ++p)
     {
-        invokeWithLookup(domainId, p->first, p->second);
+        invokeWithLookup(domainId, p->first, ICE_UNCHECKED_CAST(LookupReplyPrx, p->second->ice_identity(id)));
     }
 }
 
@@ -118,6 +121,12 @@ IceDiscovery::Request::exception()
     return false;
 }
 
+string
+IceDiscovery::Request::getRequestId() const
+{
+    return _requestId;
+}
+
 AdapterRequest::AdapterRequest(const LookupIPtr& lookup, const std::string& adapterId, int retryCount) :
     RequestT<std::string, AdapterCB>(lookup, adapterId, retryCount),
     _start(IceUtil::Time::now())
@@ -141,7 +150,7 @@ AdapterRequest::response(const Ice::ObjectPrxPtr& proxy, bool isReplicaGroup)
             _lookup->timer()->cancel(ICE_SHARED_FROM_THIS);
             _lookup->timer()->schedule(ICE_SHARED_FROM_THIS, _latency);
         }
-        _proxies.push_back(proxy);
+        _proxies.insert(proxy);
         return false;
     }
     finished(proxy);
@@ -158,13 +167,13 @@ AdapterRequest::finished(const ObjectPrxPtr& proxy)
     }
     else if(_proxies.size() == 1)
     {
-        RequestT<string, AdapterCB>::finished(_proxies[0]);
+        RequestT<string, AdapterCB>::finished(*_proxies.begin());
         return;
     }
 
     EndpointSeq endpoints;
     ObjectPrxPtr prx;
-    for(vector<ObjectPrxPtr>::const_iterator p = _proxies.begin(); p != _proxies.end(); ++p)
+    for(set<ObjectPrxPtr>::const_iterator p = _proxies.begin(); p != _proxies.end(); ++p)
     {
         if(!prx)
         {
@@ -440,11 +449,11 @@ LookupI::findAdapter(const AdapterCB& cb, const std::string& adapterId)
 }
 
 void
-LookupI::foundObject(const Ice::Identity& id, const Ice::ObjectPrxPtr& proxy)
+LookupI::foundObject(const Ice::Identity& id, const string& requestId, const Ice::ObjectPrxPtr& proxy)
 {
     Lock sync(*this);
     map<Ice::Identity, ObjectRequestPtr>::iterator p = _objectRequests.find(id);
-    if(p == _objectRequests.end())
+    if(p == _objectRequests.end() || p->second->getRequestId() != requestId) // Ignore responses from old requests
     {
         return;
     }
@@ -455,11 +464,12 @@ LookupI::foundObject(const Ice::Identity& id, const Ice::ObjectPrxPtr& proxy)
 }
 
 void
-LookupI::foundAdapter(const std::string& adapterId, const Ice::ObjectPrxPtr& proxy, bool isReplicaGroup)
+LookupI::foundAdapter(const string& adapterId, const string& requestId, const Ice::ObjectPrxPtr& proxy,
+                      bool isReplicaGroup)
 {
     Lock sync(*this);
     map<string, AdapterRequestPtr>::iterator p = _adapterRequests.find(adapterId);
-    if(p == _adapterRequests.end())
+    if(p == _adapterRequests.end() || p->second->getRequestId() != requestId) // Ignore responses from old requests
     {
         return;
     }
@@ -580,26 +590,28 @@ LookupReplyI::LookupReplyI(const LookupIPtr& lookup) : _lookup(lookup)
 
 #ifdef ICE_CPP11_MAPPING
 void
-LookupReplyI::foundObjectById(Identity id, shared_ptr<ObjectPrx> proxy, const Current&)
+LookupReplyI::foundObjectById(Identity id, shared_ptr<ObjectPrx> proxy, const Current& current)
 {
-    _lookup->foundObject(id, proxy);
+    _lookup->foundObject(id, current.id.name, proxy);
 }
 
 void
-LookupReplyI::foundAdapterById(string adapterId, shared_ptr<ObjectPrx> proxy, bool isReplicaGroup, const Current&)
+LookupReplyI::foundAdapterById(string adapterId, shared_ptr<ObjectPrx> proxy, bool isReplicaGroup,
+                               const Current& current)
 {
-    _lookup->foundAdapter(adapterId, proxy, isReplicaGroup);
+    _lookup->foundAdapter(adapterId, current.id.name, proxy, isReplicaGroup);
 }
 #else
 void
-LookupReplyI::foundObjectById(const Identity& id, const ObjectPrxPtr& proxy, const Current&)
+LookupReplyI::foundObjectById(const Identity& id, const ObjectPrxPtr& proxy, const Current& current)
 {
-    _lookup->foundObject(id, proxy);
+    _lookup->foundObject(id, current.id.name, proxy);
 }
 
 void
-LookupReplyI::foundAdapterById(const string& adapterId, const ObjectPrxPtr& proxy, bool isReplicaGroup, const Current&)
+LookupReplyI::foundAdapterById(const string& adapterId, const ObjectPrxPtr& proxy, bool isReplicaGroup,
+                               const Current& current)
 {
-    _lookup->foundAdapter(adapterId, proxy, isReplicaGroup);
+    _lookup->foundAdapter(adapterId, current.id.name, proxy, isReplicaGroup);
 }
 #endif
diff --git a/cpp/src/IceDiscovery/LookupI.h b/cpp/src/IceDiscovery/LookupI.h
index 6925cbf27da..8e196c63d21 100644
--- a/cpp/src/IceDiscovery/LookupI.h
+++ b/cpp/src/IceDiscovery/LookupI.h
@@ -16,6 +16,8 @@
 #include <IceUtil/Timer.h>
 #include <Ice/Properties.h>
 
+#include <set>
+
 namespace IceDiscovery
 {
 
@@ -30,6 +32,7 @@ public:
     virtual bool retry();
     void invoke(const std::string&, const std::vector<std::pair<LookupPrxPtr, LookupReplyPrxPtr> >&);
     bool exception();
+    std::string getRequestId() const;
 
     virtual void finished(const Ice::ObjectPrxPtr&) = 0;
 
@@ -38,6 +41,7 @@ protected:
     virtual void invokeWithLookup(const std::string&, const LookupPrxPtr&, const LookupReplyPrxPtr&) = 0;
 
     LookupIPtr _lookup;
+    const std::string _requestId;
     int _retryCount;
     size_t _lookupCount;
     size_t _failureCount;
@@ -129,7 +133,12 @@ private:
     virtual void invokeWithLookup(const std::string&, const LookupPrxPtr&, const LookupReplyPrxPtr&);
     virtual void runTimerTask();
 
-    std::vector<Ice::ObjectPrxPtr> _proxies;
+    //
+    // We use a set because the same IceDiscovery plugin might return multiple times
+    // the same proxy if it's accessible through multiple network interfaces and if we
+    // also sent the request to multiple interfaces.
+    //
+    std::set<Ice::ObjectPrxPtr> _proxies;
     IceUtil::Time _start;
     IceUtil::Time _latency;
 };
@@ -157,8 +166,8 @@ public:
     void findObject(const ObjectCB&, const Ice::Identity&);
     void findAdapter(const AdapterCB&, const std::string&);
 
-    void foundObject(const Ice::Identity&, const Ice::ObjectPrxPtr&);
-    void foundAdapter(const std::string&, const Ice::ObjectPrxPtr&, bool);
+    void foundObject(const Ice::Identity&, const std::string&, const Ice::ObjectPrxPtr&);
+    void foundAdapter(const std::string&, const std::string&, const Ice::ObjectPrxPtr&, bool);
 
     void adapterRequestTimedOut(const AdapterRequestPtr&);
     void adapterRequestException(const AdapterRequestPtr&, const Ice::LocalException&);
diff --git a/cpp/src/IceDiscovery/PluginI.cpp b/cpp/src/IceDiscovery/PluginI.cpp
index 8e03497e85e..b1e7e58edd2 100644
--- a/cpp/src/IceDiscovery/PluginI.cpp
+++ b/cpp/src/IceDiscovery/PluginI.cpp
@@ -148,8 +148,10 @@ PluginI::initialize()
     _lookup = ICE_MAKE_SHARED(LookupI, locatorRegistry, ICE_UNCHECKED_CAST(LookupPrx, lookupPrx), properties);
     _multicastAdapter->add(_lookup, Ice::stringToIdentity("IceDiscovery/Lookup"));
 
-    Ice::ObjectPrxPtr lookupReply = _replyAdapter->addWithUUID(ICE_MAKE_SHARED(LookupReplyI, _lookup))->ice_datagram();
-    _lookup->setLookupReply(ICE_UNCHECKED_CAST(LookupReplyPrx, lookupReply));
+    _replyAdapter->addDefaultServant(ICE_MAKE_SHARED(LookupReplyI, _lookup), "");
+    Ice::Identity id;
+    id.name = "dummy";
+    _lookup->setLookupReply(ICE_UNCHECKED_CAST(LookupReplyPrx, _replyAdapter->createProxy(id)->ice_datagram()));
 
     //
     // Setup locator on the communicator.