ClickHouse ConnectionPool Á´½Ó³ØµÄÓÅ»¯

  class ConnectionPool : public IConnectionPool, private PoolBase<Connection>
  {
  public:
      using Entry = IConnectionPool::Entry;
      using Base = PoolBase<Connection>;
      ConnectionPool(unsigned max_connections_,
              const String & host_,
              UInt16 port_,
              const String & default_database_,
              const String & user_,
              const String & password_,
              const String & cluster_,
              const String & cluster_secret_,
              const String & client_name_,
              Protocol::Compression compression_,
              Protocol::Secure secure_,
              Int64 priority_ = 1)
         : Base(max_connections_,
          &Poco::Logger::get("ConnectionPool (" + host_ + ":" + toString(port_) + ")")),
          host(host_),
          port(port_),
          default_database(default_database_),
          user(user_),
          password(password_),
          cluster(cluster_),
          cluster_secret(cluster_secret_),
          client_name(client_name_),
          compression(compression_),
          secure(secure_),
          priority(priority_)
      {
      }

    PoolBase(unsigned max_items_, Poco::Logger * log_)
       : max_items(max_items_), log(log_)
    {
        items.reserve(max_items);
    }
    /** Creates a new object to put into the pool. */
    virtual ObjectPtr allocObject() = 0;

std::mutex  Ëø
std::unique_lock ΨһËø
std::lock_guard
std::shared_ptr Ö¸Õ룬´øÒýÓüÆÊýÆ÷ use_count
std::vector Êý¾Ý
class Àà
template class Ä£°æÀà

connection_timeout
send_timeout
receive_timeout
tcp_keep_alive_timeout
http_keep_alive_timeout 
secure_connection_timeout
handshake_timeout 

secure °²È«Ä£Ê½ http?https
compression Êý¾Ý´«ÊäÊÇ·ñѹËõ

ip   ½Úµãip
role ½ÚµãµÄ½ÇÉ«

http/tcp port ¼¯Èº·ÃÎʵĶ˿Ú
user ̞
password ÃÜÂë
std::vector<NodeInfo> ¼¯Èº½Úµã
xxx ÆäËû 
ÍøÂçÏÞÖƲÎÊý
ÍøÂç´«Êä²ÎÊý

IClusterInfo
socket/client  Server ·ÅÌṩµÄÁ´½Ó·½Ê½µÄclient µÄ·â×°
?

struct NodeInfo
{
    explicit NodeInfo(std::string host_, std::string role_ = "follower") : host(host_), role(role_) { }
    std::string host;
    std::string role;
};
using NodeInfoPtr = std::shared_ptr<NodeInfo>;
using NodeInfoPtrs = std::vector<NodeInfoPtr>;

struct ClusterInfo
{
    enum class Compression
    {
        Disable = 0,
        Enable = 1,
    };
?
    enum class Secure
    {
        Disable = 0,
        Enable = 1,
    };
?
?
    explicit ClusterInfo(const Poco::Util::AbstractConfiguration & config, std::string config_name);
    //to do ,memory ClusterInfo(xxx);
    NodeInfoPtrs node_info_ptrs;
    std::int32_t port;
    std::string user = "root";
    std::string password;
    Secure security = Secure::Disable;
    Compression compression = Compression::Enable;
    ConnectionTimeouts timeouts;
};
using ClusterInfoPtr = std::shared_ptr<ClusterInfo>;
?
?
?
ConnectionParameters::ConnectionParameters(const Poco::Util::AbstractConfiguration & config)
{
    bool is_secure = config.getBool("secure", false);
    security = is_secure ? Protocol::Secure::Enable : Protocol::Secure::Disable;
?
    host = config.getString("host", "localhost");
    port = config.getInt(
        "port", config.getInt(is_secure ? "tcp_port_secure" : "tcp_port", is_secure ? DBMS_DEFAULT_SECURE_PORT : DBMS_DEFAULT_PORT));
?
    default_database = config.getString("database", "");
?
    /// changed the default value to "default" to fix the issue when the user in the prompt is blank
    user = config.getString("user", "default");
?
    bool password_prompt = false;
    if (config.getBool("ask-password", false))
    {
        if (config.has("password"))
            throw Exception("Specified both --password and --ask-password. Remove one of them", ErrorCodes::BAD_ARGUMENTS);
        password_prompt = true;
    }
    else
    {
        password = config.getString("password", "");
        /// if the value of --password is omitted, the password will be set implicitly to "\n"
        if (password == "\n")
            password_prompt = true;
    }
    if (password_prompt)
    {
#if !defined(ARCADIA_BUILD)
        std::string prompt{"Password for user (" + user + "): "};
        char buf[1000] = {};
        if (auto * result = readpassphrase(prompt.c_str(), buf, sizeof(buf), 0))
            password = result;
#endif
    }
?
    /// By default compression is disabled if address looks like localhost.
    compression = config.getBool("compression", !isLocalAddress(DNSResolver::instance().resolveHost(host)))
        ? Protocol::Compression::Enable : Protocol::Compression::Disable;
?
    timeouts = ConnectionTimeouts(
        Poco::Timespan(config.getInt("connect_timeout", DBMS_DEFAULT_CONNECT_TIMEOUT_SEC), 0),
        Poco::Timespan(config.getInt("send_timeout", DBMS_DEFAULT_SEND_TIMEOUT_SEC), 0),
        Poco::Timespan(config.getInt("receive_timeout", DBMS_DEFAULT_RECEIVE_TIMEOUT_SEC), 0),
        Poco::Timespan(config.getInt("tcp_keep_alive_timeout", 0), 0));
}

class ConnectionCache
{
  
}

class IConnection
{
public:
    IConnection() = default;
?
    virtual ~IConnection() = default;
?
    virtual void connect() = 0;
    virtual void close() = 0;
?
    virtual void heartbeat() = 0;
?
    virtual void isLeader() = 0;
    virtual void getleader() = 0;
    virtual void getCluster() = 0;
?
    virtual void send() = 0;
    virtual void receive() = 0;
?
?
    virtual bool inUse() = 0;
};
xxxx

template<typename TObject>
class ConnectionPool {
public:
    using ObjectPtr = std::shared_ptr<TObject>;
    using ObjectPtrs = std::vector<ObjectPtr>;
?
    ObjectPtr get() {
        std::unique_lock<std::mutex> lock(object_mutex);
?
// ͨ¹ýshared_ptr Ö±½Ó¼ò»¯·â×°£¬ºÍuse_count ÅжÏobject ÊÇ·ñÔÚʹÓÃ
        while (true) {
            for (auto object_ptr : object_ptrs)
                if (object_ptr.use_count() == 2)
                    return object_ptr;
?
            auto object_index = object_ptrs.size();
            if (object_index < max_pool_size) {
                alloObject(object_index);
                continue;
            }
?
            condition_variable.wait_for(lock, std::chrono::milliseconds(10));
        }
    }
?
    ConnectionPool(ClusterInfoPtr cluster_info_ptr_, size_t min_pool_size_, size_t max_pool_size_)
            : cluster_info_ptr(std::move(cluster_info_ptr_)), min_pool_size(min_pool_size_),
              max_pool_size(max_pool_size_) {
        reserve();
    }
?
?
    void reserve() {
        std::lock_guard<std::mutex> lock(object_mutex);
        for (size_t object_index = object_ptrs.size(); object_index < min_pool_size; object_index++)
            alloObject(object_index);
    }
?
?
    void alloObject(size_t object_index) {
        ObjectPtr object_ptr
                = std::make_shared<TObject>(
                        cluster_info_ptr->node_info_ptrs[object_index % cluster_info_ptr->node_info_ptrs.size()]);
        object_ptrs.emplace_back(object_ptr);
    }
    //to do ¼¯ÈºÐÅÏ¢¸üУ¬¸üÐÂClusterInfo
?
    ~ConnectionPool() = default;
?
private:
    ClusterInfoPtr cluster_info_ptr;
    ObjectPtrs object_ptrs;
    size_t min_pool_size;
    size_t max_pool_size;
    std::mutex object_mutex;
    std::condition_variable condition_variable;
};
?

    /** What is given to the user. */
    class Entry
    {
    public:
        friend class PoolBase<Object>;

        Entry() = default; /// For deferred initialization.

        /** The `Entry` object protects the resource from being used by another thread.
          * The following methods are forbidden for `rvalue`, so you can not write a similar to
          *
          * auto q = pool.get()->query("SELECT .."); // Oops, after this line Entry was destroyed
          * q.execute (); // Someone else can use this Connection
          */
        Object * operator->() && = delete;
        const Object * operator->() const && = delete;
        Object & operator*() && = delete;
        const Object & operator*() const && = delete;

        Object * operator->() & { return &*data->data.object; }
        const Object * operator->() const & { return &*data->data.object; }
        Object & operator*() & { return *data->data.object; }
        const Object & operator*() const & { return *data->data.object; }

        bool isNull() const { return data == nullptr; }

        PoolBase * getPool() const
        {
            if (!data)
                throw DB::Exception("Attempt to get pool from uninitialized entry", DB::ErrorCodes::LOGICAL_ERROR);
            return &data->data.pool;
        }

    private:
        std::shared_ptr<PoolEntryHelper> data;

        explicit Entry(PooledObject & object) : data(std::make_shared<PoolEntryHelper>(object)) { }
    };

    virtual ~PoolBase() = default;

    /** Allocates the object. Wait for free object in pool for 'timeout'. With 'timeout' < 0, the timeout is infinite. */
    Entry get(Poco::Timespan::TimeDiff timeout)
    {
        std::unique_lock lock(mutex);

        while (true)
        {
            for (auto & item : items)
                if (!item->in_use)
                    return Entry(*item);

            if (items.size() < max_items)
            {
                ObjectPtr object = allocObject();
                items.emplace_back(std::make_shared<PooledObject>(object, *this));
                return Entry(*items.back());
            }

            LOG_INFO(log, "No free connections in pool. Waiting.");

            if (timeout < 0)
                available.wait(lock);
            else
                available.wait_for(lock, std::chrono::microseconds(timeout));
        }
    }

?
class RedisConnection : public IConnection
{
public:
    RedisConnection(NodeInfoPtr nodeInfoPtr);
    RedisConnection(ClusterInfoPtr clusterInfoPtr);
    ~RedisConnection();
?
//redis client object 
    xxxxx
};

class RedisClient : public SConnectionPool<RedisConnection>
{
public:
    RedisClient(ClusterInfoPtr clusterInfoPtr_, size_t min_pool_size_, size_t max_pool_size_)
        : ConnectionPool(clusterInfoPtr_, min_pool_size_, max_pool_size_), cluster_info_ptr(clusterInfoPtr_)
    {
    }
?
    ~RedisClient() = default;
?
};

people1 method5
xxxxxxx  1
people1 method1
xxxxxxx  2
people1 method2
xxxxxxx  4
people1 method4
xxxxxxx  3
people1 method3
xxxxxxx  1
people1 method1
xxxxxxx  5
people1 method5
xxxxxxx  4
people1 method4
xxxxxxx  5
people1 method5
xxxxxxx  1
people1 method1
xxxxxxx  2
people1 method2
xxxxxxx  3
people1 method3
xxxxxxx  1
people1 method1
xxxxxxx  5
people1 method5
xxxxxxx  3
people1 method3
xxxxxxx  4
people1 method4
xxxxxxx  4
people1 method4
xxxxxxx  3
people1 method3
xxxxxxx  1
people1 method1
xxxxxxx  2
people1 method2
xxxxxxx  5
people1 method5
xxxxxxx  4
people1 method4