// ********************************************************************** // // Copyright (c) 2003-2006 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 using namespace std; using namespace IceGrid; pointer_to_unary_function ObjectCache::_rand(IceUtil::random); namespace IceGrid { struct ObjectLoadCI : binary_function&, pair&, bool> { bool operator()(const pair& lhs, const pair& rhs) { return lhs.second < rhs.second; } }; }; ObjectCache::TypeEntry::TypeEntry(ObjectCache& cache) : _cache(cache) { } void ObjectCache::TypeEntry::add(const Ice::ObjectPrx& obj) { // // No mutex protection here, this is called with the cache locked. // _objects.insert(lower_bound(_objects.begin(), _objects.end(), obj, ::Ice::proxyIdentityLess), obj); } bool ObjectCache::TypeEntry::remove(const Ice::ObjectPrx& obj) { // // No mutex protection here, this is called with the cache locked. // Ice::ObjectProxySeq::iterator q = lower_bound(_objects.begin(), _objects.end(), obj, ::Ice::proxyIdentityLess); assert((*q)->ice_getIdentity() == obj->ice_getIdentity()); _objects.erase(q); return _objects.empty(); } void ObjectCache::TypeEntry::addAllocationRequest(const ObjectAllocationRequestPtr& request) { // // No mutex protection here, this is called with the cache locked. // if(request->pending()) { _requests.push_back(request); } } bool ObjectCache::TypeEntry::canTryAllocate(const ObjectEntryPtr& entry) { // // No mutex protection here, this is called with the cache locked. // list::iterator p = _requests.begin(); while(p != _requests.end()) { AllocationRequestPtr request = *p; try { if(request->isCanceled()) // If the request has been canceled, we just remove it. { p = _requests.erase(p); } else if(entry->tryAllocate(request, true)) { p = _requests.erase(p); return true; // The request successfully allocated the entry! } else if(entry->getSession()) // If entry is allocated, we're done { return false; } ++p; } catch(const AllocationException&) { p = _requests.erase(p); } } return false; } ObjectCache::ObjectCache(AdapterCache& adapterCache) : _adapterCache(adapterCache) { } void ObjectCache::add(const ObjectInfo& info, const string& application, bool allocatable, const AllocatablePtr& parent) { const Ice::Identity& id = info.proxy->ice_getIdentity(); Lock sync(*this); assert(!getImpl(id)); ObjectEntryPtr entry = new ObjectEntry(*this, info, application, allocatable, parent); addImpl(id, entry); map::iterator p = _types.find(entry->getType()); if(p == _types.end()) { p = _types.insert(p, make_pair(entry->getType(), TypeEntry(*this))); } p->second.add(info.proxy); if(_traceLevels && _traceLevels->object > 0) { Ice::Trace out(_traceLevels->logger, _traceLevels->objectCat); out << "added object `" << Ice::identityToString(id) << "'"; } } ObjectEntryPtr ObjectCache::get(const Ice::Identity& id) const { Lock sync(*this); ObjectEntryPtr entry = getImpl(id); if(!entry) { throw ObjectNotRegisteredException(id); } return entry; } ObjectEntryPtr ObjectCache::remove(const Ice::Identity& id) { Lock sync(*this); ObjectEntryPtr entry = removeImpl(id); assert(entry); map::iterator p = _types.find(entry->getType()); assert(p != _types.end()); if(p->second.remove(entry->getObjectInfo().proxy)) { _types.erase(p); } if(_traceLevels && _traceLevels->object > 0) { Ice::Trace out(_traceLevels->logger, _traceLevels->objectCat); out << "removed object `" << Ice::identityToString(id) << "'"; } return entry; } void ObjectCache::allocateByType(const string& type, const ObjectAllocationRequestPtr& request) { Lock sync(*this); map::iterator p = _types.find(type); if(p == _types.end()) { request->response(0); return; } Ice::ObjectProxySeq objects = p->second.getObjects(); random_shuffle(objects.begin(), objects.end(), _rand); // TODO: OPTIMIZE try { for(Ice::ObjectProxySeq::const_iterator q = objects.begin(); q != objects.end(); ++q) { ObjectEntryPtr entry = getImpl((*q)->ice_getIdentity()); if(entry->tryAllocate(request)) { return; } } } catch(const AllocationException&) { return; } // // TODO: XXX Only add the allocation request if there's allocatables for this type!!! // p->second.addAllocationRequest(request); } void ObjectCache::allocateByTypeOnLeastLoadedNode(const string& type, const ObjectAllocationRequestPtr& request, LoadSample sample) { Lock sync(*this); map::iterator p = _types.find(type); if(p == _types.end()) { request->response(0); return; } Ice::ObjectProxySeq objects = p->second.getObjects(); random_shuffle(objects.begin(), objects.end(), _rand); // TODO: OPTIMIZE vector > objsWLoad; objsWLoad.reserve(objects.size()); for(Ice::ObjectProxySeq::const_iterator o = objects.begin(); o != objects.end(); ++o) { float load = 1.0f; if(!(*o)->ice_getAdapterId().empty()) { try { load = _adapterCache.get((*o)->ice_getAdapterId())->getLeastLoadedNodeLoad(sample); } catch(const AdapterNotExistException&) { } } objsWLoad.push_back(make_pair(*o, load)); } sort(objsWLoad.begin(), objsWLoad.end(), ObjectLoadCI()); for(vector >::const_iterator q = objsWLoad.begin(); q != objsWLoad.end(); ++q) { ObjectEntryPtr entry = getImpl(q->first->ice_getIdentity()); if(entry->allocatable()) { if(entry->tryAllocate(request)) { return; } else if(request->isCanceled()) { return; } } } // // TODO: XXX Only add the allocation request if there's allocatables for this type!!! // p->second.addAllocationRequest(request); } bool ObjectCache::canTryAllocate(const ObjectEntryPtr& entry) { // // Notify the type entry that an object was released. // Lock sync(*this); map::iterator p = _types.find(entry->getType()); if(p == _types.end()) { return false; } return p->second.canTryAllocate(entry); } Ice::ObjectProxySeq ObjectCache::getObjectsByType(const string& type) { Lock sync(*this); Ice::ObjectProxySeq proxies; map::const_iterator p = _types.find(type); if(p == _types.end()) { return proxies; } const Ice::ObjectProxySeq& objects = p->second.getObjects(); for(Ice::ObjectProxySeq::const_iterator q = objects.begin(); q != objects.end(); ++q) { ObjectEntryPtr entry = getImpl((*q)->ice_getIdentity()); if(!entry->allocatable()) // Only return non-allocatable objects. { proxies.push_back(*q); } } return proxies; } ObjectInfoSeq ObjectCache::getAll(const string& expression) { Lock sync(*this); ObjectInfoSeq infos; for(map::const_iterator p = _entries.begin(); p != _entries.end(); ++p) { if(expression.empty() || IceUtil::match(Ice::identityToString(p->first), expression, true)) { infos.push_back(p->second->getObjectInfo()); } } return infos; } ObjectInfoSeq ObjectCache::getAllByType(const string& type) { Lock sync(*this); ObjectInfoSeq infos; map::const_iterator p = _types.find(type); if(p == _types.end()) { return infos; } const Ice::ObjectProxySeq& objects = p->second.getObjects(); for(Ice::ObjectProxySeq::const_iterator q = objects.begin(); q != objects.end(); ++q) { infos.push_back(getImpl((*q)->ice_getIdentity())->getObjectInfo()); } return infos; } ObjectEntry::ObjectEntry(ObjectCache& cache, const ObjectInfo& info, const string& application, bool allocatable, const AllocatablePtr& parent) : Allocatable(allocatable, parent), _cache(cache), _info(info), _application(application) { } Ice::ObjectPrx ObjectEntry::getProxy() const { return _info.proxy; } string ObjectEntry::getType() const { return _info.type; } string ObjectEntry::getApplication() const { return _application; } const ObjectInfo& ObjectEntry::getObjectInfo() const { return _info; } bool ObjectEntry::canRemove() { return true; } void ObjectEntry::allocated(const SessionIPtr& session) { // // Add the object allocation to the session. The object will be // released once the session is destroyed. // if(!session->addAllocation(this)) { throw AllocationException("session destroyed"); } TraceLevelsPtr traceLevels = _cache.getTraceLevels(); if(traceLevels && traceLevels->object > 1) { Ice::Trace out(traceLevels->logger, traceLevels->objectCat); const Ice::Identity id = _info.proxy->ice_getIdentity(); out << "object `" << id << "' allocated by `" << session->getUserId() << "' (" << _count << ")"; } } void ObjectEntry::released(const SessionIPtr& session) { // // Remove the object allocation from the session. // session->removeAllocation(this); TraceLevelsPtr traceLevels = _cache.getTraceLevels(); if(traceLevels && traceLevels->object > 1) { Ice::Trace out(traceLevels->logger, traceLevels->objectCat); const Ice::Identity id = _info.proxy->ice_getIdentity(); out << "object `" << id << "' released by `" << session->getUserId() << "' (" << _count << ")"; } } bool ObjectEntry::canTryAllocate() { return _cache.canTryAllocate(this); }