Database_connection_pool.cpp

#include <iostream>
#include <memory>
#include <string>
#include <vector>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <chrono>
#include <atomic>
#include <stdexcept>

// ============================================================================
// 数据库连接池 — 泛型, RAII, 健康检查, 超时, 统计
// ============================================================================

// 连接工厂概念：用户提供创建 / 销毁 / 检查 的实现
// 适配 MySQL / PostgreSQL / SQLite 等任意后端
template <typename Connection>
class ConnectionFactory {
public:
    virtual ~ConnectionFactory() = default;
    virtual std::unique_ptr<Connection> Create()    = 0;
    virtual void Destroy(Connection* conn)           = 0;
    virtual bool IsValid(Connection* conn) noexcept  = 0; // 健康检查 (ping)
};

// 连接池
template <typename Connection>
class ConnectionPool {
public:
    struct Config {
        size_t min_size   = 2;   // 最小连接数 (池创建时预热)
        size_t max_size   = 16;  // 最大连接数
        size_t idle_ms    = 30'000; // 空闲连接回收时间 (ms)
        size_t check_ms   = 15'000; // 定期健康检查间隔 (ms)
        size_t wait_ms    = 5'000;  // 获取连接超时 (ms)
    };

    explicit ConnectionPool(std::unique_ptr<ConnectionFactory<Connection>> factory,
                            Config config = {})
        : factory_(std::move(factory)), config_(std::move(config)) {
        // 预热 min_size 个连接
        for (size_t i = 0; i < config_.min_size; ++i) {
            idle_.push(PooledConn{std::move(factory_->Create()), Clock::now()});
            ++total_;
        }
        // 启动后台健康检查 / 空闲回收线程
        bg_ = std::thread(&ConnectionPool::BgLoop, this);
    }

    ~ConnectionPool() {
        stop_ = true;
        bg_cv_.notify_all();
        if (bg_.joinable()) bg_.join();
        // 销毁所有空闲连接
        while (!idle_.empty()) {
            factory_->Destroy(idle_.front().conn.release());
            idle_.pop();
            --total_;
        }
    }

    // RAII 借用句柄 —— 析构时自动归还
    class Borrowed {
    public:
        Borrowed(std::shared_ptr<ConnectionPool> pool, std::unique_ptr<Connection> conn)
            : pool_(std::move(pool)), conn_(std::move(conn)) {}
        ~Borrowed() { if (conn_) pool_->Return(std::move(conn_)); }
        Borrowed(Borrowed&&) = default;
        Borrowed& operator=(Borrowed&&) = default;
        Borrowed(const Borrowed&) = delete;
        Connection* operator->() noexcept { return conn_.get(); }
        Connection& operator*() noexcept { return *conn_; }
    private:
        std::shared_ptr<ConnectionPool> pool_;
        std::unique_ptr<Connection> conn_;
    };

    // 获取连接 (带超时)
    Borrowed Get() {
        auto deadline = Clock::now() + std::chrono::milliseconds(config_.wait_ms);
        std::unique_lock lock(mtx_);

        while (idle_.empty()) {
            // 若未达上限，创建新连接
            if (total_ < config_.max_size) {
                auto conn = factory_->Create();
                ++total_;
                ++active_;
                lock.unlock();
                return Borrowed(shared_from_this(), std::move(conn));
            }
            // 已达上限，等待归还
            if (cv_.wait_until(lock, deadline) == std::cv_status::timeout) {
                throw std::runtime_error("ConnectionPool: acquire timeout");
            }
        }

        auto entry = std::move(idle_.front());
        idle_.pop();
        auto conn = std::move(entry.conn);
        ++active_;
        lock.unlock();

        // 归还前 ping 一次
        if (!factory_->IsValid(conn.get())) {
            factory_->Destroy(conn.release());
            --total_;
            return Get(); // 递归重试
        }

        return Borrowed(shared_from_this(), std::move(conn));
    }

    // 统计
    size_t Total()  const noexcept { return total_.load(); }
    size_t Active() const noexcept { return active_.load(); }
    size_t Idle()   const noexcept { return idle_size_.load(); }
    size_t Waiters()const noexcept { return waiters_.load(); }

private:
    struct PooledConn {
        std::unique_ptr<Connection> conn;
        std::chrono::steady_clock::time_point created;
    };
    using Clock = std::chrono::steady_clock;

    std::shared_ptr<ConnectionPool> shared_from_this() {
        // 简单实现：假设调用方用 shared_ptr 持有池
        return std::shared_ptr<ConnectionPool>(this, [](ConnectionPool*){}); // 不安全的简化
    }

    void Return(std::unique_ptr<Connection> conn) {
        {
            std::lock_guard lock(mtx_);
            idle_.push(PooledConn{std::move(conn), Clock::now()});
            --active_;
        }
        cv_.notify_one();
    }

    void BgLoop() {
        while (!stop_) {
            std::unique_lock lock(bg_mtx_);
            bg_cv_.wait_for(lock, std::chrono::milliseconds(5'000));
            if (stop_) break;
            lock.unlock();

            // 回收空闲超时的连接 (保留 min_size)
            std::lock_guard pool_lock(mtx_);
            std::queue<PooledConn> kept;
            while (idle_.size() > config_.min_size) {
                auto& entry = idle_.front();
                auto age = std::chrono::duration_cast<std::chrono::milliseconds>(
                    Clock::now() - entry.created).count();
                if (age > static_cast<long>(config_.idle_ms)) {
                    factory_->Destroy(entry.conn.release());
                    idle_.pop();
                    --total_;
                } else {
                    kept.push(std::move(entry));
                    idle_.pop();
                }
            }
            while (!kept.empty()) {
                idle_.push(std::move(kept.front()));
                kept.pop();
            }
            idle_size_.store(idle_.size());
        }
    }

    std::unique_ptr<ConnectionFactory<Connection>> factory_;
    Config config_;

    std::mutex mtx_;
    std::condition_variable cv_;
    std::queue<PooledConn> idle_;
    std::atomic<size_t> total_{0}, active_{0}, idle_size_{0}, waiters_{0};

    std::thread bg_;
    std::mutex bg_mtx_;
    std::condition_variable bg_cv_;
    std::atomic<bool> stop_{false};
};

// ============================================================================
// 简单 SQLite 模拟工厂 (演示用)
// ============================================================================
struct FakeSQLite {
    bool ok = true;
    void Exec(const std::string&) { std::cout << "[sql] exec" << std::endl; }
};

struct SQLiteFactory final : ConnectionFactory<FakeSQLite> {
    std::unique_ptr<FakeSQLite> Create()          override { return std::make_unique<FakeSQLite>(); }
    void Destroy(FakeSQLite* conn)                override { delete conn; }
    bool IsValid(FakeSQLite* conn) noexcept       override { return conn && conn->ok; }
};

// ============================================================================
// 演示
// ============================================================================
int main() {
    auto pool = std::make_shared<ConnectionPool<FakeSQLite>>(
        std::make_unique<SQLiteFactory>(),
        ConnectionPool<FakeSQLite>::Config{
            .min_size = 2,
            .max_size = 8,
            .idle_ms  = 10'000,
            .wait_ms  = 2'000,
        });

    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back([pool, i]() {
            for (int j = 0; j < 5; ++j) {
                try {
                    auto conn = pool->Get();
                    conn->Exec("INSERT ...");
                } catch (const std::exception& e) {
                    std::cerr << "thread " << i << " fail: " << e.what() << std::endl;
                }
            }
        });
    }
    for (auto& t : threads) t.join();

    std::cout << "Pool stats: total=" << pool->Total()
              << " active=" << pool->Active()
              << " idle=" << pool->Idle() << std::endl;
    return 0;
}