memory_definitions.cpp

Go to the documentation of this file.

#include "memory_definitions.h"
 
#include <limits>
#include <map>
#include <vector>
 
#include <ArmarXCore/core/exceptions/local/ExpressionException.h>
#include <ArmarXCore/core/logging/Logging.h>
 
#include <RobotAPI/libraries/armem/core/container_maps.h>
#include <RobotAPI/libraries/aron/core/data/variant/container/Dict.h>
 
#include "error.h"
 
namespace armarx::armem::server::wm
{
 
    void
    Entity::setMaxHistorySize(long maxSize)
    {
        std::string containerID = (this->id()).str();
        MaxHistorySize::setMaxHistorySize(maxSize, containerID);
        truncate();
    }
 
    auto
    Entity::update(const EntityUpdate& update) -> UpdateResult
    {
        UpdateResult result = EntityBase::update(update);
        result.removedSnapshots = this->truncate();
        return result;
    }
 
    std::vector<EntitySnapshot>
    Entity::truncate()
    {
        std::vector<EntitySnapshot> removedElements;
        if (_maxHistorySize >= 0)
        {
            // Limit removals per call to avoid long blocking
            // _truncateMaxBatchSize == 0 means unlimited
            const size_t maxRemovals = (_truncateMaxBatchSize > 0)
                                           ? _truncateMaxBatchSize
                                           : std::numeric_limits<size_t>::max();
            size_t removed = 0;
 
            while (this->_container.size() > size_t(_maxHistorySize) && removed < maxRemovals)
            {
                removedElements.push_back(std::move(this->_container.begin()->second));
                this->_container.erase(this->_container.begin());
                ++removed;
            }
        }
        else if (_unlimitedHistoryWarningThreshold > 0 &&
                 this->_container.size() > _unlimitedHistoryWarningThreshold)
        {
            // Unlimited history mode - warn if growing too large
            ARMARX_WARNING << deactivateSpam(60)
                           << "Entity '" << this->id().str() << "' has "
                           << this->_container.size()
                           << " snapshots with unlimited history (maxHistorySize=-1). "
                           << "Consider setting a limit to prevent memory exhaustion.";
        }
        return removedElements;
    }
 
    // TODO: add core segment if param is set
    std::vector<Memory::Base::UpdateResult>
    Memory::updateLocking(const Commit& commit)
    {
        // Group updates by core segment, then update each core segment in a batch to only lock it once.
        std::map<std::string, std::vector<const EntityUpdate*>> updatesPerCoreSegment;
        for (const EntityUpdate& update : commit.updates)
        {
            updatesPerCoreSegment[update.entityID.coreSegmentName].push_back(&update);
        }
 
        std::vector<Memory::Base::UpdateResult> result;
        // To throw an exception after the commit if a core segment is missing and the memory should not create new ones
        std::vector<std::string> missingCoreSegmentNames;
        for (const auto& [coreSegmentName, updates] : updatesPerCoreSegment)
        {
            auto it = this->_container.find(coreSegmentName);
            if (it != this->_container.end())
            {
                CoreSegment& coreSegment = it->second;
 
                // Lock the core segment for the whole batch (exclusive lock for writes).
                coreSegment.doLockedExclusive(
                    [&result, &coreSegment, updates = &updates]()
                    {
                        for (const EntityUpdate* update : *updates)
                        {
                            auto r = coreSegment.update(*update);
                            Base::UpdateResult ret{r};
                            ret.memoryUpdateType = UpdateType::UpdatedExisting;
                            result.push_back(ret);
                        }
                    });
            }
            else
            {
                // Perform the other updates first, then throw afterwards.
                missingCoreSegmentNames.push_back(coreSegmentName);
            }
        }
        // Throw an exception if something went wrong.
        if (not missingCoreSegmentNames.empty())
        {
            // Just throw an exception for the first entry. We can extend this exception in the future.
            throw armem::error::MissingEntry::create<CoreSegment>(missingCoreSegmentNames.front(),
                                                                  *this);
        }
        return result;
    }
 
    // TODO: Add core segment if param is set
    Memory::Base::UpdateResult
    Memory::updateLocking(const EntityUpdate& update)
    {
        this->_checkContainerName(update.entityID.memoryName, this->name());
 
        CoreSegment& segment = getCoreSegment(update.entityID.coreSegmentName);
        Base::UpdateResult result;
        segment.doLockedExclusive([&result, &segment, &update]() { result = segment.update(update); });
        result.memoryUpdateType = UpdateType::UpdatedExisting;
        return result;
    }
} // namespace armarx::armem::server::wm