MAJOR: conn_stream/stream-int: move the appctx to the conn-stream
Thanks to previous changes, it is now possible to set an appctx as endpoint
for a conn-stream. This means the appctx is no longer linked to the
stream-interface but to the conn-stream. Thus, a pointer to the conn-stream
is explicitly stored in the stream-interface. The endpoint (connection or
appctx) can be retrieved via the conn-stream.
diff --git a/src/backend.c b/src/backend.c
index f289337..8367a8e 100644
--- a/src/backend.c
+++ b/src/backend.c
@@ -1251,7 +1251,6 @@
{
struct connection *cli_conn = objt_conn(strm_orig(s));
struct connection *srv_conn = NULL;
- struct conn_stream *srv_cs = NULL;
struct server *srv;
const int reuse_mode = s->be->options & PR_O_REUSE_MASK;
int reuse = 0;
@@ -1496,15 +1495,11 @@
}
if (avail >= 1) {
- srv_cs = si_attach_conn(&s->si[1], srv_conn);
- if (srv_cs) {
- if (srv_conn->mux->attach(srv_conn, srv_cs, s->sess) == -1) {
- srv_conn = NULL;
- cs_init(srv_cs, NULL);
- }
- }
- else
+ si_attach_conn(&s->si[1], srv_conn);
+ if (srv_conn->mux->attach(srv_conn, s->si[1].cs, s->sess) == -1) {
+ si_reset_endpoint(&s->si[1]);
srv_conn = NULL;
+ }
}
else
srv_conn = NULL;
@@ -1518,8 +1513,6 @@
/* no reuse or failed to reuse the connection above, pick a new one */
if (!srv_conn) {
srv_conn = conn_new(s->target);
- srv_cs = NULL;
-
if (srv_conn) {
DBG_TRACE_STATE("alloc new be connection", STRM_EV_STRM_PROC|STRM_EV_SI_ST, s);
srv_conn->owner = s->sess;
@@ -1578,11 +1571,7 @@
return SF_ERR_INTERNAL; /* how did we get there ? */
}
- srv_cs = si_attach_conn(&s->si[1], srv_conn);
- if (!srv_cs) {
- conn_free(srv_conn);
- return SF_ERR_RESOURCE;
- }
+ si_attach_conn(&s->si[1], srv_conn);
#if defined(USE_OPENSSL) && defined(TLSEXT_TYPE_application_layer_protocol_negotiation)
if (!srv ||
(srv->use_ssl != 1 || (!(srv->ssl_ctx.alpn_str) && !(srv->ssl_ctx.npn_str)) ||
@@ -1690,7 +1679,7 @@
if (init_mux) {
const struct mux_ops *alt_mux =
likely(!(s->flags & SF_WEBSOCKET)) ? NULL : srv_get_ws_proto(srv);
- if (conn_install_mux_be(srv_conn, srv_cs, s->sess, alt_mux) < 0) {
+ if (conn_install_mux_be(srv_conn, s->si[1].cs, s->sess, alt_mux) < 0) {
conn_full_close(srv_conn);
return SF_ERR_INTERNAL;
}
@@ -1752,7 +1741,7 @@
* sockets, socket pairs, and occasionally TCP connections on the
* loopback on a heavily loaded system.
*/
- if ((srv_conn->flags & CO_FL_ERROR || srv_cs->flags & CS_FL_ERROR))
+ if ((srv_conn->flags & CO_FL_ERROR || (s->si[1].cs)->flags & CS_FL_ERROR))
s->si[1].flags |= SI_FL_ERR;
/* If we had early data, and the handshake ended, then
@@ -1761,7 +1750,7 @@
* the handshake.
*/
if (!(srv_conn->flags & (CO_FL_WAIT_XPRT | CO_FL_EARLY_SSL_HS)))
- srv_cs->flags &= ~CS_FL_WAIT_FOR_HS;
+ (s->si[1].cs)->flags &= ~CS_FL_WAIT_FOR_HS;
if (!si_state_in(s->si[1].state, SI_SB_EST|SI_SB_DIS|SI_SB_CLO) &&
(srv_conn->flags & CO_FL_WAIT_XPRT) == 0) {
@@ -1778,7 +1767,7 @@
* wake callback. Otherwise si_cs_recv()/si_cs_send() already take
* care of it.
*/
- if ((srv_cs->flags & CS_FL_EOI) && !(si_ic(&s->si[1])->flags & CF_EOI))
+ if (((s->si[1].cs)->flags & CS_FL_EOI) && !(si_ic(&s->si[1])->flags & CF_EOI))
si_ic(&s->si[1])->flags |= (CF_EOI|CF_READ_PARTIAL);
/* catch all sync connect while the mux is not already installed */
@@ -2096,7 +2085,7 @@
if (unlikely(obj_type(s->target) == OBJ_TYPE_APPLET)) {
/* the applet directly goes to the EST state */
- struct appctx *appctx = objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
if (!appctx || appctx->applet != __objt_applet(s->target))
appctx = si_register_handler(si, objt_applet(s->target));
@@ -2231,7 +2220,7 @@
/* we probably have to release last stream from the server */
if (objt_server(s->target)) {
- struct connection *conn = cs_conn(objt_cs(si->end));
+ struct connection *conn = cs_conn(si->cs);
health_adjust(__objt_server(s->target), HANA_STATUS_L4_ERR);
diff --git a/src/cache.c b/src/cache.c
index 04e7327..d736969 100644
--- a/src/cache.c
+++ b/src/cache.c
@@ -2649,7 +2649,7 @@
return 0;
/* Get appctx from the stream_interface. */
- appctx = si_appctx(&smp->strm->si[1]);
+ appctx = cs_appctx(smp->strm->si[1].cs);
if (appctx && appctx->rule) {
cconf = appctx->rule->arg.act.p[0];
if (cconf) {
diff --git a/src/cli.c b/src/cli.c
index 541bad8..c889112 100644
--- a/src/cli.c
+++ b/src/cli.c
@@ -1920,7 +1920,7 @@
struct cmsghdr *cmsg;
struct stream_interface *si = appctx->owner;
struct stream *s = si_strm(si);
- struct connection *remote = cs_conn(objt_cs(si_opposite(si)->end));
+ struct connection *remote = cs_conn(si_opposite(si)->cs);
struct msghdr msghdr;
struct iovec iov;
struct timeval tv = { .tv_sec = 1, .tv_usec = 0 };
diff --git a/src/connection.c b/src/connection.c
index 9d12562..ee135ba 100644
--- a/src/connection.c
+++ b/src/connection.c
@@ -2018,7 +2018,7 @@
conn = (kw[0] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[0] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
/* No connection or a connection with a RAW muxx */
if (!conn || (conn->mux && !(conn->mux->flags & MX_FL_HTX)))
@@ -2115,7 +2115,7 @@
conn = (kw[0] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[0] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
if (!conn)
return 0;
@@ -2142,7 +2142,7 @@
conn = (kw[0] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[0] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
if (!conn)
return 0;
diff --git a/src/dns.c b/src/dns.c
index 5a0724b..2caa5ad 100644
--- a/src/dns.c
+++ b/src/dns.c
@@ -887,6 +887,7 @@
{
struct appctx *appctx;
struct session *sess;
+ struct conn_stream *cs;
struct stream *s;
struct applet *applet = &dns_session_applet;
@@ -898,15 +899,21 @@
sess = session_new(ds->dss->srv->proxy, NULL, &appctx->obj_type);
if (!sess) {
- ha_alert("out of memory in peer_session_create().\n");
+ ha_alert("out of memory in dns_session_create().\n");
goto out_free_appctx;
}
- if ((s = stream_new(sess, &appctx->obj_type, &BUF_NULL)) == NULL) {
- ha_alert("Failed to initialize stream in peer_session_create().\n");
+ cs = cs_new(&appctx->obj_type);
+ if (!cs) {
+ ha_alert("out of memory in dns_session_create().\n");
goto out_free_sess;
}
+ if ((s = stream_new(sess, cs, &BUF_NULL)) == NULL) {
+ ha_alert("Failed to initialize stream in dns_session_create().\n");
+ goto out_free_cs;
+ }
+
s->target = &ds->dss->srv->obj_type;
if (!sockaddr_alloc(&s->si[1].dst, &ds->dss->srv->addr, sizeof(ds->dss->srv->addr)))
@@ -924,13 +931,14 @@
s->res.rto = TICK_ETERNITY;
s->res.rex = TICK_ETERNITY;
ds->appctx = appctx;
- task_wakeup(s->task, TASK_WOKEN_INIT);
return appctx;
/* Error unrolling */
out_free_strm:
LIST_DELETE(&s->list);
pool_free(pool_head_stream, s);
+ out_free_cs:
+ cs_free(cs);
out_free_sess:
session_free(sess);
out_free_appctx:
diff --git a/src/flt_spoe.c b/src/flt_spoe.c
index 0badee3..8b2de43 100644
--- a/src/flt_spoe.c
+++ b/src/flt_spoe.c
@@ -1988,6 +1988,7 @@
{
struct appctx *appctx;
struct session *sess;
+ struct conn_stream *cs;
struct stream *strm;
if ((appctx = appctx_new(&spoe_applet)) == NULL)
@@ -2023,9 +2024,13 @@
if (!sess)
goto out_free_spoe;
- if ((strm = stream_new(sess, &appctx->obj_type, &BUF_NULL)) == NULL)
+ cs = cs_new(&appctx->obj_type);
+ if (!cs)
goto out_free_sess;
+ if ((strm = stream_new(sess, cs, &BUF_NULL)) == NULL)
+ goto out_free_cs;
+
stream_set_backend(strm, conf->agent->b.be);
/* applet is waiting for data */
@@ -2041,10 +2046,11 @@
_HA_ATOMIC_INC(&conf->agent->counters.applets);
task_wakeup(SPOE_APPCTX(appctx)->task, TASK_WOKEN_INIT);
- task_wakeup(strm->task, TASK_WOKEN_INIT);
return appctx;
/* Error unrolling */
+ out_free_cs:
+ cs_free(cs);
out_free_sess:
session_free(sess);
out_free_spoe:
diff --git a/src/frontend.c b/src/frontend.c
index c8ea0d4..56371da 100644
--- a/src/frontend.c
+++ b/src/frontend.c
@@ -105,7 +105,7 @@
int alpn_len;
/* try to report the ALPN value when available (also works for NPN) */
- if (conn == cs_conn(objt_cs(s->si[0].end))) {
+ if (conn == cs_conn(s->si[0].cs)) {
if (conn_get_alpn(conn, &alpn_str, &alpn_len) && alpn_str) {
int len = MIN(alpn_len, sizeof(alpn) - 1);
memcpy(alpn, alpn_str, len);
diff --git a/src/hlua.c b/src/hlua.c
index 8c7dc61..c61e666 100644
--- a/src/hlua.c
+++ b/src/hlua.c
@@ -2617,7 +2617,7 @@
si = appctx->owner;
s = si_strm(si);
- conn = cs_conn(objt_cs(s->si[1].end));
+ conn = cs_conn(s->si[1].cs);
if (!conn || !conn_get_src(conn)) {
xref_unlock(&socket->xref, peer);
lua_pushnil(L);
@@ -2684,7 +2684,7 @@
return 2;
}
- appctx = __objt_appctx(s->si[0].end);
+ appctx = cs_appctx(s->si[0].cs);
/* Check for connection established. */
if (appctx->ctx.hlua_cosocket.connected) {
@@ -2916,6 +2916,7 @@
struct hlua_socket *socket;
struct appctx *appctx;
struct session *sess;
+ struct conn_stream *cs;
struct stream *strm;
/* Check stack size. */
@@ -2957,13 +2958,19 @@
sess = session_new(socket_proxy, NULL, &appctx->obj_type);
if (!sess) {
hlua_pusherror(L, "socket: out of memory");
+ goto out_fail_appctx;
+ }
+
+ cs = cs_new(&appctx->obj_type);
+ if (!cs) {
+ hlua_pusherror(L, "socket: out of memory");
goto out_fail_sess;
}
- strm = stream_new(sess, &appctx->obj_type, &BUF_NULL);
+ strm = stream_new(sess, cs, &BUF_NULL);
if (!strm) {
hlua_pusherror(L, "socket: out of memory");
- goto out_fail_stream;
+ goto out_fail_cs;
}
/* Initialise cross reference between stream and Lua socket object. */
@@ -2981,9 +2988,11 @@
return 1;
- out_fail_stream:
- session_free(sess);
+ out_fail_cs:
+ cs_free(cs);
out_fail_sess:
+ session_free(sess);
+ out_fail_appctx:
appctx_free(appctx);
out_fail_conf:
WILL_LJMP(lua_error(L));
diff --git a/src/http_ana.c b/src/http_ana.c
index 848750c..f0f49a7 100644
--- a/src/http_ana.c
+++ b/src/http_ana.c
@@ -1325,7 +1325,7 @@
if (unlikely(htx_is_empty(htx) || htx->first == -1)) {
/* 1: have we encountered a read error ? */
if (rep->flags & CF_READ_ERROR) {
- struct connection *conn = cs_conn(objt_cs(s->si[1].end));
+ struct connection *conn = cs_conn(s->si[1].cs);
/* Perform a L7 retry because server refuses the early data. */
if ((si_b->flags & SI_FL_L7_RETRY) &&
@@ -1656,7 +1656,7 @@
/* check for NTML authentication headers in 401 (WWW-Authenticate) and
* 407 (Proxy-Authenticate) responses and set the connection to private
*/
- srv_conn = cs_conn(objt_cs(s->si[1].end));
+ srv_conn = cs_conn(s->si[1].cs);
if (srv_conn) {
struct ist hdr;
struct http_hdr_ctx ctx;
@@ -3883,7 +3883,7 @@
struct htx *htx;
struct htx_sl *sl;
- appctx = si_appctx(si);
+ appctx = cs_appctx(si->cs);
memset(&appctx->ctx.stats, 0, sizeof(appctx->ctx.stats));
appctx->st1 = appctx->st2 = 0;
appctx->ctx.stats.st_code = STAT_STATUS_INIT;
@@ -5004,7 +5004,7 @@
chunk_printf(&trash, "%08x:%s.%s[%04x:%04x]: ", s->uniq_id, s->be->id,
dir,
objt_conn(sess->origin) ? (unsigned short)__objt_conn(sess->origin)->handle.fd : -1,
- cs_conn(objt_cs(s->si[1].end)) ? (unsigned short)(cs_conn(__objt_cs(s->si[1].end)))->handle.fd : -1);
+ cs_conn(s->si[1].cs) ? (unsigned short)(cs_conn(s->si[1].cs))->handle.fd : -1);
max = HTX_SL_P1_LEN(sl);
UBOUND(max, trash.size - trash.data - 3);
@@ -5035,7 +5035,7 @@
chunk_printf(&trash, "%08x:%s.%s[%04x:%04x]: ", s->uniq_id, s->be->id,
dir,
objt_conn(sess->origin) ? (unsigned short)__objt_conn(sess->origin)->handle.fd : -1,
- cs_conn(objt_cs(s->si[1].end)) ? (unsigned short)(cs_conn(__objt_cs(s->si[1].end)))->handle.fd : -1);
+ cs_conn(s->si[1].cs) ? (unsigned short)(cs_conn(s->si[1].cs))->handle.fd : -1);
max = n.len;
UBOUND(max, trash.size - trash.data - 3);
@@ -5091,7 +5091,7 @@
struct http_txn *http_create_txn(struct stream *s)
{
struct http_txn *txn;
- struct conn_stream *cs = objt_cs(s->si[0].end);
+ struct conn_stream *cs = s->si[0].cs;
txn = pool_alloc(pool_head_http_txn);
if (!txn)
diff --git a/src/http_client.c b/src/http_client.c
index 1877a0f..64e5190 100644
--- a/src/http_client.c
+++ b/src/http_client.c
@@ -454,6 +454,7 @@
struct applet *applet = &httpclient_applet;
struct appctx *appctx;
struct session *sess;
+ struct conn_stream *cs;
struct stream *s;
int len;
struct sockaddr_storage ss_url;
@@ -485,9 +486,14 @@
ha_alert("httpclient: out of memory in %s:%d.\n", __FUNCTION__, __LINE__);
goto out_free_appctx;
}
- if ((s = stream_new(sess, &appctx->obj_type, &hc->req.buf)) == NULL) {
+ cs = cs_new(&appctx->obj_type);
+ if (!cs) {
+ ha_alert("httpclient: out of memory in %s:%d.\n", __FUNCTION__, __LINE__);
+ goto out_free_sess;
+ }
+ if ((s = stream_new(sess, cs, &hc->req.buf)) == NULL) {
ha_alert("httpclient: Failed to initialize stream %s:%d.\n", __FUNCTION__, __LINE__);
- goto out_free_appctx;
+ goto out_free_cs;
}
/* set the "timeout server" */
@@ -528,7 +534,6 @@
si_cant_get(&s->si[0]);
appctx_wakeup(appctx);
- task_wakeup(s->task, TASK_WOKEN_INIT);
hc->appctx = appctx;
hc->flags |= HTTPCLIENT_FS_STARTED;
appctx->ctx.httpclient.ptr = hc;
@@ -543,6 +548,8 @@
out_free_stream:
LIST_DELETE(&s->list);
pool_free(pool_head_stream, s);
+out_free_cs:
+ cs_free(cs);
out_free_sess:
session_free(sess);
out_free_appctx:
diff --git a/src/log.c b/src/log.c
index 92eef6e..0213435 100644
--- a/src/log.c
+++ b/src/log.c
@@ -1970,7 +1970,7 @@
if (likely(s)) {
be = s->be;
txn = s->txn;
- be_conn = cs_conn(objt_cs(s->si[1].end));
+ be_conn = cs_conn(s->si[1].cs);
status = (txn ? txn->status : 0);
s_flags = s->flags;
uniq_id = s->uniq_id;
diff --git a/src/mux_h1.c b/src/mux_h1.c
index cc53364..5fdcd46 100644
--- a/src/mux_h1.c
+++ b/src/mux_h1.c
@@ -696,7 +696,7 @@
if (h1s->req.flags & H1_MF_UPG_WEBSOCKET)
cs->flags |= CS_FL_WEBSOCKET;
- if (stream_create_from_cs(cs, input) < 0) {
+ if (!stream_new(h1c->conn->owner, cs, input)) {
TRACE_DEVEL("leaving on stream creation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, h1c->conn, h1s);
goto err;
}
diff --git a/src/mux_h2.c b/src/mux_h2.c
index d142793..3a98e5d 100644
--- a/src/mux_h2.c
+++ b/src/mux_h2.c
@@ -1550,7 +1550,7 @@
* request. We can set the value now, it will be copied by stream_new().
*/
sess->t_idle = tv_ms_elapsed(&sess->tv_accept, &now) - sess->t_handshake;
- if (stream_create_from_cs(cs, input) < 0)
+ if (!stream_new(h2c->conn->owner, cs, input))
goto out_free_cs;
/* We want the accept date presented to the next stream to be the one
diff --git a/src/mux_pt.c b/src/mux_pt.c
index e048a49..13e60d8 100644
--- a/src/mux_pt.c
+++ b/src/mux_pt.c
@@ -297,7 +297,7 @@
goto fail_free_ctx;
}
- if (stream_create_from_cs(cs, &BUF_NULL) < 0) {
+ if (!stream_new(conn->owner, cs, &BUF_NULL)) {
TRACE_ERROR("stream creation failure", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn, cs);
goto fail_free;
}
diff --git a/src/peers.c b/src/peers.c
index deb5a89..df3fd9e 100644
--- a/src/peers.c
+++ b/src/peers.c
@@ -3181,6 +3181,7 @@
struct proxy *p = peers->peers_fe; /* attached frontend */
struct appctx *appctx;
struct session *sess;
+ struct conn_stream *cs;
struct stream *s;
peer->new_conn++;
@@ -3203,11 +3204,17 @@
goto out_free_appctx;
}
- if ((s = stream_new(sess, &appctx->obj_type, &BUF_NULL)) == NULL) {
- ha_alert("Failed to initialize stream in peer_session_create().\n");
+ cs = cs_new(&appctx->obj_type);
+ if (!cs) {
+ ha_alert("out of memory in peer_session_create().\n");
goto out_free_sess;
}
+ if ((s = stream_new(sess, cs, &BUF_NULL)) == NULL) {
+ ha_alert("Failed to initialize stream in peer_session_create().\n");
+ goto out_free_cs;
+ }
+
/* applet is waiting for data */
si_cant_get(&s->si[0]);
appctx_wakeup(appctx);
@@ -3225,7 +3232,6 @@
s->res.flags |= CF_READ_DONTWAIT;
peer->appctx = appctx;
- task_wakeup(s->task, TASK_WOKEN_INIT);
_HA_ATOMIC_INC(&active_peers);
return appctx;
@@ -3233,6 +3239,8 @@
out_free_strm:
LIST_DELETE(&s->list);
pool_free(pool_head_stream, s);
+ out_free_cs:
+ cs_free(cs);
out_free_sess:
session_free(sess);
out_free_appctx:
@@ -4019,4 +4027,3 @@
/* Register cli keywords */
INITCALL1(STG_REGISTER, cli_register_kw, &cli_kws);
-
diff --git a/src/proxy.c b/src/proxy.c
index b874cb1..a97b3cd 100644
--- a/src/proxy.c
+++ b/src/proxy.c
@@ -2612,7 +2612,7 @@
*/
static int dump_servers_state(struct stream_interface *si)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct proxy *px = appctx->ctx.cli.p0;
struct server *srv;
char srv_addr[INET6_ADDRSTRLEN + 1];
diff --git a/src/resolvers.c b/src/resolvers.c
index 097d007..43e755b 100644
--- a/src/resolvers.c
+++ b/src/resolvers.c
@@ -2586,7 +2586,7 @@
struct field *stats, size_t stats_count,
struct list *stat_modules)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct channel *rep = si_ic(si);
struct stats_module *mod;
size_t idx = 0;
@@ -2620,7 +2620,7 @@
struct field *stats, size_t stats_count,
struct list *stat_modules)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct channel *rep = si_ic(si);
struct resolvers *resolver = appctx->ctx.stats.obj1;
struct dns_nameserver *ns = appctx->ctx.stats.obj2;
diff --git a/src/sink.c b/src/sink.c
index c1da18c..a3464ab 100644
--- a/src/sink.c
+++ b/src/sink.c
@@ -636,6 +636,7 @@
struct proxy *p = sink->forward_px;
struct appctx *appctx;
struct session *sess;
+ struct conn_stream *cs;
struct stream *s;
struct applet *applet = &sink_forward_applet;
@@ -650,15 +651,21 @@
sess = session_new(p, NULL, &appctx->obj_type);
if (!sess) {
- ha_alert("out of memory in peer_session_create().\n");
+ ha_alert("out of memory in sink_forward_session_create().\n");
goto out_free_appctx;
}
- if ((s = stream_new(sess, &appctx->obj_type, &BUF_NULL)) == NULL) {
- ha_alert("Failed to initialize stream in peer_session_create().\n");
+ cs = cs_new(&appctx->obj_type);
+ if (!cs) {
+ ha_alert("out of memory in sink_forward_session_create");
goto out_free_sess;
}
+ if ((s = stream_new(sess, cs, &BUF_NULL)) == NULL) {
+ ha_alert("Failed to initialize stream in sink_forward_session_create().\n");
+ goto out_free_cs;
+ }
+
s->target = &sft->srv->obj_type;
if (!sockaddr_alloc(&s->si[1].dst, &sft->srv->addr, sizeof(sft->srv->addr)))
@@ -676,13 +683,14 @@
s->res.rto = TICK_ETERNITY;
s->res.rex = TICK_ETERNITY;
sft->appctx = appctx;
- task_wakeup(s->task, TASK_WOKEN_INIT);
return appctx;
/* Error unrolling */
out_free_strm:
LIST_DELETE(&s->list);
pool_free(pool_head_stream, s);
+ out_free_cs:
+ cs_free(cs);
out_free_sess:
session_free(sess);
out_free_appctx:
diff --git a/src/ssl_sample.c b/src/ssl_sample.c
index 0ba14e9..0a78ecb 100644
--- a/src/ssl_sample.c
+++ b/src/ssl_sample.c
@@ -529,7 +529,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
@@ -584,7 +584,7 @@
int i;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
@@ -647,7 +647,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
ssl = ssl_sock_get_ssl_object(conn);
@@ -700,7 +700,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
@@ -751,7 +751,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
@@ -803,7 +803,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
@@ -871,7 +871,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
@@ -923,7 +923,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
@@ -1020,7 +1020,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
ssl = ssl_sock_get_ssl_object(conn);
@@ -1065,7 +1065,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
@@ -1122,7 +1122,7 @@
SSL *ssl;
if (conn_server)
- conn = smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ conn = smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
else
conn = objt_conn(smp->sess->origin);
ssl = ssl_sock_get_ssl_object(conn);
@@ -1174,7 +1174,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
smp->data.type = SMP_T_BOOL;
smp->data.u.sint = (conn && conn->xprt == &ssl_sock);
@@ -1211,7 +1211,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
ssl = ssl_sock_get_ssl_object(conn);
@@ -1234,7 +1234,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
smp->flags = 0;
ssl = ssl_sock_get_ssl_object(conn);
@@ -1268,7 +1268,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
smp->flags = 0;
ssl = ssl_sock_get_ssl_object(conn);
@@ -1299,7 +1299,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
smp->flags = 0;
ssl = ssl_sock_get_ssl_object(conn);
@@ -1331,7 +1331,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
ssl = ssl_sock_get_ssl_object(conn);
if (!ssl)
@@ -1366,7 +1366,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
ssl = ssl_sock_get_ssl_object(conn);
if (!ssl)
@@ -1399,7 +1399,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
smp->flags = 0;
ssl = ssl_sock_get_ssl_object(conn);
@@ -1437,7 +1437,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
ssl = ssl_sock_get_ssl_object(conn);
if (!ssl)
@@ -1469,7 +1469,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
ssl = ssl_sock_get_ssl_object(conn);
if (!ssl)
@@ -1506,7 +1506,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
ssl = ssl_sock_get_ssl_object(conn);
if (!ssl)
@@ -1655,7 +1655,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
if (!conn || conn->xprt != &ssl_sock)
return 0;
@@ -1708,7 +1708,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
if (!conn || conn->xprt != &ssl_sock)
return 0;
@@ -1841,7 +1841,7 @@
const char *sfx;
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
if (!conn)
return 0;
@@ -1938,7 +1938,7 @@
conn = (kw[4] == 'b') ? cs_conn(__objt_check(smp->sess->origin)->cs) : NULL;
else
conn = (kw[4] != 'b') ? objt_conn(smp->sess->origin) :
- smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)) : NULL;
+ smp->strm ? cs_conn(smp->strm->si[1].cs) : NULL;
smp->flags = 0;
ssl = ssl_sock_get_ssl_object(conn);
diff --git a/src/ssl_sock.c b/src/ssl_sock.c
index 460eb60..b79fc71 100644
--- a/src/ssl_sock.c
+++ b/src/ssl_sock.c
@@ -7735,7 +7735,7 @@
struct conn_stream *cs;
conn = objt_conn(sess->origin);
- cs = objt_cs(s->si[0].end);
+ cs = s->si[0].cs;
if (conn && cs) {
if (conn->flags & (CO_FL_EARLY_SSL_HS | CO_FL_SSL_WAIT_HS)) {
diff --git a/src/stats.c b/src/stats.c
index bacde28..1600ef4 100644
--- a/src/stats.c
+++ b/src/stats.c
@@ -1812,7 +1812,7 @@
*/
static int stats_dump_fe_stats(struct stream_interface *si, struct proxy *px)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
struct stats_module *mod;
size_t stats_count = ST_F_TOTAL_FIELDS;
@@ -1979,7 +1979,7 @@
*/
static int stats_dump_li_stats(struct stream_interface *si, struct proxy *px, struct listener *l)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
struct stats_module *mod;
size_t stats_count = ST_F_TOTAL_FIELDS;
@@ -2490,7 +2490,7 @@
*/
static int stats_dump_sv_stats(struct stream_interface *si, struct proxy *px, struct server *sv)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct stats_module *mod;
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
size_t stats_count = ST_F_TOTAL_FIELDS;
@@ -2815,7 +2815,7 @@
*/
static int stats_dump_be_stats(struct stream_interface *si, struct proxy *px)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct field *stats = stat_l[STATS_DOMAIN_PROXY];
struct stats_module *mod;
size_t stats_count = ST_F_TOTAL_FIELDS;
@@ -2856,7 +2856,7 @@
*/
static void stats_dump_html_px_hdr(struct stream_interface *si, struct proxy *px)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
char scope_txt[STAT_SCOPE_TXT_MAXLEN + sizeof STAT_SCOPE_PATTERN];
struct stats_module *mod;
int stats_module_len = 0;
@@ -2965,7 +2965,7 @@
*/
static void stats_dump_html_px_end(struct stream_interface *si, struct proxy *px)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
chunk_appendf(&trash, "</table>");
if ((px->cap & PR_CAP_BE) && px->srv && (appctx->ctx.stats.flags & STAT_ADMIN)) {
@@ -3007,7 +3007,7 @@
int stats_dump_proxy_to_buffer(struct stream_interface *si, struct htx *htx,
struct proxy *px, struct uri_auth *uri)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct stream *s = si_strm(si);
struct channel *rep = si_ic(si);
struct server *sv, *svs; /* server and server-state, server-state=server or server->track */
@@ -3380,7 +3380,7 @@
*/
static void stats_dump_html_info(struct stream_interface *si, struct uri_auth *uri)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
unsigned int up = (now.tv_sec - start_date.tv_sec);
char scope_txt[STAT_SCOPE_TXT_MAXLEN + sizeof STAT_SCOPE_PATTERN];
const char *scope_ptr = stats_scope_ptr(appctx, si);
@@ -3662,7 +3662,7 @@
struct htx *htx,
struct uri_auth *uri)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct channel *rep = si_ic(si);
struct proxy *px;
@@ -3709,7 +3709,7 @@
static int stats_dump_stat_to_buffer(struct stream_interface *si, struct htx *htx,
struct uri_auth *uri)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct channel *rep = si_ic(si);
enum stats_domain domain = appctx->ctx.stats.domain;
@@ -3811,7 +3811,7 @@
static int stats_process_http_post(struct stream_interface *si)
{
struct stream *s = si_strm(si);
- struct appctx *appctx = objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct proxy *px = NULL;
struct server *sv = NULL;
@@ -4147,7 +4147,7 @@
{
struct stream *s = si_strm(si);
struct uri_auth *uri = s->be->uri_auth;
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct htx_sl *sl;
unsigned int flags;
@@ -4201,7 +4201,7 @@
char scope_txt[STAT_SCOPE_TXT_MAXLEN + sizeof STAT_SCOPE_PATTERN];
struct stream *s = si_strm(si);
struct uri_auth *uri = s->be->uri_auth;
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct htx_sl *sl;
unsigned int flags;
@@ -4516,7 +4516,7 @@
*/
static int stats_dump_info_to_buffer(struct stream_interface *si)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
if (!stats_fill_info(info, INF_TOTAL_FIELDS, appctx->ctx.stats.flags))
return 0;
diff --git a/src/stream.c b/src/stream.c
index 10cac1e..622d6b8 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -266,47 +266,24 @@
}
}
-/* Create a new stream for connection <conn>. Return < 0 on error. This is only
- * valid right after the handshake, before the connection's data layer is
- * initialized, because it relies on the session to be in conn->owner. On
+/* Upgrade an existing stream for conn-stream <cs>. Return < 0 on error. This
+ * is only valid right after a TCP to H1 upgrade. The stream should be
+ * "reativated" by removing SF_IGNORE flag. And the right mode must be set. On
* success, <input> buffer is transferred to the stream and thus points to
* BUF_NULL. On error, it is unchanged and it is the caller responsibility to
- * release it.
- */
-int stream_create_from_cs(struct conn_stream *cs, struct buffer *input)
-{
- struct connection *conn = cs_conn(cs);
- struct stream *strm;
-
- if (!conn)
- return -1;
-
- strm = stream_new(conn->owner, &cs->obj_type, input);
- if (strm == NULL)
- return -1;
-
- task_wakeup(strm->task, TASK_WOKEN_INIT);
- return 0;
-}
-
-/* Upgrade an existing TCP stream for connection <conn>. Return < 0 on error.
- * This is only valid right after a TCP to H1 upgrade. The stream should be
- * "reativated" by removing SF_IGNORE flag. And the right mode must be set.
- * On success, <input> buffer is transferred to the stream and thus points to
- * BUF_NULL. On error, it is unchanged and it is the caller responsibility to
* release it (this never happens for now).
*/
int stream_upgrade_from_cs(struct conn_stream *cs, struct buffer *input)
{
- struct connection *conn = cs_conn(cs);
struct stream_interface *si = cs->data;
struct stream *s = si_strm(si);
- if (!conn)
- return -1;
+ if (cs_conn_mux(cs)) {
+ const struct mux_ops *mux = cs_conn_mux(cs);
- if (conn->mux->flags & MX_FL_HTX)
- s->flags |= SF_HTX;
+ if (mux->flags & MX_FL_HTX)
+ s->flags |= SF_HTX;
+ }
if (!b_is_null(input)) {
/* Xfer the input buffer to the request channel. <input> will
@@ -362,12 +339,10 @@
* transfer to the stream and <input> is set to BUF_NULL. On error, <input>
* buffer is unchanged and it is the caller responsibility to release it.
*/
-struct stream *stream_new(struct session *sess, enum obj_type *origin, struct buffer *input)
+struct stream *stream_new(struct session *sess, struct conn_stream *cs, struct buffer *input)
{
struct stream *s;
struct task *t;
- struct conn_stream *cs = objt_cs(origin);
- struct appctx *appctx = objt_appctx(origin);
DBG_TRACE_ENTER(STRM_EV_STRM_NEW);
if (unlikely((s = pool_alloc(pool_head_stream)) == NULL))
@@ -472,9 +447,10 @@
if (sess->fe->mode == PR_MODE_HTTP)
s->flags |= SF_HTX;
- if (appctx)
- si_attach_appctx(&s->si[0], appctx);
- else if (cs) {
+ si_attach_cs(&s->si[0], cs);
+ if (cs->flags & CS_FL_WEBSOCKET)
+ s->flags |= SF_WEBSOCKET;
+ if (cs_conn(cs)) {
const struct mux_ops *mux = cs_conn_mux(cs);
if (mux) {
@@ -482,12 +458,7 @@
s->si[0].flags |= SI_FL_CLEAN_ABRT;
if (mux->flags & MX_FL_HTX)
s->flags |= SF_HTX;
-
- if (cs->flags & CS_FL_WEBSOCKET)
- s->flags |= SF_WEBSOCKET;
}
- /* attach the incoming connection to the stream interface now. */
- si_attach_cs(&s->si[0], cs);
}
@@ -504,7 +475,8 @@
if (likely(sess->fe->options2 & PR_O2_INDEPSTR))
s->si[1].flags |= SI_FL_INDEP_STR;
- if (!si_alloc_cs(&s->si[1], NULL))
+ s->si[1].cs = cs_new(NULL);
+ if (!s->si[1].cs)
goto out_fail_alloc_cs;
stream_init_srv_conn(s);
@@ -572,7 +544,7 @@
s->si[1].l7_buffer = BUF_NULL;
/* finish initialization of the accepted file descriptor */
- if (appctx)
+ if (cs_appctx(cs))
si_want_get(&s->si[0]);
if (sess->fe->accept && sess->fe->accept(s) < 0)
@@ -596,6 +568,7 @@
* the caller must handle the task_wakeup
*/
DBG_TRACE_LEAVE(STRM_EV_STRM_NEW, s);
+ task_wakeup(s->task, TASK_WOKEN_INIT);
return s;
/* Error unrolling */
@@ -622,7 +595,7 @@
struct session *sess = strm_sess(s);
struct proxy *fe = sess->fe;
struct bref *bref, *back;
- struct conn_stream *cli_cs = objt_cs(s->si[0].end);
+ /* struct conn_stream *cli_cs = objt_cs(s->si[0].end); */
int must_free_sess;
int i;
@@ -683,8 +656,9 @@
http_destroy_txn(s);
/* ensure the client-side transport layer is destroyed */
- if (cli_cs)
- cs_close(cli_cs);
+ /* Be sure it is useless !! */
+ /* if (cli_cs) */
+ /* cs_close(cli_cs); */
for (i = 0; i < s->store_count; i++) {
if (!s->store[i].ts)
@@ -747,7 +721,8 @@
LIST_DELETE(&s->list);
/* applets do not release session yet */
- must_free_sess = objt_appctx(sess->origin) && sess->origin == s->si[0].end;
+ /* FIXME: Handle it in appctx_free ??? */
+ must_free_sess = objt_appctx(sess->origin) && sess->origin == s->si[0].cs->end;
si_release_endpoint(&s->si[1]);
@@ -900,7 +875,7 @@
static void back_establish(struct stream *s)
{
struct stream_interface *si = &s->si[1];
- struct connection *conn = cs_conn(objt_cs(si->end));
+ struct connection *conn = cs_conn(si->cs);
struct channel *req = &s->req;
struct channel *rep = &s->res;
@@ -1023,12 +998,12 @@
return ACT_RET_ERR;
/* Initialise the context. */
- appctx = si_appctx(&s->si[1]);
+ appctx = cs_appctx(s->si[1].cs);
memset(&appctx->ctx, 0, sizeof(appctx->ctx));
appctx->rule = rule;
}
else
- appctx = si_appctx(&s->si[1]);
+ appctx = cs_appctx(s->si[1].cs);
/* Stops the applet scheduling, in case of the init function miss
* some data.
@@ -1501,8 +1476,8 @@
*/
int stream_set_http_mode(struct stream *s, const struct mux_proto_list *mux_proto)
{
+ struct conn_stream *cs = s->si[0].cs;
struct connection *conn;
- struct conn_stream *cs;
/* Already an HTTP stream */
if (IS_HTX_STRM(s))
@@ -1513,9 +1488,8 @@
if (unlikely(!s->txn && !http_create_txn(s)))
return 0;
- conn = objt_conn(strm_sess(s)->origin);
- cs = objt_cs(s->si[0].end);
- if (conn && cs) {
+ conn = cs_conn(cs);
+ if (conn) {
si_rx_endp_more(&s->si[0]);
/* Make sure we're unsubscribed, the the new
* mux will probably want to subscribe to
@@ -1541,8 +1515,8 @@
* silently destroyed. The new mux will create new
* streams.
*/
- cs_free(cs);
- si_detach_endpoint(&s->si[0]);
+ /* FIXME: must be tested */
+ /* si_release_endpoint(&s->si[0]); */
s->logs.logwait = 0;
s->logs.level = 0;
channel_abort(&s->req);
@@ -2180,10 +2154,10 @@
if (!(req->flags & (CF_KERN_SPLICING|CF_SHUTR)) &&
req->to_forward &&
(global.tune.options & GTUNE_USE_SPLICE) &&
- (cs_conn(objt_cs(si_f->end)) && cs_conn(__objt_cs(si_f->end))->xprt && cs_conn(__objt_cs(si_f->end))->xprt->rcv_pipe &&
- cs_conn(__objt_cs(si_f->end))->mux && cs_conn(__objt_cs(si_f->end))->mux->rcv_pipe) &&
- (cs_conn(objt_cs(si_b->end)) && cs_conn(__objt_cs(si_b->end))->xprt && cs_conn(__objt_cs(si_b->end))->xprt->snd_pipe &&
- cs_conn(__objt_cs(si_b->end))->mux && cs_conn(__objt_cs(si_b->end))->mux->snd_pipe) &&
+ (cs_conn(si_f->cs) && cs_conn(si_f->cs)->xprt && cs_conn(si_f->cs)->xprt->rcv_pipe &&
+ cs_conn(si_f->cs)->mux && cs_conn(si_f->cs)->mux->rcv_pipe) &&
+ (cs_conn(si_b->cs) && cs_conn(si_b->cs)->xprt && cs_conn(si_b->cs)->xprt->snd_pipe &&
+ cs_conn(si_b->cs)->mux && cs_conn(si_b->cs)->mux->snd_pipe) &&
(pipes_used < global.maxpipes) &&
(((sess->fe->options2|s->be->options2) & PR_O2_SPLIC_REQ) ||
(((sess->fe->options2|s->be->options2) & PR_O2_SPLIC_AUT) &&
@@ -2373,10 +2347,10 @@
if (!(res->flags & (CF_KERN_SPLICING|CF_SHUTR)) &&
res->to_forward &&
(global.tune.options & GTUNE_USE_SPLICE) &&
- (cs_conn(objt_cs(si_f->end)) && cs_conn(__objt_cs(si_f->end))->xprt && cs_conn(__objt_cs(si_f->end))->xprt->snd_pipe &&
- cs_conn(__objt_cs(si_f->end))->mux && cs_conn(__objt_cs(si_f->end))->mux->snd_pipe) &&
- (cs_conn(objt_cs(si_b->end)) && cs_conn(__objt_cs(si_b->end))->xprt && cs_conn(__objt_cs(si_b->end))->xprt->rcv_pipe &&
- cs_conn(__objt_cs(si_b->end))->mux && cs_conn(__objt_cs(si_b->end))->mux->rcv_pipe) &&
+ (cs_conn(si_f->cs) && cs_conn(si_f->cs)->xprt && cs_conn(si_f->cs)->xprt->snd_pipe &&
+ cs_conn(si_f->cs)->mux && cs_conn(si_f->cs)->mux->snd_pipe) &&
+ (cs_conn(si_b->cs) && cs_conn(si_b->cs)->xprt && cs_conn(si_b->cs)->xprt->rcv_pipe &&
+ cs_conn(si_b->cs)->mux && cs_conn(si_b->cs)->mux->rcv_pipe) &&
(pipes_used < global.maxpipes) &&
(((sess->fe->options2|s->be->options2) & PR_O2_SPLIC_RTR) ||
(((sess->fe->options2|s->be->options2) & PR_O2_SPLIC_AUT) &&
@@ -2453,8 +2427,8 @@
si_b->prev_state == SI_ST_EST) {
chunk_printf(&trash, "%08x:%s.srvcls[%04x:%04x]\n",
s->uniq_id, s->be->id,
- cs_conn(objt_cs(si_f->end)) ? (unsigned short)cs_conn(__objt_cs(si_f->end))->handle.fd : -1,
- cs_conn(objt_cs(si_b->end)) ? (unsigned short)cs_conn(__objt_cs(si_b->end))->handle.fd : -1);
+ cs_conn(si_f->cs) ? (unsigned short)cs_conn(si_f->cs)->handle.fd : -1,
+ cs_conn(si_b->cs) ? (unsigned short)cs_conn(si_b->cs)->handle.fd : -1);
DISGUISE(write(1, trash.area, trash.data));
}
@@ -2462,8 +2436,8 @@
si_f->prev_state == SI_ST_EST) {
chunk_printf(&trash, "%08x:%s.clicls[%04x:%04x]\n",
s->uniq_id, s->be->id,
- cs_conn(objt_cs(si_f->end)) ? (unsigned short)cs_conn(__objt_cs(si_f->end))->handle.fd : -1,
- cs_conn(objt_cs(si_b->end)) ? (unsigned short)cs_conn(__objt_cs(si_b->end))->handle.fd : -1);
+ cs_conn(si_f->cs) ? (unsigned short)cs_conn(si_f->cs)->handle.fd : -1,
+ cs_conn(si_b->cs) ? (unsigned short)cs_conn(si_b->cs)->handle.fd : -1);
DISGUISE(write(1, trash.area, trash.data));
}
}
@@ -2530,8 +2504,8 @@
(!(global.mode & MODE_QUIET) || (global.mode & MODE_VERBOSE)))) {
chunk_printf(&trash, "%08x:%s.closed[%04x:%04x]\n",
s->uniq_id, s->be->id,
- cs_conn(objt_cs(si_f->end)) ? (unsigned short)cs_conn(__objt_cs(si_f->end))->handle.fd : -1,
- cs_conn(objt_cs(si_b->end)) ? (unsigned short)cs_conn(__objt_cs(si_b->end))->handle.fd : -1);
+ cs_conn(si_f->cs) ? (unsigned short)cs_conn(si_f->cs)->handle.fd : -1,
+ cs_conn(si_b->cs) ? (unsigned short)cs_conn(si_b->cs)->handle.fd : -1);
DISGUISE(write(1, trash.area, trash.data));
}
@@ -2759,17 +2733,17 @@
req = &s->req;
res = &s->res;
- csf = objt_cs(si_f->end);
+ csf = si_f->cs;
cof = cs_conn(csf);
- acf = objt_appctx(si_f->end);
+ acf = cs_appctx(csf);
if (cof && cof->src && addr_to_str(cof->src, pn, sizeof(pn)) >= 0)
src = pn;
else if (acf)
src = acf->applet->name;
- csb = objt_cs(si_b->end);
+ csb = si_b->cs;
cob = cs_conn(csb);
- acb = objt_appctx(si_b->end);
+ acb = cs_appctx(csb);
srv = objt_server(s->target);
if (srv)
dst = srv->id;
@@ -2780,8 +2754,8 @@
"%sstrm=%p,%x src=%s fe=%s be=%s dst=%s%c"
"%stxn=%p,%x txn.req=%s,%x txn.rsp=%s,%x%c"
"%srqf=%x rqa=%x rpf=%x rpa=%x sif=%s,%x sib=%s,%x%c"
- "%saf=%p,%u csf=%p,%x%c"
- "%sab=%p,%u csb=%p,%x%c"
+ "%scsf=%p,%x csb=%p,%x%c"
+ "%saf=%p,%u sab=%p,%u%c"
"%scof=%p,%x:%s(%p)/%s(%p)/%s(%d)%c"
"%scob=%p,%x:%s(%p)/%s(%p)/%s(%d)%c"
"",
@@ -2792,8 +2766,8 @@
pfx, req->flags, req->analysers, res->flags, res->analysers,
si_state_str(si_f->state), si_f->flags,
si_state_str(si_b->state), si_b->flags, eol,
- pfx, acf, acf ? acf->st0 : 0, csf, csf ? csf->flags : 0, eol,
- pfx, acb, acb ? acb->st0 : 0, csb, csb ? csb->flags : 0, eol,
+ pfx, csf, csf ? csf->flags : 0, csb, csb ? csb->flags : 0, eol,
+ pfx, acf, acf ? acf->st0 : 0, acb, acb ? acb->st0 : 0, eol,
pfx, cof, cof ? cof->flags : 0, conn_get_mux_name(cof), cof?cof->ctx:0, conn_get_xprt_name(cof),
cof ? cof->xprt_ctx : 0, conn_get_ctrl_name(cof), cof ? cof->handle.fd : 0, eol,
pfx, cob, cob ? cob->flags : 0, conn_get_mux_name(cob), cob?cob->ctx:0, conn_get_xprt_name(cob),
@@ -3133,7 +3107,7 @@
*/
static int stats_dump_full_strm_to_buffer(struct stream_interface *si, struct stream *strm)
{
- struct appctx *appctx = __objt_appctx(si->end);
+ struct appctx *appctx = cs_appctx(si->cs);
struct tm tm;
extern const char *monthname[12];
char pn[INET6_ADDRSTRLEN];
@@ -3217,9 +3191,7 @@
else
chunk_appendf(&trash, " backend=<NONE> (id=-1 mode=-)");
- cs = objt_cs(strm->si[1].end);
- conn = cs_conn(cs);
-
+ conn = cs_conn(strm->si[1].cs);
switch (conn && conn_get_src(conn) ? addr_to_str(conn->src, pn, sizeof(pn)) : AF_UNSPEC) {
case AF_INET:
case AF_INET6:
@@ -3286,8 +3258,8 @@
&strm->si[0],
si_state_str(strm->si[0].state),
strm->si[0].flags,
- obj_type_name(strm->si[0].end),
- obj_base_ptr(strm->si[0].end),
+ obj_type_name(strm->si[0].cs->end),
+ obj_base_ptr(strm->si[0].cs->end),
strm->si[0].exp ?
tick_is_expired(strm->si[0].exp, now_ms) ? "<PAST>" :
human_time(TICKS_TO_MS(strm->si[0].exp - now_ms),
@@ -3299,20 +3271,20 @@
&strm->si[1],
si_state_str(strm->si[1].state),
strm->si[1].flags,
- obj_type_name(strm->si[1].end),
- obj_base_ptr(strm->si[1].end),
+ obj_type_name(strm->si[1].cs->end),
+ obj_base_ptr(strm->si[1].cs->end),
strm->si[1].exp ?
tick_is_expired(strm->si[1].exp, now_ms) ? "<PAST>" :
human_time(TICKS_TO_MS(strm->si[1].exp - now_ms),
TICKS_TO_MS(1000)) : "<NEVER>",
strm->si[1].err_type, strm->si[1].wait_event.events);
- if (cs_conn(objt_cs(strm->si[0].end)) != NULL) {
- cs = __objt_cs(strm->si[0].end);
- conn = cs_conn(cs);
+ cs = strm->si[0].cs;
+ chunk_appendf(&trash, " cs=%p csf=0x%08x ctx=%p\n", cs, cs->flags, cs->ctx);
+ if ((conn = cs_conn(cs)) != NULL) {
chunk_appendf(&trash,
- " co0=%p ctrl=%s xprt=%s mux=%s data=%s target=%s:%p\n",
+ " co0=%p ctrl=%s xprt=%s mux=%s data=%s target=%s:%p\n",
conn,
conn_get_ctrl_name(conn),
conn_get_xprt_name(conn),
@@ -3329,11 +3301,10 @@
conn->handle.fd >= 0 ? !!(fdtab[conn->handle.fd].update_mask & tid_bit) : 0,
conn->handle.fd >= 0 ? fdtab[conn->handle.fd].thread_mask: 0);
- chunk_appendf(&trash, " cs=%p csf=0x%08x ctx=%p\n", cs, cs->flags, cs->ctx);
}
- else if ((tmpctx = objt_appctx(strm->si[0].end)) != NULL) {
+ else if ((tmpctx = cs_appctx(cs)) != NULL) {
chunk_appendf(&trash,
- " app0=%p st0=%d st1=%d st2=%d applet=%s tmask=0x%lx nice=%d calls=%u rate=%u cpu=%llu lat=%llu\n",
+ " app0=%p st0=%d st1=%d st2=%d applet=%s tmask=0x%lx nice=%d calls=%u rate=%u cpu=%llu lat=%llu\n",
tmpctx,
tmpctx->st0,
tmpctx->st1,
@@ -3344,12 +3315,11 @@
(unsigned long long)tmpctx->t->cpu_time, (unsigned long long)tmpctx->t->lat_time);
}
- if (cs_conn(objt_cs(strm->si[1].end)) != NULL) {
- cs = __objt_cs(strm->si[1].end);
- conn = cs_conn(cs);
-
+ cs = strm->si[1].cs;
+ chunk_appendf(&trash, " cs=%p csf=0x%08x ctx=%p\n", cs, cs->flags, cs->ctx);
+ if ((conn = cs_conn(cs)) != NULL) {
chunk_appendf(&trash,
- " co1=%p ctrl=%s xprt=%s mux=%s data=%s target=%s:%p\n",
+ " co1=%p ctrl=%s xprt=%s mux=%s data=%s target=%s:%p\n",
conn,
conn_get_ctrl_name(conn),
conn_get_xprt_name(conn),
@@ -3366,11 +3336,10 @@
conn->handle.fd >= 0 ? !!(fdtab[conn->handle.fd].update_mask & tid_bit) : 0,
conn->handle.fd >= 0 ? fdtab[conn->handle.fd].thread_mask: 0);
- chunk_appendf(&trash, " cs=%p csf=0x%08x ctx=%p\n", cs, cs->flags, cs->ctx);
}
- else if ((tmpctx = objt_appctx(strm->si[1].end)) != NULL) {
+ else if ((tmpctx = cs_appctx(cs)) != NULL) {
chunk_appendf(&trash,
- " app1=%p st0=%d st1=%d st2=%d applet=%s tmask=0x%lx nice=%d calls=%u rate=%u cpu=%llu lat=%llu\n",
+ " app1=%p st0=%d st1=%d st2=%d applet=%s tmask=0x%lx nice=%d calls=%u rate=%u cpu=%llu lat=%llu\n",
tmpctx,
tmpctx->st0,
tmpctx->st1,
@@ -3680,7 +3649,7 @@
human_time(TICKS_TO_MS(curr_strm->res.analyse_exp - now_ms),
TICKS_TO_MS(1000)) : "");
- conn = cs_conn(objt_cs(curr_strm->si[0].end));
+ conn = cs_conn(curr_strm->si[0].cs);
chunk_appendf(&trash,
" s0=[%d,%1xh,fd=%d,ex=%s]",
curr_strm->si[0].state,
@@ -3690,7 +3659,7 @@
human_time(TICKS_TO_MS(curr_strm->si[0].exp - now_ms),
TICKS_TO_MS(1000)) : "");
- conn = cs_conn(objt_cs(curr_strm->si[1].end));
+ conn = cs_conn(curr_strm->si[1].cs);
chunk_appendf(&trash,
" s1=[%d,%1xh,fd=%d,ex=%s]",
curr_strm->si[1].state,
diff --git a/src/stream_interface.c b/src/stream_interface.c
index 2edef2b..ae07404 100644
--- a/src/stream_interface.c
+++ b/src/stream_interface.c
@@ -306,13 +306,14 @@
DPRINTF(stderr, "registering handler %p for si %p (was %p)\n", app, si, si_task(si));
- appctx = si_alloc_appctx(si, app);
+ appctx = appctx_new(app);
if (!appctx)
return NULL;
-
+ si_attach_appctx(si, appctx);
+ appctx->t->nice = si_strm(si)->task->nice;
si_cant_get(si);
appctx_wakeup(appctx);
- return si_appctx(si);
+ return appctx;
}
/* This callback is used to send a valid PROXY protocol line to a socket being
@@ -358,7 +359,7 @@
ret = make_proxy_line(trash.area, trash.size,
objt_server(conn->target),
- cs_conn(objt_cs(si_opposite(si)->end)),
+ cs_conn(si_opposite(si)->cs),
strm);
}
else {
@@ -433,7 +434,7 @@
/* process consumer side */
if (channel_is_empty(oc)) {
- struct connection *conn = cs_conn(objt_cs(si->end));
+ struct connection *conn = cs_conn(si->cs);
if (((oc->flags & (CF_SHUTW|CF_SHUTW_NOW)) == CF_SHUTW_NOW) &&
(si->state == SI_ST_EST) && (!conn || !(conn->flags & (CO_FL_WAIT_XPRT | CO_FL_EARLY_SSL_HS))))
@@ -800,7 +801,7 @@
struct task *si_cs_io_cb(struct task *t, void *ctx, unsigned int state)
{
struct stream_interface *si = ctx;
- struct conn_stream *cs = objt_cs(si->end);
+ struct conn_stream *cs = si->cs;
int ret = 0;
if (!cs_conn(cs))
@@ -916,7 +917,6 @@
void si_sync_send(struct stream_interface *si)
{
struct channel *oc = si_oc(si);
- struct conn_stream *cs;
oc->flags &= ~(CF_WRITE_NULL|CF_WRITE_PARTIAL);
@@ -929,11 +929,10 @@
if (!si_state_in(si->state, SI_SB_CON|SI_SB_RDY|SI_SB_EST))
return;
- cs = objt_cs(si->end);
- if (!cs_conn_mux(cs))
+ if (!cs_conn_mux(si->cs))
return;
- si_cs_send(cs);
+ si_cs_send(si->cs);
}
/* Updates at once the channel flags, and timers of both stream interfaces of a
@@ -964,15 +963,15 @@
/* stream ints are processed outside of process_stream() and must be
* handled at the latest moment.
*/
- if (obj_type(si_f->end) == OBJ_TYPE_APPCTX &&
+ if (cs_appctx(si_f->cs) &&
((si_rx_endp_ready(si_f) && !si_rx_blocked(si_f)) ||
(si_tx_endp_ready(si_f) && !si_tx_blocked(si_f))))
- appctx_wakeup(si_appctx(si_f));
+ appctx_wakeup(cs_appctx(si_f->cs));
- if (obj_type(si_b->end) == OBJ_TYPE_APPCTX &&
+ if (cs_appctx(si_b->cs) &&
((si_rx_endp_ready(si_b) && !si_rx_blocked(si_b)) ||
(si_tx_endp_ready(si_b) && !si_tx_blocked(si_b))))
- appctx_wakeup(si_appctx(si_b));
+ appctx_wakeup(cs_appctx(si_b->cs));
}
/*
@@ -987,9 +986,11 @@
*/
static void stream_int_shutr_conn(struct stream_interface *si)
{
- struct conn_stream *cs = __objt_cs(si->end);
+ struct conn_stream *cs = si->cs;
struct channel *ic = si_ic(si);
+ BUG_ON(!cs_conn(cs));
+
si_rx_shut_blk(si);
if (ic->flags & CF_SHUTR)
return;
@@ -1023,10 +1024,12 @@
*/
static void stream_int_shutw_conn(struct stream_interface *si)
{
- struct conn_stream *cs = __objt_cs(si->end);
+ struct conn_stream *cs = si->cs;
struct channel *ic = si_ic(si);
struct channel *oc = si_oc(si);
+ BUG_ON(!cs_conn(cs));
+
oc->flags &= ~CF_SHUTW_NOW;
if (oc->flags & CF_SHUTW)
return;
@@ -1120,7 +1123,7 @@
static void stream_int_chk_snd_conn(struct stream_interface *si)
{
struct channel *oc = si_oc(si);
- struct conn_stream *cs = __objt_cs(si->end);
+ struct conn_stream *cs = si->cs;
struct connection *conn = cs_conn(cs);
BUG_ON(!conn);
@@ -1551,10 +1554,12 @@
*/
static void stream_int_read0(struct stream_interface *si)
{
- struct conn_stream *cs = __objt_cs(si->end);
+ struct conn_stream *cs = si->cs;
struct channel *ic = si_ic(si);
struct channel *oc = si_oc(si);
+ BUG_ON(!cs_conn(cs));
+
si_rx_shut_blk(si);
if (ic->flags & CF_SHUTR)
return;
@@ -1624,7 +1629,7 @@
*/
if ((si_rx_endp_ready(si) && !si_rx_blocked(si)) ||
(si_tx_endp_ready(si) && !si_tx_blocked(si)))
- appctx_wakeup(si_appctx(si));
+ appctx_wakeup(cs_appctx(si->cs));
}
/*
@@ -1686,7 +1691,7 @@
}
/* on shutw we always wake the applet up */
- appctx_wakeup(si_appctx(si));
+ appctx_wakeup(cs_appctx(si->cs));
switch (si->state) {
case SI_ST_RDY:
@@ -1730,7 +1735,7 @@
if (!ic->pipe) {
/* (re)start reading */
- appctx_wakeup(si_appctx(si));
+ appctx_wakeup(cs_appctx(si->cs));
}
}
@@ -1756,7 +1761,7 @@
if (!channel_is_empty(oc)) {
/* (re)start sending */
- appctx_wakeup(si_appctx(si));
+ appctx_wakeup(cs_appctx(si->cs));
}
}
diff --git a/src/tcp_sample.c b/src/tcp_sample.c
index 19edcd2..d98f7c9 100644
--- a/src/tcp_sample.c
+++ b/src/tcp_sample.c
@@ -54,7 +54,7 @@
if (kw[0] == 'b') { /* bc_src */
struct connection *conn = ((obj_type(smp->sess->origin) == OBJ_TYPE_CHECK)
? cs_conn(__objt_check(smp->sess->origin)->cs)
- : (smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)): NULL));
+ : (smp->strm ? cs_conn(smp->strm->si[1].cs): NULL));
if (conn && conn_get_src(conn))
src = conn_src(conn);
}
@@ -98,7 +98,7 @@
if (kw[0] == 'b') { /* bc_src_port */
struct connection *conn = ((obj_type(smp->sess->origin) == OBJ_TYPE_CHECK)
? cs_conn(__objt_check(smp->sess->origin)->cs)
- : (smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)): NULL));
+ : (smp->strm ? cs_conn(smp->strm->si[1].cs): NULL));
if (conn && conn_get_src(conn))
src = conn_src(conn);
}
@@ -133,7 +133,7 @@
if (kw[0] == 'b') { /* bc_dst */
struct connection *conn = ((obj_type(smp->sess->origin) == OBJ_TYPE_CHECK)
? cs_conn(__objt_check(smp->sess->origin)->cs)
- : (smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)): NULL));
+ : (smp->strm ? cs_conn(smp->strm->si[1].cs): NULL));
if (conn && conn_get_dst(conn))
dst = conn_dst(conn);
}
@@ -229,7 +229,7 @@
if (kw[0] == 'b') { /* bc_dst_port */
struct connection *conn = ((obj_type(smp->sess->origin) == OBJ_TYPE_CHECK)
? cs_conn(__objt_check(smp->sess->origin)->cs)
- : (smp->strm ? cs_conn(objt_cs(smp->strm->si[1].end)): NULL));
+ : (smp->strm ? cs_conn(smp->strm->si[1].cs): NULL));
if (conn && conn_get_dst(conn))
dst = conn_dst(conn);
}
@@ -325,7 +325,7 @@
/* get the object associated with the stream interface.The
* object can be other thing than a connection. For example,
* it be a appctx. */
- conn = cs_conn(objt_cs(smp->strm->si[dir].end));
+ conn = cs_conn(smp->strm->si[dir].cs);
if (!conn)
return 0;