io_context_thread_manager.cpp

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#include "internal/integration/io_context_thread_manager.h"
#include "internal/integration/logger_integration.h"
#include "internal/core/network_context.h"
 
#include <atomic>
#include <unordered_map>
#include <vector>
 
namespace kcenon::network::integration {
 
class io_context_thread_manager::impl {
public:
    struct context_entry {
        std::weak_ptr<asio::io_context> context;
        std::string component_name;
    };
 
    impl() : total_started_(0), total_completed_(0) {}
 
    ~impl() {
        stop_all();
        wait_all();
    }
 
    std::shared_ptr<thread_pool_interface> get_thread_pool() {
        std::lock_guard<std::mutex> lock(mutex_);
        if (!thread_pool_) {
            thread_pool_ = kcenon::network::core::network_context::instance().get_thread_pool();
            if (!thread_pool_) {
                // Fallback to basic thread pool with size for concurrent io_contexts
                // Tests like ConnectionScaling need 20+ clients + server
                // Use larger size to avoid thread exhaustion
                auto pool_size = std::max(32u, std::thread::hardware_concurrency() * 4);
                thread_pool_ = std::make_shared<basic_thread_pool>(pool_size);
            }
        }
        return thread_pool_;
    }
 
    std::future<void> run_io_context(
        std::shared_ptr<asio::io_context> io_context,
        const std::string& component_name
    ) {
        auto pool = get_thread_pool();
        if (!pool) {
            auto promise = std::make_shared<std::promise<void>>();
            promise->set_exception(
                std::make_exception_ptr(
                    std::runtime_error("Thread pool not available")
                )
            );
            return promise->get_future();
        }
 
        // Submit io_context::run to thread pool
        auto ctx_weak = std::weak_ptr<asio::io_context>(io_context);
        auto name = component_name.empty() ? "unnamed" : component_name;
 
        // Create a shared promise for the caller's future
        auto caller_promise = std::make_shared<std::promise<void>>();
        auto caller_future = caller_promise->get_future();
 
        // Track the entry for metrics and stop_all functionality
        {
            std::lock_guard<std::mutex> lock(mutex_);
            context_entry entry;
            entry.context = io_context;
            entry.component_name = name;
            entries_.push_back(std::move(entry));
            total_started_.fetch_add(1, std::memory_order_relaxed);
        }
 
        // Submit the io_context::run task - this is the ONLY task we submit
        // No separate monitoring task to avoid thread pool exhaustion
        // Note: Capture total_completed_ by pointer to avoid capturing 'this',
        // which can cause heap corruption during static destruction.
        auto* completed_ptr = &total_completed_;
        // Submit io_context::run task to thread pool
        // IMPORTANT: Avoid logging inside this task to prevent heap corruption during
        // static destruction. When common_system's GlobalLoggerRegistry is destroyed
        // before this thread pool task completes, logging causes heap corruption.
        // Use detail::static_destruction_guard::is_logging_safe() for any necessary logging.
        pool->submit([ctx_weak, caller_promise, completed_ptr]() {
            auto ctx = ctx_weak.lock();
            if (!ctx) {
                // io_context expired - this can happen during rapid shutdown
                caller_promise->set_value();
                return;
            }
 
            try {
                ctx->run();
                caller_promise->set_value();
            } catch (...) {
                caller_promise->set_exception(std::current_exception());
            }
 
            completed_ptr->fetch_add(1, std::memory_order_relaxed);
        });
 
        // Clean up expired entries periodically
        cleanup_expired();
 
        return caller_future;
    }
 
    void stop_io_context(std::shared_ptr<asio::io_context> io_context) {
        // NOTE: No logging here to prevent heap corruption during static
        // destruction. This may be called when GlobalLoggerRegistry is destroyed.
        if (io_context) {
            io_context->stop();
        }
    }
 
    void stop_all() {
        // NOTE: No logging here to prevent heap corruption during static
        // destruction. This may be called when GlobalLoggerRegistry is destroyed.
        std::lock_guard<std::mutex> lock(mutex_);
        for (auto& entry : entries_) {
            auto ctx = entry.context.lock();
            if (ctx) {
                ctx->stop();
            }
        }
    }
 
    void wait_all() {
        // Since futures are returned directly to callers, wait_all just
        // waits for all tracked io_contexts to be stopped.
        // The callers should wait on their own futures.
        auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
 
        while (std::chrono::steady_clock::now() < deadline) {
            {
                std::lock_guard<std::mutex> lock(mutex_);
                bool all_stopped = true;
                for (const auto& entry : entries_) {
                    auto ctx = entry.context.lock();
                    if (ctx && !ctx->stopped()) {
                        all_stopped = false;
                        break;
                    }
                }
                if (all_stopped) {
                    entries_.clear();
                    break;
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
 
        // Clear any remaining expired entries
        std::lock_guard<std::mutex> lock(mutex_);
        entries_.clear();
    }
 
    size_t active_count() const {
        std::lock_guard<std::mutex> lock(mutex_);
        size_t count = 0;
        for (const auto& entry : entries_) {
            auto ctx = entry.context.lock();
            if (ctx && !ctx->stopped()) {
                ++count;
            }
        }
        return count;
    }
 
    bool is_active(std::shared_ptr<asio::io_context> io_context) const {
        if (!io_context) {
            return false;
        }
 
        std::lock_guard<std::mutex> lock(mutex_);
        for (const auto& entry : entries_) {
            auto ctx = entry.context.lock();
            if (ctx && ctx.get() == io_context.get() && !ctx->stopped()) {
                return true;
            }
        }
        return false;
    }
 
    void set_thread_pool(std::shared_ptr<thread_pool_interface> pool) {
        std::lock_guard<std::mutex> lock(mutex_);
        thread_pool_ = pool;
    }
 
    io_context_thread_manager::metrics get_metrics() const {
        io_context_thread_manager::metrics m;
        m.active_contexts = active_count();
        m.total_started = total_started_.load(std::memory_order_relaxed);
        m.total_completed = total_completed_.load(std::memory_order_relaxed);
        return m;
    }
 
private:
    void cleanup_expired() {
        std::lock_guard<std::mutex> lock(mutex_);
        entries_.erase(
            std::remove_if(entries_.begin(), entries_.end(),
                [](const context_entry& entry) {
                    return entry.context.expired();
                }),
            entries_.end()
        );
    }
 
    mutable std::mutex mutex_;
    std::shared_ptr<thread_pool_interface> thread_pool_;
    std::vector<context_entry> entries_;
    std::atomic<size_t> total_started_;
    std::atomic<size_t> total_completed_;
};
 
io_context_thread_manager& io_context_thread_manager::instance() {
    static io_context_thread_manager instance;
    return instance;
}
 
io_context_thread_manager::io_context_thread_manager()
    // Intentional Leak pattern: Use no-op deleter to prevent destruction
    // during static destruction phase. This avoids heap corruption when
    // thread pool tasks still reference impl's members (e.g., completed counter).
    // Memory impact: ~few KB (reclaimed by OS on process termination)
    : pimpl_(new impl(), [](impl*) { /* no-op deleter - intentional leak */ }) {
}
 
io_context_thread_manager::~io_context_thread_manager() = default;
 
std::future<void> io_context_thread_manager::run_io_context(
    std::shared_ptr<asio::io_context> io_context,
    const std::string& component_name
) {
    return pimpl_->run_io_context(io_context, component_name);
}
 
void io_context_thread_manager::stop_io_context(
    std::shared_ptr<asio::io_context> io_context
) {
    pimpl_->stop_io_context(io_context);
}
 
void io_context_thread_manager::stop_all() {
    pimpl_->stop_all();
}
 
void io_context_thread_manager::wait_all() {
    pimpl_->wait_all();
}
 
size_t io_context_thread_manager::active_count() const {
    return pimpl_->active_count();
}
 
bool io_context_thread_manager::is_active(
    std::shared_ptr<asio::io_context> io_context
) const {
    return pimpl_->is_active(io_context);
}
 
void io_context_thread_manager::set_thread_pool(
    std::shared_ptr<thread_pool_interface> pool
) {
    pimpl_->set_thread_pool(pool);
}
 
io_context_thread_manager::metrics io_context_thread_manager::get_metrics() const {
    return pimpl_->get_metrics();
}
 
} // namespace kcenon::network::integration