1 /*
2 * PgBouncer - Lightweight connection pooler for PostgreSQL.
3 *
4 * Copyright (c) 2007-2009 Marko Kreen, Skype Technologies OÜ
5 *
6 * Permission to use, copy, modify, and/or distribute this software for any
7 * purpose with or without fee is hereby granted, provided that the above
8 * copyright notice and this permission notice appear in all copies.
9 *
10 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
11 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
12 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
13 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
14 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
15 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
16 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
17 */
18
19 /*
20 * Periodic maintenance.
21 */
22
23 #include "bouncer.h"
24
25 /* do full maintenance 3x per second */
26 static struct timeval full_maint_period = {0, USEC / 3};
27 static struct event full_maint_ev;
28
29 /* close all sockets in server list */
close_server_list(struct StatList * sk_list,const char * reason)30 static void close_server_list(struct StatList *sk_list, const char *reason)
31 {
32 struct List *item, *tmp;
33 PgSocket *server;
34
35 statlist_for_each_safe(item, sk_list, tmp) {
36 server = container_of(item, PgSocket, head);
37 disconnect_server(server, true, "%s", reason);
38 }
39 }
40
close_client_list(struct StatList * sk_list,const char * reason)41 static void close_client_list(struct StatList *sk_list, const char *reason)
42 {
43 struct List *item, *tmp;
44 PgSocket *client;
45
46 statlist_for_each_safe(item, sk_list, tmp) {
47 client = container_of(item, PgSocket, head);
48 disconnect_client(client, true, "%s", reason);
49 }
50 }
51
suspend_socket(PgSocket * sk,bool force_suspend)52 bool suspend_socket(PgSocket *sk, bool force_suspend)
53 {
54 if (sk->suspended)
55 return true;
56
57 if (sbuf_is_empty(&sk->sbuf)) {
58 if (sbuf_pause(&sk->sbuf))
59 sk->suspended = true;
60 }
61
62 if (sk->suspended || !force_suspend)
63 return sk->suspended;
64
65 if (is_server_socket(sk))
66 disconnect_server(sk, true, "suspend_timeout");
67 else
68 disconnect_client(sk, true, "suspend_timeout");
69 return true;
70 }
71
72 /* suspend all sockets in socket list */
suspend_socket_list(struct StatList * list,bool force_suspend)73 static int suspend_socket_list(struct StatList *list, bool force_suspend)
74 {
75 struct List *item, *tmp;
76 PgSocket *sk;
77 int active = 0;
78
79 statlist_for_each_safe(item, list, tmp) {
80 sk = container_of(item, PgSocket, head);
81 if (!suspend_socket(sk, force_suspend))
82 active++;
83 }
84 return active;
85 }
86
87 /* resume all suspended sockets in socket list */
resume_socket_list(struct StatList * list)88 static void resume_socket_list(struct StatList *list)
89 {
90 struct List *item, *tmp;
91 PgSocket *sk;
92
93 statlist_for_each_safe(item, list, tmp) {
94 sk = container_of(item, PgSocket, head);
95 if (sk->suspended) {
96 sk->suspended = false;
97 sbuf_continue(&sk->sbuf);
98 }
99 }
100 }
101
102 /* resume all suspended sockets in all pools */
resume_sockets(void)103 static void resume_sockets(void)
104 {
105 struct List *item;
106 PgPool *pool;
107
108 statlist_for_each(item, &pool_list) {
109 pool = container_of(item, PgPool, head);
110 if (pool->db->admin)
111 continue;
112 resume_socket_list(&pool->active_client_list);
113 resume_socket_list(&pool->active_server_list);
114 resume_socket_list(&pool->idle_server_list);
115 resume_socket_list(&pool->used_server_list);
116 }
117 }
118
119 /* resume pools and listen sockets */
resume_all(void)120 void resume_all(void)
121 {
122 resume_sockets();
123 resume_pooler();
124 }
125
126 /*
127 * send test/reset query to server if needed
128 */
launch_recheck(PgPool * pool)129 static void launch_recheck(PgPool *pool)
130 {
131 const char *q = cf_server_check_query;
132 bool need_check = true;
133 PgSocket *server;
134 bool res = true;
135
136 /* find clean server */
137 while (1) {
138 server = first_socket(&pool->used_server_list);
139 if (!server)
140 return;
141 if (server->ready)
142 break;
143 disconnect_server(server, true, "idle server got dirty");
144 }
145
146 /* is the check needed? */
147 if (q == NULL || q[0] == 0) {
148 need_check = false;
149 } else if (cf_server_check_delay > 0) {
150 usec_t now = get_cached_time();
151 if (now - server->request_time < cf_server_check_delay)
152 need_check = false;
153 }
154
155 if (need_check) {
156 /* send test query, wait for result */
157 slog_debug(server, "P: checking: %s", q);
158 change_server_state(server, SV_TESTED);
159 SEND_generic(res, server, 'Q', "s", q);
160 if (!res)
161 disconnect_server(server, false, "test query failed");
162 } else {
163 /* make immediately available */
164 release_server(server);
165 }
166 }
167
168 /*
169 * make servers available
170 */
per_loop_activate(PgPool * pool)171 static void per_loop_activate(PgPool *pool)
172 {
173 struct List *item, *tmp;
174 PgSocket *client;
175 int sv_tested, sv_used;
176
177 /* if there is a cancel request waiting, open a new connection */
178 if (!statlist_empty(&pool->cancel_req_list)) {
179 launch_new_connection(pool);
180 return;
181 }
182
183 /* see if any server have been freed */
184 sv_tested = statlist_count(&pool->tested_server_list);
185 sv_used = statlist_count(&pool->used_server_list);
186 statlist_for_each_safe(item, &pool->waiting_client_list, tmp) {
187 client = container_of(item, PgSocket, head);
188 if (!statlist_empty(&pool->idle_server_list)) {
189
190 /* db not fully initialized after reboot */
191 if (client->wait_for_welcome && !pool->welcome_msg_ready) {
192 launch_new_connection(pool);
193 continue;
194 }
195
196 /* there is a ready server already */
197 activate_client(client);
198 } else if (sv_tested > 0) {
199 /* some connections are in testing process */
200 --sv_tested;
201 } else if (sv_used > 0) {
202 /* ask for more connections to be tested */
203 launch_recheck(pool);
204 --sv_used;
205 } else {
206 /* not enough connections */
207 launch_new_connection(pool);
208 break;
209 }
210 }
211 }
212
213 /*
214 * pause active clients
215 */
per_loop_pause(PgPool * pool)216 static int per_loop_pause(PgPool *pool)
217 {
218 int active = 0;
219
220 if (pool->db->admin)
221 return 0;
222
223 close_server_list(&pool->idle_server_list, "pause mode");
224 close_server_list(&pool->used_server_list, "pause mode");
225 close_server_list(&pool->new_server_list, "pause mode");
226
227 active += statlist_count(&pool->active_server_list);
228 active += statlist_count(&pool->tested_server_list);
229
230 return active;
231 }
232
233 /*
234 * suspend active clients and servers
235 */
per_loop_suspend(PgPool * pool,bool force_suspend)236 static int per_loop_suspend(PgPool *pool, bool force_suspend)
237 {
238 int active = 0;
239
240 if (pool->db->admin)
241 return 0;
242
243 active += suspend_socket_list(&pool->active_client_list, force_suspend);
244
245 /* this list is not suspendable, but still need force_suspend and counting */
246 active += suspend_socket_list(&pool->waiting_client_list, force_suspend);
247 if (active)
248 per_loop_activate(pool);
249
250 if (!active) {
251 active += suspend_socket_list(&pool->active_server_list, force_suspend);
252 active += suspend_socket_list(&pool->idle_server_list, force_suspend);
253
254 /* as all clients are done, no need for them */
255 close_server_list(&pool->tested_server_list, "close unsafe file descriptors on suspend");
256 close_server_list(&pool->used_server_list, "close unsafe file descriptors on suspend");
257 }
258
259 return active;
260 }
261
262 /*
263 * Count the servers in server_list that have close_needed set.
264 */
count_close_needed(struct StatList * server_list)265 static int count_close_needed(struct StatList *server_list)
266 {
267 struct List *item;
268 PgSocket *server;
269 int count = 0;
270
271 statlist_for_each(item, server_list) {
272 server = container_of(item, PgSocket, head);
273 if (server->close_needed)
274 count++;
275 }
276
277 return count;
278 }
279
280 /*
281 * Per-loop tasks for WAIT_CLOSE
282 */
per_loop_wait_close(PgPool * pool)283 static int per_loop_wait_close(PgPool *pool)
284 {
285 int count = 0;
286
287 if (pool->db->admin)
288 return 0;
289
290 count += count_close_needed(&pool->active_server_list);
291 count += count_close_needed(&pool->idle_server_list);
292 count += count_close_needed(&pool->new_server_list);
293 count += count_close_needed(&pool->tested_server_list);
294 count += count_close_needed(&pool->used_server_list);
295
296 return count;
297 }
298
299 /*
300 * this function is called for each event loop.
301 */
per_loop_maint(void)302 void per_loop_maint(void)
303 {
304 struct List *item;
305 PgPool *pool;
306 int active_count = 0;
307 int waiting_count = 0;
308 bool partial_pause = false;
309 bool partial_wait = false;
310 bool force_suspend = false;
311
312 if (cf_pause_mode == P_SUSPEND && cf_suspend_timeout > 0) {
313 usec_t stime = get_cached_time() - g_suspend_start;
314 if (stime >= cf_suspend_timeout)
315 force_suspend = true;
316 }
317
318 statlist_for_each(item, &pool_list) {
319 pool = container_of(item, PgPool, head);
320 if (pool->db->admin)
321 continue;
322 switch (cf_pause_mode) {
323 case P_NONE:
324 if (pool->db->db_paused) {
325 partial_pause = true;
326 active_count += per_loop_pause(pool);
327 } else {
328 per_loop_activate(pool);
329 }
330 break;
331 case P_PAUSE:
332 active_count += per_loop_pause(pool);
333 break;
334 case P_SUSPEND:
335 active_count += per_loop_suspend(pool, force_suspend);
336 break;
337 }
338
339 if (pool->db->db_wait_close) {
340 partial_wait = true;
341 waiting_count += per_loop_wait_close(pool);
342 }
343 }
344
345 switch (cf_pause_mode) {
346 case P_SUSPEND:
347 if (force_suspend) {
348 close_client_list(&login_client_list, "suspend_timeout");
349 } else {
350 active_count += statlist_count(&login_client_list);
351 }
352 /* fallthrough */
353 case P_PAUSE:
354 if (!active_count)
355 admin_pause_done();
356 break;
357 case P_NONE:
358 if (partial_pause && !active_count)
359 admin_pause_done();
360 break;
361 }
362
363 if (partial_wait && !waiting_count)
364 admin_wait_close_done();
365 }
366
367 /* maintaining clients in pool */
pool_client_maint(PgPool * pool)368 static void pool_client_maint(PgPool *pool)
369 {
370 struct List *item, *tmp;
371 usec_t now = get_cached_time();
372 PgSocket *client;
373 usec_t age;
374
375 /* force client_idle_timeout */
376 if (cf_client_idle_timeout > 0) {
377 statlist_for_each_safe(item, &pool->active_client_list, tmp) {
378 client = container_of(item, PgSocket, head);
379 Assert(client->state == CL_ACTIVE);
380 if (client->link)
381 continue;
382 if (now - client->request_time > cf_client_idle_timeout)
383 disconnect_client(client, true, "client_idle_timeout");
384 }
385 }
386
387 /* force timeouts for waiting queries */
388 if (cf_query_timeout > 0 || cf_query_wait_timeout > 0) {
389 statlist_for_each_safe(item, &pool->waiting_client_list, tmp) {
390 client = container_of(item, PgSocket, head);
391 Assert(client->state == CL_WAITING || client->state == CL_WAITING_LOGIN);
392 if (client->query_start == 0) {
393 age = now - client->request_time;
394 /* log_warning("query_start==0"); */
395 } else {
396 age = now - client->query_start;
397 }
398
399 if (cf_query_timeout > 0 && age > cf_query_timeout) {
400 disconnect_client(client, true, "query_timeout");
401 } else if (cf_query_wait_timeout > 0 && age > cf_query_wait_timeout) {
402 disconnect_client(client, true, "query_wait_timeout");
403 }
404 }
405 }
406
407 /* apply client_login_timeout to clients waiting for welcome pkt */
408 if (cf_client_login_timeout > 0 && !pool->welcome_msg_ready) {
409 statlist_for_each_safe(item, &pool->waiting_client_list, tmp) {
410 client = container_of(item, PgSocket, head);
411 if (!client->wait_for_welcome)
412 continue;
413 age = now - client->connect_time;
414 if (age > cf_client_login_timeout)
415 disconnect_client(client, true, "client_login_timeout (server down)");
416 }
417 }
418 }
419
check_unused_servers(PgPool * pool,struct StatList * slist,bool idle_test)420 static void check_unused_servers(PgPool *pool, struct StatList *slist, bool idle_test)
421 {
422 usec_t now = get_cached_time();
423 struct List *item, *tmp;
424 usec_t idle, age;
425 PgSocket *server;
426 usec_t lifetime_kill_gap = 0;
427
428 /*
429 * Calculate the time that disconnects because of server_lifetime
430 * must be separated. This avoids the need to re-launch lot
431 * of connections together.
432 */
433 if (pool_pool_size(pool) > 0)
434 lifetime_kill_gap = cf_server_lifetime / pool_pool_size(pool);
435
436 /* disconnect idle servers if needed */
437 statlist_for_each_safe(item, slist, tmp) {
438 server = container_of(item, PgSocket, head);
439
440 age = now - server->connect_time;
441 idle = now - server->request_time;
442
443 if (server->close_needed) {
444 disconnect_server(server, true, "database configuration changed");
445 } else if (server->state == SV_IDLE && !server->ready) {
446 disconnect_server(server, true, "SV_IDLE server got dirty");
447 } else if (server->state == SV_USED && !server->ready) {
448 disconnect_server(server, true, "SV_USED server got dirty");
449 } else if (cf_server_idle_timeout > 0 && idle > cf_server_idle_timeout
450 && (pool_min_pool_size(pool) == 0 || pool_connected_server_count(pool) > pool_min_pool_size(pool))) {
451 disconnect_server(server, true, "server idle timeout");
452 } else if (age >= cf_server_lifetime) {
453 if (pool->last_lifetime_disconnect + lifetime_kill_gap <= now) {
454 disconnect_server(server, true, "server lifetime over");
455 pool->last_lifetime_disconnect = now;
456 }
457 } else if (cf_pause_mode == P_PAUSE) {
458 disconnect_server(server, true, "pause mode");
459 } else if (idle_test && *cf_server_check_query) {
460 if (idle > cf_server_check_delay)
461 change_server_state(server, SV_USED);
462 }
463 }
464 }
465
466 /*
467 * Check pool size, close conns if too many. Makes pooler
468 * react faster to the case when admin decreased pool size.
469 */
check_pool_size(PgPool * pool)470 static void check_pool_size(PgPool *pool)
471 {
472 PgSocket *server;
473 int cur = pool_connected_server_count(pool);
474 int many = cur - (pool_pool_size(pool) + pool_res_pool_size(pool));
475
476 Assert(pool_pool_size(pool) >= 0);
477
478 while (many > 0) {
479 server = first_socket(&pool->used_server_list);
480 if (!server)
481 server = first_socket(&pool->idle_server_list);
482 if (!server)
483 break;
484 disconnect_server(server, true, "too many servers in the pool");
485 many--;
486 cur--;
487 }
488
489 /* launch extra connections to satisfy min_pool_size */
490 if (cur < pool_min_pool_size(pool) &&
491 cur < pool_pool_size(pool) &&
492 cf_pause_mode == P_NONE &&
493 cf_reboot == 0 &&
494 pool_client_count(pool) > 0)
495 {
496 log_debug("launching new connection to satisfy min_pool_size");
497 launch_new_connection(pool);
498 }
499 }
500
501 /* maintain servers in a pool */
pool_server_maint(PgPool * pool)502 static void pool_server_maint(PgPool *pool)
503 {
504 struct List *item, *tmp;
505 usec_t now = get_cached_time();
506 PgSocket *server;
507
508 /* find and disconnect idle servers */
509 check_unused_servers(pool, &pool->used_server_list, 0);
510 check_unused_servers(pool, &pool->tested_server_list, 0);
511 check_unused_servers(pool, &pool->idle_server_list, 1);
512
513 /* disconnect close_needed active servers if server_fast_close is set */
514 if (cf_server_fast_close) {
515 statlist_for_each_safe(item, &pool->active_server_list, tmp) {
516 server = container_of(item, PgSocket, head);
517 Assert(server->state == SV_ACTIVE);
518 if (server->ready && server->close_needed)
519 disconnect_server(server, true, "database configuration changed");
520 }
521 }
522
523 /* handle query_timeout and idle_transaction_timeout */
524 if (cf_query_timeout > 0 || cf_idle_transaction_timeout > 0) {
525 statlist_for_each_safe(item, &pool->active_server_list, tmp) {
526 usec_t age_client, age_server;
527
528 server = container_of(item, PgSocket, head);
529 Assert(server->state == SV_ACTIVE);
530 if (server->ready)
531 continue;
532
533 /*
534 * Note the different age calculations:
535 * query_timeout counts from the last request
536 * of the client (the client started the
537 * query), idle_transaction_timeout counts
538 * from the last request of the server (the
539 * server sent the idle information).
540 */
541 age_client = now - server->link->request_time;
542 age_server = now - server->request_time;
543
544 if (cf_query_timeout > 0 && age_client > cf_query_timeout) {
545 disconnect_server(server, true, "query timeout");
546 } else if (cf_idle_transaction_timeout > 0 &&
547 server->idle_tx &&
548 age_server > cf_idle_transaction_timeout)
549 {
550 disconnect_server(server, true, "idle transaction timeout");
551 }
552 }
553 }
554
555 /* find connections that got connect, but could not log in */
556 if (cf_server_connect_timeout > 0) {
557 statlist_for_each_safe(item, &pool->new_server_list, tmp) {
558 usec_t age;
559
560 server = container_of(item, PgSocket, head);
561 Assert(server->state == SV_LOGIN);
562
563 age = now - server->connect_time;
564 if (age > cf_server_connect_timeout)
565 disconnect_server(server, true, "connect timeout");
566 }
567 }
568
569 check_pool_size(pool);
570 }
571
cleanup_client_logins(void)572 static void cleanup_client_logins(void)
573 {
574 struct List *item, *tmp;
575 PgSocket *client;
576 usec_t age;
577 usec_t now = get_cached_time();
578
579 if (cf_client_login_timeout <= 0)
580 return;
581
582 statlist_for_each_safe(item, &login_client_list, tmp) {
583 client = container_of(item, PgSocket, head);
584 age = now - client->connect_time;
585 if (age > cf_client_login_timeout)
586 disconnect_client(client, true, "client_login_timeout");
587 }
588 }
589
cleanup_inactive_autodatabases(void)590 static void cleanup_inactive_autodatabases(void)
591 {
592 struct List *item, *tmp;
593 PgDatabase *db;
594 usec_t age;
595 usec_t now = get_cached_time();
596
597 if (cf_autodb_idle_timeout <= 0)
598 return;
599
600 /* now kill the old ones */
601 statlist_for_each_safe(item, &autodatabase_idle_list, tmp) {
602 db = container_of(item, PgDatabase, head);
603 if (db->db_paused)
604 continue;
605 age = now - db->inactive_time;
606 if (age > cf_autodb_idle_timeout) {
607 kill_database(db);
608 } else {
609 break;
610 }
611 }
612 }
613
614 /* full-scale maintenance, done only occasionally */
do_full_maint(evutil_socket_t sock,short flags,void * arg)615 static void do_full_maint(evutil_socket_t sock, short flags, void *arg)
616 {
617 struct List *item, *tmp;
618 PgPool *pool;
619 PgDatabase *db;
620
621 static unsigned int seq;
622 seq++;
623
624 /*
625 * Avoid doing anything that may surprise other pgbouncer.
626 */
627 if (cf_pause_mode == P_SUSPEND)
628 return;
629
630 statlist_for_each_safe(item, &pool_list, tmp) {
631 pool = container_of(item, PgPool, head);
632 if (pool->db->admin)
633 continue;
634 pool_server_maint(pool);
635 pool_client_maint(pool);
636
637 /* is autodb active? */
638 if (pool->db->db_auto && pool->db->inactive_time == 0) {
639 if (pool_client_count(pool) > 0 || pool_server_count(pool) > 0)
640 pool->db->active_stamp = seq;
641 }
642 }
643
644 /* find inactive autodbs */
645 statlist_for_each_safe(item, &database_list, tmp) {
646 db = container_of(item, PgDatabase, head);
647 if (db->db_auto && db->inactive_time == 0) {
648 if (db->active_stamp == seq)
649 continue;
650 db->inactive_time = get_cached_time();
651 statlist_remove(&database_list, &db->head);
652 statlist_append(&autodatabase_idle_list, &db->head);
653 }
654 }
655
656 cleanup_inactive_autodatabases();
657
658 cleanup_client_logins();
659
660 if (cf_shutdown == 1 && get_active_server_count() == 0) {
661 log_info("server connections dropped, exiting");
662 cf_shutdown = 2;
663 event_base_loopbreak(pgb_event_base);
664 return;
665 }
666
667 if (requires_auth_file(cf_auth_type))
668 loader_users_check();
669
670 adns_zone_cache_maint(adns);
671 }
672
673 /* first-time initialization */
janitor_setup(void)674 void janitor_setup(void)
675 {
676 /* launch maintenance */
677 event_assign(&full_maint_ev, pgb_event_base, -1, EV_PERSIST, do_full_maint, NULL);
678 event_add(&full_maint_ev, &full_maint_period);
679 }
680
kill_pool(PgPool * pool)681 void kill_pool(PgPool *pool)
682 {
683 const char *reason = "database removed";
684
685 close_client_list(&pool->active_client_list, reason);
686 close_client_list(&pool->waiting_client_list, reason);
687 close_client_list(&pool->cancel_req_list, reason);
688
689 close_server_list(&pool->active_server_list, reason);
690 close_server_list(&pool->idle_server_list, reason);
691 close_server_list(&pool->used_server_list, reason);
692 close_server_list(&pool->tested_server_list, reason);
693 close_server_list(&pool->new_server_list, reason);
694
695 pktbuf_free(pool->welcome_msg);
696
697 list_del(&pool->map_head);
698 statlist_remove(&pool_list, &pool->head);
699 varcache_clean(&pool->orig_vars);
700 slab_free(pool_cache, pool);
701 }
702
kill_database(PgDatabase * db)703 void kill_database(PgDatabase *db)
704 {
705 PgPool *pool;
706 struct List *item, *tmp;
707
708 log_warning("dropping database '%s' as it does not exist anymore or inactive auto-database", db->name);
709
710 statlist_for_each_safe(item, &pool_list, tmp) {
711 pool = container_of(item, PgPool, head);
712 if (pool->db == db)
713 kill_pool(pool);
714 }
715
716 pktbuf_free(db->startup_params);
717 free(db->host);
718
719 if (db->forced_user)
720 slab_free(user_cache, db->forced_user);
721 if (db->connect_query)
722 free(db->connect_query);
723 if (db->inactive_time) {
724 statlist_remove(&autodatabase_idle_list, &db->head);
725 } else {
726 statlist_remove(&database_list, &db->head);
727 }
728 aatree_destroy(&db->user_tree);
729 slab_free(db_cache, db);
730 }
731
732 /* as [pgbouncer] section can be loaded after databases,
733 there's need for review */
config_postprocess(void)734 void config_postprocess(void)
735 {
736 struct List *item, *tmp;
737 PgDatabase *db;
738
739 statlist_for_each_safe(item, &database_list, tmp) {
740 db = container_of(item, PgDatabase, head);
741 if (db->db_dead) {
742 kill_database(db);
743 continue;
744 }
745 }
746 }
747