Memory.cpp

Go to the documentation of this file.

#include "Memory.h"
 
#include <memory>
 
#include <ArmarXCore/core/logging/Logging.h>
#include <ArmarXCore/core/time/TimeUtil.h>
 
#include "RobotAPI/libraries/armem/core/MemoryID.h"
#include "RobotAPI/libraries/armem/server/ltm/persistence/MemoryPersistenceStrategy.h"
#include "RobotAPI/libraries/armem/server/ltm/persistence/RedundantPersistenceStrategy.h"
#include "RobotAPI/libraries/armem/server/ltm/persistence/disk/DiskPersistence.h"
#include "RobotAPI/libraries/armem/server/ltm/persistence/rest/RestPersistence.h"
#include <RobotAPI/libraries/armem/server/wm/memory_definitions.h>
 
namespace armarx::armem::server::ltm
{
    void
    Memory::_configure(const nlohmann::json& json)
    {
        ARMARX_TRACE;
 
        BufferedBase::configureMixin(json);
        CachedBase::configureMixin(json);
 
        std::string exportPath = p.export_path;
 
        std::string persistenceStrategiesStr = p.persistenceStrategies;
 
        ARMARX_INFO << "Persistence strategies=" << persistenceStrategiesStr;
 
        std::vector<std::string> persistenceStrategies = split(persistenceStrategiesStr, ',');
 
        ARMARX_INFO << "Found " << persistenceStrategies.size() << " persistence strategies";
 
 
        std::string identifiersStr = p.persistenceStrategyIdentifier;
        std::vector<std::string> identifiers = split(identifiersStr, ',');
        std::string identifier = "defaultStrategy";
        ARMARX_INFO << "Persistence identifiers=" << identifiersStr;
        ARMARX_INFO << "Found " << identifiers.size() << " persistence strategies identifiers";
 
 
        std::shared_ptr<armem::server::ltm::persistence::RedundantPersistenceStrategy>
            redundantPersistence =
                std::make_shared<armem::server::ltm::persistence::RedundantPersistenceStrategy>();
        redundantPersistence->setExportName(getExportName());
 
        for (size_t i = 0; i < persistenceStrategies.size(); i++)
        {
            std::string s = persistenceStrategies.at(i);
 
            if (i < identifiers.size())
            {
                identifier = identifiers.at(i);
            }
 
            if (s == "disk")
            {
                ARMARX_IMPORTANT << "Using persistence strategy=" << s;
                ARMARX_INFO << "MemoryID=" << getMemoryID().str();
                ARMARX_INFO << "Identifier=" << identifier;
                ARMARX_INFO << "Export name=" << getExportName();
                ARMARX_INFO << "Export path=" << exportPath;
                ARMARX_INFO << "Min available disk space=" << p.minDiskSpace;
 
                std::shared_ptr<armem::server::ltm::persistence::DiskPersistence> diskPersistence =
                    std::make_shared<armem::server::ltm::persistence::DiskPersistence>(
                        std::filesystem::path(exportPath));
                diskPersistence->setIdentifier(identifier);
                diskPersistence->setExportName(getExportName());
                diskPersistence->setMinAvailableDiskSpace(p.minDiskSpace);
 
                // Configure compression if enabled
                if (p.compressionEnabled)
                {
                    diskPersistence->setCompressionLevel(p.compressionLevel);
                    diskPersistence->setCompressionMinBytes(
                        static_cast<std::size_t>(p.compressionMinBytes));
                    diskPersistence->setCompressionEnabled(true);
                    ARMARX_INFO << "LTM export compression enabled: gzip level "
                                << p.compressionLevel << " for payloads >= "
                                << p.compressionMinBytes << " bytes";
                }
 
                // Configure batch writing if enabled
                if (p.batchWriteEnabled)
                {
                    diskPersistence->setBatchSizeThreshold(static_cast<size_t>(p.batchSizeThreshold));
                    diskPersistence->setBatchTimeThresholdMs(static_cast<size_t>(p.batchTimeThresholdMs));
                    diskPersistence->setBatchWriteEnabled(true);
                    ARMARX_INFO << "Batch write mode enabled: threshold=" << p.batchSizeThreshold
                               << " items or " << p.batchTimeThresholdMs << "ms";
                }
 
                redundantPersistence->addStrategy(diskPersistence);
            }
            else if (s == "rest")
            {
                ARMARX_IMPORTANT << "Using persistence strategy=" << s;
                ARMARX_WARNING << "Persistence strategy=" << s << " currently deactivated";
                /** 
                ARMARX_INFO << "MemoryID=" << getMemoryID().str();
                ARMARX_INFO << "Identifier=" << identifier;
                ARMARX_INFO << "Export name=" << getExportName();
                ARMARX_INFO << "Host=" << p.restHost;
                ARMARX_INFO << "Port=" << p.restPort;
                ARMARX_INFO << "DisableIfNotAvailable=" << p.restDisableIfNotAvailable;
 
                std::shared_ptr<armem::server::ltm::persistence::RestPersistence> restPersistence =
                    std::make_shared<armem::server::ltm::persistence::RestPersistence>(
                        identifier,
                        getExportName(),
                        p.restHost,
                        p.restPort,
                        p.restDisableIfNotAvailable);
 
 
                redundantPersistence->addStrategy(restPersistence);
                */
            }
            else if (s == "mongodb")
            {
                ARMARX_IMPORTANT << "Using persistence strategy=" << s;
                ARMARX_WARNING << "Persistence strategy=" << s << " not implemented";
            }
            else
            {
                ARMARX_WARNING << "Persistence strategy=" << s << " unknown";
            }
        }
 
        setPersistenceStrategy(redundantPersistence);
    }
 
    Memory::Memory() : Memory("MemoryExport", "Test")
    {
    }
 
    Memory::Memory(const std::string& exportName, const std::string& memoryName) :
        MemoryBase(exportName, MemoryID(memoryName, "")),
        BufferedBase(MemoryID(memoryName, "")),
        CachedBase(MemoryID(memoryName, "")),
        persistenceStrategy_(std::make_shared<persistence::RedundantPersistenceStrategy>())
    {
    }
 
    Memory::Memory(
        const std::string& exportName,
        const std::string& memoryName,
        const std::shared_ptr<persistence::RedundantPersistenceStrategy>& persistenceStrategy) :
        MemoryBase(exportName, MemoryID(memoryName, "")),
        BufferedBase(MemoryID(memoryName, "")),
        CachedBase(MemoryID(memoryName, "")),
        persistenceStrategy_(persistenceStrategy)
    {
    }
 
    void
    Memory::_setExportName(const std::string& memoryName)
    {
        ARMARX_TRACE;
 
        if (persistenceStrategy_)
        {
            persistenceStrategy_->setExportName(memoryName);
        }
    }
 
    void
    Memory::_enable()
    {
        ARMARX_TRACE;
 
        BufferedBase::start();
    }
 
    void
    Memory::_disable()
    {
        ARMARX_TRACE;
 
        BufferedBase::stop();
        ARMARX_IMPORTANT << "Storing of data finished, starting to generate and save statistics...";
        getAndSaveStatistics();
    }
 
    void
    Memory::_setMemoryID(const MemoryID& memoryId)
    {
        ARMARX_TRACE;
 
        ARMARX_DEBUG << "Setting memory=" << memoryId.memoryName
                     << ", id=" << memoryId.cleanID().str();
 
        BufferedBase::setMixinMemoryID(memoryId);
        CachedBase::setMixinMemoryID(memoryId);
    }
 
    bool
    Memory::_implForEachCoreSegment(std::function<void(CoreSegment&)> func) const
    {
        ARMARX_TRACE;
 
        ARMARX_DEBUG << "For each core segment (mem id=" << id().cleanID().str() << ")";
 
 
        for (auto& core_segment : persistenceStrategy_->getContainerKeys(id()))
        {
            ARMARX_DEBUG << "Found core segment=" << core_segment;
 
            std::shared_ptr<persistence::MemoryPersistenceStrategy> coreSegmentPersistenceStrategy(
                persistenceStrategy_);
 
            CoreSegment c(getExportName(),
                          id().withCoreSegmentName(core_segment),
                          processors,
                          coreSegmentPersistenceStrategy);
            func(c);
        }
 
 
        return true;
    }
 
    bool
    Memory::_implHasCoreSegment(const std::string& coreSegmentName) const
    {
        ARMARX_TRACE;
 
        bool foundCoreSegment = cacheHasCoreSegment(coreSegmentName) ||
                                persistenceStrategy_->containsContainer(id(), coreSegmentName);
 
        return foundCoreSegment;
    }
 
    std::shared_ptr<CoreSegment>
    Memory::_implFindCoreSegment(const std::string& coreSegmentName) const
    {
        ARMARX_TRACE;
 
        if (!_implHasCoreSegment(coreSegmentName))  // Call _impl version
        {
            return nullptr;
        }
 
        std::shared_ptr<persistence::MemoryPersistenceStrategy> coreSegmentPersistenceStrategy(
            persistenceStrategy_);
 
        return std::make_shared<CoreSegment>(getExportName(),
                                             id().withCoreSegmentName(coreSegmentName),
                                             processors,
                                             coreSegmentPersistenceStrategy);
    }
 
    void
    Memory::createPropertyDefinitions(PropertyDefinitionsPtr& defs, const std::string& prefix)
    {
        ARMARX_TRACE;
 
        MemoryBase::createPropertyDefinitions(defs, prefix);
        BufferedBase::createPropertyDefinitions(defs, prefix);
    }
 
    void
    Memory::_loadAllReferences(armem::wm::Memory& wmMemory)
    {
        ARMARX_TRACE;
 
        wmMemory.id() = id().getMemoryID().cleanID();
 
        ARMARX_DEBUG << "Memory: Load all references (id=" << wmMemory.id().cleanID().str() << ")";
 
        forEachCoreSegment(
            [&wmMemory](auto& ltmCoreSegment)
            {
                armem::wm::CoreSegment wmCoreSegment;
                ltmCoreSegment.loadAllReferences(wmCoreSegment);
 
                // Because there might be multiple LTMs providers with the same provider segment
                if (wmMemory.hasCoreSegment(wmCoreSegment.name()))
                {
                    armem::wm::CoreSegment existingWmCoreSegment =
                        wmMemory.getCoreSegment(wmCoreSegment.name());
                    existingWmCoreSegment.append(wmCoreSegment);
                }
                else
                {
                    wmMemory.addCoreSegment(wmCoreSegment);
                }
            });
    }
 
    void
    Memory::_loadAllReferences(armem::wm::Memory& wmMemory, std::list<std::string> coreSegmentNames)
    {
        //TODO
        ARMARX_TRACE;
 
        wmMemory.id() = id().getMemoryID().cleanID();
 
        ARMARX_DEBUG << "Memory: Load all references (id=" << wmMemory.id().cleanID().str() << ")";
 
        forEachCoreSegment(
            [&wmMemory, &coreSegmentNames](auto& ltmCoreSegment)
            {
                bool loadCoreSeg =
                    (std::find(coreSegmentNames.begin(),
                               coreSegmentNames.end(),
                               ltmCoreSegment.id().coreSegmentName) != coreSegmentNames.end());
                if (loadCoreSeg)
                {
                    armem::wm::CoreSegment wmCoreSegment;
                    ltmCoreSegment.loadAllReferences(wmCoreSegment);
 
                    // Because there might be multiple LTMs providers with the same provider segment
                    if (wmMemory.hasCoreSegment(wmCoreSegment.name()))
                    {
                        armem::wm::CoreSegment existingWmCoreSegment =
                            wmMemory.getCoreSegment(wmCoreSegment.name());
                        existingWmCoreSegment.append(wmCoreSegment);
                    }
                    else
                    {
                        wmMemory.addCoreSegment(wmCoreSegment);
                    }
                }
                else
                {
                    ARMARX_DEBUG << "Skipping loading CoreSegment with name "
                                 << wmMemory.id().coreSegmentName
                                 << " from LTM into WM as it is not in the defined list";
                }
            });
    }
 
    void
    Memory::_loadLatestNReferences(int n, armem::wm::Memory& wmMemory)
    {
        ARMARX_TRACE;
        wmMemory.id() = id().getMemoryID().cleanID();
 
        ARMARX_DEBUG << "Memory: Load latest N references for all core segments (id="
                     << wmMemory.id().cleanID().str() << ")";
 
        forEachCoreSegment(
            [&wmMemory, &n](auto& ltmCoreSegment)
            {
                armem::wm::CoreSegment wmCoreSegment;
                ltmCoreSegment.loadLatestNReferences(n, wmCoreSegment);
 
                // Because there might be multiple LTMs providers with the same provider segment
                if (wmMemory.hasCoreSegment(wmCoreSegment.name()))
                {
                    armem::wm::CoreSegment existingWmCoreSegment =
                        wmMemory.getCoreSegment(wmCoreSegment.name());
                    existingWmCoreSegment.append(wmCoreSegment);
                }
                else
                {
                    wmMemory.addCoreSegment(wmCoreSegment);
                }
            });
    }
 
    void
    Memory::_loadLatestNReferences(int n,
                                   armem::wm::Memory& wmMemory,
                                   std::list<std::string> coreSegNames)
    {
        ARMARX_TRACE;
        wmMemory.id() = id().getMemoryID().cleanID();
 
        ARMARX_DEBUG << "Memory: Load latest references for set of core segments (id="
                     << wmMemory.id().cleanID().str() << ")";
 
        forEachCoreSegment(
            [&wmMemory, &n, &coreSegNames](auto& ltmCoreSegment)
            {
                bool loadCoreSeg =
                    (std::find(coreSegNames.begin(),
                               coreSegNames.end(),
                               ltmCoreSegment.id().coreSegmentName) != coreSegNames.end());
                if (loadCoreSeg)
                {
                    ARMARX_DEBUG << "Load core segment=" << ltmCoreSegment.id().coreSegmentName;
                    armem::wm::CoreSegment wmCoreSegment;
                    ltmCoreSegment.loadLatestNReferences(n, wmCoreSegment);
 
                    // Because there might be multiple LTMs providers with the same provider segment
                    if (wmMemory.hasCoreSegment(wmCoreSegment.name()))
                    {
                        armem::wm::CoreSegment existingWmCoreSegment =
                            wmMemory.getCoreSegment(wmCoreSegment.name());
                        existingWmCoreSegment.append(wmCoreSegment);
                    }
                    else
                    {
                        wmMemory.addCoreSegment(wmCoreSegment);
                    }
                }
                else
                {
                    ARMARX_DEBUG << "Skipping loading CoreSegment with name "
                                 << wmMemory.id().coreSegmentName
                                 << " from LTM into WM as it is not in the defined list";
                }
            });
    }
 
    void
    Memory::_resolve(armem::wm::Memory& wmMemory)
    {
        ARMARX_TRACE;
 
        ARMARX_DEBUG << "Resolve memory id=" << id().cleanID().str();
 
        wmMemory.forEachCoreSegment(
            [&](auto& wmCoreSegment)
            {
                std::shared_ptr<persistence::MemoryPersistenceStrategy>
                    coreSegmentPersistenceStrategy(persistenceStrategy_);
 
                CoreSegment ltmCoreSegment(
                    getExportName(),
                    id().withCoreSegmentName(wmCoreSegment.id().coreSegmentName),
                    processors,
                    coreSegmentPersistenceStrategy);
 
                ltmCoreSegment.resolve(wmCoreSegment);
            });
    }
 
    void
    Memory::_store(const armem::wm::Memory& wmMemory)
    {
        ARMARX_TRACE;
        BufferedBase::addToBuffer(wmMemory);
    }
 
    void
    Memory::_directlyStore(const armem::wm::Memory& wmMemory, bool simulatedVersion)
    {
        ARMARX_TRACE;
 
        if (id().memoryName.empty())
        {
            ARMARX_WARNING
                << "During storage of memory '" << wmMemory.id().str()
                << "' I noticed that the corresponding LTM has no id set. "
                << "I set the id of the LTM to the same name, however this should not happen!";
            setMemoryID(wmMemory.id());
        }
 
        ARMARX_DEBUG << "Directly store memory id=" << id().cleanID().str();
        ARMARX_DEBUG << "CoreSegments: " << wmMemory.getCoreSegmentNames().size();
 
        wmMemory.forEachCoreSegment(
            [&](const auto& wmCoreSegment)
            {
                std::shared_ptr<persistence::MemoryPersistenceStrategy>
                    coreSegmentPersistenceStrategy(persistenceStrategy_);
 
                CoreSegment ltmCoreSegment(
                    getExportName(),
                    id().withCoreSegmentName(wmCoreSegment.id().coreSegmentName),
                    processors,
                    coreSegmentPersistenceStrategy);
 
                // 2. store data
                ltmCoreSegment.store(wmCoreSegment, simulatedVersion);
 
 
                // 3. update statistics
                statistics.recordedCoreSegments++;
            });
 
        // 4. update cache
        //CachedBase::addToCache(memory);
    }
 
    std::shared_ptr<armem::wm::Memory>
    Memory::_preFilterMemory(const armem::wm::Memory& memory,
                             uint64_t& filteredCount,
                             uint64_t& passedCount)
    {
        ARMARX_TRACE;
 
        filteredCount = 0;
        passedCount = 0;
 
        // If no filters and no segment selection configured, return a copy of the original memory
        if (!processors || (processors->snapFilters.empty() && !processors->hasSegmentSelection()))
        {
            // Count all snapshots as passed
            memory.forEachCoreSegment([&](const auto& cs) {
                cs.forEachProviderSegment([&](const auto& ps) {
                    ps.forEachEntity([&](const auto& entity) {
                        entity.forEachSnapshot([&](const auto&) {
                            passedCount++;
                        });
                    });
                });
            });
 
            // Return a copy wrapped in shared_ptr
            return std::make_shared<armem::wm::Memory>(memory);
        }
 
        // Create a new memory for filtered snapshots
        auto filteredMemory = std::make_shared<armem::wm::Memory>(memory.id());
 
        // Iterate through the memory hierarchy and apply filters
        memory.forEachCoreSegment([&](const auto& wmCoreSegment) {
            wmCoreSegment.forEachProviderSegment([&](const auto& wmProviderSegment) {
                wmProviderSegment.forEachEntity([&](const auto& wmEntity) {
                    wmEntity.forEachSnapshot([&](const auto& wmSnapshot) {
                        // Apply the segment whitelist and per-segment / global snapshot filters
                        bool accepted = processors->acceptSnapshotForSegment(
                            wmCoreSegment.name(), wmProviderSegment.name(), wmSnapshot, false);
 
                        if (accepted)
                        {
                            // Add snapshot to filtered memory
                            // Need to ensure the hierarchy exists
                            auto& cs = filteredMemory->hasCoreSegment(wmCoreSegment.name())
                                ? filteredMemory->getCoreSegment(wmCoreSegment.name())
                                : filteredMemory->addCoreSegment(wmCoreSegment.name());
 
                            auto& ps = cs.hasProviderSegment(wmProviderSegment.name())
                                ? cs.getProviderSegment(wmProviderSegment.name())
                                : cs.addProviderSegment(wmProviderSegment.name());
 
                            auto& entity = ps.hasEntity(wmEntity.name())
                                ? ps.getEntity(wmEntity.name())
                                : ps.addEntity(wmEntity.name());
 
                            entity.addSnapshot(wmSnapshot);
                            passedCount++;
                        }
                        else
                        {
                            filteredCount++;
                        }
                    });
                });
            });
        });
 
        ARMARX_DEBUG << "Pre-filter: " << passedCount << " passed, "
                    << filteredCount << " filtered out";
 
        return filteredMemory;
    }
 
    void
    Memory::_loadOnStartup()
    {
        //not used any more
    }
 
    void
    Memory::_enqueueForAsyncStorage(std::shared_ptr<const armem::wm::Memory> memory)
    {
        ARMARX_TRACE;
        // Forward to the BufferedMemoryMixin's async storage queue
        BufferedBase::enqueueForAsyncStoragePublic(std::move(memory));
    }
 
    void
    Memory::_enqueuePendingConversion(detail::mixin::PendingConversion pending)
    {
        ARMARX_TRACE;
        // Forward to the BufferedMemoryMixin's async storage queue
        BufferedBase::enqueuePendingConversion(std::move(pending));
    }
 
    void
    Memory::getAndSaveStatistics()
    {
        ARMARX_TRACE;
        auto firstTimeStarted = this->statistics.firstStarted;
        if (!firstTimeStarted.isValid())
        {
            //No statistics to be saved, as recording was never started
            ARMARX_DEBUG << "No Statistics will be saved because firstStarted is invalid: "
                         << VAROUT(this->statistics.firstStarted);
            return;
        }
 
        ARMARX_DEBUG << "Preparing to save statistics for " << this->name();
        try
        {
            auto first_stats = this->getFilterStatistics();
            if (first_stats.empty())
            {
                //empty statistics mean no data was recorded and no statistics should be saved
                ARMARX_DEBUG << "No Statistics will be saved because no actual data was recorded.";
                return;
            }
            std::map<std::string,
                     std::map<std::string, ltm::processor::SnapshotFilter::FilterStatistics>>
                information;
            std::map<std::string, armarx::core::time::DateTime> times;
 
            try
            {
                times["Started LTM1"] = this->getStatistics().lastStarted;
                times["Stopped LTM1"] = this->getStatistics().lastStopped;
                information["LTM"] = first_stats;
            }
            catch (...)
            {
                ARMARX_DEBUG << "Something went wrong after getting the statistics";
            }
            // auto exportPath = this->getMemoryBasePath();
            auto exportName = this->getExportName();
            auto recording_started = this->getStatistics().lastStarted;
            auto recording_stopped = this->getStatistics().lastStopped;
            // test::save_statistics(information,
            //                       times,
            //                       recording_started,
            //                       recording_stopped,
            //                       exportPath,
            //                       exportName,
            //                       this->name());
        }
        catch (...)
        {
            ARMARX_DEBUG << "Something went wrong with the statistics saving process";
        }
    }
 
    std::vector<std::string>
    Memory::split(std::string str, char delimiter)
    {
        ARMARX_TRACE;
        // Using str in a string stream
        std::stringstream ss(str);
        std::vector<std::string> res;
        std::string token;
        while (std::getline(ss, token, delimiter))
        {
            res.push_back(token);
        }
        return res;
    }
} // namespace armarx::armem::server::ltm