MemoryBase.h

Go to the documentation of this file.

#pragma once
 
#include <functional>
#include <list>
#include <map>
#include <set>
#include <string>
#include <vector>
 
 
// BaseClass
#include "MemoryItem.h"
 
// ChildType
 
// ArmarX
#include <ArmarXCore/core/Component.h>
#include <ArmarXCore/core/logging/LoggingUtil.h>
#include <ArmarXCore/core/time.h>
#include <ArmarXCore/interface/core/UserException.h>
#include <ArmarXCore/interface/observers/ObserverInterface.h>
#include <ArmarXCore/observers/variant/Variant.h>
 
#include <RobotAPI/libraries/armem/core/MemoryID.h>
#include <RobotAPI/libraries/armem/core/operations.h>
#include <RobotAPI/libraries/armem/core/wm/aron_conversions.h>
#include <RobotAPI/libraries/armem/core/wm/memory_definitions.h>
#include <RobotAPI/libraries/armem/server/test/ForgettingExperiments.h>
#include <RobotAPI/libraries/armem/server/wm/memory_definitions.h>
 
// PendingConversion is defined separately to avoid coupling MemoryBase to BufferedMemoryMixin
// This allows different storage backends (disk, MongoDB, etc.) to be used independently
#include "PendingConversion.h"
 
namespace armarx::armem::server::ltm
{
    enum class RecordingMode
    {
        CONSOLIDATE_REMOVED, // only store whats removed from the WM
        CONSOLIDATE_ALL, // store whats in the WM (on commit)
        CONSOLIDATE_LATEST // store the latest snapshots at a fixed frequency (for now 1Hz)
    };
} // namespace armarx::armem::server::ltm
 
namespace armarx::armem::server::ltm::detail
{
    /// @brief Interface functions for the longterm memory classes
    template <class _CoreSegmentT>
    class MemoryBase : public MemoryItem
    {
    public:
        struct Statistics
        {
            armarx::core::time::DateTime lastEnabled = armarx::core::time::DateTime::Invalid();
            long recordedCoreSegments = 0;
            bool firstStart = true;
            bool firstStop = true;
            armarx::core::time::DateTime firstStarted = armarx::core::time::DateTime::Invalid();
            armarx::core::time::DateTime firstStopped = armarx::core::time::DateTime::Invalid();
            armarx::core::time::DateTime lastStarted = armarx::core::time::DateTime::Invalid();
            armarx::core::time::DateTime lastStopped = armarx::core::time::DateTime::Invalid();
        };
 
    public:
        using CoreSegmentT = _CoreSegmentT;
 
        MemoryBase(const std::string& exportName, const MemoryID& id) :
            MemoryItem(exportName, id), enabled(false)
        {
        }
 
        void
        setRecordingMode(const std::string& m)
        {
            if (m == "CONSOLIDATE_REMOVED")
            {
                this->recordingMode = RecordingMode::CONSOLIDATE_REMOVED;
            }
            else if (m == "CONSOLIDATE_ALL")
            {
                this->recordingMode = RecordingMode::CONSOLIDATE_ALL;
            }
            else if (m == "CONSOLIDATE_LATEST")
            {
                this->recordingMode = RecordingMode::CONSOLIDATE_LATEST;
            }
            else
            {
                ARMARX_WARNING << "Unknown recording mode: " << m;
            }
        }
 
        void
        setRecordingMode(const RecordingMode m)
        {
            this->recordingMode = m;
        }
 
        RecordingMode
        getRecordingMode() const
        {
            return recordingMode;
        }
 
        /// initialize config
        void
        configure()
        {
            bool en = p.enabled_on_startup;
            ARMARX_INFO << VAROUT(p.configuration_on_startup);
            ARMARX_INFO << VAROUT(p.export_name);
            ARMARX_INFO << VAROUT(p.export_path);
 
            this->setExportName(p.export_name);
            this->setRecordingMode(p.recordingMode);
 
            try
            {
                const auto json = nlohmann::json::parse(p.configuration_on_startup);
 
                // Processors are shared. So we only need to configure the root
                processors->configure(json);
 
                this->_configure(json);
            }
            catch (...)
            {
                ARMARX_WARNING << "Failed to parse `" << p.configuration_on_startup << "`";
                en = false;
            }
 
            if (en)
            {
                this->startRecording();
            }
        }
 
        /// enable this LTM
        void
        enable()
        {
            processors->resetFilterStatisticsForNewEpisode(); //make sure the statistics are
            //resetted before the recording starts
            auto now = armarx::core::time::DateTime::Now();
            ARMARX_INFO << "Enabling LTM " << id().str() << " at " << now.toDateTimeString();
            ARMARX_INFO << "Storing information at " << this->getExportName();
            enabled = true;
            if (statistics.firstStart)
            {
                statistics.firstStarted = now;
                statistics.firstStart = false;
            }
            statistics.lastStarted = now;
            this->_enable();
        }
 
        /// disable this LTM
        void
        disable()
        {
            auto now = armarx::core::time::DateTime::Now();
            ARMARX_INFO << "Disabling LTM " << id().str() << " at " << now.toDateTimeString();
            enabled = false;
            if (statistics.firstStop)
            {
                statistics.firstStopped = now;
                statistics.firstStop = false;
            }
            statistics.lastStopped = now;
            this->_disable();
            ARMARX_INFO << "Disabling of LTM finished";
        }
 
        /// return the full ltm as a wm::Memory with only references
        /// the ltm may be huge, use with caution
        armem::wm::Memory
        loadAllReferences()
        {
            armem::wm::Memory ret;
            loadAllReferences(ret);
            return ret;
        }
 
        void
        loadAllReferences(armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_LoadAll);
            _loadAllReferences(memory);
            TIMING_END_STREAM(LTM_Memory_LoadAll, ARMARX_DEBUG);
        }
 
        //return the newest part of the ltm as a wm::Memory with only references
        //will return the newest n snapshots of each entity
        armem::wm::Memory
        loadLatestNReferences(int n)
        {
            armem::wm::Memory ret;
            loadLatestNReferences(n, ret);
            return ret;
        }
 
        //return the newest part of the ltm as a wm::Memory with only references
        //will return the newest n snapshots of each entity for only the CoreSegments matching one of the names in the list
        void
        loadLatestNReferences(int n, armem::wm::Memory& memory, std::list<std::string> coreSegNames)
        {
            TIMING_START(LTM_Memory_loadNewestn_someCoreSegs);
            _loadLatestNReferences(n, memory, coreSegNames);
            TIMING_END_STREAM(LTM_Memory_loadNewestn_someCoreSegs, ARMARX_DEBUG);
        }
 
        void
        loadLatestNReferences(int n, armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_loadNewestn);
            _loadLatestNReferences(n, memory);
            TIMING_END_STREAM(LTM_Memory_loadNewestn, ARMARX_DEBUG);
        }
 
        /// return the full ltm as a wm::Memory and resolves the references
        /// the ltm may be huge, use with caution
        armem::wm::Memory
        loadAllAndResolve()
        {
            armem::wm::Memory ret;
            loadAllAndResolve(ret);
            return ret;
        }
 
        void
        loadAllAndResolve(armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_LoadAllAndResolve);
            _loadAllReferences(memory);
            _resolve(memory);
            TIMING_END_STREAM(LTM_Memory_LoadAllAndResolve, ARMARX_DEBUG);
        }
 
        void
        loadAllAndResolve(armem::wm::Memory& memory, std::list<std::string> coreSegmentNames)
        {
            TIMING_START(LTM_Memory_LoadAllAndResolve);
            _loadAllReferences(memory, coreSegmentNames);
            _resolve(memory);
            TIMING_END_STREAM(LTM_Memory_LoadAllAndResolve, ARMARX_DEBUG);
        }
 
        /// convert the references of the input into a wm::Memory
        void
        resolve(armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_Load);
            _resolve(memory);
            TIMING_END_STREAM(LTM_Memory_Load, ARMARX_DEBUG);
        }
 
        /// append a wm::Memory instance to the ltm (asynchronously via thread pool)
        void
        store(std::shared_ptr<const armem::wm::Memory> memory)
        {
            TIMING_START(LTM_Memory_Append);
 
            IceUtil::Time startTime;
            if (debugObserver)
            {
                startTime = IceUtil::Time::now();
            }
 
            for (auto& f : processors->memFilters)
            {
                if (!f->accept(*memory))
                {
                    ARMARX_DEBUG
                        << deactivateSpam(60)
                        << "Ignoring to put a Memory into the LTM because it got filtered.";
                    return;
                }
            }
            // Store asynchronously using the thread pool in BufferedMemoryMixin
            // No copy needed - just move the shared_ptr
            _enqueueForAsyncStorage(std::move(memory));
 
            if (debugObserver)
            {
                IceUtil::Time endTime = IceUtil::Time::now();
                IceUtil::Time elapsed = endTime - startTime;
                float elapsedMs = elapsed.toMilliSecondsDouble();
 
                std::string channelName = id().memoryName + "Memory_LTM";
                debugObserver->setDebugChannel(
                    channelName,
                    {
                        {"LTM | t store [ms]", new Variant(elapsedMs)},
                    });
            }
 
            TIMING_END_STREAM(LTM_Memory_Append, ARMARX_DEBUG);
        }
 
        /// append a wm::Memory instance to the ltm
        void
        store(const armem::server::wm::Memory& serverMemory)
        {
            auto memory = std::make_shared<armem::wm::Memory>(serverMemory.name());
            memory->update(armem::toCommit(serverMemory), true, false);
            this->store(std::move(memory));
        }
 
        /// Store snapshots with deferred conversion (happens in async thread)
        /// This is more efficient than calling store() with a pre-converted Memory
        /// because the expensive toMemory() conversion happens off the critical path
        ///
        /// IMPORTANT: This method does NOT access the WM structure to avoid deadlocks.
        /// The caller must pre-extract metadata before calling this method.
        void
        storeSnapshotsAsync(const std::string& memoryName,
                           const std::vector<wm::EntitySnapshot>& snapshots,
                           const std::vector<mixin::PendingConversion::SegmentMetadata>& segmentMetadata)
        {
            if (!enabled)
            {
                return;
            }
 
            // Pre-filter snapshots before enqueuing to reduce queue pressure.
            // Applies the segment whitelist (`storeSegments`) and per-segment / global
            // snapshot filters via Processors::acceptSnapshotForSegment.
            std::vector<wm::EntitySnapshot> filteredSnapshots;
            if (processors &&
                (!processors->snapFilters.empty() || processors->hasSegmentSelection()))
            {
                filteredSnapshots.reserve(snapshots.size());
                for (const auto& snapshot : snapshots)
                {
                    if (processors->acceptSnapshotForSegment(snapshot.id().coreSegmentName,
                                                             snapshot.id().providerSegmentName,
                                                             snapshot,
                                                             false))
                    {
                        filteredSnapshots.push_back(snapshot);
                    }
                }
 
                // If all snapshots were filtered, skip enqueuing
                if (filteredSnapshots.empty())
                {
                    ARMARX_DEBUG << deactivateSpam(10) << "All " << snapshots.size()
                                << " snapshots were pre-filtered in storeSnapshotsAsync";
                    return;
                }
            }
            else
            {
                // No filters and no segment selection, use all snapshots
                filteredSnapshots = snapshots;
            }
 
            // Build pending conversion with pre-filtered snapshots
            mixin::PendingConversion pending;
            pending.memoryName = memoryName;
            pending.snapshots = std::move(filteredSnapshots);
            pending.segmentMetadata = segmentMetadata;
 
            // Enqueue for async conversion and storage
            this->_enqueuePendingConversion(std::move(pending));
        }
 
        /// iterate over all core segments of this ltm
        bool
        forEachCoreSegment(std::function<void(CoreSegmentT&)> func) const
        {
            return _implForEachCoreSegment(func);
        }
 
        /// check if core segment exists
        bool
        hasCoreSegment(const std::string& coreSegmentName) const
        {
            return _implHasCoreSegment(coreSegmentName);
        }
 
        /// find core segment
        std::shared_ptr<CoreSegmentT>
        findCoreSegment(const std::string& coreSegmentName) const
        {
            return _implFindCoreSegment(coreSegmentName);
        }
 
        /// default parameters. Implementation should use the configuration to configure
        virtual void
        createPropertyDefinitions(PropertyDefinitionsPtr& defs, const std::string& prefix)
        {
            defs->optional(p.enabled_on_startup, prefix + "enabled");
            defs->optional(p.recordingMode, prefix + "recordingMode");
            defs->optional(
                p.configuration_on_startup,
                prefix + "configuration",
                "JSON LTM configuration. Top-level keys configure global snapshot filters "
                "(e.g. SnapshotFrequencyFilter). Optional \"storeSegments\": [\"Core\", "
                "\"Core/Provider\"] whitelists which segments to store (absent => store all). "
                "Optional \"segments\": { \"Core\" | \"Core/Provider\": { <filters> } } sets "
                "per-segment filter overrides (falling back to the global filters).");
            defs->optional(p.export_name, prefix + "exportName");
            defs->optional(p.storeOnStop, prefix + "storeOnStop");
            defs->optional(p.export_path, prefix + "exportPath");
 
            defs->optional(p.importOnStartUp, prefix + "importOnStartUp");
            defs->optional(p.maxAmountOfSnapshotsLoaded, prefix + "maxAmountSnapshotsLoaded");
            defs->optional(p.coreSegmentsToLoad, prefix + "loadedCoreSegments");
 
            defs->optional(p.persistenceStrategyIdentifier,
                           prefix + "persistenceStrategyIdentifier");
            defs->optional(p.persistenceStrategies, prefix + "persistenceStrategies");
            defs->optional(p.restHost, prefix + "rest.host");
            defs->optional(p.restPort, prefix + "rest.port");
            defs->optional(p.restDisableIfNotAvailable, prefix + "rest.disableIfNotAvailable");
 
            defs->optional(p.minDiskSpace, prefix + "minDiskSpace");
 
            // Batch write configuration
            defs->optional(p.batchWriteEnabled, prefix + "batchWriteEnabled",
                          "Enable batch writing for improved I/O throughput");
            defs->optional(p.batchSizeThreshold, prefix + "batchSizeThreshold",
                          "Number of items to accumulate before flushing to disk");
            defs->optional(p.batchTimeThresholdMs, prefix + "batchTimeThresholdMs",
                          "Maximum time in ms to hold items before flushing");
        }
 
        /// enable/disable
        void
        startRecording()
        {
            statistics.lastEnabled = armarx::core::time::DateTime::Now();
 
            enable();
        }
 
        void
        stopRecording()
        {
 
            disable();
        }
 
        bool
        isRecording() const
        {
            return enabled;
        }
 
        /// statistics
        virtual void
        resetStatistics()
        {
            // enabled stays the same
            statistics.lastEnabled = armarx::core::time::DateTime::Invalid();
            statistics.recordedCoreSegments = 0;
        }
 
        Statistics
        getStatistics() const
        {
            return statistics;
        }
 
        std::map<std::string, processor::SnapshotFilter::FilterStatistics>
        getFilterStatistics()
        {
            try
            {
                ARMARX_INFO << "Trying to save statistics";
                auto stats = processors->getSnapshotFilterStatistics();
                return stats;
            }
            catch (InvalidArgumentException& e)
            { //no data was actually recorded
                ARMARX_INFO << e.what();
            }
            catch (...)
            {
                ARMARX_WARNING << "Saving statistics did not work";
            }
            std::map<std::string, processor::SnapshotFilter::FilterStatistics> emptyStatistics;
            return emptyStatistics;
        }
 
        /// get level name1
        static std::string
        getLevelName()
        {
            return "LT-Memory";
        }
 
        void
        loadOnStartup()
        {
            _loadOnStartup();
        }
 
        /**
         * @brief Set an optional debug observer for timing measurements
         * @param observer The debug observer proxy (can be nullptr to disable)
         *
         * This allows optional timing reporting to a debug observer.
         * If set, you can report timing information like:
         *
         * @code
         * if (debugObserver)
         * {
         *     debugObserver->setDebugChannel("ChannelName", {
         *         {"MetricName [ms]", new Variant(timeInMs)}
         *     });
         * }
         * @endcode
         */
        void
        setDebugObserver(DebugObserverInterfacePrx observer)
        {
            debugObserver = observer;
        }
 
        /**
         * @brief Get the current debug observer (may be nullptr)
         */
        DebugObserverInterfacePrx
        getDebugObserver() const
        {
            return debugObserver;
        }
 
 
    protected:
        /// configuration
        virtual void
        _configure(const nlohmann::json&)
        {
        }
 
        virtual void
        _enable()
        {
        }
 
        virtual void
        _disable()
        {
        }
 
        virtual void
        _setExportName(const std::string&)
        {
        }
 
        virtual void _loadAllReferences(armem::wm::Memory& memory) = 0;
        virtual void _loadAllReferences(armem::wm::Memory& memory,
                                        std::list<std::string> coreSegNames) = 0;
        virtual void _resolve(armem::wm::Memory& memory) = 0;
        virtual void _store(const armem::wm::Memory& memory) = 0;
        virtual void _directlyStore(const armem::wm::Memory& memory,
                                    bool simulatedVersion = false) = 0;
        virtual void _loadOnStartup() = 0;
        virtual void _loadLatestNReferences(int n, armem::wm::Memory& memory) = 0;
        virtual void _loadLatestNReferences(int n,
                                            armem::wm::Memory& memory,
                                            std::list<std::string> coreSegNames) = 0;
        virtual void _enqueueForAsyncStorage(std::shared_ptr<const armem::wm::Memory> memory) = 0;
        virtual void _enqueuePendingConversion(mixin::PendingConversion pending) = 0;
 
        // Implementation methods - subclasses override these
        // Thread safety is handled by DiskPersistence's per-directory mutexes
        virtual bool _implForEachCoreSegment(std::function<void(CoreSegmentT&)> func) const = 0;
        virtual bool _implHasCoreSegment(const std::string& coreSegmentName) const = 0;
        virtual std::shared_ptr<CoreSegmentT> _implFindCoreSegment(const std::string& coreSegmentName) const = 0;
 
    public:
        // stuff for scenario parameters
        struct Properties
        {
            // whether the LTM is enabled on startup
            bool enabled_on_startup = false;
 
            // How data is stored, if the LTM is enabled
            std::string recordingMode = "CONSOLIDATE_REMOVED";
 
            // Name and export path (TODO: Should this be part of the base class? Export path is a disk memory thing)
            std::string export_name = "MemoryExport";
            std::string export_path = "/tmp/ltm";
            bool storeOnStop = false;
            int minDiskSpace = 50; //GB
 
            // TODO: belong more to WM, but no good solution for that currently:
            bool importOnStartUp = false;
            int maxAmountOfSnapshotsLoaded = 1;
            std::string coreSegmentsToLoad = "";
 
            // Other configuration
            std::string configuration_on_startup =
                "{ \"SnapshotFrequencyFilter\": "
                "{\"WaitingTimeInMsForFilter\" : 50}}"; //record with 20 fps as standard
 
            std::string persistenceStrategies = "disk";
            std::string persistenceStrategyIdentifier = "localDisk";
            std::string restHost = "localhost";
            int restPort = 8080;
            bool restDisableIfNotAvailable = false;
 
            bool overwriteIdentifierOnLtmLoad = true;
 
            // Batch write configuration for disk persistence
            bool batchWriteEnabled = false;      // Enable batch writing for improved I/O throughput
            int batchSizeThreshold = 100;        // Flush batch when it reaches this many items
            int batchTimeThresholdMs = 100;      // Flush batch after this many milliseconds
 
        } p;
 
    protected:
        mutable Statistics statistics;
 
        std::atomic_bool enabled = true;
 
        RecordingMode recordingMode = RecordingMode::CONSOLIDATE_REMOVED;
 
        /// Optional debug observer for timing measurements
        DebugObserverInterfacePrx debugObserver;
    };
} // namespace armarx::armem::server::ltm::detail