MemoryToIceAdapter.h

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <chrono>
#include <mutex>
 
#include <RobotAPI/interface/armem/client/MemoryListenerInterface.h>
#include <RobotAPI/interface/armem/server/MemoryInterface.h>
#include <RobotAPI/libraries/armem/client/Query.h>
#include <RobotAPI/libraries/armem/core/wm/memory_definitions.h>
#include <RobotAPI/libraries/armem/server/ltm/Memory.h>
#include <RobotAPI/libraries/armem/server/wm/memory_definitions.h>
 
namespace armarx::armem::server
{
    /**
     * @brief Statistics for memory read/write operations.
     *
     * Thread-safe statistics tracking for debugging memory performance.
     * Tracks counts, rates, and timing information for commits and queries.
     */
    struct MemoryOperationStatistics
    {
        // Commit (write) statistics
        std::atomic<uint64_t> totalCommitCount{0};
        std::atomic<uint64_t> totalEntityUpdates{0};
        std::atomic<uint64_t> successfulUpdates{0};
        std::atomic<uint64_t> failedUpdates{0};
        std::atomic<double> totalCommitBlockingTimeMs{0.0};
        std::atomic<double> maxCommitBlockingTimeMs{0.0};
 
        // Query (read) statistics
        std::atomic<uint64_t> totalQueryCount{0};
        std::atomic<double> totalQueryBlockingTimeMs{0.0};
        std::atomic<double> maxQueryBlockingTimeMs{0.0};
 
        // Rate calculation helpers (protected by mutex for rate calculations)
        mutable std::mutex rateMutex;
        std::chrono::steady_clock::time_point lastRateCalculation{std::chrono::steady_clock::now()};
        uint64_t lastCommitCount{0};
        uint64_t lastQueryCount{0};
        double commitsPerSecond{0.0};
        double queriesPerSecond{0.0};
 
        /**
         * @brief Calculate and update the current rates.
         * @return Pair of (commits/sec, queries/sec)
         */
        std::pair<double, double> updateRates()
        {
            std::lock_guard<std::mutex> lock(rateMutex);
            auto now = std::chrono::steady_clock::now();
            double elapsedSec = std::chrono::duration<double>(now - lastRateCalculation).count();
 
            if (elapsedSec >= 1.0)  // Update rates at most once per second
            {
                uint64_t currentCommits = totalCommitCount.load(std::memory_order_relaxed);
                uint64_t currentQueries = totalQueryCount.load(std::memory_order_relaxed);
 
                commitsPerSecond = (currentCommits - lastCommitCount) / elapsedSec;
                queriesPerSecond = (currentQueries - lastQueryCount) / elapsedSec;
 
                lastCommitCount = currentCommits;
                lastQueryCount = currentQueries;
                lastRateCalculation = now;
            }
 
            return {commitsPerSecond, queriesPerSecond};
        }
 
        void reset()
        {
            totalCommitCount = 0;
            totalEntityUpdates = 0;
            successfulUpdates = 0;
            failedUpdates = 0;
            totalCommitBlockingTimeMs = 0.0;
            maxCommitBlockingTimeMs = 0.0;
            totalQueryCount = 0;
            totalQueryBlockingTimeMs = 0.0;
            maxQueryBlockingTimeMs = 0.0;
 
            std::lock_guard<std::mutex> lock(rateMutex);
            lastRateCalculation = std::chrono::steady_clock::now();
            lastCommitCount = 0;
            lastQueryCount = 0;
            commitsPerSecond = 0.0;
            queriesPerSecond = 0.0;
        }
    };
 
 
    /**
     * @brief Helps connecting a Memory server to the Ice interface.
     *
     * This involves conversion of ice types to C++ types as well as
     * catchin exceptions and converting them to error messages
     */
    class MemoryToIceAdapter
    {
    public:
        /// Construct an MemoryToIceAdapter from an existing Memory.
        MemoryToIceAdapter(server::wm::Memory* workingMemory = nullptr,
                           server::ltm::Memory* longtermMemory = nullptr);
 
        void setMemoryListener(client::MemoryListenerInterfacePrx memoryListenerTopic);
 
        // WRITING
        data::AddSegmentResult addSegment(const data::AddSegmentInput& input,
                                          bool addCoreSegments = false);
 
        data::AddSegmentsResult addSegments(const data::AddSegmentsInput& input,
                                            bool addCoreSegments = false);
 
 
        data::CommitResult commit(const data::Commit& commitIce, Time timeArrived);
        data::CommitResult commit(const data::Commit& commitIce);
        armem::CommitResult commit(const armem::Commit& commit);
        data::CommitResult commitLocking(const data::Commit& commitIce, Time timeArrived);
        data::CommitResult commitLocking(const data::Commit& commitIce);
        armem::CommitResult commitLocking(const armem::Commit& commit);
 
 
        // READING
        query::data::Result query(const armem::query::data::Input& input);
        client::QueryResult query(const client::QueryInput& input);
 
        /**
         * Query the LTMs of the memory server. Stored into WM if recording mode is enabled.
         * @param storeIntoWM if set the result is also stored into the wm, no matter the recording mode.
         */
        query::data::Result queryLTM(const armem::query::data::Input& input, bool storeIntoWM);
 
        armem::structure::data::GetServerStructureResult getServerStructure();
 
        // LTM LOADING AND REPLAYING
 
        /**
         * Loads all core segments and their data from the LTM.
         */
        armem::CommitResult reloadAllFromLTM();
 
        /**
         * Only load specific core segments and their data from the LTM.
         */
        armem::CommitResult reloadCoreSegmentsFromLTM(std::list<std::string>& coreSegmentname);
 
        armem::CommitResult reloadPropertyDefinedCoreSegmentsFromLTM();
 
        /**
         * Triggers a reload (@see reloadFromLTM) only if load on startup property is set.
         * Loads only the core segments defined in the property 'loadedCoreSegments'
         */
        armem::CommitResult reloadFromLTMOnStartup();
 
        // LTM STORING AND RECORDING
        dto::DirectlyStoreResult directlyStore(const dto::DirectlyStoreInput& directlStoreInput);
        dto::StartRecordResult startRecord(const dto::StartRecordInput& startRecordInput);
        dto::StopRecordResult stopRecord();
        dto::RecordStatusResult getRecordStatus();
 
        // PREDICTION
        prediction::data::PredictionResultSeq
        predict(prediction::data::PredictionRequestSeq requests);
 
        prediction::data::EngineSupportMap getAvailableEngines();
 
        /// Get statistics for memory operations (commits and queries)
        const MemoryOperationStatistics& getStatistics() const { return statistics; }
 
        /// Reset statistics counters
        void resetStatistics() { statistics.reset(); }
 
        /**
         * @brief Report all debug metrics to the debug observer.
         *
         * This should be called periodically (e.g., every second) to update
         * the debug observer with current memory performance metrics.
         */
        void reportDebugMetrics();
 
    public:
        server::wm::Memory* workingMemory;
        server::ltm::Memory* longtermMemory;
 
        client::MemoryListenerInterfacePrx memoryListenerTopic;
 
 
    private:
        armem::CommitResult _commit(const armem::Commit& commit, bool locking);
 
        /// Statistics for tracking memory operations
        MemoryOperationStatistics statistics;
    };
 
 
} // namespace armarx::armem::server