MEDIUM: conn-stream: Be prepared to use an appctx as conn-stream endpoint

To be able to use an appctx as conn-stream endpoint, the connection is no
longer stored as is in the conn-stream. The obj-type is used instead.

include/haproxy/conn_stream-t.h

@@ -25,8 +25,6 @@
 
 #include <haproxy/obj_type-t.h>
 
-struct connection;
-
 /* conn_stream flags */
 enum {
 	CS_FL_NONE          = 0x00000000,  /* Just for initialization purposes */
@@ -92,7 +90,7 @@ struct conn_stream {
 	enum obj_type obj_type;              /* differentiates connection from applet context */
 	/* 3 bytes hole here */
 	unsigned int flags;                  /* CS_FL_* */
-	struct connection *conn;             /* xprt-level connection */
+	enum obj_type *end;                  /* points to the end point (connection or appctx) */
 	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

@@ -23,6 +23,7 @@
 #define _HAPROXY_CONN_STREAM_H
 
 #include <haproxy/api.h>
+#include <haproxy/applet.h>
 #include <haproxy/connection.h>
 #include <haproxy/conn_stream-t.h>
 #include <haproxy/obj_type.h>
@@ -32,24 +33,29 @@ 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(struct connection *conn, void *target);
+struct conn_stream *cs_new(enum obj_type *endp);
 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, struct connection *conn)
+static inline void cs_init(struct conn_stream *cs, enum obj_type *endp)
 {
 	cs->obj_type = OBJ_TYPE_CS;
 	cs->flags = CS_FL_NONE;
-	cs->conn = conn;
+	cs->end = endp;
-	cs->ctx = conn;
+	if (objt_conn(endp))
+		cs->ctx = endp;
+	cs->data = NULL;
+	cs->data_cb = NULL;
 }
 
-/* Returns the conn from a cs. If cs is NULL, returns NULL */
+/* Returns the connection from a cs if the endpoint is a connection. Otherwise
+ * NULL is returned.
+ */
 static inline struct connection *cs_conn(const struct conn_stream *cs)
 {
-	return cs ? cs->conn : NULL;
+	return (cs ? objt_conn(cs->end) : NULL);
 }
 
 /* Returns the mux of the connection from a cs if the endpoint is a
@@ -62,6 +68,14 @@ static inline const struct mux_ops *cs_conn_mux(const struct conn_stream *cs)
 	return (conn ? conn->mux : NULL);
 }
 
+/* Returns the appctx from a cs if the endpoint is an appctx. Otherwise NULL is
+ * returned.
+ */
+static inline struct appctx *cs_appctx(const struct conn_stream *cs)
+{
+	return (cs ? objt_appctx(cs->end) : 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)
 {
@@ -77,6 +91,7 @@ static inline void cs_attach(struct conn_stream *cs, void *data, const struct da
 static inline void cs_detach(struct conn_stream *cs)
 {
 	struct connection *conn;
+	struct appctx *appctx;
 
 	if ((conn = cs_conn(cs))) {
 		if (conn->mux)
@@ -92,6 +107,11 @@ static inline void cs_detach(struct conn_stream *cs)
 			conn_free(conn);
 		}
 	}
+	else if ((appctx = cs_appctx(cs))) {
+		if (appctx->applet->release)
+			appctx->applet->release(appctx);
+		appctx_free(appctx);
+	}
 	cs_init(cs, NULL);
 }
 

include/haproxy/stream_interface.h

@@ -240,9 +240,9 @@ static inline struct conn_stream *si_attach_conn(struct stream_interface *si, st
 	si_reset_endpoint(si);
 	cs = objt_cs(si->end);
 	if (!cs)
-		cs = cs_new(conn, conn->target);
+		cs = cs_new(&conn->obj_type);
 	if (cs) {
-		cs_init(cs, conn);
+		cs_init(cs, &conn->obj_type);
 		if (!conn->ctx)
 			conn->ctx = cs;
 		si_attach_cs(si, cs);
@@ -431,7 +431,7 @@ static inline struct conn_stream *si_alloc_cs(struct stream_interface *si, struc
 
 	si_release_endpoint(si);
 
-	cs = cs_new(conn, (conn ? conn->target : NULL));
+	cs = cs_new(&conn->obj_type);
 	if (cs)
 		si_attach_cs(si, cs);
 

src/conn_stream.c

@@ -19,30 +19,17 @@
 DECLARE_POOL(pool_head_connstream, "conn_stream", sizeof(struct conn_stream));
 
 
-/* Tries to allocate a new conn_stream and initialize its main fields. If
+/* Tries to allocate a new conn_stream and initialize its main fields. On
- * <conn> is NULL, then a new connection is allocated on the fly, initialized,
+ * failure, nothing is allocated and NULL is returned.
- * and assigned to cs->conn ; this connection will then have to be released
- * using pool_free() or conn_free(). The conn_stream is initialized and added
- * to the mux's stream list on success, then returned. On failure, nothing is
- * allocated and NULL is returned.
  */
-struct conn_stream *cs_new(struct connection *conn, void *target)
+struct conn_stream *cs_new(enum obj_type *endp)
 {
 	struct conn_stream *cs;
 
 	cs = pool_alloc(pool_head_connstream);
 	if (unlikely(!cs))
 		return NULL;
-
+	cs_init(cs, endp);
-	if (!conn) {
-		conn = conn_new(target);
-		if (unlikely(!conn)) {
-			cs_free(cs);
-			return NULL;
-		}
-	}
-
-	cs_init(cs, conn);
 	return cs;
 }
 

src/h3.c

@@ -176,7 +176,7 @@ static int h3_headers_to_htx(struct qcs *qcs, struct buffer *buf, uint64_t len,
 	if (fin)
 		htx->flags |= HTX_FL_EOM;
 
-	cs = cs_new(qcs->qcc->conn, qcs->qcc->conn->target);
+	cs = cs_new(qcs->qcc->conn->obj_type);
 	if (!cs)
 		return 1;
 

src/hq_interop.c

@@ -72,7 +72,7 @@ static int hq_interop_decode_qcs(struct qcs *qcs, int fin, void *ctx)
 	htx_add_endof(htx, HTX_BLK_EOH);
 	htx_to_buf(htx, &htx_buf);
 
-	cs = cs_new(qcs->qcc->conn, qcs->qcc->conn->target);
+	cs = cs_new(&qcs->qcc->conn->obj_type);
 	if (!cs)
 		return -1;
 

src/mux_h1.c

@@ -678,12 +678,13 @@ static inline size_t h1s_data_pending(const struct h1s *h1s)
  * success or NULL on error. */
 static struct conn_stream *h1s_new_cs(struct h1s *h1s, struct buffer *input)
 {
+	struct h1c *h1c = h1s->h1c;
 	struct conn_stream *cs;
 
-	TRACE_ENTER(H1_EV_STRM_NEW, h1s->h1c->conn, h1s);
+	TRACE_ENTER(H1_EV_STRM_NEW, h1c->conn, h1s);
-	cs = cs_new(h1s->h1c->conn, h1s->h1c->conn->target);
+	cs = cs_new(&h1c->conn->obj_type);
 	if (!cs) {
-		TRACE_ERROR("CS allocation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, h1s->h1c->conn, h1s);
+		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;
@@ -696,21 +697,21 @@ static struct conn_stream *h1s_new_cs(struct h1s *h1s, struct buffer *input)
 		cs->flags |= CS_FL_WEBSOCKET;
 
 	if (stream_create_from_cs(cs, input) < 0) {
-		TRACE_DEVEL("leaving on stream creation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, h1s->h1c->conn, h1s);
+		TRACE_DEVEL("leaving on stream creation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, h1c->conn, h1s);
 		goto err;
 	}
 
-	HA_ATOMIC_INC(&h1s->h1c->px_counters->open_streams);
+	HA_ATOMIC_INC(&h1c->px_counters->open_streams);
-	HA_ATOMIC_INC(&h1s->h1c->px_counters->total_streams);
+	HA_ATOMIC_INC(&h1c->px_counters->total_streams);
 
-	h1s->h1c->flags = (h1s->h1c->flags & ~H1C_F_ST_EMBRYONIC) | H1C_F_ST_ATTACHED | H1C_F_ST_READY;
+	h1c->flags = (h1c->flags & ~H1C_F_ST_EMBRYONIC) | H1C_F_ST_ATTACHED | H1C_F_ST_READY;
-	TRACE_LEAVE(H1_EV_STRM_NEW, h1s->h1c->conn, h1s);
+	TRACE_LEAVE(H1_EV_STRM_NEW, h1c->conn, h1s);
 	return cs;
 
   err:
 	cs_free(cs);
 	h1s->cs = NULL;
-	TRACE_DEVEL("leaving on error", H1_EV_STRM_NEW|H1_EV_STRM_ERR, h1s->h1c->conn, h1s);
+	TRACE_DEVEL("leaving on error", H1_EV_STRM_NEW|H1_EV_STRM_ERR, h1c->conn, h1s);
 	return NULL;
 }
 

src/mux_h2.c

@@ -1529,7 +1529,7 @@ static struct h2s *h2c_frt_stream_new(struct h2c *h2c, int id, struct buffer *in
 	if (!h2s)
 		goto out;
 
-	cs = cs_new(h2c->conn, h2c->conn->target);
+	cs = cs_new(&h2c->conn->obj_type);
 	if (!cs)
 		goto out_close;
 

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, conn->target);
+		cs = cs_new(&conn->obj_type);
 		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/tcpcheck.c

@@ -1091,8 +1091,10 @@ enum tcpcheck_eval_ret tcpcheck_eval_connect(struct check *check, struct tcpchec
 	check_release_buf(check, &check->bo);
 
 	/* No connection, prepare a new one */
-	cs = cs_new(NULL, (s ? &s->obj_type : &proxy->obj_type));
+	conn = conn_new((s ? &s->obj_type : &proxy->obj_type));
-	if (!cs) {
+	if (conn)
+		cs = cs_new(&conn->obj_type);
+	if (!conn || !cs) {
 		chunk_printf(&trash, "TCPCHK error allocating connection at step %d",
 			     tcpcheck_get_step_id(check, rule));
 		if (rule->comment)
@@ -1100,13 +1102,14 @@ enum tcpcheck_eval_ret tcpcheck_eval_connect(struct check *check, struct tcpchec
 		set_server_check_status(check, HCHK_STATUS_SOCKERR, trash.area);
 		ret = TCPCHK_EVAL_STOP;
 		TRACE_ERROR("conn-stream allocation error", CHK_EV_TCPCHK_CONN|CHK_EV_TCPCHK_ERR, check);
+		if (conn)
+			conn_free(conn);
 		goto out;
 	}
 
 	tasklet_set_tid(check->wait_list.tasklet, tid);
 
 	check->cs = cs;
-	conn = cs->conn;
 	conn_set_owner(conn, check->sess, NULL);
 
 	/* Maybe there were an older connection we were waiting on */