MemoryToIceAdapter.cpp

Go to the documentation of this file.

#include "MemoryToIceAdapter.h"
 
#include <iostream>
#include <list>
#include <sstream>
#include <thread>
 
#include <IceUtil/Time.h>
 
#include <ArmarXCore/core/exceptions/LocalException.h>
#include <ArmarXCore/core/exceptions/local/ExpressionException.h>
#include <ArmarXCore/core/ice_conversions/ice_conversions_templates.h>
#include <ArmarXCore/core/logging/Logging.h>
#include <ArmarXCore/core/time/ice_conversions.h>
#include <ArmarXCore/observers/variant/Variant.h>
 
#include <RobotAPI/libraries/armem/client/query/Builder.h>
#include <RobotAPI/libraries/armem/core/error.h>
#include <RobotAPI/libraries/armem/core/ice_conversions.h>
#include <RobotAPI/libraries/armem/core/wm/ice_conversions.h>
#include <RobotAPI/libraries/armem/core/wm/memory_conversions.h>
#include <RobotAPI/libraries/armem/server/ltm/detail/PendingConversion.h>
#include <RobotAPI/libraries/armem/server/ltm/persistence/MemoryPersistenceStrategy.h>
#include <RobotAPI/libraries/armem/server/ltm/persistence/rest/RestPersistence.h>
#include <RobotAPI/libraries/armem/server/wm/ice_conversions.h>
#include <RobotAPI/libraries/aron/core/Exception.h>
 
#include "query_proc/ltm/disk/ltm.h"
#include "query_proc/wm/wm.h"
 
namespace armarx::armem::server
{
    namespace
    {
        /**
         * @brief Extract metadata from WM structure for given snapshots
         *
         * This function is called OUTSIDE of any WM locks to avoid deadlocks.
         * It extracts only the metadata needed for async conversion (aron types).
         *
         * @param structure The WM structure to extract from
         * @param snapshots The snapshots to extract metadata for
         * @return Vector of segment metadata
         */
        std::vector<ltm::detail::mixin::PendingConversion::SegmentMetadata>
        extractSegmentMetadata(const wm::Memory& structure,
                              const std::vector<wm::EntitySnapshot>& snapshots)
        {
            std::vector<ltm::detail::mixin::PendingConversion::SegmentMetadata> metadata;
 
            // Build unique set of core/provider segments from snapshots
            std::map<std::string, std::set<std::string>> segmentMap; // core -> set of providers
            for (const auto& snapshot : snapshots)
            {
                segmentMap[snapshot.id().coreSegmentName].insert(snapshot.id().providerSegmentName);
            }
 
            // Extract metadata for each segment
            for (const auto& [coreSegmentName, providerNames] : segmentMap)
            {
                auto* coreStructure = structure.findCoreSegment(coreSegmentName);
                if (!coreStructure)
                {
                    ARMARX_WARNING << "Core segment not found in structure: " << coreSegmentName;
                    continue;
                }
 
                ltm::detail::mixin::PendingConversion::SegmentMetadata meta;
                meta.coreSegmentName = coreSegmentName;
                meta.coreSegmentAronType = coreStructure->aronType();
 
                for (const auto& providerName : providerNames)
                {
                    auto* providerStructure = coreStructure->findProviderSegment(providerName);
                    if (providerStructure)
                    {
                        meta.providerSegmentAronTypes[providerName] = providerStructure->aronType();
                    }
                }
 
                metadata.push_back(std::move(meta));
            }
 
            return metadata;
        }
    } // anonymous namespace
 
 
    MemoryToIceAdapter::MemoryToIceAdapter(wm::Memory* workingMemory,
                                           server::ltm::Memory* longtermMemory) :
        workingMemory(workingMemory), longtermMemory(longtermMemory)
    {
        // Both are dereferenced unconditionally on the commit path (e.g. `longtermMemory->
        // isRecording()` below, `*workingMemory` in extractSegmentMetadata). Fail here, where
        // the caller is still on the stack, rather than in a servant thread later.
        ARMARX_CHECK_NOT_NULL(workingMemory);
        ARMARX_CHECK_NOT_NULL(longtermMemory);
    }
 
    void
    MemoryToIceAdapter::setMemoryListener(client::MemoryListenerInterfacePrx memoryListener)
    {
        this->memoryListenerTopic = memoryListener;
    }
 
    // WRITING
    data::AddSegmentResult
    MemoryToIceAdapter::addSegment(const data::AddSegmentInput& input, bool addCoreSegments)
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(workingMemory);
 
        ARMARX_DEBUG << "Adding segment using MemoryToIceAdapter";
 
        data::AddSegmentResult output;
 
        server::wm::CoreSegment* coreSegment = nullptr;
        try
        {
            coreSegment = &workingMemory->getCoreSegment(input.coreSegmentName);
        }
        catch (const armem::error::MissingEntry& e)
        {
            if (addCoreSegments)
            {
                coreSegment = &workingMemory->addCoreSegment(input.coreSegmentName);
            }
            else
            {
                output.success = false;
                output.errorMessage = e.what();
                return output;
            }
        }
        ARMARX_CHECK_NOT_NULL(coreSegment);
 
        if (input.providerSegmentName.size() > 0)
        {
            coreSegment->doLockedExclusive(
                [&coreSegment, &input]()
                {
                    try
                    {
                        coreSegment->addProviderSegment(input.providerSegmentName);
                    }
                    catch (const armem::error::ContainerEntryAlreadyExists&)
                    {
                        // This is ok.
                        if (input.clearWhenExists)
                        {
                            server::wm::ProviderSegment& provider =
                                coreSegment->getProviderSegment(input.providerSegmentName);
                            provider.clear();
                        }
                    }
                });
        }
 
        armem::MemoryID segmentID;
        segmentID.memoryName = workingMemory->name();
        segmentID.coreSegmentName = input.coreSegmentName;
        segmentID.providerSegmentName = input.providerSegmentName;
 
        output.success = true;
        output.segmentID = segmentID.str();
        return output;
    }
 
    data::AddSegmentsResult
    MemoryToIceAdapter::addSegments(const data::AddSegmentsInput& input, bool addCoreSegments)
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(workingMemory);
 
        data::AddSegmentsResult output;
        for (const auto& i : input)
        {
            output.push_back(addSegment(i, addCoreSegments));
        }
        return output;
    }
 
    data::CommitResult
    MemoryToIceAdapter::commit(const data::Commit& commitIce, Time timeArrived)
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(workingMemory);
        auto handleException = [](const std::string& what)
        {
            data::CommitResult result;
            data::EntityUpdateResult& r = result.results.emplace_back();
            r.success = false;
            r.errorMessage = what;
            return result;
        };
 
        armem::Commit commit;
        try
        {
            ::armarx::armem::fromIce(commitIce, commit, timeArrived);
        }
        catch (const Ice::Exception& e)
        {
            return handleException(e.what());
        }
        catch (const std::exception& e)
        {
            // Deliberately the whole std::exception family. The clause this replaces named
            // aron::error::AronNotValidException, which nothing in the tree ever throws and which
            // is a sibling of ValueNotValidException rather than a base of it -- so a malformed
            // payload escaped the servant and reached the client as an Ice::UnknownException with
            // the diagnostic gone. What conversion really throws is ValueNotValidException, or a
            // std::out_of_range from the descriptor lookup in Aron2Descriptor() when the payload
            // is not one of the known dto types. Neither has anything in common but std::exception.
            return handleException(e.what());
        }
 
        armem::CommitResult result = this->commit(commit);
        data::CommitResult resultIce;
        toIce(resultIce, result);
 
        return resultIce;
    }
 
    data::CommitResult
    MemoryToIceAdapter::commit(const data::Commit& commitIce)
    {
        ARMARX_TRACE;
        return commit(commitIce, armem::Time::Now());
    }
 
    armem::CommitResult
    MemoryToIceAdapter::commit(const armem::Commit& commit)
    {
        ARMARX_TRACE;
        return this->_commit(commit, false);
    }
 
    data::CommitResult
    MemoryToIceAdapter::commitLocking(const data::Commit& commitIce, Time timeArrived)
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(workingMemory);
        auto handleException = [](const std::string& what)
        {
            data::CommitResult result;
            data::EntityUpdateResult& r = result.results.emplace_back();
            r.success = false;
            r.errorMessage = what;
            return result;
        };
 
        armem::Commit commit;
        try
        {
            ::armarx::armem::fromIce(commitIce, commit, timeArrived);
        }
        catch (const Ice::Exception& e)
        {
            return handleException(e.what());
        }
        catch (const std::exception& e)
        {
            // Deliberately the whole std::exception family. The clause this replaces named
            // aron::error::AronNotValidException, which nothing in the tree ever throws and which
            // is a sibling of ValueNotValidException rather than a base of it -- so a malformed
            // payload escaped the servant and reached the client as an Ice::UnknownException with
            // the diagnostic gone. What conversion really throws is ValueNotValidException, or a
            // std::out_of_range from the descriptor lookup in Aron2Descriptor() when the payload
            // is not one of the known dto types. Neither has anything in common but std::exception.
            return handleException(e.what());
        }
 
        armem::CommitResult result = this->commitLocking(commit);
        data::CommitResult resultIce;
        toIce(resultIce, result);
 
        return resultIce;
    }
 
    data::CommitResult
    MemoryToIceAdapter::commitLocking(const data::Commit& commitIce)
    {
        ARMARX_TRACE;
        return commitLocking(commitIce, armem::Time::Now());
    }
 
    armem::CommitResult
    MemoryToIceAdapter::commitLocking(const armem::Commit& commit)
    {
        ARMARX_TRACE;
        return this->_commit(commit, true);
    }
 
    armem::CommitResult
    MemoryToIceAdapter::_commit(const armem::Commit& commit, bool locking)
    {
        ARMARX_TRACE;
        TIMING_START(MemoryToIceAdapter_commit);
 
        // Start timing for this commit
        auto commitStartTime = std::chrono::steady_clock::now();
 
        IceUtil::Time startTime;
        auto debugObserver = longtermMemory->getDebugObserver();
        if (debugObserver)
        {
            startTime = IceUtil::Time::now();
        }
 
        // Update statistics - increment commit count and entity update count
        statistics.totalCommitCount.fetch_add(1, std::memory_order_relaxed);
        statistics.totalEntityUpdates.fetch_add(commit.updates.size(), std::memory_order_relaxed);
 
        std::vector<data::MemoryID> updatedIDs;
        const bool publishUpdates = bool(memoryListenerTopic);
 
        CommitResult commitResult;
        for (const EntityUpdate& update : commit.updates)
        {
            EntityUpdateResult& result = commitResult.results.emplace_back();
            try
            {
                IceUtil::Time updateStartTime;
                if (debugObserver)
                {
                    updateStartTime = IceUtil::Time::now();
                }
 
                auto updateResult =
                    locking ? workingMemory->updateLocking(update) : workingMemory->update(update);
 
                result.success = true;
                result.snapshotID = updateResult.id;
                result.arrivedTime = update.arrivedTime;
 
                // Track successful update
                statistics.successfulUpdates.fetch_add(1, std::memory_order_relaxed);
 
                if (debugObserver)
                {
                    IceUtil::Time updateEndTime = IceUtil::Time::now();
                    IceUtil::Time updateElapsed = updateEndTime - updateStartTime;
                    float updateElapsedMs = updateElapsed.toMilliSecondsDouble();
 
                    std::string channelName = workingMemory->name() + "Memory";
                    debugObserver->setDebugChannel(
                        channelName,
                        {
                            {"Memory | Commit | updateResult step [ms]", new armarx::Variant(updateElapsedMs)},
                        });
                }
 
                for (const auto& snapshot : updateResult.removedSnapshots)
                {
                    ARMARX_DEBUG << "The id " << snapshot.id() << " was removed from wm";
                }
 
                // Consolidate to ltm(s) if recording mode is CLONE_WM
                if (longtermMemory->isRecording() &&
                    longtermMemory->getRecordingMode() ==
                        server::ltm::RecordingMode::CONSOLIDATE_ALL)
                {
                    // PERFORMANCE OPTIMIZATION: Defer conversion to async thread
                    // Previously, toMemory() was called synchronously here, blocking the commit path
                    // Now we pass snapshots directly and conversion happens in async storage thread
                    //
                    // DEADLOCK FIX: Extract metadata BEFORE calling storeSnapshotsAsync to ensure
                    // we don't access WM structure while any WM locks might be held
                    IceUtil::Time storeStartTime;
                    if (debugObserver)
                    {
                        storeStartTime = IceUtil::Time::now();
                    }
 
                    // Extract metadata from WM structure (done outside any locks)
                    auto segmentMetadata = extractSegmentMetadata(*workingMemory, updateResult.updatedSnapshots);
 
                    // Store snapshots with deferred conversion (happens in async thread)
                    // This no longer accesses WM structure - all metadata is pre-extracted
                    longtermMemory->storeSnapshotsAsync(
                        longtermMemory->name(),
                        updateResult.updatedSnapshots,
                        segmentMetadata);
 
                    if (debugObserver)
                    {
                        IceUtil::Time storeEndTime = IceUtil::Time::now();
                        IceUtil::Time storeElapsed = storeEndTime - storeStartTime;
                        float storeElapsedMs = storeElapsed.toMilliSecondsDouble();
 
                        std::string channelName = workingMemory->name() + "Memory";
                        debugObserver->setDebugChannel(
                            channelName,
                            {
                                {"Memory | Commit | LTM enqueue (CONSOLIDATE_ALL) [ms]", new armarx::Variant(storeElapsedMs)},
                            });
                    }
                }
 
 
                // Consolidate to ltm(s) if recording mode is CONSOLIDATE_REMOVED
                if (longtermMemory->isRecording() &&
                    longtermMemory->getRecordingMode() ==
                        server::ltm::RecordingMode::CONSOLIDATE_REMOVED)
                {
                    // PERFORMANCE OPTIMIZATION: Defer conversion to async thread
                    // Previously, toMemory() was called synchronously here, blocking the commit path
                    // Now we pass snapshots directly and conversion happens in async storage thread
                    //
                    // DEADLOCK FIX: Extract metadata BEFORE calling storeSnapshotsAsync to ensure
                    // we don't access WM structure while any WM locks might be held
                    IceUtil::Time storeStartTime;
                    if (debugObserver)
                    {
                        storeStartTime = IceUtil::Time::now();
                    }
 
                    // Extract metadata from WM structure (done outside any locks)
                    auto segmentMetadata = extractSegmentMetadata(*workingMemory, updateResult.removedSnapshots);
 
                    // Store snapshots with deferred conversion (happens in async thread)
                    // This no longer accesses WM structure - all metadata is pre-extracted
                    longtermMemory->storeSnapshotsAsync(
                        longtermMemory->name(),
                        updateResult.removedSnapshots,
                        segmentMetadata);
 
                    if (debugObserver)
                    {
                        IceUtil::Time storeEndTime = IceUtil::Time::now();
                        IceUtil::Time storeElapsed = storeEndTime - storeStartTime;
                        float storeElapsedMs = storeElapsed.toMilliSecondsDouble();
 
                        std::string channelName = workingMemory->name() + "Memory";
                        debugObserver->setDebugChannel(
                            channelName,
                            {
                                {"Memory | Commit | LTM enqueue (CONSOLIDATE_REMOVED) [ms]", new armarx::Variant(storeElapsedMs)},
                            });
                    }
                }
 
                if (longtermMemory->isRecording() &&
                    longtermMemory->getRecordingMode() ==
                        server::ltm::RecordingMode::CONSOLIDATE_LATEST)
                {
                    ARMARX_WARNING << deactivateSpam() << "THIS IS NOT IMPLEMENTED YET!!!";
                }
 
                if (publishUpdates)
                {
                    data::MemoryID& id = updatedIDs.emplace_back();
                    toIce(id, result.snapshotID);
                }
            }
            catch (const error::ArMemError& e)
            {
                result.success = false;
                result.errorMessage = e.what();
                statistics.failedUpdates.fetch_add(1, std::memory_order_relaxed);
            }
            catch (const aron::error::AronException& e)
            {
                result.success = false;
                result.errorMessage = e.what();
                statistics.failedUpdates.fetch_add(1, std::memory_order_relaxed);
            }
            catch (const Ice::Exception& e)
            {
                result.success = false;
                result.errorMessage = e.what();
                statistics.failedUpdates.fetch_add(1, std::memory_order_relaxed);
            }
            catch (...)
            {
                // The three handlers above all report the failure; this one used to only log, so
                // an unrecognised exception left result.success at the true set after the working
                // memory update and the client was told the commit had gone through. Nothing here
                // is specific to LTM consolidation either -- the try covers the WM update too.
                result.success = false;
                result.errorMessage = GetHandledExceptionString();
                ARMARX_ERROR << "Unhandled exception during commit: " << result.errorMessage;
                statistics.failedUpdates.fetch_add(1, std::memory_order_relaxed);
            }
        }
 
        if (publishUpdates)
        {
            memoryListenerTopic->memoryUpdated(updatedIDs);
        }
 
        // Calculate commit blocking time
        auto commitEndTime = std::chrono::steady_clock::now();
        double commitBlockingTimeMs = std::chrono::duration<double, std::milli>(commitEndTime - commitStartTime).count();
 
        // Update blocking time statistics (using compare-exchange for max)
        {
            double currentTotal = statistics.totalCommitBlockingTimeMs.load(std::memory_order_relaxed);
            statistics.totalCommitBlockingTimeMs.store(currentTotal + commitBlockingTimeMs, std::memory_order_relaxed);
 
            double currentMax = statistics.maxCommitBlockingTimeMs.load(std::memory_order_relaxed);
            while (commitBlockingTimeMs > currentMax)
            {
                if (statistics.maxCommitBlockingTimeMs.compare_exchange_weak(currentMax, commitBlockingTimeMs, std::memory_order_relaxed))
                {
                    break;
                }
            }
        }
 
        if (debugObserver)
        {
            IceUtil::Time endTime = IceUtil::Time::now();
            IceUtil::Time elapsed = endTime - startTime;
            float elapsedMs = elapsed.toMilliSecondsDouble();
 
            // Calculate current rates
            auto [commitsPerSec, queriesPerSec] = statistics.updateRates();
 
            const size_t queueSize = longtermMemory->getAsyncQueueSize();
 
            std::string channelName = workingMemory->name() + "Memory";
            debugObserver->setDebugChannel(
                channelName,
                {
                    // Timing metrics
                    {"Memory | Commit | t blocked [ms]", new armarx::Variant(elapsedMs)},
                    {"Memory | Commit | max blocked [ms]", new armarx::Variant(static_cast<float>(statistics.maxCommitBlockingTimeMs.load()))},
                    // Rate metrics
                    {"Memory | writes/sec", new armarx::Variant(static_cast<float>(commitsPerSec))},
                    {"Memory | reads/sec", new armarx::Variant(static_cast<float>(queriesPerSec))},
                    // Count metrics
                    {"Memory | total commits", new armarx::Variant(static_cast<int>(statistics.totalCommitCount.load()))},
                    {"Memory | total entity updates", new armarx::Variant(static_cast<int>(statistics.totalEntityUpdates.load()))},
                    {"Memory | successful updates", new armarx::Variant(static_cast<int>(statistics.successfulUpdates.load()))},
                    {"Memory | failed updates", new armarx::Variant(static_cast<int>(statistics.failedUpdates.load()))},
                    // Queue metrics
                    {"Memory | LTM async queue size", new armarx::Variant(static_cast<int>(queueSize))},
                });
        }
 
        TIMING_END_STREAM(MemoryToIceAdapter_commit, ARMARX_DEBUG);
        return commitResult;
    }
 
    // READING
    armem::query::data::Result
    MemoryToIceAdapter::query(const armem::query::data::Input& input)
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(workingMemory);
        ARMARX_CHECK_NOT_NULL(longtermMemory);
 
        // Start timing for this query
        auto queryStartTime = std::chrono::steady_clock::now();
 
        // Update statistics - increment query count
        statistics.totalQueryCount.fetch_add(1, std::memory_order_relaxed);
 
        // Core segment processors will aquire the core segment locks.
        query_proc::wm_server::MemoryQueryProcessor wmServerProcessor(
            armem::query::boolToDataMode(input.withData));
        armem::wm::Memory wmResult = wmServerProcessor.process(input, *workingMemory);
 
        armem::query::data::Result result;
 
 
        result.memory = armarx::toIce<data::MemoryPtr>(wmResult);
 
        result.success = true;
        if (result.memory->coreSegments.size() == 0)
        {
            ARMARX_DEBUG << "No data in memory found after query.";
        }
 
        // Calculate query blocking time
        auto queryEndTime = std::chrono::steady_clock::now();
        double queryBlockingTimeMs = std::chrono::duration<double, std::milli>(queryEndTime - queryStartTime).count();
 
        // Update blocking time statistics (using compare-exchange for max)
        {
            double currentTotal = statistics.totalQueryBlockingTimeMs.load(std::memory_order_relaxed);
            statistics.totalQueryBlockingTimeMs.store(currentTotal + queryBlockingTimeMs, std::memory_order_relaxed);
 
            double currentMax = statistics.maxQueryBlockingTimeMs.load(std::memory_order_relaxed);
            while (queryBlockingTimeMs > currentMax)
            {
                if (statistics.maxQueryBlockingTimeMs.compare_exchange_weak(currentMax, queryBlockingTimeMs, std::memory_order_relaxed))
                {
                    break;
                }
            }
        }
 
        // Report to debug observer if available
        auto debugObserver = longtermMemory->getDebugObserver();
        if (debugObserver)
        {
            std::string channelName = workingMemory->name() + "Memory";
            debugObserver->setDebugChannel(
                channelName,
                {
                    {"Memory | Query | t blocked [ms]", new armarx::Variant(static_cast<float>(queryBlockingTimeMs))},
                    {"Memory | Query | max blocked [ms]", new armarx::Variant(static_cast<float>(statistics.maxQueryBlockingTimeMs.load()))},
                    {"Memory | total queries", new armarx::Variant(static_cast<int>(statistics.totalQueryCount.load()))},
                });
        }
 
        return result;
    }
 
    armem::query::data::Result
    MemoryToIceAdapter::queryLTM(const armem::query::data::Input& input, bool storeIntoWM)
    {
        ARMARX_TRACE;
 
        query_proc::ltm_server::MemoryQueryProcessor ltmProcessor;
        armem::wm::Memory ltmResult = ltmProcessor.process(input, *longtermMemory);
 
        // convert memory ==> meaning resolving references
        // upon query, the LTM only returns a structure of the data (memory without data)
        if (input.withData)
        {
            longtermMemory->resolve(ltmResult);
        }
 
        if (longtermMemory->isRecording() || storeIntoWM)
        {
            this->commit(toCommit(ltmResult));
 
            // mark removed entries of wm in viewer
            // TODO
        }
 
        armem::query::data::Result result;
 
        result.memory = armarx::toIce<data::MemoryPtr>(ltmResult);
 
        result.success = true;
        if (result.memory->coreSegments.size() == 0)
        {
            ARMARX_DEBUG << "No data in ltm found after query.";
        }
 
        return result;
    }
 
    client::QueryResult
    MemoryToIceAdapter::query(const client::QueryInput& input)
    {
        ARMARX_TRACE;
 
        return client::QueryResult::fromIce(query(input.toIce()));
    }
 
    armem::structure::data::GetServerStructureResult
    MemoryToIceAdapter::getServerStructure()
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(workingMemory);
        ARMARX_CHECK_NOT_NULL(longtermMemory);
 
        armem::structure::data::GetServerStructureResult ret;
        ret.success = true;
 
        wm::Memory structure;
        structure.id() = workingMemory->id();
 
        // Get all info from the WM
        client::QueryBuilder builder(armem::query::DataMode::NoData);
        builder.all();
 
        auto query_result = this->query(builder.buildQueryInput());
        if (query_result.success)
        {
            structure.append(query_result.memory);
        }
 
        // Get all info from the LTM
        structure.append(longtermMemory->loadAllReferences());
 
        ret.serverStructure = armarx::toIce<data::MemoryPtr>(structure);
 
        return ret;
    }
 
    armem::CommitResult
    MemoryToIceAdapter::reloadAllFromLTM()
    {
        ARMARX_TRACE;
 
        ARMARX_INFO << "Reloading of all core segments from LTM into WM triggered";
 
        int maxAmountOfSnapshots = this->longtermMemory->p.maxAmountOfSnapshotsLoaded;
        //create WM and load latest references
        armem::wm::Memory m;
        this->longtermMemory->loadLatestNReferences(maxAmountOfSnapshots, m);
 
        //construct a commit of the loaded data and commit it to the working memory
        auto com = armem::toCommit(m);
        auto res = this->commit(com);
 
        //the CommitResult contains some information which might be helpful:
        return res;
    }
 
    armem::CommitResult
    MemoryToIceAdapter::reloadCoreSegmentsFromLTM(std::list<std::string>& coreSegmentNames)
    {
        ARMARX_TRACE;
 
        ARMARX_INFO << "Reloading of specific core segments from LTM into WM triggered";
 
        std::ostringstream namesStr;
        for (auto it = coreSegmentNames.begin(); it != coreSegmentNames.end(); ++it)
        {
            if (it != coreSegmentNames.begin())
                namesStr << ", "; // Add comma before every element except the first
            namesStr << *it;
        }
 
        ARMARX_INFO << "Loading core segments=" << namesStr.str();
 
 
        int maxAmountOfSnapshots = this->longtermMemory->p.maxAmountOfSnapshotsLoaded;
        //create WM and load latest references
        armem::wm::Memory m;
        this->longtermMemory->loadLatestNReferences(maxAmountOfSnapshots, m, coreSegmentNames);
 
        //construct a commit of the loaded data and commit it to the working memory
        auto com = armem::toCommit(m);
        auto res = this->commit(com);
 
        //the CommitResult contains some information which might be helpful:
        return res;
    }
 
    armem::CommitResult
    MemoryToIceAdapter::reloadPropertyDefinedCoreSegmentsFromLTM()
    {
        ARMARX_INFO << "Reloading of coresegment defined in 'loadedCoreSegments' from LTM into WM "
                       "on startup triggered";
 
        auto coreNames = this->longtermMemory->p.coreSegmentsToLoad;
 
        ARMARX_INFO << "Loading core segments=" << coreNames
                    << " defined in property 'loadedCoreSegments'";
 
        //convert string to list of names:
        std::list<std::string> names;
        std::stringstream ss(coreNames);
        std::string item;
 
        while (std::getline(ss, item, ','))
        {
            names.push_back(item);
        }
 
        return reloadCoreSegmentsFromLTM(names);
    }
 
    // WM LOADING FROM LTM
    armem::CommitResult
    MemoryToIceAdapter::reloadFromLTMOnStartup()
    {
        ARMARX_TRACE;
 
        ARMARX_INFO << "Reloading of data from LTM into WM on startup triggered";
 
        if (this->longtermMemory->p.importOnStartUp)
        {
            return reloadPropertyDefinedCoreSegmentsFromLTM();
        }
        else
        {
            ARMARX_INFO << "Not loading initial data from LTM due to importOnStartup being "
                        << this->longtermMemory->p.importOnStartUp;
            armem::CommitResult r;
            return r;
        }
    }
 
    // LTM STORING AND RECORDING
    dto::DirectlyStoreResult
    MemoryToIceAdapter::directlyStore(const dto::DirectlyStoreInput& directlStoreInput)
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(longtermMemory);
 
        dto::DirectlyStoreResult output;
        output.success = true;
 
        armem::wm::Memory m = armarx::fromIce<armem::wm::Memory>(directlStoreInput.memory);
        longtermMemory->directlyStore(m);
 
        return output;
    }
 
    dto::StartRecordResult
    MemoryToIceAdapter::startRecord(const dto::StartRecordInput& startRecordInput)
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(longtermMemory);
        ARMARX_IMPORTANT << "Enabling the recording of memory " << longtermMemory->id().str();
        longtermMemory->startRecording();
 
        dto::StartRecordResult ret;
        ret.success = true;
 
        return ret;
    }
 
    dto::StopRecordResult
    MemoryToIceAdapter::stopRecord()
    {
        ARMARX_TRACE;
        ARMARX_CHECK_NOT_NULL(longtermMemory);
        ARMARX_CHECK_NOT_NULL(workingMemory);
        ARMARX_IMPORTANT << "Disabling the recording of memory " << longtermMemory->id().str();
 
        if (longtermMemory->p.storeOnStop)
        { //if true this means when stopping LTM recording leftover snapshots are transferred to WM using the simulated consolidation
            ARMARX_INFO << "Starting to save left-over WM data into LTM";
            longtermMemory->directlyStore(*workingMemory, true);
            ARMARX_INFO << "Stored leftover WM data into LTM";
        }
        else
        {
            ARMARX_INFO << "Not storing WM data into LTM on stop, because storeOnStop is "
                        << longtermMemory->p.storeOnStop;
        }
 
        // Stop the recording (stops the periodic buffer task)
        longtermMemory->stopRecording();
 
        // Use the new async flush mechanism to ensure all queued data is written
        // This runs in a separate thread to avoid blocking the caller
        auto ltm = longtermMemory;
        std::thread flushThread(
            [ltm]()
            {
                ARMARX_INFO << "Flushing async storage queue...";
                bool success = ltm->flushAsyncStorage(0); // 0 = wait indefinitely
                if (success)
                {
                    ARMARX_INFO << "All pending data stored successfully";
                }
                else
                {
                    ARMARX_WARNING << "Flush completed with timeout or errors";
                }
                ltm->bufferFinished();
            });
        flushThread.detach();
 
        ARMARX_IMPORTANT
            << "Stopped all LTM recordings, flushing async queue in background. "
            << "Please wait with stopping the component until all files are written";
 
        dto::StopRecordResult ret;
        ret.success = true;
 
        return ret;
    }
 
    dto::RecordStatusResult
    MemoryToIceAdapter::getRecordStatus()
    {
        dto::RecordStatusResult ret;
        ret.success = true;
 
        long savedSnapshots;
        long totalSnapshots;
 
        ARMARX_DEBUG << "Get record status";
 
        longtermMemory->forEachCoreSegment(
            [&savedSnapshots, &totalSnapshots](const auto& c)
            {
                c.forEachProviderSegment(
                    [&savedSnapshots, &totalSnapshots](const auto& p)
                    {
                        p.forEachEntity(
                            [&savedSnapshots, &totalSnapshots](const auto& e)
                            {
                                savedSnapshots += e.getStatistics().recordedSnapshots;
 
                                e.forEachSnapshot([&totalSnapshots](const auto&)
                                                  { totalSnapshots++; });
                            });
                    });
            });
 
        ret.status.savedSnapshots = savedSnapshots;
        ret.status.totalSnapshots = totalSnapshots;
 
        return ret;
    }
 
    // PREDICTION
    prediction::data::PredictionResultSeq
    MemoryToIceAdapter::predict(prediction::data::PredictionRequestSeq requests)
    {
        auto res = workingMemory->dispatchPredictions(
            armarx::fromIce<std::vector<PredictionRequest>>(requests));
        return armarx::toIce<prediction::data::PredictionResultSeq>(res);
    }
 
    prediction::data::EngineSupportMap
    MemoryToIceAdapter::getAvailableEngines()
    {
        prediction::data::EngineSupportMap result;
        armarx::toIce(result, workingMemory->getAllPredictionEngines());
 
        // Uncomment once LTM also supports prediction engines.
 
        /*prediction::data::EngineSupportMap ltmMap;
        armarx::toIce(ltmMap, longtermMemory->getAllPredictionEngines());
        for (const auto& [memoryID, engines] : ltmMap)
        {
            auto entryIter = result.find(memoryID);
            if (entryIter == result.end())
            {
                result.emplace(memoryID, engines);
            }
            else
            {
                // Merge LTM-supported engines with WM-supported engines, removing duplicates
                std::set<prediction::data::PredictionEngine> engineSet;
                engineSet.insert(entryIter->second.begin(), entryIter->second.end());
                engineSet.insert(engines.begin(), engines.end());
                entryIter->second.assign(engineSet.begin(), engineSet.end());
            }
        }*/
 
        return result;
    }
 
    void
    MemoryToIceAdapter::reportDebugMetrics()
    {
        auto debugObserver = longtermMemory->getDebugObserver();
        if (!debugObserver)
        {
            return;
        }
 
        // Calculate current rates
        auto [commitsPerSec, queriesPerSec] = statistics.updateRates();
 
        // Queue size and async stats from the LTM.
        const size_t queueSize = longtermMemory->getAsyncQueueSize();
        const size_t numThreadsProcessing = longtermMemory->getNumThreadsProcessing();
        const auto& asyncStats = longtermMemory->getAsyncStorageStatistics();
        const uint64_t asyncItemsEnqueued =
            asyncStats.totalItemsEnqueued.load(std::memory_order_relaxed);
        const uint64_t asyncItemsProcessed =
            asyncStats.totalItemsProcessed.load(std::memory_order_relaxed);
        const uint64_t asyncSnapshotsStored =
            asyncStats.totalSnapshotsStored.load(std::memory_order_relaxed);
        // Not snapshotsDropped on its own: that counter covers only the error path, so while
        // the queue was shedding load under backpressure this channel read zero throughout.
        const uint64_t asyncSnapshotsDropped = asyncStats.totalSnapshotsDropped();
        const uint64_t asyncSnapshotsDroppedBackpressure =
            asyncStats.snapshotsDroppedBackpressure.load(std::memory_order_relaxed);
        const uint64_t asyncBackpressureEvents =
            asyncStats.backpressureEvents.load(std::memory_order_relaxed);
        const double asyncAvgStorageTimeMs = asyncStats.getAvgStorageTimeMs();
        const double asyncMaxStorageTimeMs = asyncStats.getMaxStorageTimeMs();
 
        // Calculate average blocking times
        uint64_t commitCount = statistics.totalCommitCount.load(std::memory_order_relaxed);
        uint64_t queryCount = statistics.totalQueryCount.load(std::memory_order_relaxed);
        double avgCommitBlockingMs = commitCount > 0
            ? statistics.totalCommitBlockingTimeMs.load(std::memory_order_relaxed) / commitCount
            : 0.0;
        double avgQueryBlockingMs = queryCount > 0
            ? statistics.totalQueryBlockingTimeMs.load(std::memory_order_relaxed) / queryCount
            : 0.0;
 
        std::string channelName = workingMemory->name() + "Memory";
        debugObserver->setDebugChannel(
            channelName,
            {
                // Rate metrics
                {"Memory | writes/sec", new armarx::Variant(static_cast<float>(commitsPerSec))},
                {"Memory | reads/sec", new armarx::Variant(static_cast<float>(queriesPerSec))},
                // Count metrics
                {"Memory | total commits", new armarx::Variant(static_cast<int>(statistics.totalCommitCount.load()))},
                {"Memory | total queries", new armarx::Variant(static_cast<int>(statistics.totalQueryCount.load()))},
                {"Memory | total entity updates", new armarx::Variant(static_cast<int>(statistics.totalEntityUpdates.load()))},
                {"Memory | successful updates", new armarx::Variant(static_cast<int>(statistics.successfulUpdates.load()))},
                {"Memory | failed updates", new armarx::Variant(static_cast<int>(statistics.failedUpdates.load()))},
                // Timing metrics
                {"Memory | Commit | avg blocked [ms]", new armarx::Variant(static_cast<float>(avgCommitBlockingMs))},
                {"Memory | Commit | max blocked [ms]", new armarx::Variant(static_cast<float>(statistics.maxCommitBlockingTimeMs.load()))},
                {"Memory | Query | avg blocked [ms]", new armarx::Variant(static_cast<float>(avgQueryBlockingMs))},
                {"Memory | Query | max blocked [ms]", new armarx::Variant(static_cast<float>(statistics.maxQueryBlockingTimeMs.load()))},
                // Async storage queue metrics
                {"Memory | LTM async queue size", new armarx::Variant(static_cast<int>(queueSize))},
                {"Memory | LTM threads processing", new armarx::Variant(static_cast<int>(numThreadsProcessing))},
                {"Memory | LTM items enqueued", new armarx::Variant(static_cast<int>(asyncItemsEnqueued))},
                {"Memory | LTM items processed", new armarx::Variant(static_cast<int>(asyncItemsProcessed))},
                {"Memory | LTM snapshots stored", new armarx::Variant(static_cast<int>(asyncSnapshotsStored))},
                {"Memory | LTM snapshots dropped", new armarx::Variant(static_cast<int>(asyncSnapshotsDropped))},
                {"Memory | LTM snapshots dropped (backpressure)", new armarx::Variant(static_cast<int>(asyncSnapshotsDroppedBackpressure))},
                {"Memory | LTM backpressure events", new armarx::Variant(static_cast<int>(asyncBackpressureEvents))},
                {"Memory | LTM avg storage [ms]", new armarx::Variant(static_cast<float>(asyncAvgStorageTimeMs))},
                {"Memory | LTM max storage [ms]", new armarx::Variant(static_cast<float>(asyncMaxStorageTimeMs))},
            });
    }
 
} // namespace armarx::armem::server