Line data Source code
1 : #include "connection_pool.hpp"
2 : #include "../../util/logger.hpp"
3 :
4 : #include <algorithm>
5 :
6 : namespace thinger::http {
7 :
8 : namespace {
9 :
10 : // A key for the logs, without its last field: the credentials of a proxy, if it has one
11 : // (see proxy_endpoint::key())
12 3945 : [[maybe_unused]] std::string_view loggable(std::string_view key) {
13 3945 : return key.substr(0, key.rfind('|'));
14 : }
15 :
16 : }
17 :
18 : struct connection_pool::state : std::enable_shared_from_this<state> {
19 : struct idle_entry {
20 : std::string key;
21 : std::shared_ptr<client_connection> connection;
22 : std::chrono::steady_clock::time_point expires;
23 : uint64_t id;
24 : };
25 : using idle_list = std::list<idle_entry>;
26 :
27 : struct host {
28 : std::string origin;
29 : std::vector<idle_list::iterator> idle; // oldest first
30 : size_t active = 0;
31 : };
32 :
33 : mutable std::mutex mutex;
34 : config limits;
35 : idle_list idle; // oldest first, of every key
36 : std::unordered_map<std::string, host> hosts;
37 : uint64_t next_id = 0;
38 : size_t created = 0;
39 : size_t reused = 0;
40 : bool closed = false;
41 : bool timers = true;
42 :
43 : // Take an idle connection out of the pool (lock held), to close or lend
44 48111 : std::shared_ptr<client_connection> remove(idle_list::iterator entry) {
45 48111 : auto& entries = hosts[entry->key].idle;
46 48111 : entries.erase(std::find(entries.begin(), entries.end(), entry));
47 48111 : auto connection = std::move(entry->connection);
48 48111 : idle.erase(entry);
49 48111 : return connection;
50 : }
51 :
52 : // Forget a key with nothing idle nor lent (lock held)
53 18774 : void prune(const std::string& key) {
54 18774 : auto it = hosts.find(key);
55 18774 : if (it != hosts.end() && it->second.idle.empty() && it->second.active == 0) hosts.erase(it);
56 18774 : }
57 :
58 : // Take the expired idle connections out, to close (lock held)
59 59029 : void sweep(std::chrono::steady_clock::time_point now, std::vector<std::shared_ptr<client_connection>>& closing) {
60 408324 : for (auto it = idle.begin(); it != idle.end();) {
61 349295 : auto entry = it++;
62 349295 : if (entry->expires > now) continue;
63 33 : auto key = entry->key;
64 33 : closing.push_back(remove(entry));
65 33 : prune(key);
66 33 : }
67 59029 : }
68 :
69 : // Close the oldest idle connections beyond the limits (lock held)
70 928 : void trim(std::vector<std::shared_ptr<client_connection>>& closing) {
71 1313 : for (auto& [key, entry] : hosts) {
72 1329 : while (entry.idle.size() > limits.max_idle_per_host) closing.push_back(remove(entry.idle.front()));
73 : }
74 928 : while (idle.size() > limits.max_idle) closing.push_back(remove(idle.begin()));
75 1313 : for (auto it = hosts.begin(); it != hosts.end();) {
76 430 : it = it->second.idle.empty() && it->second.active == 0 ? hosts.erase(it) : std::next(it);
77 : }
78 928 : }
79 :
80 : // The server closes a connection once idle for the time it announced: it is not reused
81 : // near that time, as undici does, nor at all if that is too short
82 56278 : std::chrono::milliseconds idle_time_for(std::optional<std::chrono::seconds> announced) const {
83 56278 : auto timeout = limits.idle_timeout;
84 56278 : if (announced) {
85 90 : timeout = std::min(timeout, std::chrono::duration_cast<std::chrono::milliseconds>(*announced) -
86 45 : keep_alive_margin);
87 : }
88 56278 : return timeout;
89 : }
90 :
91 : // Room for one more idle connection of `key`, closing the oldest idle ones: the most
92 : // likely to be closed by their servers soon (lock held)
93 49771 : void make_room(const std::string& key, host& entry, std::vector<std::shared_ptr<client_connection>>& closing) {
94 50788 : while (entry.idle.size() >= limits.max_idle_per_host) closing.push_back(remove(entry.idle.front()));
95 50773 : while (idle.size() >= limits.max_idle) {
96 1002 : auto oldest = idle.front().key;
97 1002 : closing.push_back(remove(idle.begin()));
98 1002 : if (oldest != key) prune(oldest);
99 1002 : }
100 49771 : }
101 :
102 : // Keep a connection idle for `timeout` (lock held)
103 49771 : void park(const std::string& key, host& entry, const std::shared_ptr<client_connection>& connection,
104 : std::chrono::milliseconds timeout) {
105 49771 : auto id = ++next_id;
106 49771 : connection->unclaim();
107 49771 : if (timers) {
108 47543 : connection->expire_after(timeout, [pool = weak_from_this(), key, id]() {
109 18 : if (auto self = pool.lock()) self->expire(key, id);
110 18 : });
111 : }
112 49771 : idle.push_back(idle_entry{key, connection, std::chrono::steady_clock::now() + timeout, id});
113 99542 : entry.idle.push_back(std::prev(idle.end()));
114 49771 : }
115 :
116 56278 : void release(const std::string& key, std::shared_ptr<client_connection> connection) {
117 : // The connection is not shared with another request now
118 56278 : bool reusable = connection->reusable();
119 56278 : auto announced = connection->keep_alive_timeout();
120 56278 : bool keep = false;
121 56278 : std::vector<std::shared_ptr<client_connection>> closing;
122 : {
123 56278 : std::lock_guard lock(mutex);
124 : // Without timers, the expired ones are closed as a request ends too, before they
125 : // take the room of this one. This key is not pruned: the lease counts as active
126 56278 : if (!timers) sweep(std::chrono::steady_clock::now(), closing);
127 56278 : if (auto it = hosts.find(key); it != hosts.end()) {
128 56278 : if (it->second.active > 0) --it->second.active;
129 56278 : auto timeout = idle_time_for(announced);
130 106052 : keep = reusable && !closed && timeout.count() > 0 && limits.max_idle_per_host > 0 &&
131 49774 : limits.max_idle > 0;
132 56278 : if (keep) {
133 49771 : make_room(key, it->second, closing);
134 49771 : park(key, it->second, connection, timeout);
135 : } else {
136 6507 : prune(key);
137 : }
138 : }
139 56278 : }
140 : // Not reusable, it is already closed, or belongs to a new protocol (a 101 response)
141 56278 : if (reusable && !keep) connection->close();
142 58306 : for (auto& stale : closing) stale->close();
143 56278 : }
144 :
145 : // The idle timeout of a connection elapsed (on its io_context)
146 18 : void expire(const std::string& key, uint64_t id) {
147 18 : std::shared_ptr<client_connection> connection;
148 : {
149 18 : std::lock_guard lock(mutex);
150 18 : auto it = hosts.find(key);
151 18 : if (it == hosts.end()) return;
152 18 : auto& entries = it->second.idle;
153 30 : auto entry = std::find_if(entries.begin(), entries.end(), [id](auto e) { return e->id == id; });
154 : // Lent again, or closed, meanwhile
155 18 : if (entry == entries.end()) return;
156 12 : connection = remove(*entry);
157 12 : prune(key);
158 18 : }
159 12 : LOG_DEBUG("closing a connection idle for too long ({})", loggable(key));
160 12 : connection->close();
161 18 : }
162 : };
163 :
164 10529 : connection_pool::lease& connection_pool::lease::operator=(lease&& other) noexcept {
165 10529 : if (this != &other) {
166 10529 : release();
167 10529 : pool_ = std::move(other.pool_);
168 10529 : key_ = std::move(other.key_);
169 10529 : connection_ = std::move(other.connection_);
170 : }
171 10529 : return *this;
172 : }
173 :
174 140159 : void connection_pool::lease::release() {
175 140159 : if (!connection_) return;
176 56281 : auto connection = std::move(connection_);
177 56281 : connection_.reset();
178 56281 : if (auto pool = pool_.lock()) {
179 56278 : pool->release(key_, std::move(connection));
180 3 : } else if (connection->reusable()) {
181 3 : connection->close();
182 56281 : }
183 56281 : }
184 :
185 2764 : connection_pool::connection_pool() : state_(std::make_shared<state>()) {
186 2764 : }
187 :
188 2764 : connection_pool::~connection_pool() {
189 : {
190 2764 : std::lock_guard lock(state_->mutex);
191 2764 : state_->closed = true;
192 2764 : }
193 : // Nothing can reach the idle connections anymore. Those lent are closed by their
194 : // requests: they are not kept once given back
195 2764 : clear();
196 2764 : }
197 :
198 928 : void connection_pool::configure(const config& config) {
199 928 : std::vector<std::shared_ptr<client_connection>> closing;
200 : {
201 928 : std::lock_guard lock(state_->mutex);
202 928 : state_->limits = config;
203 928 : state_->trim(closing);
204 928 : }
205 1872 : for (auto& connection : closing) connection->close();
206 928 : }
207 :
208 2103 : void connection_pool::use_idle_timers(bool enabled) {
209 2103 : std::lock_guard lock(state_->mutex);
210 2103 : state_->timers = enabled;
211 2103 : }
212 :
213 904 : connection_pool::config connection_pool::get_config() const {
214 904 : std::lock_guard lock(state_->mutex);
215 1808 : return state_->limits;
216 904 : }
217 :
218 56167 : connection_pool::lease connection_pool::acquire(const std::string& key) {
219 56167 : std::shared_ptr<client_connection> found;
220 56167 : std::vector<std::shared_ptr<client_connection>> closing;
221 : {
222 56167 : std::lock_guard lock(state_->mutex);
223 : // Expired ones of any key too, in case their timers did not run
224 56167 : state_->sweep(std::chrono::steady_clock::now(), closing);
225 56167 : if (auto it = state_->hosts.find(key); it != state_->hosts.end()) {
226 52303 : auto& entry = it->second;
227 : // The most recently used first, so the others expire if not needed. A server may
228 : // close an idle connection at any time: one it closed (or wrote to) is not used
229 56437 : while (!entry.idle.empty()) {
230 45103 : auto connection = state_->remove(entry.idle.back());
231 45103 : if (!connection->stale() && connection->claim()) {
232 40969 : found = std::move(connection);
233 40969 : break;
234 : }
235 4134 : closing.push_back(std::move(connection));
236 45103 : }
237 52303 : if (found) {
238 40969 : ++entry.active;
239 40969 : ++state_->reused;
240 : } else {
241 11334 : state_->prune(key);
242 : }
243 : }
244 56167 : }
245 56167 : if (!closing.empty()) LOG_DEBUG("closing {} idle connections closed or expired ({})", closing.size(), loggable(key));
246 60325 : for (auto& connection : closing) connection->close();
247 56167 : if (!found) return {};
248 40969 : return lease(state_, key, std::move(found));
249 56167 : }
250 :
251 15312 : connection_pool::lease connection_pool::add(const std::string& key, const std::string& origin,
252 : std::shared_ptr<client_connection> connection) {
253 : {
254 15312 : std::lock_guard lock(state_->mutex);
255 15312 : auto& entry = state_->hosts[key];
256 15312 : if (entry.origin.empty()) entry.origin = origin;
257 15312 : ++entry.active;
258 15312 : ++state_->created;
259 15312 : }
260 15312 : connection->claim();
261 15312 : return lease(state_, key, std::move(connection));
262 : }
263 :
264 137 : size_t connection_pool::size() const {
265 137 : std::lock_guard lock(state_->mutex);
266 274 : return state_->idle.size();
267 137 : }
268 :
269 114 : connection_pool::stats connection_pool::get_stats() const {
270 114 : std::lock_guard lock(state_->mutex);
271 114 : stats result;
272 114 : result.created = state_->created;
273 114 : result.reused = state_->reused;
274 229 : for (const auto& [key, entry] : state_->hosts) {
275 115 : auto& host = result.hosts[entry.origin];
276 115 : host.idle += entry.idle.size();
277 115 : host.active += entry.active;
278 115 : result.idle += entry.idle.size();
279 115 : result.active += entry.active;
280 : }
281 228 : return result;
282 114 : }
283 :
284 7747 : void connection_pool::clear() {
285 7747 : std::vector<std::shared_ptr<client_connection>> closing;
286 : {
287 7747 : std::lock_guard lock(state_->mutex);
288 9407 : for (auto& entry : state_->idle) closing.push_back(std::move(entry.connection));
289 7747 : state_->idle.clear();
290 9445 : for (auto it = state_->hosts.begin(); it != state_->hosts.end();) {
291 1698 : it->second.idle.clear();
292 1893 : it = it->second.active == 0 ? state_->hosts.erase(it) : std::next(it);
293 : }
294 7747 : }
295 9407 : for (auto& connection : closing) connection->close();
296 7747 : }
297 :
298 : } // namespace thinger::http
|
