Memory.cpp

Go to the documentation of this file.

#include "Memory.h"
 
#include <ArmarXCore/core/logging/Logging.h>
#include <ArmarXCore/core/time/TimeUtil.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);
 
        BufferedBase::configureMixin(json);
        CachedBase::configureMixin(json);
        DiskMemoryBase::configureMixin(json);
        MongoDBStorageMixin::configureMixin(json);
    }
 
    Memory::Memory() : Memory(std::filesystem::path("/tmp/ltm"), {}, "MemoryExport", "Test")
    {
    }
 
    Memory::Memory(const detail::mixin::Path& p,
 
                   const std::string& exportName,
                   const std::string& memoryName /* UNESCAPED */) :
        Memory(p, {}, exportName, memoryName)
    {
    }
 
    Memory::Memory(const detail::mixin::Path& p,
                   const detail::mixin::MongoDBSettings& s,
 
                   const std::string& exportName,
                   const std::string& memoryName /* UNESCAPED */) :
        MemoryBase(exportName, MemoryID(memoryName, "")),
        BufferedBase(MemoryID(memoryName, "")),
        CachedBase(MemoryID(memoryName, "")),
        DiskMemoryItemMixin(p, exportName, MemoryID(memoryName, "")),
        MongoDBStorageMixin(s, exportName, MemoryID(memoryName, ""))
    {
        ARMARX_INFO << "Creating a new memory at " << p.string() << " and " << exportName;
    }
 
    void
    Memory::_setExportName(const std::string& n)
    {
        DiskMemoryBase::setMixinExportName(n);
        MongoDBStorageMixin::setMixinExportName(n);
    }
 
    void
    Memory::_enable()
    {
        BufferedBase::start();
        //MongoDBStorageMixin::start();
    }
 
    void
    Memory::_disable()
    {
        BufferedBase::stop();
        MongoDBStorageMixin::stop();
        ARMARX_IMPORTANT << "Storing of data finished, starting to generate and save statistics...";
        getAndSaveStatistics();
    }
 
    void
    Memory::_setMemoryID(const MemoryID& id)
    {
        std::lock_guard l(ltm_mutex);
 
        BufferedBase::setMixinMemoryID(id);
        CachedBase::setMixinMemoryID(id);
        DiskMemoryBase::setMixinMemoryID(id);
        MongoDBStorageMixin::setMixinMemoryID(id);
    }
 
    bool
    Memory::forEachCoreSegment(std::function<void(CoreSegment&)> func) const
    {
        std::lock_guard l(ltm_mutex);
 
        // legacy: check fs
        if (fullPathExists())
        {
            for (const auto& subdirName : getAllDirectories())
            {
                const std::string& segmentName_one = util::fs::detail::unescapeName(subdirName);
                auto segmentName = util::fs::detail::extractLastDirectoryFromPath(segmentName_one);
                CoreSegment c(getMemoryBasePath(),
                              getSettings(),
                              getExportName(),
                              id().withCoreSegmentName(segmentName),
                              processors);
                func(c);
            }
        }
        else
        {
            ARMARX_WARNING << "Could not load the core segments of LTM " 
                           << id().str() 
                           << " as the path " 
                           << getFullPath().string() 
                           << " does not exist.";
        }
 
        ARMARX_DEBUG << "All CoreSegments handeled";
 
        return true;
    }
 
    bool
    Memory::hasCoreSegment(const std::string& name) const
    {
        std::lock_guard l(ltm_mutex);
 
        if (cacheHasCoreSegment(name))
        {
            return true;
        }
 
        // check if collection exists
        /*if (connected() && collectionExists())
        {
            auto c = CoreSegment(getMemoryBasePath(),
                                 getSettings(),
                                 getExportName(),
                                 id().withCoreSegmentName(name),
                                 processors);
 
            return (bool)c.collectionExists();
        }*/
 
        // legacy: check if segment is stored on hard drive without db
        if (fullPathExists())
        {
            auto c = CoreSegment(getMemoryBasePath(),
                                 getSettings(),
                                 getExportName(),
                                 id().withCoreSegmentName(name),
                                 processors);
 
            return c.fullPathExists();
        }
 
        return false;
    }
 
    std::unique_ptr<CoreSegment>
    Memory::findCoreSegment(const std::string& coreSegmentName) const
    {
        std::lock_guard l(ltm_mutex);
 
        if (!hasCoreSegment(coreSegmentName))
        {
            return nullptr;
        }
 
        return std::make_unique<CoreSegment>(getMemoryBasePath(),
                                             getSettings(),
                                             getExportName(),
                                             id().withCoreSegmentName(coreSegmentName),
                                             processors);
    }
 
    void
    Memory::createPropertyDefinitions(PropertyDefinitionsPtr& defs, const std::string& prefix)
    {
        MemoryBase::createPropertyDefinitions(defs, prefix);
        DiskMemoryBase::createPropertyDefinitions(defs, prefix);
        BufferedBase::createPropertyDefinitions(defs, prefix);
    }
 
    void
    Memory::_loadAllReferences(armem::wm::Memory& m)
    {
        std::lock_guard l(ltm_mutex);
 
        m.id() = id().getMemoryID().cleanID();
 
        forEachCoreSegment(
            [&m](auto& x)
            {
                armem::wm::CoreSegment s;
                x.loadAllReferences(s);
                m.addCoreSegment(s);
            });
    }
 
    void
    Memory::_loadLatestNReferences(int n, armem::wm::Memory& m)
    {
        std::lock_guard l(ltm_mutex);
        m.id() = id().getMemoryID().cleanID();
 
        forEachCoreSegment(
            [&m, &n](auto& x)
            {
                armem::wm::CoreSegment s;
                x.loadLatestNReferences(n, s);
                m.addCoreSegment(s);
            });
    }
 
    void
    Memory::_loadLatestNReferences(int n, armem::wm::Memory& m, std::list<std::string> coreSegNames)
    {
        std::lock_guard l(ltm_mutex);
        m.id() = id().getMemoryID().cleanID();
 
        forEachCoreSegment(
            [&m, &n, &coreSegNames](auto& x)
            {
                bool loadCoreSeg = (std::find(coreSegNames.begin(), coreSegNames.end(), x.id().coreSegmentName) != coreSegNames.end());
                if(loadCoreSeg){
                    armem::wm::CoreSegment s;
                    x.loadLatestNReferences(n, s);
                    m.addCoreSegment(s);
                } else {
                    ARMARX_DEBUG << "Skipping loading CoreSegment with name " 
                                 << x.id().coreSegmentName 
                                 << " from LTM into WM as it is not in the defined list";
                }
                
            });
    }
 
    void
    Memory::_resolve(armem::wm::Memory& m)
    {
        std::lock_guard l(ltm_mutex); // we cannot load a memory multiple times simultaneously
 
        ARMARX_DEBUG << VAROUT(fullPathExists());
 
        if (/*(connected() && collectionExists()) ||*/ fullPathExists())
        {
            m.forEachCoreSegment(
                [&](auto& e)
                {
                    //ARMARX_INFO << "resolve for CoreSegment " << e.id().str();
                    CoreSegment c(getMemoryBasePath(),
                                  getSettings(),
                                  getExportName(),
                                  id().withCoreSegmentName(e.id().coreSegmentName),
                                  processors);
                    c.resolve(e);
                });
        } else {
            ARMARX_WARNING << "You are trying to resolve an LTM from a path that does not exist: " 
                           << getFullPath();
        }
    }
 
    void
    Memory::_store(const armem::wm::Memory& m)
    {
        BufferedBase::addToBuffer(m);
    }
 
    void
    Memory::_directlyStore(const armem::wm::Memory& memory)
    {
        std::lock_guard l(ltm_mutex); // we cannot store a memory multiple times simultaneously
 
        if (id().memoryName.empty())
        {
            ARMARX_WARNING
                << "During storage of memory '" << memory.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(memory.id());
        }
 
        /*if (!connected())
        {
            ARMARX_WARNING << "LTM NOT CONNECTED ALTHOUGH ENABLED " << id().str();
            return;
        }*/
 
        memory.forEachCoreSegment(
            [&](const auto& core)
            {
                CoreSegment c(getMemoryBasePath(),
                              getSettings(),
                              getExportName(),
                              id().withCoreSegmentName(core.id().coreSegmentName),
                              processors);
 
                // 2. store data
                c.store(core);
                
 
                // 3. update statistics
                statistics.recordedCoreSegments++;
            });
 
        // 4. update cache
        //CachedBase::addToCache(memory);
    }
 
    void
    Memory::_loadOnStartup(){
        //not used any more
        
    }
 
    void
    Memory::getAndSaveStatistics()
    {
        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_INFO << "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_INFO << "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_INFO << "Something went wrong with the statistics saving process";
        }
    }
} // namespace armarx::armem::server::ltm