WIP: httpterm: add httpterm sources

Makefile

@@ -951,11 +951,12 @@ all:
 	@echo
 	@exit 1
 else
-all: dev/flags/flags haproxy $(EXTRA)
+all: dev/flags/flags haproxy httpterm $(EXTRA)
 endif # obsolete targets
 endif # TARGET
 
 OBJS =
+HTTPTERM_OBJS =
 
 ifneq ($(EXTRA_OBJS),)
   OBJS += $(EXTRA_OBJS)
@@ -1003,12 +1004,14 @@ OBJS += src/mux_h2.o src/mux_h1.o src/mux_fcgi.o src/log.o		\
         src/http_acl.o src/dict.o src/dgram.o src/pipe.o		\
         src/hpack-huff.o src/hpack-enc.o src/ebtree.o src/hash.o	\
         src/httpclient_cli.o src/version.o src/ncbmbuf.o src/ech.o	\
-        src/cfgparse-peers.o
+        src/cfgparse-peers.o src/httpterm.o
 
 ifneq ($(TRACE),)
   OBJS += src/calltrace.o
 endif
 
+HTTPTERM_OBJS += $(OBJS)
+
 # Used only for forced dependency checking. May be cleared during development.
 INCLUDES = $(wildcard include/*/*.h)
 DEP = $(INCLUDES) .build_opts
@@ -1056,6 +1059,9 @@ endif # non-empty target
 haproxy: $(OPTIONS_OBJS) $(OBJS)
 	$(cmd_LD) $(ARCH_FLAGS) $(LDFLAGS) -o $@ $^ $(LDOPTS)
 
+httpterm: $(OPTIONS_OBJS) $(HTTPTERM_OBJS)
+	$(cmd_LD) $(ARCH_FLAGS) $(LDFLAGS) -o $@ $^ $(LDOPTS)
+
 objsize: haproxy
 	$(Q)objdump -t $^|grep ' g '|grep -F '.text'|awk '{print $$5 FS $$6}'|sort
 

include/haproxy/hstream-t.h

@@ -0,0 +1,35 @@
+#ifndef _HAPROXY_HSTREAM_T_H
+#define _HAPROXY_HSTREAM_T_H
+
+#include <haproxy/dynbuf-t.h>
+#include <haproxy/http-t.h>
+#include <haproxy/obj_type-t.h>
+
+/* httpterm stream */
+struct hstream {
+	enum obj_type obj_type;
+	struct session *sess;
+
+	struct stconn *sc;
+	struct task *task;
+
+	struct buffer req;
+	struct buffer res;
+	unsigned long long to_write;    /* #of response data bytes to write after headers */
+	/* Wait list for buffer allocation */
+	struct buffer_wait buf_wait;
+
+	int flags;
+
+	int ka;                      /* .0: keep-alive  .1: forced  .2: http/1.1, .3: was_reused */
+	unsigned long long req_size; /* values passed in the URI to override the server's */
+	unsigned long long req_body; /* remaining body to be consumed from the request */
+	int req_code;
+	int req_cache, req_time;
+	int req_chunked;
+	int req_random;
+	int req_after_res;           /* Drain the request body after having sent the response */
+	enum http_meth_t req_meth;
+};
+
+#endif /* _HAPROXY_HSTREAM_T_H */

include/haproxy/hstream.h

@@ -0,0 +1,13 @@
+#ifndef _HAPROXY_HSTREAM_H
+#define  _HAPROXY_HSTREAM_H
+
+#include <haproxy/cfgparse.h>
+#include <haproxy/hstream-t.h>
+
+void init_httpterm_cfg(int argc, char **argv, struct cfgfile *cfg);
+struct task *sc_hstream_io_cb(struct task *t, void *ctx, unsigned int state);
+int hstream_wake(struct stconn *sc);
+void hstream_shutdown(struct stconn *sc);
+void *hstream_new(struct session *sess, struct stconn *sc, struct buffer *input);
+
+#endif /* _HAPROXY_HSTREAM_H */

include/haproxy/obj_type-t.h

@@ -46,6 +46,7 @@ enum obj_type {
 #ifdef USE_QUIC
 	OBJ_TYPE_DGRAM,        /* object is a struct quic_dgram */
 #endif
+	OBJ_TYPE_HTTPTERM,     /* object is a struct hstream */
 	OBJ_TYPE_ENTRIES       /* last one : number of entries */
 } __attribute__((packed)) ;
 

include/haproxy/obj_type.h

@@ -26,6 +26,7 @@
 #include <haproxy/applet-t.h>
 #include <haproxy/check-t.h>
 #include <haproxy/connection-t.h>
+#include <haproxy/hstream-t.h>
 #include <haproxy/listener-t.h>
 #include <haproxy/obj_type-t.h>
 #include <haproxy/pool.h>
@@ -189,6 +190,19 @@ static inline struct check *objt_check(enum obj_type *t)
 	return __objt_check(t);
 }
 
+static inline struct hstream *__objt_hstream(enum obj_type *t)
+{
+	return container_of(t, struct hstream, obj_type);
+}
+
+static inline struct hstream *objt_hstream(enum obj_type *t)
+{
+	if (!t || *t != OBJ_TYPE_HTTPTERM)
+		return NULL;
+
+	return __objt_hstream(t);
+}
+
 #ifdef USE_QUIC
 static inline struct quic_dgram *__objt_dgram(enum obj_type *t)
 {

include/haproxy/proxy-t.h

@@ -412,6 +412,7 @@ struct proxy {
 	int redispatch_after;			/* number of retries before redispatch */
 	unsigned down_time;			/* total time the proxy was down */
 	int (*accept)(struct stream *s);       /* application layer's accept() */
+	void *(*stream_new_from_sc)(struct session *sess, struct stconn *sc, struct buffer *in); /* stream connector creator function from mux stream connector */
 	struct conn_src conn_src;               /* connection source settings */
 	enum obj_type *default_target;		/* default target to use for accepted streams or NULL */
 	struct proxy *next;

include/haproxy/stconn.h

@@ -24,6 +24,7 @@
 
 #include <haproxy/api.h>
 #include <haproxy/connection.h>
+#include <haproxy/hstream-t.h>
 #include <haproxy/htx-t.h>
 #include <haproxy/obj_type.h>
 #include <haproxy/stconn-t.h>
@@ -45,10 +46,12 @@ void se_shutdown(struct sedesc *sedesc, enum se_shut_mode mode);
 struct stconn *sc_new_from_endp(struct sedesc *sedesc, struct session *sess, struct buffer *input);
 struct stconn *sc_new_from_strm(struct stream *strm, unsigned int flags);
 struct stconn *sc_new_from_check(struct check *check, unsigned int flags);
+struct stconn *sc_new_from_httpterm(struct sedesc *sd, struct session *sess, struct buffer *input);
 void sc_free(struct stconn *sc);
 
 int sc_attach_mux(struct stconn *sc, void *target, void *ctx);
 int sc_attach_strm(struct stconn *sc, struct stream *strm);
+int sc_attach_hstream(struct stconn *sc, struct hstream *hs);
 
 void sc_destroy(struct stconn *sc);
 int sc_reset_endp(struct stconn *sc);
@@ -331,6 +334,21 @@ static inline struct check *sc_check(const struct stconn *sc)
 	return NULL;
 }
 
+/* Returns the httpterm stream from a sc if the application is a
+ * httpterm stream. Otherwise NULL is returned. __sc_hstream() returns the httpterm
+ * stream without any control while sc_hstream() check the application type.
+ */
+static inline struct hstream *__sc_hstream(const struct stconn *sc)
+{
+	return __objt_hstream(sc->app);
+}
+static inline struct hstream *sc_hstream(const struct stconn *sc)
+{
+	if (obj_type(sc->app) == OBJ_TYPE_HTTPTERM)
+		return __objt_hstream(sc->app);
+	return NULL;
+}
+
 /* Returns the name of the application layer's name for the stconn,
  * or "NONE" when none is attached.
  */

include/haproxy/stream.h

@@ -59,7 +59,7 @@ extern struct pool_head *pool_head_uniqueid;
 
 extern struct data_cb sess_conn_cb;
 
-struct stream *stream_new(struct session *sess, struct stconn *sc, struct buffer *input);
+void *stream_new(struct session *sess, struct stconn *sc, struct buffer *input);
 void stream_free(struct stream *s);
 int stream_upgrade_from_sc(struct stconn *sc, struct buffer *input);
 int stream_set_http_mode(struct stream *s, const struct mux_proto_list *mux_proto);

src/cfgparse-listen.c

@@ -18,6 +18,7 @@
 #include <haproxy/compression-t.h>
 #include <haproxy/connection.h>
 #include <haproxy/extcheck.h>
+#include <haproxy/hstream.h>
 #include <haproxy/http_ana.h>
 #include <haproxy/http_htx.h>
 #include <haproxy/http_ext.h>
@@ -719,6 +720,10 @@ int cfg_parse_listen(const char *file, int linenum, char **args, int kwm)
 			err_code |= ERR_ALERT | ERR_FATAL;
 			goto out;
 		}
+		else if (strcmp(args[1], "httpterm") == 0 && (curproxy->cap & PR_CAP_FE)) {
+			curproxy->mode = PR_MODE_HTTP;
+			curproxy->stream_new_from_sc = hstream_new;
+		}
 		else {
 			ha_alert("parsing [%s:%d] : unknown proxy mode '%s'.\n", file, linenum, args[1]);
 			err_code |= ERR_ALERT | ERR_FATAL;

src/proxy.c

@@ -31,6 +31,7 @@
 #include <haproxy/filters.h>
 #include <haproxy/global.h>
 #include <haproxy/guid.h>
+#include <haproxy/hstream.h>
 #include <haproxy/http_ana.h>
 #include <haproxy/http_htx.h>
 #include <haproxy/http_ext.h>
@@ -1550,6 +1551,7 @@ void init_new_proxy(struct proxy *p)
 	/* Default to only allow L4 retries */
 	p->retry_type = PR_RE_CONN_FAILED;
 
+	p->stream_new_from_sc = stream_new;
 	guid_init(&p->guid);
 
 	p->extra_counters_fe = NULL;
@@ -1914,6 +1916,7 @@ static int proxy_defproxy_cpy(struct proxy *curproxy, const struct proxy *defpro
 		curproxy->clitcpka_cnt   = defproxy->clitcpka_cnt;
 		curproxy->clitcpka_idle  = defproxy->clitcpka_idle;
 		curproxy->clitcpka_intvl = defproxy->clitcpka_intvl;
+		curproxy->stream_new_from_sc = defproxy->stream_new_from_sc;
 	}
 
 	if (curproxy->cap & PR_CAP_BE) {

src/stconn.c

@@ -16,6 +16,7 @@
 #include <haproxy/connection.h>
 #include <haproxy/check.h>
 #include <haproxy/filters.h>
+#include <haproxy/hstream.h>
 #include <haproxy/http_ana.h>
 #include <haproxy/pipe.h>
 #include <haproxy/pool.h>
@@ -91,6 +92,15 @@ struct sc_app_ops sc_app_check_ops = {
 	.name    = "CHCK",
 };
 
+struct sc_app_ops sc_app_hstream_ops = {
+	.chk_rcv = NULL,
+	.chk_snd = NULL,
+	.abort   = NULL,
+	.shutdown= NULL,
+	.wake    = hstream_wake,
+	.name    = "HTERM",
+};
+
 /* Initializes an endpoint */
 void sedesc_init(struct sedesc *sedesc)
 {
@@ -244,7 +254,7 @@ struct stconn *sc_new_from_endp(struct sedesc *sd, struct session *sess, struct
 	sc = sc_new(sd);
 	if (unlikely(!sc))
 		return NULL;
-	if (unlikely(!stream_new(sess, sc, input))) {
+	if (unlikely(!sess->fe->stream_new_from_sc(sess, sc, input))) {
 		sd->sc = NULL;
 		if (sc->sedesc != sd) {
 			/* none was provided so sc_new() allocated one */
@@ -415,6 +425,30 @@ int sc_attach_strm(struct stconn *sc, struct stream *strm)
 	return 0;
 }
 
+/* Attach a stconn to a httpterm layer and sets the relevant
+ * callbacks. Returns -1 on error and 0 on success. SE_FL_ORPHAN flag is
+ * removed. This function is called by a httpterm stream when it is created
+ * to attach it on the stream connector on the client side.
+ */
+int sc_attach_hstream(struct stconn *sc, struct hstream *hs)
+{
+	BUG_ON(!sc_ep_test(sc, SE_FL_T_MUX));
+
+	sc->app = &hs->obj_type;
+	sc_ep_clr(sc, SE_FL_ORPHAN);
+	sc_ep_report_read_activity(sc);
+	sc->wait_event.tasklet = tasklet_new();
+	if (!sc->wait_event.tasklet)
+		return -1;
+
+	sc->wait_event.tasklet->process = sc_hstream_io_cb;
+	sc->wait_event.tasklet->context = sc;
+	sc->wait_event.events = 0;
+
+	sc->app_ops = &sc_app_hstream_ops;
+	return 0;
+}
+
 /* Detaches the stconn from the endpoint, if any. For a connecrion, if a
  * mux owns the connection ->detach() callback is called. Otherwise, it means
  * the stream connector owns the connection. In this case the connection is closed

src/stream.c

@@ -344,7 +344,7 @@ int stream_buf_available(void *arg)
  * 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, struct stconn *sc, struct buffer *input)
+void *stream_new(struct session *sess, struct stconn *sc, struct buffer *input)
 {
 	struct stream *s;
 	struct task *t;