// ********************************************************************** // // Copyright (c) 2003-2007 ZeroC, Inc. All rights reserved. // // This copy of Ice is licensed to you under the terms described in the // ICE_LICENSE file included in this distribution. // // ********************************************************************** #include #include #include #include #include #include #include #include #include #include using namespace std; using namespace IceGrid; namespace IceGrid { struct ReplicaLoadComp : binary_function { bool operator()(const pair& lhs, const pair& rhs) { return lhs.first < rhs.first; } }; struct ReplicaPriorityComp : binary_function { bool operator()(const ServerAdapterEntryPtr& lhs, const ServerAdapterEntryPtr& rhs) { return lhs->getPriority() < rhs->getPriority(); } }; struct TransformToReplicaLoad : public unary_function > { public: TransformToReplicaLoad(LoadSample loadSample) : _loadSample(loadSample) { } pair operator()(const ServerAdapterEntryPtr& value) { return make_pair(value->getLeastLoadedNodeLoad(_loadSample), value); } LoadSample _loadSample; }; struct TransformToReplica : public unary_function&, ServerAdapterEntryPtr> { ServerAdapterEntryPtr operator()(const pair& value) { return value.second; } }; } AdapterCache::AdapterCache(const Ice::CommunicatorPtr& communicator) : _communicator(communicator) { } void AdapterCache::addServerAdapter(const AdapterDescriptor& desc, const ServerEntryPtr& server, const string& app) { Lock sync(*this); if(getImpl(desc.id)) { Ice::Error out(_communicator->getLogger()); out << "can't add duplicate adapter `" << desc.id << "'"; return; } istringstream is(desc.priority); int priority = 0; is >> priority; ServerAdapterEntryPtr entry = new ServerAdapterEntry(*this, desc.id, app, desc.replicaGroupId, priority, server); addImpl(desc.id, entry); if(!desc.replicaGroupId.empty()) { ReplicaGroupEntryPtr repEntry = ReplicaGroupEntryPtr::dynamicCast(getImpl(desc.replicaGroupId)); if(!repEntry) { Ice::Error out(_communicator->getLogger()); out << "can't add adapter `" << desc.id << "' to unknown replica group `" << desc.replicaGroupId << "'"; } repEntry->addReplica(desc.id, entry); } } void AdapterCache::addReplicaGroup(const ReplicaGroupDescriptor& desc, const string& app) { Lock sync(*this); if(getImpl(desc.id)) { Ice::Error out(_communicator->getLogger()); out << "can't add duplicate replica group `" << desc.id << "'"; return; } addImpl(desc.id, new ReplicaGroupEntry(*this, desc.id, app, desc.loadBalancing)); } AdapterEntryPtr AdapterCache::get(const string& id) const { Lock sync(*this); AdapterEntryPtr entry = getImpl(id); if(!entry) { throw AdapterNotExistException(id); } return entry; } void AdapterCache::removeServerAdapter(const string& id) { Lock sync(*this); ServerAdapterEntryPtr entry = ServerAdapterEntryPtr::dynamicCast(getImpl(id)); if(!entry) { Ice::Error out(_communicator->getLogger()); out << "can't remove unknown adapter `" << id << "'"; return; } removeImpl(id); string replicaGroupId = entry->getReplicaGroupId(); if(!replicaGroupId.empty()) { ReplicaGroupEntryPtr repEntry = ReplicaGroupEntryPtr::dynamicCast(getImpl(replicaGroupId)); if(!repEntry) { Ice::Error out(_communicator->getLogger()); out << "can't remove adapter `" << id << "' from unknown replica group `" << replicaGroupId << "'"; } repEntry->removeReplica(id); } } void AdapterCache::removeReplicaGroup(const string& id) { Lock sync(*this); ReplicaGroupEntryPtr entry = ReplicaGroupEntryPtr::dynamicCast(getImpl(id)); if(!entry) { Ice::Error out(_communicator->getLogger()); out << "can't remove unknown replica group `" << id << "'"; return; } removeImpl(id); } AdapterEntryPtr AdapterCache::addImpl(const string& id, const AdapterEntryPtr& entry) { if(_traceLevels && _traceLevels->adapter > 0) { Ice::Trace out(_traceLevels->logger, _traceLevels->adapterCat); out << "added adapter `" << id << "'"; } return Cache::addImpl(id, entry); } void AdapterCache::removeImpl(const string& id) { if(_traceLevels && _traceLevels->adapter > 0) { Ice::Trace out(_traceLevels->logger, _traceLevels->adapterCat); out << "removed adapter `" << id << "'"; } Cache::removeImpl(id); } AdapterEntry::AdapterEntry(AdapterCache& cache, const string& id, const string& application) : _cache(cache), _id(id), _application(application) { } bool AdapterEntry::canRemove() { return true; } string AdapterEntry::getId() const { return _id; } string AdapterEntry::getApplication() const { return _application; } ServerAdapterEntry::ServerAdapterEntry(AdapterCache& cache, const string& id, const string& application, const string& replicaGroupId, int priority, const ServerEntryPtr& server) : AdapterEntry(cache, id, application), _replicaGroupId(replicaGroupId), _priority(priority), _server(server) { } void ServerAdapterEntry::getLocatorAdapterInfo(LocatorAdapterInfoSeq& adapters, int& nReplicas, bool& replicaGroup) { nReplicas = 1; replicaGroup = false; LocatorAdapterInfo info; info.id = _id; info.proxy = _server->getAdapter(info.activationTimeout, info.deactivationTimeout, _id, true); adapters.push_back(info); } float ServerAdapterEntry::getLeastLoadedNodeLoad(LoadSample loadSample) const { try { return _server->getLoad(loadSample); } catch(const ServerNotExistException&) { // This might happen if the application is updated concurrently. } catch(const NodeNotExistException&) { // This might happen if the application is updated concurrently. } catch(const NodeUnreachableException&) { } catch(const Ice::Exception& ex) { Ice::Error error(_cache.getTraceLevels()->logger); error << "unexpected exception while getting node load:\n" << ex; } return 999.9f; } AdapterInfoSeq ServerAdapterEntry::getAdapterInfo() const { AdapterInfo info; info.id = _id; info.replicaGroupId = _replicaGroupId; try { info.proxy = _server->getAdapter(_id, true)->getDirectProxy(); } catch(const Ice::Exception&) { } AdapterInfoSeq infos; infos.push_back(info); return infos; } AdapterPrx ServerAdapterEntry::getProxy(const string& replicaGroupId, bool upToDate) const { if(replicaGroupId.empty()) { return _server->getAdapter(_id, upToDate); } else { if(_replicaGroupId != replicaGroupId) // Validate the replica group. { throw Ice::InvalidReplicaGroupIdException(); } return _server->getAdapter(_id, upToDate); } } int ServerAdapterEntry::getPriority() const { return _priority; } ReplicaGroupEntry::ReplicaGroupEntry(AdapterCache& cache, const string& id, const string& application, const LoadBalancingPolicyPtr& policy) : AdapterEntry(cache, id, application), _lastReplica(0) { update(policy); } void ReplicaGroupEntry::addReplica(const string& replicaId, const ServerAdapterEntryPtr& adapter) { Lock sync(*this); _replicas.push_back(adapter); } void ReplicaGroupEntry::removeReplica(const string& replicaId) { Lock sync(*this); for(vector::iterator p = _replicas.begin(); p != _replicas.end(); ++p) { if(replicaId == (*p)->getId()) { _replicas.erase(p); // Make sure _lastReplica is still within the bounds. _lastReplica = _replicas.empty() ? 0 : _lastReplica % static_cast(_replicas.size()); break; } } } void ReplicaGroupEntry::update(const LoadBalancingPolicyPtr& policy) { Lock sync(*this); assert(policy); _loadBalancing = policy; istringstream is(_loadBalancing->nReplicas); int nReplicas = 0; is >> nReplicas; _loadBalancingNReplicas = nReplicas < 0 ? 1 : nReplicas; AdaptiveLoadBalancingPolicyPtr alb = AdaptiveLoadBalancingPolicyPtr::dynamicCast(_loadBalancing); if(alb) { if(alb->loadSample == "1") { _loadSample = LoadSample1; } else if(alb->loadSample == "5") { _loadSample = LoadSample5; } else if(alb->loadSample == "15") { _loadSample = LoadSample15; } else { _loadSample = LoadSample1; } } } void ReplicaGroupEntry::getLocatorAdapterInfo(LocatorAdapterInfoSeq& adapters, int& nReplicas, bool& replicaGroup) { vector replicas; bool adaptive = false; LoadSample loadSample = LoadSample1; { Lock sync(*this); replicaGroup = true; nReplicas = _loadBalancingNReplicas > 0 ? _loadBalancingNReplicas : static_cast(_replicas.size()); if(_replicas.empty()) { return; } replicas.reserve(_replicas.size()); if(RoundRobinLoadBalancingPolicyPtr::dynamicCast(_loadBalancing)) { for(unsigned int i = 0; i < _replicas.size(); ++i) { replicas.push_back(_replicas[(_lastReplica + i) % _replicas.size()]); } _lastReplica = (_lastReplica + 1) % static_cast(_replicas.size()); } else if(AdaptiveLoadBalancingPolicyPtr::dynamicCast(_loadBalancing)) { replicas = _replicas; RandomNumberGenerator rng; random_shuffle(replicas.begin(), replicas.end(), rng); loadSample = _loadSample; adaptive = true; } else if(OrderedLoadBalancingPolicyPtr::dynamicCast(_loadBalancing)) { replicas = _replicas; sort(replicas.begin(), replicas.end(), ReplicaPriorityComp()); } else if(RandomLoadBalancingPolicyPtr::dynamicCast(_loadBalancing)) { replicas = _replicas; RandomNumberGenerator rng; random_shuffle(replicas.begin(), replicas.end(), rng); } } if(adaptive) { // // This must be done outside the synchronization block since // the trasnform() might call and lock each server adapter // entry. We also can't sort directly as the load of each // server adapter is not stable so we first take a snapshot of // each adapter and sort the snapshot. // vector > rl; transform(replicas.begin(), replicas.end(), back_inserter(rl), TransformToReplicaLoad(loadSample)); sort(rl.begin(), rl.end(), ReplicaLoadComp()); replicas.clear(); transform(rl.begin(), rl.end(), back_inserter(replicas), TransformToReplica()); } // // Retrieve the proxy of each adapter from the server. The adapter // might not exist anymore at this time or the node might not be // reachable. // for(vector::const_iterator p = replicas.begin(); p != replicas.end(); ++p) { try { int dummy; bool dummy2; (*p)->getLocatorAdapterInfo(adapters, dummy, dummy2); } catch(const AdapterNotExistException&) { } catch(const NodeUnreachableException&) { } catch(const DeploymentException&) { } } } float ReplicaGroupEntry::getLeastLoadedNodeLoad(LoadSample loadSample) const { vector replicas; { Lock sync(*this); replicas = _replicas; } if(replicas.empty()) { return 999.9f; } else if(replicas.size() == 1) { return replicas.back()->getLeastLoadedNodeLoad(loadSample); } else { RandomNumberGenerator rng; random_shuffle(replicas.begin(), replicas.end(), rng); vector > rl; transform(replicas.begin(), replicas.end(), back_inserter(rl), TransformToReplicaLoad(loadSample)); return min_element(rl.begin(), rl.end(), ReplicaLoadComp())->first; } } AdapterInfoSeq ReplicaGroupEntry::getAdapterInfo() const { // // This method is called with the database locked so we're sure // that no new adapters will be added or removed concurrently. // vector replicas; { Lock sync(*this); replicas = _replicas; } AdapterInfoSeq infos; for(vector::const_iterator p = replicas.begin(); p != replicas.end(); ++p) { AdapterInfoSeq infs = (*p)->getAdapterInfo(); assert(infs.size() == 1); infos.push_back(infs[0]); } return infos; } bool ReplicaGroupEntry::hasAdaptersFromOtherApplications() const { vector replicas; { Lock sync(*this); replicas = _replicas; } AdapterInfoSeq infos; for(vector::const_iterator p = replicas.begin(); p != replicas.end(); ++p) { if((*p)->getApplication() != _application) { return true; } } return false; }