From: Willy Tarreau Date: Sat, 4 Apr 2015 16:50:31 +0000 (+0200) Subject: REORG: session: move the session parts out of stream.c X-Git-Tag: v1.6-dev2~245 X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=9903f0e1a276f635a3cc43bdb988454028357df7;p=thirdparty%2Fhaproxy.git REORG: session: move the session parts out of stream.c This concerns everythins related to accepting a new session and expiring the embryonic session. There's still a hard-coded call to stream_accept_session() which could be set somewhere in the frontend, but for now it's not a problem. --- diff --git a/include/proto/session.h b/include/proto/session.h index 6d45d58059..3f008e16e3 100644 --- a/include/proto/session.h +++ b/include/proto/session.h @@ -35,6 +35,7 @@ extern struct pool_head *pool2_session; void session_free(struct session *sess); int init_session(); +int session_accept_fd(struct listener *l, int cfd, struct sockaddr_storage *addr); /* Remove the refcount from the session to the tracked counters, and clear the * pointer to ensure this is only performed once. The caller is responsible for diff --git a/include/proto/stream.h b/include/proto/stream.h index f8eba1298b..02992fd51e 100644 --- a/include/proto/stream.h +++ b/include/proto/stream.h @@ -36,7 +36,7 @@ extern struct list buffer_wq; extern struct data_cb sess_conn_cb; -int stream_accept(struct listener *l, int cfd, struct sockaddr_storage *addr); +int stream_accept_session(struct session *sess, struct task *t); /* perform minimal intializations, report 0 in case of error, 1 if OK. */ int init_stream(); diff --git a/src/cfgparse.c b/src/cfgparse.c index 92f072e4df..edaee56a02 100644 --- a/src/cfgparse.c +++ b/src/cfgparse.c @@ -73,6 +73,7 @@ #include #include #include +#include #include #include #include @@ -1884,7 +1885,7 @@ int cfg_parse_peers(const char *file, int linenum, char **args, int kwm) l->maxaccept = 1; l->maxconn = ((struct proxy *)curpeers->peers_fe)->maxconn; l->backlog = ((struct proxy *)curpeers->peers_fe)->backlog; - l->accept = stream_accept; + l->accept = session_accept_fd; l->handler = process_stream; l->analysers |= ((struct proxy *)curpeers->peers_fe)->fe_req_ana; l->default_target = ((struct proxy *)curpeers->peers_fe)->default_target; @@ -7708,7 +7709,7 @@ out_uri_auth_compat: listener->maxaccept = (listener->maxaccept + nbproc - 1) / nbproc; } - listener->accept = stream_accept; + listener->accept = session_accept_fd; listener->handler = process_stream; listener->analysers |= curproxy->fe_req_ana; listener->default_target = curproxy->default_target; diff --git a/src/dumpstats.c b/src/dumpstats.c index de86339142..d82ce85388 100644 --- a/src/dumpstats.c +++ b/src/dumpstats.c @@ -56,6 +56,7 @@ #include #include #include +#include #include #include #include @@ -337,7 +338,7 @@ static int stats_parse_global(char **args, int section_type, struct proxy *curpx list_for_each_entry(l, &bind_conf->listeners, by_bind) { l->maxconn = global.stats_fe->maxconn; l->backlog = global.stats_fe->backlog; - l->accept = stream_accept; + l->accept = session_accept_fd; l->handler = process_stream; l->default_target = global.stats_fe->default_target; l->options |= LI_O_UNLIMITED; /* don't make the peers subject to global limits */ diff --git a/src/session.c b/src/session.c index caf8c1439f..070c340b89 100644 --- a/src/session.c +++ b/src/session.c @@ -1,7 +1,7 @@ /* - * Stream management functions. + * Session management functions. * - * Copyright 2000-2012 Willy Tarreau + * Copyright 2000-2015 Willy Tarreau * * This program is free software; you can redistribute it and/or * modify it under the terms of the GNU General Public License @@ -18,10 +18,30 @@ #include #include +#include +#include +#include +#include +#include +#include +#include #include +#include struct pool_head *pool2_session; +static int conn_complete_session(struct connection *conn); +static int conn_update_session(struct connection *conn); +static struct task *session_expire_embryonic(struct task *t); + +/* data layer callbacks for an embryonic stream */ +struct data_cb sess_conn_cb = { + .recv = NULL, + .send = NULL, + .wake = conn_update_session, + .init = conn_complete_session, +}; + void session_free(struct session *sess) { session_store_counters(sess); @@ -35,6 +55,301 @@ int init_session() return pool2_session != NULL; } +/* 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 can be ignored, + * or a negative value upon critical failure. The accepted file descriptor is + * closed if we return <= 0. If no handshake is needed, it immediately tries + * to instanciate a new stream. + */ +int session_accept_fd(struct listener *l, int cfd, struct sockaddr_storage *addr) +{ + struct connection *cli_conn; + struct proxy *p = l->frontend; + struct session *sess; + struct task *t; + int ret; + + + ret = -1; /* assume unrecoverable error by default */ + + if (unlikely((cli_conn = conn_new()) == NULL)) + goto out_close; + + conn_prepare(cli_conn, l->proto, l->xprt); + + cli_conn->t.sock.fd = cfd; + cli_conn->addr.from = *addr; + cli_conn->flags |= CO_FL_ADDR_FROM_SET; + cli_conn->target = &l->obj_type; + cli_conn->proxy_netns = l->netns; + + conn_ctrl_init(cli_conn); + + /* wait for a PROXY protocol header */ + if (l->options & LI_O_ACC_PROXY) { + cli_conn->flags |= CO_FL_ACCEPT_PROXY; + conn_sock_want_recv(cli_conn); + } + + conn_data_want_recv(cli_conn); + if (conn_xprt_init(cli_conn) < 0) + goto out_free_conn; + + sess = pool_alloc2(pool2_session); + if (!sess) + goto out_free_conn; + + p->feconn++; + /* This session was accepted, count it now */ + if (p->feconn > p->fe_counters.conn_max) + p->fe_counters.conn_max = p->feconn; + + proxy_inc_fe_conn_ctr(l, p); + + sess->listener = l; + sess->fe = p; + sess->origin = &cli_conn->obj_type; + sess->accept_date = date; /* user-visible date for logging */ + sess->tv_accept = now; /* corrected date for internal use */ + memset(sess->stkctr, 0, sizeof(sess->stkctr)); + + /* now evaluate the tcp-request layer4 rules. We only need a session + * and no stream for these rules. + */ + if ((l->options & LI_O_TCP_RULES) && !tcp_exec_req_rules(sess)) { + /* 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)); + ret = 0; /* successful termination */ + goto out_free_sess; + } + + /* monitor-net and health mode are processed immediately after TCP + * connection rules. This way it's possible to block them, but they + * never use the lower data layers, they send directly over the socket, + * as they were designed for. We first flush the socket receive buffer + * in order to avoid emission of an RST by the system. We ignore any + * error. + */ + if (unlikely((p->mode == PR_MODE_HEALTH) || + ((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))) { + /* we have 4 possibilities here : + * - HTTP mode, from monitoring address => send "HTTP/1.0 200 OK" + * - HEALTH mode with HTTP check => send "HTTP/1.0 200 OK" + * - HEALTH mode without HTTP check => just send "OK" + * - TCP mode from monitoring address => just close + */ + if (l->proto->drain) + l->proto->drain(cfd); + if (p->mode == PR_MODE_HTTP || + (p->mode == PR_MODE_HEALTH && (p->options2 & PR_O2_CHK_ANY) == PR_O2_HTTP_CHK)) + send(cfd, "HTTP/1.0 200 OK\r\n\r\n", 19, MSG_DONTWAIT|MSG_NOSIGNAL|MSG_MORE); + else if (p->mode == PR_MODE_HEALTH) + send(cfd, "OK\n", 3, MSG_DONTWAIT|MSG_NOSIGNAL|MSG_MORE); + ret = 0; + goto out_free_sess; + } + + if (unlikely((t = task_new()) == NULL)) + goto out_free_sess; + + t->context = sess; + t->nice = l->nice; + + /* OK, now either we have a pending handshake to execute with and + * then we must return to the I/O layer, or we can proceed with the + * end of the stream initialization. In case of handshake, we also + * set the I/O timeout to the frontend's client timeout. + * + * At this point we set the relation between sess/task/conn this way : + * + * orig -- sess <-- context + * | | + * v | + * conn -- owner ---> task + */ + if (cli_conn->flags & CO_FL_HANDSHAKE) { + conn_attach(cli_conn, t, &sess_conn_cb); + t->process = session_expire_embryonic; + t->expire = tick_add_ifset(now_ms, p->timeout.client); + task_queue(t); + cli_conn->flags |= CO_FL_INIT_DATA | CO_FL_WAKE_DATA; + return 1; + } + + ret = stream_accept_session(sess, t); + if (ret > 0) + return ret; + + task_free(t); + out_free_sess: + p->feconn--; + session_free(sess); + out_free_conn: + cli_conn->flags &= ~CO_FL_XPRT_TRACKED; + conn_xprt_close(cli_conn); + conn_free(cli_conn); + out_close: + if (ret < 0 && l->xprt == &raw_sock && p->mode == PR_MODE_HTTP) { + /* critical error, no more memory, try to emit a 500 response */ + struct chunk *err_msg = &p->errmsg[HTTP_ERR_500]; + if (!err_msg->str) + err_msg = &http_err_chunks[HTTP_ERR_500]; + send(cfd, err_msg->str, err_msg->len, MSG_DONTWAIT|MSG_NOSIGNAL); + } + + if (fdtab[cfd].owner) + fd_delete(cfd); + else + close(cfd); + return ret; +} + + +/* prepare the trash with a log prefix for session . It only works with + * embryonic sessions based on a real connection. This function requires that + * at sess->origin points to the incoming connection. + */ +static void session_prepare_log_prefix(struct session *sess) +{ + struct tm tm; + char pn[INET6_ADDRSTRLEN]; + int ret; + char *end; + struct connection *cli_conn = __objt_conn(sess->origin); + + ret = addr_to_str(&cli_conn->addr.from, pn, sizeof(pn)); + if (ret <= 0) + chunk_printf(&trash, "unknown ["); + else if (ret == AF_UNIX) + chunk_printf(&trash, "%s:%d [", pn, sess->listener->luid); + else + chunk_printf(&trash, "%s:%d [", pn, get_host_port(&cli_conn->addr.from)); + + get_localtime(sess->accept_date.tv_sec, &tm); + end = date2str_log(trash.str + trash.len, &tm, &(sess->accept_date), trash.size - trash.len); + trash.len = end - trash.str; + if (sess->listener->name) + chunk_appendf(&trash, "] %s/%s", sess->fe->id, sess->listener->name); + else + chunk_appendf(&trash, "] %s/%d", sess->fe->id, sess->listener->luid); +} + +/* This function kills an existing embryonic session. It stops the connection's + * transport layer, releases assigned resources, resumes the listener if it was + * disabled and finally kills the file descriptor. This function requires that + * sess->origin points to the incoming connection. + */ +static void session_kill_embryonic(struct session *sess) +{ + int level = LOG_INFO; + struct connection *conn = __objt_conn(sess->origin); + struct task *task = conn->owner; + unsigned int log = sess->fe->to_log; + const char *err_msg; + + if (sess->fe->options2 & PR_O2_LOGERRORS) + level = LOG_ERR; + + if (log && (sess->fe->options & PR_O_NULLNOLOG)) { + /* with "option dontlognull", we don't log connections with no transfer */ + if (!conn->err_code || + conn->err_code == CO_ER_PRX_EMPTY || conn->err_code == CO_ER_PRX_ABORT || + conn->err_code == CO_ER_SSL_EMPTY || conn->err_code == CO_ER_SSL_ABORT) + log = 0; + } + + if (log) { + if (!conn->err_code && (task->state & TASK_WOKEN_TIMER)) { + if (conn->flags & CO_FL_ACCEPT_PROXY) + conn->err_code = CO_ER_PRX_TIMEOUT; + else if (conn->flags & CO_FL_SSL_WAIT_HS) + conn->err_code = CO_ER_SSL_TIMEOUT; + } + + session_prepare_log_prefix(sess); + err_msg = conn_err_code_str(conn); + if (err_msg) + send_log(sess->fe, level, "%s: %s\n", trash.str, err_msg); + else + send_log(sess->fe, level, "%s: unknown connection error (code=%d flags=%08x)\n", + trash.str, conn->err_code, conn->flags); + } + + /* kill the connection now */ + conn_force_close(conn); + conn_free(conn); + + sess->fe->feconn--; + + if (!(sess->listener->options & LI_O_UNLIMITED)) + actconn--; + jobs--; + sess->listener->nbconn--; + if (sess->listener->state == LI_FULL) + resume_listener(sess->listener); + + /* Dequeues all of the listeners waiting for a resource */ + if (!LIST_ISEMPTY(&global_listener_queue)) + dequeue_all_listeners(&global_listener_queue); + + if (!LIST_ISEMPTY(&sess->fe->listener_queue) && + (!sess->fe->fe_sps_lim || freq_ctr_remain(&sess->fe->fe_sess_per_sec, sess->fe->fe_sps_lim, 0) > 0)) + dequeue_all_listeners(&sess->fe->listener_queue); + + task_delete(task); + task_free(task); + session_free(sess); +} + +/* Manages the embryonic session timeout. It is only called when the timeout + * strikes and performs the required cleanup. + */ +static struct task *session_expire_embryonic(struct task *t) +{ + struct session *sess = t->context; + + if (!(t->state & TASK_WOKEN_TIMER)) + return t; + + session_kill_embryonic(sess); + return NULL; +} + +/* Finish initializing a session from a connection, or kills it if the + * connection shows and error. Returns <0 if the connection was killed. + */ +static int conn_complete_session(struct connection *conn) +{ + struct task *task = conn->owner; + struct session *sess = task->context; + + if (!(conn->flags & CO_FL_ERROR) && (stream_accept_session(sess, task) > 0)) { + conn->flags &= ~CO_FL_INIT_DATA; + return 0; + } + + session_kill_embryonic(sess); + return -1; +} + +/* Update a session status. The connection is killed in case of + * error, and <0 will be returned. Otherwise it does nothing. + */ +static int conn_update_session(struct connection *conn) +{ + struct task *task = conn->owner; + struct session *sess = task->context; + + if (conn->flags & CO_FL_ERROR) { + session_kill_embryonic(sess); + return -1; + } + return 0; +} + /* * Local variables: * c-indent-level: 8 diff --git a/src/stream.c b/src/stream.c index d55b0ba218..e0dc824a8c 100644 --- a/src/stream.c +++ b/src/stream.c @@ -56,328 +56,15 @@ struct list streams; /* list of streams waiting for at least one buffer */ struct list buffer_wq = LIST_HEAD_INIT(buffer_wq); -static int conn_stream_complete(struct connection *conn); -static int conn_stream_update(struct connection *conn); -static struct task *expire_mini_session(struct task *t); -int stream_complete(struct session *s, struct task *t); - -/* data layer callbacks for an embryonic stream */ -struct data_cb sess_conn_cb = { - .recv = NULL, - .send = NULL, - .wake = conn_stream_update, - .init = conn_stream_complete, -}; - -/* This function is called from the protocol layer accept() in order to - * instanciate a new embryonic stream on behalf of a given listener and - * frontend. It returns a positive value upon success, 0 if the connection - * can be ignored, or a negative value upon critical failure. The accepted - * file descriptor is closed if we return <= 0. - */ -int stream_accept(struct listener *l, int cfd, struct sockaddr_storage *addr) -{ - struct connection *cli_conn; - struct proxy *p = l->frontend; - struct session *sess; - struct task *t; - int ret; - - - ret = -1; /* assume unrecoverable error by default */ - - if (unlikely((cli_conn = conn_new()) == NULL)) - goto out_close; - - conn_prepare(cli_conn, l->proto, l->xprt); - - cli_conn->t.sock.fd = cfd; - cli_conn->addr.from = *addr; - cli_conn->flags |= CO_FL_ADDR_FROM_SET; - cli_conn->target = &l->obj_type; - cli_conn->proxy_netns = l->netns; - - conn_ctrl_init(cli_conn); - - /* wait for a PROXY protocol header */ - if (l->options & LI_O_ACC_PROXY) { - cli_conn->flags |= CO_FL_ACCEPT_PROXY; - conn_sock_want_recv(cli_conn); - } - - /* Finish setting the callbacks. Right now the transport layer is present - * but not initialized. Also note we need to be careful as the stream - * int is not initialized yet. - */ - conn_data_want_recv(cli_conn); - if (conn_xprt_init(cli_conn) < 0) - goto out_free_conn; - - sess = pool_alloc2(pool2_session); - if (!sess) - goto out_free_conn; - - p->feconn++; - /* This session was accepted, count it now */ - if (p->feconn > p->fe_counters.conn_max) - p->fe_counters.conn_max = p->feconn; - - proxy_inc_fe_conn_ctr(l, p); - - sess->listener = l; - sess->fe = p; - sess->origin = &cli_conn->obj_type; - sess->accept_date = date; /* user-visible date for logging */ - sess->tv_accept = now; /* corrected date for internal use */ - memset(sess->stkctr, 0, sizeof(sess->stkctr)); - - /* 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(sess)) { - /* 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)); - ret = 0; /* successful termination */ - goto out_free_sess; - } - - /* monitor-net and health mode are processed immediately after TCP - * connection rules. This way it's possible to block them, but they - * never use the lower data layers, they send directly over the socket, - * as they were designed for. We first flush the socket receive buffer - * in order to avoid emission of an RST by the system. We ignore any - * error. - */ - if (unlikely((p->mode == PR_MODE_HEALTH) || - ((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))) { - /* we have 4 possibilities here : - * - HTTP mode, from monitoring address => send "HTTP/1.0 200 OK" - * - HEALTH mode with HTTP check => send "HTTP/1.0 200 OK" - * - HEALTH mode without HTTP check => just send "OK" - * - TCP mode from monitoring address => just close - */ - if (l->proto->drain) - l->proto->drain(cfd); - if (p->mode == PR_MODE_HTTP || - (p->mode == PR_MODE_HEALTH && (p->options2 & PR_O2_CHK_ANY) == PR_O2_HTTP_CHK)) - send(cfd, "HTTP/1.0 200 OK\r\n\r\n", 19, MSG_DONTWAIT|MSG_NOSIGNAL|MSG_MORE); - else if (p->mode == PR_MODE_HEALTH) - send(cfd, "OK\n", 3, MSG_DONTWAIT|MSG_NOSIGNAL|MSG_MORE); - ret = 0; - goto out_free_sess; - } - - if (unlikely((t = task_new()) == NULL)) - goto out_free_sess; - - t->context = sess; - t->nice = l->nice; - - /* OK, now either we have a pending handshake to execute with and - * then we must return to the I/O layer, or we can proceed with the - * end of the stream initialization. In case of handshake, we also - * set the I/O timeout to the frontend's client timeout. - * - * At this point we set the relation between sess/task/conn this way : - * - * orig -- sess <-- context - * | | - * v | - * conn -- owner ---> task - */ - if (cli_conn->flags & CO_FL_HANDSHAKE) { - conn_attach(cli_conn, t, &sess_conn_cb); - t->process = expire_mini_session; - t->expire = tick_add_ifset(now_ms, p->timeout.client); - task_queue(t); - cli_conn->flags |= CO_FL_INIT_DATA | CO_FL_WAKE_DATA; - return 1; - } - - ret = stream_complete(sess, t); - if (ret > 0) - return ret; - - task_free(t); - out_free_sess: - p->feconn--; - session_free(sess); - out_free_conn: - cli_conn->flags &= ~CO_FL_XPRT_TRACKED; - conn_xprt_close(cli_conn); - conn_free(cli_conn); - out_close: - if (ret < 0 && l->xprt == &raw_sock && p->mode == PR_MODE_HTTP) { - /* critical error, no more memory, try to emit a 500 response */ - struct chunk *err_msg = &p->errmsg[HTTP_ERR_500]; - if (!err_msg->str) - err_msg = &http_err_chunks[HTTP_ERR_500]; - send(cfd, err_msg->str, err_msg->len, MSG_DONTWAIT|MSG_NOSIGNAL); - } - - if (fdtab[cfd].owner) - fd_delete(cfd); - else - close(cfd); - return ret; -} - - -/* prepare the trash with a log prefix for session . It only works with - * embryonic streams based on a real connection. This function requires that - * at sess->origin points to the incoming connection. - */ -static void prepare_mini_sess_log_prefix(struct session *sess) -{ - struct tm tm; - char pn[INET6_ADDRSTRLEN]; - int ret; - char *end; - struct connection *cli_conn = __objt_conn(sess->origin); - - ret = addr_to_str(&cli_conn->addr.from, pn, sizeof(pn)); - if (ret <= 0) - chunk_printf(&trash, "unknown ["); - else if (ret == AF_UNIX) - chunk_printf(&trash, "%s:%d [", pn, sess->listener->luid); - else - chunk_printf(&trash, "%s:%d [", pn, get_host_port(&cli_conn->addr.from)); - - get_localtime(sess->accept_date.tv_sec, &tm); - end = date2str_log(trash.str + trash.len, &tm, &(sess->accept_date), trash.size - trash.len); - trash.len = end - trash.str; - if (sess->listener->name) - chunk_appendf(&trash, "] %s/%s", sess->fe->id, sess->listener->name); - else - chunk_appendf(&trash, "] %s/%d", sess->fe->id, sess->listener->luid); -} - -/* This function kills an existing embryonic stream. It stops the connection's - * transport layer, releases assigned resources, resumes the listener if it was - * disabled and finally kills the file descriptor. This function requires that - * at sess->origin points to the incoming connection. - */ -static void kill_mini_session(struct session *sess) -{ - int level = LOG_INFO; - struct connection *conn = __objt_conn(sess->origin); - struct task *task = conn->owner; - unsigned int log = sess->fe->to_log; - const char *err_msg; - - if (sess->fe->options2 & PR_O2_LOGERRORS) - level = LOG_ERR; - - if (log && (sess->fe->options & PR_O_NULLNOLOG)) { - /* with "option dontlognull", we don't log connections with no transfer */ - if (!conn->err_code || - conn->err_code == CO_ER_PRX_EMPTY || conn->err_code == CO_ER_PRX_ABORT || - conn->err_code == CO_ER_SSL_EMPTY || conn->err_code == CO_ER_SSL_ABORT) - log = 0; - } - - if (log) { - if (!conn->err_code && (task->state & TASK_WOKEN_TIMER)) { - if (conn->flags & CO_FL_ACCEPT_PROXY) - conn->err_code = CO_ER_PRX_TIMEOUT; - else if (conn->flags & CO_FL_SSL_WAIT_HS) - conn->err_code = CO_ER_SSL_TIMEOUT; - } - - prepare_mini_sess_log_prefix(sess); - err_msg = conn_err_code_str(conn); - if (err_msg) - send_log(sess->fe, level, "%s: %s\n", trash.str, err_msg); - else - send_log(sess->fe, level, "%s: unknown connection error (code=%d flags=%08x)\n", - trash.str, conn->err_code, conn->flags); - } - - /* kill the connection now */ - conn_force_close(conn); - conn_free(conn); - - sess->fe->feconn--; - - if (!(sess->listener->options & LI_O_UNLIMITED)) - actconn--; - jobs--; - sess->listener->nbconn--; - if (sess->listener->state == LI_FULL) - resume_listener(sess->listener); - - /* Dequeues all of the listeners waiting for a resource */ - if (!LIST_ISEMPTY(&global_listener_queue)) - dequeue_all_listeners(&global_listener_queue); - - if (!LIST_ISEMPTY(&sess->fe->listener_queue) && - (!sess->fe->fe_sps_lim || freq_ctr_remain(&sess->fe->fe_sess_per_sec, sess->fe->fe_sps_lim, 0) > 0)) - dequeue_all_listeners(&sess->fe->listener_queue); - - task_delete(task); - task_free(task); - session_free(sess); -} - -/* Finish initializing a stream from a connection, or kills it if the - * connection shows and error. Returns <0 if the connection was killed. - */ -static int conn_stream_complete(struct connection *conn) -{ - struct task *task = conn->owner; - struct session *sess = task->context; - - if (!(conn->flags & CO_FL_ERROR) && (stream_complete(sess, task) > 0)) { - conn->flags &= ~CO_FL_INIT_DATA; - return 0; - } - - /* kill the connection now */ - kill_mini_session(sess); - return -1; -} - -/* Update an embryonic stream status. The connection is killed in case of - * error, and <0 will be returned. Otherwise it does nothing. - */ -static int conn_stream_update(struct connection *conn) -{ - struct task *task = conn->owner; - struct session *sess = task->context; - - if (conn->flags & CO_FL_ERROR) { - kill_mini_session(sess); - return -1; - } - return 0; -} - -/* Manages embryonic streams timeout. It is only called when the timeout - * strikes and performs the required cleanup. - */ -static struct task *expire_mini_session(struct task *t) -{ - struct session *sess = t->context; - - if (!(t->state & TASK_WOKEN_TIMER)) - return t; - - kill_mini_session(sess); - return NULL; -} - -/* This function is called from the I/O handler which detects the end of +/* This function is called from the session handler which detects the end of * handshake, in order to complete initialization of a valid stream. It must - * be called with an embryonic stream. It returns a positive value upon + * be called with an embryonic session. It returns a positive value upon * success, 0 if the connection can be ignored, or a negative value upon * critical failure. The accepted file descriptor is closed if we return <= 0. * The client-side end point is assumed to be a connection, whose pointer is * taken from sess->origin which is assumed to be valid. */ -int stream_complete(struct session *sess, struct task *t) +int stream_accept_session(struct session *sess, struct task *t) { struct stream *s; struct listener *l = sess->listener;