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 "RemoteReferenceCount.h"
 
#include <chrono>
#include <queue>
#include <thread>
 
#include <Ice/Initialize.h> // for stringToIdentity
#include <Ice/ObjectAdapter.h>
#include <IceUtil/UUID.h>
 
#include "ArmarXManager.h"
#include "exceptions/local/ExpressionException.h"
#include "system/FactoryCollectionBase.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
 
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::detail
 
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::ObjectFactories
 
namespace armarx
{
    const Ice::Long RemoteReferenceCountControlBlockManager::DefaultDeletionDelayMs = 3000;
    const Ice::Long RemoteReferenceCountControlBlockManager::DefaultOrphantDeletionDelayMs = 10000;
} // namespace armarx