BufferedMemoryMixin.h

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <iomanip>
#include <limits>
#include <map>
#include <mutex>
#include <queue>
#include <thread>
#include <variant>
 
#include <boost/lockfree/queue.hpp>
 
#include <SimoxUtility/json.h>
 
#include <ArmarXCore/core/services/tasks/PeriodicTask.h>
#include <ArmarXCore/interface/observers/ObserverInterface.h>
#include <ArmarXCore/observers/variant/Variant.h>
 
#include <RobotAPI/libraries/armem/core/operations.h>
#include <RobotAPI/libraries/armem/core/wm/memory_definitions.h>
#include <RobotAPI/libraries/armem/core/wm/memory_conversions.h>
#include <RobotAPI/libraries/armem/server/wm/memory_definitions.h>
 
// Include PendingConversion definition (shared with MemoryBase)
#include "../PendingConversion.h"
 
namespace armarx::armem::server::ltm::detail::mixin
{
    /**
     * @brief Statistics for async storage queue operations.
     *
     * Thread-safe statistics for tracking queue performance and throughput.
     */
    struct AsyncStorageStatistics
    {
        // Queue statistics
        std::atomic<uint64_t> totalItemsEnqueued{0};
        std::atomic<uint64_t> totalItemsProcessed{0};
        std::atomic<uint64_t> totalSnapshotsStored{0};
 
        // Timing statistics (in nanoseconds)
        std::atomic<uint64_t> totalStorageTimeNs{0};
        std::atomic<uint64_t> maxStorageTimeNs{0};
        std::atomic<uint64_t> totalConversionTimeNs{0};
 
        // Backpressure statistics
        std::atomic<uint64_t> backpressureEvents{0};
 
        // Dropped snapshots due to queue being full (non-blocking drop policy)
        std::atomic<uint64_t> snapshotsDroppedBackpressure{0};
 
        // Dropped snapshots (other reasons, e.g., errors)
        std::atomic<uint64_t> snapshotsDropped{0};
 
        // Pre-filter statistics (snapshots filtered before enqueue)
        std::atomic<uint64_t> snapshotsPreFiltered{0};
        std::atomic<uint64_t> snapshotsPassedPreFilter{0};
        std::atomic<uint64_t> totalPreFilterTimeNs{0};
 
        void reset()
        {
            totalItemsEnqueued = 0;
            totalItemsProcessed = 0;
            totalSnapshotsStored = 0;
            totalStorageTimeNs = 0;
            maxStorageTimeNs = 0;
            totalConversionTimeNs = 0;
            backpressureEvents = 0;
            snapshotsDroppedBackpressure = 0;
            snapshotsDropped = 0;
            snapshotsPreFiltered = 0;
            snapshotsPassedPreFilter = 0;
            totalPreFilterTimeNs = 0;
        }
 
        double getAvgStorageTimeMs() const
        {
            uint64_t count = totalItemsProcessed.load(std::memory_order_relaxed);
            if (count == 0) return 0.0;
            return totalStorageTimeNs.load(std::memory_order_relaxed) / (count * 1e6);
        }
 
        double getMaxStorageTimeMs() const
        {
            return maxStorageTimeNs.load(std::memory_order_relaxed) / 1e6;
        }
 
        double getAvgConversionTimeMs() const
        {
            uint64_t count = totalItemsProcessed.load(std::memory_order_relaxed);
            if (count == 0) return 0.0;
            return totalConversionTimeNs.load(std::memory_order_relaxed) / (count * 1e6);
        }
 
        double getAvgPreFilterTimeMs() const
        {
            uint64_t total = snapshotsPreFiltered.load(std::memory_order_relaxed) +
                            snapshotsPassedPreFilter.load(std::memory_order_relaxed);
            if (total == 0) return 0.0;
            return totalPreFilterTimeNs.load(std::memory_order_relaxed) / (total * 1e6);
        }
 
        double getPreFilterRejectionRate() const
        {
            uint64_t filtered = snapshotsPreFiltered.load(std::memory_order_relaxed);
            uint64_t passed = snapshotsPassedPreFilter.load(std::memory_order_relaxed);
            uint64_t total = filtered + passed;
            if (total == 0) return 0.0;
            return static_cast<double>(filtered) / static_cast<double>(total);
        }
    };
 
 
    template <class _CoreSegmentT>
    class BufferedMemoryMixin
    {
    public:
        /// Can hold either pre-converted Memory or pending conversion data
        using StorageItem = std::variant<std::shared_ptr<const armem::wm::Memory>, PendingConversion>;
 
        BufferedMemoryMixin(const MemoryID& id) :
            buffer(std::make_unique<armem::wm::Memory>(id)),
            to_store(std::make_unique<armem::wm::Memory>(id)),
            storageQueue(1000), // Initialize lock-free queue with capacity
            queueSize(0),
            stopWorkerThread(false)
        {
        }
 
        virtual ~BufferedMemoryMixin()
        {
            stopAsyncStorageWorker();
 
            // Clean up any remaining items in the lock-free queue
            StorageItem* item;
            while (storageQueue.pop(item))
            {
                delete item;
            }
        }
 
        void
        directlyStore(const armem::wm::Memory& memory, bool simulatedVersion = false)
        {
            // DEADLOCK FIX: Removed storeMutex to prevent lock order inversion
            // Previously: storeMutex → ltm_mutex (in _directlyStore)
            // This could deadlock with code that holds ltm_mutex → storeMutex
            //
            // The _directlyStore() implementations are responsible for their own
            // thread safety (e.g., Memory::_directlyStore uses ltm_mutex)
 
            TIMING_START(LTM_Memory_DirectlyStore);
            _directlyStore(memory, simulatedVersion);
            TIMING_END_STREAM(LTM_Memory_DirectlyStore, ARMARX_DEBUG);
        }
 
        void
        directlyStore(const armem::server::wm::Memory& serverMemory, bool simulatedVersion = false)
        {
            armem::wm::Memory memory;
            memory.setName(serverMemory.name());
            auto ids = memory.update(armem::toCommit(serverMemory), true);
            ARMARX_DEBUG << "Amount of ids in update: " << ids.size();
            this->directlyStore(memory, simulatedVersion);
        }
 
        void
        bufferFinished()
        {
            //use this to count how much work is still left
        }
 
        /**
         * @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)
        {
            ARMARX_DEBUG << "Flushing async storage queue...";
 
            auto startTime = std::chrono::steady_clock::now();
 
            while (true)
            {
                // Lock-free check: no mutex needed for atomic operations
                if (queueSize.load(std::memory_order_acquire) == 0 &&
                    numThreadsProcessing.load(std::memory_order_acquire) == 0)
                {
                    ARMARX_DEBUG << "Async storage queue flushed successfully";
                    return true;
                }
 
                // Check timeout
                if (timeoutMs > 0)
                {
                    auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
                        std::chrono::steady_clock::now() - startTime).count();
                    if (elapsed >= timeoutMs)
                    {
                        ARMARX_WARNING << "Flush timeout after " << elapsed << "ms with "
                                       << getQueueSize() << " items still in queue";
                        return false;
                    }
                }
 
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        }
 
        /**
         * @brief Get the current size of the async storage queue
         */
        size_t
        getQueueSize() const
        {
            return queueSize.load(std::memory_order_acquire);
        }
 
        /**
         * @brief Get the async storage statistics
         */
        const AsyncStorageStatistics&
        getAsyncStorageStatistics() const
        {
            return asyncStats;
        }
 
        /**
         * @brief Reset the async storage statistics
         */
        void
        resetAsyncStorageStatistics()
        {
            asyncStats.reset();
        }
 
        /**
         * @brief Get the number of threads currently processing items
         */
        size_t
        getNumThreadsProcessing() const
        {
            return numThreadsProcessing.load(std::memory_order_acquire);
        }
 
    protected:
        void
        setMixinMemoryID(const MemoryID& id)
        {
            ARMARX_CHECK_NOT_EMPTY(id.memoryName) << " The full id was: " << id.str();
 
            buffer->id() = id.getMemoryID();
            to_store->id() = id.getMemoryID();
        }
 
        void
        start()
        {
            // Start the async storage worker thread
            startAsyncStorageWorker();
 
            // create task if not already exists
            if (!task)
            {
                int waitingTimeMs = 1000.f / storeFrequency;
                task = new armarx::PeriodicTask<BufferedMemoryMixin>(
                    this, &BufferedMemoryMixin::storeBuffer, waitingTimeMs);
                task->start();
                task->setDelayWarningTolerance(
                    waitingTimeMs); //a warning will be issued if the task takes longer than the waitingTime
            }
        }
 
        void
        stop()
        {
            if (task)
            {
                task->stop();
                task = nullptr;
            }
 
            // Stop the async storage worker thread (but don't flush yet)
            // The caller should call flushAsyncStorage() if they want to wait for completion
        }
 
        armem::wm::Memory
        getBuffer() const
        {
            std::lock_guard l(bufferMutex);
            return *buffer;
        }
 
        void
        storeBuffer()
        {
            std::shared_ptr<const armem::wm::Memory> memoryToStore;
            {
                std::lock_guard l(bufferMutex);
                to_store = std::move(buffer);
                buffer = std::make_unique<armem::wm::Memory>(to_store->id());
                // Convert unique_ptr to shared_ptr<const>
                memoryToStore = std::shared_ptr<const armem::wm::Memory>(std::move(to_store));
            }
 
            if (memoryToStore->empty())
            {
                ARMARX_DEBUG << deactivateSpam(120) << "Cannot store an empty buffer. Ignoring.";
                return;
            }
 
            // Pre-filter the memory before enqueuing to reduce queue pressure
            auto preFilterStart = std::chrono::steady_clock::now();
            uint64_t filteredCount = 0;
            uint64_t passedCount = 0;
            auto filteredMemory = _preFilterMemory(*memoryToStore, filteredCount, passedCount);
            auto preFilterEnd = std::chrono::steady_clock::now();
 
            // Update pre-filter statistics
            auto preFilterTimeNs = std::chrono::duration_cast<std::chrono::nanoseconds>(
                preFilterEnd - preFilterStart).count();
            asyncStats.snapshotsPreFiltered.fetch_add(filteredCount, std::memory_order_relaxed);
            asyncStats.snapshotsPassedPreFilter.fetch_add(passedCount, std::memory_order_relaxed);
            asyncStats.totalPreFilterTimeNs.fetch_add(preFilterTimeNs, std::memory_order_relaxed);
 
            // If all snapshots were filtered, skip enqueuing
            if (!filteredMemory || filteredMemory->empty())
            {
                ARMARX_DEBUG << deactivateSpam(10) << "All " << filteredCount
                            << " snapshots were pre-filtered, skipping enqueue";
                return;
            }
 
            // Push filtered memory to async storage queue
            enqueueForAsyncStorage(std::move(filteredMemory));
        }
 
        /// configuration
        void
        configureMixin(const nlohmann::json& json)
        {
            if (json.find("BufferedMemory.storeFrequency") != json.end())
            {
                storeFrequency = json.at("BufferedMemory.storeFrequency");
                ARMARX_INFO << "Setting store frequency from configuration json to "
                            << storeFrequency;
            }
            if (json.find("BufferedMemory.maxAsyncQueueSize") != json.end())
            {
                maxAsyncQueueSize = json.at("BufferedMemory.maxAsyncQueueSize");
                ARMARX_INFO << "Setting max async queue size from configuration json to "
                            << maxAsyncQueueSize;
            }
            if (json.find("BufferedMemory.numAsyncStorageThreads") != json.end())
            {
                numAsyncStorageThreads = json.at("BufferedMemory.numAsyncStorageThreads");
                ARMARX_INFO << "Setting number of async storage threads from configuration json to "
                            << numAsyncStorageThreads;
            }
            if (json.find("BufferedMemory.workerShutdownTimeoutSeconds") != json.end())
            {
                workerShutdownTimeoutSeconds = json.at("BufferedMemory.workerShutdownTimeoutSeconds");
                ARMARX_INFO << "Setting worker shutdown timeout from configuration json to "
                            << workerShutdownTimeoutSeconds << " seconds";
            }
        }
 
        void
        createPropertyDefinitions(PropertyDefinitionsPtr& defs, const std::string& prefix)
        {
            defs->optional(storeFrequency, prefix + "storeFrequency");
            defs->optional(maxAsyncQueueSize, prefix + "maxAsyncQueueSize");
            defs->optional(numAsyncStorageThreads, prefix + "numAsyncStorageThreads");
            defs->optional(workerShutdownTimeoutSeconds, prefix + "workerShutdownTimeoutSeconds");
        }
 
        virtual void _directlyStore(const armem::wm::Memory& memory,
                                    bool simulatedVersion = false) = 0;
 
        /**
         * @brief Pre-filter a memory object before enqueuing for async storage.
         *
         * This method applies snapshot filters BEFORE the memory enters the async queue,
         * reducing queue pressure and avoiding work on the async worker threads.
         *
         * @param memory The memory to filter
         * @param filteredCount Output: number of snapshots that were filtered out
         * @param passedCount Output: number of snapshots that passed the filter
         * @return A new memory containing only the snapshots that passed the filters,
         *         or nullptr if all snapshots were filtered out
         */
        virtual std::shared_ptr<armem::wm::Memory> _preFilterMemory(
            const armem::wm::Memory& memory,
            uint64_t& filteredCount,
            uint64_t& passedCount) = 0;
 
        void
        addToBuffer(const armem::wm::Memory& memory)
        {
            std::lock_guard l(bufferMutex);
            buffer->append(memory);
        }
 
        /**
         * @brief Public interface to enqueue memory for async storage
         * This allows MemoryBase::store() to use the async thread pool
         */
        void
        enqueueForAsyncStoragePublic(std::shared_ptr<const armem::wm::Memory> memory)
        {
            enqueueForAsyncStorage(std::move(memory));
        }
 
    protected:
        /**
         * @brief Enqueue a memory object for async storage
         */
        void
        enqueueForAsyncStorage(std::shared_ptr<const armem::wm::Memory> memory)
        {
            enqueueStorageItem(StorageItem{std::move(memory)});
        }
 
        /**
         * @brief Enqueue snapshots for deferred conversion and async storage
         * This defers the expensive toMemory() conversion to the async thread
         */
        void
        enqueuePendingConversion(PendingConversion pending)
        {
            enqueueStorageItem(StorageItem{std::move(pending)});
        }
 
    private:
        /**
         * @brief Start the async storage worker threads (thread pool)
         */
        void
        startAsyncStorageWorker()
        {
            std::lock_guard<std::mutex> lock(workerMutex);
            if (workerThreads.empty())
            {
                stopWorkerThread = false;
                size_t numThreads = std::max(size_t(1), numAsyncStorageThreads);
                workerThreads.reserve(numThreads);
                for (size_t i = 0; i < numThreads; ++i)
                {
                    workerThreads.emplace_back(&BufferedMemoryMixin::asyncStorageWorker, this, i);
                }
                ARMARX_INFO << "Async storage thread pool started with " << numThreads << " worker threads";
            }
        }
 
        /**
         * @brief Stop the async storage worker threads with configurable timeout
         */
        void
        stopAsyncStorageWorker()
        {
            {
                std::lock_guard<std::mutex> lock(workerMutex);
                if (workerThreads.empty())
                {
                    return;
                }
                stopWorkerThread = true;
            }
            queueCondition.notify_all();
 
            // Wait for threads to finish with timeout
            auto startTime = std::chrono::steady_clock::now();
            auto timeout = std::chrono::seconds(workerShutdownTimeoutSeconds);
            bool allThreadsFinished = false;
 
            // Poll until all threads exit or timeout
            while (true)
            {
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
 
                // Check if all threads have finished processing
                if (numThreadsProcessing.load(std::memory_order_acquire) == 0 &&
                    queueSize.load(std::memory_order_acquire) == 0)
                {
                    allThreadsFinished = true;
                    break;
                }
 
                // Check timeout
                auto elapsed = std::chrono::steady_clock::now() - startTime;
                if (elapsed >= timeout)
                {
                    ARMARX_WARNING << "Worker threads did not finish within timeout of "
                                   << workerShutdownTimeoutSeconds << " seconds. "
                                   << "Still processing: " << numThreadsProcessing.load()
                                   << " threads, queue size: " << queueSize.load();
                    break;
                }
            }
 
            // RACE CONDITION FIX: Hold lock during thread join to protect workerThreads vector
            {
                std::lock_guard<std::mutex> lock(workerMutex);
 
                if (allThreadsFinished)
                {
                    // Threads finished gracefully, join them
                    for (auto& thread : workerThreads)
                    {
                        if (thread.joinable())
                        {
                            thread.join();
                        }
                    }
                    ARMARX_INFO << "Async storage thread pool stopped gracefully";
                }
                else
                {
                    // Timeout occurred - detach threads to avoid blocking
                    ARMARX_WARNING << "Detaching worker threads that did not terminate in time. "
                                   << "This may indicate slow disk I/O or stuck operations.";
                    for (auto& thread : workerThreads)
                    {
                        if (thread.joinable())
                        {
                            thread.detach();
                        }
                    }
                    ARMARX_WARNING << "Async storage thread pool stopped with timeout - threads detached";
                }
 
                workerThreads.clear();
            }
        }
 
    protected:
        /**
         * @brief Internal method to enqueue any storage item (Memory or PendingConversion)
         * PERFORMANCE: Uses lock-free queue to avoid mutex contention at 50Hz commit rate
         * NON-BLOCKING: If the queue is full, the item is DROPPED immediately to prevent
         * upstream queue (e.g., RobotWriterQueue) from backing up.
         */
        void
        enqueueStorageItem(StorageItem item)
        {
            // Check queue size BEFORE allocating to avoid unnecessary allocation
            size_t currentSize = queueSize.load(std::memory_order_acquire);
            if (currentSize >= maxAsyncQueueSize)
            {
                // NON-BLOCKING DROP POLICY: Immediately drop new items when queue is full
                // This prevents upstream queues (RobotWriterQueue) from backing up
                asyncStats.backpressureEvents.fetch_add(1, std::memory_order_relaxed);
 
                // Count snapshots in the item we're about to drop
                size_t snapshotCount = countSnapshotsInItem(item);
                asyncStats.snapshotsDroppedBackpressure.fetch_add(snapshotCount, std::memory_order_relaxed);
 
                ARMARX_WARNING << deactivateSpam(1)
                               << "LTM async storage queue full (" << currentSize << "/" << maxAsyncQueueSize
                               << " items). DROPPING " << snapshotCount << " snapshots to prevent upstream backup. "
                               << "Consider increasing queue size, reducing commit rate, or enabling more aggressive filtering.";
 
                return; // Non-blocking: return immediately without waiting
            }
 
            // Allocate on heap for lock-free queue (requires pointer type)
            StorageItem* itemPtr = new StorageItem(std::move(item));
 
            // Lock-free push - should succeed since we checked size above
            // (there's a small race window but that's acceptable)
            if (!storageQueue.push(itemPtr))
            {
                // Rare case: queue filled up between size check and push
                size_t snapshotCount = countSnapshotsInItem(*itemPtr);
                asyncStats.snapshotsDroppedBackpressure.fetch_add(snapshotCount, std::memory_order_relaxed);
                ARMARX_WARNING << deactivateSpam(1) << "Lock-free queue race: dropping " << snapshotCount << " snapshots";
                delete itemPtr;
                return;
            }
 
            // Track enqueued item
            asyncStats.totalItemsEnqueued.fetch_add(1, std::memory_order_relaxed);
 
            // Atomically increment size counter
            queueSize.fetch_add(1, std::memory_order_release);
 
            // Signal waiting worker threads (use mutex only for condition variable)
            {
                std::lock_guard<std::mutex> lock(queueMutex);
            }
            queueCondition.notify_one();
        }
 
        /**
         * @brief Count the number of snapshots in a storage item
         */
        size_t
        countSnapshotsInItem(const StorageItem& item) const
        {
            size_t count = 0;
            std::visit([&count](auto&& arg) {
                using T = std::decay_t<decltype(arg)>;
                if constexpr (std::is_same_v<T, std::shared_ptr<const armem::wm::Memory>>)
                {
                    arg->forEachCoreSegment([&count](const auto& coreSegment) {
                        coreSegment.forEachProviderSegment([&count](const auto& providerSegment) {
                            providerSegment.forEachEntity([&count](const auto& entity) {
                                entity.forEachSnapshot([&count](const auto&) { count++; });
                            });
                        });
                    });
                }
                else if constexpr (std::is_same_v<T, PendingConversion>)
                {
                    count = arg.snapshots.size();
                }
            }, item);
            return count;
        }
 
        /**
         * @brief Worker thread that processes the async storage queue
         * @param threadId ID of this worker thread (for logging)
         */
        void
        asyncStorageWorker(size_t threadId)
        {
            ARMARX_INFO << "Async storage worker thread #" << threadId << " started";
 
            size_t itemsProcessed = 0;
            double totalTimeMs = 0.0;
            double minTimeMs = std::numeric_limits<double>::max();
            double maxTimeMs = 0.0;
 
            while (true)
            {
                StorageItem* itemPtr = nullptr;
                size_t queueSizeBeforePop = 0;
                bool hasItem = false;
 
                // Try lock-free pop first (fast path - no mutex)
                if (storageQueue.pop(itemPtr))
                {
                    // Got an item from the queue
                    queueSizeBeforePop = queueSize.fetch_sub(1, std::memory_order_acq_rel);
                    numThreadsProcessing.fetch_add(1, std::memory_order_release);
                    hasItem = true;
                }
                else
                {
                    // Queue is empty - wait for signal or stop
                    std::unique_lock<std::mutex> lock(queueMutex);
 
                    // Exit if stop requested and queue is still empty
                    if (stopWorkerThread.load(std::memory_order_acquire))
                    {
                        // Double-check queue is empty after acquiring lock
                        if (!storageQueue.pop(itemPtr))
                        {
                            break;
                        }
                        // Found an item, process it before exiting
                        queueSizeBeforePop = queueSize.fetch_sub(1, std::memory_order_acq_rel);
                        numThreadsProcessing.fetch_add(1, std::memory_order_release);
                        hasItem = true;
                    }
                    else
                    {
                        // Wait for notification
                        queueCondition.wait(lock, [this]() {
                            return queueSize.load(std::memory_order_acquire) > 0 ||
                                   stopWorkerThread.load(std::memory_order_acquire);
                        });
                        // Loop back to try popping again
                        continue;
                    }
                }
 
                // Store the memory (outside the lock to avoid blocking enqueue operations)
                if (hasItem)
                {
                    auto startTime = std::chrono::high_resolution_clock::now();
                    size_t snapshotCount = 0;
                    double conversionTimeMs = 0.0;
 
                    try
                    {
                        TIMING_START(LTM_AsyncStorage);
 
                        // Handle variant: either pre-converted Memory or pending conversion
                        // Note: itemPtr is a pointer to StorageItem, dereference it
                        std::visit([&](auto&& item) {
                            using T = std::decay_t<decltype(item)>;
                            if constexpr (std::is_same_v<T, std::shared_ptr<const armem::wm::Memory>>)
                            {
                                // Already converted Memory - just store it
                                item->forEachCoreSegment(
                                    [&snapshotCount](const auto& coreSegment)
                                    {
                                        coreSegment.forEachProviderSegment(
                                            [&snapshotCount](const auto& providerSegment)
                                            {
                                                providerSegment.forEachEntity(
                                                    [&snapshotCount](const auto& entity)
                                                    {
                                                        entity.forEachSnapshot(
                                                            [&snapshotCount](const auto&) { snapshotCount++; });
                                                    });
                                            });
                                    });
                                this->directlyStore(*item);
                            }
                            else if constexpr (std::is_same_v<T, PendingConversion>)
                            {
                                // Pending conversion - do toMemory() here in async thread
                                auto conversionStart = std::chrono::high_resolution_clock::now();
 
                                snapshotCount = item.snapshots.size();
                                auto memory = std::make_shared<armem::wm::Memory>(item.memoryName);
 
                                // Perform the conversion using captured metadata
                                for (const auto& snapshot : item.snapshots)
                                {
                                    const std::string& coreSegmentName = snapshot.id().coreSegmentName;
                                    const std::string& providerSegmentName = snapshot.id().providerSegmentName;
 
                                    // Find metadata for this core segment
                                    auto coreMeta = std::find_if(item.segmentMetadata.begin(), item.segmentMetadata.end(),
                                        [&coreSegmentName](const auto& meta) { return meta.coreSegmentName == coreSegmentName; });
 
                                    if (coreMeta == item.segmentMetadata.end())
                                    {
                                        ARMARX_WARNING << "Missing metadata for core segment: " << coreSegmentName;
                                        continue;
                                    }
 
                                    // Add core segment if needed
                                    if (!memory->hasCoreSegment(coreSegmentName))
                                    {
                                        memory->addCoreSegment(coreSegmentName, coreMeta->coreSegmentAronType);
                                    }
                                    auto* coreSegment = memory->findCoreSegment(coreSegmentName);
 
                                    // Add provider segment if needed
                                    if (!coreSegment->hasProviderSegment(providerSegmentName))
                                    {
                                        auto providerTypeIt = coreMeta->providerSegmentAronTypes.find(providerSegmentName);
                                        aron::type::ObjectPtr providerType = (providerTypeIt != coreMeta->providerSegmentAronTypes.end())
                                            ? providerTypeIt->second : nullptr;
                                        coreSegment->addProviderSegment(providerSegmentName, providerType);
                                    }
                                    auto* providerSegment = coreSegment->findProviderSegment(providerSegmentName);
 
                                    // Add entity and snapshot
                                    if (!providerSegment->hasEntity(snapshot.id().entityName))
                                    {
                                        providerSegment->addEntity(snapshot.id().entityName);
                                    }
                                    auto* entity = providerSegment->findEntity(snapshot.id().entityName);
                                    entity->addSnapshot(snapshot);
                                }
 
                                auto conversionEnd = std::chrono::high_resolution_clock::now();
                                conversionTimeMs = std::chrono::duration<double, std::milli>(conversionEnd - conversionStart).count();
 
                                // Now store the converted memory
                                this->directlyStore(*memory);
                            }
                        }, *itemPtr); // Dereference the pointer to get the StorageItem
 
                        TIMING_END_STREAM(LTM_AsyncStorage, ARMARX_DEBUG);
                    }
                    catch (const std::exception& e)
                    {
                        ARMARX_ERROR << "Error during async storage: " << e.what();
                    }
                    catch (...)
                    {
                        ARMARX_ERROR << "Unknown error during async storage";
                    }
 
                    // Clean up the heap-allocated item
                    delete itemPtr;
 
                    auto endTime = std::chrono::high_resolution_clock::now();
                    double durationMs = std::chrono::duration<double, std::milli>(endTime - startTime).count();
                    uint64_t durationNs = std::chrono::duration_cast<std::chrono::nanoseconds>(endTime - startTime).count();
 
                    // Update local statistics
                    itemsProcessed++;
                    totalTimeMs += durationMs;
                    minTimeMs = std::min(minTimeMs, durationMs);
                    maxTimeMs = std::max(maxTimeMs, durationMs);
                    double avgTimeMs = totalTimeMs / itemsProcessed;
 
                    // Update global async storage statistics
                    asyncStats.totalItemsProcessed.fetch_add(1, std::memory_order_relaxed);
                    asyncStats.totalSnapshotsStored.fetch_add(snapshotCount, std::memory_order_relaxed);
                    asyncStats.totalStorageTimeNs.fetch_add(durationNs, std::memory_order_relaxed);
                    if (conversionTimeMs > 0)
                    {
                        uint64_t conversionNs = static_cast<uint64_t>(conversionTimeMs * 1e6);
                        asyncStats.totalConversionTimeNs.fetch_add(conversionNs, std::memory_order_relaxed);
                    }
 
                    // Update max storage time (using compare-exchange)
                    uint64_t currentMax = asyncStats.maxStorageTimeNs.load(std::memory_order_relaxed);
                    while (durationNs > currentMax)
                    {
                        if (asyncStats.maxStorageTimeNs.compare_exchange_weak(currentMax, durationNs, std::memory_order_relaxed))
                        {
                            break;
                        }
                    }
 
                    // Decrement processing counter and get queue size (all lock-free)
                    numThreadsProcessing.fetch_sub(1, std::memory_order_release);
                    size_t remainingInQueue = queueSize.load(std::memory_order_acquire);
 
                    ARMARX_DEBUG << "Async storage completed: "
                                << "snapshots=" << snapshotCount
                                << ", time=" << std::fixed << std::setprecision(2) << durationMs << "ms"
                                << (conversionTimeMs > 0 ? " (conversion=" + std::to_string(static_cast<int>(conversionTimeMs)) + "ms)" : "")
                                << ", queue_before=" << queueSizeBeforePop
                                << ", queue_after=" << remainingInQueue
                                << " | Stats: items=" << itemsProcessed
                                << ", avg=" << std::fixed << std::setprecision(2) << avgTimeMs << "ms"
                                << ", min=" << std::fixed << std::setprecision(2) << minTimeMs << "ms"
                                << ", max=" << std::fixed << std::setprecision(2) << maxTimeMs << "ms";
                }
            }
 
            ARMARX_DEBUG << "Async storage worker thread #" << threadId << " exiting. Final stats: "
                        << "total_items=" << itemsProcessed
                        << ", total_time=" << std::fixed << std::setprecision(2) << totalTimeMs << "ms"
                        << ", avg_time=" << std::fixed << std::setprecision(2) << (itemsProcessed > 0 ? totalTimeMs / itemsProcessed : 0.0) << "ms"
                        << ", min_time=" << std::fixed << std::setprecision(2) << (itemsProcessed > 0 ? minTimeMs : 0.0) << "ms"
                        << ", max_time=" << std::fixed << std::setprecision(2) << maxTimeMs << "ms";
        }
 
 
    protected:
        /// Internal memory for data consolidated from wm to ltm (double-buffer)
        /// The to-put-to-ltm buffer (contains data in plain text)
        /// This buffer may still be filtered (e.g. snapshot filters).
        /// This means that it is not guaranteed that all data in the buffer will be stored in the ltm
        std::unique_ptr<armem::wm::Memory> buffer;
        std::unique_ptr<armem::wm::Memory> to_store;
        std::atomic_flag storeFlag = ATOMIC_FLAG_INIT;
 
        /// The frequency (Hz) to store data to the ltm
        float storeFrequency = 10;
 
        /// Maximum size of the async storage queue (default 100 items)
        size_t maxAsyncQueueSize = 1000;
 
        /// Number of worker threads for async storage (default 4)
        size_t numAsyncStorageThreads = 4;
 
        /// Timeout in seconds for worker thread shutdown (default 30 seconds)
        int workerShutdownTimeoutSeconds = 30;
 
    private:
        /// The periodic'task to store the content of the buffer to the ltm
        typename armarx::PeriodicTask<BufferedMemoryMixin>::pointer_type task = nullptr;
 
        /// A mutex to access the buffer object
        /// PERFORMANCE: Using std::mutex instead of recursive_mutex since no recursion occurs
        /// All buffer access methods (getBuffer, storeBuffer, addToBuffer) are non-recursive
        mutable std::mutex bufferMutex;
        // NOTE: storeMutex removed - was causing lock order inversion deadlock
        // _directlyStore() implementations handle their own synchronization
 
        /// Async storage queue infrastructure
        /// PERFORMANCE: Using lock-free queue to eliminate mutex contention at 50Hz commit rate
        boost::lockfree::queue<StorageItem*> storageQueue; // Lock-free queue (holds pointers)
        std::atomic<size_t> queueSize{0}; // Atomic counter for queue size
        mutable std::mutex queueMutex; // Only used for condition variable signaling, not queue access
        std::condition_variable queueCondition;
        std::vector<std::thread> workerThreads;
        std::mutex workerMutex;
        std::atomic<bool> stopWorkerThread;
        std::atomic<size_t> numThreadsProcessing{0}; // Tracks how many threads are currently processing items
 
        /// Statistics for async storage operations
        mutable AsyncStorageStatistics asyncStats;
    };
} // namespace armarx::armem::server::ltm::detail::mixin