thread_safe_container.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <unordered_map>
#include <mutex>
#include <shared_mutex>
#include <atomic>
#include <algorithm>
#include <concepts>
#include <thread>
#include <chrono>
#include <condition_variable>
#include "value.h"
#include "core/concepts.h"
 
namespace kcenon::container
{
    // Forward declarations
    class lockfree_container_reader;
    class auto_refresh_reader;
 
    class thread_safe_container : public std::enable_shared_from_this<thread_safe_container>
    {
    public:
        using value_map = std::unordered_map<std::string, value>;
        using const_iterator = value_map::const_iterator;
        using iterator = value_map::iterator;
 
        thread_safe_container() = default;
 
        thread_safe_container(std::initializer_list<std::pair<std::string, value>> init);
 
        thread_safe_container(const thread_safe_container& other);
 
        thread_safe_container(thread_safe_container&& other) noexcept;
 
        thread_safe_container& operator=(const thread_safe_container& other);
 
        thread_safe_container& operator=(thread_safe_container&& other) noexcept;
 
        std::optional<value> get(std::string_view key) const;
 
        template<concepts::ValueVariantType T>
        std::optional<T> get_typed(std::string_view key) const {
            std::shared_lock lock(mutex_);
            auto it = values_.find(std::string(key));
            if (it != values_.end()) {
                return it->second.get<T>();
            }
            return std::nullopt;
        }
 
        void set(std::string_view key, value value);
 
        template<concepts::ValueVariantType T>
        void set_typed(std::string_view key, T&& val) {
            set(key, value(key, std::forward<T>(val)));
        }
 
        bool remove(std::string_view key);
 
        void clear();
 
        size_t size() const;
 
        bool empty() const;
 
        bool contains(std::string_view key) const;
 
        std::vector<std::string> keys() const;
 
        void set_variant(const value& val);
 
        std::optional<value> get_variant(const std::string& key) const;
 
        void set_container(const std::string& key,
                          std::shared_ptr<thread_safe_container> container);
 
        std::shared_ptr<thread_safe_container> get_container(const std::string& key) const;
 
        template<concepts::KeyValueCallback Func>
        void for_each(Func&& func) const {
            std::shared_lock lock(mutex_);
            for (const auto& [key, val] : values_) {
                func(key, val);
            }
        }
 
        template<concepts::MutableKeyValueCallback Func>
        void for_each_mut(Func&& func) {
            std::unique_lock lock(mutex_);
            for (auto& [key, val] : values_) {
                func(key, val);
            }
        }
 
        template<concepts::ValueMapCallback<value_map> Func>
        void bulk_update(Func&& updater) {
            std::unique_lock lock(mutex_);
            updater(values_);
            bulk_write_count_.fetch_add(1, std::memory_order_relaxed);
        }
 
        template<concepts::ConstValueMapCallback<value_map> Func>
        auto bulk_read(Func&& reader) const {
            std::shared_lock lock(mutex_);
            bulk_read_count_.fetch_add(1, std::memory_order_relaxed);
            return reader(values_);
        }
 
        bool compare_exchange(std::string_view key,
                            const value& expected,
                            const value& desired);
 
        struct Statistics {
            size_t read_count;
            size_t write_count;
            size_t bulk_read_count;
            size_t bulk_write_count;
            size_t size;
        };
 
        Statistics get_statistics() const;
 
        std::string to_json() const;
 
        std::vector<uint8_t> serialize() const;
 
        static std::shared_ptr<thread_safe_container> deserialize(
            const std::vector<uint8_t>& data);
 
        value& operator[](const std::string& key);
 
        std::shared_ptr<lockfree_container_reader> create_lockfree_reader();
 
        std::shared_ptr<auto_refresh_reader> create_auto_refresh_reader(
            std::chrono::milliseconds refresh_interval);
 
    private:
        mutable std::shared_mutex mutex_;
        value_map values_;
        
        // Statistics
        mutable std::atomic<size_t> read_count_{0};
        mutable std::atomic<size_t> write_count_{0};
        mutable std::atomic<size_t> bulk_read_count_{0};
        mutable std::atomic<size_t> bulk_write_count_{0};
    };
 
    class snapshot_reader {
    public:
        using snapshot_ptr = std::shared_ptr<const thread_safe_container::value_map>;
 
        explicit snapshot_reader(std::shared_ptr<thread_safe_container> container)
            : container_(container) {
            update_snapshot();
        }
 
        template<concepts::ValueVariantType T>
        std::optional<T> get(std::string_view key) const {
            std::shared_lock lock(snapshot_mutex_);
            if (!snapshot_) return std::nullopt;
 
            auto it = snapshot_->find(std::string(key));
            if (it != snapshot_->end()) {
                return it->second.get<T>();
            }
            return std::nullopt;
        }
 
        void update_snapshot();
 
    private:
        std::shared_ptr<thread_safe_container> container_;
        snapshot_ptr snapshot_;
        mutable std::shared_mutex snapshot_mutex_;
    };
 
    class lockfree_container_reader {
    public:
        using snapshot_type = std::unordered_map<std::string, value>;
        using snapshot_ptr = std::shared_ptr<const snapshot_type>;
 
        explicit lockfree_container_reader(std::shared_ptr<thread_safe_container> container)
            : container_(std::move(container))
            , snapshot_(std::make_shared<const snapshot_type>())
        {
            refresh();
        }
 
        lockfree_container_reader(const lockfree_container_reader& other)
            : container_(other.container_)
            , snapshot_(std::atomic_load_explicit(&other.snapshot_, std::memory_order_acquire))
        {}
 
        lockfree_container_reader(lockfree_container_reader&& other) noexcept
            : container_(std::move(other.container_))
            , snapshot_(std::atomic_load_explicit(&other.snapshot_, std::memory_order_acquire))
        {}
 
        lockfree_container_reader& operator=(const lockfree_container_reader& other) {
            if (this != &other) {
                container_ = other.container_;
                auto new_snapshot = std::atomic_load_explicit(&other.snapshot_, std::memory_order_acquire);
                std::atomic_store_explicit(&snapshot_, new_snapshot, std::memory_order_release);
            }
            return *this;
        }
 
        lockfree_container_reader& operator=(lockfree_container_reader&& other) noexcept {
            if (this != &other) {
                container_ = std::move(other.container_);
                auto new_snapshot = std::atomic_load_explicit(&other.snapshot_, std::memory_order_acquire);
                std::atomic_store_explicit(&snapshot_, new_snapshot, std::memory_order_release);
            }
            return *this;
        }
 
        ~lockfree_container_reader() = default;
 
        template<concepts::ValueVariantType T>
        [[nodiscard]] std::optional<T> get(std::string_view key) const noexcept {
            auto snap = std::atomic_load_explicit(&snapshot_, std::memory_order_acquire);
            if (!snap) {
                return std::nullopt;
            }
 
            auto it = snap->find(std::string(key));
            if (it != snap->end()) {
                return it->second.get<T>();
            }
            return std::nullopt;
        }
 
        [[nodiscard]] bool contains(std::string_view key) const noexcept {
            auto snap = std::atomic_load_explicit(&snapshot_, std::memory_order_acquire);
            if (!snap) {
                return false;
            }
            return snap->find(std::string(key)) != snap->end();
        }
 
        [[nodiscard]] size_t size() const noexcept {
            auto snap = std::atomic_load_explicit(&snapshot_, std::memory_order_acquire);
            return snap ? snap->size() : 0;
        }
 
        [[nodiscard]] bool empty() const noexcept {
            return size() == 0;
        }
 
        [[nodiscard]] std::vector<std::string> keys() const {
            auto snap = std::atomic_load_explicit(&snapshot_, std::memory_order_acquire);
            std::vector<std::string> result;
            if (snap) {
                result.reserve(snap->size());
                for (const auto& [key, val] : *snap) {
                    result.push_back(key);
                }
            }
            return result;
        }
 
        template<typename Func>
        void for_each(Func&& func) const {
            auto snap = std::atomic_load_explicit(&snapshot_, std::memory_order_acquire);
            if (snap) {
                for (const auto& [key, val] : *snap) {
                    func(key, val);
                }
            }
        }
 
        void refresh() {
            auto new_snapshot = std::make_shared<snapshot_type>();
            container_->for_each([&new_snapshot](const std::string& key, const value& val) {
                new_snapshot->emplace(key, val);
            });
            std::atomic_store_explicit(&snapshot_,
                                      snapshot_ptr(new_snapshot),
                                      std::memory_order_release);
            refresh_count_.fetch_add(1, std::memory_order_relaxed);
        }
 
        [[nodiscard]] size_t refresh_count() const noexcept {
            return refresh_count_.load(std::memory_order_relaxed);
        }
 
        [[nodiscard]] std::shared_ptr<thread_safe_container> source() const noexcept {
            return container_;
        }
 
    private:
        std::shared_ptr<thread_safe_container> container_;
        snapshot_ptr snapshot_;
        std::atomic<size_t> refresh_count_{0};
    };
 
    class auto_refresh_reader {
    public:
        explicit auto_refresh_reader(std::shared_ptr<thread_safe_container> container,
                                     std::chrono::milliseconds refresh_interval)
            : reader_(std::make_shared<lockfree_container_reader>(std::move(container)))
            , refresh_interval_(refresh_interval)
            , running_(true)
        {
            refresh_thread_ = std::thread(&auto_refresh_reader::refresh_loop, this);
        }
 
        ~auto_refresh_reader() {
            stop();
        }
 
        // Non-copyable
        auto_refresh_reader(const auto_refresh_reader&) = delete;
        auto_refresh_reader& operator=(const auto_refresh_reader&) = delete;
 
        // Non-movable (due to thread)
        auto_refresh_reader(auto_refresh_reader&&) = delete;
        auto_refresh_reader& operator=(auto_refresh_reader&&) = delete;
 
        void stop() {
            {
                std::lock_guard<std::mutex> lock(mutex_);
                if (!running_) {
                    return;
                }
                running_ = false;
            }
            cv_.notify_one();
            if (refresh_thread_.joinable()) {
                refresh_thread_.join();
            }
        }
 
        [[nodiscard]] bool is_running() const noexcept {
            std::lock_guard<std::mutex> lock(mutex_);
            return running_;
        }
 
        [[nodiscard]] std::chrono::milliseconds refresh_interval() const noexcept {
            return refresh_interval_;
        }
 
        template<concepts::ValueVariantType T>
        [[nodiscard]] std::optional<T> get(std::string_view key) const noexcept {
            return reader_->get<T>(key);
        }
 
        [[nodiscard]] bool contains(std::string_view key) const noexcept {
            return reader_->contains(key);
        }
 
        [[nodiscard]] size_t size() const noexcept {
            return reader_->size();
        }
 
        [[nodiscard]] bool empty() const noexcept {
            return reader_->empty();
        }
 
        [[nodiscard]] std::vector<std::string> keys() const {
            return reader_->keys();
        }
 
        template<typename Func>
        void for_each(Func&& func) const {
            reader_->for_each(std::forward<Func>(func));
        }
 
        void refresh() {
            reader_->refresh();
        }
 
        [[nodiscard]] size_t refresh_count() const noexcept {
            return reader_->refresh_count();
        }
 
        [[nodiscard]] std::shared_ptr<lockfree_container_reader> reader() const noexcept {
            return reader_;
        }
 
        [[nodiscard]] std::shared_ptr<thread_safe_container> source() const noexcept {
            return reader_->source();
        }
 
    private:
        void refresh_loop() {
            while (true) {
                std::unique_lock<std::mutex> lock(mutex_);
                if (cv_.wait_for(lock, refresh_interval_, [this] { return !running_; })) {
                    break;  // Stopped
                }
                lock.unlock();
                reader_->refresh();
            }
        }
 
        std::shared_ptr<lockfree_container_reader> reader_;
        std::chrono::milliseconds refresh_interval_;
        std::thread refresh_thread_;
        mutable std::mutex mutex_;
        std::condition_variable cv_;
        bool running_;
    };
 
    // Type alias for backward compatibility
    using lockfree_reader = snapshot_reader;
 
    // Inline definition of create_lockfree_reader (must be after lockfree_container_reader is defined)
    inline std::shared_ptr<lockfree_container_reader> thread_safe_container::create_lockfree_reader() {
        return std::make_shared<lockfree_container_reader>(shared_from_this());
    }
 
    // Inline definition of create_auto_refresh_reader (must be after auto_refresh_reader is defined)
    inline std::shared_ptr<auto_refresh_reader> thread_safe_container::create_auto_refresh_reader(
        std::chrono::milliseconds refresh_interval) {
        return std::make_shared<auto_refresh_reader>(shared_from_this(), refresh_interval);
    }
 
} // namespace kcenon::container