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.

include/haproxy/conn_stream-t.h

@@ -91,6 +91,7 @@ struct conn_stream {
 	/* 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 */

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;
+	cs->end = NULL;
-	if (conn) {
+	cs->app = NULL;
-		cs->ctx = conn;
+	cs->ctx = NULL;
-		if (!conn->ctx)
-			conn->ctx = cs;
-	}
-	else if (appctx) {
-		cs->ctx = appctx;
-		/* appctx->owner must be set by the caller for now */
-	}
 	cs->data = NULL;
 	cs->data_cb = NULL;
 }
@@ -88,14 +82,32 @@ static inline struct appctx *cs_appctx(const struct conn_stream *cs)
 
 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 struct stream *cs_strm(const struct conn_stream *cs)
-static inline void cs_attach(struct conn_stream *cs, void *data, const struct data_cb *data_cb)
 {
-	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 @@ static inline void cs_detach(struct conn_stream *cs)
 			appctx->applet->release(appctx);
 		appctx_free(appctx);
 	}
-	cs_init(cs, NULL);
+	cs_init(cs);
 }
 
 /* Release a conn_stream */

include/haproxy/stream_interface.h

@@ -187,18 +187,18 @@ static inline void si_attach_cs(struct stream_interface *si, struct conn_stream
 	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_cs(struct stream_interface *si, struct conn_stream
 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_conn(struct stream_interface *si, struct connection
 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() */

src/check.c

@@ -1019,7 +1019,7 @@ int httpchk_build_status_header(struct server *s, struct buffer *buf)
 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;
 

src/conn_stream.c

@@ -22,14 +22,16 @@ DECLARE_POOL(pool_head_connstream, "conn_stream", sizeof(struct conn_stream));
 /* 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;
 }
 

src/dns.c

@@ -903,7 +903,7 @@ static struct appctx *dns_session_create(struct dns_session *ds)
 		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;

src/flt_spoe.c

@@ -2024,7 +2024,7 @@ spoe_create_appctx(struct spoe_config *conf)
 	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;
 

src/hlua.c

@@ -2961,7 +2961,7 @@ __LJMP static int hlua_socket_new(lua_State *L)
 		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;

src/http_client.c

@@ -486,7 +486,7 @@ struct appctx *httpclient_start(struct httpclient *hc)
 		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;

src/mux_fcgi.c

@@ -1230,8 +1230,8 @@ static int fcgi_set_default_param(struct fcgi_conn *fconn, struct fcgi_strm *fst
 				  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 *src = si_src(si_opposite(cs_si(fstrm->cs)));
-	const struct sockaddr_storage *dst = si_dst(si_opposite(fstrm->cs->data));
+	const struct sockaddr_storage *dst = si_dst(si_opposite(cs_si(fstrm->cs)));
 	struct ist p;
 
 	if (!sl)
@@ -3312,11 +3312,11 @@ static void fcgi_strm_capture_bad_message(struct fcgi_conn *fconn, struct fcgi_s
 	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 @@ static int fcgi_show_fd(struct buffer *msg, struct connection *conn)
 			      (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",
+			chunk_appendf(msg, " .cs.flg=0x%08x .cs.app=%p",
-				      fstrm->cs->flags, fstrm->cs->data);
+				      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);

src/mux_h1.c

@@ -682,13 +682,12 @@ static struct conn_stream *h1s_new_cs(struct h1s *h1s, struct buffer *input)
 	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 @@ static void h1_capture_bad_message(struct h1c *h1c, struct h1s *h1s,
 	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 @@ static int h1_show_fd(struct buffer *msg, struct connection *conn)
 			      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",
+			chunk_appendf(msg, " .cs.flg=0x%08x .cs.app=%p",
-				      h1s->cs->flags, h1s->cs->data);
+				      h1s->cs->flags, h1s->cs->app);
 
 		chunk_appendf(&trash, " .subs=%p", h1s->subs);
 		if (h1s->subs) {

src/mux_h2.c

@@ -1529,13 +1529,12 @@ static struct h2s *h2c_frt_stream_new(struct h2c *h2c, int id, struct buffer *in
 	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 @@ static int h2_show_fd(struct buffer *msg, struct connection *conn)
 			      (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)",
+			chunk_appendf(msg, "(.flg=0x%08x .app=%p)",
-				      h2s->cs->flags, h2s->cs->data);
+				      h2s->cs->flags, h2s->cs->app);
 
 		chunk_appendf(&trash, " .subs=%p", h2s->subs);
 		if (h2s->subs) {

src/mux_pt.c

@@ -291,7 +291,7 @@ static int mux_pt_init(struct connection *conn, struct proxy *prx, struct sessio
 	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;

src/peers.c

@@ -3204,7 +3204,7 @@ static struct appctx *peer_session_create(struct peers *peers, struct peer *peer
 		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;

src/sink.c

@@ -655,7 +655,7 @@ static struct appctx *sink_forward_session_create(struct sink *sink, struct sink
 		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;

src/stream.c

@@ -275,8 +275,7 @@ static void strm_trace(enum trace_level level, uint64_t mask, const struct trace
  */
 int stream_upgrade_from_cs(struct conn_stream *cs, struct buffer *input)
 {
-	struct stream_interface *si = cs->data;
+	struct stream *s = cs_strm(cs);
-	struct stream *s = si_strm(si);
 
 	if (cs_conn_mux(cs)) {
 		const struct mux_ops *mux = cs_conn_mux(cs);
@@ -475,7 +474,7 @@ struct stream *stream_new(struct session *sess, struct conn_stream *cs, struct b
 	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;
 

src/stream_interface.c

@@ -354,12 +354,11 @@ int conn_si_send_proxy(struct connection *conn, unsigned int flag)
 		 */
 
 		if (cs && cs->data_cb == &si_conn_cb) {
-			struct stream_interface *si = cs->data;
+			struct stream *strm = cs_strm(cs);
-			struct stream *strm = si_strm(si);
 
 			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 void stream_int_notify(struct stream_interface *si)
 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 @@ static int si_cs_process(struct conn_stream *cs)
 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 @@ static void stream_int_chk_snd_conn(struct stream_interface *si)
 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;

src/tcpcheck.c

@@ -1093,7 +1093,7 @@ enum tcpcheck_eval_ret tcpcheck_eval_connect(struct check *check, struct tcpchec
 	/* 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 @@ enum tcpcheck_eval_ret tcpcheck_eval_connect(struct check *check, struct tcpchec
 		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;