2 * PgBouncer - Lightweight connection pooler for PostgreSQL.
4 * Copyright (c) 2007-2009 Marko Kreen, Skype Technologies OÜ
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.
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.
20 * Handling of pooler listening sockets
25 static int fd_net = 0;
26 static int fd_unix = 0;
28 static struct event ev_net;
29 static struct event ev_unix;
31 /* if sockets are registered in libevent */
32 static bool reg_net = false;
33 static bool reg_unix = false;
35 /* should listening sockets be active or suspended? */
36 static bool pooler_active = false;
38 /* on accept() failure sleep 5 seconds */
39 static struct event ev_err;
40 static struct timeval err_timeout = {5, 0};
42 /* atexit() cleanup func */
43 static void cleanup_unix_socket(void)
47 /* avoid cleanup if exit() while suspended */
51 snprintf(fn, sizeof(fn), "%s/.s.PGSQL.%d",
52 cf_unix_socket_dir, cf_listen_port);
56 void get_pooler_fds(int *p_net, int *p_unix)
62 static int create_unix_socket(const char *socket_dir, int listen_port)
64 struct sockaddr_un un;
69 /* fill sockaddr struct */
70 memset(&un, 0, sizeof(un));
71 un.sun_family = AF_UNIX;
72 snprintf(un.sun_path, sizeof(un.sun_path),
73 "%s/.s.PGSQL.%d", socket_dir, listen_port);
75 /* check for lockfile */
76 snprintf(lockfile, sizeof(lockfile), "%s.lock", un.sun_path);
77 res = lstat(lockfile, &st);
79 fatal("unix port %d is in use", listen_port);
81 /* expect old bouncer gone */
85 sock = socket(PF_UNIX, SOCK_STREAM, 0);
87 fatal_perror("socket");
90 res = bind(sock, (const struct sockaddr *)&un, sizeof(un));
94 /* remove socket on shutdown */
95 atexit(cleanup_unix_socket);
97 /* set common options */
98 tune_socket(sock, true);
100 /* finally, accept connections */
101 res = listen(sock, 100);
103 fatal_perror("listen");
105 res = chmod(un.sun_path, 0777);
107 fatal_perror("chmod");
109 log_info("listening on unix:%s", un.sun_path);
115 * Notify pooler only when also data is arrived.
117 * optval specifies how long after connection attempt to wait for data.
119 * Related to tcp_synack_retries sysctl, default 5 (corresponds 180 secs).
121 * SO_ACCEPTFILTER needs to be set after listern(), maybe TCP_DEFER_ACCEPT too.
123 static void tune_accept(int sock, bool on)
125 const char *act = on ? "install" : "uninstall";
127 #ifdef TCP_DEFER_ACCEPT
128 int val = 45; /* fixme: proper value */
129 socklen_t vlen = sizeof(val);
130 res = getsockopt(sock, IPPROTO_TCP, TCP_DEFER_ACCEPT, &val, &vlen);
131 log_noise("old TCP_DEFER_ACCEPT on %d = %d", sock, val);
133 log_noise("%s TCP_DEFER_ACCEPT on %d", act, sock);
134 res = setsockopt(sock, IPPROTO_TCP, TCP_DEFER_ACCEPT, &val, sizeof(val));
137 #ifdef SO_ACCEPTFILTER
138 struct accept_filter_arg af, *afp = on ? &af : NULL;
139 socklen_t af_len = on ? sizeof(af) : 0;
140 memset(&af, 0, sizeof(af));
141 strcpy(af.af_name, "dataready");
142 log_noise("%s SO_ACCEPTFILTER on %d", act, sock);
143 res = setsockopt(sock, SOL_SOCKET, SO_ACCEPTFILTER, afp, af_len);
148 log_warning("tune_accept: %s TCP_DEFER_ACCEPT/SO_ACCEPTFILTER: %s",
149 act, strerror(errno));
152 void pooler_tune_accept(bool on)
155 tune_accept(fd_net, on);
158 static int create_net_socket(const char *listen_addr, int listen_port)
161 struct sockaddr_in sa;
166 sock = socket(AF_INET, SOCK_STREAM, 0);
168 fatal_perror("socket");
171 memset(&sa, 0, sizeof(sa));
172 sa.sin_family = AF_INET;
173 sa.sin_port = htons(cf_listen_port);
174 if (strcmp(listen_addr, "*") == 0) {
175 sa.sin_addr.s_addr = htonl(INADDR_ANY);
177 sa.sin_addr.s_addr = inet_addr(listen_addr);
178 if (sa.sin_addr.s_addr == INADDR_NONE)
179 fatal("cannot parse addr: '%s'", listen_addr);
182 /* relaxed binding */
184 res = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
186 fatal_perror("setsockopt");
188 /* bind to address */
189 res = bind(sock, (struct sockaddr *)&sa, sizeof(sa));
191 fatal_perror("bind");
193 /* set common options */
194 tune_socket(sock, false);
196 /* make it accept connections */
197 res = listen(sock, 100);
199 fatal_perror("listen");
201 tune_accept(sock, cf_tcp_defer_accept);
203 log_info("listening on %s:%d", cf_listen_addr, cf_listen_port);
208 static void err_wait_func(int sock, short flags, void *arg)
210 if (cf_pause_mode != P_SUSPEND)
214 static const char *addrpair(const PgAddr *src, const PgAddr *dst)
216 static char ip1buf[64], ip2buf[64], buf[256];
217 const char *ip1, *ip2;
221 ip1 = inet_ntop(AF_INET, &src->ip_addr, ip1buf, sizeof(ip1buf));
223 ip1 = strerror(errno);
224 ip2 = inet_ntop(AF_INET, &dst->ip_addr, ip2buf, sizeof(ip2buf));
226 ip2 = strerror(errno);
227 snprintf(buf, sizeof(buf), "%s:%d -> %s:%d",
228 ip1, src->port, ip2, dst->port);
232 static const char *conninfo(const PgSocket *sk)
234 if (is_server_socket(sk))
235 return addrpair(&sk->local_addr, &sk->remote_addr);
237 return addrpair(&sk->remote_addr, &sk->local_addr);
240 /* got new connection, associate it with client struct */
241 static void pool_accept(int sock, short flags, void *is_unix)
246 struct sockaddr_in in;
247 struct sockaddr_un un;
250 socklen_t len = sizeof(addr);
252 if(!(flags & EV_READ)) {
253 log_warning("No EV_READ in pool_accept");
258 fd = safe_accept(sock, &addr.sa, &len);
262 else if (errno == ECONNABORTED)
266 * probably fd limit, pointless to try often
267 * wait a bit, hope that admin resolves somehow
269 log_error("accept() failed: %s", strerror(errno));
270 evtimer_set(&ev_err, err_wait_func, NULL);
271 safe_evtimer_add(&ev_err, &err_timeout);
276 log_noise("new fd from accept=%d", fd);
281 log_noise("getuid(): %d", (int)getuid());
282 if (getpeereid(fd, &uid, &gid) >= 0)
283 log_noise("unix peer uid: %d", (int)uid);
285 log_warning("unix peer uid failed: %s", strerror(errno));
287 client = accept_client(fd, NULL, true);
289 client = accept_client(fd, &addr.in, false);
293 slog_debug(client, "P: got connection: %s", conninfo(client));
296 * there may be several clients waiting,
297 * avoid context switch by looping
302 bool use_pooler_socket(int sock, bool is_unix)
304 tune_socket(sock, is_unix);
313 void suspend_pooler(void)
315 pooler_active = false;
317 if (fd_net && reg_net) {
318 if (event_del(&ev_net) < 0) {
319 log_warning("suspend_pooler, event_del: %s", strerror(errno));
324 if (fd_unix && reg_unix) {
325 if (event_del(&ev_unix) < 0) {
326 log_warning("suspend_pooler, event_del: %s", strerror(errno));
333 void resume_pooler(void)
335 pooler_active = true;
337 if (fd_unix && !reg_unix) {
338 event_set(&ev_unix, fd_unix, EV_READ | EV_PERSIST, pool_accept, "1");
339 if (event_add(&ev_unix, NULL) < 0) {
340 log_warning("event_add failed: %s", strerror(errno));
346 if (fd_net && !reg_net) {
347 event_set(&ev_net, fd_net, EV_READ | EV_PERSIST, pool_accept, NULL);
348 if (event_add(&ev_net, NULL) < 0) {
349 log_warning("event_add failed: %s", strerror(errno));
355 /* listen on socket - should happen after all other initializations */
356 void pooler_setup(void)
358 if (cf_listen_addr && !fd_net)
359 fd_net = create_net_socket(cf_listen_addr, cf_listen_port);
361 if (*cf_unix_socket_dir && !fd_unix)
362 fd_unix = create_unix_socket(cf_unix_socket_dir, cf_listen_port);
364 if (!fd_net && !fd_unix)
365 fatal("nowhere to listen on");
370 /* retry previously failed suspend_pooler() / resume_pooler() */
371 void per_loop_pooler_maint(void)
374 if ((fd_unix && !reg_unix) || (fd_net && !reg_net))
377 if ((fd_unix && reg_unix) || (fd_net && reg_net))