Forge
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/ClickHouse

Public

mirrored fromhttps://github.com/cloudflare/ClickHouse

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
70034b6232840c23f0ccb4e5d3ad5fb78bd625e7

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

ClickHouse/base/mysqlxx

base/mysqlxx/Pool.h

245lines · modecode

RawDownload
Clang Tidy, part 7 (#9799)c7863e1
6 years ago
1#pragma once
2
3#include <list>
4#include <memory>
5#include <mutex>
6
7#include <Poco/Exception.h>
8#include <mysqlxx/Connection.h>
9
10
11#define MYSQLXX_POOL_DEFAULT_START_CONNECTIONS 1
12#define MYSQLXX_POOL_DEFAULT_MAX_CONNECTIONS 16
13#define MYSQLXX_POOL_SLEEP_ON_CONNECT_FAIL 1
14
15
16namespace mysqlxx
17{
18
19/** MySQL connections pool.
20 * This class is poorly connected with mysqlxx and is made in different style (was taken from old code).
21 * Usage:
22 * mysqlxx::Pool pool("mysql_params");
23 *
24 * void thread()
25 * {
26 * mysqlxx::Pool::Entry connection = pool.Get();
27 * std::string s = connection->query("SELECT 'Hello, world!' AS world").use().fetch()["world"].getString();
28 * }
29 * TODO: simplify with PoolBase.
30 */
31class Pool final
32{
33protected:
34 /// Information about connection.
35 struct Connection
36 {
37 mysqlxx::Connection conn;
38 int ref_count = 0;
39 };
40
41public:
42 /// Connection with database.
43 class Entry
44 {
45 public:
46 Entry() {}
47
48 Entry(const Entry & src)
49 : data(src.data), pool(src.pool)
50 {
51 incrementRefCount();
52 }
53
54 ~Entry()
55 {
56 decrementRefCount();
57 }
58
59 Entry & operator= (const Entry & src)
60 {
61 pool = src.pool;
62 if (data)
63 decrementRefCount();
64 data = src.data;
65 if (data)
66 incrementRefCount();
67 return * this;
68 }
69
70 bool isNull() const
71 {
72 return data == nullptr;
73 }
74
75 operator mysqlxx::Connection & () &
76 {
77 forceConnected();
78 return data->conn;
79 }
80
81 operator const mysqlxx::Connection & () const &
82 {
83 forceConnected();
84 return data->conn;
85 }
86
87 const mysqlxx::Connection * operator->() const &
88 {
89 forceConnected();
90 return &data->conn;
91 }
92
93 mysqlxx::Connection * operator->() &
94 {
95 forceConnected();
96 return &data->conn;
97 }
98
99 Entry(Pool::Connection * conn, Pool * p)
100 : data(conn), pool(p)
101 {
102 incrementRefCount();
103 }
104
105 std::string getDescription() const
106 {
107 if (pool)
108 return pool->getDescription();
109 else
110 return "pool is null";
111 }
112
113 void disconnect();
114
115 friend class Pool;
116
117 private:
118 /// Pointer to mysqlxx connection.
119 Connection * data = nullptr;
120 /// Pointer to pool we are belonging to.
121 Pool * pool = nullptr;
122
123 /// Connects to database. If connection is failed then waits and repeats again.
124 void forceConnected() const;
125
126 /// Connects to database. If connection is failed then returns false.
127 bool tryForceConnected() const
128 {
129 return data->conn.ping();
130 }
131
132 void incrementRefCount();
133 void decrementRefCount();
134 };
135
136
137 Pool(const std::string & config_name,
138 unsigned default_connections_ = MYSQLXX_POOL_DEFAULT_START_CONNECTIONS,
139 unsigned max_connections_ = MYSQLXX_POOL_DEFAULT_MAX_CONNECTIONS,
140 const char * parent_config_name_ = nullptr)
141 : Pool{Poco::Util::Application::instance().config(), config_name,
142 default_connections_, max_connections_, parent_config_name_}
143 {}
144
145 /**
146 * @param config_name Setting name in configuration file
147 * @param default_connections_ Number of default connections
148 * @param max_connections_ Maximum number of connections
149 */
150 Pool(const Poco::Util::AbstractConfiguration & cfg, const std::string & config_name,
151 unsigned default_connections_ = MYSQLXX_POOL_DEFAULT_START_CONNECTIONS,
152 unsigned max_connections_ = MYSQLXX_POOL_DEFAULT_MAX_CONNECTIONS,
153 const char * parent_config_name_ = nullptr);
154
155 /** Like with mysqlxx::Connection, either port either socket should be specified.
156 * If server is localhost and socket is not empty, than socket is used. Otherwise, server and port is used.
157 */
158 Pool(const std::string & db_,
159 const std::string & server_,
160 const std::string & user_ = "",
161 const std::string & password_ = "",
162 unsigned port_ = 0,
163 const std::string & socket_ = "",
164 unsigned connect_timeout_ = MYSQLXX_DEFAULT_TIMEOUT,
165 unsigned rw_timeout_ = MYSQLXX_DEFAULT_RW_TIMEOUT,
166 unsigned default_connections_ = MYSQLXX_POOL_DEFAULT_START_CONNECTIONS,
167 unsigned max_connections_ = MYSQLXX_POOL_DEFAULT_MAX_CONNECTIONS,
168 unsigned enable_local_infile_ = MYSQLXX_DEFAULT_ENABLE_LOCAL_INFILE)
169 : default_connections(default_connections_), max_connections(max_connections_),
170 db(db_), server(server_), user(user_), password(password_), port(port_), socket(socket_),
171 connect_timeout(connect_timeout_), rw_timeout(rw_timeout_), enable_local_infile(enable_local_infile_) {}
172
173 Pool(const Pool & other)
174 : default_connections{other.default_connections},
175 max_connections{other.max_connections},
176 db{other.db}, server{other.server},
177 user{other.user}, password{other.password},
178 port{other.port}, socket{other.socket},
179 connect_timeout{other.connect_timeout}, rw_timeout{other.rw_timeout},
180 enable_local_infile{other.enable_local_infile}
181 {}
182
183 Pool & operator=(const Pool &) = delete;
184
185 ~Pool();
186
187 /// Allocates connection.
188 Entry get();
189
190 /// Allocates connection.
191 /// If database is not accessible, returns empty Entry object.
192 /// If pool is overflowed, throws exception.
193 Entry tryGet();
194
195 /// Get description of database.
196 std::string getDescription() const
197 {
198 return description;
199 }
200
201 void removeConnection(Connection * connection);
202
203protected:
204 /// Number of MySQL connections which are created at launch.
205 unsigned default_connections;
206 /// Maximum possible number of connections
207 unsigned max_connections;
208
209private:
210 /// Initialization flag.
211 bool initialized{false};
212 /// List of connections.
213 using Connections = std::list<Connection *>;
214 /// List of connections.
215 Connections connections;
216 /// Lock for connections list access
217 std::mutex mutex;
218 /// Description of connection.
219 std::string description;
220
221 /// Connection settings.
222 std::string db;
223 std::string server;
224 std::string user;
225 std::string password;
226 unsigned port;
227 std::string socket;
228 unsigned connect_timeout;
229 unsigned rw_timeout;
230 std::string ssl_ca;
231 std::string ssl_cert;
232 std::string ssl_key;
233 bool enable_local_infile;
234
235 /// True if connection was established at least once.
236 bool was_successful{false};
237
238 /// Initialises class if it wasn't.
239 void initialize();
240
241 /** Create new connection. */
242 Connection * allocConnection(bool dont_throw_if_failed_first_time = false);
243};
244
245}