MEDIUM: stream: No longer release backend conn-stream on connection retry

The backend conn-stream is no longer released on connection retry. This
means the conn-stream is detached from the underlying connection but not
released. Thus, during connection retries, the stream has always an
allocated conn-stream with no connection. All previous changes were made to
make this possible.

Note that .attach() mux callback function was changed to get the conn-stream
as argument. The muxes are no longer responsible to create the conn-stream
when a server connection is attached to a stream.
This commit is contained in:
Christopher Faulet 2021-12-16 14:44:31 +01:00
parent e39827de0d
commit e00ad358c9
9 changed files with 26 additions and 56 deletions

View File

@ -421,7 +421,7 @@ struct mux_ops {
void (*shutr)(struct conn_stream *cs, enum cs_shr_mode); /* shutr function */ void (*shutr)(struct conn_stream *cs, enum cs_shr_mode); /* shutr function */
void (*shutw)(struct conn_stream *cs, enum cs_shw_mode); /* shutw function */ void (*shutw)(struct conn_stream *cs, enum cs_shw_mode); /* shutw function */
struct conn_stream *(*attach)(struct connection *, struct session *sess); /* Create and attach a conn_stream to an outgoing connection */ int (*attach)(struct connection *conn, struct conn_stream *, struct session *sess); /* attach a conn_stream to an outgoing connection */
const struct conn_stream *(*get_first_cs)(const struct connection *); /* retrieves any valid conn_stream from this connection */ const struct conn_stream *(*get_first_cs)(const struct connection *); /* retrieves any valid conn_stream from this connection */
void (*detach)(struct conn_stream *); /* Detach a conn_stream from an outgoing connection, when the request is done */ void (*detach)(struct conn_stream *); /* Detach a conn_stream from an outgoing connection, when the request is done */
int (*show_fd)(struct buffer *, struct connection *); /* append some data about connection into chunk for "show fd"; returns non-zero if suspicious */ int (*show_fd)(struct buffer *, struct connection *); /* append some data about connection into chunk for "show fd"; returns non-zero if suspicious */

View File

@ -1496,9 +1496,13 @@ int connect_server(struct stream *s)
} }
if (avail >= 1) { if (avail >= 1) {
srv_cs = srv_conn->mux->attach(srv_conn, s->sess); srv_cs = si_attach_conn(&s->si[1], srv_conn);
if (srv_cs) if (srv_cs) {
si_attach_cs(&s->si[1], srv_cs); if (srv_conn->mux->attach(srv_conn, srv_cs, s->sess) == -1) {
srv_conn = NULL;
cs_init(srv_cs, NULL);
}
}
else else
srv_conn = NULL; srv_conn = NULL;
} }
@ -1574,12 +1578,11 @@ skip_reuse:
return SF_ERR_INTERNAL; /* how did we get there ? */ return SF_ERR_INTERNAL; /* how did we get there ? */
} }
srv_cs = si_alloc_cs(&s->si[1], srv_conn); srv_cs = si_attach_conn(&s->si[1], srv_conn);
if (!srv_cs) { if (!srv_cs) {
conn_free(srv_conn); conn_free(srv_conn);
return SF_ERR_RESOURCE; return SF_ERR_RESOURCE;
} }
srv_conn->ctx = srv_cs;
#if defined(USE_OPENSSL) && defined(TLSEXT_TYPE_application_layer_protocol_negotiation) #if defined(USE_OPENSSL) && defined(TLSEXT_TYPE_application_layer_protocol_negotiation)
if (!srv || if (!srv ||
(srv->use_ssl != 1 || (!(srv->ssl_ctx.alpn_str) && !(srv->ssl_ctx.npn_str)) || (srv->use_ssl != 1 || (!(srv->ssl_ctx.alpn_str) && !(srv->ssl_ctx.npn_str)) ||
@ -2288,7 +2291,7 @@ void back_handle_st_cer(struct stream *s)
} }
/* At this stage, we will trigger a connection retry (with or without /* At this stage, we will trigger a connection retry (with or without
* redispatch). Thus we must release the SI endpoint on the server side * redispatch). Thus we must reset the SI endpoint on the server side
* an close the attached connection. It is especially important to do it * an close the attached connection. It is especially important to do it
* now if the retry is not immediately performed, to be sure to release * now if the retry is not immediately performed, to be sure to release
* resources as soon as possible and to not catch errors from the lower * resources as soon as possible and to not catch errors from the lower
@ -2297,7 +2300,7 @@ void back_handle_st_cer(struct stream *s)
* Note: the stream-interface will be switched to ST_REQ, ST_ASS or * Note: the stream-interface will be switched to ST_REQ, ST_ASS or
* ST_TAR and SI_FL_ERR and SI_FL_EXP flags will be unset. * ST_TAR and SI_FL_ERR and SI_FL_EXP flags will be unset.
*/ */
si_release_endpoint(&s->si[1]); si_reset_endpoint(&s->si[1]);
stream_choose_redispatch(s); stream_choose_redispatch(s);

View File

@ -2715,7 +2715,7 @@ int pcli_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
* connection. * connection.
*/ */
if (!si_conn_ready(&s->si[1])) { if (!si_conn_ready(&s->si[1])) {
si_release_endpoint(&s->si[1]); si_reset_endpoint(&s->si[1]);
s->srv_conn = NULL; s->srv_conn = NULL;
} }

View File

@ -1257,7 +1257,7 @@ static __inline int do_l7_retry(struct stream *s, struct stream_interface *si)
res->to_forward = 0; res->to_forward = 0;
res->analyse_exp = TICK_ETERNITY; res->analyse_exp = TICK_ETERNITY;
res->total = 0; res->total = 0;
si_release_endpoint(&s->si[1]); si_reset_endpoint(&s->si[1]);
b_free(&req->buf); b_free(&req->buf);
/* Swap the L7 buffer with the channel buffer */ /* Swap the L7 buffer with the channel buffer */

View File

@ -3515,34 +3515,26 @@ static size_t fcgi_strm_parse_response(struct fcgi_strm *fstrm, struct buffer *b
* Attach a new stream to a connection * Attach a new stream to a connection
* (Used for outgoing connections) * (Used for outgoing connections)
*/ */
static struct conn_stream *fcgi_attach(struct connection *conn, struct session *sess) static int fcgi_attach(struct connection *conn, struct conn_stream *cs, struct session *sess)
{ {
struct conn_stream *cs;
struct fcgi_strm *fstrm; struct fcgi_strm *fstrm;
struct fcgi_conn *fconn = conn->ctx; struct fcgi_conn *fconn = conn->ctx;
TRACE_ENTER(FCGI_EV_FSTRM_NEW, conn); TRACE_ENTER(FCGI_EV_FSTRM_NEW, conn);
cs = cs_new(conn, conn->target);
if (!cs) {
TRACE_ERROR("CS allocation failure", FCGI_EV_FSTRM_NEW|FCGI_EV_FSTRM_ERR, conn);
goto err;
}
fstrm = fcgi_conn_stream_new(fconn, cs, sess); fstrm = fcgi_conn_stream_new(fconn, cs, sess);
if (!fstrm) { if (!fstrm)
cs_free(cs);
goto err; goto err;
}
/* the connection is not idle anymore, let's mark this */ /* the connection is not idle anymore, let's mark this */
HA_ATOMIC_AND(&fconn->wait_event.tasklet->state, ~TASK_F_USR1); HA_ATOMIC_AND(&fconn->wait_event.tasklet->state, ~TASK_F_USR1);
xprt_set_used(conn, conn->xprt, conn->xprt_ctx); xprt_set_used(conn, conn->xprt, conn->xprt_ctx);
TRACE_LEAVE(FCGI_EV_FSTRM_NEW, conn, fstrm); TRACE_LEAVE(FCGI_EV_FSTRM_NEW, conn, fstrm);
return cs; return 0;
err: err:
TRACE_DEVEL("leaving on error", FCGI_EV_FSTRM_NEW|FCGI_EV_FSTRM_ERR, conn); TRACE_DEVEL("leaving on error", FCGI_EV_FSTRM_NEW|FCGI_EV_FSTRM_ERR, conn);
return NULL; return -1;
} }
/* Retrieves the first valid conn_stream from this connection, or returns NULL. /* Retrieves the first valid conn_stream from this connection, or returns NULL.

View File

@ -3219,10 +3219,9 @@ struct task *h1_timeout_task(struct task *t, void *context, unsigned int state)
* Attach a new stream to a connection * Attach a new stream to a connection
* (Used for outgoing connections) * (Used for outgoing connections)
*/ */
static struct conn_stream *h1_attach(struct connection *conn, struct session *sess) static int h1_attach(struct connection *conn, struct conn_stream *cs, struct session *sess)
{ {
struct h1c *h1c = conn->ctx; struct h1c *h1c = conn->ctx;
struct conn_stream *cs = NULL;
struct h1s *h1s; struct h1s *h1s;
TRACE_ENTER(H1_EV_STRM_NEW, conn); TRACE_ENTER(H1_EV_STRM_NEW, conn);
@ -3231,12 +3230,6 @@ static struct conn_stream *h1_attach(struct connection *conn, struct session *se
goto err; goto err;
} }
cs = cs_new(h1c->conn, h1c->conn->target);
if (!cs) {
TRACE_ERROR("CS allocation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, conn);
goto err;
}
h1s = h1c_bck_stream_new(h1c, cs, sess); h1s = h1c_bck_stream_new(h1c, cs, sess);
if (h1s == NULL) { if (h1s == NULL) {
TRACE_ERROR("h1s creation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, conn); TRACE_ERROR("h1s creation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, conn);
@ -3248,11 +3241,10 @@ static struct conn_stream *h1_attach(struct connection *conn, struct session *se
xprt_set_used(conn, conn->xprt, conn->xprt_ctx); xprt_set_used(conn, conn->xprt, conn->xprt_ctx);
TRACE_LEAVE(H1_EV_STRM_NEW, conn, h1s); TRACE_LEAVE(H1_EV_STRM_NEW, conn, h1s);
return cs; return 0;
err: err:
cs_free(cs);
TRACE_DEVEL("leaving on error", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, conn); TRACE_DEVEL("leaving on error", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, conn);
return NULL; return -1;
} }
/* Retrieves a valid conn_stream from this connection, or returns NULL. For /* Retrieves a valid conn_stream from this connection, or returns NULL. For

View File

@ -4180,23 +4180,16 @@ do_leave:
* Attach a new stream to a connection * Attach a new stream to a connection
* (Used for outgoing connections) * (Used for outgoing connections)
*/ */
static struct conn_stream *h2_attach(struct connection *conn, struct session *sess) static int h2_attach(struct connection *conn, struct conn_stream *cs, struct session *sess)
{ {
struct conn_stream *cs;
struct h2s *h2s; struct h2s *h2s;
struct h2c *h2c = conn->ctx; struct h2c *h2c = conn->ctx;
TRACE_ENTER(H2_EV_H2S_NEW, conn); TRACE_ENTER(H2_EV_H2S_NEW, conn);
cs = cs_new(conn, conn->target);
if (!cs) {
TRACE_DEVEL("leaving on CS allocation failure", H2_EV_H2S_NEW|H2_EV_H2S_ERR, conn);
return NULL;
}
h2s = h2c_bck_stream_new(h2c, cs, sess); h2s = h2c_bck_stream_new(h2c, cs, sess);
if (!h2s) { if (!h2s) {
TRACE_DEVEL("leaving on stream creation failure", H2_EV_H2S_NEW|H2_EV_H2S_ERR, conn); TRACE_DEVEL("leaving on stream creation failure", H2_EV_H2S_NEW|H2_EV_H2S_ERR, conn);
cs_free(cs); return -1;
return NULL;
} }
/* the connection is not idle anymore, let's mark this */ /* the connection is not idle anymore, let's mark this */
@ -4204,7 +4197,7 @@ static struct conn_stream *h2_attach(struct connection *conn, struct session *se
xprt_set_used(h2c->conn, h2c->conn->xprt, h2c->conn->xprt_ctx); xprt_set_used(h2c->conn, h2c->conn->xprt, h2c->conn->xprt_ctx);
TRACE_LEAVE(H2_EV_H2S_NEW, conn, h2s); TRACE_LEAVE(H2_EV_H2S_NEW, conn, h2s);
return cs; return 0;
} }
/* Retrieves the first valid conn_stream from this connection, or returns NULL. /* Retrieves the first valid conn_stream from this connection, or returns NULL.

View File

@ -363,28 +363,18 @@ static int mux_pt_wake(struct connection *conn)
* Attach a new stream to a connection * Attach a new stream to a connection
* (Used for outgoing connections) * (Used for outgoing connections)
*/ */
static struct conn_stream *mux_pt_attach(struct connection *conn, struct session *sess) static int mux_pt_attach(struct connection *conn, struct conn_stream *cs, struct session *sess)
{ {
struct conn_stream *cs;
struct mux_pt_ctx *ctx = conn->ctx; struct mux_pt_ctx *ctx = conn->ctx;
TRACE_ENTER(PT_EV_STRM_NEW, conn); TRACE_ENTER(PT_EV_STRM_NEW, conn);
if (ctx->wait_event.events) if (ctx->wait_event.events)
conn->xprt->unsubscribe(ctx->conn, conn->xprt_ctx, SUB_RETRY_RECV, &ctx->wait_event); conn->xprt->unsubscribe(ctx->conn, conn->xprt_ctx, SUB_RETRY_RECV, &ctx->wait_event);
cs = cs_new(conn, conn->target);
if (!cs) {
TRACE_ERROR("CS allocation failure", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn);
goto fail;
}
ctx->cs = cs; ctx->cs = cs;
cs->flags |= CS_FL_RCV_MORE; cs->flags |= CS_FL_RCV_MORE;
TRACE_LEAVE(PT_EV_STRM_NEW, conn, cs); TRACE_LEAVE(PT_EV_STRM_NEW, conn, cs);
return (cs); return 0;
fail:
TRACE_DEVEL("leaving on error", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn);
return NULL;
} }
/* Retrieves a valid conn_stream from this connection, or returns NULL. For /* Retrieves a valid conn_stream from this connection, or returns NULL. For

View File

@ -2200,7 +2200,7 @@ struct task *process_stream(struct task *t, void *context, unsigned int state)
} }
} }
else { else {
si_release_endpoint(si_b); si_reset_endpoint(si_b);
si_b->state = SI_ST_CLO; /* shutw+ini = abort */ si_b->state = SI_ST_CLO; /* shutw+ini = abort */
channel_shutw_now(req); /* fix buffer flags upon abort */ channel_shutw_now(req); /* fix buffer flags upon abort */
channel_shutr_now(res); channel_shutr_now(res);