SegmentUtilImplementations.cpp

Go to the documentation of this file.

/*
* This file is part of ArmarX.
*
* 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    ArmarX
* @author     Mirko Waechter( mirko.waechter at kit dot edu)
* @date       2016
* @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
*             GNU General Public License
*/
#include "SegmentUtilImplementations.h"
 
#include <Ice/Connection.h>
#include <Ice/ObjectAdapter.h>
#include <IceUtil/UUID.h>
 
#include <ArmarXCore/core/ArmarXManager.h>
#include <ArmarXCore/core/exceptions/local/ExpressionException.h>
#include <ArmarXCore/core/time/TimeUtil.h>
#include <ArmarXCore/observers/variant/Variant.h>
#include <ArmarXCore/util/json/JSONObject.h>
 
#include <MemoryX/core/entity/Entity.h>
#include <MemoryX/core/entity/EntityRef.h>
 
namespace memoryx
{
 
    SegmentUtilImplementations::SegmentUtilImplementations()
    {
    }
 
    auto
    SegmentUtilImplementations::getReadLock(const Ice::Current& c) const -> ScopedSharedLockPtr
    {
        std::unique_lock lock(tokenMutex);
 
        if (!lockToken.empty() && c.ctx.count("token") > 0 && c.ctx.at("token") == lockToken)
        {
            return ScopedSharedLockPtr();
        }
        return ScopedSharedLockPtr(new ScopedSharedLock(segmentMutex));
    }
 
    auto
    SegmentUtilImplementations::getWriteLock(const Ice::Current& c) const -> ScopedUniqueLockPtr
    {
        std::unique_lock lock(tokenMutex);
 
        if (!lockToken.empty() && c.ctx.count("token") > 0 && c.ctx.at("token") == lockToken)
        {
            return ScopedUniqueLockPtr();
        }
        return ScopedUniqueLockPtr(new ScopedUniqueLock(segmentMutex));
    }
 
    std::string
    SegmentUtilImplementations::getJSONEntityById(const std::string& id, const Ice::Current&) const
    {
        EntityBasePtr entity = getEntityById(id);
        if (!entity)
        {
            throw EntityNotFoundException();
        }
        armarx::JSONObjectPtr json = new armarx::JSONObject();
        json->setIceObject("entity", entity);
        return json->asString(true);
    }
 
    EntityBaseList
    SegmentUtilImplementations::getEntityWithChildrenById(const std::string& id,
                                                          bool includeMetaEntities,
                                                          const Ice::Current& c) const
    {
        auto comp = [](const EntityBasePtr& e1, const EntityBasePtr& e2)
        { return e1->getId() < e2->getId(); };
 
        std::set<EntityBasePtr, decltype(comp)> result(comp);
        std::set<EntityBasePtr, decltype(comp)> searchList(comp);
        EntityPtr root = EntityPtr::dynamicCast(getEntityById(id, c));
        searchList.insert(root);
        result.insert(root);
        auto all = getAllEntities(c);
        std::string attribName = "parentEntityRefs";
 
        while (!searchList.empty())
        {
            EntityPtr curParent = EntityPtr::dynamicCast(*searchList.begin());
            auto curParentRef = getEntityRefById(curParent->getId(), c);
            searchList.erase(searchList.begin());
            for (EntityBasePtr& entity : all)
            {
                if (!entity->hasAttribute(attribName, c))
                {
                    continue;
                }
                auto attr = entity->getAttribute(attribName, c);
                for (int i = 0; i < attr->size(); ++i)
                {
                    memoryx::EntityRefPtr ref =
                        armarx::VariantPtr::dynamicCast(attr->getValueAt(i, c))->get<EntityRef>();
                    if (ref->equals(curParentRef))
                    {
                        ARMARX_INFO_S << ref->entityName << " is parent  of " << entity->getName();
                        if (result.count(entity) ==
                            0) // if already in result list, this item was already in search list
                        {
                            searchList.insert(entity);
                        }
                        result.insert(entity);
                    }
                }
            }
        }
 
        if (!includeMetaEntities)
        {
            EntityBaseList filteredResult;
            std::copy_if(result.begin(),
                         result.end(),
                         std::back_inserter(filteredResult),
                         [](const EntityBasePtr& entity) { return !entity->isMetaEntity(); });
            return filteredResult;
        }
        return EntityBaseList(result.begin(), result.end());
    }
 
    EntityBaseList
    SegmentUtilImplementations::getEntityWithChildrenByName(const std::string& entityName,
                                                            bool includeMetaEntities,
                                                            const Ice::Current& c) const
    {
        auto entity = getEntityByName(entityName);
        if (entity)
        {
            return getEntityWithChildrenById(entity->getId(), includeMetaEntities);
        }
        else
        {
            return EntityBaseList();
        }
    }
 
    SegmentLockBasePtr
    SegmentUtilImplementations::lockSegment(const Ice::Current& c)
    {
        writeLock.reset(new ScopedUniqueLock(segmentMutex));
        SegmentLockBasePtr lock = new SegmentLock(!c.con);
        {
            std::unique_lock mutexLock(tokenMutex);
            lockToken = lock->token = IceUtil::generateUUID();
        }
        ARMARX_CHECK_EXPRESSION(c.adapter) << "Ice::Current must not be empty";
        lock->segment =
            EntityMemorySegmentInterfacePrx::uncheckedCast(c.adapter->createProxy(c.id));
        if (keepAliveCheckTask)
        {
            keepAliveCheckTask->stop();
        }
        keepAliveCheckTask = new armarx::PeriodicTask<SegmentUtilImplementations>(
            this,
            &SegmentUtilImplementations::keepAliveCheck,
            lock->keepAliveIntervalMs * 2,
            false,
            "KeepAliveChecker",
            false);
        {
            std::unique_lock lock(keepAliveTimestampMutex);
            keepAliveTimestamp = armarx::TimeUtil::GetTime();
        }
        keepAliveCheckTask->start();
 
        return lock;
    }
 
    bool
    SegmentUtilImplementations::keepLockAlive(const std::string& token, const Ice::Current&)
    {
        {
            std::unique_lock lock(tokenMutex);
            if (token != lockToken)
            {
                throw InvalidLockTokenException();
            }
        }
 
        if (writeLock)
        {
            std::unique_lock lock(keepAliveTimestampMutex);
            keepAliveTimestamp = armarx::TimeUtil::GetTime();
            return true;
        }
 
        return false;
    }
 
    bool
    SegmentUtilImplementations::unlockSegment(const SegmentLockBasePtr& lock, const Ice::Current& c)
    {
        std::unique_lock mutexLock(tokenMutex);
 
        if (lock->token != lockToken)
        {
            throw InvalidLockTokenException();
        }
        if (!writeLock)
        {
            return false;
        }
        writeLock.reset();
        lockToken.clear();
        return true;
    }
 
    bool
    SegmentUtilImplementations::unlockSegmentWithToken(const std::string& token,
                                                       const Ice::Current& c)
    {
        std::unique_lock mutexLock(tokenMutex);
 
        if (token != lockToken)
        {
            throw InvalidLockTokenException();
        }
        if (!writeLock)
        {
            return false;
        }
        writeLock.reset();
        lockToken.clear();
        return true;
    }
 
    void
    SegmentUtilImplementations::keepAliveCheck()
    {
        std::unique_lock lock(keepAliveTimestampMutex);
 
        auto age = (IceUtil::Time::now() - keepAliveTimestamp);
        if (age.toMilliSeconds() > keepAliveCheckTask->getInterval() && writeLock)
        {
            writeLock.reset();
 
            {
                std::unique_lock mutexLock(tokenMutex);
                lockToken.clear();
            }
 
            ARMARX_WARNING_S << "Segment lock of segment '" << getSegmentName()
                             << "' did not receive keep-alive in time ("
                             << age.toMilliSecondsDouble()
                             << " ms, interval:  " << keepAliveCheckTask->getInterval()
                             << " ) - unlocking segment";
        }
    }
 
    SegmentLock::~SegmentLock()
    {
        if (keepAliveTask)
        {
            unlock();
            keepAliveTask->stop();
        }
    }
 
    SegmentLock::SegmentLock(bool start_watchdog)
    {
        if (start_watchdog)
        {
            startWatchdog();
        }
    }
 
    void
    SegmentLock::unlock(const Ice::Current& c)
    {
        try
        {
            if (keepAliveTask)
            {
                keepAliveTask->stop();
            }
 
            segment->unlockSegmentWithToken(this->token);
        }
        catch (...)
        {
        }
    }
 
    void
    SegmentLock::startWatchdog()
    {
        keepAliveTask = new armarx::PeriodicTask<SegmentLock>(
            this, &SegmentLock::keepAlive, keepAliveIntervalMs, false, "KeepAliveTask", false);
        keepAliveTask->start();
    }
 
    void
    SegmentLock::ice_postUnmarshal()
    {
        startWatchdog();
    }
 
    void
    SegmentLock::keepAlive()
    {
        try
        {
            segment->keepLockAlive(this->token);
        }
        catch (...)
        {
        }
    }
 
} // namespace memoryx