stable_reference_buffer.hpp

Go to the documentation of this file.

/*
Copyright (c) 2025 acrion innovations GmbH
Authors: Stefan Zipproth, s.zipproth@acrion.ch
 
This file is part of Cbeam, see https://github.com/acrion/cbeam and https://cbeam.org
 
Cbeam is offered under a commercial and under the AGPL license.
For commercial licensing, contact us at https://acrion.ch/sales. For AGPL licensing, see below.
 
AGPL licensing:
 
Cbeam is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
 
Cbeam is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
 
You should have received a copy of the GNU Affero General Public License
along with Cbeam. If not, see <https://www.gnu.org/licenses/>.
*/
 
#pragma once
 
#include <cbeam/concurrency/process.hpp>               // for cbeam::concurrency::get_current_process_id
#include <cbeam/container/buffer.hpp>                  // for cbeam::container::buffer
#include <cbeam/container/stable_interprocess_map.hpp> // for cbeam::container::stable_interprocess_map
#include <cbeam/convert/string.hpp>                    // for cbeam::convert::to_string
#include <cbeam/error/logic_error.hpp>                 // for cbeam::error::logic_error
#include <cbeam/error/runtime_error.hpp>               // for cbeam::error::runtime_error
#include <cbeam/lifecycle/singleton.hpp>               // for cbeam::lifecycle::singleton
#include <cbeam/logging/log_manager.hpp>               // for CBEAM_LOG_DEBUG, CBEAM_LOG
 
#include <stdlib.h> // for free
 
#include <cassert> // for assert
#include <cstddef> // for size_t, std::size_t
#include <cstdint> // for uint8_t
 
#include <memory>    // for std::shared_ptr, std::__shared_ptr_access, std::allocator
#include <set>       // for std::operator!=, std::set
#include <string>    // for std::operator+, std::to_string, std::char_traits, std::basic_string, std::string
 
namespace cbeam::container
{
    class stable_reference_buffer : public buffer
    {
    public:
        class delay_deallocation
        {
        public:
            delay_deallocation()
            {
                auto lock = _use_count->get_lock_guard();
                _use_count->foreach ([this](auto it)
                                     { _old_entries.insert(it.first); return true; });
                _use_count->insert(nullptr, get_initial_use_count() + 1);
                // CBEAM_LOG_DEBUG("Incrementing initial useCount to " + std::to_string(get_initial_use_count()) + " to delay deallocation of memory created in this scope");
            }

            ~delay_deallocation() noexcept
            {
                auto lock = _use_count->get_lock_guard();
                bool done;
                do
                {
                    done = false;
                    _use_count->foreach ([this, &done](auto it)
                                         {
                        if (_old_entries.count(it.first) == 0)
                        {
                            if (--it.second == 0)
                            {
                                CBEAM_LOG_DEBUG("cbeam::container::container::stable_reference_buffer: Deallocating " + convert::to_string((void*)it.first) + " in context of destruction of class delay_deallocation");
                                free(it.first);
                                _use_count->erase(it.first);
                                done = true;
                                return false;
                            }
                            else
                            {
                                _old_entries.insert(it.first); // protect this entry in case done is set to true and we perform a second scan
                                CBEAM_LOG_DEBUG("cbeam::container::container::stable_reference_buffer: Removed reference to " + convert::to_string((void*)it.first) + " (" + std::to_string(it.second) + " left) in context of destruction of class delay_deallocation");
                            }
 
                            return true;
                        }
 
                        return true; });
                } while (done);
                _use_count->insert(nullptr, get_initial_use_count() - 1);
                // CBEAM_LOG_DEBUG("Resetting initial useCount to " + std::to_string(get_initial_use_count()) + " to leave scope of delayed deallocation");
            }
 
        private:
            std::set<uint8_t*>                                      _old_entries;
            std::shared_ptr<stable_interprocess_map<uint8_t*, int>> _use_count{get_use_count()};
        };
 
        stable_reference_buffer() 
            : buffer()
        {
        }

        stable_reference_buffer(const std::size_t size, const size_t size_of_type = 1)
            : buffer(size, size_of_type)
        {
            _use_count->update_or_insert(
                _buffer, [](int& count)
                { ++count; },
                get_initial_use_count());
 
            CBEAM_LOG_DEBUG("cbeam::container::container::stable_reference_buffer: Allocated   " + convert::to_string((void*)_buffer) + " with useCount=" + std::to_string(get_initial_use_count()));
        }
 
        stable_reference_buffer(const buffer& base)
            : buffer(base)
        {
            _use_count->update_or_insert(
                _buffer, [](int& count)
                { ++count; },
                get_initial_use_count());
 
            CBEAM_LOG_DEBUG("cbeam::container::container::stable_reference_buffer: Allocated   " + convert::to_string((void*)_buffer) + " with useCount=" + std::to_string(get_initial_use_count()));
        }

        ~stable_reference_buffer() noexcept override
        {
            stable_reference_buffer::reset();
        }

        explicit stable_reference_buffer(const void* address)
        {
            _buffer = (uint8_t*)address;
 
            _use_count->update(
                _buffer, [](int& count)
                { ++count; },
                "cbeam::container::container::stable_reference_buffer: memory address " + convert::to_string((void*)address) + " was not created by cbeam::stable_reference_buffer");
 
            CBEAM_LOG_DEBUG("cbeam::container::container::stable_reference_buffer: " + std::to_string(_use_count->at(_buffer)) + ". reference to " + convert::to_string((void*)_buffer) + " (added from raw pointer)");
        }

        stable_reference_buffer(const stable_reference_buffer& other)
        {
            if (other._buffer == nullptr)
            {
                throw cbeam::error::runtime_error("cbeam::container::container::stable_reference_buffer: copy constructor has been passed a default constructed (therefore invalid) instance of Memory");
            }
 
            _buffer = other._buffer;
 
            _use_count->update(_buffer, [this, &other](int& count)
                               { ++count; });
 
            CBEAM_LOG_DEBUG("cbeam::container::container::stable_reference_buffer: " + std::to_string(_use_count->at(_buffer)) + ". reference to " + convert::to_string((void*)_buffer) + " (added from copy constructor)");
        }

        void append(const void* bufferToAppend, const size_t lengthOfBuffer) override
        {
            auto lock = _use_count->get_lock_guard();
 
            if (_size == 0 && _buffer != nullptr)
            {
                throw cbeam::error::logic_error("cbeam::container::container::stable_reference_buffer::append: This instance was created from a raw pointer without knowledge of its buffer length, so appending is not possible.");
            }
 
            uint8_t* old_buffer = _buffer;
 
            buffer::append(bufferToAppend, lengthOfBuffer);
 
            if (_buffer != old_buffer)
            {
                if (old_buffer)
                {
                    _use_count->insert(_buffer, _use_count->at(old_buffer));
                    _use_count->erase(old_buffer);
                }
                else
                {
                    _use_count->insert(_buffer, get_initial_use_count());
                }
            }
        }

        void* safe_get() const noexcept
        {
            if (use_count() <= 1)
            {
                CBEAM_LOG("Error: Attempt to access the raw pointer via cbeam::container::stable_reference_buffer::safe_get() without adequate reference count. This operation is blocked and returns a nullptr to prevent unsafe memory access, as the buffer could be invalidated upon destruction of this instance.");
                return nullptr;
            }
 
            return get();
        }

        stable_reference_buffer& operator=(const stable_reference_buffer& other)
        {
            if (this != &other)
            {
                if (other._buffer == nullptr)
                {
                    throw cbeam::error::runtime_error("cbeam::container::container::stable_reference_buffer: copy assignment operator has been passed a default constructed (therefore invalid) instance of Memory");
                }
 
                auto lock = _use_count->get_lock_guard();
                reset();
 
                assert(is_known(other._buffer));
                _buffer = other._buffer;
                _use_count->update(_buffer, [this, &other](auto& count)
                                   { ++count; });
                CBEAM_LOG_DEBUG("cbeam::container::container::stable_reference_buffer: " + std::to_string(_use_count->at(_buffer)) + ". reference to " + convert::to_string((void*)_buffer) + " (added from copy assignment operator)");
            }
            return *this;
        }
 
        stable_reference_buffer& operator=(const buffer& other) override
        {
            if (this == &other)
            {
                return *this;
            }
 
            auto lock = _use_count->get_lock_guard();
            reset();
            buffer::operator=(other);
 
            if (is_known(_buffer))
            {
                _use_count->insert(_buffer, _use_count->at(_buffer) + 1);
            }
            else
            {
                _use_count->insert(_buffer, get_initial_use_count());
            }
 
            return *this;
        }

        static bool is_known(const void* address)
        {
            if (!address)
            {
                return false; // performance optimization for nullptr
            }
 
            auto use_count = get_use_count();
            auto lock      = use_count->get_lock_guard();
            return use_count->count((uint8_t*)address) == 1;
        }

        size_t use_count() const
        {
            if (_buffer == nullptr)
            {
                return 0;
            }
            return _use_count->at_or_default(_buffer, 0); // return 0 if the buffer is unknown
        }

        void reset() noexcept override
        {
            try
            {
                auto lock = _use_count->get_lock_guard();
 
                if (is_known(_buffer))
                {
                    auto updated_value = _use_count->update(_buffer, [](auto& count)
                                                            { --count; });
 
                    if (updated_value == 0)
                    {
                        CBEAM_LOG_DEBUG("Deallocating " + convert::to_string((void*)_buffer));
                        _use_count->erase(_buffer);
                        buffer::reset();
                    }
                    else
                    {
                        CBEAM_LOG_DEBUG("Removed reference to " + convert::to_string((void*)_buffer) + " (" + std::to_string(_use_count->at(_buffer)) + " left)");
 
                        if (updated_value < 0)
                        {
                            CBEAM_LOG("cbeam::stable_reference_buffer::reset: Detected invalid pointer to " + convert::to_string((void*)_buffer));
                            assert(false);
                        }
                    }
                }
                _buffer = nullptr;
                _size   = 0;
            }
            catch (const std::exception& ex)
            {
                CBEAM_LOG("cbeam::container::container::stable_reference_buffer: ");
                assert(false);
            }
        }
 
        using buffer::swap;

        void swap(stable_reference_buffer& other)
        {
            auto lock = _use_count->get_lock_guard();
            buffer::swap(other);
        }
 
    private:
        static std::shared_ptr<stable_interprocess_map<uint8_t*, int>> get_use_count()
        {
            return lifecycle::singleton<stable_interprocess_map<uint8_t*, int>, std::string, size_t>::get(
                "cbeam::memory::stable_reference_buffer::_use_count",
                std::to_string(concurrency::get_current_process_id()) + ".srb.cbeam",
                1024);
        }

        static int get_initial_use_count()
        {
            return get_use_count()->at_or_default(nullptr, 1); // we use the key nullptr to store the initial use count
        }

        std::shared_ptr<stable_interprocess_map<uint8_t*, int>> _use_count{get_use_count()};
    };
}