RemoteReferenceCount.cpp

Go to the documentation of this file.

/*
 * This file is part of ArmarX.
 *
 * Copyright (C) 2011-2017, 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    ArmarX
 * @author     Raphael Grimm( raphael dor grimm at kit dot edu)
 * @date       2017
 * @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
 *             GNU General Public License
 */
#include <thread>
#include <queue>
#include <chrono>
 
#include <Ice/Initialize.h>             // for stringToIdentity
#include <Ice/ObjectAdapter.h>
#include <IceUtil/UUID.h>
 
#include "RemoteReferenceCount.h"
#include "system/FactoryCollectionBase.h"
#include "exceptions/local/ExpressionException.h"
#include "ArmarXManager.h"
 
 
namespace armarx
{
 
    class RemoteReferenceCounter;
    using RemoteReferenceCounterPtr = IceUtil::Handle<RemoteReferenceCounter>;
 
    class HeartbeatMonitor
    {
    public:
        static void Add(RemoteReferenceCounter* ptr)
        {
            Instance().add(ptr);
        }
        static void Remove(RemoteReferenceCounter* ptr)
        {
            Instance().remove(ptr);
        }
    private:
        void add(RemoteReferenceCounter* ptr);
        void remove(RemoteReferenceCounter* ptr);
        static HeartbeatMonitor& Instance();
        ~HeartbeatMonitor();
 
        void heartbeatTask();
 
        std::mutex mtx;
        std::thread t;
        std::atomic_bool shutdown {false};
        HeartbeatMonitor() = default;
        using PqEntry = std::pair<IceUtil::Time, RemoteReferenceCounter*>;
        struct PqEntryCompare
        {
            bool operator()(const PqEntry& lhs, const PqEntry& rhs) const
            {
                return lhs.first > rhs.first;
            }
        };
        std::priority_queue<PqEntry,
                            std::vector<PqEntry>,
                            PqEntryCompare> entries;
        std::set<void*> entriesToDelete;
        static const IceUtil::Time maximalSleepTime;
    };
    const IceUtil::Time HeartbeatMonitor::maximalSleepTime = IceUtil::Time::milliSeconds(100);
 
    class RemoteReferenceCounter : virtual public RemoteReferenceCounterBase
    {
    public:
        RemoteReferenceCounter() = default;
        RemoteReferenceCounter(const RemoteReferenceCountControlBlockInterfacePrx& prx, const std::string& id, Ice::Long heartBeatMs)
            : RemoteReferenceCounterBase(prx, id, heartBeatMs)
        {
            ice_postUnmarshal();
        }
        RemoteReferenceCounter(const RemoteReferenceCounter& other) : RemoteReferenceCounter(other.block, other.id, other.heartBeatMs) {}
 
        ~RemoteReferenceCounter() override
        {
            if (block)
            {
                block->removeCounter(counterId);
            }
            HeartbeatMonitor::Remove(this);
        }
 
        RemoteReferenceCounterBasePtr copy(const Ice::Current& = Ice::emptyCurrent) const final override
        {
            return new RemoteReferenceCounter {*this};
        }
        std::string getId(const Ice::Current& = Ice::emptyCurrent) const final override
        {
            return id;
        }
        void ice_postUnmarshal() override
        {
            block->addCounter(counterId);
            HeartbeatMonitor::Add(this);
        }
        IceUtil::Time heartbeat(IceUtil::Time now)
        {
            block->heartbeat(counterId);
            return now + IceUtil::Time::milliSeconds(heartBeatMs);
        }
    private:
        const std::string counterId
        {
            IceUtil::generateUUID()
        };
    };
 
    class SimpleRemoteReferenceCounter : virtual public SimpleRemoteReferenceCounterBase
    {
    public:
        SimpleRemoteReferenceCounter() = default;
        SimpleRemoteReferenceCounter(const SimpleRemoteReferenceCountControlBlockInterfacePrx& prx, const std::string& id) : SimpleRemoteReferenceCounterBase(prx, id)
        {
            ice_postUnmarshal();
        }
        SimpleRemoteReferenceCounter(const SimpleRemoteReferenceCounter& other) : SimpleRemoteReferenceCounter(other.block, other.id) {}
 
        ~SimpleRemoteReferenceCounter() override
        {
            if (block)
            {
                block->removeCounter(counterId);
            }
        }
 
        SimpleRemoteReferenceCounterBasePtr copy(const Ice::Current& = Ice::emptyCurrent) const final override
        {
            return new SimpleRemoteReferenceCounter {*this};
        }
        std::string getId(const Ice::Current& = Ice::emptyCurrent) const final override
        {
            return id;
        }
        void ice_postUnmarshal() override
        {
            block->addCounter(counterId);
        }
    private:
        const std::string counterId
        {
            IceUtil::generateUUID()
        };
    };
 
    HeartbeatMonitor& HeartbeatMonitor::Instance()
    {
        static HeartbeatMonitor monitor;
        return monitor;
    }
    void HeartbeatMonitor::add(RemoteReferenceCounter* ptr)
    {
        std::lock_guard<std::mutex> guard {mtx};
 
        if (!t.joinable())
        {
            t = std::thread {[this]{heartbeatTask();}};
        }
        entries.emplace(ptr->heartbeat(IceUtil::Time::now()), ptr);
    }
    void HeartbeatMonitor::remove(RemoteReferenceCounter* ptr)
    {
        std::lock_guard<std::mutex> guard {mtx};
        entriesToDelete.emplace(ptr);
    }
    HeartbeatMonitor::~HeartbeatMonitor()
    {
        std::lock_guard<std::mutex> guard {mtx};
        shutdown = true;
        if (t.joinable())
        {
            t.join();
        }
    }
    void HeartbeatMonitor::heartbeatTask()
    {
        while (!shutdown)
        {
            IceUtil::Time now;
            {
                // this stops all dtors of RemoteReferenceCounter
                // -> if the ptr is not already in entriesToDelete,
                // the object can't be deleted
                std::lock_guard<std::mutex> guard {mtx};
                now = IceUtil::Time::now();
                while (!shutdown && !entries.empty() && entries.top().first <= now)
                {
                    RemoteReferenceCounter* ptr;
                    std::tie(std::ignore, ptr) = entries.top();
                    entries.pop();
                    if (entriesToDelete.count(ptr))
                    {
                        continue;
                    }
                    try
                    {
                        // the remote may be dead!
                        entries.emplace(ptr->heartbeat(now), ptr);
                    }
                    catch (...)
                    {}
                }
            }
            IceUtil::Time msToSleep = maximalSleepTime;
            if (!entries.empty())
            {
                msToSleep = std::min(maximalSleepTime, entries.top().first - now);
            }
            std::this_thread::sleep_for(std::chrono::milliseconds {msToSleep.toMilliSeconds()});
        }
    }
 
    void RemoteReferenceCountControlBlockManager::stop()
    {
        if (!shutdown.exchange(true))
        {
            if (thread.joinable())
            {
                thread.join();
            }
        }
    }
 
    void RemoteReferenceCountControlBlockManager::manage()
    {
        while (!shutdown)
        {
            IceUtil::Time now;
            {
                std::lock_guard<std::mutex> guard {stateMutex};
                now = IceUtil::Time::now();
                //sweep pendingForActivation
                {
                    auto newPendingForActivation = pendingForActivation;
                    for (const auto& pending : pendingForActivation)
                    {
                        if (pending->isCountingActivated())
                        {
                            rrccbs.emplace(pending->nextCheckTimePoint(), pending);
                            newPendingForActivation.erase(pending);
                        }
                    }
                    pendingForActivation = std::move(newPendingForActivation);
                }
                //sweep rrccbs
                while (!shutdown && !rrccbs.empty() && rrccbs.top().first <= now)
                {
                    detail::RemoteReferenceCountControlBlockManagementInterfacePtr ptr;
                    std::tie(std::ignore, ptr) = rrccbs.top();
                    rrccbs.pop();
                    const auto newT = ptr->nextCheckTimePoint();
                    if (newT < now)
                    {
                        try
                        {
                            ptr->countReachedZero();
                        }
                        catch (...)
                        {
                            ARMARX_WARNING << " function countReachedZero threw an exception";
                        }
                        continue;
                    }
                    rrccbs.emplace(newT, std::move(ptr));
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds {period.toMilliSeconds()});
        }
        //exiting -> call all reach 0 + warn
        ARMARX_CHECK_EXPRESSION(shutdown);
        std::lock_guard<std::mutex> guard {stateMutex};
        auto processRemovedEntry = [](const detail::RemoteReferenceCountControlBlockManagementInterfacePtr & ptr)
        {
            ARMARX_CHECK_EXPRESSION(ptr);
            ARMARX_CHECK_EXPRESSION(ptr->getProxy());
            ARMARX_WARNING << "REMOVING RemoteReferenceCount on shutdown!:\n"
                           << "    counting activated : " << ptr->isCountingActivated() << "\n"
                           << "    identity (cat/name): " << ptr->getProxy()->ice_getIdentity().category << " / " << ptr->getProxy()->ice_getIdentity().name << "\n"
                           << "    type id            : " << ptr->getProxy()->ice_id();
            try
            {
                ptr->countReachedZero();
            }
            catch (...)
            {}
        };
 
        while (!rrccbs.empty())
        {
            detail::RemoteReferenceCountControlBlockManagementInterfacePtr ptr;
            std::tie(std::ignore, ptr) = rrccbs.top();
            rrccbs.pop();
            processRemovedEntry(ptr);
        }
        for (const auto& ptr : pendingForActivation)
        {
            processRemovedEntry(ptr);
        }
        pendingForActivation.clear();
    }
 
    void RemoteReferenceCountControlBlockManager::add(detail::RemoteReferenceCountControlBlockManagementInterfacePtr ptr)
    {
        ARMARX_CHECK_EXPRESSION(ptr);
        ARMARX_CHECK_EXPRESSION(ptr->getProxy());
        std::lock_guard<std::mutex> guard {stateMutex};
        ARMARX_CHECK_EXPRESSION(!shutdown);
        if (!thread.joinable())
        {
            thread = std::thread {[this]{manage();}};
        }
        pendingForActivation.emplace(ptr);
    }
}
 
namespace armarx::detail
{
    //RemoteReferenceCountControlBlockManagementInterface
    RemoteReferenceCountControlBlockManagementInterface::RemoteReferenceCountControlBlockManagementInterface(const ArmarXManagerPtr& manager, const std::string& id)
        : armarXManager {manager}, id {id}
    {
        ARMARX_CHECK_EXPRESSION(armarXManager);
        const auto iceId = Ice::stringToIdentity("RRCB_" + IceUtil::generateUUID());
        __setNoDelete(true);
        selfProxy = armarXManager->getAdapter()->add(this, iceId);
        __setNoDelete(false);
    }
 
    RemoteReferenceCountControlBlockManagementInterface::~RemoteReferenceCountControlBlockManagementInterface()
    {
        if (!armarXManager)
        {
            ARMARX_FATAL << "armarXManager NULL\n(line " << __LINE__ << " in " << __FUNCTION__ << " in " << __FILE__ ")";
            std::terminate(); //throwing an exception calls terminate anyways (the dtor is noexcept) -> call terminate directly";
        }
        if (!selfProxy)
        {
            ARMARX_FATAL << "selfProxy NULL\n(line " << __LINE__ << " in " << __FUNCTION__ << " in " << __FILE__ ")";
            std::terminate(); //throwing an exception calls terminate anyways (the dtor is noexcept) -> call terminate directly";
        }
        try
        {
            //removing may fail if the adapter was deactivated in a different thread
            armarXManager->getAdapter()->remove(selfProxy->ice_getIdentity());
        }
        catch (Ice::ObjectAdapterDeactivatedException&) {}
    }
 
    void RemoteReferenceCountControlBlockManagementInterface::countReachedZero()
    {
        ARMARX_CHECK_EXPRESSION(isCountingActivated());
        ARMARX_CHECK_EXPRESSION(!countReachedZero_.exchange(true));
        onCountReachedZero();
    }
 
    //AbstractRemoteReferenceCountControlBlock
    void AbstractRemoteReferenceCountControlBlock::addCounter(const std::string& counterId, const Ice::Current&)
    {
        std::lock_guard<std::mutex> guard {mtx};
        if (counterIds.count(counterId))
        {
            ARMARX_WARNING << "an already existing counter id was added! " << VAROUT(counterId);
        }
        else
        {
            ARMARX_DEBUG << "adding counter id '" << counterId << "'. "
                         << "New number of counters: " << counterIds.size() + 1;
        }
        counterIds[counterId] = IceUtil::Time::now();
    }
    void AbstractRemoteReferenceCountControlBlock::heartbeat(const std::string& counterId, const Ice::Current&)
    {
        std::lock_guard<std::mutex> guard {mtx};
        if (!counterIds.count(counterId))
        {
            ARMARX_WARNING << "heartbeat of nonexistent counter '" << counterId << "'. "
                           << "The counter was added.";
        }
        counterIds[counterId] = IceUtil::Time::now();
    }
    void AbstractRemoteReferenceCountControlBlock::removeCounter(const std::string& counterId, const Ice::Current&)
    {
        std::lock_guard<std::mutex> guard {mtx};
        if (counterIds.count(counterId))
        {
            counterIds.erase(counterId);
            ARMARX_DEBUG << "deleteing counter id '" << counterId << "'. "
                         << "Remaining number of counters: " << counterIds.size();
        }
        else
        {
            ARMARX_WARNING << "an non-existent counter id was deleted! " << VAROUT(counterId);
        }
        if (counterIds.empty())
        {
            lastTimeReachedZero = IceUtil::Time::now();
        }
    }
 
    AbstractRemoteReferenceCountControlBlock::AbstractRemoteReferenceCountControlBlock(const ArmarXManagerPtr& manager, const std::string& id,
            IceUtil::Time deletionDelay, IceUtil::Time orphantDeletionDelay, long heartBeatMs)
        : RemoteReferenceCountControlBlockManagementInterface(manager, id),
          deletionDelay {deletionDelay}, orphantDeletionDelay {orphantDeletionDelay}, heartBeatMs {heartBeatMs}
    {
        ARMARX_CHECK_GREATER(heartBeatMs, 0);
        ARMARX_CHECK_GREATER(deletionDelay.toMicroSeconds(), 0);
        ARMARX_CHECK_GREATER(orphantDeletionDelay.toMicroSeconds(), 0);
    }
    IceUtil::Time AbstractRemoteReferenceCountControlBlock::nextCheckTimePoint()
    {
        std::lock_guard<std::mutex> guard {mtx};
        if (counterIds.empty())
        {
            return lastTimeReachedZero + deletionDelay;
        }
        else
        {
            IceUtil::Time max;
            for (const auto& entry : counterIds)
            {
                max = std::max(max, entry.second);
            }
            return max + orphantDeletionDelay;
        }
    }
    RemoteReferenceCounterBasePtr AbstractRemoteReferenceCountControlBlock::getReferenceCounter()
    {
        ARMARX_DEBUG << "creating RemoteReferenceCounterBasePtr";
        auto proxy = RemoteReferenceCountControlBlockInterfacePrx::checkedCast(selfProxy);
        ARMARX_CHECK_NOT_NULL(proxy);
        return new RemoteReferenceCounter {proxy, id, heartBeatMs};
    }
 
    //AbstractSimpleRemoteReferenceCountControlBlock
    void AbstractSimpleRemoteReferenceCountControlBlock::addCounter(const std::string& counterId, const Ice::Current&)
    {
        std::lock_guard<std::mutex> guard {mtx};
        if (counterIds.count(counterId))
        {
            ARMARX_WARNING << "an already existing counter id was added! " << VAROUT(counterId);
        }
        else
        {
            counterIds.emplace(counterId);
            ARMARX_DEBUG << "adding counter id '" << counterId << "'. "
                         << "New number of counters: " << counterIds.size();
        }
    }
    void AbstractSimpleRemoteReferenceCountControlBlock::removeCounter(const std::string& counterId, const Ice::Current&)
    {
        std::lock_guard<std::mutex> guard {mtx};
        if (counterIds.count(counterId))
        {
            counterIds.erase(counterId);
            ARMARX_DEBUG << "deleteing counter id '" << counterId << "'. "
                         << "Remaining number of counters: " << counterIds.size();
        }
        else
        {
            ARMARX_WARNING << "an non-existent counter id was deleted! " << VAROUT(counterId);
        }
        if (counterIds.empty())
        {
            lastTimeReachedZero = IceUtil::Time::now();
        }
    }
 
    AbstractSimpleRemoteReferenceCountControlBlock::AbstractSimpleRemoteReferenceCountControlBlock(const ArmarXManagerPtr& manager, const std::string& id, IceUtil::Time deletionDelay)
        : RemoteReferenceCountControlBlockManagementInterface(manager, id), deletionDelay {deletionDelay}
    {
        ARMARX_CHECK_GREATER(deletionDelay.toMicroSeconds(), 0);
    }
 
    SimpleRemoteReferenceCounterBasePtr AbstractSimpleRemoteReferenceCountControlBlock::getReferenceCounter()
    {
        ARMARX_DEBUG << "creating SimpleRemoteReferenceCounterBasePtr";
        auto proxy = SimpleRemoteReferenceCountControlBlockInterfacePrx::checkedCast(selfProxy);
        ARMARX_CHECK_NOT_NULL(proxy);
        return new SimpleRemoteReferenceCounter {proxy, id};
    }
    IceUtil::Time AbstractSimpleRemoteReferenceCountControlBlock::nextCheckTimePoint()
    {
        std::lock_guard<std::mutex> guard {mtx};
        if (counterIds.empty())
        {
            return lastTimeReachedZero + deletionDelay;
        }
        return IceUtil::Time::now() + deletionDelay;
    }
}
 
namespace armarx::ObjectFactories
{
    class RemoteReferenceCounterObjectFactories : public FactoryCollectionBase
    {
    public:
        ObjectFactoryMap getFactories() override
        {
            ObjectFactoryMap map;
            add<armarx::RemoteReferenceCounterBase, armarx::RemoteReferenceCounter>(map);
            add<armarx::SimpleRemoteReferenceCounterBase, armarx::SimpleRemoteReferenceCounter>(map);
            return map;
        }
    };
    const FactoryCollectionBaseCleanUp RemoteReferenceCounterObjectFactoriesVar =
        FactoryCollectionBase::addToPreregistration(new RemoteReferenceCounterObjectFactories());
}
 
namespace armarx
{
    const Ice::Long RemoteReferenceCountControlBlockManager::DefaultDeletionDelayMs = 3000;
    const Ice::Long RemoteReferenceCountControlBlockManager::DefaultOrphantDeletionDelayMs = 10000;
}