mirrors/haproxy
1
0
Fork 0
mirror of https://git.haproxy.org/git/haproxy.git/ synced 2025-12-08 03:01:14 +01:00
Code Issues Projects Releases Wiki Activity

[MAJOR] frontend: split accept() into frontend_accept() and session_accept()

Browse Source 
A new function session_accept() is now called from the lower layer to
instanciate a new session. Once the session is instanciated, the upper
layer's frontent_accept() is called. This one can be service-dependant.

That way, we have a 3-phase accept() sequence :
  1) protocol-specific, session-less accept(), which is pointed to by
     the listener. It defaults to the generic stream_sock_accept().
  2) session_accept() which relies on a frontend but not necessarily
     for use in a proxy (eg: stats or any future service).
  3) frontend_accept() which performs the accept for the service
     offerred by the frontend. It defaults to frontend_accept() which
     is really what is used by a proxy.

The TCP/HTTP proxies have been moved to this mode so that we can now rely on
frontend_accept() for any type of session initialization relying on a frontend.

The next step will be to convert the stats to use the same system for the stats.
This commit is contained in:
Willy Tarreau 2010-06-01 17:45:26 +02:00
parent f229eb8f8f
commit 81f9aa3bf2
5 changed files with 293 additions and 261 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/proto/frontend.h
View File

@ -26,7 +26,7 @@
#include <types/session.h> #include <types/session.h>
void get_frt_addr(struct session *s); void get_frt_addr(struct session *s);
int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr); int frontend_accept(struct session *s);
#endif /* _PROTO_FRONTEND_H */ #endif /* _PROTO_FRONTEND_H */

39
include/proto/session.h
View File

@ -1,23 +1,23 @@
/* /*
include/proto/session.h * include/proto/session.h
This file defines everything related to sessions. * This file defines everything related to sessions.
*
Copyright (C) 2000-2008 Willy Tarreau - w@1wt.eu * Copyright (C) 2000-2010 Willy Tarreau - w@1wt.eu
*
This library is free software; you can redistribute it and/or * This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public * modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation, version 2.1 * License as published by the Free Software Foundation, version 2.1
exclusively. * exclusively.
*
This library is distributed in the hope that it will be useful, * This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of * but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details. * Lesser General Public License for more details.
*
You should have received a copy of the GNU Lesser General Public * You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software * License along with this library; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/ */
#ifndef _PROTO_SESSION_H #ifndef _PROTO_SESSION_H
#define _PROTO_SESSION_H #define _PROTO_SESSION_H
@ -29,6 +29,7 @@
extern struct pool_head *pool2_session; extern struct pool_head *pool2_session;
extern struct list sessions; extern struct list sessions;
int session_accept(struct listener *l, int cfd, struct sockaddr_storage *addr);
void session_free(struct session *s); void session_free(struct session *s);
/* perform minimal intializations, report 0 in case of error, 1 if OK. */ /* perform minimal intializations, report 0 in case of error, 1 if OK. */

4
src/cfgparse.c
View File

@ -5280,6 +5280,8 @@ out_uri_auth_compat:
} }
if (curproxy->cap & PR_CAP_FE) { if (curproxy->cap & PR_CAP_FE) {
curproxy->accept = frontend_accept;
if (curproxy->tcp_req.inspect_delay || if (curproxy->tcp_req.inspect_delay ||
!LIST_ISEMPTY(&curproxy->tcp_req.inspect_rules)) !LIST_ISEMPTY(&curproxy->tcp_req.inspect_rules))
curproxy->fe_req_ana |= AN_REQ_INSPECT; curproxy->fe_req_ana |= AN_REQ_INSPECT;
@ -5351,7 +5353,7 @@ out_uri_auth_compat:
listener->maxconn = curproxy->maxconn; listener->maxconn = curproxy->maxconn;
listener->backlog = curproxy->backlog; listener->backlog = curproxy->backlog;
listener->timeout = &curproxy->timeout.client; listener->timeout = &curproxy->timeout.client;
listener->accept = frontend_accept; listener->accept = session_accept;
listener->frontend = curproxy; listener->frontend = curproxy;
listener->handler = process_session; listener->handler = process_session;
listener->analysers |= curproxy->fe_req_ana; listener->analysers |= curproxy->fe_req_ana;

272
src/frontend.c
View File

@ -53,200 +53,13 @@ void get_frt_addr(struct session *s)
s->flags |= SN_FRT_ADDR_SET; s->flags |= SN_FRT_ADDR_SET;
} }
/* This function is called from the protocol layer accept() in order to instanciate /* Finish a session accept() for a proxy (TCP or HTTP). It returns a negative
* a new proxy. It returns a positive value upon success, 0 if the connection needs * value in case of failure, a positive value in case of success, or zero if
* to be closed and ignored, or a negative value upon critical failure. * it is a success but the session must be closed ASAP.
*/ */
int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr) int frontend_accept(struct session *s)
{ {
struct proxy *p = l->frontend; int cfd = s->si[0].fd;
struct session *s;
struct http_txn *txn;
struct task *t;
if (unlikely((s = pool_alloc2(pool2_session)) == NULL)) {
Alert("out of memory in event_accept().\n");
goto out_close;
}
/* minimum session initialization required for monitor mode below */
s->flags = 0;
s->logs.logwait = p->to_log;
/* 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, but
* cleanly.
*/
if (unlikely((l->options & LI_O_CHK_MONNET) &&
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;
s->logs.logwait = 0;
}
/* OK, we're keeping the session, so let's properly initialize the session */
LIST_ADDQ(&sessions, &s->list);
LIST_INIT(&s->back_refs);
if (unlikely((t = task_new()) == NULL)) { /* disable this proxy for a while */
Alert("out of memory in event_accept().\n");
goto out_free_session;
}
s->term_trace = 0;
s->cli_addr = *addr;
s->logs.accept_date = date; /* user-visible date for logging */
s->logs.tv_accept = now; /* corrected date for internal use */
s->uniq_id = totalconn;
proxy_inc_fe_ctr(l, p); /* note: cum_beconn will be increased once assigned */
t->process = l->handler;
t->context = s;
t->nice = l->nice;
t->expire = TICK_ETERNITY;
s->task = t;
s->listener = l;
/* 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 */
/* now evaluate the tcp-request layer4 rules. Since we expect to be able
* to abort right here as soon as possible, we check the rules before
* even initializing the stream interfaces.
*/
if ((l->options & LI_O_TCP_RULES) && !tcp_exec_req_rules(s)) {
task_free(t);
LIST_DEL(&s->list);
pool_free2(pool2_session, s);
/* let's do a no-linger now to close with a single RST. */
setsockopt(cfd, SOL_SOCKET, SO_LINGER, (struct linger *) &nolinger, sizeof(struct linger));
return 0;
}
/* this part should be common with other protocols */
s->si[0].fd = cfd;
s->si[0].owner = t;
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].connect = NULL;
s->si[0].iohandler = NULL;
s->si[0].exp = TICK_ETERNITY;
s->si[0].flags = SI_FL_NONE;
if (likely(s->fe->options2 & PR_O2_INDEPSTR))
s->si[0].flags |= SI_FL_INDEP_STR;
if (addr->ss_family == AF_INET || addr->ss_family == AF_INET6)
s->si[0].flags = SI_FL_CAP_SPLTCP; /* TCP/TCPv6 splicing possible */
/* add the various callbacks */
stream_sock_prepare_interface(&s->si[0]);
/* pre-initialize the other side's stream interface to an INIT state. The
* callbacks will be initialized before attempting to connect.
*/
s->si[1].fd = -1; /* just to help with debugging */
s->si[1].owner = t;
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].connect = NULL;
s->si[1].iohandler = NULL;
s->si[1].shutr = stream_int_shutr;
s->si[1].shutw = stream_int_shutw;
s->si[1].exp = TICK_ETERNITY;
s->si[1].flags = SI_FL_NONE;
if (likely(s->fe->options2 & PR_O2_INDEPSTR))
s->si[1].flags |= SI_FL_INDEP_STR;
s->srv = s->prev_srv = s->srv_conn = NULL;
s->pend_pos = NULL;
/* init store persistence */
s->store_count = 0;
/* Adjust some socket options */
if (unlikely(fcntl(cfd, F_SETFL, O_NONBLOCK) == -1)) {
Alert("accept(): cannot set the socket in non blocking mode. Giving up\n");
goto out_free_task;
}
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;
if (unlikely((s->req = pool_alloc2(pool2_buffer)) == NULL))
goto out_free_task; /* no memory */
if (unlikely((s->rep = pool_alloc2(pool2_buffer)) == NULL))
goto out_free_req; /* no memory */
/* initialize the request buffer */
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 */
/* activate default analysers enabled for this listener */
s->req->analysers = l->analysers;
s->req->wto = TICK_ETERNITY;
s->req->rto = TICK_ETERNITY;
s->req->rex = TICK_ETERNITY;
s->req->wex = TICK_ETERNITY;
s->req->analyse_exp = TICK_ETERNITY;
/* initialize response buffer */
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->wto = TICK_ETERNITY;
s->rep->rex = TICK_ETERNITY;
s->rep->wex = TICK_ETERNITY;
s->rep->analyse_exp = TICK_ETERNITY;
/* finish initialization of the accepted file descriptor */
fd_insert(cfd);
fdtab[cfd].owner = &s->si[0];
fdtab[cfd].state = FD_STREADY;
fdtab[cfd].flags = 0;
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);
/***************** to be moved to the TCP/HTTP frontend's accept() **************/
tv_zero(&s->logs.tv_request); tv_zero(&s->logs.tv_request);
s->logs.t_queue = -1; s->logs.t_queue = -1;
@ -277,10 +90,10 @@ int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
goto out_delete_cfd; goto out_delete_cfd;
} }
if (p->options & PR_O_TCP_CLI_KA) if (s->fe->options & PR_O_TCP_CLI_KA)
setsockopt(cfd, SOL_SOCKET, SO_KEEPALIVE, (char *) &one, sizeof(one)); setsockopt(cfd, SOL_SOCKET, SO_KEEPALIVE, (char *) &one, sizeof(one));
if (p->options & PR_O_TCP_NOLING) if (s->fe->options & PR_O_TCP_NOLING)
setsockopt(cfd, SOL_SOCKET, SO_LINGER, (struct linger *) &nolinger, sizeof(struct linger)); setsockopt(cfd, SOL_SOCKET, SO_LINGER, (struct linger *) &nolinger, sizeof(struct linger));
if (global.tune.client_sndbuf) if (global.tune.client_sndbuf)
@ -289,34 +102,34 @@ int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
if (global.tune.client_rcvbuf) if (global.tune.client_rcvbuf)
setsockopt(cfd, SOL_SOCKET, SO_RCVBUF, &global.tune.client_rcvbuf, sizeof(global.tune.client_rcvbuf)); setsockopt(cfd, SOL_SOCKET, SO_RCVBUF, &global.tune.client_rcvbuf, sizeof(global.tune.client_rcvbuf));
if (p->mode == PR_MODE_HTTP) { if (s->fe->mode == PR_MODE_HTTP) {
/* the captures are only used in HTTP frontends */ /* the captures are only used in HTTP frontends */
if (unlikely(p->nb_req_cap > 0 && if (unlikely(s->fe->nb_req_cap > 0 &&
(txn->req.cap = pool_alloc2(p->req_cap_pool)) == NULL)) (s->txn.req.cap = pool_alloc2(s->fe->req_cap_pool)) == NULL))
goto out_delete_cfd; /* no memory */ goto out_delete_cfd; /* no memory */
if (unlikely(p->nb_rsp_cap > 0 && if (unlikely(s->fe->nb_rsp_cap > 0 &&
(txn->rsp.cap = pool_alloc2(p->rsp_cap_pool)) == NULL)) (s->txn.rsp.cap = pool_alloc2(s->fe->rsp_cap_pool)) == NULL))
goto out_free_reqcap; /* no memory */ goto out_free_reqcap; /* no memory */
} }
if (p->acl_requires & ACL_USE_L7_ANY) { if (s->fe->acl_requires & ACL_USE_L7_ANY) {
/* we have to allocate header indexes only if we know /* we have to allocate header indexes only if we know
* that we may make use of them. This of course includes * that we may make use of them. This of course includes
* (mode == PR_MODE_HTTP). * (mode == PR_MODE_HTTP).
*/ */
txn->hdr_idx.size = MAX_HTTP_HDR; s->txn.hdr_idx.size = MAX_HTTP_HDR;
if (unlikely((txn->hdr_idx.v = pool_alloc2(p->hdr_idx_pool)) == NULL)) if (unlikely((s->txn.hdr_idx.v = pool_alloc2(s->fe->hdr_idx_pool)) == NULL))
goto out_free_rspcap; /* no memory */ goto out_free_rspcap; /* no memory */
/* and now initialize the HTTP transaction state */ /* and now initialize the HTTP transaction state */
http_init_txn(s); http_init_txn(s);
} }
if ((p->mode == PR_MODE_TCP || p->mode == PR_MODE_HTTP) if ((s->fe->mode == PR_MODE_TCP || s->fe->mode == PR_MODE_HTTP)
&& (p->logfac1 >= 0 || p->logfac2 >= 0)) { && (s->fe->logfac1 >= 0 || s->fe->logfac2 >= 0)) {
if (likely(p->to_log)) { if (likely(s->fe->to_log)) {
/* we have the client ip */ /* we have the client ip */
if (s->logs.logwait & LW_CLIP) if (s->logs.logwait & LW_CLIP)
if (!(s->logs.logwait &= ~LW_CLIP)) if (!(s->logs.logwait &= ~LW_CLIP))
@ -332,10 +145,10 @@ int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
sn, sizeof(sn)) && sn, sizeof(sn)) &&
inet_ntop(AF_INET, (const void *)&((struct sockaddr_in *)&s->cli_addr)->sin_addr, inet_ntop(AF_INET, (const void *)&((struct sockaddr_in *)&s->cli_addr)->sin_addr,
pn, sizeof(pn))) { pn, sizeof(pn))) {
send_log(p, LOG_INFO, "Connect from %s:%d to %s:%d (%s/%s)\n", send_log(s->fe, LOG_INFO, "Connect from %s:%d to %s:%d (%s/%s)\n",
pn, ntohs(((struct sockaddr_in *)&s->cli_addr)->sin_port), pn, ntohs(((struct sockaddr_in *)&s->cli_addr)->sin_port),
sn, ntohs(((struct sockaddr_in *)&s->frt_addr)->sin_port), sn, ntohs(((struct sockaddr_in *)&s->frt_addr)->sin_port),
p->id, (p->mode == PR_MODE_HTTP) ? "HTTP" : "TCP"); s->fe->id, (s->fe->mode == PR_MODE_HTTP) ? "HTTP" : "TCP");
} }
} }
else { else {
@ -348,10 +161,10 @@ int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
sn, sizeof(sn)) && sn, sizeof(sn)) &&
inet_ntop(AF_INET6, (const void *)&((struct sockaddr_in6 *)&s->cli_addr)->sin6_addr, inet_ntop(AF_INET6, (const void *)&((struct sockaddr_in6 *)&s->cli_addr)->sin6_addr,
pn, sizeof(pn))) { pn, sizeof(pn))) {
send_log(p, LOG_INFO, "Connect from %s:%d to %s:%d (%s/%s)\n", send_log(s->fe, LOG_INFO, "Connect from %s:%d to %s:%d (%s/%s)\n",
pn, ntohs(((struct sockaddr_in6 *)&s->cli_addr)->sin6_port), pn, ntohs(((struct sockaddr_in6 *)&s->cli_addr)->sin6_port),
sn, ntohs(((struct sockaddr_in6 *)&s->frt_addr)->sin6_port), sn, ntohs(((struct sockaddr_in6 *)&s->frt_addr)->sin6_port),
p->id, (p->mode == PR_MODE_HTTP) ? "HTTP" : "TCP"); s->fe->id, (s->fe->mode == PR_MODE_HTTP) ? "HTTP" : "TCP");
} }
} }
} }
@ -369,7 +182,7 @@ int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
pn, sizeof(pn)); pn, sizeof(pn));
len = sprintf(trash, "%08x:%s.accept(%04x)=%04x from [%s:%d]\n", len = sprintf(trash, "%08x:%s.accept(%04x)=%04x from [%s:%d]\n",
s->uniq_id, p->id, (unsigned short)l->fd, (unsigned short)cfd, s->uniq_id, s->fe->id, (unsigned short)s->listener->fd, (unsigned short)cfd,
pn, ntohs(((struct sockaddr_in *)&s->cli_addr)->sin_port)); pn, ntohs(((struct sockaddr_in *)&s->cli_addr)->sin_port));
} }
else { else {
@ -379,14 +192,14 @@ int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
pn, sizeof(pn)); pn, sizeof(pn));
len = sprintf(trash, "%08x:%s.accept(%04x)=%04x from [%s:%d]\n", len = sprintf(trash, "%08x:%s.accept(%04x)=%04x from [%s:%d]\n",
s->uniq_id, p->id, (unsigned short)l->fd, (unsigned short)cfd, s->uniq_id, s->fe->id, (unsigned short)s->listener->fd, (unsigned short)cfd,
pn, ntohs(((struct sockaddr_in6 *)(&s->cli_addr))->sin6_port)); pn, ntohs(((struct sockaddr_in6 *)(&s->cli_addr))->sin6_port));
} }
write(1, trash, len); write(1, trash, len);
} }
if (p->mode == PR_MODE_HTTP) if (s->fe->mode == PR_MODE_HTTP)
s->req->flags |= BF_READ_DONTWAIT; /* one read is usually enough */ s->req->flags |= BF_READ_DONTWAIT; /* one read is usually enough */
/* note: this should not happen anymore since there's always at least the switching rules */ /* note: this should not happen anymore since there's always at least the switching rules */
@ -399,11 +212,11 @@ int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
s->rep->wto = s->fe->timeout.client; s->rep->wto = s->fe->timeout.client;
fdtab[cfd].flags = FD_FL_TCP | FD_FL_TCP_NODELAY; fdtab[cfd].flags = FD_FL_TCP | FD_FL_TCP_NODELAY;
if (p->options & PR_O_TCP_NOLING) if (s->fe->options & PR_O_TCP_NOLING)
fdtab[cfd].flags |= FD_FL_TCP_NOLING; fdtab[cfd].flags |= FD_FL_TCP_NOLING;
if (unlikely((p->mode == PR_MODE_HTTP && (s->flags & SN_MONITOR)) || if (unlikely((s->fe->mode == PR_MODE_HTTP && (s->flags & SN_MONITOR)) ||
(p->mode == PR_MODE_HEALTH && (p->options & PR_O_HTTP_CHK)))) { (s->fe->mode == PR_MODE_HEALTH && (s->fe->options & PR_O_HTTP_CHK)))) {
/* Either we got a request from a monitoring system on an HTTP instance, /* 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 * 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. * both cases, we return a fake "HTTP/1.0 200 OK" response and we exit.
@ -412,44 +225,27 @@ int frontend_accept(struct listener *l, int cfd, struct sockaddr_storage *addr)
chunk_initstr(&msg, "HTTP/1.0 200 OK\r\n\r\n"); chunk_initstr(&msg, "HTTP/1.0 200 OK\r\n\r\n");
stream_int_retnclose(&s->si[0], &msg); /* forge a 200 response */ stream_int_retnclose(&s->si[0], &msg); /* forge a 200 response */
s->req->analysers = 0; s->req->analysers = 0;
t->expire = s->rep->wex; s->task->expire = s->rep->wex;
EV_FD_CLR(cfd, DIR_RD); EV_FD_CLR(cfd, DIR_RD);
} }
else if (unlikely(p->mode == PR_MODE_HEALTH)) { /* health check mode, no client reading */ else if (unlikely(s->fe->mode == PR_MODE_HEALTH)) { /* health check mode, no client reading */
struct chunk msg; struct chunk msg;
chunk_initstr(&msg, "OK\n"); chunk_initstr(&msg, "OK\n");
stream_int_retnclose(&s->si[0], &msg); /* forge an "OK" response */ stream_int_retnclose(&s->si[0], &msg); /* forge an "OK" response */
s->req->analysers = 0; s->req->analysers = 0;
t->expire = s->rep->wex; s->task->expire = s->rep->wex;
EV_FD_CLR(cfd, DIR_RD); EV_FD_CLR(cfd, DIR_RD);
} }
/* everything's OK, let's go on */
/**********************************************/
/* 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);
return 1; return 1;
/* Error unrolling */ /* Error unrolling */
out_free_rspcap: out_free_rspcap:
pool_free2(p->rsp_cap_pool, txn->rsp.cap); pool_free2(s->fe->rsp_cap_pool, s->txn.rsp.cap);
out_free_reqcap: out_free_reqcap:
pool_free2(p->req_cap_pool, txn->req.cap); pool_free2(s->fe->req_cap_pool, s->txn.req.cap);
out_delete_cfd: out_delete_cfd:
fd_delete(cfd); fd_delete(cfd);
pool_free2(pool2_buffer, s->rep);
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; return -1;
} }

237
src/session.c
View File

@ -1,7 +1,7 @@
/* /*
* Server management functions. * Session management functions.
* *
* Copyright 2000-2008 Willy Tarreau <w@1wt.eu> * Copyright 2000-2010 Willy Tarreau <w@1wt.eu>
* *
* This program is free software; you can redistribute it and/or * This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License * modify it under the terms of the GNU General Public License
@ -11,6 +11,8 @@
*/ */
#include <stdlib.h> #include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <common/config.h> #include <common/config.h>
#include <common/debug.h> #include <common/debug.h>
@ -42,6 +44,237 @@
struct pool_head *pool2_session; struct pool_head *pool2_session;
struct list sessions; struct list sessions;
/* This function is called from the protocol layer accept() in order to instanciate
* a new session on behalf of a given listener and frontend. 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 session_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;
if (unlikely((s = pool_alloc2(pool2_session)) == NULL)) {
Alert("out of memory in event_accept().\n");
goto out_close;
}
/* minimum session initialization required for monitor mode below */
s->flags = 0;
s->logs.logwait = p->to_log;
/* 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, but
* cleanly.
*/
if (unlikely((l->options & LI_O_CHK_MONNET) &&
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;
s->logs.logwait = 0;
}
/* OK, we're keeping the session, so let's properly initialize the session */
LIST_ADDQ(&sessions, &s->list);
LIST_INIT(&s->back_refs);
if (unlikely((t = task_new()) == NULL)) { /* disable this proxy for a while */
Alert("out of memory in event_accept().\n");
goto out_free_session;
}
s->term_trace = 0;
s->cli_addr = *addr;
s->logs.accept_date = date; /* user-visible date for logging */
s->logs.tv_accept = now; /* corrected date for internal use */
s->uniq_id = totalconn;
proxy_inc_fe_ctr(l, p); /* note: cum_beconn will be increased once assigned */
t->process = l->handler;
t->context = s;
t->nice = l->nice;
t->expire = TICK_ETERNITY;
s->task = t;
s->listener = l;
/* 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 */
/* now evaluate the tcp-request layer4 rules. Since we expect to be able
* to abort right here as soon as possible, we check the rules before
* even initializing the stream interfaces.
*/
if ((l->options & LI_O_TCP_RULES) && !tcp_exec_req_rules(s)) {
task_free(t);
LIST_DEL(&s->list);
pool_free2(pool2_session, s);
/* let's do a no-linger now to close with a single RST. */
setsockopt(cfd, SOL_SOCKET, SO_LINGER, (struct linger *) &nolinger, sizeof(struct linger));
return 0;
}
/* this part should be common with other protocols */
s->si[0].fd = cfd;
s->si[0].owner = t;
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].connect = NULL;
s->si[0].iohandler = NULL;
s->si[0].exp = TICK_ETERNITY;
s->si[0].flags = SI_FL_NONE;
if (likely(s->fe->options2 & PR_O2_INDEPSTR))
s->si[0].flags |= SI_FL_INDEP_STR;
if (addr->ss_family == AF_INET || addr->ss_family == AF_INET6)
s->si[0].flags = SI_FL_CAP_SPLTCP; /* TCP/TCPv6 splicing possible */
/* add the various callbacks */
stream_sock_prepare_interface(&s->si[0]);
/* pre-initialize the other side's stream interface to an INIT state. The
* callbacks will be initialized before attempting to connect.
*/
s->si[1].fd = -1; /* just to help with debugging */
s->si[1].owner = t;
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].connect = NULL;
s->si[1].iohandler = NULL;
s->si[1].shutr = stream_int_shutr;
s->si[1].shutw = stream_int_shutw;
s->si[1].exp = TICK_ETERNITY;
s->si[1].flags = SI_FL_NONE;
if (likely(s->fe->options2 & PR_O2_INDEPSTR))
s->si[1].flags |= SI_FL_INDEP_STR;
s->srv = s->prev_srv = s->srv_conn = NULL;
s->pend_pos = NULL;
/* init store persistence */
s->store_count = 0;
/* Adjust some socket options */
if (unlikely(fcntl(cfd, F_SETFL, O_NONBLOCK) == -1)) {
Alert("accept(): cannot set the socket in non blocking mode. Giving up\n");
goto out_free_task;
}
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;
if (unlikely((s->req = pool_alloc2(pool2_buffer)) == NULL))
goto out_free_task; /* no memory */
if (unlikely((s->rep = pool_alloc2(pool2_buffer)) == NULL))
goto out_free_req; /* no memory */
/* initialize the request buffer */
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 */
/* activate default analysers enabled for this listener */
s->req->analysers = l->analysers;
s->req->wto = TICK_ETERNITY;
s->req->rto = TICK_ETERNITY;
s->req->rex = TICK_ETERNITY;
s->req->wex = TICK_ETERNITY;
s->req->analyse_exp = TICK_ETERNITY;
/* initialize response buffer */
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->wto = TICK_ETERNITY;
s->rep->rex = TICK_ETERNITY;
s->rep->wex = TICK_ETERNITY;
s->rep->analyse_exp = TICK_ETERNITY;
/* finish initialization of the accepted file descriptor */
fd_insert(cfd);
fdtab[cfd].owner = &s->si[0];
fdtab[cfd].state = FD_STREADY;
fdtab[cfd].flags = 0;
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);
if (p->accept) {
int ret = p->accept(s);
if (unlikely(ret < 0))
goto out_free_rep;
if (unlikely(ret == 0)) {
/* work is finished, we can release everything (eg: monitoring) */
pool_free2(pool2_buffer, s->rep);
pool_free2(pool2_buffer, s->req);
task_free(t);
LIST_DEL(&s->list);
pool_free2(pool2_session, s);
return 0;
}
}
/* 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);
return 1;
/* Error unrolling */
out_free_rep:
pool_free2(pool2_buffer, s->rep);
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;
}
/* /*
* frees the context associated to a session. It must have been removed first. * frees the context associated to a session. It must have been removed first.
*/ */