session_manager_base.h

Go to the documentation of this file.

/*****************************************************************************
BSD 3-Clause License
 
Copyright (c) 2025, kcenon
All rights reserved.
*****************************************************************************/
 
#pragma once
 
#include "internal/core/session_info.h"
#include "internal/core/session_traits.h"
 
#include <atomic>
#include <chrono>
#include <memory>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <string>
#include <type_traits>
#include <unordered_map>
#include <vector>
 
namespace kcenon::network::core {
 
struct session_config {
    size_t max_sessions{1000};
    std::chrono::milliseconds idle_timeout{std::chrono::minutes(5)};
    std::chrono::milliseconds cleanup_interval{std::chrono::seconds(30)};
    bool enable_backpressure{true};
    double backpressure_threshold{0.8};
};
 
template <typename SessionType>
class session_manager_base {
public:
    using session_ptr = std::shared_ptr<SessionType>;
    using traits = session_traits<SessionType>;
    using info_type = session_info_t<SessionType>;
 
    explicit session_manager_base(const session_config& config = session_config())
        : config_(config)
        , session_count_(0)
        , total_accepted_(0)
        , total_rejected_(0)
        , total_cleaned_up_(0) {}
 
    // Non-copyable, non-movable
    session_manager_base(const session_manager_base&) = delete;
    session_manager_base& operator=(const session_manager_base&) = delete;
    session_manager_base(session_manager_base&&) = delete;
    session_manager_base& operator=(session_manager_base&&) = delete;
 
    virtual ~session_manager_base() = default;
 
    // =========================================================================
    // Connection Acceptance
    // =========================================================================
 
    [[nodiscard]] auto can_accept_connection() const -> bool {
        return session_count_.load(std::memory_order_acquire) < config_.max_sessions;
    }
 
    [[nodiscard]] auto is_backpressure_active() const -> bool {
        if (!config_.enable_backpressure) {
            return false;
        }
        auto count = session_count_.load(std::memory_order_acquire);
        auto threshold =
            static_cast<size_t>(config_.max_sessions * config_.backpressure_threshold);
        return count >= threshold;
    }
 
    // =========================================================================
    // Session CRUD
    // =========================================================================
 
    auto add_session(session_ptr session, const std::string& session_id = "") -> bool {
        if (!can_accept_connection()) {
            total_rejected_.fetch_add(1, std::memory_order_relaxed);
            return false;
        }
 
        std::unique_lock lock(sessions_mutex_);
        std::string id = session_id.empty() ? generate_id() : session_id;
        active_sessions_.emplace(id, info_type(std::move(session)));
        session_count_.fetch_add(1, std::memory_order_release);
        total_accepted_.fetch_add(1, std::memory_order_relaxed);
 
        return true;
    }
 
    auto add_session_with_id(session_ptr session, const std::string& session_id = "")
        -> std::string {
        if (!can_accept_connection()) {
            total_rejected_.fetch_add(1, std::memory_order_relaxed);
            return "";
        }
 
        std::unique_lock lock(sessions_mutex_);
        std::string id = session_id.empty() ? generate_id() : session_id;
        active_sessions_.emplace(id, info_type(std::move(session)));
        session_count_.fetch_add(1, std::memory_order_release);
        total_accepted_.fetch_add(1, std::memory_order_relaxed);
 
        return id;
    }
 
    auto remove_session(const std::string& session_id) -> bool {
        std::unique_lock lock(sessions_mutex_);
        auto it = active_sessions_.find(session_id);
        if (it != active_sessions_.end()) {
            active_sessions_.erase(it);
            session_count_.fetch_sub(1, std::memory_order_release);
            return true;
        }
        return false;
    }
 
    [[nodiscard]] auto get_session(const std::string& session_id) const -> session_ptr {
        std::shared_lock lock(sessions_mutex_);
        auto it = active_sessions_.find(session_id);
        return (it != active_sessions_.end()) ? it->second.session : nullptr;
    }
 
    [[nodiscard]] auto get_all_sessions() const -> std::vector<session_ptr> {
        std::shared_lock lock(sessions_mutex_);
        std::vector<session_ptr> sessions;
        sessions.reserve(active_sessions_.size());
        for (const auto& [id, info] : active_sessions_) {
            sessions.push_back(info.session);
        }
        return sessions;
    }
 
    [[nodiscard]] auto get_all_session_ids() const -> std::vector<std::string> {
        std::shared_lock lock(sessions_mutex_);
        std::vector<std::string> ids;
        ids.reserve(active_sessions_.size());
        for (const auto& [id, info] : active_sessions_) {
            ids.push_back(id);
        }
        return ids;
    }
 
    // =========================================================================
    // Activity Tracking (only available when traits::has_activity_tracking)
    // =========================================================================
 
    template <typename T = SessionType>
    auto update_activity(const std::string& session_id)
        -> std::enable_if_t<session_traits<T>::has_activity_tracking, void> {
        std::unique_lock lock(sessions_mutex_);
        auto it = active_sessions_.find(session_id);
        if (it != active_sessions_.end()) {
            it->second.update_activity();
        }
    }
 
    template <typename T = SessionType>
    [[nodiscard]] auto get_idle_duration(const std::string& session_id) const
        -> std::enable_if_t<session_traits<T>::has_activity_tracking,
                            std::optional<std::chrono::milliseconds>> {
        std::shared_lock lock(sessions_mutex_);
        auto it = active_sessions_.find(session_id);
        if (it != active_sessions_.end()) {
            return it->second.idle_duration();
        }
        return std::nullopt;
    }
 
    template <typename T = SessionType>
    auto cleanup_idle_sessions()
        -> std::enable_if_t<session_traits<T>::has_activity_tracking, size_t> {
        std::vector<std::pair<std::string, session_ptr>> to_remove;
 
        // Identify idle sessions under read lock
        {
            std::shared_lock lock(sessions_mutex_);
            for (const auto& [id, info] : active_sessions_) {
                if (info.idle_duration() > config_.idle_timeout) {
                    to_remove.emplace_back(id, info.session);
                }
            }
        }
 
        // Stop and remove idle sessions
        size_t removed = 0;
        for (const auto& [id, session] : to_remove) {
            if constexpr (traits::stop_on_clear) {
                if (session) {
                    try {
                        session->stop_session();
                    } catch (...) {
                    }
                }
            }
            if (remove_session(id)) {
                removed++;
            }
        }
 
        if (removed > 0) {
            total_cleaned_up_.fetch_add(removed, std::memory_order_relaxed);
        }
        return removed;
    }
 
    // =========================================================================
    // Lifecycle Management
    // =========================================================================
 
    auto clear_all_sessions() -> void {
        if constexpr (traits::stop_on_clear) {
            // Get sessions under read lock
            std::vector<session_ptr> sessions;
            {
                std::shared_lock lock(sessions_mutex_);
                sessions.reserve(active_sessions_.size());
                for (const auto& [id, info] : active_sessions_) {
                    sessions.push_back(info.session);
                }
            }
 
            // Stop sessions without holding lock
            for (auto& session : sessions) {
                if (session) {
                    try {
                        session->stop_session();
                    } catch (...) {
                    }
                }
            }
        }
 
        std::unique_lock lock(sessions_mutex_);
        active_sessions_.clear();
        session_count_.store(0, std::memory_order_release);
    }
 
    auto stop_all_sessions() -> void { clear_all_sessions(); }
 
    // =========================================================================
    // Metrics
    // =========================================================================
 
    [[nodiscard]] auto get_session_count() const -> size_t {
        return session_count_.load(std::memory_order_acquire);
    }
 
    [[nodiscard]] auto get_total_accepted() const -> uint64_t {
        return total_accepted_.load(std::memory_order_relaxed);
    }
 
    [[nodiscard]] auto get_total_rejected() const -> uint64_t {
        return total_rejected_.load(std::memory_order_relaxed);
    }
 
    [[nodiscard]] auto get_total_cleaned_up() const -> uint64_t {
        return total_cleaned_up_.load(std::memory_order_relaxed);
    }
 
    [[nodiscard]] auto get_utilization() const -> double {
        if (config_.max_sessions == 0) {
            return 0.0;
        }
        return static_cast<double>(session_count_.load(std::memory_order_acquire)) /
               config_.max_sessions;
    }
 
    // =========================================================================
    // Configuration
    // =========================================================================
 
    auto set_max_sessions(size_t max_sessions) -> void {
        config_.max_sessions = max_sessions;
    }
 
    [[nodiscard]] auto get_config() const -> const session_config& { return config_; }
 
    // =========================================================================
    // Statistics (comprehensive view)
    // =========================================================================
 
    struct stats {
        size_t active_sessions;
        size_t max_sessions;
        uint64_t total_accepted;
        uint64_t total_rejected;
        uint64_t total_cleaned_up;
        double utilization;
        bool backpressure_active;
        std::chrono::milliseconds idle_timeout;
    };
 
    [[nodiscard]] auto get_stats() const -> stats {
        return stats{.active_sessions = session_count_.load(std::memory_order_acquire),
                     .max_sessions = config_.max_sessions,
                     .total_accepted = total_accepted_.load(std::memory_order_relaxed),
                     .total_rejected = total_rejected_.load(std::memory_order_relaxed),
                     .total_cleaned_up = total_cleaned_up_.load(std::memory_order_relaxed),
                     .utilization = get_utilization(),
                     .backpressure_active = is_backpressure_active(),
                     .idle_timeout = config_.idle_timeout};
    }
 
    // =========================================================================
    // ID Generation (public for external use)
    // =========================================================================
 
    static auto generate_id() -> std::string {
        static std::atomic<uint64_t> counter{0};
        return std::string(traits::id_prefix) +
               std::to_string(counter.fetch_add(1, std::memory_order_relaxed));
    }
 
protected:
    session_config config_;
    mutable std::shared_mutex sessions_mutex_;
    std::unordered_map<std::string, info_type> active_sessions_;
 
    std::atomic<size_t> session_count_;
    std::atomic<uint64_t> total_accepted_;
    std::atomic<uint64_t> total_rejected_;
    std::atomic<uint64_t> total_cleaned_up_;
};
 
} // namespace kcenon::network::core