connection_pool.cpp

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.
 
#include "internal/core/connection_pool.h"
#include "internal/integration/logger_integration.h"
 
#include <future>
 
namespace kcenon::network::core
{
 
    connection_pool::connection_pool(std::string host, unsigned short port,
                                     size_t pool_size,
                                     std::chrono::seconds acquire_timeout)
        : host_(std::move(host)), port_(port), pool_size_(pool_size),
          acquire_timeout_(acquire_timeout)
    {
    }
 
    connection_pool::~connection_pool() noexcept
    {
        try
        {
            // Set shutdown flag
            is_shutdown_.store(true);
 
            // Notify all waiting threads
            cv_.notify_all();
 
            // Stop all connections
            std::lock_guard<std::mutex> lock(mutex_);
            while (!available_.empty())
            {
                auto client = std::move(available_.front());
                available_.pop();
                if (client)
                {
                    (void)client->stop_client();
                }
            }
        }
        catch (...)
        {
            // Destructor must not throw
        }
    }
 
    auto connection_pool::initialize() -> VoidResult
    {
        NETWORK_LOG_INFO("[connection_pool] Initializing pool with " +
            std::to_string(pool_size_) + " connections to " + host_ + ":" +
            std::to_string(port_));
 
        // Zero-size pool requires no connections — succeed immediately
        if (pool_size_ == 0)
        {
            NETWORK_LOG_INFO("[connection_pool] Pool size is 0, no connections to create");
            return ok();
        }
 
        // Launch all connections in parallel using std::async
        struct connect_result
        {
            std::unique_ptr<messaging_client> client;
            bool success = false;
            std::string error_message;
        };
 
        std::vector<std::future<connect_result>> futures;
        futures.reserve(pool_size_);
 
        for (size_t i = 0; i < pool_size_; ++i)
        {
            futures.push_back(std::async(std::launch::async,
                [this, i]() -> connect_result {
                    connect_result cr;
                    cr.client = std::make_unique<messaging_client>(
                        "pool_client_" + std::to_string(i));
 
                    auto result = cr.client->start_client(host_, port_);
                    if (result.is_err())
                    {
                        cr.success = false;
                        cr.error_message = result.error().message;
                        cr.client.reset();
                    }
                    else
                    {
                        cr.success = true;
                    }
                    return cr;
                }));
        }
 
        // Collect results
        size_t success_count = 0;
        size_t fail_count = 0;
        {
            std::lock_guard<std::mutex> lock(mutex_);
            for (size_t i = 0; i < futures.size(); ++i)
            {
                auto cr = futures[i].get();
                if (cr.success && cr.client)
                {
                    available_.push(std::move(cr.client));
                    ++success_count;
                }
                else
                {
                    ++fail_count;
                    NETWORK_LOG_WARN(
                        "[connection_pool] Connection " + std::to_string(i) +
                        " failed: " + cr.error_message);
                }
            }
        }
 
        if (success_count == 0)
        {
            return error_void(
                -1,
                "All " + std::to_string(pool_size_) +
                    " pool connections failed",
                "connection_pool::initialize",
                "Host: " + host_ + ", Port: " + std::to_string(port_)
            );
        }
 
        if (fail_count > 0)
        {
            NETWORK_LOG_WARN("[connection_pool] Initialized with " +
                std::to_string(success_count) + "/" +
                std::to_string(pool_size_) +
                " connections (" + std::to_string(fail_count) + " failed)");
        }
        else
        {
            NETWORK_LOG_INFO("[connection_pool] Successfully initialized " +
                std::to_string(success_count) + " connections");
        }
 
        return ok();
    }
 
    auto connection_pool::acquire() -> std::unique_ptr<messaging_client>
    {
        std::unique_lock<std::mutex> lock(mutex_);
 
        // Wait until a connection is available, shutdown, or timeout
        if (acquire_timeout_.count() > 0)
        {
            if (!cv_.wait_for(lock, acquire_timeout_, [this] {
                    return !available_.empty() || is_shutdown_.load();
                }))
            {
                NETWORK_LOG_WARN(
                    "[connection_pool] Acquire timed out after " +
                    std::to_string(acquire_timeout_.count()) +
                    "s. Active: " +
                    std::to_string(active_count_.load()) + "/" +
                    std::to_string(pool_size_));
                return nullptr;
            }
        }
        else
        {
            cv_.wait(lock, [this] {
                return !available_.empty() || is_shutdown_.load();
            });
        }
 
        // Check if shutdown
        if (is_shutdown_.load())
        {
            return nullptr;
        }
 
        // Get connection from queue
        auto client = std::move(available_.front());
        available_.pop();
        active_count_.fetch_add(1, std::memory_order_relaxed);
 
        NETWORK_LOG_DEBUG("[connection_pool] Acquired connection. Active: " +
            std::to_string(active_count_.load()) + "/" +
            std::to_string(pool_size_));
 
        return client;
    }
 
    auto connection_pool::release(std::unique_ptr<messaging_client> client)
        -> void
    {
        if (!client)
        {
            return;
        }
 
        // Check if client is still connected, reconnect if necessary
        if (!client->is_connected())
        {
            NETWORK_LOG_WARN("[connection_pool] Connection lost, reconnecting...");
 
            auto result = client->start_client(host_, port_);
            if (result.is_err())
            {
                NETWORK_LOG_ERROR("[connection_pool] Failed to reconnect: " +
                    result.error().message);
                // Drop this connection and decrement active count
                active_count_.fetch_sub(1, std::memory_order_relaxed);
                return;
            }
        }
 
        // Return to pool
        {
            std::lock_guard<std::mutex> lock(mutex_);
            available_.push(std::move(client));
        }
 
        active_count_.fetch_sub(1, std::memory_order_relaxed);
 
        // Notify one waiting thread
        cv_.notify_one();
 
        NETWORK_LOG_DEBUG("[connection_pool] Released connection. Active: " +
            std::to_string(active_count_.load()) + "/" +
            std::to_string(pool_size_));
    }
 
} // namespace kcenon::network::core