BufferedMemoryMixin.h

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <chrono>
#include <thread>
 
#include <SimoxUtility/json.h>
 
#include <ArmarXCore/core/services/tasks/PeriodicTask.h>
 
#include <RobotAPI/libraries/armem/core/operations.h>
#include <RobotAPI/libraries/armem/core/wm/memory_definitions.h>
#include <RobotAPI/libraries/armem/server/wm/memory_definitions.h>
 
namespace armarx::armem::server::ltm::detail::mixin
{
    template <class _CoreSegmentT>
    class BufferedMemoryMixin
    {
 
    public:
        BufferedMemoryMixin(const MemoryID& id) :
            buffer(std::make_unique<armem::wm::Memory>(id)),
            to_store(std::make_unique<armem::wm::Memory>(id))
        {
        }
 
        virtual ~BufferedMemoryMixin() = default;
 
        void
        directlyStore(const armem::wm::Memory& memory, bool simulatedVersion = false)
        {
            std::lock_guard l(storeMutex);
 
            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
        }
 
    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()
        {
            // 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;
            }
        }
 
        armem::wm::Memory
        getBuffer() const
        {
            std::lock_guard l(bufferMutex);
            return *buffer;
        }
 
        void
        storeBuffer()
        {
            std::lock_guard l(storeMutex);
            {
                std::lock_guard l(bufferMutex);
                to_store = std::move(buffer);
                buffer = std::make_unique<armem::wm::Memory>(to_store->id());
            }
 
            if (to_store->empty())
            {
                ARMARX_DEBUG << deactivateSpam(120) << "Cannot store an empty buffer. Ignoring.";
                return;
            }
 
            while (storeFlag.test_and_set(std::memory_order_acquire))
            {
                std::this_thread::yield();
            }
            this->directlyStore(*to_store);
            storeFlag.clear(std::memory_order_release);
        }
 
        /// 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;
            }
        }
 
        void
        createPropertyDefinitions(PropertyDefinitionsPtr& defs, const std::string& prefix)
        {
            defs->optional(storeFrequency, prefix + "storeFrequency");
        }
 
        virtual void _directlyStore(const armem::wm::Memory& memory,
                                    bool simulatedVersion = false) = 0;
 
        void
        addToBuffer(const armem::wm::Memory& memory)
        {
            std::lock_guard l(bufferMutex);
            buffer->append(memory);
        }
 
 
    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;
 
    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
        mutable std::recursive_mutex bufferMutex;
        mutable std::recursive_mutex storeMutex;
    };
} // namespace armarx::armem::server::ltm::detail::mixin