IceManager.cpp

Go to the documentation of this file.

/*
 * This file is part of ArmarX.
 *
 * Copyright (C) 2011-2016, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
 *
 * ArmarX is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation.
 *
 * ArmarX is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 *
 * @package    ArmarXCore::core
 * @author     Nils Adermann (naderman at naderman dot de)
 * @author     Jan Issac (jan dot issac at gmx dot de)
 * @date       2010
 * @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
 *             GNU General Public License
 */
 
#include "IceManager.h"
 
#include <Ice/Current.h> // for Current
#include <Ice/DispatchInterceptor.h> // for DispatchInterceptor, etc
#include <Ice/Initialize.h> // for stringToIdentity
#include <Ice/LocalException.h> // for AlreadyRegisteredException, etc
#include <Ice/Object.h> // for DispatchStatus, Object, etc
#include <Ice/ObjectAdapter.h> // for ObjectAdapterPtr, etc
#include <Ice/ProxyHandle.h> // for ProxyHandle
#include <IceGrid/Admin.h> // for AdminPrx, Admin
#include <IceGrid/Registry.h> // for RegistryPrx
#include <IceStorm/IceStorm.h> // for TopicPrx, TopicManagerPrx, etc
#include <IceUtil/Thread.h> // for ThreadControl
#include <IceUtil/Time.h> // for Time
#include <IceUtil/UUID.h> // for Time
 
#include <ArmarXCore/core/IceGridAdmin.h> // for IceGridAdmin
#include <ArmarXCore/core/exceptions/Exception.h>
#include <ArmarXCore/core/logging/LogSender.h> // for LogSender, flush
#include <ArmarXCore/core/logging/Logging.h> // for ARMARX_INFO, etc
#include <ArmarXCore/interface/core/UserException.h> // for UserException
 
#include "IceGridAdmin.h" // for IceGridAdmin
#include "IceManagerImpl.h"
 
namespace IceGrid
{
    class DeploymentException;
    class ObjectNotRegisteredException;
} // namespace IceGrid
 
namespace armarx
{
    IceManager::IceManager(const Ice::CommunicatorPtr& communicator,
                           std::string name,
                           const std::string topicSuffix) :
        impl(new Impl)
    {
        impl->communicator = communicator;
        impl->name = name;
        impl->topicSuffix = topicSuffix;
        impl->forceShutdown = false;
        setTag("IceManager");
    }
 
    IceManager::~IceManager()
    {
        destroy();
    }
 
    class ExeceptionHandlingInterceptor : public Ice::DispatchInterceptor
    {
    public:
        ExeceptionHandlingInterceptor(const Ice::ObjectPtr& servant) : _servant(servant)
        {
        }
 
        bool
        dispatch(Ice::Request& request) override
        {
            try
            {
                return _servant->ice_dispatch(request);
            }
            catch (...)
            {
 
                ARMARX_WARNING << deactivateSpam(30,
                                                 request.getCurrent().operation +
                                                     request.getCurrent().id.name)
                               << "Calling interface function '" << request.getCurrent().operation
                               << "' of object '" << request.getCurrent().id.name << "' failed:\n"
                               << GetHandledExceptionString();
                throw;
            }
        }
 
        Ice::ObjectPtr _servant;
    };
 
    ObjectHandles
    IceManager::registerObject(const Ice::ObjectPtr& object,
                               const std::string& objectName,
                               const Ice::ObjectAdapterPtr& adapterToAddTo)
    {
        std::unique_lock lock(impl->objectRegistryMutex);
 
        if (isObjectReachable(objectName))
        {
            throw Ice::AlreadyRegisteredException(__FILE__, __LINE__, object->ice_id(), objectName);
        }
 
        ObjectEntryPtr objectEntry = getOrCreateObjectEntry(objectName);
 
        objectEntry->id = Ice::stringToIdentity(objectName);
 
        if (adapterToAddTo)
        {
            objectEntry->adapter = adapterToAddTo;
            objectEntry->ownAdapter = false;
        }
        else
        {
            objectEntry->adapter =
                getCommunicator()->createObjectAdapterWithEndpoints(objectName, "tcp");
            objectEntry->adapter->activate();
            objectEntry->ownAdapter = true;
        }
 
 
        Ice::DispatchInterceptorPtr interceptor = new ExeceptionHandlingInterceptor(object);
        objectEntry->adapter->add(interceptor, objectEntry->id);
        objectEntry->proxy = objectEntry->adapter->createProxy(objectEntry->id);
 
 
        ARMARX_VERBOSE << objectEntry->name << " registered" << flush;
 
        return ObjectHandles(objectEntry->proxy, objectEntry->adapter);
    }
 
    void
    IceManager::removeObject(const std::string& objectName)
    {
        IceGrid::AdminPrx admin = getIceGridSession()->getAdmin();
 
        std::unique_lock lock(impl->objectRegistryMutex);
        {
            auto objectIt = impl->objectRegistry.find(objectName);
 
            if (objectIt != impl->objectRegistry.end())
            {
                try
                {
                    ObjectEntryPtr objectEntry = objectIt->second;
                    Ice::ObjectAdapterPtr adapter = objectIt->second->adapter;
 
                    for (auto& topic : objectEntry->usedTopics)
                    {
                        unsubscribeTopic(objectEntry->proxy, topic);
                    }
 
                    if (adapter)
                    {
                        if (objectEntry->ownAdapter)
                        {
                            adapter->destroy();
                        }
                        else if (!adapter->isDeactivated())
                        {
                            adapter->remove(objectEntry->id);
                        }
                    }
 
 
                    //                objectIt->second->adapter->remove(objectIt->second->id); // deactivate object adapter
                    ARMARX_VERBOSE << "removing object from ice: " << objectName
                                   << " with id: " << objectIt->second->id.name;
                    admin->removeObject(objectIt->second->id);
                }
                catch (IceGrid::ObjectNotRegisteredException& notRegisteredException)
                {
                    // removing an unregistered object
                    //                 //!!!
                    //                ARMARX_ERROR << "*** ARMARX_ERROR: IceManager >> removing "
                    //                          << objectName
                    //                          << " object failed due to ObjectNotRegisteredException"
                    //                          << flush;
                }
                catch (IceGrid::DeploymentException& deploymentException)
                {
                    // cannot remove object due to deployment
                    ARMARX_ERROR << "*** removing " << objectName
                                 << " object failed due to DeploymentException" << flush;
                }
                catch (Ice::ObjectAdapterDeactivatedException& e)
                {
                    ARMARX_INFO << "ObjectAdapterDeactivatedException for "
                                << objectIt->second->id.name;
                }
 
                impl->objectRegistry.erase(objectIt);
            }
        }
    }
 
    void
    IceManager::throwUserException(const std::string& message)
    {
        throw armarx::UserException(message);
    }
 
    bool
    IceManager::removeProxyFromCache(const std::string& name,
                                     const std::string& typeName,
                                     const std::string& endpoints)
    {
        std::string proxyString = name;
 
        if (!endpoints.empty())
        {
            proxyString += std::string(":") + endpoints;
        }
 
        std::string proxyTypedId = proxyString + std::string(":") + typeName;
        return (impl->checkedProxies.erase(proxyTypedId) > 0);
    }
 
    bool
    IceManager::removeProxyFromCache(const Ice::ObjectPrx& proxy)
    {
        if (!proxy)
        {
            return false;
        }
        for (auto& proxyEntry : impl->checkedProxies)
        {
            if (proxyEntry.second == proxy)
            {
                impl->checkedProxies.erase(proxyEntry.first);
                return true;
            }
        }
        return false;
    }
 
    IceStorm::TopicManagerPrx
    IceManager::getTopicManager()
    {
        std::unique_lock lock(impl->topicManagerMutex);
 
        if (!impl->topicManagerProxy)
        {
            impl->topicManagerProxy = GetTopicManager(getCommunicator());
        }
 
        return impl->topicManagerProxy;
    }
 
    IceStorm::TopicManagerPrx
    IceManager::GetTopicManager(Ice::CommunicatorPtr communicator)
    {
        Ice::ObjectPrx obj = communicator->stringToProxy("IceStorm/TopicManager");
 
        return IceStorm::TopicManagerPrx::checkedCast(obj);
    }
 
    void
    IceManager::subscribeTopic(Ice::ObjectPrx subscriberProxy,
                               const std::string& topicName,
                               bool orderedPublishing)
    {
        IceStorm::TopicPrx topic = retrieveTopic(topicName);
 
        IceStorm::QoS qos;
        //    // ensure ordered. If CPU is heavily used, the messages might arrive out of sequence even with one publisher if not enabled
        //    // might be a performance issue. But as long as this is not proved leave ordered in order to prevent unexpected behavior (as publishers would assume blocking calls)
        if (orderedPublishing)
        {
            qos["reliability"] = "ordered";
        }
 
        try
        {
            topic->subscribeAndGetPublisher(
                qos, orderedPublishing ? subscriberProxy : subscriberProxy->ice_oneway());
        }
        catch (IceStorm::AlreadySubscribed& e)
        {
            try
            {
                unsubscribeTopic(subscriberProxy, topicName);
                topic->subscribeAndGetPublisher(
                    qos, orderedPublishing ? subscriberProxy : subscriberProxy->ice_oneway());
            }
            catch (IceStorm::AlreadySubscribed& e)
            {
                ARMARX_INFO << topicName << " already subscribed" << flush;
            }
        }
 
        ARMARX_INFO << "Subscribed to topic " << topicName;
        std::unique_lock lock(impl->topicSubscriptionMutex);
        {
            impl->subscriptions.push_back(std::make_pair(topicName, subscriberProxy));
        }
    }
 
    void
    IceManager::registerAndSubscribeTopic(Ice::ObjectPtr subscriber,
                                          const std::string& topicName,
                                          bool orderedPublishing)
    {
        auto prx = registerObject(subscriber, topicName + "Listener" + IceUtil::generateUUID());
        subscribeTopic(prx.first, topicName, orderedPublishing);
    }
 
    void
    IceManager::unsubscribeTopic(Ice::ObjectPrx subscriberProxy, const std::string& topicName)
    {
        IceStorm::TopicPrx topic = retrieveTopic(topicName);
 
        topic->unsubscribe(subscriberProxy);
        ARMARX_INFO << "Unsubscribed from topic " << topicName;
        std::unique_lock lock(impl->topicSubscriptionMutex);
        {
            std::vector<std::pair<std::string, Ice::ObjectPrx>>::iterator toDelete =
                impl->subscriptions.end();
            std::vector<std::pair<std::string, Ice::ObjectPrx>>::iterator it;
 
            for (it = impl->subscriptions.begin(); it != impl->subscriptions.end(); ++it)
            {
                if (it->first == topicName && it->second == subscriberProxy)
                {
                    toDelete = it;
                }
            }
 
            // we have to check because an old component which is not in our list might still be subscribed
            if (toDelete != impl->subscriptions.end())
            {
                impl->subscriptions.erase(toDelete);
            }
        }
    }
 
    IceStorm::TopicPrx
    IceManager::retrieveTopic(const std::string& name)
    {
        auto topicName = name + getTopicSuffix();
        std::unique_lock lock(impl->topicRetrievalMutex);
 
        IceStorm::TopicPrx& topic = impl->topics[topicName];
 
        while (!isShutdown() && !topic)
        {
            try
            {
                topic = getTopicManager()->retrieve(topicName);
            }
            catch (const IceStorm::NoSuchTopic&)
            {
                try
                {
                    //this also adds to the map
                    topic = getTopicManager()->create(topicName);
                    ARMARX_INFO << "Topic " << topicName << " created " << flush;
                }
                catch (const IceStorm::TopicExists&)
                {
                    // if the topic has been created in the meanwhile (retry via while)
                }
            }
        }
 
        return topic;
    }
 
    Ice::ObjectPrx
    IceManager::implGetCheckedProxy(std::string const& proxyTypedId)
    {
        std::unique_lock lock(impl->proxyCacheMutex);
        return impl->checkedProxies.at(proxyTypedId);
    }
 
    void
    IceManager::implSetCheckedProxy(std::string const& proxyTypedId, const Ice::ObjectPrx& proxy)
    {
        std::unique_lock lock(impl->proxyCacheMutex);
        impl->checkedProxies[proxyTypedId] = proxy;
    }
 
    void
    IceManager::shutdown()
    {
        cleanUp();
 
        getCommunicator()->shutdown();
    }
 
    void
    IceManager::waitForShutdown()
    {
        getCommunicator()->waitForShutdown();
    }
 
    bool
    IceManager::isShutdown()
    {
        return getCommunicator()->isShutdown();
    }
 
    void
    IceManager::destroy()
    {
        if (impl->iceGridAdmin)
        {
            impl->iceGridAdmin->stop();
        }
 
        getCommunicator()->destroy();
    }
 
    std::string
    IceManager::getTopicSuffix() const
    {
        return impl->topicSuffix;
    }
 
    Ice::ObjectPrx
    IceManager::communicator_stringToProxy(const std::string& proxyString)
    {
        return getCommunicator()->stringToProxy(proxyString);
    }
 
    Ice::ObjectPrx
    IceManager::__getTopic(const std::string& topicName, bool useUDP)
    {
        IceStorm::TopicPrx topic = retrieveTopic(topicName);
 
        auto prx = topic->getPublisher();
        if (useUDP)
        {
            prx = prx->ice_datagram();
        }
        else
        {
            prx = prx->ice_oneway();
        }
        return prx;
    }
 
    void
    IceManager::cleanUp()
    {
        ARMARX_DEBUG << " *** CLEAN UP ***" << flush;
 
        // unsubscribe all and remove all objects
        if (getCommunicator())
        {
            IceGrid::AdminPrx admin = getIceGridSession()->getAdmin();
            std::vector<std::pair<std::string, Ice::ObjectPrx>>::iterator it;
 
            {
                /* topicSubscription lock scope */
                std::unique_lock lock(impl->topicSubscriptionMutex);
                {
                    for (it = impl->subscriptions.begin(); it != impl->subscriptions.end(); ++it)
                    {
                        retrieveTopic(it->first)->unsubscribe(it->second);
                    }
 
                    impl->subscriptions.clear();
                }
            }
 
            std::unique_lock lock(impl->objectRegistryMutex);
            {
                auto objListIt = impl->objectRegistry.begin();
 
                for (; objListIt != impl->objectRegistry.end(); ++objListIt)
                {
                    try
                    {
                        objListIt->second->adapter->deactivate();
                        //                    objListIt->second->adapter->remove(objListIt->second->id);  // deactivate object adapter
                        admin->removeObject(objListIt->second->id);
                    }
                    catch (...)
                    {
                    }
                }
            }
 
            if (impl->iceGridAdmin)
            {
                impl->iceGridAdmin->removeObservers();
            }
        }
    }
 
    void
    IceManager::registerObjectDependency(const std::string& registrantName,
                                         const std::string& dependencyObjectName)
    {
        std::unique_lock lock(impl->objectRegistryMutex);
        {
            ObjectEntryPtr objectEntry = getOrCreateObjectEntry(registrantName);
 
            objectEntry->dependencies.push_back(new DependencyObjectEntry(
                dependencyObjectName, getCommunicator()->stringToProxy(dependencyObjectName)));
        }
    }
 
    void
    IceManager::resolveObjectDependencies()
    {
        auto objectIt = impl->objectRegistry.begin();
 
        for (; objectIt != impl->objectRegistry.end(); ++objectIt)
        {
            std::string missingObjects;
            ObjectEntryPtr objectEntry = objectIt->second;
 
            if (!objectEntry->active && objectEntry->proxy)
            {
                objectEntry->dependenciesResolved = true;
                DependencyList::iterator depIt = objectEntry->dependencies.begin();
 
                for (; depIt != objectEntry->dependencies.end(); ++depIt)
                {
                    DependencyObjectEntryPtr dependencyEntry = *depIt;
 
                    if (!dependencyEntry->resolved)
                    {
                        try
                        {
                            dependencyEntry->proxy->ice_timeout(2000)->ice_ping();
 
                            ARMARX_INFO << objectEntry->name << " found " << dependencyEntry->name
                                        << flush;
 
                            dependencyEntry->resolved = true;
 
                            objectEntry->updated = true;
                        }
                        catch (...)
                        {
                            objectEntry->dependenciesResolved = false;
 
                            missingObjects += "\t" + dependencyEntry->name + "\n";
                        }
 
                        IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(10));
                    }
                }
            }
 
            if (objectEntry->updated)
            {
                if (missingObjects.length() > 0)
                {
                    ARMARX_INFO << objectEntry->name << " still waiting for:\n"
                                << missingObjects << flush;
                }
 
                objectEntry->updated = false;
 
                if (objectEntry->dependenciesResolved)
                {
                    ARMARX_INFO << " all " << objectEntry->name << " dependencies resolved"
                                << flush;
                }
            }
        }
    }
 
    void
    IceManager::registerDelayedTopicSubscription(const std::string& registrantName,
                                                 const std::string& topicName)
    {
        std::unique_lock lock(impl->objectRegistryMutex);
 
        ObjectEntryPtr objectEntry = getOrCreateObjectEntry(registrantName);
 
        objectEntry->usedTopics.push_back(topicName);
    }
 
    void
    IceManager::registerDelayedTopicRetrieval(const std::string& registrantName,
                                              const std::string& topicName)
    {
        std::unique_lock lock(impl->objectRegistryMutex);
 
        ObjectEntryPtr objectEntry = getOrCreateObjectEntry(registrantName);
 
        objectEntry->offeredTopics.push_back(topicName);
    }
 
    void
    IceManager::subscribeTopics(Ice::ObjectPrx subscriber,
                                const TopicList& topics,
                                bool orderedPublishing)
    {
        TopicList::const_iterator it = topics.begin();
 
        for (; it != topics.end(); ++it)
        {
            subscribeTopic(subscriber, *it, orderedPublishing);
        }
    }
 
    void
    IceManager::retrieveTopics(const TopicList& topics)
    {
        TopicList::const_iterator it = topics.begin();
 
        for (; it != topics.end(); ++it)
        {
            retrieveTopic(*it);
        }
    }
 
    void
    IceManager::setName(std::string name)
    {
        this->impl->name = name;
    }
 
    const Ice::CommunicatorPtr&
    IceManager::getCommunicator()
    {
        return impl->communicator;
    }
 
    const IceGridAdminPtr&
    IceManager::getIceGridSession()
    {
        std::unique_lock lock(impl->iceGridAdminMutex);
 
        if (!impl->iceGridAdmin)
        {
            impl->iceGridAdmin = IceGridAdmin::Create(getCommunicator(), impl->name);
        }
 
        return impl->iceGridAdmin;
    }
 
    IceGrid::RegistryPrx
    IceManager::getIceGridRegistry()
    {
        return getIceGridSession()->registry();
    }
 
    IceManager::ObjectEntryPtr
    IceManager::getOrCreateObjectEntry(const std::string& objectName)
    {
        auto objIt = impl->objectRegistry.find(objectName);
 
        if (objIt == impl->objectRegistry.end() || objIt->second->name.empty())
        {
            ObjectEntryPtr objectEntry = new ObjectEntry();
            objectEntry->name = objectName;
 
            impl->objectRegistry[objectName] = objectEntry;
 
            return objectEntry;
        }
 
        return objIt->second;
    }
 
    bool
    IceManager::isObjectReachable(std::string objectName)
    {
        try
        {
            Ice::ObjectPrx prx = getProxy<Ice::ObjectPrx>(objectName);
            prx->ice_timeout(500)->ice_ping();
            return true; // if ping'able, object is already registered
            //        throw Ice::AlreadyRegisteredException(__FILE__,__LINE__, object->ice_id(), object->getName());
        }
        catch (...)
        {
 
            return false;
        }
    }
} // namespace armarx