Memory.h

Go to the documentation of this file.

#pragma once
 
#include <list>
 
// Base Class
#include "detail/MemoryBase.h"
#include "detail/mixins/BufferedMemoryMixin.h"
#include "detail/mixins/CachedMemoryMixin.h"
 
// Persistence
#include "persistence/MemoryPersistenceStrategy.h"
#include "persistence/RedundantPersistenceStrategy.h"
 
// Segmnet Type
#include "CoreSegment.h"
 
namespace armarx::armem::server::ltm
{
    /// @brief A memory storing data on the hard drive and in mongodb (needs 'armarx memory start' to start the mongod instance)
    class Memory :
        public detail::MemoryBase<CoreSegment>,
        public detail::mixin::BufferedMemoryMixin<CoreSegment>,
        public detail::mixin::CachedMemoryMixin<CoreSegment>
    {
    public:
        using MemoryBase = detail::MemoryBase<CoreSegment>;
        using BufferedBase = detail::mixin::BufferedMemoryMixin<CoreSegment>;
        using CachedBase = detail::mixin::CachedMemoryMixin<CoreSegment>;
 
        Memory();
 
        Memory(const std::string& exportName, const std::string& memoryName);
 
        Memory(
            const std::string& exportName,
            const std::string& memoryName,
            const std::shared_ptr<persistence::RedundantPersistenceStrategy>& persistenceStrategy);
 
        void _setExportName(const std::string& memoryName) final;
        void _setMemoryID(const MemoryID& memoryId) final;
        void _enable() final;
        void _disable() final;
        void _configure(const nlohmann::json& config) final;
 
        void createPropertyDefinitions(PropertyDefinitionsPtr& defs,
                                       const std::string& prefix) override;
 
        /**
         * @brief getAndSaveStatistics generates and saves statistics for a LTM recording
         */
        void getAndSaveStatistics();
 
        /**
         * @brief Flush the async storage queue and wait for all pending items to be stored.
         * This blocks until the queue is empty and all storage operations are complete.
         * @param timeoutMs Maximum time to wait in milliseconds (0 = wait indefinitely)
         * @return true if queue was flushed successfully, false if timeout occurred
         */
        bool flushAsyncStorage(int timeoutMs = 0)
        {
            return BufferedBase::flushAsyncStorage(timeoutMs);
        }
 
        /**
         * @brief Get the current size of the async storage queue
         */
        size_t getAsyncQueueSize() const
        {
            return BufferedBase::getQueueSize();
        }
 
        /**
         * @brief Get the async storage statistics
         */
        const detail::mixin::AsyncStorageStatistics& getAsyncStorageStatistics() const
        {
            return BufferedBase::getAsyncStorageStatistics();
        }
 
        /**
         * @brief Get the number of threads currently processing items
         */
        size_t getNumThreadsProcessing() const
        {
            return BufferedBase::getNumThreadsProcessing();
        }
 
        void
        setPersistenceStrategy(
            std::shared_ptr<persistence::RedundantPersistenceStrategy> persistenceStrategy)
        {
            persistenceStrategy_ = persistenceStrategy;
        }
 
        std::shared_ptr<persistence::RedundantPersistenceStrategy>
        getPersistenceStrategy() const
        {
            return persistenceStrategy_;
        }
 
    protected:
        // Implementation methods
        bool _implForEachCoreSegment(std::function<void(CoreSegment&)> func) const final;
        bool _implHasCoreSegment(const std::string& coreSegmentName) const final;
        std::shared_ptr<CoreSegment> _implFindCoreSegment(const std::string& coreSegmentName) const final;
 
        void _loadAllReferences(armem::wm::Memory& wmMemory) final;
        void _loadAllReferences(armem::wm::Memory& wmMemory,
                                std::list<std::string> coreSegNames) final;
        void _loadLatestNReferences(int n, armem::wm::Memory& wmMemory) final;
        void _loadLatestNReferences(int n,
                                    armem::wm::Memory& wmMemory,
                                    std::list<std::string> coreSegNames) final;
        void _resolve(armem::wm::Memory& wmMemory) final;
        void _store(const armem::wm::Memory& wmMemory) final;
        void _directlyStore(const armem::wm::Memory& wmMemory, bool simulatedVersion) final;
        std::shared_ptr<armem::wm::Memory> _preFilterMemory(
            const armem::wm::Memory& memory,
            uint64_t& filteredCount,
            uint64_t& passedCount) final;
        void _loadOnStartup() final;
        void _enqueueForAsyncStorage(std::shared_ptr<const armem::wm::Memory> memory) final;
        void _enqueuePendingConversion(detail::mixin::PendingConversion pending) final;
 
    private:
        std::time_t current_date;
 
        std::shared_ptr<persistence::RedundantPersistenceStrategy> persistenceStrategy_;
 
        std::vector<std::string> split(std::string str, char delimiter);
    };
} // namespace armarx::armem::server::ltm