MEDIUM: conn-stream: Be able to pass endpoint to create a conn-stream

It is a transient commit to prepare next changes. It is possible to pass a
pre-allocated endpoint to create a new conn-stream. If it is NULL, a new
endpoint is created, otherwise the existing one is used. There no more
change at the conn-stream level.

In the applets, all conn-stream are created with no pre-allocated
endpoint. But for multiplexers, an endpoint is systematically created before
creating the conn-stream.

include/haproxy/conn_stream.h

@@ -37,7 +37,7 @@ struct check;
 struct cs_endpoint *cs_endpoint_new();
 void cs_endpoint_free(struct cs_endpoint *endp);
 
-struct conn_stream *cs_new();
+struct conn_stream *cs_new(struct cs_endpoint *endp);
 void cs_free(struct conn_stream *cs);
 void cs_attach_endp_mux(struct conn_stream *cs, void *endp, void *ctx);
 void cs_attach_endp_app(struct conn_stream *cs, void *endp, void *ctx);

include/haproxy/mux_quic.h

@@ -107,9 +107,19 @@ static inline struct qc_stream_desc *qcc_get_stream(struct qcc *qcc, uint64_t id
 
 static inline struct conn_stream *qc_attach_cs(struct qcs *qcs, struct buffer *buf)
 {
-	struct conn_stream *cs = cs_new();
-	if (!cs)
+	struct cs_endpoint *endp;
+	struct conn_stream *cs;
+
+	endp = cs_endpoint_new();
+	if (!endp)
 		return NULL;
+	endp->target = qcs;
+	endp->ctx = qcs->qcc->conn;
+	cs = cs_new(endp);
+	if (!cs) {
+		cs_endpoint_free(endp);
+		return NULL;
+	}
 	cs_attach_endp_mux(cs, qcs, qcs->qcc->conn);
 	qcs->cs = cs;
 	stream_new(qcs->qcc->conn->owner, cs, buf);

src/check.c

@@ -1391,7 +1391,7 @@ int start_check_task(struct check *check, int mininter,
 	if (check->type == PR_O2_EXT_CHK)
 		t = task_new_on(0);
 	else {
-		check->cs = cs_new();
+		check->cs = cs_new(NULL);
 		if (!check->cs)
 			goto fail_alloc_cs;
 		if (cs_attach_app(check->cs, &check->obj_type) < 0)

src/conn_stream.c

@@ -46,10 +46,9 @@ void cs_endpoint_free(struct cs_endpoint *endp)
 /* 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()
+struct conn_stream *cs_new(struct cs_endpoint *endp)
 {
 	struct conn_stream *cs;
-	struct cs_endpoint *endp;
 
 	cs = pool_alloc(pool_head_connstream);
 
@@ -62,9 +61,11 @@ struct conn_stream *cs_new()
 	cs->si = NULL;
 	cs->data_cb = NULL;
 
-	endp = cs_endpoint_new();
-	if (unlikely(!endp))
-		goto alloc_error;
+	if (!endp) {
+		endp = cs_endpoint_new();
+		if (unlikely(!endp))
+			goto alloc_error;
+	}
 	cs->endp = endp;
 
 	return cs;

src/dns.c

@@ -891,7 +891,7 @@ static struct appctx *dns_session_create(struct dns_session *ds)
 	struct stream *s;
 	struct applet *applet = &dns_session_applet;
 
-	cs = cs_new();
+	cs = cs_new(NULL);
 	if (!cs) {
 		ha_alert("out of memory in dns_session_create().\n");
 		goto out_close;

src/flt_spoe.c

@@ -1991,7 +1991,7 @@ spoe_create_appctx(struct spoe_config *conf)
 	struct conn_stream *cs;
 	struct stream      *strm;
 
-	cs = cs_new();
+	cs = cs_new(NULL);
 	if (!cs)
 		goto out_error;
 

src/hlua.c

@@ -2944,7 +2944,7 @@ __LJMP static int hlua_socket_new(lua_State *L)
 	lua_rawgeti(L, LUA_REGISTRYINDEX, class_socket_ref);
 	lua_setmetatable(L, -2);
 
-	cs = cs_new();
+	cs = cs_new(NULL);
 	if (!cs) {
 		hlua_pusherror(L, "socket: out of memory");
 		goto out_fail_conf;

src/http_client.c

@@ -476,7 +476,7 @@ struct appctx *httpclient_start(struct httpclient *hc)
 		goto out;
 	}
 
-	cs = cs_new();
+	cs = cs_new(NULL);
 	if (!cs) {
 		ha_alert("httpclient: out of memory in %s:%d.\n", __FUNCTION__, __LINE__);
 		goto out;

src/mux_h1.c

@@ -716,23 +716,31 @@ static inline size_t h1s_data_pending(const struct h1s *h1s)
 static struct conn_stream *h1s_new_cs(struct h1s *h1s, struct buffer *input)
 {
 	struct h1c *h1c = h1s->h1c;
+	struct cs_endpoint *endp;
 	struct conn_stream *cs;
 
 	TRACE_ENTER(H1_EV_STRM_NEW, h1c->conn, h1s);
-	cs = cs_new();
+	endp = cs_endpoint_new();
+	if (!endp) {
+		TRACE_ERROR("CS endp allocation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, h1c->conn, h1s);
+		goto err;
+	}
+	endp->target = h1s;
+	endp->ctx = h1c->conn;
+	if (h1s->flags & H1S_F_NOT_FIRST)
+		endp->flags |= CS_EP_NOT_FIRST;
+	if (h1s->req.flags & H1_MF_UPG_WEBSOCKET)
+		endp->flags |= CS_EP_WEBSOCKET;
+
+	cs = cs_new(endp);
 	if (!cs) {
 		TRACE_ERROR("CS allocation failure", H1_EV_STRM_NEW|H1_EV_STRM_END|H1_EV_STRM_ERR, h1c->conn, h1s);
+		cs_endpoint_free(endp);
 		goto err;
 	}
 	cs_attach_endp_mux(cs, h1s, h1c->conn);
 	h1s->cs = cs;
 
-	if (h1s->flags & H1S_F_NOT_FIRST)
-		cs->endp->flags |= CS_EP_NOT_FIRST;
-
-	if (h1s->req.flags & H1_MF_UPG_WEBSOCKET)
-		cs->endp->flags |= CS_EP_WEBSOCKET;
-
 	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_cs;

src/mux_h2.c

@@ -1590,6 +1590,7 @@ static struct h2s *h2s_new(struct h2c *h2c, int id)
 static struct h2s *h2c_frt_stream_new(struct h2c *h2c, int id, struct buffer *input, uint32_t flags)
 {
 	struct session *sess = h2c->conn->owner;
+	struct cs_endpoint *endp;
 	struct conn_stream *cs;
 	struct h2s *h2s;
 
@@ -1602,19 +1603,26 @@ static struct h2s *h2c_frt_stream_new(struct h2c *h2c, int id, struct buffer *in
 	if (!h2s)
 		goto out;
 
-	cs = cs_new();
-	if (!cs)
+	endp = cs_endpoint_new();
+	if (!endp)
 		goto out_close;
-	cs->endp->flags |= CS_EP_NOT_FIRST;
-	cs_attach_endp_mux(cs, h2s, h2c->conn);
-	h2s->cs = cs;
-	h2c->nb_cs++;
-
+	endp->target = h2s;
+	endp->ctx = h2c->conn;
+	endp->flags |= CS_EP_NOT_FIRST;
 	/* FIXME wrong analogy between ext-connect and websocket, this need to
 	 * be refine.
 	 */
 	if (flags & H2_SF_EXT_CONNECT_RCVD)
-		cs->endp->flags |= CS_EP_WEBSOCKET;
+		endp->flags |= CS_EP_WEBSOCKET;
+
+	cs = cs_new(endp);
+	if (!cs) {
+		cs_endpoint_free(endp);
+		goto out_close;
+	}
+	cs_attach_endp_mux(cs, h2s, h2c->conn);
+	h2s->cs = cs;
+	h2c->nb_cs++;
 
 	/* The stream will record the request's accept date (which is either the
 	 * end of the connection's or the date immediately after the previous

src/mux_pt.c

@@ -272,6 +272,7 @@ struct task *mux_pt_io_cb(struct task *t, void *tctx, unsigned int status)
 static int mux_pt_init(struct connection *conn, struct proxy *prx, struct session *sess,
 		       struct buffer *input)
 {
+	struct cs_endpoint *endp;
 	struct conn_stream *cs = conn->ctx;
 	struct mux_pt_ctx *ctx = pool_alloc(pool_head_pt_ctx);
 
@@ -291,9 +292,15 @@ static int mux_pt_init(struct connection *conn, struct proxy *prx, struct sessio
 	ctx->conn = conn;
 
 	if (!cs) {
-		cs = cs_new();
+		endp = cs_endpoint_new();
+		if (!endp)
+			goto fail_free_ctx;
+		endp->target = ctx;
+		endp->ctx = conn;
+		cs = cs_new(endp);
 		if (!cs) {
 			TRACE_ERROR("CS allocation failure", PT_EV_STRM_NEW|PT_EV_STRM_END|PT_EV_STRM_ERR, conn);
+			cs_endpoint_free(endp);
 			goto fail_free_ctx;
 		}
 		cs_attach_endp_mux(cs, ctx, conn);

src/peers.c

@@ -3191,7 +3191,7 @@ static struct appctx *peer_session_create(struct peers *peers, struct peer *peer
 	peer->last_hdshk = now_ms;
 	s = NULL;
 
-	cs = cs_new();
+	cs = cs_new(NULL);
 	if (!cs) {
 		ha_alert("out of memory in peer_session_create().\n");
 		goto out_close;

src/sink.c

@@ -643,7 +643,7 @@ static struct appctx *sink_forward_session_create(struct sink *sink, struct sink
 	if (sft->srv->log_proto == SRV_LOG_PROTO_OCTET_COUNTING)
 		applet = &sink_forward_oc_applet;
 
-	cs = cs_new();
+	cs = cs_new(NULL);
 	if (!cs) {
 		ha_alert("out of memory in sink_forward_session_create");
 		goto out_close;

src/stream.c

@@ -443,7 +443,7 @@ struct stream *stream_new(struct session *sess, struct conn_stream *cs, struct b
 		s->flags |= SF_HTX;
 
 	s->csf = cs;
-	s->csb = cs_new();
+	s->csb = cs_new(NULL);
 	if (!s->csb)
 		goto out_fail_alloc_csb;