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)
    {
        std::lock_guard l(ltm_mutex);
 
        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);
 
                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)
    {
        std::lock_guard l(ltm_mutex);
 
        ARMARX_TRACE;
 
        ARMARX_DEBUG << "Setting memory=" << memoryId.memoryName
                     << ", id=" << memoryId.cleanID().str();
 
        BufferedBase::setMixinMemoryID(memoryId);
        CachedBase::setMixinMemoryID(memoryId);
    }
 
    bool
    Memory::forEachCoreSegment(std::function<void(CoreSegment&)> func) const
    {
        std::lock_guard l(ltm_mutex);
 
        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::hasCoreSegment(const std::string& coreSegmentName) const
    {
        std::lock_guard l(ltm_mutex);
 
        ARMARX_TRACE;
 
        bool foundCoreSegment = cacheHasCoreSegment(coreSegmentName) ||
                                persistenceStrategy_->containsContainer(id(), coreSegmentName);
 
        return foundCoreSegment;
    }
 
    std::shared_ptr<CoreSegment>
    Memory::findCoreSegment(const std::string& coreSegmentName) const
    {
        std::lock_guard l(ltm_mutex);
 
        ARMARX_TRACE;
 
        if (!hasCoreSegment(coreSegmentName))
        {
            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;
        std::lock_guard l(ltm_mutex);
 
        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::_loadLatestNReferences(int n, armem::wm::Memory& wmMemory)
    {
        ARMARX_TRACE;
        std::lock_guard l(ltm_mutex);
        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;
        std::lock_guard l(ltm_mutex);
        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;
        std::lock_guard l(ltm_mutex); // we cannot load a memory multiple times simultaneously
 
        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;
        std::lock_guard l(ltm_mutex); // we cannot store a memory multiple times simultaneously
 
        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);
    }
 
    void
    Memory::_loadOnStartup()
    {
        //not used any more
    }
 
    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