Better support for slow server activation and IcePack registry timeout.

8 files changed, 264 insertions, 99 deletions

diff --git a/cpp/src/IcePack/AdapterI.cpp b/cpp/src/IcePack/AdapterI.cpp
index ca7d77c203e..54c5d6ac1ca 100644
--- a/cpp/src/IcePack/AdapterI.cpp
+++ b/cpp/src/IcePack/AdapterI.cpp
@@ -25,14 +25,19 @@ IcePack::StandaloneAdapterI::StandaloneAdapterI()
 }
 
 void
-IcePack::StandaloneAdapterI::getDirectProxy_async(const AMD_Adapter_getDirectProxyPtr& cb,
-						  bool activate,
-						  const Ice::Current&)
+StandaloneAdapterI::activate_async(const AMD_Adapter_activatePtr& cb, const Ice::Current&)
 {
     IceUtil::Mutex::Lock sync(*this);
     cb->ice_response(proxy);
 }
 
+Ice::ObjectPrx
+IcePack::StandaloneAdapterI::getDirectProxy(const Ice::Current&) const
+{
+    IceUtil::Mutex::Lock sync(*this);
+    return proxy;
+}
+
 void
 IcePack::StandaloneAdapterI::setDirectProxy(const Ice::ObjectPrx& prx, const Ice::Current&)
 {
diff --git a/cpp/src/IcePack/AdapterI.h b/cpp/src/IcePack/AdapterI.h
index 1c7e94f68d2..e9f98d6614e 100644
--- a/cpp/src/IcePack/AdapterI.h
+++ b/cpp/src/IcePack/AdapterI.h
@@ -27,7 +27,8 @@ public:
     StandaloneAdapterI(const AdapterFactoryPtr&);
     StandaloneAdapterI();
 
-    virtual void getDirectProxy_async(const AMD_Adapter_getDirectProxyPtr& cb, bool, const Ice::Current&);
+    virtual void activate_async(const AMD_Adapter_activatePtr&, const Ice::Current&);
+    virtual Ice::ObjectPrx getDirectProxy(const Ice::Current&) const;
     virtual void setDirectProxy(const ::Ice::ObjectPrx&, const ::Ice::Current&);
     virtual void destroy(const ::Ice::Current&);
 
diff --git a/cpp/src/IcePack/AdminI.cpp b/cpp/src/IcePack/AdminI.cpp
index fa57199fd67..9c21ab78345 100644
--- a/cpp/src/IcePack/AdminI.cpp
+++ b/cpp/src/IcePack/AdminI.cpp
@@ -1359,7 +1359,11 @@ IcePack::AdminI::getAdapterEndpoints(const string& id, const Current&) const
     AdapterPrx adapter = _adapterRegistry->findById(id);
     try
     {
-	return _communicator->proxyToString(adapter->getDirectProxy(true));
+	return _communicator->proxyToString(adapter->getDirectProxy());
+    }
+    catch(AdapterNotActiveException&)
+    {
+	return "";
     }
     catch(const Ice::ObjectNotExistException&)
     {
diff --git a/cpp/src/IcePack/Internal.ice b/cpp/src/IcePack/Internal.ice
index 4bdb541316d..654c60618d7 100644
--- a/cpp/src/IcePack/Internal.ice
+++ b/cpp/src/IcePack/Internal.ice
@@ -81,23 +81,40 @@ exception AdapterActiveException
 {
 };
 
+exception AdapterNotActiveException
+{
+    /** True if the adapter can be activated on demand. */
+    bool activatable;
+    
+    /** How long to wait for the adapter to become active. */
+    int timeout;
+};
+
 interface Adapter
 {
     /**
      *
+     * Activate this adapter. If this adapter can be activated, this 
+     * will activate the adapter and return the direct proxy of the 
+     * adapter once it's active. If this adapter can be activated on
+     * demand, this will return 0 if the adapter is inactive or the
+     * adapter direct proxy it's active.
+     *
+     **/
+    ["ami", "amd"] Object* activate();    
+
+    /**
+     *
      * Get the adapter direct proxy. The adapter direct proxy is a
      * proxy created with the object adapter. The proxy contains the
      * last known adapter endpoints.
      *
-     * @param activate If true and if the adapter is not active,
-     * [getDirectProxy] will activate the adapter and wait for its
-     * activation.
-     *
      * @return A direct proxy containing the last known adapter
-     * endpoints.
+     * endpoints if the adapter is already active.
      *
      **/
-    ["amd", "ami"] Object* getDirectProxy(bool activate);
+     ["ami"] nonmutating Object* getDirectProxy()
+	throws AdapterNotActiveException;
 
     /**
      *
diff --git a/cpp/src/IcePack/LocatorI.cpp b/cpp/src/IcePack/LocatorI.cpp
index 882c9c2c27c..4a245b79d28 100644
--- a/cpp/src/IcePack/LocatorI.cpp
+++ b/cpp/src/IcePack/LocatorI.cpp
@@ -23,70 +23,51 @@ class AMI_Adapter_getDirectProxyI : public AMI_Adapter_getDirectProxy
 {
 public:
 
-    AMI_Adapter_getDirectProxyI(const Ice::AMD_Locator_findAdapterByIdPtr& cb, 
-				const AdapterRegistryPtr& registry,
-				const string& id,
-				const AdapterPrx& adapter) : 
-	_cb(cb), _adapterRegistry(registry), _id(id), _adapter(adapter)
+    AMI_Adapter_getDirectProxyI(const LocatorIPtr& locator, const string& id, const AdapterPrx& adapter) : 
+	_locator(locator), _id(id), _adapter(adapter)
     {
     }
 
     virtual void ice_response(const ::Ice::ObjectPrx& obj)
     {
-	//
-	// Return the adapter dummy direct proxy.
-	//
-	_cb->ice_response(obj);
+	assert(obj);
+	_locator->getDirectProxyCallback(_adapter->ice_getIdentity(), obj);
     }
 
     virtual void ice_exception(const ::Ice::Exception& ex)
     {
-	try
-	{
-	    ex.ice_throw();
-	}
-	catch(const Ice::ObjectNotExistException&)
-	{
-	    //
-	    // Expected if the adapter is destroyed, if that's the case, we remove it from the adapter
-	    // registry.
-	    //
-	    try
-	    {
-		_adapterRegistry->remove(_id, _adapter);
-	    }
-	    catch(const AdapterNotExistException&)
-	    {
-	    }
-	    _cb->ice_exception(Ice::AdapterNotFoundException());
-	    return;
-	}
-	catch(const Ice::LocalException&)
-	{
-	    //
-	    // Expected if we couldn't contact the adapter object (possibly because the IcePack node is 
-	    // down). Return a null proxy in this case (the client will get empty endpoints and throw a
-	    // NoEndpointException).
-	    //
-	    _cb->ice_response(0);
-	    return;
-	}
-	catch(const Ice::Exception& ex)
-	{
-	    //
-	    // Rethrow unexpected exception.
-	    //
-	    _cb->ice_exception(ex);
-	    return;
-	}
+	_locator->getDirectProxyException(_adapter, _id, ex);
+    }
 
-	assert(false);
+private:
+
+    const LocatorIPtr _locator;
+    const string _id;
+    const AdapterPrx _adapter;
+};
+
+class AMI_Adapter_activateI : public AMI_Adapter_activate
+{
+public:
+
+    AMI_Adapter_activateI(const LocatorIPtr& locator, const string& id, const AdapterPrx& adapter) : 
+	_locator(locator), _id(id), _adapter(adapter)
+    {
+    }
+
+    virtual void ice_response(const ::Ice::ObjectPrx& obj)
+    {
+	_locator->getDirectProxyCallback(_adapter->ice_getIdentity(), obj);
+    }
+
+    virtual void ice_exception(const ::Ice::Exception& ex)
+    {
+	_locator->getDirectProxyException(_adapter, _id, ex);
     }
 
 private:
 
-    const Ice::AMD_Locator_findAdapterByIdPtr _cb;
-    const AdapterRegistryPtr& _adapterRegistry;
+    const LocatorIPtr _locator;
     const string _id;
     const AdapterPrx _adapter;
 };
@@ -169,9 +150,9 @@ private:
 
 }
 
-IcePack::LocatorI::LocatorI(const AdapterRegistryPtr& adapterRegistry, 
-			    const ObjectRegistryPtr& objectRegistry, 
-			    const Ice::LocatorRegistryPrx& locatorRegistry) :
+LocatorI::LocatorI(const AdapterRegistryPtr& adapterRegistry, 
+		   const ObjectRegistryPtr& objectRegistry, 
+		   const Ice::LocatorRegistryPrx& locatorRegistry) :
     _adapterRegistry(adapterRegistry),
     _objectRegistry(objectRegistry),
     _locatorRegistry(locatorRegistry)
@@ -183,9 +164,9 @@ IcePack::LocatorI::LocatorI(const AdapterRegistryPtr& adapterRegistry,
 // registry.
 //
 void
-IcePack::LocatorI::findObjectById_async(const Ice::AMD_Locator_findObjectByIdPtr& cb,
-					const Ice::Identity& id, 
-					const Ice::Current& current) const
+LocatorI::findObjectById_async(const Ice::AMD_Locator_findObjectByIdPtr& cb, 
+			       const Ice::Identity& id, 
+			       const Ice::Current& current) const
 {
     ObjectDescriptor obj;
     try
@@ -218,24 +199,138 @@ IcePack::LocatorI::findObjectById_async(const Ice::AMD_Locator_findObjectByIdPtr
 // registry. If found, we try to get its direct proxy.
 //
 void
-IcePack::LocatorI::findAdapterById_async(const Ice::AMD_Locator_findAdapterByIdPtr& cb, 
-					 const string& id, 
-					 const Ice::Current&) const
+LocatorI::findAdapterById_async(const Ice::AMD_Locator_findAdapterByIdPtr& cb, 
+				const string& id, 
+				const Ice::Current&) const
 {
+    AdapterPrx adapter;
     try
     {
-	AdapterPrx adapter = AdapterPrx::uncheckedCast(_adapterRegistry->findById(id));
-	AMI_Adapter_getDirectProxyPtr amiCB = new AMI_Adapter_getDirectProxyI(cb, _adapterRegistry, id, adapter);
-	adapter->getDirectProxy_async(amiCB, true);
+	adapter = AdapterPrx::uncheckedCast(_adapterRegistry->findById(id));
     }
     catch(const AdapterNotExistException&)
     {
 	throw Ice::AdapterNotFoundException();
     }
+
+    LocatorIPtr self = const_cast<LocatorI*>(this);
+    if(self->getDirectProxyRequest(cb, adapter))
+    {
+	try
+	{
+	    AMI_Adapter_getDirectProxyPtr amiCB = new AMI_Adapter_getDirectProxyI(self, id, adapter);
+	    adapter->getDirectProxy_async(amiCB);
+	}
+	catch(const Ice::LocalException& ex)
+	{
+	    self->getDirectProxyException(adapter, id, ex);
+	}
+    }
 }
 
 Ice::LocatorRegistryPrx
-IcePack::LocatorI::getRegistry(const Ice::Current&) const
+LocatorI::getRegistry(const Ice::Current&) const
 {
     return _locatorRegistry;
 }
+
+bool
+LocatorI::getDirectProxyRequest(const Ice::AMD_Locator_findAdapterByIdPtr& cb, const AdapterPrx& adapter)
+{
+    Lock sync(*this);
+    
+    //
+    // Check if there's already pending requests for this adapter. If that's the case,
+    // we just add this one to the queue. If not, we add it to the queue and initiate
+    // a call on the adapter to get its direct proxy.
+    //
+    PendingRequestsMap::iterator p;
+    p = _pendingRequests.insert(make_pair(adapter->ice_getIdentity(), PendingRequests())).first;
+    p->second.push_back(cb);
+    return p->second.size() == 1;
+}
+
+void
+LocatorI::getDirectProxyException(const AdapterPrx& adapter, const string& id, const Ice::Exception& ex)
+{
+    Lock sync(*this);
+
+    PendingRequestsMap::iterator p = _pendingRequests.find(adapter->ice_getIdentity());
+    assert(p != _pendingRequests.end());
+    try
+    {
+	ex.ice_throw();
+    }
+    catch(const AdapterNotActiveException& ex)
+    {
+	if(ex.activatable)
+	{
+	    //
+	    // Activate the adapter if it can be activated on demand. NOTE: we use the timeout
+	    // provided in the exception to activate the adapter. The timeout correspond to the
+	    // wait time configured for the server.
+	    //
+	    try
+	    {
+		AMI_Adapter_activatePtr amiCB = new AMI_Adapter_activateI(this, id, adapter);
+		AdapterPrx::uncheckedCast(adapter->ice_timeout(ex.timeout))->activate_async(amiCB);
+	    }
+	    catch(const Ice::LocalException& ex)
+	    {
+		getDirectProxyException(adapter, id, ex);
+	    }
+	    return;
+	}
+	else
+	{
+	    for(PendingRequests::const_iterator q = p->second.begin(); q != p->second.end(); ++q)
+	    {
+		(*q)->ice_response(0);
+	    }
+	}
+    }
+    catch(const Ice::ObjectNotExistException& ex)
+    {
+	//
+	// Expected if the adapter is destroyed, if that's the case, we remove it from the adapter registry.
+	//
+	try
+	{
+	    _adapterRegistry->remove(id, adapter);
+	}
+	catch(const AdapterNotExistException&)
+	{
+	}
+
+	for(PendingRequests::const_iterator q = p->second.begin(); q != p->second.end(); ++q)
+	{
+	    (*q)->ice_exception(Ice::AdapterNotFoundException());
+	}
+    }
+    catch(const Ice::LocalException&)
+    {
+	for(PendingRequests::const_iterator q = p->second.begin(); q != p->second.end(); ++q)
+	{
+	    (*q)->ice_response(0);
+	}
+    }
+    catch(const Ice::Exception&)
+    {
+	assert(false);
+    }
+    _pendingRequests.erase(p);
+}
+
+void
+LocatorI::getDirectProxyCallback(const Ice::Identity& adapterId, const Ice::ObjectPrx& proxy)
+{
+    Lock sync(*this);
+
+    PendingRequestsMap::iterator p = _pendingRequests.find(adapterId);
+    assert(p != _pendingRequests.end());
+    for(PendingRequests::const_iterator q = p->second.begin(); q != p->second.end(); ++q)
+    {
+	(*q)->ice_response(proxy);
+    }
+    _pendingRequests.erase(p);
+}
diff --git a/cpp/src/IcePack/LocatorI.h b/cpp/src/IcePack/LocatorI.h
index 08d18fd083b..b7f3ff62490 100644
--- a/cpp/src/IcePack/LocatorI.h
+++ b/cpp/src/IcePack/LocatorI.h
@@ -16,7 +16,7 @@
 namespace IcePack
 {
 
-class LocatorI : public Ice::Locator
+class LocatorI : public Ice::Locator, public IceUtil::Mutex
 {
 public:
 
@@ -30,13 +30,24 @@ public:
 
     virtual ::Ice::LocatorRegistryPrx getRegistry(const ::Ice::Current&) const;
 
-private:
+    bool getDirectProxyRequest(const ::Ice::AMD_Locator_findAdapterByIdPtr&, const AdapterPrx&);
+    void getDirectProxyException(const AdapterPrx&, const std::string&, const Ice::Exception&);
+    void getDirectProxyCallback(const Ice::Identity&, const Ice::ObjectPrx&);
     
+protected:
+
     const AdapterRegistryPtr _adapterRegistry;
     const ObjectRegistryPtr _objectRegistry;
     const Ice::LocatorRegistryPrx _locatorRegistry;
+
+    typedef std::vector<Ice::AMD_Locator_findAdapterByIdPtr> PendingRequests;
+    typedef std::map<Ice::Identity, PendingRequests> PendingRequestsMap;
+
+    PendingRequestsMap _pendingRequests;
 };
 
+typedef IceUtil::Handle<LocatorI> LocatorIPtr;
+
 }
 
 #endif
diff --git a/cpp/src/IcePack/ServerAdapterI.cpp b/cpp/src/IcePack/ServerAdapterI.cpp
index a8a61e61253..5ca0754d4fc 100644
--- a/cpp/src/IcePack/ServerAdapterI.cpp
+++ b/cpp/src/IcePack/ServerAdapterI.cpp
@@ -25,7 +25,7 @@ public:
  
     WaitForAdapterActivation(const ServerAdapterPtr& adapter, 
 			     const TraceLevelsPtr traceLevels,
-			     const AMD_Adapter_getDirectProxyPtr& cb) : 
+			     const AMD_Adapter_activatePtr& cb) : 
 	WaitItem(adapter),
 	_adapter(adapter),
 	_traceLevels(traceLevels),
@@ -35,7 +35,18 @@ public:
     
     virtual void execute()
     {
-	_adapter->getDirectProxy_async(_cb, false);
+	try
+	{
+	    _cb->ice_response(_adapter->getDirectProxy());
+	}
+	catch(const AdapterNotActiveException&)
+	{
+	    _cb->ice_response(0);
+	}
+	catch(const Ice::LocalException&)
+	{
+	    _cb->ice_response(0);
+	}
     }
 
     virtual void expired(bool destroyed)
@@ -50,14 +61,14 @@ public:
 
 private:
     
-    ServerAdapterPtr _adapter;
-    TraceLevelsPtr _traceLevels;
-    AMD_Adapter_getDirectProxyPtr _cb;
+    const ServerAdapterPtr _adapter;
+    const TraceLevelsPtr _traceLevels;
+    const AMD_Adapter_activatePtr _cb;
 };
 
 }
 
-IcePack::ServerAdapterI::ServerAdapterI(const ServerFactoryPtr& factory, const TraceLevelsPtr& traceLevels, 
+ServerAdapterI::ServerAdapterI(const ServerFactoryPtr& factory, const TraceLevelsPtr& traceLevels, 
 					Ice::Int waitTime) :
     _factory(factory),
     _traceLevels(traceLevels),
@@ -65,24 +76,22 @@ IcePack::ServerAdapterI::ServerAdapterI(const ServerFactoryPtr& factory, const T
 {
 }
 
-IcePack::ServerAdapterI::~ServerAdapterI()
+ServerAdapterI::~ServerAdapterI()
 {
 }
 
 string
-IcePack::ServerAdapterI::getId(const Ice::Current&)
+ServerAdapterI::getId(const Ice::Current&)
 {
     return id;
 }
 
 void
-IcePack::ServerAdapterI::getDirectProxy_async(const AMD_Adapter_getDirectProxyPtr& cb,
-					      bool activate, 
-					      const Ice::Current& current)
+ServerAdapterI::activate_async(const AMD_Adapter_activatePtr& cb, const Ice::Current& current)
 {
     {
-	::IceUtil::Mutex::Lock sync(*this);
-	if(_proxy || !activate)
+	Lock sync(*this);
+	if(_proxy)
 	{
 	    //
 	    // Return the adapter direct proxy.
@@ -111,7 +120,7 @@ IcePack::ServerAdapterI::getDirectProxy_async(const AMD_Adapter_getDirectProxyPt
 	    // Now that the server is activated, wait for the adapter
 	    // direct proxy to be set.
 	    //
-	    ::IceUtil::Mutex::Lock sync(*this);
+	    Lock sync(*this);
 	    if(!_proxy)
 	    {
 		_factory->getWaitQueue()->add(new WaitForAdapterActivation(this, _traceLevels, cb), _waitTime);
@@ -122,10 +131,9 @@ IcePack::ServerAdapterI::getDirectProxy_async(const AMD_Adapter_getDirectProxyPt
     catch(const Ice::ObjectNotExistException&)
     {
 	//
-	// The server associated to this adapter doesn't exist
-	// anymore. Somehow the database is inconsistent if this
-	// happens. The best thing to do is to destroy the adapter
-	// and throw an ObjectNotExist exception.
+	// The server associated to this adapter doesn't exist anymore. Somehow the database is 
+	// inconsistent if this happens. The best thing to do is to destroy the adapter and throw
+	// an ObjectNotExist exception.
 	//
 	destroy(current);
 
@@ -139,7 +147,7 @@ IcePack::ServerAdapterI::getDirectProxy_async(const AMD_Adapter_getDirectProxyPt
     // adapter proxy.
     //
     {
-	::IceUtil::Mutex::Lock sync(*this);
+	Lock sync(*this);
 	if(_traceLevels->adapter > 1)
 	{
 	    Ice::Trace out(_traceLevels->logger, _traceLevels->adapterCat);
@@ -150,10 +158,33 @@ IcePack::ServerAdapterI::getDirectProxy_async(const AMD_Adapter_getDirectProxyPt
     }   
 }
 
+Ice::ObjectPrx
+ServerAdapterI::getDirectProxy(const Ice::Current& current) const
+{
+    Lock sync(*this);
+
+    //
+    // Return the adapter direct proxy if it's set. Otherwise, throw. The caller can eventually
+    // activate the adapter if it's activatable.
+    //
+    if(_proxy)
+    {
+	return _proxy;
+    }
+    else
+    {
+	AdapterNotActiveException ex;
+	ServerState state = svr->getState();
+	ex.activatable = svr->getActivationMode() == OnDemand || state == Activating || state == Active;
+	ex.timeout = _waitTime.toMilliSeconds();
+	throw ex;
+    }
+}
+
 void
-IcePack::ServerAdapterI::setDirectProxy(const Ice::ObjectPrx& prx, const Ice::Current& current)
+ServerAdapterI::setDirectProxy(const Ice::ObjectPrx& prx, const Ice::Current& current)
 {
-    ::IceUtil::Mutex::Lock sync(*this);
+    Lock sync(*this);
 
     //
     // If the adapter proxy is not null the given proxy can only be null. We don't allow to overide an 
@@ -191,7 +222,7 @@ IcePack::ServerAdapterI::setDirectProxy(const Ice::ObjectPrx& prx, const Ice::Cu
 }
 
 void
-IcePack::ServerAdapterI::destroy(const Ice::Current& current)
+ServerAdapterI::destroy(const Ice::Current& current)
 {
     _factory->destroy(this, current.id);
 }
diff --git a/cpp/src/IcePack/ServerAdapterI.h b/cpp/src/IcePack/ServerAdapterI.h
index 4d20c179df9..9af116ff325 100644
--- a/cpp/src/IcePack/ServerAdapterI.h
+++ b/cpp/src/IcePack/ServerAdapterI.h
@@ -32,7 +32,8 @@ public:
 
     virtual std::string getId(const Ice::Current&);
 
-    virtual void getDirectProxy_async(const AMD_Adapter_getDirectProxyPtr& cb, bool, const Ice::Current&);
+    virtual void activate_async(const AMD_Adapter_activatePtr& cb, const Ice::Current&);
+    virtual Ice::ObjectPrx getDirectProxy(const Ice::Current&) const;
     virtual void setDirectProxy(const ::Ice::ObjectPrx&, const ::Ice::Current&);
     virtual void destroy(const ::Ice::Current&);