MEDIUM: conn_stream: Add a pointer to the app object into the conn-stream
In the same way the conn-stream has a pointer to the stream endpoint , this
patch adds a pointer to the application entity in the conn-stream
structure. For now, it is a stream or a health-check. It is mandatory to
merge the stream-interface with the conn-stream.
diff --git a/include/haproxy/conn_stream-t.h b/include/haproxy/conn_stream-t.h
index fcf893f..88775e9 100644
--- a/include/haproxy/conn_stream-t.h
+++ b/include/haproxy/conn_stream-t.h
@@ -91,6 +91,7 @@
/* 3 bytes hole here */
unsigned int flags; /* CS_FL_* */
enum obj_type *end; /* points to the end point (connection or appctx) */
+ enum obj_type *app; /* points to the applicative point (stream or check) */
void *data; /* pointer to upper layer's entity (eg: stream interface) */
const struct data_cb *data_cb; /* data layer callbacks. Must be set before xprt->init() */
void *ctx; /* mux-specific context */
diff --git a/include/haproxy/conn_stream.h b/include/haproxy/conn_stream.h
index c7f736d..7a7df2a 100644
--- a/include/haproxy/conn_stream.h
+++ b/include/haproxy/conn_stream.h
@@ -29,33 +29,27 @@
#include <haproxy/obj_type.h>
#include <haproxy/pool-t.h>
+struct stream;
+struct check;
+
extern struct pool_head *pool_head_connstream;
#define IS_HTX_CS(cs) (cs_conn(cs) && IS_HTX_CONN(cs_conn(cs)))
-struct conn_stream *cs_new(enum obj_type *endp);
+struct conn_stream *cs_new(enum obj_type *endp, void *ctx, enum obj_type *app, void *data, const struct data_cb *data_cb);
void cs_free(struct conn_stream *cs);
+
/*
* Initializes all required fields for a new conn_strema.
*/
-static inline void cs_init(struct conn_stream *cs, enum obj_type *endp)
+static inline void cs_init(struct conn_stream *cs)
{
- struct connection *conn = objt_conn(endp);
- struct appctx *appctx = objt_appctx(endp);
-
cs->obj_type = OBJ_TYPE_CS;
cs->flags = CS_FL_NONE;
- cs->end = endp;
- if (conn) {
- cs->ctx = conn;
- if (!conn->ctx)
- conn->ctx = cs;
- }
- else if (appctx) {
- cs->ctx = appctx;
- /* appctx->owner must be set by the caller for now */
- }
+ cs->end = NULL;
+ cs->app = NULL;
+ cs->ctx = NULL;
cs->data = NULL;
cs->data_cb = NULL;
}
@@ -88,14 +82,32 @@
static inline struct stream_interface *cs_si(const struct conn_stream *cs)
{
- return ((cs_conn(cs) || cs_appctx(cs)) ? cs->data : NULL);
+ return (cs ? cs->data : NULL);
}
-/* Attaches a conn_stream to a data layer and sets the relevant callbacks */
-static inline void cs_attach(struct conn_stream *cs, void *data, const struct data_cb *data_cb)
+static inline struct stream *cs_strm(const struct conn_stream *cs)
{
- cs->data_cb = data_cb;
+ return (cs ? objt_stream(cs->app) : NULL);
+}
+
+static inline struct check *cs_check(const struct conn_stream *cs)
+{
+ return (cs ? objt_check(cs->app) : NULL);
+}
+
+/* Attaches a conn_stream to an endpoint and sets the endpoint ctx */
+static inline void cs_attach_endp(struct conn_stream *cs, enum obj_type *endp, void *ctx)
+{
+ cs->end = endp;
+ cs->ctx = ctx;
+}
+
+/* Attaches a conn_stream to a app layer and sets the relevant callbacks */
+static inline void cs_attach_app(struct conn_stream *cs, enum obj_type *app, void *data, const struct data_cb *data_cb)
+{
+ cs->app = app;
cs->data = data;
+ cs->data_cb = data_cb;
}
/* Detach the conn_stream from the connection, if any. If a mux owns the
@@ -127,7 +139,7 @@
appctx->applet->release(appctx);
appctx_free(appctx);
}
- cs_init(cs, NULL);
+ cs_init(cs);
}
/* Release a conn_stream */
diff --git a/include/haproxy/stream_interface.h b/include/haproxy/stream_interface.h
index 314f54c..6da05bc 100644
--- a/include/haproxy/stream_interface.h
+++ b/include/haproxy/stream_interface.h
@@ -187,18 +187,18 @@
si->cs = cs;
if (cs_conn(cs)) {
si->ops = &si_conn_ops;
- cs_attach(cs, si, &si_conn_cb);
+ cs_attach_app(cs, &si_strm(si)->obj_type, si, &si_conn_cb);
}
else if (cs_appctx(cs)) {
struct appctx *appctx = cs_appctx(cs);
si->ops = &si_applet_ops;
appctx->owner = cs;
- cs_attach(cs, si, NULL);
+ cs_attach_app(cs, &si_strm(si)->obj_type, si, NULL);
}
else {
si->ops = &si_embedded_ops;
- cs_attach(cs, si, NULL);
+ cs_attach_app(cs, &si_strm(si)->obj_type, si, NULL);
}
}
@@ -208,10 +208,11 @@
static inline void si_attach_conn(struct stream_interface *si, struct connection *conn)
{
si_reset_endpoint(si);
- cs_init(si->cs, &conn->obj_type);
if (!conn->ctx)
conn->ctx = si->cs;
- si_attach_cs(si, si->cs);
+ si->ops = &si_conn_ops;
+ cs_attach_endp(si->cs, &conn->obj_type, conn);
+ cs_attach_app(si->cs, &si_strm(si)->obj_type, si, &si_conn_cb);
}
/* Attach appctx <appctx> to the stream interface <si>. The stream interface
@@ -220,9 +221,10 @@
static inline void si_attach_appctx(struct stream_interface *si, struct appctx *appctx)
{
si_reset_endpoint(si);
- cs_init(si->cs, &appctx->obj_type);
appctx->owner = si->cs;
- si_attach_cs(si, si->cs);
+ si->ops = &si_applet_ops;
+ cs_attach_endp(si->cs, &appctx->obj_type, appctx);
+ cs_attach_app(si->cs, &si_strm(si)->obj_type, si, NULL);
}
/* call the applet's release function if any. Needs to be called upon close() */
diff --git a/src/check.c b/src/check.c
index 97d340a..cbdf620 100644
--- a/src/check.c
+++ b/src/check.c
@@ -1019,7 +1019,7 @@
static int wake_srv_chk(struct conn_stream *cs)
{
struct connection *conn;
- struct check *check = cs->data;
+ struct check *check = cs_check(cs);
struct email_alertq *q = container_of(check, typeof(*q), check);
int ret = 0;
diff --git a/src/conn_stream.c b/src/conn_stream.c
index 7b53ecc..41d4a2e 100644
--- a/src/conn_stream.c
+++ b/src/conn_stream.c
@@ -22,14 +22,16 @@
/* Tries to allocate a new conn_stream and initialize its main fields. On
* failure, nothing is allocated and NULL is returned.
*/
-struct conn_stream *cs_new(enum obj_type *endp)
+struct conn_stream *cs_new(enum obj_type *endp, void *ctx, enum obj_type *app, void *data, const struct data_cb *data_cb)
{
struct conn_stream *cs;
cs = pool_alloc(pool_head_connstream);
if (unlikely(!cs))
return NULL;
- cs_init(cs, endp);
+ cs_init(cs);
+ cs_attach_endp(cs, endp, ctx);
+ cs_attach_app(cs, app, data, data_cb);
return cs;
}
diff --git a/src/dns.c b/src/dns.c
index 719aed6..ed2f875 100644
--- a/src/dns.c
+++ b/src/dns.c
@@ -903,7 +903,7 @@
goto out_free_appctx;
}
- cs = cs_new(&appctx->obj_type);
+ cs = cs_new(&appctx->obj_type, appctx, NULL, NULL, NULL);
if (!cs) {
ha_alert("out of memory in dns_session_create().\n");
goto out_free_sess;
diff --git a/src/flt_spoe.c b/src/flt_spoe.c
index a12d75d..6c4c9f8 100644
--- a/src/flt_spoe.c
+++ b/src/flt_spoe.c
@@ -2024,7 +2024,7 @@
if (!sess)
goto out_free_spoe;
- cs = cs_new(&appctx->obj_type);
+ cs = cs_new(&appctx->obj_type, appctx, NULL, NULL, NULL);
if (!cs)
goto out_free_sess;
diff --git a/src/hlua.c b/src/hlua.c
index 35ed2e5..219cd26 100644
--- a/src/hlua.c
+++ b/src/hlua.c
@@ -2961,7 +2961,7 @@
goto out_fail_appctx;
}
- cs = cs_new(&appctx->obj_type);
+ cs = cs_new(&appctx->obj_type, appctx, NULL, NULL, NULL);
if (!cs) {
hlua_pusherror(L, "socket: out of memory");
goto out_fail_sess;
diff --git a/src/http_client.c b/src/http_client.c
index 0149eeb..6b060a1 100644
--- a/src/http_client.c
+++ b/src/http_client.c
@@ -486,7 +486,7 @@
ha_alert("httpclient: out of memory in %s:%d.\n", __FUNCTION__, __LINE__);
goto out_free_appctx;
}
- cs = cs_new(&appctx->obj_type);
+ cs = cs_new(&appctx->obj_type, appctx, NULL, NULL, NULL);
if (!cs) {
ha_alert("httpclient: out of memory in %s:%d.\n", __FUNCTION__, __LINE__);
goto out_free_sess;
diff --git a/src/mux_fcgi.c b/src/mux_fcgi.c
index 3a4f7c0..d14a166 100644
--- a/src/mux_fcgi.c
+++ b/src/mux_fcgi.c
@@ -1230,8 +1230,8 @@
struct fcgi_strm_params *params)
{
struct connection *cli_conn = objt_conn(fstrm->sess->origin);
- const struct sockaddr_storage *src = si_src(si_opposite(fstrm->cs->data));
- const struct sockaddr_storage *dst = si_dst(si_opposite(fstrm->cs->data));
+ const struct sockaddr_storage *src = si_src(si_opposite(cs_si(fstrm->cs)));
+ const struct sockaddr_storage *dst = si_dst(si_opposite(cs_si(fstrm->cs)));
struct ist p;
if (!sl)
@@ -3312,11 +3312,11 @@
struct proxy *other_end;
union error_snapshot_ctx ctx;
- if (fstrm->cs && fstrm->cs->data) {
+ if (fstrm->cs && cs_strm(fstrm->cs)) {
if (sess == NULL)
- sess = si_strm(fstrm->cs->data)->sess;
+ sess = cs_strm(fstrm->cs)->sess;
if (!(h1m->flags & H1_MF_RESP))
- other_end = si_strm(fstrm->cs->data)->be;
+ other_end = cs_strm(fstrm->cs)->be;
else
other_end = sess->fe;
} else
@@ -4189,8 +4189,8 @@
(unsigned int)b_head_ofs(&fstrm->rxbuf), (unsigned int)b_size(&fstrm->rxbuf),
fstrm->cs);
if (fstrm->cs)
- chunk_appendf(msg, " .cs.flg=0x%08x .cs.data=%p",
- fstrm->cs->flags, fstrm->cs->data);
+ chunk_appendf(msg, " .cs.flg=0x%08x .cs.app=%p",
+ fstrm->cs->flags, fstrm->cs->app);
chunk_appendf(&trash, " .subs=%p", fstrm->subs);
if (fstrm->subs) {
chunk_appendf(&trash, "(ev=%d tl=%p", fstrm->subs->events, fstrm->subs->tasklet);
diff --git a/src/mux_h1.c b/src/mux_h1.c
index 5fdcd46..48ba0a3 100644
--- a/src/mux_h1.c
+++ b/src/mux_h1.c
@@ -682,13 +682,12 @@
struct conn_stream *cs;
TRACE_ENTER(H1_EV_STRM_NEW, h1c->conn, h1s);
- cs = cs_new(&h1c->conn->obj_type);
+ cs = cs_new(&h1c->conn->obj_type, h1s, NULL, NULL, NULL);
if (!cs) {
TRACE_ERROR("CS allocation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, h1c->conn, h1s);
goto err;
}
h1s->cs = cs;
- cs->ctx = h1s;
if (h1s->flags & H1S_F_NOT_FIRST)
cs->flags |= CS_FL_NOT_FIRST;
@@ -1351,11 +1350,11 @@
struct proxy *other_end;
union error_snapshot_ctx ctx;
- if ((h1c->flags & H1C_F_ST_ATTACHED) && h1s->cs->data) {
+ if ((h1c->flags & H1C_F_ST_ATTACHED) && cs_strm(h1s->cs)) {
if (sess == NULL)
- sess = si_strm(h1s->cs->data)->sess;
+ sess = cs_strm(h1s->cs)->sess;
if (!(h1m->flags & H1_MF_RESP))
- other_end = si_strm(h1s->cs->data)->be;
+ other_end = cs_strm(h1s->cs)->be;
else
other_end = sess->fe;
} else
@@ -3828,8 +3827,8 @@
h1m_state_str(h1s->req.state),
h1m_state_str(h1s->res.state), method, h1s->status);
if (h1s->cs)
- chunk_appendf(msg, " .cs.flg=0x%08x .cs.data=%p",
- h1s->cs->flags, h1s->cs->data);
+ chunk_appendf(msg, " .cs.flg=0x%08x .cs.app=%p",
+ h1s->cs->flags, h1s->cs->app);
chunk_appendf(&trash, " .subs=%p", h1s->subs);
if (h1s->subs) {
diff --git a/src/mux_h2.c b/src/mux_h2.c
index 3a98e5d..2049b51 100644
--- a/src/mux_h2.c
+++ b/src/mux_h2.c
@@ -1529,13 +1529,12 @@
if (!h2s)
goto out;
- cs = cs_new(&h2c->conn->obj_type);
+ cs = cs_new(&h2c->conn->obj_type, h2s, NULL, NULL, NULL);
if (!cs)
goto out_close;
cs->flags |= CS_FL_NOT_FIRST;
h2s->cs = cs;
- cs->ctx = h2s;
h2c->nb_cs++;
/* FIXME wrong analogy between ext-connect and websocket, this need to
@@ -6652,8 +6651,8 @@
(unsigned int)b_head_ofs(&h2s->rxbuf), (unsigned int)b_size(&h2s->rxbuf),
h2s->cs);
if (h2s->cs)
- chunk_appendf(msg, "(.flg=0x%08x .data=%p)",
- h2s->cs->flags, h2s->cs->data);
+ chunk_appendf(msg, "(.flg=0x%08x .app=%p)",
+ h2s->cs->flags, h2s->cs->app);
chunk_appendf(&trash, " .subs=%p", h2s->subs);
if (h2s->subs) {
diff --git a/src/mux_pt.c b/src/mux_pt.c
index 13e60d8..7ba9da9 100644
--- a/src/mux_pt.c
+++ b/src/mux_pt.c
@@ -291,7 +291,7 @@
ctx->conn = conn;
if (!cs) {
- cs = cs_new(&conn->obj_type);
+ cs = cs_new(&conn->obj_type, NULL, NULL, NULL, NULL);
if (!cs) {
TRACE_ERROR("CS allocation failure", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn);
goto fail_free_ctx;
diff --git a/src/peers.c b/src/peers.c
index 1b80452..76a0917 100644
--- a/src/peers.c
+++ b/src/peers.c
@@ -3204,7 +3204,7 @@
goto out_free_appctx;
}
- cs = cs_new(&appctx->obj_type);
+ cs = cs_new(&appctx->obj_type, appctx, NULL, NULL, NULL);
if (!cs) {
ha_alert("out of memory in peer_session_create().\n");
goto out_free_sess;
diff --git a/src/sink.c b/src/sink.c
index a3464ab..b09b895 100644
--- a/src/sink.c
+++ b/src/sink.c
@@ -655,7 +655,7 @@
goto out_free_appctx;
}
- cs = cs_new(&appctx->obj_type);
+ cs = cs_new(&appctx->obj_type, appctx, NULL, NULL, NULL);
if (!cs) {
ha_alert("out of memory in sink_forward_session_create");
goto out_free_sess;
diff --git a/src/stream.c b/src/stream.c
index dd6ae6f..f018c37 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -275,8 +275,7 @@
*/
int stream_upgrade_from_cs(struct conn_stream *cs, struct buffer *input)
{
- struct stream_interface *si = cs->data;
- struct stream *s = si_strm(si);
+ struct stream *s = cs_strm(cs);
if (cs_conn_mux(cs)) {
const struct mux_ops *mux = cs_conn_mux(cs);
@@ -475,7 +474,7 @@
if (likely(sess->fe->options2 & PR_O2_INDEPSTR))
s->si[1].flags |= SI_FL_INDEP_STR;
- s->si[1].cs = cs_new(NULL);
+ s->si[1].cs = cs_new(NULL, NULL, &s->obj_type, &s->si[1], NULL);
if (!s->si[1].cs)
goto out_fail_alloc_cs;
diff --git a/src/stream_interface.c b/src/stream_interface.c
index ae07404..5b86bf8 100644
--- a/src/stream_interface.c
+++ b/src/stream_interface.c
@@ -354,12 +354,11 @@
*/
if (cs && cs->data_cb == &si_conn_cb) {
- struct stream_interface *si = cs->data;
- struct stream *strm = si_strm(si);
+ struct stream *strm = cs_strm(cs);
ret = make_proxy_line(trash.area, trash.size,
objt_server(conn->target),
- cs_conn(si_opposite(si)->cs),
+ cs_conn(si_opposite(cs_si(cs))->cs),
strm);
}
else {
@@ -564,7 +563,7 @@
static int si_cs_process(struct conn_stream *cs)
{
struct connection *conn = cs_conn(cs);
- struct stream_interface *si = cs->data;
+ struct stream_interface *si = cs_si(cs);
struct channel *ic = si_ic(si);
struct channel *oc = si_oc(si);
@@ -652,7 +651,7 @@
int si_cs_send(struct conn_stream *cs)
{
struct connection *conn = cs_conn(cs);
- struct stream_interface *si = cs->data;
+ struct stream_interface *si = cs_si(cs);
struct channel *oc = si_oc(si);
int ret;
int did_send = 0;
@@ -1220,7 +1219,7 @@
int si_cs_recv(struct conn_stream *cs)
{
struct connection *conn = cs_conn(cs);
- struct stream_interface *si = cs->data;
+ struct stream_interface *si = cs_si(cs);
struct channel *ic = si_ic(si);
int ret, max, cur_read = 0;
int read_poll = MAX_READ_POLL_LOOPS;
diff --git a/src/tcpcheck.c b/src/tcpcheck.c
index 5fec53e..3697f41 100644
--- a/src/tcpcheck.c
+++ b/src/tcpcheck.c
@@ -1093,7 +1093,7 @@
/* No connection, prepare a new one */
conn = conn_new((s ? &s->obj_type : &proxy->obj_type));
if (conn)
- cs = cs_new(&conn->obj_type);
+ cs = cs_new(&conn->obj_type, conn, &check->obj_type, NULL, &check_conn_cb);
if (!conn || !cs) {
chunk_printf(&trash, "TCPCHK error allocating connection at step %d",
tcpcheck_get_step_id(check, rule));
@@ -1166,8 +1166,6 @@
goto fail_check;
}
- cs_attach(cs, check, &check_conn_cb);
-
if ((connect->options & TCPCHK_OPT_SOCKS4) && s && (s->flags & SRV_F_SOCKS4_PROXY)) {
conn->send_proxy_ofs = 1;
conn->flags |= CO_FL_SOCKS4;