memory_definitions.h

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <chrono>
#include <mutex>
#include <shared_mutex>
 
#include <RobotAPI/libraries/armem/core/base/CoreSegmentBase.h>
#include <RobotAPI/libraries/armem/core/base/EntityBase.h>
#include <RobotAPI/libraries/armem/core/base/EntityInstanceBase.h>
#include <RobotAPI/libraries/armem/core/base/EntitySnapshotBase.h>
#include <RobotAPI/libraries/armem/core/base/MemoryBase.h>
#include <RobotAPI/libraries/armem/core/base/ProviderSegmentBase.h>
#include <RobotAPI/libraries/armem/core/wm/detail/data_lookup_mixins.h>
#include <RobotAPI/libraries/armem/core/wm/memory_definitions.h>
 
#include "detail/MaxHistorySize.h"
#include "detail/Prediction.h"
 
namespace armarx::armem::server::wm
{
    /**
     * @brief Statistics for CoreSegment lock operations.
     *
     * Thread-safe statistics for tracking read/write lock contention and timing.
     */
    struct CoreSegmentLockStatistics
    {
        // Read lock statistics
        std::atomic<uint64_t> readLockCount{0};
        std::atomic<uint64_t> totalReadLockWaitNs{0};
        std::atomic<uint64_t> maxReadLockWaitNs{0};
 
        // Write lock statistics
        std::atomic<uint64_t> writeLockCount{0};
        std::atomic<uint64_t> totalWriteLockWaitNs{0};
        std::atomic<uint64_t> maxWriteLockWaitNs{0};
 
        void reset()
        {
            readLockCount = 0;
            totalReadLockWaitNs = 0;
            maxReadLockWaitNs = 0;
            writeLockCount = 0;
            totalWriteLockWaitNs = 0;
            maxWriteLockWaitNs = 0;
        }
 
        double getAvgReadLockWaitMs() const
        {
            uint64_t count = readLockCount.load(std::memory_order_relaxed);
            if (count == 0) return 0.0;
            return totalReadLockWaitNs.load(std::memory_order_relaxed) / (count * 1e6);
        }
 
        double getAvgWriteLockWaitMs() const
        {
            uint64_t count = writeLockCount.load(std::memory_order_relaxed);
            if (count == 0) return 0.0;
            return totalWriteLockWaitNs.load(std::memory_order_relaxed) / (count * 1e6);
        }
 
        double getMaxReadLockWaitMs() const
        {
            return maxReadLockWaitNs.load(std::memory_order_relaxed) / 1e6;
        }
 
        double getMaxWriteLockWaitMs() const
        {
            return maxWriteLockWaitNs.load(std::memory_order_relaxed) / 1e6;
        }
    };
 
    using EntityInstanceMetadata = base::EntityInstanceMetadata;
    using EntityInstanceData = armarx::aron::data::Dict;
    using EntityInstanceDataPtr = armarx::aron::data::DictPtr;
 
    using EntityInstance = armem::wm::EntityInstance;
    using EntitySnapshot = armem::wm::EntitySnapshot;
 
    /// @see base::EntityBase
    class Entity :
        public base::EntityBase<EntitySnapshot, Entity>,
        public detail::MaxHistorySize,
        public armem::wm::detail::FindInstanceDataMixinForEntity<Entity>
    {
    public:
        using base::EntityBase<EntitySnapshot, Entity>::EntityBase;
 
 
        /**
         * @brief Sets the maximum history size.
         *
         * The current history is truncated if necessary.
         */
        void setMaxHistorySize(long maxSize);
 
        UpdateResult update(const EntityUpdate& update);
 
 
    protected:
        /// If maximum size is set, ensure `history`'s is not higher.
        std::vector<EntitySnapshotT> truncate();
    };
 
    /// @see base::ProviderSegmentBase
    class ProviderSegment :
        public base::ProviderSegmentBase<Entity, ProviderSegment>,
        public detail::MaxHistorySizeParent<ProviderSegment>,
        public armem::wm::detail::FindInstanceDataMixin<ProviderSegment>,
        public armem::server::wm::detail::Prediction<ProviderSegment>
    {
    public:
        using base::ProviderSegmentBase<Entity, ProviderSegment>::ProviderSegmentBase;
 
 
        using ProviderSegmentBase::addEntity;
 
        template <class... Args>
        Entity&
        addEntity(const std::string& name, Args... args)
        {
            Entity& added = ProviderSegmentBase::addEntity(name, args...);
            added.setMaxHistorySize(this->getMaxHistorySize());
            added.setUnlimitedHistoryWarningThreshold(this->getUnlimitedHistoryWarningThreshold());
            added.setTruncateMaxBatchSize(this->getTruncateMaxBatchSize());
            return added;
        }
    };
 
    /// @brief base::CoreSegmentBase
    class CoreSegment :
        public base::CoreSegmentBase<ProviderSegment, CoreSegment>,
        public detail::MaxHistorySizeParent<CoreSegment>,
        public armem::wm::detail::FindInstanceDataMixin<CoreSegment>,
        public armem::server::wm::detail::PredictionContainer<CoreSegment>
    {
        using Base = base::CoreSegmentBase<ProviderSegment, CoreSegment>;
 
    public:
        using Base::CoreSegmentBase;
 
        /// @see base::CoreSegmentBase::addProviderSegment()
        using CoreSegmentBase::addProviderSegment;
 
        template <class... Args>
        ProviderSegment&
        addProviderSegment(const std::string& name, Args... args)
        {
            ProviderSegmentT& added = CoreSegmentBase::addProviderSegment(name, args...);
            int maxHistorySize = this->getMaxHistorySize();
            if (maxHistorySize < 0)
            {
                ARMARX_INFO << "The maxHistorySize for this core segment is set to < 0. "
                            << "This means nothing will ever be forgotten in working memory. "
                            << "This may slow down the memory server. \n"
                            << "Core Segment Name: " << (this->id()).str();
            }
            added.setMaxHistorySize(maxHistorySize);
            added.setUnlimitedHistoryWarningThreshold(this->getUnlimitedHistoryWarningThreshold());
            added.setTruncateMaxBatchSize(this->getTruncateMaxBatchSize());
            return added;
        }
 
        // Locking interface
 
        /**
         * @brief Execute function under shared (read) lock.
         * Multiple readers can hold the lock simultaneously.
         * Use this for read-only operations (queries, GUI, etc.)
         * Tracks lock acquisition time for debugging.
         */
        template <class FunctionT>
        auto
        doLocked(FunctionT&& function) const
        {
            auto startTime = std::chrono::steady_clock::now();
            std::shared_lock lock(_mutex);
            auto lockAcquiredTime = std::chrono::steady_clock::now();
 
            // Track lock wait time
            uint64_t waitNs = std::chrono::duration_cast<std::chrono::nanoseconds>(
                lockAcquiredTime - startTime).count();
            _lockStats.readLockCount.fetch_add(1, std::memory_order_relaxed);
            _lockStats.totalReadLockWaitNs.fetch_add(waitNs, std::memory_order_relaxed);
 
            // Update max wait time (using compare-exchange)
            uint64_t currentMax = _lockStats.maxReadLockWaitNs.load(std::memory_order_relaxed);
            while (waitNs > currentMax)
            {
                if (_lockStats.maxReadLockWaitNs.compare_exchange_weak(currentMax, waitNs, std::memory_order_relaxed))
                {
                    break;
                }
            }
 
            return function();
        }
 
        /**
         * @brief Execute function under exclusive (write) lock.
         * Only one writer can hold the lock, and no readers.
         * Use this for write operations (update, add, clear, etc.)
         * Tracks lock acquisition time for debugging.
         */
        template <class FunctionT>
        auto
        doLockedExclusive(FunctionT&& function)
        {
            auto startTime = std::chrono::steady_clock::now();
            std::unique_lock lock(_mutex);
            auto lockAcquiredTime = std::chrono::steady_clock::now();
 
            // Track lock wait time
            uint64_t waitNs = std::chrono::duration_cast<std::chrono::nanoseconds>(
                lockAcquiredTime - startTime).count();
            _lockStats.writeLockCount.fetch_add(1, std::memory_order_relaxed);
            _lockStats.totalWriteLockWaitNs.fetch_add(waitNs, std::memory_order_relaxed);
 
            // Update max wait time (using compare-exchange)
            uint64_t currentMax = _lockStats.maxWriteLockWaitNs.load(std::memory_order_relaxed);
            while (waitNs > currentMax)
            {
                if (_lockStats.maxWriteLockWaitNs.compare_exchange_weak(currentMax, waitNs, std::memory_order_relaxed))
                {
                    break;
                }
            }
 
            return function();
        }
 
        /// Get lock statistics for this core segment
        const CoreSegmentLockStatistics& getLockStatistics() const { return _lockStats; }
 
        /// Reset lock statistics
        void resetLockStatistics() { _lockStats.reset(); }
 
 
    private:
        mutable std::shared_mutex _mutex;
        mutable CoreSegmentLockStatistics _lockStats;
    };
 
    /// @see base::MemoryBase
    class Memory :
        public base::MemoryBase<CoreSegment, Memory>,
        public armem::wm::detail::FindInstanceDataMixin<Memory>,
        public armem::server::wm::detail::PredictionContainer<Memory>
    {
        using Base = base::MemoryBase<CoreSegment, Memory>;
 
    public:
        using Base::MemoryBase;
 
 
        /**
         * @brief Perform the commit, locking the core segments.
         *
         * Groups the commits by core segment, and updates each core segment
         * in a batch, locking the core segment.
         */
        std::vector<Base::UpdateResult> updateLocking(const Commit& commit);
 
        /**
         * @brief Update the memory, locking the updated core segment.
         */
        Base::UpdateResult updateLocking(const EntityUpdate& update);
 
        /// @see base::MemoryBase::addCoreSegment()
        using MemoryBase::addCoreSegment;
 
        template <class... Args>
        CoreSegment&
        addCoreSegment(const std::string& name, Args... args)
        {
            CoreSegmentT& added = MemoryBase::addCoreSegment(name, args...);
            return added;
        }
    };
 
} // namespace armarx::armem::server::wm