mirrors/haproxy
1
0
Fork 0
mirror of https://git.haproxy.org/git/haproxy.git/ synced 2025-09-20 21:31:28 +02:00
Code Issues Projects Releases Wiki Activity

[MEDIUM] got rid of event_{cli,srv}_read() in favor of stream_sock_read()

Browse Source
This commit is contained in:
Willy Tarreau 2006-07-29 18:36:34 +02:00
parent 0f9f5056f9
commit d797128d6e
8 changed files with 176 additions and 276 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/proto/stream_sock.h
View File

@ -29,10 +29,10 @@
#include <common/config.h> #include <common/config.h>
int stream_sock_read(int fd);
/* FIXME: merge those ones together */ /* FIXME: merge those ones together */
int event_cli_read(int fd);
int event_cli_write(int fd); int event_cli_write(int fd);
int event_srv_read(int fd);
int event_srv_write(int fd); int event_srv_write(int fd);

6
include/types/buffers.h
View File

@ -59,6 +59,12 @@ struct chunk {
struct buffer { struct buffer {
u_int32_t flags; /* BF_* */ u_int32_t flags; /* BF_* */
struct timeval rex; /* expiration date for a read */
struct timeval wex; /* expiration date for a write */
struct timeval cex; /* expiration date for a connect */
int rto; /* read timeout */
int wto; /* write timeout */
int cto; /* connect timeout */
unsigned int l; /* data length */ unsigned int l; /* data length */
char *r, *w, *h, *lr; /* read ptr, write ptr, last header ptr, last read */ char *r, *w, *h, *lr; /* read ptr, write ptr, last header ptr, last read */
char *rlim; /* read limit, used for header rewriting */ char *rlim; /* read limit, used for header rewriting */

5
include/types/session.h
View File

@ -101,11 +101,6 @@
struct session { struct session {
struct task *task; /* the task associated with this session */ struct task *task; /* the task associated with this session */
/* application specific below */ /* application specific below */
struct timeval crexpire; /* expiration date for a client read */
struct timeval cwexpire; /* expiration date for a client write */
struct timeval srexpire; /* expiration date for a server read */
struct timeval swexpire; /* expiration date for a server write */
struct timeval cnexpire; /* expiration date for a connect */
struct proxy *proxy; /* the proxy this socket belongs to */ struct proxy *proxy; /* the proxy this socket belongs to */
int cli_fd; /* the client side fd */ int cli_fd; /* the client side fd */
int srv_fd; /* the server side fd */ int srv_fd; /* the server side fd */

18
src/backend.c
View File

@ -427,7 +427,7 @@ int connect_server(struct session *s)
fdtab[fd].owner = s->task; fdtab[fd].owner = s->task;
fdtab[fd].state = FD_STCONN; /* connection in progress */ fdtab[fd].state = FD_STCONN; /* connection in progress */
fdtab[fd].cb[DIR_RD].f = &event_srv_read; fdtab[fd].cb[DIR_RD].f = &stream_sock_read;
fdtab[fd].cb[DIR_RD].b = s->rep; fdtab[fd].cb[DIR_RD].b = s->rep;
fdtab[fd].cb[DIR_WR].f = &event_srv_write; fdtab[fd].cb[DIR_WR].f = &event_srv_write;
fdtab[fd].cb[DIR_WR].b = s->req; fdtab[fd].cb[DIR_WR].b = s->req;
@ -448,9 +448,9 @@ int connect_server(struct session *s)
} }
if (s->proxy->contimeout) if (s->proxy->contimeout)
tv_delayfrom(&s->cnexpire, &now, s->proxy->contimeout); tv_delayfrom(&s->req->cex, &now, s->proxy->contimeout);
else else
tv_eternity(&s->cnexpire); tv_eternity(&s->req->cex);
return SN_ERR_NONE; /* connection is OK */ return SN_ERR_NONE; /* connection is OK */
} }
@ -467,7 +467,7 @@ int srv_count_retry_down(struct session *t, int conn_err)
t->conn_retries--; t->conn_retries--;
if (t->conn_retries < 0) { if (t->conn_retries < 0) {
/* if not retryable anymore, let's abort */ /* if not retryable anymore, let's abort */
tv_eternity(&t->cnexpire); tv_eternity(&t->req->cex);
srv_close_with_err(t, conn_err, SN_FINST_C, srv_close_with_err(t, conn_err, SN_FINST_C,
503, t->proxy->errmsg.len503, t->proxy->errmsg.msg503); 503, t->proxy->errmsg.len503, t->proxy->errmsg.msg503);
if (t->srv) if (t->srv)
@ -509,7 +509,7 @@ int srv_retryable_connect(struct session *t)
return 1; return 1;
case SN_ERR_INTERNAL: case SN_ERR_INTERNAL:
tv_eternity(&t->cnexpire); tv_eternity(&t->req->cex);
srv_close_with_err(t, SN_ERR_INTERNAL, SN_FINST_C, srv_close_with_err(t, SN_ERR_INTERNAL, SN_FINST_C,
500, t->proxy->errmsg.len500, t->proxy->errmsg.msg500); 500, t->proxy->errmsg.len500, t->proxy->errmsg.msg500);
if (t->srv) if (t->srv)
@ -572,7 +572,7 @@ int srv_redispatch_connect(struct session *t)
case SRV_STATUS_NOSRV: case SRV_STATUS_NOSRV:
/* note: it is guaranteed that t->srv == NULL here */ /* note: it is guaranteed that t->srv == NULL here */
tv_eternity(&t->cnexpire); tv_eternity(&t->req->cex);
srv_close_with_err(t, SN_ERR_SRVTO, SN_FINST_C, srv_close_with_err(t, SN_ERR_SRVTO, SN_FINST_C,
503, t->proxy->errmsg.len503, t->proxy->errmsg.msg503); 503, t->proxy->errmsg.len503, t->proxy->errmsg.msg503);
if (t->srv) if (t->srv)
@ -584,9 +584,9 @@ int srv_redispatch_connect(struct session *t)
case SRV_STATUS_QUEUED: case SRV_STATUS_QUEUED:
/* FIXME-20060503 : we should use the queue timeout instead */ /* FIXME-20060503 : we should use the queue timeout instead */
if (t->proxy->contimeout) if (t->proxy->contimeout)
tv_delayfrom(&t->cnexpire, &now, t->proxy->contimeout); tv_delayfrom(&t->req->cex, &now, t->proxy->contimeout);
else else
tv_eternity(&t->cnexpire); tv_eternity(&t->req->cex);
t->srv_state = SV_STIDLE; t->srv_state = SV_STIDLE;
/* do nothing else and do not wake any other session up */ /* do nothing else and do not wake any other session up */
return 1; return 1;
@ -594,7 +594,7 @@ int srv_redispatch_connect(struct session *t)
case SRV_STATUS_FULL: case SRV_STATUS_FULL:
case SRV_STATUS_INTERNAL: case SRV_STATUS_INTERNAL:
default: default:
tv_eternity(&t->cnexpire); tv_eternity(&t->req->cex);
srv_close_with_err(t, SN_ERR_INTERNAL, SN_FINST_C, srv_close_with_err(t, SN_ERR_INTERNAL, SN_FINST_C,
500, t->proxy->errmsg.len500, t->proxy->errmsg.msg500); 500, t->proxy->errmsg.len500, t->proxy->errmsg.msg500);
if (t->srv) if (t->srv)

26
src/client.c
View File

@ -314,6 +314,10 @@ int event_accept(int fd) {
if (s->cli_state == CL_STHEADERS) /* reserve some space for header rewriting */ if (s->cli_state == CL_STHEADERS) /* reserve some space for header rewriting */
s->req->rlim -= MAXREWRITE; s->req->rlim -= MAXREWRITE;
s->req->rto = s->proxy->clitimeout;
s->req->wto = s->proxy->srvtimeout;
s->req->cto = s->proxy->srvtimeout;
if ((s->rep = pool_alloc(buffer)) == NULL) { /* no memory */ if ((s->rep = pool_alloc(buffer)) == NULL) { /* no memory */
pool_free(buffer, s->req); pool_free(buffer, s->req);
if (s->rsp_cap != NULL) if (s->rsp_cap != NULL)
@ -328,9 +332,13 @@ int event_accept(int fd) {
buffer_init(s->rep); buffer_init(s->rep);
s->rep->rto = s->proxy->srvtimeout;
s->rep->wto = s->proxy->clitimeout;
s->rep->cto = 0;
fdtab[cfd].owner = t; fdtab[cfd].owner = t;
fdtab[cfd].state = FD_STREADY; fdtab[cfd].state = FD_STREADY;
fdtab[cfd].cb[DIR_RD].f = &event_cli_read; fdtab[cfd].cb[DIR_RD].f = &stream_sock_read;
fdtab[cfd].cb[DIR_RD].b = s->req; fdtab[cfd].cb[DIR_RD].b = s->req;
fdtab[cfd].cb[DIR_WR].f = &event_cli_write; fdtab[cfd].cb[DIR_WR].f = &event_cli_write;
fdtab[cfd].cb[DIR_WR].b = s->rep; fdtab[cfd].cb[DIR_WR].b = s->rep;
@ -357,20 +365,20 @@ int event_accept(int fd) {
#endif #endif
fd_insert(cfd); fd_insert(cfd);
tv_eternity(&s->cnexpire); tv_eternity(&s->req->rex);
tv_eternity(&s->srexpire); tv_eternity(&s->req->wex);
tv_eternity(&s->swexpire); tv_eternity(&s->req->cex);
tv_eternity(&s->crexpire); tv_eternity(&s->rep->rex);
tv_eternity(&s->cwexpire); tv_eternity(&s->rep->wex);
if (s->proxy->clitimeout) { if (s->proxy->clitimeout) {
if (FD_ISSET(cfd, StaticReadEvent)) if (FD_ISSET(cfd, StaticReadEvent))
tv_delayfrom(&s->crexpire, &now, s->proxy->clitimeout); tv_delayfrom(&s->req->rex, &now, s->proxy->clitimeout);
if (FD_ISSET(cfd, StaticWriteEvent)) if (FD_ISSET(cfd, StaticWriteEvent))
tv_delayfrom(&s->cwexpire, &now, s->proxy->clitimeout); tv_delayfrom(&s->rep->wex, &now, s->proxy->clitimeout);
} }
tv_min(&t->expire, &s->crexpire, &s->cwexpire); tv_min(&t->expire, &s->req->rex, &s->rep->wex);
task_queue(t); task_queue(t);

4
src/fd.c
View File

@ -15,9 +15,7 @@
* - we still use 'listeners' to check whether we want to stop or not. * - we still use 'listeners' to check whether we want to stop or not.
* - the various pollers should be moved to other external files, possibly * - the various pollers should be moved to other external files, possibly
* dynamic libs. * dynamic libs.
* - merge event_cli_read() and event_srv_read(). The difference is res_*, * - stream_sock_read() : It may be called from event_accept().
* buffer (at the beginning) and timeouts (at the end).
* => event_tcp_read(). It may be called from event_accept().
* - extract the connect code from event_srv_write() * - extract the connect code from event_srv_write()
* => event_tcp_connect(). It must then call event_write(). * => event_tcp_connect(). It must then call event_write().
* - merge the remaining event_cli_write() and event_srv_write() * - merge the remaining event_cli_write() and event_srv_write()

236
src/proto_http.c
View File

@ -87,11 +87,11 @@ void client_retnclose(struct session *s, int len, const char *msg)
{ {
FD_CLR(s->cli_fd, StaticReadEvent); FD_CLR(s->cli_fd, StaticReadEvent);
FD_SET(s->cli_fd, StaticWriteEvent); FD_SET(s->cli_fd, StaticWriteEvent);
tv_eternity(&s->crexpire); tv_eternity(&s->req->rex);
if (s->proxy->clitimeout) if (s->proxy->clitimeout)
tv_delayfrom(&s->cwexpire, &now, s->proxy->clitimeout); tv_delayfrom(&s->rep->wex, &now, s->proxy->clitimeout);
else else
tv_eternity(&s->cwexpire); tv_eternity(&s->rep->wex);
shutdown(s->cli_fd, SHUT_RD); shutdown(s->cli_fd, SHUT_RD);
s->cli_state = CL_STSHUTR; s->cli_state = CL_STSHUTR;
buffer_flush(s->rep); buffer_flush(s->rep);
@ -157,9 +157,9 @@ int process_session(struct task *t)
s->req->flags &= BF_CLEAR_READ & BF_CLEAR_WRITE; s->req->flags &= BF_CLEAR_READ & BF_CLEAR_WRITE;
s->rep->flags &= BF_CLEAR_READ & BF_CLEAR_WRITE; s->rep->flags &= BF_CLEAR_READ & BF_CLEAR_WRITE;
tv_min(&min1, &s->crexpire, &s->cwexpire); tv_min(&min1, &s->req->rex, &s->req->wex);
tv_min(&min2, &s->srexpire, &s->swexpire); tv_min(&min2, &s->rep->rex, &s->rep->wex);
tv_min(&min1, &min1, &s->cnexpire); tv_min(&min1, &min1, &s->req->cex);
tv_min(&t->expire, &min1, &min2); tv_min(&t->expire, &min1, &min2);
/* restore t to its place in the task list */ /* restore t to its place in the task list */
@ -226,8 +226,8 @@ int process_cli(struct session *t)
fprintf(stderr,"process_cli: c=%s s=%s set(r,w)=%d,%d exp(r,w)=%d.%d,%d.%d\n", fprintf(stderr,"process_cli: c=%s s=%s set(r,w)=%d,%d exp(r,w)=%d.%d,%d.%d\n",
cli_stnames[c], srv_stnames[s], cli_stnames[c], srv_stnames[s],
FD_ISSET(t->cli_fd, StaticReadEvent), FD_ISSET(t->cli_fd, StaticWriteEvent), FD_ISSET(t->cli_fd, StaticReadEvent), FD_ISSET(t->cli_fd, StaticWriteEvent),
t->crexpire.tv_sec, t->crexpire.tv_usec, req->rex.tv_sec, req->rex.tv_usec,
t->cwexpire.tv_sec, t->cwexpire.tv_usec); rep->wex.tv_sec, rep->wex.tv_usec);
#endif #endif
//fprintf(stderr,"process_cli: c=%d, s=%d, cr=%d, cw=%d, sr=%d, sw=%d\n", c, s, //fprintf(stderr,"process_cli: c=%d, s=%d, cr=%d, cw=%d, sr=%d, sw=%d\n", c, s,
//FD_ISSET(t->cli_fd, StaticReadEvent), FD_ISSET(t->cli_fd, StaticWriteEvent), //FD_ISSET(t->cli_fd, StaticReadEvent), FD_ISSET(t->cli_fd, StaticWriteEvent),
@ -429,7 +429,7 @@ int process_cli(struct session *t)
* CL_STDATA also has to take care of this, which is done below. * CL_STDATA also has to take care of this, which is done below.
*/ */
//FD_CLR(t->cli_fd, StaticReadEvent); //FD_CLR(t->cli_fd, StaticReadEvent);
//tv_eternity(&t->crexpire); //tv_eternity(&req->rex);
/* FIXME: if we break here (as up to 1.1.23), having the client /* FIXME: if we break here (as up to 1.1.23), having the client
* shutdown its connection can lead to an abort further. * shutdown its connection can lead to an abort further.
@ -449,7 +449,7 @@ int process_cli(struct session *t)
* when it switches its state, otherwise a client can stay connected * when it switches its state, otherwise a client can stay connected
* indefinitely. This now seems to be OK. * indefinitely. This now seems to be OK.
*/ */
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
goto process_data; goto process_data;
} }
@ -966,14 +966,14 @@ int process_cli(struct session *t)
if ((req->l < req->rlim - req->data) && ! FD_ISSET(t->cli_fd, StaticReadEvent)) { if ((req->l < req->rlim - req->data) && ! FD_ISSET(t->cli_fd, StaticReadEvent)) {
/* fd in StaticReadEvent was disabled, perhaps because of a previous buffer /* fd in StaticReadEvent was disabled, perhaps because of a previous buffer
* full. We cannot loop here since event_cli_read will disable it only if * full. We cannot loop here since stream_sock_read will disable it only if
* req->l == rlim-data * req->l == rlim-data
*/ */
FD_SET(t->cli_fd, StaticReadEvent); FD_SET(t->cli_fd, StaticReadEvent);
if (t->proxy->clitimeout) if (t->proxy->clitimeout)
tv_delayfrom(&t->crexpire, &now, t->proxy->clitimeout); tv_delayfrom(&req->rex, &now, t->proxy->clitimeout);
else else
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
} }
/* Since we are in header mode, if there's no space left for headers, we /* Since we are in header mode, if there's no space left for headers, we
@ -990,7 +990,7 @@ int process_cli(struct session *t)
} }
else if (req->flags & (BF_READ_ERROR | BF_READ_NULL)) { else if (req->flags & (BF_READ_ERROR | BF_READ_NULL)) {
/* read error, or last read : give up. */ /* read error, or last read : give up. */
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -999,7 +999,7 @@ int process_cli(struct session *t)
t->flags |= SN_FINST_R; t->flags |= SN_FINST_R;
return 1; return 1;
} }
else if (tv_cmp2_ms(&t->crexpire, &now) <= 0) { else if (tv_cmp2_ms(&req->rex, &now) <= 0) {
/* read timeout : give up with an error message. /* read timeout : give up with an error message.
*/ */
@ -1023,8 +1023,8 @@ int process_cli(struct session *t)
*/ */
/* read or write error */ /* read or write error */
if (rep->flags & BF_WRITE_ERROR || req->flags & BF_READ_ERROR) { if (rep->flags & BF_WRITE_ERROR || req->flags & BF_READ_ERROR) {
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -1042,7 +1042,7 @@ int process_cli(struct session *t)
/* last read, or end of server write */ /* last read, or end of server write */
else if (req->flags & BF_READ_NULL || s == SV_STSHUTW || s == SV_STCLOSE) { else if (req->flags & BF_READ_NULL || s == SV_STSHUTW || s == SV_STCLOSE) {
FD_CLR(t->cli_fd, StaticReadEvent); FD_CLR(t->cli_fd, StaticReadEvent);
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
shutdown(t->cli_fd, SHUT_RD); shutdown(t->cli_fd, SHUT_RD);
t->cli_state = CL_STSHUTR; t->cli_state = CL_STSHUTR;
return 1; return 1;
@ -1050,21 +1050,21 @@ int process_cli(struct session *t)
/* last server read and buffer empty */ /* last server read and buffer empty */
else if ((s == SV_STSHUTR || s == SV_STCLOSE) && (rep->l == 0)) { else if ((s == SV_STSHUTR || s == SV_STCLOSE) && (rep->l == 0)) {
FD_CLR(t->cli_fd, StaticWriteEvent); FD_CLR(t->cli_fd, StaticWriteEvent);
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
shutdown(t->cli_fd, SHUT_WR); shutdown(t->cli_fd, SHUT_WR);
/* We must ensure that the read part is still alive when switching /* We must ensure that the read part is still alive when switching
* to shutw */ * to shutw */
FD_SET(t->cli_fd, StaticReadEvent); FD_SET(t->cli_fd, StaticReadEvent);
if (t->proxy->clitimeout) if (t->proxy->clitimeout)
tv_delayfrom(&t->crexpire, &now, t->proxy->clitimeout); tv_delayfrom(&req->rex, &now, t->proxy->clitimeout);
t->cli_state = CL_STSHUTW; t->cli_state = CL_STSHUTW;
//fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state); //fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state);
return 1; return 1;
} }
/* read timeout */ /* read timeout */
else if (tv_cmp2_ms(&t->crexpire, &now) <= 0) { else if (tv_cmp2_ms(&req->rex, &now) <= 0) {
FD_CLR(t->cli_fd, StaticReadEvent); FD_CLR(t->cli_fd, StaticReadEvent);
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
shutdown(t->cli_fd, SHUT_RD); shutdown(t->cli_fd, SHUT_RD);
t->cli_state = CL_STSHUTR; t->cli_state = CL_STSHUTR;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -1080,15 +1080,15 @@ int process_cli(struct session *t)
return 1; return 1;
} }
/* write timeout */ /* write timeout */
else if (tv_cmp2_ms(&t->cwexpire, &now) <= 0) { else if (tv_cmp2_ms(&rep->wex, &now) <= 0) {
FD_CLR(t->cli_fd, StaticWriteEvent); FD_CLR(t->cli_fd, StaticWriteEvent);
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
shutdown(t->cli_fd, SHUT_WR); shutdown(t->cli_fd, SHUT_WR);
/* We must ensure that the read part is still alive when switching /* We must ensure that the read part is still alive when switching
* to shutw */ * to shutw */
FD_SET(t->cli_fd, StaticReadEvent); FD_SET(t->cli_fd, StaticReadEvent);
if (t->proxy->clitimeout) if (t->proxy->clitimeout)
tv_delayfrom(&t->crexpire, &now, t->proxy->clitimeout); tv_delayfrom(&req->rex, &now, t->proxy->clitimeout);
t->cli_state = CL_STSHUTW; t->cli_state = CL_STSHUTW;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -1109,7 +1109,7 @@ int process_cli(struct session *t)
if (FD_ISSET(t->cli_fd, StaticReadEvent)) { if (FD_ISSET(t->cli_fd, StaticReadEvent)) {
/* stop reading until we get some space */ /* stop reading until we get some space */
FD_CLR(t->cli_fd, StaticReadEvent); FD_CLR(t->cli_fd, StaticReadEvent);
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
} }
} else { } else {
/* there's still some space in the buffer */ /* there's still some space in the buffer */
@ -1122,9 +1122,9 @@ int process_cli(struct session *t)
* timeout on the client side so that too low values cannot make the * timeout on the client side so that too low values cannot make the
* sessions abort too early. * sessions abort too early.
*/ */
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
else else
tv_delayfrom(&t->crexpire, &now, t->proxy->clitimeout); tv_delayfrom(&req->rex, &now, t->proxy->clitimeout);
} }
} }
@ -1132,27 +1132,27 @@ int process_cli(struct session *t)
((s < SV_STDATA) /* FIXME: this may be optimized && (rep->w == rep->h)*/)) { ((s < SV_STDATA) /* FIXME: this may be optimized && (rep->w == rep->h)*/)) {
if (FD_ISSET(t->cli_fd, StaticWriteEvent)) { if (FD_ISSET(t->cli_fd, StaticWriteEvent)) {
FD_CLR(t->cli_fd, StaticWriteEvent); /* stop writing */ FD_CLR(t->cli_fd, StaticWriteEvent); /* stop writing */
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
} }
} else { } else {
/* buffer not empty */ /* buffer not empty */
if (! FD_ISSET(t->cli_fd, StaticWriteEvent)) { if (! FD_ISSET(t->cli_fd, StaticWriteEvent)) {
FD_SET(t->cli_fd, StaticWriteEvent); /* restart writing */ FD_SET(t->cli_fd, StaticWriteEvent); /* restart writing */
if (t->proxy->clitimeout) { if (t->proxy->clitimeout) {
tv_delayfrom(&t->cwexpire, &now, t->proxy->clitimeout); tv_delayfrom(&rep->wex, &now, t->proxy->clitimeout);
/* FIXME: to prevent the client from expiring read timeouts during writes, /* FIXME: to prevent the client from expiring read timeouts during writes,
* we refresh it. */ * we refresh it. */
t->crexpire = t->cwexpire; req->rex = rep->wex;
} }
else else
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
} }
} }
return 0; /* other cases change nothing */ return 0; /* other cases change nothing */
} }
else if (c == CL_STSHUTR) { else if (c == CL_STSHUTR) {
if (rep->flags & BF_WRITE_ERROR) { if (rep->flags & BF_WRITE_ERROR) {
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -1169,13 +1169,13 @@ int process_cli(struct session *t)
} }
else if ((s == SV_STSHUTR || s == SV_STCLOSE) && (rep->l == 0) else if ((s == SV_STSHUTR || s == SV_STCLOSE) && (rep->l == 0)
&& !(t->flags & SN_SELF_GEN)) { && !(t->flags & SN_SELF_GEN)) {
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
return 1; return 1;
} }
else if (tv_cmp2_ms(&t->cwexpire, &now) <= 0) { else if (tv_cmp2_ms(&rep->wex, &now) <= 0) {
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -1194,7 +1194,7 @@ int process_cli(struct session *t)
if (t->flags & SN_SELF_GEN) { if (t->flags & SN_SELF_GEN) {
produce_content(t); produce_content(t);
if (rep->l == 0) { if (rep->l == 0) {
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
return 1; return 1;
@ -1205,27 +1205,27 @@ int process_cli(struct session *t)
|| ((s == SV_STHEADERS) /* FIXME: this may be optimized && (rep->w == rep->h)*/)) { || ((s == SV_STHEADERS) /* FIXME: this may be optimized && (rep->w == rep->h)*/)) {
if (FD_ISSET(t->cli_fd, StaticWriteEvent)) { if (FD_ISSET(t->cli_fd, StaticWriteEvent)) {
FD_CLR(t->cli_fd, StaticWriteEvent); /* stop writing */ FD_CLR(t->cli_fd, StaticWriteEvent); /* stop writing */
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
} }
} else { } else {
/* buffer not empty */ /* buffer not empty */
if (! FD_ISSET(t->cli_fd, StaticWriteEvent)) { if (! FD_ISSET(t->cli_fd, StaticWriteEvent)) {
FD_SET(t->cli_fd, StaticWriteEvent); /* restart writing */ FD_SET(t->cli_fd, StaticWriteEvent); /* restart writing */
if (t->proxy->clitimeout) { if (t->proxy->clitimeout) {
tv_delayfrom(&t->cwexpire, &now, t->proxy->clitimeout); tv_delayfrom(&rep->wex, &now, t->proxy->clitimeout);
/* FIXME: to prevent the client from expiring read timeouts during writes, /* FIXME: to prevent the client from expiring read timeouts during writes,
* we refresh it. */ * we refresh it. */
t->crexpire = t->cwexpire; req->rex = rep->wex;
} }
else else
tv_eternity(&t->cwexpire); tv_eternity(&rep->wex);
} }
} }
return 0; return 0;
} }
else if (c == CL_STSHUTW) { else if (c == CL_STSHUTW) {
if (req->flags & BF_READ_ERROR) { if (req->flags & BF_READ_ERROR) {
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -1241,13 +1241,13 @@ int process_cli(struct session *t)
return 1; return 1;
} }
else if (req->flags & BF_READ_NULL || s == SV_STSHUTW || s == SV_STCLOSE) { else if (req->flags & BF_READ_NULL || s == SV_STSHUTW || s == SV_STCLOSE) {
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
return 1; return 1;
} }
else if (tv_cmp2_ms(&t->crexpire, &now) <= 0) { else if (tv_cmp2_ms(&req->rex, &now) <= 0) {
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
fd_delete(t->cli_fd); fd_delete(t->cli_fd);
t->cli_state = CL_STCLOSE; t->cli_state = CL_STCLOSE;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -1272,7 +1272,7 @@ int process_cli(struct session *t)
if (FD_ISSET(t->cli_fd, StaticReadEvent)) { if (FD_ISSET(t->cli_fd, StaticReadEvent)) {
/* stop reading until we get some space */ /* stop reading until we get some space */
FD_CLR(t->cli_fd, StaticReadEvent); FD_CLR(t->cli_fd, StaticReadEvent);
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
//fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state); //fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state);
} }
} else { } else {
@ -1280,9 +1280,9 @@ int process_cli(struct session *t)
if (! FD_ISSET(t->cli_fd, StaticReadEvent)) { if (! FD_ISSET(t->cli_fd, StaticReadEvent)) {
FD_SET(t->cli_fd, StaticReadEvent); FD_SET(t->cli_fd, StaticReadEvent);
if (t->proxy->clitimeout) if (t->proxy->clitimeout)
tv_delayfrom(&t->crexpire, &now, t->proxy->clitimeout); tv_delayfrom(&req->rex, &now, t->proxy->clitimeout);
else else
tv_eternity(&t->crexpire); tv_eternity(&req->rex);
//fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state); //fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state);
} }
} }
@ -1327,7 +1327,7 @@ int process_srv(struct session *t)
else if (c == CL_STCLOSE || c == CL_STSHUTW || else if (c == CL_STCLOSE || c == CL_STSHUTW ||
(c == CL_STSHUTR && (c == CL_STSHUTR &&
(t->req->l == 0 || t->proxy->options & PR_O_ABRT_CLOSE))) { /* give up */ (t->req->l == 0 || t->proxy->options & PR_O_ABRT_CLOSE))) { /* give up */
tv_eternity(&t->cnexpire); tv_eternity(&req->cex);
if (t->pend_pos) if (t->pend_pos)
t->logs.t_queue = tv_diff(&t->logs.tv_accept, &now); t->logs.t_queue = tv_diff(&t->logs.tv_accept, &now);
/* note that this must not return any error because it would be able to /* note that this must not return any error because it would be able to
@ -1344,11 +1344,11 @@ int process_srv(struct session *t)
* to any other session to release it and wake us up again. * to any other session to release it and wake us up again.
*/ */
if (t->pend_pos) { if (t->pend_pos) {
if (tv_cmp2_ms(&t->cnexpire, &now) > 0) if (tv_cmp2_ms(&req->cex, &now) > 0)
return 0; return 0;
else { else {
/* we've been waiting too long here */ /* we've been waiting too long here */
tv_eternity(&t->cnexpire); tv_eternity(&req->cex);
t->logs.t_queue = tv_diff(&t->logs.tv_accept, &now); t->logs.t_queue = tv_diff(&t->logs.tv_accept, &now);
srv_close_with_err(t, SN_ERR_SRVTO, SN_FINST_Q, srv_close_with_err(t, SN_ERR_SRVTO, SN_FINST_Q,
503, t->proxy->errmsg.len503, t->proxy->errmsg.msg503); 503, t->proxy->errmsg.len503, t->proxy->errmsg.msg503);
@ -1379,7 +1379,7 @@ int process_srv(struct session *t)
if (c == CL_STCLOSE || c == CL_STSHUTW || if (c == CL_STCLOSE || c == CL_STSHUTW ||
(c == CL_STSHUTR && (c == CL_STSHUTR &&
(t->req->l == 0 || t->proxy->options & PR_O_ABRT_CLOSE))) { /* give up */ (t->req->l == 0 || t->proxy->options & PR_O_ABRT_CLOSE))) { /* give up */
tv_eternity(&t->cnexpire); tv_eternity(&req->cex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) if (t->srv)
t->srv->cur_sess--; t->srv->cur_sess--;
@ -1390,8 +1390,8 @@ int process_srv(struct session *t)
srv_close_with_err(t, SN_ERR_CLICL, SN_FINST_C, 0, 0, NULL); srv_close_with_err(t, SN_ERR_CLICL, SN_FINST_C, 0, 0, NULL);
return 1; return 1;
} }
if (!(req->flags & BF_WRITE_STATUS) && tv_cmp2_ms(&t->cnexpire, &now) > 0) { if (!(req->flags & BF_WRITE_STATUS) && tv_cmp2_ms(&req->cex, &now) > 0) {
//fprintf(stderr,"1: c=%d, s=%d, now=%d.%06d, exp=%d.%06d\n", c, s, now.tv_sec, now.tv_usec, t->cnexpire.tv_sec, t->cnexpire.tv_usec); //fprintf(stderr,"1: c=%d, s=%d, now=%d.%06d, exp=%d.%06d\n", c, s, now.tv_sec, now.tv_usec, req->cex.tv_sec, req->cex.tv_usec);
return 0; /* nothing changed */ return 0; /* nothing changed */
} }
else if (!(req->flags & BF_WRITE_STATUS) || (req->flags & BF_WRITE_ERROR)) { else if (!(req->flags & BF_WRITE_STATUS) || (req->flags & BF_WRITE_ERROR)) {
@ -1433,25 +1433,25 @@ int process_srv(struct session *t)
//fprintf(stderr,"3: c=%d, s=%d\n", c, s); //fprintf(stderr,"3: c=%d, s=%d\n", c, s);
if (req->l == 0) /* nothing to write */ { if (req->l == 0) /* nothing to write */ {
FD_CLR(t->srv_fd, StaticWriteEvent); FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
} else /* need the right to write */ { } else /* need the right to write */ {
FD_SET(t->srv_fd, StaticWriteEvent); FD_SET(t->srv_fd, StaticWriteEvent);
if (t->proxy->srvtimeout) { if (t->proxy->srvtimeout) {
tv_delayfrom(&t->swexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&req->wex, &now, t->proxy->srvtimeout);
/* FIXME: to prevent the server from expiring read timeouts during writes, /* FIXME: to prevent the server from expiring read timeouts during writes,
* we refresh it. */ * we refresh it. */
t->srexpire = t->swexpire; rep->rex = req->wex;
} }
else else
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
} }
if (t->proxy->mode == PR_MODE_TCP) { /* let's allow immediate data connection in this case */ if (t->proxy->mode == PR_MODE_TCP) { /* let's allow immediate data connection in this case */
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
else else
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
t->srv_state = SV_STDATA; t->srv_state = SV_STDATA;
if (t->srv) if (t->srv)
@ -1471,7 +1471,7 @@ int process_srv(struct session *t)
t->srv->cum_sess++; t->srv->cum_sess++;
rep->rlim = rep->data + BUFSIZE - MAXREWRITE; /* rewrite needed */ rep->rlim = rep->data + BUFSIZE - MAXREWRITE; /* rewrite needed */
} }
tv_eternity(&t->cnexpire); tv_eternity(&req->cex);
return 1; return 1;
} }
} }
@ -1502,8 +1502,8 @@ int process_srv(struct session *t)
* a set-cookie header. We'll block it as requested by * a set-cookie header. We'll block it as requested by
* the 'checkcache' option, and send an alert. * the 'checkcache' option, and send an alert.
*/ */
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) { if (t->srv) {
t->srv->cur_sess--; t->srv->cur_sess--;
@ -1533,8 +1533,8 @@ int process_srv(struct session *t)
/* next, we'll block if an 'rspideny' or 'rspdeny' filter matched */ /* next, we'll block if an 'rspideny' or 'rspdeny' filter matched */
if (t->flags & SN_SVDENY) { if (t->flags & SN_SVDENY) {
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) { if (t->srv) {
t->srv->cur_sess--; t->srv->cur_sess--;
@ -1607,13 +1607,13 @@ int process_srv(struct session *t)
if ((req->l == 0) && if ((req->l == 0) &&
(c == CL_STSHUTR || c == CL_STCLOSE || t->proxy->options & PR_O_FORCE_CLO)) { (c == CL_STSHUTR || c == CL_STCLOSE || t->proxy->options & PR_O_FORCE_CLO)) {
FD_CLR(t->srv_fd, StaticWriteEvent); FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
/* We must ensure that the read part is still alive when switching /* We must ensure that the read part is still alive when switching
* to shutw */ * to shutw */
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
shutdown(t->srv_fd, SHUT_WR); shutdown(t->srv_fd, SHUT_WR);
t->srv_state = SV_STSHUTW; t->srv_state = SV_STSHUTW;
@ -1964,20 +1964,20 @@ int process_srv(struct session *t)
if ((rep->l < rep->rlim - rep->data) && ! FD_ISSET(t->srv_fd, StaticReadEvent)) { if ((rep->l < rep->rlim - rep->data) && ! FD_ISSET(t->srv_fd, StaticReadEvent)) {
/* fd in StaticReadEvent was disabled, perhaps because of a previous buffer /* fd in StaticReadEvent was disabled, perhaps because of a previous buffer
* full. We cannot loop here since event_srv_read will disable it only if * full. We cannot loop here since stream_sock_read will disable it only if
* rep->l == rlim-data * rep->l == rlim-data
*/ */
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
else else
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
} }
/* read error, write error */ /* read error, write error */
if (req->flags & BF_WRITE_ERROR || rep->flags & BF_READ_ERROR) { if (req->flags & BF_WRITE_ERROR || rep->flags & BF_READ_ERROR) {
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) { if (t->srv) {
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2006,7 +2006,7 @@ int process_srv(struct session *t)
*/ */
else if (rep->flags & BF_READ_NULL || c == CL_STSHUTW || c == CL_STCLOSE || rep->l >= rep->rlim - rep->data) { else if (rep->flags & BF_READ_NULL || c == CL_STSHUTW || c == CL_STCLOSE || rep->l >= rep->rlim - rep->data) {
FD_CLR(t->srv_fd, StaticReadEvent); FD_CLR(t->srv_fd, StaticReadEvent);
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
shutdown(t->srv_fd, SHUT_RD); shutdown(t->srv_fd, SHUT_RD);
t->srv_state = SV_STSHUTR; t->srv_state = SV_STSHUTR;
//fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state); //fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state);
@ -2014,9 +2014,9 @@ int process_srv(struct session *t)
} }
/* read timeout : return a 504 to the client. /* read timeout : return a 504 to the client.
*/ */
else if (FD_ISSET(t->srv_fd, StaticReadEvent) && tv_cmp2_ms(&t->srexpire, &now) <= 0) { else if (FD_ISSET(t->srv_fd, StaticReadEvent) && tv_cmp2_ms(&rep->rex, &now) <= 0) {
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) { if (t->srv) {
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2048,13 +2048,13 @@ int process_srv(struct session *t)
*/ */
else if ((/*c == CL_STSHUTR ||*/ c == CL_STCLOSE) && (req->l == 0)) { else if ((/*c == CL_STSHUTR ||*/ c == CL_STCLOSE) && (req->l == 0)) {
FD_CLR(t->srv_fd, StaticWriteEvent); FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
/* We must ensure that the read part is still alive when switching /* We must ensure that the read part is still alive when switching
* to shutw */ * to shutw */
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
shutdown(t->srv_fd, SHUT_WR); shutdown(t->srv_fd, SHUT_WR);
t->srv_state = SV_STSHUTW; t->srv_state = SV_STSHUTW;
@ -2065,21 +2065,21 @@ int process_srv(struct session *t)
* client shuts read too early, because we may still have * client shuts read too early, because we may still have
* some work to do on the headers. * some work to do on the headers.
*/ */
else if (FD_ISSET(t->srv_fd, StaticWriteEvent) && tv_cmp2_ms(&t->swexpire, &now) <= 0) { else if (FD_ISSET(t->srv_fd, StaticWriteEvent) && tv_cmp2_ms(&req->wex, &now) <= 0) {
FD_CLR(t->srv_fd, StaticWriteEvent); FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
shutdown(t->srv_fd, SHUT_WR); shutdown(t->srv_fd, SHUT_WR);
/* We must ensure that the read part is still alive when switching /* We must ensure that the read part is still alive when switching
* to shutw */ * to shutw */
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
/* We must ensure that the read part is still alive when switching /* We must ensure that the read part is still alive when switching
* to shutw */ * to shutw */
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
t->srv_state = SV_STSHUTW; t->srv_state = SV_STSHUTW;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -2092,20 +2092,20 @@ int process_srv(struct session *t)
if (req->l == 0) { if (req->l == 0) {
if (FD_ISSET(t->srv_fd, StaticWriteEvent)) { if (FD_ISSET(t->srv_fd, StaticWriteEvent)) {
FD_CLR(t->srv_fd, StaticWriteEvent); /* stop writing */ FD_CLR(t->srv_fd, StaticWriteEvent); /* stop writing */
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
} }
} }
else { /* client buffer not empty */ else { /* client buffer not empty */
if (! FD_ISSET(t->srv_fd, StaticWriteEvent)) { if (! FD_ISSET(t->srv_fd, StaticWriteEvent)) {
FD_SET(t->srv_fd, StaticWriteEvent); /* restart writing */ FD_SET(t->srv_fd, StaticWriteEvent); /* restart writing */
if (t->proxy->srvtimeout) { if (t->proxy->srvtimeout) {
tv_delayfrom(&t->swexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&req->wex, &now, t->proxy->srvtimeout);
/* FIXME: to prevent the server from expiring read timeouts during writes, /* FIXME: to prevent the server from expiring read timeouts during writes,
* we refresh it. */ * we refresh it. */
t->srexpire = t->swexpire; rep->rex = req->wex;
} }
else else
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
} }
} }
@ -2120,8 +2120,8 @@ int process_srv(struct session *t)
else if (s == SV_STDATA) { else if (s == SV_STDATA) {
/* read or write error */ /* read or write error */
if (req->flags & BF_WRITE_ERROR || rep->flags & BF_READ_ERROR) { if (req->flags & BF_WRITE_ERROR || rep->flags & BF_READ_ERROR) {
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) { if (t->srv) {
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2144,7 +2144,7 @@ int process_srv(struct session *t)
/* last read, or end of client write */ /* last read, or end of client write */
else if (rep->flags & BF_READ_NULL || c == CL_STSHUTW || c == CL_STCLOSE) { else if (rep->flags & BF_READ_NULL || c == CL_STSHUTW || c == CL_STCLOSE) {
FD_CLR(t->srv_fd, StaticReadEvent); FD_CLR(t->srv_fd, StaticReadEvent);
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
shutdown(t->srv_fd, SHUT_RD); shutdown(t->srv_fd, SHUT_RD);
t->srv_state = SV_STSHUTR; t->srv_state = SV_STSHUTR;
//fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state); //fprintf(stderr,"%p:%s(%d), c=%d, s=%d\n", t, __FUNCTION__, __LINE__, t->cli_state, t->cli_state);
@ -2153,21 +2153,21 @@ int process_srv(struct session *t)
/* end of client read and no more data to send */ /* end of client read and no more data to send */
else if ((c == CL_STSHUTR || c == CL_STCLOSE) && (req->l == 0)) { else if ((c == CL_STSHUTR || c == CL_STCLOSE) && (req->l == 0)) {
FD_CLR(t->srv_fd, StaticWriteEvent); FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
shutdown(t->srv_fd, SHUT_WR); shutdown(t->srv_fd, SHUT_WR);
/* We must ensure that the read part is still alive when switching /* We must ensure that the read part is still alive when switching
* to shutw */ * to shutw */
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
t->srv_state = SV_STSHUTW; t->srv_state = SV_STSHUTW;
return 1; return 1;
} }
/* read timeout */ /* read timeout */
else if (tv_cmp2_ms(&t->srexpire, &now) <= 0) { else if (tv_cmp2_ms(&rep->rex, &now) <= 0) {
FD_CLR(t->srv_fd, StaticReadEvent); FD_CLR(t->srv_fd, StaticReadEvent);
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
shutdown(t->srv_fd, SHUT_RD); shutdown(t->srv_fd, SHUT_RD);
t->srv_state = SV_STSHUTR; t->srv_state = SV_STSHUTR;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
@ -2177,15 +2177,15 @@ int process_srv(struct session *t)
return 1; return 1;
} }
/* write timeout */ /* write timeout */
else if (tv_cmp2_ms(&t->swexpire, &now) <= 0) { else if (tv_cmp2_ms(&req->wex, &now) <= 0) {
FD_CLR(t->srv_fd, StaticWriteEvent); FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
shutdown(t->srv_fd, SHUT_WR); shutdown(t->srv_fd, SHUT_WR);
/* We must ensure that the read part is still alive when switching /* We must ensure that the read part is still alive when switching
* to shutw */ * to shutw */
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
t->srv_state = SV_STSHUTW; t->srv_state = SV_STSHUTW;
if (!(t->flags & SN_ERR_MASK)) if (!(t->flags & SN_ERR_MASK))
t->flags |= SN_ERR_SRVTO; t->flags |= SN_ERR_SRVTO;
@ -2198,20 +2198,20 @@ int process_srv(struct session *t)
if (req->l == 0) { if (req->l == 0) {
if (FD_ISSET(t->srv_fd, StaticWriteEvent)) { if (FD_ISSET(t->srv_fd, StaticWriteEvent)) {
FD_CLR(t->srv_fd, StaticWriteEvent); /* stop writing */ FD_CLR(t->srv_fd, StaticWriteEvent); /* stop writing */
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
} }
} }
else { /* buffer not empty, there are still data to be transferred */ else { /* buffer not empty, there are still data to be transferred */
if (! FD_ISSET(t->srv_fd, StaticWriteEvent)) { if (! FD_ISSET(t->srv_fd, StaticWriteEvent)) {
FD_SET(t->srv_fd, StaticWriteEvent); /* restart writing */ FD_SET(t->srv_fd, StaticWriteEvent); /* restart writing */
if (t->proxy->srvtimeout) { if (t->proxy->srvtimeout) {
tv_delayfrom(&t->swexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&req->wex, &now, t->proxy->srvtimeout);
/* FIXME: to prevent the server from expiring read timeouts during writes, /* FIXME: to prevent the server from expiring read timeouts during writes,
* we refresh it. */ * we refresh it. */
t->srexpire = t->swexpire; rep->rex = req->wex;
} }
else else
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
} }
} }
@ -2219,16 +2219,16 @@ int process_srv(struct session *t)
if (rep->l == BUFSIZE) { /* no room to read more data */ if (rep->l == BUFSIZE) { /* no room to read more data */
if (FD_ISSET(t->srv_fd, StaticReadEvent)) { if (FD_ISSET(t->srv_fd, StaticReadEvent)) {
FD_CLR(t->srv_fd, StaticReadEvent); FD_CLR(t->srv_fd, StaticReadEvent);
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
} }
} }
else { else {
if (! FD_ISSET(t->srv_fd, StaticReadEvent)) { if (! FD_ISSET(t->srv_fd, StaticReadEvent)) {
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
else else
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
} }
} }
@ -2237,7 +2237,7 @@ int process_srv(struct session *t)
else if (s == SV_STSHUTR) { else if (s == SV_STSHUTR) {
if (req->flags & BF_WRITE_ERROR) { if (req->flags & BF_WRITE_ERROR) {
//FD_CLR(t->srv_fd, StaticWriteEvent); //FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) { if (t->srv) {
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2260,7 +2260,7 @@ int process_srv(struct session *t)
} }
else if ((c == CL_STSHUTR || c == CL_STCLOSE) && (req->l == 0)) { else if ((c == CL_STSHUTR || c == CL_STCLOSE) && (req->l == 0)) {
//FD_CLR(t->srv_fd, StaticWriteEvent); //FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) if (t->srv)
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2274,9 +2274,9 @@ int process_srv(struct session *t)
return 1; return 1;
} }
else if (tv_cmp2_ms(&t->swexpire, &now) <= 0) { else if (tv_cmp2_ms(&req->wex, &now) <= 0) {
//FD_CLR(t->srv_fd, StaticWriteEvent); //FD_CLR(t->srv_fd, StaticWriteEvent);
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) if (t->srv)
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2297,20 +2297,20 @@ int process_srv(struct session *t)
else if (req->l == 0) { else if (req->l == 0) {
if (FD_ISSET(t->srv_fd, StaticWriteEvent)) { if (FD_ISSET(t->srv_fd, StaticWriteEvent)) {
FD_CLR(t->srv_fd, StaticWriteEvent); /* stop writing */ FD_CLR(t->srv_fd, StaticWriteEvent); /* stop writing */
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
} }
} }
else { /* buffer not empty */ else { /* buffer not empty */
if (! FD_ISSET(t->srv_fd, StaticWriteEvent)) { if (! FD_ISSET(t->srv_fd, StaticWriteEvent)) {
FD_SET(t->srv_fd, StaticWriteEvent); /* restart writing */ FD_SET(t->srv_fd, StaticWriteEvent); /* restart writing */
if (t->proxy->srvtimeout) { if (t->proxy->srvtimeout) {
tv_delayfrom(&t->swexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&req->wex, &now, t->proxy->srvtimeout);
/* FIXME: to prevent the server from expiring read timeouts during writes, /* FIXME: to prevent the server from expiring read timeouts during writes,
* we refresh it. */ * we refresh it. */
t->srexpire = t->swexpire; rep->rex = req->wex;
} }
else else
tv_eternity(&t->swexpire); tv_eternity(&req->wex);
} }
} }
return 0; return 0;
@ -2318,7 +2318,7 @@ int process_srv(struct session *t)
else if (s == SV_STSHUTW) { else if (s == SV_STSHUTW) {
if (rep->flags & BF_READ_ERROR) { if (rep->flags & BF_READ_ERROR) {
//FD_CLR(t->srv_fd, StaticReadEvent); //FD_CLR(t->srv_fd, StaticReadEvent);
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) { if (t->srv) {
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2341,7 +2341,7 @@ int process_srv(struct session *t)
} }
else if (rep->flags & BF_READ_NULL || c == CL_STSHUTW || c == CL_STCLOSE) { else if (rep->flags & BF_READ_NULL || c == CL_STSHUTW || c == CL_STCLOSE) {
//FD_CLR(t->srv_fd, StaticReadEvent); //FD_CLR(t->srv_fd, StaticReadEvent);
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) if (t->srv)
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2355,9 +2355,9 @@ int process_srv(struct session *t)
return 1; return 1;
} }
else if (tv_cmp2_ms(&t->srexpire, &now) <= 0) { else if (tv_cmp2_ms(&rep->rex, &now) <= 0) {
//FD_CLR(t->srv_fd, StaticReadEvent); //FD_CLR(t->srv_fd, StaticReadEvent);
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
fd_delete(t->srv_fd); fd_delete(t->srv_fd);
if (t->srv) if (t->srv)
t->srv->cur_sess--; t->srv->cur_sess--;
@ -2378,16 +2378,16 @@ int process_srv(struct session *t)
else if (rep->l == BUFSIZE) { /* no room to read more data */ else if (rep->l == BUFSIZE) { /* no room to read more data */
if (FD_ISSET(t->srv_fd, StaticReadEvent)) { if (FD_ISSET(t->srv_fd, StaticReadEvent)) {
FD_CLR(t->srv_fd, StaticReadEvent); FD_CLR(t->srv_fd, StaticReadEvent);
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
} }
} }
else { else {
if (! FD_ISSET(t->srv_fd, StaticReadEvent)) { if (! FD_ISSET(t->srv_fd, StaticReadEvent)) {
FD_SET(t->srv_fd, StaticReadEvent); FD_SET(t->srv_fd, StaticReadEvent);
if (t->proxy->srvtimeout) if (t->proxy->srvtimeout)
tv_delayfrom(&t->srexpire, &now, t->proxy->srvtimeout); tv_delayfrom(&rep->rex, &now, t->proxy->srvtimeout);
else else
tv_eternity(&t->srexpire); tv_eternity(&rep->rex);
} }
} }
return 0; return 0;

153
src/stream_sock.c
View File

@ -41,17 +41,15 @@
/* /*
* this function is called on a read event from a client socket. * this function is called on a read event from a stream socket.
* It returns 0. * It returns 0.
*/ */
int event_cli_read(int fd) { int stream_sock_read(int fd) {
struct task *t = fdtab[fd].owner;
struct buffer *b = fdtab[fd].cb[DIR_RD].b; struct buffer *b = fdtab[fd].cb[DIR_RD].b;
struct session *s = t->context;
int ret, max; int ret, max;
#ifdef DEBUG_FULL #ifdef DEBUG_FULL
fprintf(stderr,"event_cli_read : fd=%d, s=%p\n", fd, s); fprintf(stderr,"stream_sock_read : fd=%d, owner=%p\n", fd, fdtab[fd].owner);
#endif #endif
if (fdtab[fd].state != FD_STERROR) { if (fdtab[fd].state != FD_STERROR) {
@ -133,12 +131,12 @@ int event_cli_read(int fd) {
} }
if (b->flags & BF_READ_STATUS) { if (b->flags & BF_READ_STATUS) {
if (s->proxy->clitimeout && FD_ISSET(fd, StaticReadEvent)) if (b->rto && FD_ISSET(fd, StaticReadEvent))
tv_delayfrom(&s->crexpire, &now, s->proxy->clitimeout); tv_delayfrom(&b->rex, &now, b->rto);
else else
tv_eternity(&s->crexpire); tv_eternity(&b->rex);
task_wakeup(&rq, t); task_wakeup(&rq, fdtab[fd].owner);
} }
return 0; return 0;
@ -152,11 +150,10 @@ int event_cli_read(int fd) {
int event_cli_write(int fd) { int event_cli_write(int fd) {
struct task *t = fdtab[fd].owner; struct task *t = fdtab[fd].owner;
struct buffer *b = fdtab[fd].cb[DIR_WR].b; struct buffer *b = fdtab[fd].cb[DIR_WR].b;
struct session *s = t->context;
int ret, max; int ret, max;
#ifdef DEBUG_FULL #ifdef DEBUG_FULL
fprintf(stderr,"event_cli_write : fd=%d, s=%p\n", fd, s); fprintf(stderr,"event_cli_write : fd=%d, owner=%p\n", fd, fdtab[fd].owner);
#endif #endif
if (b->l == 0) { /* let's realign the buffer to optimize I/O */ if (b->l == 0) { /* let's realign the buffer to optimize I/O */
@ -174,7 +171,7 @@ int event_cli_write(int fd) {
if (max == 0) { if (max == 0) {
b->flags |= BF_WRITE_NULL; b->flags |= BF_WRITE_NULL;
task_wakeup(&rq, t); task_wakeup(&rq, t);
tv_eternity(&s->cwexpire); tv_eternity(&b->wex);
FD_CLR(fd, StaticWriteEvent); FD_CLR(fd, StaticWriteEvent);
return 0; return 0;
} }
@ -221,125 +218,22 @@ int event_cli_write(int fd) {
fdtab[fd].state = FD_STERROR; fdtab[fd].state = FD_STERROR;
} }
if (s->proxy->clitimeout) { if (b->wto) {
tv_delayfrom(&s->cwexpire, &now, s->proxy->clitimeout); tv_delayfrom(&b->wex, &now, b->wto);
/* FIXME: to prevent the client from expiring read timeouts during writes, /* FIXME: to prevent the client from expiring read timeouts during writes,
* we refresh it. A solution would be to merge read+write timeouts into a * we refresh it. A solution would be to merge read+write timeouts into a
* unique one, although that needs some study particularly on full-duplex * unique one, although that needs some study particularly on full-duplex
* TCP connections. */ * TCP connections. */
s->crexpire = s->cwexpire; b->rex = b->wex;
} }
else else
tv_eternity(&s->cwexpire); tv_eternity(&b->wex);
task_wakeup(&rq, t); task_wakeup(&rq, t);
return 0; return 0;
} }
/*
* this function is called on a read event from a server socket.
* It returns 0.
*/
int event_srv_read(int fd) {
struct task *t = fdtab[fd].owner;
struct buffer *b = fdtab[fd].cb[DIR_RD].b;
struct session *s = t->context;
int ret, max;
#ifdef DEBUG_FULL
fprintf(stderr,"event_srv_read : fd=%d, s=%p\n", fd, s);
#endif
if (fdtab[fd].state != FD_STERROR) {
#ifdef FILL_BUFFERS
while (1)
#else
do
#endif
{
if (b->l == 0) { /* let's realign the buffer to optimize I/O */
b->r = b->w = b->h = b->lr = b->data;
max = b->rlim - b->data;
}
else if (b->r > b->w) {
max = b->rlim - b->r;
}
else {
max = b->w - b->r;
/* FIXME: theorically, if w>0, we shouldn't have rlim < data+size anymore
* since it means that the rewrite protection has been removed. This
* implies that the if statement can be removed.
*/
if (max > b->rlim - b->data)
max = b->rlim - b->data;
}
if (max == 0) { /* not anymore room to store data */
FD_CLR(fd, StaticReadEvent);
break;
}
#ifndef MSG_NOSIGNAL
{
int skerr;
socklen_t lskerr = sizeof(skerr);
getsockopt(fd, SOL_SOCKET, SO_ERROR, &skerr, &lskerr);
if (skerr)
ret = -1;
else
ret = recv(fd, b->r, max, 0);
}
#else
ret = recv(fd, b->r, max, MSG_NOSIGNAL);
#endif
if (ret > 0) {
b->r += ret;
b->l += ret;
b->flags |= BF_PARTIAL_READ;
if (b->r == b->data + BUFSIZE) {
b->r = b->data; /* wrap around the buffer */
}
b->total += ret;
/* we hope to read more data or to get a close on next round */
continue;
}
else if (ret == 0) {
b->flags |= BF_READ_NULL;
break;
}
else if (errno == EAGAIN) {/* ignore EAGAIN */
break;
}
else {
b->flags |= BF_READ_ERROR;
fdtab[fd].state = FD_STERROR;
break;
}
} /* while(1) */
#ifndef FILL_BUFFERS
while (0);
#endif
}
else {
b->flags |= BF_READ_ERROR;
fdtab[fd].state = FD_STERROR;
}
if (b->flags & BF_READ_STATUS) {
if (s->proxy->srvtimeout && FD_ISSET(fd, StaticReadEvent))
tv_delayfrom(&s->srexpire, &now, s->proxy->srvtimeout);
else
tv_eternity(&s->srexpire);
task_wakeup(&rq, t);
}
return 0;
}
/* /*
@ -378,7 +272,7 @@ int event_srv_write(int fd) {
b->flags |= BF_WRITE_ERROR; b->flags |= BF_WRITE_ERROR;
fdtab[fd].state = FD_STERROR; fdtab[fd].state = FD_STERROR;
task_wakeup(&rq, t); task_wakeup(&rq, t);
tv_eternity(&s->swexpire); tv_eternity(&b->wex);
FD_CLR(fd, StaticWriteEvent); FD_CLR(fd, StaticWriteEvent);
return 0; return 0;
} }
@ -387,7 +281,7 @@ int event_srv_write(int fd) {
b->flags |= BF_WRITE_NULL; b->flags |= BF_WRITE_NULL;
task_wakeup(&rq, t); task_wakeup(&rq, t);
fdtab[fd].state = FD_STREADY; fdtab[fd].state = FD_STREADY;
tv_eternity(&s->swexpire); tv_eternity(&b->wex);
FD_CLR(fd, StaticWriteEvent); FD_CLR(fd, StaticWriteEvent);
return 0; return 0;
} }
@ -437,17 +331,16 @@ int event_srv_write(int fd) {
* otherwise it could loop indefinitely ! * otherwise it could loop indefinitely !
*/ */
if (s->srv_state != SV_STCONN) { if (s->srv_state != SV_STCONN) {
if (s->proxy->srvtimeout) { if (b->wto) {
tv_delayfrom(&s->swexpire, &now, s->proxy->srvtimeout); tv_delayfrom(&b->wex, &now, b->wto);
/* FIXME: to prevent the server from expiring read timeouts during writes, /* FIXME: to prevent the client from expiring read timeouts during writes,
* we refresh it. A solution would be to merge read+write+connect timeouts * we refresh it. A solution would be to merge read+write timeouts into a
* into a unique one since we don't mind expiring on read or write, and none * unique one, although that needs some study particularly on full-duplex
* of them is enabled while waiting for connect(), although that needs some * TCP connections. */
* study particularly on full-duplex TCP connections. */ b->rex = b->wex;
s->srexpire = s->swexpire;
} }
else else
tv_eternity(&s->swexpire); tv_eternity(&b->wex);
} }
task_wakeup(&rq, t); task_wakeup(&rq, t);