[MEDIUM] separate protocol-level accept() from the frontend's
For a long time we had two large accept() functions, one for TCP
sockets instanciating proxies, and another one for UNIX sockets
instanciating the stats interface.
A lot of code was duplicated and both did not work exactly the same way.
Now we have a stream_sock layer accept() called for either TCP or UNIX
sockets, and this function calls the frontend-specific accept() function
which does the rest of the frontend-specific initialisation.
Some code is still duplicated (session & task allocation, stream interface
initialization), and might benefit from having an intermediate session-level
accept() callback to perform such initializations. Still there are some
minor differences that need to be addressed first. For instance, the monitor
nets should only be checked for proxies and not for other connection templates.
Last, we renamed l->private as l->frontend. The "private" pointer in
the listener is only used to store a frontend, so let's rename it to
eliminate this ambiguity. When we later support detached listeners
(eg: FTP), we'll add another field to avoid the confusion.
diff --git a/src/cfgparse.c b/src/cfgparse.c
index cc82544..06ebf36 100644
--- a/src/cfgparse.c
+++ b/src/cfgparse.c
@@ -42,6 +42,7 @@
#include <proto/buffers.h>
#include <proto/checks.h>
#include <proto/dumpstats.h>
+#include <proto/frontend.h>
#include <proto/httperr.h>
#include <proto/lb_chash.h>
#include <proto/lb_fwlc.h>
@@ -5349,8 +5350,8 @@
listener->maxconn = curproxy->maxconn;
listener->backlog = curproxy->backlog;
listener->timeout = &curproxy->timeout.client;
- listener->accept = event_accept;
- listener->private = curproxy;
+ listener->accept = frontend_accept;
+ listener->frontend = curproxy;
listener->handler = process_session;
listener->analysers |= curproxy->fe_req_ana;
diff --git a/src/dumpstats.c b/src/dumpstats.c
index 15174b8..1856f21 100644
--- a/src/dumpstats.c
+++ b/src/dumpstats.c
@@ -1,7 +1,7 @@
/*
- * Functions dedicated to statistics output
+ * Functions dedicated to statistics output and the stats socket
*
- * Copyright 2000-2009 Willy Tarreau <w@1wt.eu>
+ * Copyright 2000-2010 Willy Tarreau <w@1wt.eu>
* Copyright 2007-2009 Krzysztof Piotr Oledzki <ole@ans.pl>
*
* This program is free software; you can redistribute it and/or
@@ -44,12 +44,14 @@
#include <proto/dumpstats.h>
#include <proto/fd.h>
#include <proto/freq_ctr.h>
+#include <proto/log.h>
#include <proto/pipe.h>
#include <proto/proto_uxst.h>
#include <proto/proxy.h>
#include <proto/session.h>
#include <proto/server.h>
#include <proto/stream_interface.h>
+#include <proto/stream_sock.h>
#include <proto/task.h>
const char stats_sock_usage_msg[] =
@@ -73,6 +75,167 @@
"Permission denied\n"
"";
+/* This function is called from the protocol layer accept() in order to instanciate
+ * a new stats socket. It returns a positive value upon success, 0 if the connection
+ * needs to be closed and ignored, or a negative value upon critical failure.
+ */
+int stats_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
+{
+ struct proxy *p = l->frontend; /* attached frontend */
+ struct session *s;
+ struct task *t;
+
+ if ((s = pool_alloc2(pool2_session)) == NULL) {
+ Alert("out of memory in stats_accept().\n");
+ goto out_close;
+ }
+
+ LIST_ADDQ(&sessions, &s->list);
+ LIST_INIT(&s->back_refs);
+
+ s->flags = 0;
+ s->term_trace = 0;
+ s->cli_addr = *addr;
+
+ if ((t = task_new()) == NULL) {
+ Alert("out of memory in stats_accept().\n");
+ goto out_free_session;
+ }
+
+ t->process = l->handler;
+ t->context = s;
+ t->nice = l->nice;
+
+ s->task = t;
+ s->listener = l;
+ s->fe = s->be = p;
+
+ s->req = s->rep = NULL; /* will be allocated later */
+
+ s->si[0].state = s->si[0].prev_state = SI_ST_EST;
+ s->si[0].err_type = SI_ET_NONE;
+ s->si[0].err_loc = NULL;
+ s->si[0].owner = t;
+ s->si[0].update = stream_sock_data_finish;
+ s->si[0].shutr = stream_sock_shutr;
+ s->si[0].shutw = stream_sock_shutw;
+ s->si[0].chk_rcv = stream_sock_chk_rcv;
+ s->si[0].chk_snd = stream_sock_chk_snd;
+ s->si[0].connect = NULL;
+ s->si[0].iohandler = NULL;
+ s->si[0].fd = cfd;
+ s->si[0].flags = SI_FL_NONE;
+ if (s->fe->options2 & PR_O2_INDEPSTR)
+ s->si[0].flags |= SI_FL_INDEP_STR;
+ s->si[0].exp = TICK_ETERNITY;
+
+ s->si[1].state = s->si[1].prev_state = SI_ST_INI;
+ s->si[1].err_type = SI_ET_NONE;
+ s->si[1].err_loc = NULL;
+ s->si[1].owner = t;
+ s->si[1].exp = TICK_ETERNITY;
+ s->si[1].fd = -1; /* just to help with debugging */
+ s->si[1].flags = SI_FL_NONE;
+ if (s->be->options2 & PR_O2_INDEPSTR)
+ s->si[1].flags |= SI_FL_INDEP_STR;
+
+ stream_int_register_handler(&s->si[1], stats_io_handler);
+ s->si[1].private = s;
+ s->si[1].st1 = 0;
+ s->si[1].st0 = STAT_CLI_INIT;
+
+ s->srv = s->prev_srv = s->srv_conn = NULL;
+ s->pend_pos = NULL;
+
+ s->store_count = 0;
+
+ memset(&s->logs, 0, sizeof(s->logs));
+ memset(&s->txn, 0, sizeof(s->txn));
+
+ s->logs.accept_date = date; /* user-visible date for logging */
+ s->logs.tv_accept = now; /* corrected date for internal use */
+
+ s->data_state = DATA_ST_INIT;
+ s->data_source = DATA_SRC_NONE;
+ s->uniq_id = totalconn;
+ proxy_inc_fe_ctr(l, p); /* note: cum_beconn will be increased once assigned */
+
+ if (fcntl(cfd, F_SETFL, O_NONBLOCK) == -1) {
+ Alert("accept(): cannot set the socket in non blocking mode. Giving up.\n");
+ goto out_free_task;
+ }
+
+ if ((s->req = pool_alloc2(pool2_buffer)) == NULL)
+ goto out_free_task;
+
+ s->req->size = global.tune.bufsize;
+ buffer_init(s->req);
+ s->req->prod = &s->si[0];
+ s->req->cons = &s->si[1];
+ s->si[0].ib = s->si[1].ob = s->req;
+ s->req->flags |= BF_READ_ATTACHED; /* the producer is already connected */
+ s->req->flags |= BF_READ_DONTWAIT; /* we plan to read small requests */
+
+ s->req->analysers = l->analysers;
+
+ s->req->wto = TICK_ETERNITY;
+ s->req->cto = TICK_ETERNITY;
+ s->req->rto = TICK_ETERNITY;
+
+ if ((s->rep = pool_alloc2(pool2_buffer)) == NULL)
+ goto out_free_req;
+
+ s->rep->size = global.tune.bufsize;
+ buffer_init(s->rep);
+
+ s->rep->prod = &s->si[1];
+ s->rep->cons = &s->si[0];
+ s->si[0].ob = s->si[1].ib = s->rep;
+
+ s->rep->rto = TICK_ETERNITY;
+ s->rep->cto = TICK_ETERNITY;
+ s->rep->wto = TICK_ETERNITY;
+
+ s->req->rex = TICK_ETERNITY;
+ s->req->wex = TICK_ETERNITY;
+ s->req->analyse_exp = TICK_ETERNITY;
+ s->rep->rex = TICK_ETERNITY;
+ s->rep->wex = TICK_ETERNITY;
+ s->rep->analyse_exp = TICK_ETERNITY;
+
+ t->expire = TICK_ETERNITY;
+
+ if (l->timeout) {
+ s->req->rto = *l->timeout;
+ s->rep->wto = *l->timeout;
+ }
+
+ fd_insert(cfd);
+ fdtab[cfd].owner = &s->si[0];
+ fdtab[cfd].state = FD_STREADY;
+ fdtab[cfd].cb[DIR_RD].f = l->proto->read;
+ fdtab[cfd].cb[DIR_RD].b = s->req;
+ fdtab[cfd].cb[DIR_WR].f = l->proto->write;
+ fdtab[cfd].cb[DIR_WR].b = s->rep;
+ fdinfo[cfd].peeraddr = (struct sockaddr *)&s->cli_addr;
+ fdinfo[cfd].peerlen = sizeof(s->cli_addr);
+
+ EV_FD_SET(cfd, DIR_RD);
+
+ task_wakeup(t, TASK_WOKEN_INIT);
+ return 1;
+
+ out_free_req:
+ pool_free2(pool2_buffer, s->req);
+ out_free_task:
+ task_free(t);
+ out_free_session:
+ LIST_DEL(&s->list);
+ pool_free2(pool2_session, s);
+ out_close:
+ return -1;
+}
+
/* This function parses a "stats" statement in the "global" section. It returns
* -1 if there is any error, otherwise zero. If it returns -1, it may write an
* error message into ther <err> buffer, for at most <errlen> bytes, trailing
@@ -124,15 +287,16 @@
global.stats_fe->last_change = now.tv_sec;
global.stats_fe->id = strdup("GLOBAL");
global.stats_fe->cap = PR_CAP_FE;
+ global.stats_fe->maxconn = global.stats_sock.maxconn;
}
global.stats_sock.state = LI_INIT;
global.stats_sock.options = LI_O_NONE;
- global.stats_sock.accept = uxst_event_accept;
+ global.stats_sock.accept = stats_accept;
global.stats_sock.handler = process_session;
global.stats_sock.analysers = 0;
global.stats_sock.nice = -64; /* we want to boost priority for local stats */
- global.stats_sock.private = global.stats_fe; /* must point to the frontend */
+ global.stats_sock.frontend = global.stats_fe;
global.stats_sock.perm.ux.level = ACCESS_LVL_OPER; /* default access level */
global.stats_fe->timeout.client = MS_TO_TICKS(10000); /* default timeout of 10 seconds */
@@ -224,6 +388,8 @@
global.maxsock -= global.stats_sock.maxconn;
global.stats_sock.maxconn = maxconn;
global.maxsock += global.stats_sock.maxconn;
+ if (global.stats_fe)
+ global.stats_fe->maxconn = global.stats_sock.maxconn;
}
else {
snprintf(err, errlen, "'stats' only supports 'socket', 'maxconn' and 'timeout' in 'global' section");
diff --git a/src/frontend.c b/src/frontend.c
index bc354b9..ef1001d 100644
--- a/src/frontend.c
+++ b/src/frontend.c
@@ -53,443 +53,359 @@
s->flags |= SN_FRT_ADDR_SET;
}
-/*
- * FIXME: This should move to the STREAM_SOCK code then split into TCP and HTTP.
+/* This function is called from the protocol layer accept() in order to instanciate
+ * a new proxy. It returns a positive value upon success, 0 if the connection needs
+ * to be closed and ignored, or a negative value upon critical failure.
*/
-
-/*
- * this function is called on a read event from a listen socket, corresponding
- * to an accept. It tries to accept as many connections as possible.
- * It returns 0.
- */
-int event_accept(int fd) {
- struct listener *l = fdtab[fd].owner;
- struct proxy *p = (struct proxy *)l->private; /* attached frontend */
+int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
+{
+ struct proxy *p = l->frontend;
struct session *s;
struct http_txn *txn;
struct task *t;
- int cfd;
- int max_accept = global.tune.maxaccept;
- if (p->fe_sps_lim) {
- int max = freq_ctr_remain(&p->fe_sess_per_sec, p->fe_sps_lim, 0);
- if (max_accept > max)
- max_accept = max;
+ if ((s = pool_alloc2(pool2_session)) == NULL) { /* disable this proxy for a while */
+ Alert("out of memory in event_accept().\n");
+ goto out_close;
}
- while (p->feconn < p->maxconn && actconn < global.maxconn && max_accept--) {
- struct sockaddr_storage addr;
- socklen_t laddr = sizeof(addr);
+ LIST_ADDQ(&sessions, &s->list);
+ LIST_INIT(&s->back_refs);
- if ((cfd = accept(fd, (struct sockaddr *)&addr, &laddr)) == -1) {
- switch (errno) {
- case EAGAIN:
- case EINTR:
- case ECONNABORTED:
- return 0; /* nothing more to accept */
- case ENFILE:
- send_log(p, LOG_EMERG,
- "Proxy %s reached system FD limit at %d. Please check system tunables.\n",
- p->id, maxfd);
- return 0;
- case EMFILE:
- send_log(p, LOG_EMERG,
- "Proxy %s reached process FD limit at %d. Please check 'ulimit-n' and restart.\n",
- p->id, maxfd);
- return 0;
- case ENOBUFS:
- case ENOMEM:
- send_log(p, LOG_EMERG,
- "Proxy %s reached system memory limit at %d sockets. Please check system tunables.\n",
- p->id, maxfd);
- return 0;
- default:
- return 0;
- }
- }
+ s->flags = 0;
+ s->term_trace = 0;
+ s->cli_addr = *addr;
- if (l->nbconn >= l->maxconn) {
- /* too many connections, we shoot this one and return.
- * FIXME: it would be better to simply switch the listener's
- * state to LI_FULL and disable the FD. We could re-enable
- * it upon fd_delete(), but this requires all protocols to
- * be switched.
- */
- goto out_close;
+ /* if this session comes from a known monitoring system, we want to ignore
+ * it as soon as possible, which means closing it immediately for TCP.
+ */
+ if (p->mon_mask.s_addr &&
+ addr->ss_family == AF_INET &&
+ (((struct sockaddr_in *)addr)->sin_addr.s_addr & p->mon_mask.s_addr) == p->mon_net.s_addr) {
+ if (p->mode == PR_MODE_TCP) {
+ pool_free2(pool2_session, s);
+ return 0;
}
+ s->flags |= SN_MONITOR;
+ }
- if ((s = pool_alloc2(pool2_session)) == NULL) { /* disable this proxy for a while */
- Alert("out of memory in event_accept().\n");
- EV_FD_CLR(fd, DIR_RD);
- p->state = PR_STIDLE;
- goto out_close;
- }
+ if ((t = task_new()) == NULL) { /* disable this proxy for a while */
+ Alert("out of memory in event_accept().\n");
+ goto out_free_session;
+ }
- LIST_ADDQ(&sessions, &s->list);
- LIST_INIT(&s->back_refs);
+ if ((fcntl(cfd, F_SETFL, O_NONBLOCK) == -1) ||
+ (setsockopt(cfd, IPPROTO_TCP, TCP_NODELAY,
+ (char *) &one, sizeof(one)) == -1)) {
+ Alert("accept(): cannot set the socket in non blocking mode. Giving up\n");
+ goto out_free_task;
+ }
- s->flags = 0;
- s->term_trace = 0;
+ if (p->options & PR_O_TCP_CLI_KA)
+ setsockopt(cfd, SOL_SOCKET, SO_KEEPALIVE, (char *) &one, sizeof(one));
- /* if this session comes from a known monitoring system, we want to ignore
- * it as soon as possible, which means closing it immediately for TCP.
- */
- if (addr.ss_family == AF_INET &&
- p->mon_mask.s_addr &&
- (((struct sockaddr_in *)&addr)->sin_addr.s_addr & p->mon_mask.s_addr) == p->mon_net.s_addr) {
- if (p->mode == PR_MODE_TCP) {
- close(cfd);
- pool_free2(pool2_session, s);
- continue;
- }
- s->flags |= SN_MONITOR;
- }
+ if (p->options & PR_O_TCP_NOLING)
+ setsockopt(cfd, SOL_SOCKET, SO_LINGER, (struct linger *) &nolinger, sizeof(struct linger));
- if ((t = task_new()) == NULL) { /* disable this proxy for a while */
- Alert("out of memory in event_accept().\n");
- EV_FD_CLR(fd, DIR_RD);
- p->state = PR_STIDLE;
- goto out_free_session;
- }
+ if (global.tune.client_sndbuf)
+ setsockopt(cfd, SOL_SOCKET, SO_SNDBUF, &global.tune.client_sndbuf, sizeof(global.tune.client_sndbuf));
- s->cli_addr = addr;
- if (cfd >= global.maxsock) {
- Alert("accept(): not enough free sockets. Raise -n argument. Giving up.\n");
- goto out_free_task;
- }
+ if (global.tune.client_rcvbuf)
+ setsockopt(cfd, SOL_SOCKET, SO_RCVBUF, &global.tune.client_rcvbuf, sizeof(global.tune.client_rcvbuf));
- if ((fcntl(cfd, F_SETFL, O_NONBLOCK) == -1) ||
- (setsockopt(cfd, IPPROTO_TCP, TCP_NODELAY,
- (char *) &one, sizeof(one)) == -1)) {
- Alert("accept(): cannot set the socket in non blocking mode. Giving up\n");
- goto out_free_task;
- }
+ t->process = l->handler;
+ t->context = s;
+ t->nice = l->nice;
- if (p->options & PR_O_TCP_CLI_KA)
- setsockopt(cfd, SOL_SOCKET, SO_KEEPALIVE, (char *) &one, sizeof(one));
+ s->task = t;
+ s->listener = l;
- if (p->options & PR_O_TCP_NOLING)
- setsockopt(cfd, SOL_SOCKET, SO_LINGER, (struct linger *) &nolinger, sizeof(struct linger));
+ /* Note: initially, the session's backend points to the frontend.
+ * This changes later when switching rules are executed or
+ * when the default backend is assigned.
+ */
+ s->be = s->fe = p;
- if (global.tune.client_sndbuf)
- setsockopt(cfd, SOL_SOCKET, SO_SNDBUF, &global.tune.client_sndbuf, sizeof(global.tune.client_sndbuf));
+ s->req = s->rep = NULL; /* will be allocated later */
- if (global.tune.client_rcvbuf)
- setsockopt(cfd, SOL_SOCKET, SO_RCVBUF, &global.tune.client_rcvbuf, sizeof(global.tune.client_rcvbuf));
+ s->si[0].state = s->si[0].prev_state = SI_ST_EST;
+ s->si[0].err_type = SI_ET_NONE;
+ s->si[0].err_loc = NULL;
+ s->si[0].owner = t;
+ s->si[0].update = stream_sock_data_finish;
+ s->si[0].shutr = stream_sock_shutr;
+ s->si[0].shutw = stream_sock_shutw;
+ s->si[0].chk_rcv = stream_sock_chk_rcv;
+ s->si[0].chk_snd = stream_sock_chk_snd;
+ s->si[0].connect = NULL;
+ s->si[0].iohandler = NULL;
+ s->si[0].fd = cfd;
+ s->si[0].flags = SI_FL_NONE | SI_FL_CAP_SPLTCP; /* TCP splicing capable */
+ if (s->fe->options2 & PR_O2_INDEPSTR)
+ s->si[0].flags |= SI_FL_INDEP_STR;
+ s->si[0].exp = TICK_ETERNITY;
- t->process = l->handler;
- t->context = s;
- t->nice = l->nice;
+ s->si[1].state = s->si[1].prev_state = SI_ST_INI;
+ s->si[1].err_type = SI_ET_NONE;
+ s->si[1].err_loc = NULL;
+ s->si[1].owner = t;
+ s->si[1].update = stream_sock_data_finish;
+ s->si[1].shutr = stream_sock_shutr;
+ s->si[1].shutw = stream_sock_shutw;
+ s->si[1].chk_rcv = stream_sock_chk_rcv;
+ s->si[1].chk_snd = stream_sock_chk_snd;
+ s->si[1].connect = tcpv4_connect_server;
+ s->si[1].iohandler = NULL;
+ s->si[1].exp = TICK_ETERNITY;
+ s->si[1].fd = -1; /* just to help with debugging */
+ s->si[1].flags = SI_FL_NONE;
+ if (s->be->options2 & PR_O2_INDEPSTR)
+ s->si[1].flags |= SI_FL_INDEP_STR;
- s->task = t;
- s->listener = l;
+ s->srv = s->prev_srv = s->srv_conn = NULL;
+ s->pend_pos = NULL;
+ s->conn_retries = s->be->conn_retries;
- /* Note: initially, the session's backend points to the frontend.
- * This changes later when switching rules are executed or
- * when the default backend is assigned.
- */
- s->be = s->fe = p;
-
- s->req = s->rep = NULL; /* will be allocated later */
+ /* init store persistence */
+ s->store_count = 0;
- s->si[0].state = s->si[0].prev_state = SI_ST_EST;
- s->si[0].err_type = SI_ET_NONE;
- s->si[0].err_loc = NULL;
- s->si[0].owner = t;
- s->si[0].update = stream_sock_data_finish;
- s->si[0].shutr = stream_sock_shutr;
- s->si[0].shutw = stream_sock_shutw;
- s->si[0].chk_rcv = stream_sock_chk_rcv;
- s->si[0].chk_snd = stream_sock_chk_snd;
- s->si[0].connect = NULL;
- s->si[0].iohandler = NULL;
- s->si[0].fd = cfd;
- s->si[0].flags = SI_FL_NONE | SI_FL_CAP_SPLTCP; /* TCP splicing capable */
- if (s->fe->options2 & PR_O2_INDEPSTR)
- s->si[0].flags |= SI_FL_INDEP_STR;
- s->si[0].exp = TICK_ETERNITY;
+ /* FIXME: the logs are horribly complicated now, because they are
+ * defined in <p>, <p>, and later <be> and <be>.
+ */
- s->si[1].state = s->si[1].prev_state = SI_ST_INI;
- s->si[1].err_type = SI_ET_NONE;
- s->si[1].err_loc = NULL;
- s->si[1].owner = t;
- s->si[1].update = stream_sock_data_finish;
- s->si[1].shutr = stream_sock_shutr;
- s->si[1].shutw = stream_sock_shutw;
- s->si[1].chk_rcv = stream_sock_chk_rcv;
- s->si[1].chk_snd = stream_sock_chk_snd;
- s->si[1].connect = tcpv4_connect_server;
- s->si[1].iohandler = NULL;
- s->si[1].exp = TICK_ETERNITY;
- s->si[1].fd = -1; /* just to help with debugging */
- s->si[1].flags = SI_FL_NONE;
- if (s->be->options2 & PR_O2_INDEPSTR)
- s->si[1].flags |= SI_FL_INDEP_STR;
+ if (s->flags & SN_MONITOR)
+ s->logs.logwait = 0;
+ else
+ s->logs.logwait = p->to_log;
- s->srv = s->prev_srv = s->srv_conn = NULL;
- s->pend_pos = NULL;
- s->conn_retries = s->be->conn_retries;
+ if (s->logs.logwait & LW_REQ)
+ s->do_log = http_sess_log;
+ else
+ s->do_log = tcp_sess_log;
- /* init store persistence */
- s->store_count = 0;
+ /* default error reporting function, may be changed by analysers */
+ s->srv_error = default_srv_error;
- /* FIXME: the logs are horribly complicated now, because they are
- * defined in <p>, <p>, and later <be> and <be>.
- */
-
- if (s->flags & SN_MONITOR)
- s->logs.logwait = 0;
- else
- s->logs.logwait = p->to_log;
+ s->logs.accept_date = date; /* user-visible date for logging */
+ s->logs.tv_accept = now; /* corrected date for internal use */
+ tv_zero(&s->logs.tv_request);
+ s->logs.t_queue = -1;
+ s->logs.t_connect = -1;
+ s->logs.t_data = -1;
+ s->logs.t_close = 0;
+ s->logs.bytes_in = s->logs.bytes_out = 0;
+ s->logs.prx_queue_size = 0; /* we get the number of pending conns before us */
+ s->logs.srv_queue_size = 0; /* we will get this number soon */
- if (s->logs.logwait & LW_REQ)
- s->do_log = http_sess_log;
- else
- s->do_log = tcp_sess_log;
+ s->data_source = DATA_SRC_NONE;
- /* default error reporting function, may be changed by analysers */
- s->srv_error = default_srv_error;
+ s->uniq_id = totalconn;
+ proxy_inc_fe_ctr(l, p); /* note: cum_beconn will be increased once assigned */
- s->logs.accept_date = date; /* user-visible date for logging */
- s->logs.tv_accept = now; /* corrected date for internal use */
- tv_zero(&s->logs.tv_request);
- s->logs.t_queue = -1;
- s->logs.t_connect = -1;
- s->logs.t_data = -1;
- s->logs.t_close = 0;
- s->logs.bytes_in = s->logs.bytes_out = 0;
- s->logs.prx_queue_size = 0; /* we get the number of pending conns before us */
- s->logs.srv_queue_size = 0; /* we will get this number soon */
+ txn = &s->txn;
+ /* Those variables will be checked and freed if non-NULL in
+ * session.c:session_free(). It is important that they are
+ * properly initialized.
+ */
+ txn->sessid = NULL;
+ txn->srv_cookie = NULL;
+ txn->cli_cookie = NULL;
+ txn->uri = NULL;
+ txn->req.cap = NULL;
+ txn->rsp.cap = NULL;
+ txn->hdr_idx.v = NULL;
+ txn->hdr_idx.size = txn->hdr_idx.used = 0;
- s->data_source = DATA_SRC_NONE;
+ if (p->mode == PR_MODE_HTTP) {
+ /* the captures are only used in HTTP frontends */
+ if (p->nb_req_cap > 0 &&
+ (txn->req.cap = pool_alloc2(p->req_cap_pool)) == NULL)
+ goto out_fail_reqcap; /* no memory */
- s->uniq_id = totalconn;
- totalconn++;
- proxy_inc_fe_ctr(l, p); /* note: cum_beconn will be increased once assigned */
+ if (p->nb_rsp_cap > 0 &&
+ (txn->rsp.cap = pool_alloc2(p->rsp_cap_pool)) == NULL)
+ goto out_fail_rspcap; /* no memory */
+ }
- txn = &s->txn;
- /* Those variables will be checked and freed if non-NULL in
- * session.c:session_free(). It is important that they are
- * properly initialized.
+ if (p->acl_requires & ACL_USE_L7_ANY) {
+ /* we have to allocate header indexes only if we know
+ * that we may make use of them. This of course includes
+ * (mode == PR_MODE_HTTP).
*/
- txn->sessid = NULL;
- txn->srv_cookie = NULL;
- txn->cli_cookie = NULL;
- txn->uri = NULL;
- txn->req.cap = NULL;
- txn->rsp.cap = NULL;
- txn->hdr_idx.v = NULL;
- txn->hdr_idx.size = txn->hdr_idx.used = 0;
+ txn->hdr_idx.size = MAX_HTTP_HDR;
- if (p->mode == PR_MODE_HTTP) {
- /* the captures are only used in HTTP frontends */
- if (p->nb_req_cap > 0 &&
- (txn->req.cap = pool_alloc2(p->req_cap_pool)) == NULL)
- goto out_fail_reqcap; /* no memory */
-
- if (p->nb_rsp_cap > 0 &&
- (txn->rsp.cap = pool_alloc2(p->rsp_cap_pool)) == NULL)
- goto out_fail_rspcap; /* no memory */
- }
+ if ((txn->hdr_idx.v = pool_alloc2(p->hdr_idx_pool)) == NULL)
+ goto out_fail_idx; /* no memory */
- if (p->acl_requires & ACL_USE_L7_ANY) {
- /* we have to allocate header indexes only if we know
- * that we may make use of them. This of course includes
- * (mode == PR_MODE_HTTP).
- */
- txn->hdr_idx.size = MAX_HTTP_HDR;
-
- if ((txn->hdr_idx.v = pool_alloc2(p->hdr_idx_pool)) == NULL)
- goto out_fail_idx; /* no memory */
+ /* and now initialize the HTTP transaction state */
+ http_init_txn(s);
+ }
- /* and now initialize the HTTP transaction state */
- http_init_txn(s);
+ if ((p->mode == PR_MODE_TCP || p->mode == PR_MODE_HTTP)
+ && (p->logfac1 >= 0 || p->logfac2 >= 0)) {
+ if (p->to_log) {
+ /* we have the client ip */
+ if (s->logs.logwait & LW_CLIP)
+ if (!(s->logs.logwait &= ~LW_CLIP))
+ s->do_log(s);
}
-
- if ((p->mode == PR_MODE_TCP || p->mode == PR_MODE_HTTP)
- && (p->logfac1 >= 0 || p->logfac2 >= 0)) {
- if (p->to_log) {
- /* we have the client ip */
- if (s->logs.logwait & LW_CLIP)
- if (!(s->logs.logwait &= ~LW_CLIP))
- s->do_log(s);
- }
- else if (s->cli_addr.ss_family == AF_INET) {
- char pn[INET_ADDRSTRLEN], sn[INET_ADDRSTRLEN];
-
- if (!(s->flags & SN_FRT_ADDR_SET))
- get_frt_addr(s);
-
- if (inet_ntop(AF_INET, (const void *)&((struct sockaddr_in *)&s->frt_addr)->sin_addr,
- sn, sizeof(sn)) &&
- inet_ntop(AF_INET, (const void *)&((struct sockaddr_in *)&s->cli_addr)->sin_addr,
- pn, sizeof(pn))) {
- send_log(p, LOG_INFO, "Connect from %s:%d to %s:%d (%s/%s)\n",
- pn, ntohs(((struct sockaddr_in *)&s->cli_addr)->sin_port),
- sn, ntohs(((struct sockaddr_in *)&s->frt_addr)->sin_port),
- p->id, (p->mode == PR_MODE_HTTP) ? "HTTP" : "TCP");
- }
- }
- else {
- char pn[INET6_ADDRSTRLEN], sn[INET6_ADDRSTRLEN];
+ else if (s->cli_addr.ss_family == AF_INET) {
+ char pn[INET_ADDRSTRLEN], sn[INET_ADDRSTRLEN];
- if (!(s->flags & SN_FRT_ADDR_SET))
- get_frt_addr(s);
+ if (!(s->flags & SN_FRT_ADDR_SET))
+ get_frt_addr(s);
- if (inet_ntop(AF_INET6, (const void *)&((struct sockaddr_in6 *)&s->frt_addr)->sin6_addr,
- sn, sizeof(sn)) &&
- inet_ntop(AF_INET6, (const void *)&((struct sockaddr_in6 *)&s->cli_addr)->sin6_addr,
- pn, sizeof(pn))) {
- send_log(p, LOG_INFO, "Connect from %s:%d to %s:%d (%s/%s)\n",
- pn, ntohs(((struct sockaddr_in6 *)&s->cli_addr)->sin6_port),
- sn, ntohs(((struct sockaddr_in6 *)&s->frt_addr)->sin6_port),
- p->id, (p->mode == PR_MODE_HTTP) ? "HTTP" : "TCP");
- }
+ if (inet_ntop(AF_INET, (const void *)&((struct sockaddr_in *)&s->frt_addr)->sin_addr,
+ sn, sizeof(sn)) &&
+ inet_ntop(AF_INET, (const void *)&((struct sockaddr_in *)&s->cli_addr)->sin_addr,
+ pn, sizeof(pn))) {
+ send_log(p, LOG_INFO, "Connect from %s:%d to %s:%d (%s/%s)\n",
+ pn, ntohs(((struct sockaddr_in *)&s->cli_addr)->sin_port),
+ sn, ntohs(((struct sockaddr_in *)&s->frt_addr)->sin_port),
+ p->id, (p->mode == PR_MODE_HTTP) ? "HTTP" : "TCP");
}
}
-
- if ((global.mode & MODE_DEBUG) && (!(global.mode & MODE_QUIET) || (global.mode & MODE_VERBOSE))) {
- int len;
+ else {
+ char pn[INET6_ADDRSTRLEN], sn[INET6_ADDRSTRLEN];
if (!(s->flags & SN_FRT_ADDR_SET))
get_frt_addr(s);
- if (s->cli_addr.ss_family == AF_INET) {
- char pn[INET_ADDRSTRLEN];
- inet_ntop(AF_INET,
- (const void *)&((struct sockaddr_in *)&s->cli_addr)->sin_addr,
- pn, sizeof(pn));
-
- len = sprintf(trash, "%08x:%s.accept(%04x)=%04x from [%s:%d]\n",
- s->uniq_id, p->id, (unsigned short)fd, (unsigned short)cfd,
- pn, ntohs(((struct sockaddr_in *)&s->cli_addr)->sin_port));
+ if (inet_ntop(AF_INET6, (const void *)&((struct sockaddr_in6 *)&s->frt_addr)->sin6_addr,
+ sn, sizeof(sn)) &&
+ inet_ntop(AF_INET6, (const void *)&((struct sockaddr_in6 *)&s->cli_addr)->sin6_addr,
+ pn, sizeof(pn))) {
+ send_log(p, LOG_INFO, "Connect from %s:%d to %s:%d (%s/%s)\n",
+ pn, ntohs(((struct sockaddr_in6 *)&s->cli_addr)->sin6_port),
+ sn, ntohs(((struct sockaddr_in6 *)&s->frt_addr)->sin6_port),
+ p->id, (p->mode == PR_MODE_HTTP) ? "HTTP" : "TCP");
}
- else {
- char pn[INET6_ADDRSTRLEN];
- inet_ntop(AF_INET6,
- (const void *)&((struct sockaddr_in6 *)(&s->cli_addr))->sin6_addr,
- pn, sizeof(pn));
+ }
+ }
- len = sprintf(trash, "%08x:%s.accept(%04x)=%04x from [%s:%d]\n",
- s->uniq_id, p->id, (unsigned short)fd, (unsigned short)cfd,
- pn, ntohs(((struct sockaddr_in6 *)(&s->cli_addr))->sin6_port));
- }
+ if ((global.mode & MODE_DEBUG) && (!(global.mode & MODE_QUIET) || (global.mode & MODE_VERBOSE))) {
+ int len;
- write(1, trash, len);
- }
+ if (!(s->flags & SN_FRT_ADDR_SET))
+ get_frt_addr(s);
- if ((s->req = pool_alloc2(pool2_buffer)) == NULL)
- goto out_fail_req; /* no memory */
+ if (s->cli_addr.ss_family == AF_INET) {
+ char pn[INET_ADDRSTRLEN];
+ inet_ntop(AF_INET,
+ (const void *)&((struct sockaddr_in *)&s->cli_addr)->sin_addr,
+ pn, sizeof(pn));
- s->req->size = global.tune.bufsize;
- buffer_init(s->req);
- s->req->prod = &s->si[0];
- s->req->cons = &s->si[1];
- s->si[0].ib = s->si[1].ob = s->req;
+ len = sprintf(trash, "%08x:%s.accept(%04x)=%04x from [%s:%d]\n",
+ s->uniq_id, p->id, (unsigned short)l->fd, (unsigned short)cfd,
+ pn, ntohs(((struct sockaddr_in *)&s->cli_addr)->sin_port));
+ }
+ else {
+ char pn[INET6_ADDRSTRLEN];
+ inet_ntop(AF_INET6,
+ (const void *)&((struct sockaddr_in6 *)(&s->cli_addr))->sin6_addr,
+ pn, sizeof(pn));
- s->req->flags |= BF_READ_ATTACHED; /* the producer is already connected */
+ len = sprintf(trash, "%08x:%s.accept(%04x)=%04x from [%s:%d]\n",
+ s->uniq_id, p->id, (unsigned short)l->fd, (unsigned short)cfd,
+ pn, ntohs(((struct sockaddr_in6 *)(&s->cli_addr))->sin6_port));
+ }
- if (p->mode == PR_MODE_HTTP)
- s->req->flags |= BF_READ_DONTWAIT; /* one read is usually enough */
+ write(1, trash, len);
+ }
- /* activate default analysers enabled for this listener */
- s->req->analysers = l->analysers;
+ if ((s->req = pool_alloc2(pool2_buffer)) == NULL)
+ goto out_fail_req; /* no memory */
- /* note: this should not happen anymore since there's always at least the switching rules */
- if (!s->req->analysers) {
- buffer_auto_connect(s->req); /* don't wait to establish connection */
- buffer_auto_close(s->req); /* let the producer forward close requests */
- }
+ s->req->size = global.tune.bufsize;
+ buffer_init(s->req);
+ s->req->prod = &s->si[0];
+ s->req->cons = &s->si[1];
+ s->si[0].ib = s->si[1].ob = s->req;
- s->req->rto = s->fe->timeout.client;
- s->req->wto = s->be->timeout.server;
- s->req->cto = s->be->timeout.connect;
+ s->req->flags |= BF_READ_ATTACHED; /* the producer is already connected */
- if ((s->rep = pool_alloc2(pool2_buffer)) == NULL)
- goto out_fail_rep; /* no memory */
+ if (p->mode == PR_MODE_HTTP)
+ s->req->flags |= BF_READ_DONTWAIT; /* one read is usually enough */
- s->rep->size = global.tune.bufsize;
- buffer_init(s->rep);
- s->rep->prod = &s->si[1];
- s->rep->cons = &s->si[0];
- s->si[0].ob = s->si[1].ib = s->rep;
- s->rep->analysers = 0;
+ /* activate default analysers enabled for this listener */
+ s->req->analysers = l->analysers;
- s->rep->rto = s->be->timeout.server;
- s->rep->wto = s->fe->timeout.client;
- s->rep->cto = TICK_ETERNITY;
+ /* note: this should not happen anymore since there's always at least the switching rules */
+ if (!s->req->analysers) {
+ buffer_auto_connect(s->req); /* don't wait to establish connection */
+ buffer_auto_close(s->req); /* let the producer forward close requests */
+ }
- s->req->rex = TICK_ETERNITY;
- s->req->wex = TICK_ETERNITY;
- s->req->analyse_exp = TICK_ETERNITY;
- s->rep->rex = TICK_ETERNITY;
- s->rep->wex = TICK_ETERNITY;
- s->rep->analyse_exp = TICK_ETERNITY;
- t->expire = TICK_ETERNITY;
+ s->req->rto = s->fe->timeout.client;
+ s->req->wto = s->be->timeout.server;
+ s->req->cto = s->be->timeout.connect;
- fd_insert(cfd);
- fdtab[cfd].owner = &s->si[0];
- fdtab[cfd].state = FD_STREADY;
- fdtab[cfd].flags = FD_FL_TCP | FD_FL_TCP_NODELAY;
- if (p->options & PR_O_TCP_NOLING)
- fdtab[cfd].flags |= FD_FL_TCP_NOLING;
+ if ((s->rep = pool_alloc2(pool2_buffer)) == NULL)
+ goto out_fail_rep; /* no memory */
- fdtab[cfd].cb[DIR_RD].f = l->proto->read;
- fdtab[cfd].cb[DIR_RD].b = s->req;
- fdtab[cfd].cb[DIR_WR].f = l->proto->write;
- fdtab[cfd].cb[DIR_WR].b = s->rep;
- fdinfo[cfd].peeraddr = (struct sockaddr *)&s->cli_addr;
- fdinfo[cfd].peerlen = sizeof(s->cli_addr);
+ s->rep->size = global.tune.bufsize;
+ buffer_init(s->rep);
+ s->rep->prod = &s->si[1];
+ s->rep->cons = &s->si[0];
+ s->si[0].ob = s->si[1].ib = s->rep;
+ s->rep->analysers = 0;
- if ((p->mode == PR_MODE_HTTP && (s->flags & SN_MONITOR)) ||
- (p->mode == PR_MODE_HEALTH && (p->options & PR_O_HTTP_CHK))) {
- /* Either we got a request from a monitoring system on an HTTP instance,
- * or we're in health check mode with the 'httpchk' option enabled. In
- * both cases, we return a fake "HTTP/1.0 200 OK" response and we exit.
- */
- struct chunk msg;
- chunk_initstr(&msg, "HTTP/1.0 200 OK\r\n\r\n");
- stream_int_retnclose(&s->si[0], &msg); /* forge a 200 response */
- s->req->analysers = 0;
- t->expire = s->rep->wex;
- }
- else if (p->mode == PR_MODE_HEALTH) { /* health check mode, no client reading */
- struct chunk msg;
- chunk_initstr(&msg, "OK\n");
- stream_int_retnclose(&s->si[0], &msg); /* forge an "OK" response */
- s->req->analysers = 0;
- t->expire = s->rep->wex;
- }
- else {
- EV_FD_SET(cfd, DIR_RD);
- }
+ s->rep->rto = s->be->timeout.server;
+ s->rep->wto = s->fe->timeout.client;
+ s->rep->cto = TICK_ETERNITY;
- /* it is important not to call the wakeup function directly but to
- * pass through task_wakeup(), because this one knows how to apply
- * priorities to tasks.
- */
- task_wakeup(t, TASK_WOKEN_INIT);
+ s->req->rex = TICK_ETERNITY;
+ s->req->wex = TICK_ETERNITY;
+ s->req->analyse_exp = TICK_ETERNITY;
+ s->rep->rex = TICK_ETERNITY;
+ s->rep->wex = TICK_ETERNITY;
+ s->rep->analyse_exp = TICK_ETERNITY;
+ t->expire = TICK_ETERNITY;
- l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
- if (l->nbconn >= l->maxconn) {
- EV_FD_CLR(l->fd, DIR_RD);
- l->state = LI_FULL;
- }
+ fd_insert(cfd);
+ fdtab[cfd].owner = &s->si[0];
+ fdtab[cfd].state = FD_STREADY;
+ fdtab[cfd].flags = FD_FL_TCP | FD_FL_TCP_NODELAY;
+ if (p->options & PR_O_TCP_NOLING)
+ fdtab[cfd].flags |= FD_FL_TCP_NOLING;
- p->feconn++; /* beconn will be increased later */
- if (p->feconn > p->counters.feconn_max)
- p->counters.feconn_max = p->feconn;
+ fdtab[cfd].cb[DIR_RD].f = l->proto->read;
+ fdtab[cfd].cb[DIR_RD].b = s->req;
+ fdtab[cfd].cb[DIR_WR].f = l->proto->write;
+ fdtab[cfd].cb[DIR_WR].b = s->rep;
+ fdinfo[cfd].peeraddr = (struct sockaddr *)&s->cli_addr;
+ fdinfo[cfd].peerlen = sizeof(s->cli_addr);
- if (l->counters) {
- if (l->nbconn > l->counters->conn_max)
- l->counters->conn_max = l->nbconn;
- }
+ if ((p->mode == PR_MODE_HTTP && (s->flags & SN_MONITOR)) ||
+ (p->mode == PR_MODE_HEALTH && (p->options & PR_O_HTTP_CHK))) {
+ /* Either we got a request from a monitoring system on an HTTP instance,
+ * or we're in health check mode with the 'httpchk' option enabled. In
+ * both cases, we return a fake "HTTP/1.0 200 OK" response and we exit.
+ */
+ struct chunk msg;
+ chunk_initstr(&msg, "HTTP/1.0 200 OK\r\n\r\n");
+ stream_int_retnclose(&s->si[0], &msg); /* forge a 200 response */
+ s->req->analysers = 0;
+ t->expire = s->rep->wex;
+ }
+ else if (p->mode == PR_MODE_HEALTH) { /* health check mode, no client reading */
+ struct chunk msg;
+ chunk_initstr(&msg, "OK\n");
+ stream_int_retnclose(&s->si[0], &msg); /* forge an "OK" response */
+ s->req->analysers = 0;
+ t->expire = s->rep->wex;
+ }
+ else {
+ EV_FD_SET(cfd, DIR_RD);
+ }
- actconn++;
+ /* it is important not to call the wakeup function directly but to
+ * pass through task_wakeup(), because this one knows how to apply
+ * priorities to tasks.
+ */
+ task_wakeup(t, TASK_WOKEN_INIT);
- // fprintf(stderr, "accepting from %p => %d conn, %d total, task=%p\n", p, actconn, totalconn, t);
- } /* end of while (p->feconn < p->maxconn) */
- return 0;
+ return 1;
/* Error unrolling */
out_fail_rep:
@@ -507,8 +423,7 @@
LIST_DEL(&s->list);
pool_free2(pool2_session, s);
out_close:
- close(cfd);
- return 0;
+ return -1;
}
/* set test->i to the id of the frontend */
diff --git a/src/proto_tcp.c b/src/proto_tcp.c
index 1662e90..fc020c9 100644
--- a/src/proto_tcp.c
+++ b/src/proto_tcp.c
@@ -63,6 +63,7 @@
.sock_family = AF_INET,
.sock_addrlen = sizeof(struct sockaddr_in),
.l3_addrlen = 32/8,
+ .accept = &stream_sock_accept,
.read = &stream_sock_read,
.write = &stream_sock_write,
.bind_all = tcp_bind_listeners,
@@ -81,6 +82,7 @@
.sock_family = AF_INET6,
.sock_addrlen = sizeof(struct sockaddr_in6),
.l3_addrlen = 128/8,
+ .accept = &stream_sock_accept,
.read = &stream_sock_read,
.write = &stream_sock_write,
.bind_all = tcp_bind_listeners,
@@ -539,19 +541,17 @@
listener->fd = fd;
listener->state = LI_LISTEN;
- /* the function for the accept() event */
- fd_insert(fd);
- fdtab[fd].cb[DIR_RD].f = listener->accept;
- fdtab[fd].cb[DIR_WR].f = NULL; /* never called */
- fdtab[fd].cb[DIR_RD].b = fdtab[fd].cb[DIR_WR].b = NULL;
fdtab[fd].owner = listener; /* reference the listener instead of a task */
fdtab[fd].state = FD_STLISTEN;
- fdtab[fd].flags = FD_FL_TCP;
- if (listener->options & LI_O_NOLINGER)
- fdtab[fd].flags |= FD_FL_TCP_NOLING;
+ fdtab[fd].flags = FD_FL_TCP | ((listener->options & LI_O_NOLINGER) ? FD_FL_TCP_NOLING : 0);
+ fdtab[fd].cb[DIR_RD].f = listener->proto->accept;
+ fdtab[fd].cb[DIR_WR].f = NULL; /* never called */
+ fdtab[fd].cb[DIR_RD].b = fdtab[fd].cb[DIR_WR].b = NULL;
fdinfo[fd].peeraddr = NULL;
fdinfo[fd].peerlen = 0;
+ fd_insert(fd);
+
tcp_return:
if (msg && errlen)
strlcpy2(errmsg, msg, errlen);
diff --git a/src/proto_uxst.c b/src/proto_uxst.c
index 177c5a3..98c9555 100644
--- a/src/proto_uxst.c
+++ b/src/proto_uxst.c
@@ -1,7 +1,7 @@
/*
* UNIX SOCK_STREAM protocol layer (uxst)
*
- * Copyright 2000-2009 Willy Tarreau <w@1wt.eu>
+ * Copyright 2000-2010 Willy Tarreau <w@1wt.eu>
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
@@ -29,26 +29,17 @@
#include <common/config.h>
#include <common/debug.h>
#include <common/errors.h>
-#include <common/memory.h>
#include <common/mini-clist.h>
#include <common/standard.h>
-#include <common/ticks.h>
#include <common/time.h>
#include <common/version.h>
#include <types/global.h>
-#include <proto/acl.h>
-#include <proto/backend.h>
-#include <proto/buffers.h>
-#include <proto/dumpstats.h>
#include <proto/fd.h>
#include <proto/log.h>
#include <proto/protocols.h>
#include <proto/proto_uxst.h>
-#include <proto/queue.h>
-#include <proto/session.h>
-#include <proto/stream_interface.h>
#include <proto/stream_sock.h>
#include <proto/task.h>
@@ -68,6 +59,7 @@
.sock_family = AF_UNIX,
.sock_addrlen = sizeof(struct sockaddr_un),
.l3_addrlen = sizeof(((struct sockaddr_un*)0)->sun_path),/* path len */
+ .accept = &stream_sock_accept,
.read = &stream_sock_read,
.write = &stream_sock_write,
.bind_all = uxst_bind_listeners,
@@ -262,7 +254,7 @@
/* the function for the accept() event */
fd_insert(fd);
- fdtab[fd].cb[DIR_RD].f = listener->accept;
+ fdtab[fd].cb[DIR_RD].f = listener->proto->accept;
fdtab[fd].cb[DIR_WR].f = NULL; /* never called */
fdtab[fd].cb[DIR_RD].b = fdtab[fd].cb[DIR_WR].b = NULL;
fdtab[fd].owner = listener; /* reference the listener instead of a task */
@@ -347,229 +339,6 @@
* 4) high-level functions
********************************/
-
-/*
- * This function is called on a read event from a listen socket, corresponding
- * to an accept. It tries to accept as many connections as possible.
- * It returns 0. Since we use UNIX sockets on the local system for monitoring
- * purposes and other related things, we do not need to output as many messages
- * as with TCP which can fall under attack.
- */
-int uxst_event_accept(int fd) {
- struct listener *l = fdtab[fd].owner;
- struct session *s;
- struct task *t;
- int cfd;
- int max_accept;
-
- if (global.nbproc > 1)
- max_accept = 8; /* let other processes catch some connections too */
- else
- max_accept = -1;
-
- while (max_accept--) {
- struct sockaddr_storage addr;
- socklen_t laddr = sizeof(addr);
-
- if ((cfd = accept(fd, (struct sockaddr *)&addr, &laddr)) == -1) {
- switch (errno) {
- case EAGAIN:
- case EINTR:
- case ECONNABORTED:
- return 0; /* nothing more to accept */
- case ENFILE:
- /* Process reached system FD limit. Check system tunables. */
- return 0;
- case EMFILE:
- /* Process reached process FD limit. Check 'ulimit-n'. */
- return 0;
- case ENOBUFS:
- case ENOMEM:
- /* Process reached system memory limit. Check system tunables. */
- return 0;
- default:
- return 0;
- }
- }
-
- if (l->nbconn >= l->maxconn || actconn >= global.maxconn) {
- /* too many connections, we shoot this one and return.
- * FIXME: it would be better to simply switch the listener's
- * state to LI_FULL and disable the FD. We could re-enable
- * it upon fd_delete(), but this requires all protocols to
- * be switched.
- */
- goto out_close;
- }
-
- if ((s = pool_alloc2(pool2_session)) == NULL) {
- Alert("out of memory in uxst_event_accept().\n");
- goto out_close;
- }
-
- LIST_ADDQ(&sessions, &s->list);
- LIST_INIT(&s->back_refs);
-
- s->flags = 0;
- s->term_trace = 0;
-
- if ((t = task_new()) == NULL) {
- Alert("out of memory in uxst_event_accept().\n");
- goto out_free_session;
- }
-
- s->cli_addr = addr;
-
- /* FIXME: should be checked earlier */
- if (cfd >= global.maxsock) {
- Alert("accept(): not enough free sockets. Raise -n argument. Giving up.\n");
- goto out_free_task;
- }
-
- if (fcntl(cfd, F_SETFL, O_NONBLOCK) == -1) {
- Alert("accept(): cannot set the socket in non blocking mode. Giving up.\n");
- goto out_free_task;
- }
-
- t->process = l->handler;
- t->context = s;
- t->nice = l->nice;
-
- s->task = t;
- s->listener = l;
- s->fe = s->be = l->private;
-
- s->req = s->rep = NULL; /* will be allocated later */
-
- s->si[0].state = s->si[0].prev_state = SI_ST_EST;
- s->si[0].err_type = SI_ET_NONE;
- s->si[0].err_loc = NULL;
- s->si[0].owner = t;
- s->si[0].update = stream_sock_data_finish;
- s->si[0].shutr = stream_sock_shutr;
- s->si[0].shutw = stream_sock_shutw;
- s->si[0].chk_rcv = stream_sock_chk_rcv;
- s->si[0].chk_snd = stream_sock_chk_snd;
- s->si[0].connect = NULL;
- s->si[0].iohandler = NULL;
- s->si[0].fd = cfd;
- s->si[0].flags = SI_FL_NONE;
- if (s->fe->options2 & PR_O2_INDEPSTR)
- s->si[0].flags |= SI_FL_INDEP_STR;
- s->si[0].exp = TICK_ETERNITY;
-
- s->si[1].state = s->si[1].prev_state = SI_ST_INI;
- s->si[1].err_type = SI_ET_NONE;
- s->si[1].err_loc = NULL;
- s->si[1].owner = t;
- s->si[1].exp = TICK_ETERNITY;
- s->si[1].fd = -1; /* just to help with debugging */
- s->si[1].flags = SI_FL_NONE;
- if (s->be->options2 & PR_O2_INDEPSTR)
- s->si[1].flags |= SI_FL_INDEP_STR;
-
- stream_int_register_handler(&s->si[1], stats_io_handler);
- s->si[1].private = s;
- s->si[1].st1 = 0;
- s->si[1].st0 = STAT_CLI_INIT;
-
- s->srv = s->prev_srv = s->srv_conn = NULL;
- s->pend_pos = NULL;
-
- s->store_count = 0;
-
- memset(&s->logs, 0, sizeof(s->logs));
- memset(&s->txn, 0, sizeof(s->txn));
-
- s->logs.accept_date = date; /* user-visible date for logging */
- s->logs.tv_accept = now; /* corrected date for internal use */
-
- s->data_state = DATA_ST_INIT;
- s->data_source = DATA_SRC_NONE;
- s->uniq_id = totalconn;
- totalconn++;
-
- if ((s->req = pool_alloc2(pool2_buffer)) == NULL)
- goto out_free_task;
-
- s->req->size = global.tune.bufsize;
- buffer_init(s->req);
- s->req->prod = &s->si[0];
- s->req->cons = &s->si[1];
- s->si[0].ib = s->si[1].ob = s->req;
- s->req->flags |= BF_READ_ATTACHED; /* the producer is already connected */
- s->req->flags |= BF_READ_DONTWAIT; /* we plan to read small requests */
-
- s->req->analysers = l->analysers;
-
- s->req->wto = TICK_ETERNITY;
- s->req->cto = TICK_ETERNITY;
- s->req->rto = TICK_ETERNITY;
-
- if ((s->rep = pool_alloc2(pool2_buffer)) == NULL)
- goto out_free_req;
-
- s->rep->size = global.tune.bufsize;
- buffer_init(s->rep);
-
- s->rep->prod = &s->si[1];
- s->rep->cons = &s->si[0];
- s->si[0].ob = s->si[1].ib = s->rep;
- s->rep->analysers = 0;
-
- s->rep->rto = TICK_ETERNITY;
- s->rep->cto = TICK_ETERNITY;
- s->rep->wto = TICK_ETERNITY;
-
- s->req->rex = TICK_ETERNITY;
- s->req->wex = TICK_ETERNITY;
- s->req->analyse_exp = TICK_ETERNITY;
- s->rep->rex = TICK_ETERNITY;
- s->rep->wex = TICK_ETERNITY;
- s->rep->analyse_exp = TICK_ETERNITY;
-
- t->expire = TICK_ETERNITY;
-
- if (l->timeout) {
- s->req->rto = *l->timeout;
- s->rep->wto = *l->timeout;
- }
-
- fd_insert(cfd);
- fdtab[cfd].owner = &s->si[0];
- fdtab[cfd].state = FD_STREADY;
- fdtab[cfd].cb[DIR_RD].f = l->proto->read;
- fdtab[cfd].cb[DIR_RD].b = s->req;
- fdtab[cfd].cb[DIR_WR].f = l->proto->write;
- fdtab[cfd].cb[DIR_WR].b = s->rep;
- fdinfo[cfd].peeraddr = (struct sockaddr *)&s->cli_addr;
- fdinfo[cfd].peerlen = sizeof(s->cli_addr);
-
- EV_FD_SET(cfd, DIR_RD);
-
- task_wakeup(t, TASK_WOKEN_INIT);
-
- l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
- if (l->nbconn >= l->maxconn) {
- EV_FD_CLR(l->fd, DIR_RD);
- l->state = LI_FULL;
- }
- actconn++;
- }
- return 0;
-
- out_free_req:
- pool_free2(pool2_buffer, s->req);
- out_free_task:
- task_free(t);
- out_free_session:
- LIST_DEL(&s->list);
- pool_free2(pool2_session, s);
- out_close:
- close(cfd);
- return 0;
-}
-
__attribute__((constructor))
static void __uxst_protocol_init(void)
{
diff --git a/src/stream_sock.c b/src/stream_sock.c
index d588762..171ed29 100644
--- a/src/stream_sock.c
+++ b/src/stream_sock.c
@@ -31,6 +31,8 @@
#include <proto/buffers.h>
#include <proto/fd.h>
+#include <proto/freq_ctr.h>
+#include <proto/log.h>
#include <proto/pipe.h>
#include <proto/stream_sock.h>
#include <proto/task.h>
@@ -1118,6 +1120,107 @@
}
}
+/* This function is called on a read event from a listening socket, corresponding
+ * to an accept. It tries to accept as many connections as possible, and for each
+ * calls the listener's accept handler (generally the frontend's accept handler).
+ */
+int stream_sock_accept(int fd)
+{
+ struct listener *l = fdtab[fd].owner;
+ struct proxy *p = l->frontend;
+ int max_accept = global.tune.maxaccept;
+ int cfd;
+ int ret;
+
+ if (unlikely(l->nbconn >= l->maxconn)) {
+ EV_FD_CLR(l->fd, DIR_RD);
+ l->state = LI_FULL;
+ return 0;
+ }
+
+ if (p && p->fe_sps_lim) {
+ int max = freq_ctr_remain(&p->fe_sess_per_sec, p->fe_sps_lim, 0);
+ if (max_accept > max)
+ max_accept = max;
+ }
+
+ while ((!p || p->feconn < p->maxconn) && actconn < global.maxconn && max_accept--) {
+ struct sockaddr_storage addr;
+ socklen_t laddr = sizeof(addr);
+
+ cfd = accept(fd, (struct sockaddr *)&addr, &laddr);
+ if (unlikely(cfd == -1)) {
+ switch (errno) {
+ case EAGAIN:
+ case EINTR:
+ case ECONNABORTED:
+ return 0; /* nothing more to accept */
+ case ENFILE:
+ send_log(p, LOG_EMERG,
+ "Proxy %s reached system FD limit at %d. Please check system tunables.\n",
+ p->id, maxfd);
+ return 0;
+ case EMFILE:
+ send_log(p, LOG_EMERG,
+ "Proxy %s reached process FD limit at %d. Please check 'ulimit-n' and restart.\n",
+ p->id, maxfd);
+ return 0;
+ case ENOBUFS:
+ case ENOMEM:
+ send_log(p, LOG_EMERG,
+ "Proxy %s reached system memory limit at %d sockets. Please check system tunables.\n",
+ p->id, maxfd);
+ return 0;
+ default:
+ return 0;
+ }
+ }
+
+ if (unlikely(cfd >= global.maxsock)) {
+ Alert("accept(): not enough free sockets. Raise -n argument. Giving up.\n");
+ goto out_close;
+ }
+
+ ret = l->accept(l, cfd, &addr);
+ if (unlikely(ret < 0)) {
+ /* critical error encountered, generally a resource shortage */
+ EV_FD_CLR(fd, DIR_RD);
+ p->state = PR_STIDLE;
+ goto out_close;
+ }
+ else if (unlikely(ret == 0)) {
+ /* ignore this connection */
+ close(cfd);
+ continue;
+ }
+
+ actconn++;
+ totalconn++;
+ l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
+ if (l->nbconn >= l->maxconn) {
+ EV_FD_CLR(l->fd, DIR_RD);
+ l->state = LI_FULL;
+ }
+
+ if (p) {
+ p->feconn++; /* beconn will be increased later */
+ if (p->feconn > p->counters.feconn_max)
+ p->counters.feconn_max = p->feconn;
+ }
+
+ if (l->counters) {
+ if (l->nbconn > l->counters->conn_max)
+ l->counters->conn_max = l->nbconn;
+ }
+ } /* end of while (p->feconn < p->maxconn) */
+ return 0;
+
+ /* Error unrolling */
+ out_close:
+ close(cfd);
+ return 0;
+}
+
/*
* Local variables: