Adding upstream version 2.9.5.upstream/2.9.5

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 2050 insertions, 0 deletions

diff --git a/src/stconn.c b/src/stconn.c
new file mode 100644
index 0000000..8e3ae7e
--- /dev/null
+++ b/src/stconn.c
@@ -0,0 +1,2050 @@
+/*
+ * stream connector management functions
+ *
+ * Copyright 2021 Christopher Faulet <cfaulet@haproxy.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version
+ * 2 of the License, or (at your option) any later version.
+ *
+ */
+
+#include <haproxy/api.h>
+#include <haproxy/applet.h>
+#include <haproxy/connection.h>
+#include <haproxy/check.h>
+#include <haproxy/http_ana.h>
+#include <haproxy/pipe.h>
+#include <haproxy/pool.h>
+#include <haproxy/sample.h>
+#include <haproxy/sc_strm.h>
+#include <haproxy/stconn.h>
+#include <haproxy/xref.h>
+
+DECLARE_POOL(pool_head_connstream, "stconn", sizeof(struct stconn));
+DECLARE_POOL(pool_head_sedesc, "sedesc", sizeof(struct sedesc));
+
+/* functions used by default on a detached stream connector */
+static void sc_app_abort(struct stconn *sc);
+static void sc_app_shut(struct stconn *sc);
+static void sc_app_chk_rcv(struct stconn *sc);
+static void sc_app_chk_snd(struct stconn *sc);
+
+/* functions used on a mux-based stream connector */
+static void sc_app_abort_conn(struct stconn *sc);
+static void sc_app_shut_conn(struct stconn *sc);
+static void sc_app_chk_rcv_conn(struct stconn *sc);
+static void sc_app_chk_snd_conn(struct stconn *sc);
+
+/* functions used on an applet-based stream connector */
+static void sc_app_abort_applet(struct stconn *sc);
+static void sc_app_shut_applet(struct stconn *sc);
+static void sc_app_chk_rcv_applet(struct stconn *sc);
+static void sc_app_chk_snd_applet(struct stconn *sc);
+
+static int sc_conn_process(struct stconn *sc);
+static int sc_conn_recv(struct stconn *sc);
+static int sc_conn_send(struct stconn *sc);
+static int sc_applet_process(struct stconn *sc);
+
+/* stream connector operations for connections */
+struct sc_app_ops sc_app_conn_ops = {
+	.chk_rcv = sc_app_chk_rcv_conn,
+	.chk_snd = sc_app_chk_snd_conn,
+	.abort   = sc_app_abort_conn,
+	.shutdown= sc_app_shut_conn,
+	.wake    = sc_conn_process,
+	.name    = "STRM",
+};
+
+/* stream connector operations for embedded tasks */
+struct sc_app_ops sc_app_embedded_ops = {
+	.chk_rcv = sc_app_chk_rcv,
+	.chk_snd = sc_app_chk_snd,
+	.abort   = sc_app_abort,
+	.shutdown= sc_app_shut,
+	.wake    = NULL,   /* may never be used */
+	.name    = "NONE", /* may never be used */
+};
+
+/* stream connector operations for applets */
+struct sc_app_ops sc_app_applet_ops = {
+	.chk_rcv = sc_app_chk_rcv_applet,
+	.chk_snd = sc_app_chk_snd_applet,
+	.abort   = sc_app_abort_applet,
+	.shutdown= sc_app_shut_applet,
+	.wake    = sc_applet_process,
+	.name    = "STRM",
+};
+
+/* stream connector for health checks on connections */
+struct sc_app_ops sc_app_check_ops = {
+	.chk_rcv = NULL,
+	.chk_snd = NULL,
+	.abort   = NULL,
+	.shutdown= NULL,
+	.wake    = wake_srv_chk,
+	.name    = "CHCK",
+};
+
+/* Initializes an endpoint */
+void sedesc_init(struct sedesc *sedesc)
+{
+	sedesc->se = NULL;
+	sedesc->conn = NULL;
+	sedesc->sc = NULL;
+	sedesc->lra = TICK_ETERNITY;
+	sedesc->fsb = TICK_ETERNITY;
+	sedesc->xref.peer = NULL;
+	se_fl_setall(sedesc, SE_FL_NONE);
+
+	sedesc->iobuf.pipe = NULL;
+	sedesc->iobuf.buf = NULL;
+	sedesc->iobuf.offset = sedesc->iobuf.data = 0;
+	sedesc->iobuf.flags = IOBUF_FL_NONE;
+}
+
+/* Tries to alloc an endpoint and initialize it. Returns NULL on failure. */
+struct sedesc *sedesc_new()
+{
+	struct sedesc *sedesc;
+
+	sedesc = pool_alloc(pool_head_sedesc);
+	if (unlikely(!sedesc))
+		return NULL;
+
+	sedesc_init(sedesc);
+	return sedesc;
+}
+
+/* Releases an endpoint. It is the caller responsibility to be sure it is safe
+ * and it is not shared with another entity
+ */
+void sedesc_free(struct sedesc *sedesc)
+{
+	if (sedesc) {
+		if (sedesc->iobuf.pipe)
+			put_pipe(sedesc->iobuf.pipe);
+		pool_free(pool_head_sedesc, sedesc);
+	}
+}
+
+/* Tries to allocate a new stconn and initialize its main fields. On
+ * failure, nothing is allocated and NULL is returned. It is an internal
+ * function. The caller must, at least, set the SE_FL_ORPHAN or SE_FL_DETACHED
+ * flag.
+ */
+static struct stconn *sc_new(struct sedesc *sedesc)
+{
+	struct stconn *sc;
+
+	sc = pool_alloc(pool_head_connstream);
+
+	if (unlikely(!sc))
+		goto alloc_error;
+
+	sc->obj_type = OBJ_TYPE_SC;
+	sc->flags = SC_FL_NONE;
+	sc->state = SC_ST_INI;
+	sc->ioto = TICK_ETERNITY;
+	sc->room_needed = 0;
+	sc->app = NULL;
+	sc->app_ops = NULL;
+	sc->src = NULL;
+	sc->dst = NULL;
+	sc->wait_event.tasklet = NULL;
+	sc->wait_event.events = 0;
+
+	/* If there is no endpoint, allocate a new one now */
+	if (!sedesc) {
+		sedesc = sedesc_new();
+		if (unlikely(!sedesc))
+			goto alloc_error;
+	}
+	sc->sedesc = sedesc;
+	sedesc->sc = sc;
+
+	return sc;
+
+  alloc_error:
+	pool_free(pool_head_connstream, sc);
+	return NULL;
+}
+
+/* Creates a new stream connector and its associated stream from a mux. <sd> must
+ * be defined. It returns NULL on error. On success, the new stream connector is
+ * returned. In this case, SE_FL_ORPHAN flag is removed.
+ */
+struct stconn *sc_new_from_endp(struct sedesc *sd, struct session *sess, struct buffer *input)
+{
+	struct stconn *sc;
+
+	sc = sc_new(sd);
+	if (unlikely(!sc))
+		return NULL;
+	if (unlikely(!stream_new(sess, sc, input))) {
+		sd->sc = NULL;
+		if (sc->sedesc != sd) {
+			/* none was provided so sc_new() allocated one */
+			sedesc_free(sc->sedesc);
+		}
+		pool_free(pool_head_connstream, sc);
+		se_fl_set(sd, SE_FL_ORPHAN);
+		return NULL;
+	}
+	se_fl_clr(sd, SE_FL_ORPHAN);
+	return sc;
+}
+
+/* Creates a new stream connector from an stream. There is no endpoint here, thus it
+ * will be created by sc_new(). So the SE_FL_DETACHED flag is set. It returns
+ * NULL on error. On success, the new stream connector is returned.
+ */
+struct stconn *sc_new_from_strm(struct stream *strm, unsigned int flags)
+{
+	struct stconn *sc;
+
+	sc = sc_new(NULL);
+	if (unlikely(!sc))
+		return NULL;
+	sc->flags |= flags;
+	sc_ep_set(sc, SE_FL_DETACHED);
+	sc->app = &strm->obj_type;
+	sc->app_ops = &sc_app_embedded_ops;
+	return sc;
+}
+
+/* Creates a new stream connector from an health-check. There is no endpoint here,
+ * thus it will be created by sc_new(). So the SE_FL_DETACHED flag is set. It
+ * returns NULL on error. On success, the new stream connector is returned.
+ */
+struct stconn *sc_new_from_check(struct check *check, unsigned int flags)
+{
+	struct stconn *sc;
+
+	sc = sc_new(NULL);
+	if (unlikely(!sc))
+		return NULL;
+	sc->flags |= flags;
+	sc_ep_set(sc, SE_FL_DETACHED);
+	sc->app = &check->obj_type;
+	sc->app_ops = &sc_app_check_ops;
+	return sc;
+}
+
+/* Releases a stconn previously allocated by sc_new(), as well as its
+ * endpoint, if it exists. This function is called internally or on error path.
+ */
+void sc_free(struct stconn *sc)
+{
+	sockaddr_free(&sc->src);
+	sockaddr_free(&sc->dst);
+	if (sc->sedesc) {
+		BUG_ON(!sc_ep_test(sc, SE_FL_DETACHED));
+		sedesc_free(sc->sedesc);
+	}
+	tasklet_free(sc->wait_event.tasklet);
+	pool_free(pool_head_connstream, sc);
+}
+
+/* Conditionally removes a stream connector if it is detached and if there is no app
+ * layer defined. Except on error path, this one must be used. if release, the
+ * pointer on the SC is set to NULL.
+ */
+static void sc_free_cond(struct stconn **scp)
+{
+	struct stconn *sc = *scp;
+
+	if (!sc->app && (!sc->sedesc || sc_ep_test(sc, SE_FL_DETACHED))) {
+		sc_free(sc);
+		*scp = NULL;
+	}
+}
+
+
+/* Attaches a stconn to a mux endpoint and sets the endpoint ctx. Returns
+ * -1 on error and 0 on success. SE_FL_DETACHED flag is removed. This function is
+ * called from a mux when it is attached to a stream or a health-check.
+ */
+int sc_attach_mux(struct stconn *sc, void *sd, void *ctx)
+{
+	struct connection *conn = ctx;
+	struct sedesc *sedesc = sc->sedesc;
+
+	if (sc_strm(sc)) {
+		if (!sc->wait_event.tasklet) {
+			sc->wait_event.tasklet = tasklet_new();
+			if (!sc->wait_event.tasklet)
+				return -1;
+			sc->wait_event.tasklet->process = sc_conn_io_cb;
+			sc->wait_event.tasklet->context = sc;
+			sc->wait_event.events = 0;
+		}
+
+		sc->app_ops = &sc_app_conn_ops;
+		xref_create(&sc->sedesc->xref, &sc_opposite(sc)->sedesc->xref);
+	}
+	else if (sc_check(sc)) {
+		if (!sc->wait_event.tasklet) {
+			sc->wait_event.tasklet = tasklet_new();
+			if (!sc->wait_event.tasklet)
+				return -1;
+			sc->wait_event.tasklet->process = srv_chk_io_cb;
+			sc->wait_event.tasklet->context = sc;
+			sc->wait_event.events = 0;
+		}
+
+		sc->app_ops = &sc_app_check_ops;
+	}
+
+	sedesc->se = sd;
+	sedesc->conn = ctx;
+	se_fl_set(sedesc, SE_FL_T_MUX);
+	se_fl_clr(sedesc, SE_FL_DETACHED);
+	if (!conn->ctx)
+		conn->ctx = sc;
+	return 0;
+}
+
+/* Attaches a stconn to an applet endpoint and sets the endpoint
+ * ctx. Returns -1 on error and 0 on success. SE_FL_DETACHED flag is
+ * removed. This function is called by a stream when a backend applet is
+ * registered.
+ */
+static void sc_attach_applet(struct stconn *sc, void *sd)
+{
+	sc->sedesc->se = sd;
+	sc_ep_set(sc, SE_FL_T_APPLET);
+	sc_ep_clr(sc, SE_FL_DETACHED);
+	if (sc_strm(sc)) {
+		sc->app_ops = &sc_app_applet_ops;
+		xref_create(&sc->sedesc->xref, &sc_opposite(sc)->sedesc->xref);
+	}
+}
+
+/* Attaches a stconn to a app 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 stream when it is created to attach it
+ * on the stream connector on the client side.
+ */
+int sc_attach_strm(struct stconn *sc, struct stream *strm)
+{
+	sc->app = &strm->obj_type;
+	sc_ep_clr(sc, SE_FL_ORPHAN);
+	sc_ep_report_read_activity(sc);
+	if (sc_ep_test(sc, SE_FL_T_MUX)) {
+		sc->wait_event.tasklet = tasklet_new();
+		if (!sc->wait_event.tasklet)
+			return -1;
+		sc->wait_event.tasklet->process = sc_conn_io_cb;
+		sc->wait_event.tasklet->context = sc;
+		sc->wait_event.events = 0;
+
+		sc->app_ops = &sc_app_conn_ops;
+	}
+	else if (sc_ep_test(sc, SE_FL_T_APPLET)) {
+		sc->app_ops = &sc_app_applet_ops;
+	}
+	else {
+		sc->app_ops = &sc_app_embedded_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
+ * and released. For an applet, the appctx is released. If still allocated, the
+ * endpoint is reset and flag as detached. If the app layer is also detached,
+ * the stream connector is released.
+ */
+static void sc_detach_endp(struct stconn **scp)
+{
+	struct stconn *sc = *scp;
+	struct xref *peer;
+
+	if (!sc)
+		return;
+
+
+	/* Remove my link in the original objects. */
+	peer = xref_get_peer_and_lock(&sc->sedesc->xref);
+	if (peer)
+		xref_disconnect(&sc->sedesc->xref, peer);
+
+	if (sc_ep_test(sc, SE_FL_T_MUX)) {
+		struct connection *conn = __sc_conn(sc);
+		struct sedesc *sedesc = sc->sedesc;
+
+		if (conn->mux) {
+			if (sc->wait_event.events != 0)
+				conn->mux->unsubscribe(sc, sc->wait_event.events, &sc->wait_event);
+			se_fl_set(sedesc, SE_FL_ORPHAN);
+			sedesc->sc = NULL;
+			sc->sedesc = NULL;
+			conn->mux->detach(sedesc);
+		}
+		else {
+			/* It's too early to have a mux, let's just destroy
+			 * the connection
+			 */
+			conn_stop_tracking(conn);
+			conn_full_close(conn);
+			if (conn->destroy_cb)
+				conn->destroy_cb(conn);
+			conn_free(conn);
+		}
+	}
+	else if (sc_ep_test(sc, SE_FL_T_APPLET)) {
+		struct appctx *appctx = __sc_appctx(sc);
+
+		sc_ep_set(sc, SE_FL_ORPHAN);
+		sc->sedesc->sc = NULL;
+		sc->sedesc = NULL;
+		appctx_shut(appctx);
+		appctx_free(appctx);
+	}
+
+	if (sc->sedesc) {
+		/* the SD wasn't used and can be recycled */
+		sc->sedesc->se     = NULL;
+		sc->sedesc->conn   = NULL;
+		sc->sedesc->flags  = 0;
+		sc_ep_set(sc, SE_FL_DETACHED);
+	}
+
+	/* FIXME: Rest SC for now but must be reviewed. SC flags are only
+	 *        connection related for now but this will evolved
+	 */
+	sc->flags &= SC_FL_ISBACK;
+	if (sc_strm(sc))
+		sc->app_ops = &sc_app_embedded_ops;
+	else
+		sc->app_ops = NULL;
+	sc_free_cond(scp);
+}
+
+/* Detaches the stconn from the app layer. If there is no endpoint attached
+ * to the stconn
+ */
+static void sc_detach_app(struct stconn **scp)
+{
+	struct stconn *sc = *scp;
+
+	if (!sc)
+		return;
+
+	sc->app = NULL;
+	sc->app_ops = NULL;
+	sockaddr_free(&sc->src);
+	sockaddr_free(&sc->dst);
+
+	tasklet_free(sc->wait_event.tasklet);
+	sc->wait_event.tasklet = NULL;
+	sc->wait_event.events = 0;
+	sc_free_cond(scp);
+}
+
+/* Destroy the stconn. It is detached from its endpoint and its
+ * application. After this call, the stconn must be considered as released.
+ */
+void sc_destroy(struct stconn *sc)
+{
+	sc_detach_endp(&sc);
+	sc_detach_app(&sc);
+	BUG_ON_HOT(sc);
+}
+
+/* Resets the stream connector endpoint. It happens when the app layer want to renew
+ * its endpoint. For a connection retry for instance. If a mux or an applet is
+ * attached, a new endpoint is created. Returns -1 on error and 0 on success.
+ */
+int sc_reset_endp(struct stconn *sc)
+{
+	struct sedesc *new_sd;
+
+	BUG_ON(!sc->app);
+
+	if (!__sc_endp(sc)) {
+		/* endpoint not attached or attached to a mux with no
+		 * target. Thus the endpoint will not be release but just
+		 * reset. The app is still attached, the sc will not be
+		 * released.
+		 */
+		sc_detach_endp(&sc);
+		return 0;
+	}
+
+	/* allocate the new endpoint first to be able to set error if it
+	 * fails */
+	new_sd = sedesc_new();
+	if (!unlikely(new_sd))
+		return -1;
+
+	/* The app is still attached, the sc will not be released */
+	sc_detach_endp(&sc);
+	BUG_ON(!sc);
+	BUG_ON(sc->sedesc);
+	sc->sedesc = new_sd;
+	sc->sedesc->sc = sc;
+	sc_ep_set(sc, SE_FL_DETACHED);
+	return 0;
+}
+
+
+/* Create an applet to handle a stream connector as a new appctx. The SC will
+ * wake it up every time it is solicited. The appctx must be deleted by the task
+ * handler using sc_detach_endp(), possibly from within the function itself.
+ * It also pre-initializes the applet's context and returns it (or NULL in case
+ * it could not be allocated).
+ */
+struct appctx *sc_applet_create(struct stconn *sc, struct applet *app)
+{
+	struct appctx *appctx;
+
+	appctx = appctx_new_here(app, sc->sedesc);
+	if (!appctx)
+		return NULL;
+	sc_attach_applet(sc, appctx);
+	appctx->t->nice = __sc_strm(sc)->task->nice;
+	applet_need_more_data(appctx);
+	appctx_wakeup(appctx);
+
+	sc->state = SC_ST_RDY;
+	return appctx;
+}
+
+/* Conditionally forward the close to the write side. It return 1 if it can be
+ * forwarded. It is the caller responsibility to forward the close to the write
+ * side. Otherwise, 0 is returned. In this case, SC_FL_SHUT_WANTED flag may be set on
+ * the consumer SC if we are only waiting for the outgoing data to be flushed.
+ */
+static inline int sc_cond_forward_shut(struct stconn *sc)
+{
+	/* The close must not be forwarded */
+	if (!(sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE)) || !(sc->flags & SC_FL_NOHALF))
+		return 0;
+
+	if (co_data(sc_ic(sc)) && !(sc_ic(sc)->flags & CF_WRITE_TIMEOUT)) {
+		/* the shutdown cannot be forwarded now because
+		 * we should flush outgoing data first. But instruct the output
+		 * channel it should be done ASAP.
+		 */
+		sc_schedule_shutdown(sc);
+		return 0;
+	}
+
+	/* the close can be immediately forwarded to the write side */
+	return 1;
+}
+
+
+static inline int sc_is_fastfwd_supported(struct stconn *sc)
+{
+	return (!(global.tune.no_zero_copy_fwd & NO_ZERO_COPY_FWD) &&
+		sc_ep_test(sc, SE_FL_MAY_FASTFWD_PROD) &&
+		sc_ep_test(sc_opposite(sc), SE_FL_MAY_FASTFWD_CONS) &&
+		sc_ic(sc)->to_forward);
+}
+/*
+ * This function performs a shutdown-read on a detached stream connector in a
+ * connected or init state (it does nothing for other states). It either shuts
+ * the read side or marks itself as closed. The buffer flags are updated to
+ * reflect the new state. If the stream connector has SC_FL_NOHALF, we also
+ * forward the close to the write side. The owner task is woken up if it exists.
+ */
+static void sc_app_abort(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+
+	if (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE))
+		return;
+
+	sc->flags |= SC_FL_ABRT_DONE;
+	ic->flags |= CF_READ_EVENT;
+
+	if (!sc_state_in(sc->state, SC_SB_CON|SC_SB_RDY|SC_SB_EST))
+		return;
+
+	if (sc->flags & SC_FL_SHUT_DONE) {
+		sc->state = SC_ST_DIS;
+		if (sc->flags & SC_FL_ISBACK)
+			__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+	}
+	else if (sc_cond_forward_shut(sc))
+		return sc_app_shut(sc);
+
+	/* note that if the task exists, it must unregister itself once it runs */
+	if (!(sc->flags & SC_FL_DONT_WAKE))
+		task_wakeup(sc_strm_task(sc), TASK_WOKEN_IO);
+}
+
+/*
+ * This function performs a shutdown-write on a detached stream connector in a
+ * connected or init state (it does nothing for other states). It either shuts
+ * the write side or marks itself as closed. The buffer flags are updated to
+ * reflect the new state. It does also close everything if the SC was marked as
+ * being in error state. The owner task is woken up if it exists.
+ */
+static void sc_app_shut(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+	struct channel *oc = sc_oc(sc);
+
+	sc->flags &= ~SC_FL_SHUT_WANTED;
+	if (sc->flags & SC_FL_SHUT_DONE)
+		return;
+	sc->flags |= SC_FL_SHUT_DONE;
+	oc->flags |= CF_WRITE_EVENT;
+	sc_set_hcto(sc);
+
+	switch (sc->state) {
+	case SC_ST_RDY:
+	case SC_ST_EST:
+		/* we have to shut before closing, otherwise some short messages
+		 * may never leave the system, especially when there are remaining
+		 * unread data in the socket input buffer, or when nolinger is set.
+		 * However, if SC_FL_NOLINGER is explicitly set, we know there is
+		 * no risk so we close both sides immediately.
+		 */
+		if (!(sc->flags & (SC_FL_ERROR|SC_FL_NOLINGER|SC_FL_EOS|SC_FL_ABRT_DONE)) &&
+		    !(ic->flags & CF_DONT_READ))
+			return;
+
+		__fallthrough;
+	case SC_ST_CON:
+	case SC_ST_CER:
+	case SC_ST_QUE:
+	case SC_ST_TAR:
+		/* Note that none of these states may happen with applets */
+		sc->state = SC_ST_DIS;
+		__fallthrough;
+	default:
+		sc->flags &= ~SC_FL_NOLINGER;
+		sc->flags |= SC_FL_ABRT_DONE;
+		if (sc->flags & SC_FL_ISBACK)
+			__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+	}
+
+	/* note that if the task exists, it must unregister itself once it runs */
+	if (!(sc->flags & SC_FL_DONT_WAKE))
+		task_wakeup(sc_strm_task(sc), TASK_WOKEN_IO);
+}
+
+/* default chk_rcv function for scheduled tasks */
+static void sc_app_chk_rcv(struct stconn *sc)
+{
+	if (sc_ep_have_ff_data(sc_opposite(sc))) {
+		/* stop reading */
+		sc_need_room(sc, -1);
+	}
+	else {
+		/* (re)start reading */
+		if (!(sc->flags & SC_FL_DONT_WAKE))
+			task_wakeup(sc_strm_task(sc), TASK_WOKEN_IO);
+	}
+}
+
+/* default chk_snd function for scheduled tasks */
+static void sc_app_chk_snd(struct stconn *sc)
+{
+	struct channel *oc = sc_oc(sc);
+
+	if (unlikely(sc->state != SC_ST_EST || (sc->flags & SC_FL_SHUT_DONE)))
+		return;
+
+	if (!sc_ep_test(sc, SE_FL_WAIT_DATA) ||  /* not waiting for data */
+	    (!co_data(oc) && !sc_ep_have_ff_data(sc)))  /* called with nothing to send ! */
+		return;
+
+	/* Otherwise there are remaining data to be sent in the buffer,
+	 * so we tell the handler.
+	 */
+	sc_ep_clr(sc, SE_FL_WAIT_DATA);
+	if (!(sc->flags & SC_FL_DONT_WAKE))
+		task_wakeup(sc_strm_task(sc), TASK_WOKEN_IO);
+}
+
+/*
+ * This function performs a shutdown-read on a stream connector attached to
+ * a connection in a connected or init state (it does nothing for other
+ * states). It either shuts the read side or marks itself as closed. The buffer
+ * flags are updated to reflect the new state. If the stream connector has
+ * SC_FL_NOHALF, we also forward the close to the write side. If a control
+ * layer is defined, then it is supposed to be a socket layer and file
+ * descriptors are then shutdown or closed accordingly. The function
+ * automatically disables polling if needed.
+ */
+static void sc_app_abort_conn(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+
+	BUG_ON(!sc_conn(sc));
+
+	if (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE))
+		return;
+	sc->flags |= SC_FL_ABRT_DONE;
+	ic->flags |= CF_READ_EVENT;
+
+	if (!sc_state_in(sc->state, SC_SB_CON|SC_SB_RDY|SC_SB_EST))
+		return;
+
+	if (sc->flags & SC_FL_SHUT_DONE) {
+		sc_conn_shut(sc);
+		sc->state = SC_ST_DIS;
+		if (sc->flags & SC_FL_ISBACK)
+			__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+	}
+	else if (sc_cond_forward_shut(sc))
+		return sc_app_shut_conn(sc);
+}
+
+/*
+ * This function performs a shutdown-write on a stream connector attached to
+ * a connection in a connected or init state (it does nothing for other
+ * states). It either shuts the write side or marks itself as closed. The
+ * buffer flags are updated to reflect the new state.  It does also close
+ * everything if the SC was marked as being in error state. If there is a
+ * data-layer shutdown, it is called.
+ */
+static void sc_app_shut_conn(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+	struct channel *oc = sc_oc(sc);
+
+	BUG_ON(!sc_conn(sc));
+
+	sc->flags &= ~SC_FL_SHUT_WANTED;
+	if (sc->flags & SC_FL_SHUT_DONE)
+		return;
+	sc->flags |= SC_FL_SHUT_DONE;
+	oc->flags |= CF_WRITE_EVENT;
+	sc_set_hcto(sc);
+
+	switch (sc->state) {
+	case SC_ST_RDY:
+	case SC_ST_EST:
+		/* we have to shut before closing, otherwise some short messages
+		 * may never leave the system, especially when there are remaining
+		 * unread data in the socket input buffer, or when nolinger is set.
+		 * However, if SC_FL_NOLINGER is explicitly set, we know there is
+		 * no risk so we close both sides immediately.
+		 */
+		if (sc->flags & SC_FL_NOLINGER) {
+			/* unclean data-layer shutdown, typically an aborted request
+			 * or a forwarded shutdown from a client to a server due to
+			 * option abortonclose. No need for the TLS layer to try to
+			 * emit a shutdown message.
+			 */
+			sc_conn_shutw(sc, CO_SHW_SILENT);
+		}
+		else {
+			/* clean data-layer shutdown. This only happens on the
+			 * frontend side, or on the backend side when forwarding
+			 * a client close in TCP mode or in HTTP TUNNEL mode
+			 * while option abortonclose is set. We want the TLS
+			 * layer to try to signal it to the peer before we close.
+			 */
+			sc_conn_shutw(sc, CO_SHW_NORMAL);
+
+			if (!(sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE)) && !(ic->flags & CF_DONT_READ))
+				return;
+		}
+
+		__fallthrough;
+	case SC_ST_CON:
+		/* we may have to close a pending connection, and mark the
+		 * response buffer as abort
+		 */
+		sc_conn_shut(sc);
+		__fallthrough;
+	case SC_ST_CER:
+	case SC_ST_QUE:
+	case SC_ST_TAR:
+		sc->state = SC_ST_DIS;
+		__fallthrough;
+	default:
+		sc->flags &= ~SC_FL_NOLINGER;
+		sc->flags |= SC_FL_ABRT_DONE;
+		if (sc->flags & SC_FL_ISBACK)
+			__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+	}
+}
+
+/* This function is used for inter-stream connector calls. It is called by the
+ * consumer to inform the producer side that it may be interested in checking
+ * for free space in the buffer. Note that it intentionally does not update
+ * timeouts, so that we can still check them later at wake-up. This function is
+ * dedicated to connection-based stream connectors.
+ */
+static void sc_app_chk_rcv_conn(struct stconn *sc)
+{
+	BUG_ON(!sc_conn(sc));
+
+	/* (re)start reading */
+	if (sc_state_in(sc->state, SC_SB_CON|SC_SB_RDY|SC_SB_EST))
+		tasklet_wakeup(sc->wait_event.tasklet);
+}
+
+
+/* This function is used for inter-stream connector calls. It is called by the
+ * producer to inform the consumer side that it may be interested in checking
+ * for data in the buffer. Note that it intentionally does not update timeouts,
+ * so that we can still check them later at wake-up.
+ */
+static void sc_app_chk_snd_conn(struct stconn *sc)
+{
+	struct channel *oc = sc_oc(sc);
+
+	BUG_ON(!sc_conn(sc));
+
+	if (unlikely(!sc_state_in(sc->state, SC_SB_RDY|SC_SB_EST) ||
+		     (sc->flags & SC_FL_SHUT_DONE)))
+		return;
+
+	if (unlikely(!co_data(oc) && !sc_ep_have_ff_data(sc)))  /* called with nothing to send ! */
+		return;
+
+	if (!sc_ep_have_ff_data(sc) &&              /* data wants to be fast-forwarded ASAP */
+	    !sc_ep_test(sc, SE_FL_WAIT_DATA))       /* not waiting for data */
+		return;
+
+	if (!(sc->wait_event.events & SUB_RETRY_SEND))
+		sc_conn_send(sc);
+
+	if (sc_ep_test(sc, SE_FL_ERROR | SE_FL_ERR_PENDING) || sc_is_conn_error(sc)) {
+		/* Write error on the file descriptor */
+		BUG_ON(sc_ep_test(sc, SE_FL_EOS|SE_FL_ERROR|SE_FL_ERR_PENDING) == (SE_FL_EOS|SE_FL_ERR_PENDING));
+		goto out_wakeup;
+	}
+
+	/* OK, so now we know that some data might have been sent, and that we may
+	 * have to poll first. We have to do that too if the buffer is not empty.
+	 */
+	if (!co_data(oc)) {
+		/* the connection is established but we can't write. Either the
+		 * buffer is empty, or we just refrain from sending because the
+		 * ->o limit was reached. Maybe we just wrote the last
+		 * chunk and need to close.
+		 */
+		if ((oc->flags & CF_AUTO_CLOSE) &&
+		    ((sc->flags & (SC_FL_SHUT_DONE|SC_FL_SHUT_WANTED)) == SC_FL_SHUT_WANTED) &&
+		    sc_state_in(sc->state, SC_SB_RDY|SC_SB_EST)) {
+			sc_shutdown(sc);
+			goto out_wakeup;
+		}
+
+		if ((sc->flags & (SC_FL_SHUT_DONE|SC_FL_SHUT_WANTED)) == 0)
+			sc_ep_set(sc, SE_FL_WAIT_DATA);
+	}
+	else {
+		/* Otherwise there are remaining data to be sent in the buffer,
+		 * which means we have to poll before doing so.
+		 */
+		sc_ep_clr(sc, SE_FL_WAIT_DATA);
+	}
+
+	/* in case of special condition (error, shutdown, end of write...), we
+	 * have to notify the task.
+	 */
+	if (likely((sc->flags & SC_FL_SHUT_DONE) ||
+		   ((oc->flags & CF_WRITE_EVENT) && sc->state < SC_ST_EST) ||
+		   ((oc->flags & CF_WAKE_WRITE) &&
+		    ((!co_data(oc) && !oc->to_forward) ||
+		     !sc_state_in(sc->state, SC_SB_EST))))) {
+	out_wakeup:
+		if (!(sc->flags & SC_FL_DONT_WAKE))
+			task_wakeup(sc_strm_task(sc), TASK_WOKEN_IO);
+	}
+}
+
+/*
+ * This function performs a shutdown-read on a stream connector attached to an
+ * applet in a connected or init state (it does nothing for other states). It
+ * either shuts the read side or marks itself as closed. The buffer flags are
+ * updated to reflect the new state. If the stream connector has SC_FL_NOHALF,
+ * we also forward the close to the write side. The owner task is woken up if
+ * it exists.
+ */
+static void sc_app_abort_applet(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+
+	BUG_ON(!sc_appctx(sc));
+
+	if (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE))
+		return;
+	sc->flags |= SC_FL_ABRT_DONE;
+	ic->flags |= CF_READ_EVENT;
+
+	/* Note: on abort, we don't call the applet */
+
+	if (!sc_state_in(sc->state, SC_SB_CON|SC_SB_RDY|SC_SB_EST))
+		return;
+
+	if (sc->flags & SC_FL_SHUT_DONE) {
+		appctx_shut(__sc_appctx(sc));
+		sc->state = SC_ST_DIS;
+		if (sc->flags & SC_FL_ISBACK)
+			__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+	}
+	else if (sc_cond_forward_shut(sc))
+		return sc_app_shut_applet(sc);
+}
+
+/*
+ * This function performs a shutdown-write on a stream connector attached to an
+ * applet in a connected or init state (it does nothing for other states). It
+ * either shuts the write side or marks itself as closed. The buffer flags are
+ * updated to reflect the new state. It does also close everything if the SI
+ * was marked as being in error state. The owner task is woken up if it exists.
+ */
+static void sc_app_shut_applet(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+	struct channel *oc = sc_oc(sc);
+
+	BUG_ON(!sc_appctx(sc));
+
+	sc->flags &= ~SC_FL_SHUT_WANTED;
+	if (sc->flags & SC_FL_SHUT_DONE)
+		return;
+	sc->flags |= SC_FL_SHUT_DONE;
+	oc->flags |= CF_WRITE_EVENT;
+	sc_set_hcto(sc);
+
+	/* on shutw we always wake the applet up */
+	appctx_wakeup(__sc_appctx(sc));
+
+	switch (sc->state) {
+	case SC_ST_RDY:
+	case SC_ST_EST:
+		/* we have to shut before closing, otherwise some short messages
+		 * may never leave the system, especially when there are remaining
+		 * unread data in the socket input buffer, or when nolinger is set.
+		 * However, if SC_FL_NOLINGER is explicitly set, we know there is
+		 * no risk so we close both sides immediately.
+		 */
+		if (!(sc->flags & (SC_FL_ERROR|SC_FL_NOLINGER|SC_FL_EOS|SC_FL_ABRT_DONE)) &&
+		    !(ic->flags & CF_DONT_READ))
+			return;
+
+		__fallthrough;
+	case SC_ST_CON:
+	case SC_ST_CER:
+	case SC_ST_QUE:
+	case SC_ST_TAR:
+		/* Note that none of these states may happen with applets */
+		appctx_shut(__sc_appctx(sc));
+		sc->state = SC_ST_DIS;
+		__fallthrough;
+	default:
+		sc->flags &= ~SC_FL_NOLINGER;
+		sc->flags |= SC_FL_ABRT_DONE;
+		if (sc->flags & SC_FL_ISBACK)
+			__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+	}
+}
+
+/* chk_rcv function for applets */
+static void sc_app_chk_rcv_applet(struct stconn *sc)
+{
+	BUG_ON(!sc_appctx(sc));
+
+	if (!sc_ep_have_ff_data(sc_opposite(sc))) {
+		/* (re)start reading */
+		appctx_wakeup(__sc_appctx(sc));
+	}
+}
+
+/* chk_snd function for applets */
+static void sc_app_chk_snd_applet(struct stconn *sc)
+{
+	struct channel *oc = sc_oc(sc);
+
+	BUG_ON(!sc_appctx(sc));
+
+	if (unlikely(sc->state != SC_ST_EST || (sc->flags & SC_FL_SHUT_DONE)))
+		return;
+
+	/* we only wake the applet up if it was waiting for some data  and is ready to consume it */
+	if (!sc_ep_test(sc, SE_FL_WAIT_DATA|SE_FL_WONT_CONSUME))
+		return;
+
+	if (co_data(oc) || sc_ep_have_ff_data(sc)) {
+		/* (re)start sending */
+		appctx_wakeup(__sc_appctx(sc));
+	}
+}
+
+
+/* This function is designed to be called from within the stream handler to
+ * update the input channel's expiration timer and the stream connector's
+ * Rx flags based on the channel's flags. It needs to be called only once
+ * after the channel's flags have settled down, and before they are cleared,
+ * though it doesn't harm to call it as often as desired (it just slightly
+ * hurts performance). It must not be called from outside of the stream
+ * handler, as what it does will be used to compute the stream task's
+ * expiration.
+ */
+void sc_update_rx(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+
+	if (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE))
+		return;
+
+	/* Unblock the SC if it needs room and the free space is large enough (0
+	 * means it can always be unblocked). Do not unblock it if -1 was
+	 * specified.
+	 */
+	if (!sc->room_needed || (sc->room_needed > 0 && channel_recv_max(ic) >= sc->room_needed))
+		sc_have_room(sc);
+
+	/* Read not closed, update FD status and timeout for reads */
+	if (ic->flags & CF_DONT_READ)
+		sc_wont_read(sc);
+	else
+		sc_will_read(sc);
+
+	sc_chk_rcv(sc);
+}
+
+/* This function is designed to be called from within the stream handler to
+ * update the output channel's expiration timer and the stream connector's
+ * Tx flags based on the channel's flags. It needs to be called only once
+ * after the channel's flags have settled down, and before they are cleared,
+ * though it doesn't harm to call it as often as desired (it just slightly
+ * hurts performance). It must not be called from outside of the stream
+ * handler, as what it does will be used to compute the stream task's
+ * expiration.
+ */
+void sc_update_tx(struct stconn *sc)
+{
+	struct channel *oc = sc_oc(sc);
+
+	if (sc->flags & SC_FL_SHUT_DONE)
+		return;
+
+	/* Write not closed, update FD status and timeout for writes */
+	if (!co_data(oc)) {
+		/* stop writing */
+		if (!sc_ep_test(sc, SE_FL_WAIT_DATA)) {
+			if ((sc->flags & SC_FL_SHUT_WANTED) == 0)
+				sc_ep_set(sc, SE_FL_WAIT_DATA);
+		}
+		return;
+	}
+
+	/* (re)start writing */
+	sc_ep_clr(sc, SE_FL_WAIT_DATA);
+}
+
+/* This function is the equivalent to sc_update() except that it's
+ * designed to be called from outside the stream handlers, typically the lower
+ * layers (applets, connections) after I/O completion. After updating the stream
+ * interface and timeouts, it will try to forward what can be forwarded, then to
+ * wake the associated task up if an important event requires special handling.
+ * It may update SE_FL_WAIT_DATA and/or SC_FL_NEED_ROOM, that the callers are
+ * encouraged to watch to take appropriate action.
+ * It should not be called from within the stream itself, sc_update()
+ * is designed for this. Please do not statify this function, it's often
+ * present in backtraces, it's useful to recognize it.
+ */
+void sc_notify(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+	struct channel *oc = sc_oc(sc);
+	struct stconn *sco = sc_opposite(sc);
+	struct task *task = sc_strm_task(sc);
+
+	/* process consumer side */
+	if (!co_data(oc)) {
+		struct connection *conn = sc_conn(sc);
+
+		if (((sc->flags & (SC_FL_SHUT_DONE|SC_FL_SHUT_WANTED)) == SC_FL_SHUT_WANTED) &&
+		    (sc->state == SC_ST_EST) && (!conn || !(conn->flags & (CO_FL_WAIT_XPRT | CO_FL_EARLY_SSL_HS))))
+			sc_shutdown(sc);
+	}
+
+	/* indicate that we may be waiting for data from the output channel or
+	 * we're about to close and can't expect more data if SC_FL_SHUT_WANTED is there.
+	 */
+	if (!(sc->flags & (SC_FL_SHUT_DONE|SC_FL_SHUT_WANTED)))
+		sc_ep_set(sc, SE_FL_WAIT_DATA);
+	else if ((sc->flags & (SC_FL_SHUT_DONE|SC_FL_SHUT_WANTED)) == SC_FL_SHUT_WANTED)
+		sc_ep_clr(sc, SE_FL_WAIT_DATA);
+
+	if (oc->flags & CF_DONT_READ)
+		sc_wont_read(sco);
+	else
+		sc_will_read(sco);
+
+	/* Notify the other side when we've injected data into the IC that
+	 * needs to be forwarded. We can do fast-forwarding as soon as there
+	 * are output data, but we avoid doing this if some of the data are
+	 * not yet scheduled for being forwarded, because it is very likely
+	 * that it will be done again immediately afterwards once the following
+	 * data are parsed (eg: HTTP chunking). We only clear SC_FL_NEED_ROOM
+	 * once we've emptied *some* of the output buffer, and not just when
+	 * there is available room, because applets are often forced to stop
+	 * before the buffer is full. We must not stop based on input data
+	 * alone because an HTTP parser might need more data to complete the
+	 * parsing.
+	 */
+	if (sc_ep_have_ff_data(sc_opposite(sc)) ||
+	    (co_data(ic) && sc_ep_test(sco, SE_FL_WAIT_DATA) &&
+	     (!(sc->flags & SC_FL_SND_EXP_MORE) || channel_full(ic, co_data(ic)) || channel_input_data(ic) == 0))) {
+		int new_len, last_len;
+
+		last_len = co_data(ic) + sc_ep_ff_data(sco);
+		sc_chk_snd(sco);
+		new_len = co_data(ic) + sc_ep_ff_data(sco);
+
+		/* check if the consumer has freed some space either in the
+		 * buffer or in the pipe.
+		 */
+		if (!sc->room_needed || (new_len < last_len && (sc->room_needed < 0 || channel_recv_max(ic) >= sc->room_needed)))
+			sc_have_room(sc);
+	}
+
+	if (!(ic->flags & CF_DONT_READ))
+		sc_will_read(sc);
+
+	sc_chk_rcv(sc);
+	sc_chk_rcv(sco);
+
+	/* wake the task up only when needed */
+	if (/* changes on the production side that must be handled:
+	     *  - An error on receipt: SC_FL_ERROR
+	     *  - A read event: shutdown for reads (CF_READ_EVENT + EOS/ABRT_DONE)
+	     *                  end of input (CF_READ_EVENT + SC_FL_EOI)
+	     *                  data received and no fast-forwarding (CF_READ_EVENT + !to_forward)
+	     *                  read event while consumer side is not established (CF_READ_EVENT + sco->state != SC_ST_EST)
+	     */
+		((ic->flags & CF_READ_EVENT) && ((sc->flags & SC_FL_EOI) || (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE)) || !ic->to_forward || sco->state != SC_ST_EST)) ||
+		(sc->flags & SC_FL_ERROR) ||
+
+	    /* changes on the consumption side */
+	    sc_ep_test(sc, SE_FL_ERR_PENDING) ||
+	    ((oc->flags & CF_WRITE_EVENT) &&
+	     ((sc->state < SC_ST_EST) ||
+	      (sc->flags & SC_FL_SHUT_DONE) ||
+	      (((oc->flags & CF_WAKE_WRITE) ||
+		(!(oc->flags & CF_AUTO_CLOSE) &&
+		 !(sc->flags & (SC_FL_SHUT_WANTED|SC_FL_SHUT_DONE)))) &&
+	       (sco->state != SC_ST_EST ||
+		(!co_data(oc) && !oc->to_forward)))))) {
+		task_wakeup(task, TASK_WOKEN_IO);
+	}
+	else {
+		/* Update expiration date for the task and requeue it if not already expired */
+		if (!tick_is_expired(task->expire, now_ms)) {
+			task->expire = tick_first(task->expire, sc_ep_rcv_ex(sc));
+			task->expire = tick_first(task->expire, sc_ep_snd_ex(sc));
+			task->expire = tick_first(task->expire, sc_ep_rcv_ex(sco));
+			task->expire = tick_first(task->expire, sc_ep_snd_ex(sco));
+			task->expire = tick_first(task->expire, ic->analyse_exp);
+			task->expire = tick_first(task->expire, oc->analyse_exp);
+			task->expire = tick_first(task->expire, __sc_strm(sc)->conn_exp);
+
+			/* WARNING: Don't forget to remove this BUG_ON before 2.9.0 */
+			BUG_ON(tick_is_expired(task->expire, now_ms));
+			task_queue(task);
+		}
+	}
+
+	if (ic->flags & CF_READ_EVENT)
+		sc->flags &= ~SC_FL_RCV_ONCE;
+}
+
+/*
+ * This function propagates an end-of-stream received on a socket-based connection.
+ * It updates the stream connector. If the stream connector has SC_FL_NOHALF,
+ * the close is also forwarded to the write side as an abort.
+ */
+static void sc_conn_eos(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+
+	BUG_ON(!sc_conn(sc));
+
+	if (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE))
+		return;
+	sc->flags |= SC_FL_EOS;
+	ic->flags |= CF_READ_EVENT;
+	sc_ep_report_read_activity(sc);
+
+	if (!sc_state_in(sc->state, SC_SB_CON|SC_SB_RDY|SC_SB_EST))
+		return;
+
+	if (sc->flags & SC_FL_SHUT_DONE)
+		goto do_close;
+
+	if (sc_cond_forward_shut(sc)) {
+		/* we want to immediately forward this close to the write side */
+		/* force flag on ssl to keep stream in cache */
+		sc_conn_shutw(sc, CO_SHW_SILENT);
+		goto do_close;
+	}
+
+	/* otherwise that's just a normal read shutdown */
+	return;
+
+ do_close:
+	/* OK we completely close the socket here just as if we went through sc_shut[rw]() */
+	sc_conn_shut(sc);
+
+	sc->flags &= ~SC_FL_SHUT_WANTED;
+	sc->flags |= SC_FL_SHUT_DONE;
+
+	sc->state = SC_ST_DIS;
+	if (sc->flags & SC_FL_ISBACK)
+		__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+	return;
+}
+
+/*
+ * This is the callback which is called by the connection layer to receive data
+ * into the buffer from the connection. It iterates over the mux layer's
+ * rcv_buf function. Please do not statify this function, it's often present in
+ * backtraces, it's useful to recognize it.
+ */
+int sc_conn_recv(struct stconn *sc)
+{
+	struct connection *conn = __sc_conn(sc);
+	struct channel *ic = sc_ic(sc);
+	int ret, max, cur_read = 0;
+	int read_poll = MAX_READ_POLL_LOOPS;
+	int flags = 0;
+
+	/* If not established yet, do nothing. */
+	if (sc->state != SC_ST_EST)
+		return 0;
+
+	/* If another call to sc_conn_recv() failed, and we subscribed to
+	 * recv events already, give up now.
+	 */
+	if ((sc->wait_event.events & SUB_RETRY_RECV) || sc_waiting_room(sc))
+		return 0;
+
+	/* maybe we were called immediately after an asynchronous abort */
+	if (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE))
+		return 1;
+
+	/* we must wait because the mux is not installed yet */
+	if (!conn->mux)
+		return 0;
+
+	/* stop immediately on errors. Note that we DON'T want to stop on
+	 * POLL_ERR, as the poller might report a write error while there
+	 * are still data available in the recv buffer. This typically
+	 * happens when we send too large a request to a backend server
+	 * which rejects it before reading it all.
+	 */
+	if (!sc_ep_test(sc, SE_FL_RCV_MORE)) {
+		if (!conn_xprt_ready(conn))
+			return 0;
+		if (sc_ep_test(sc, SE_FL_ERROR))
+			goto end_recv;
+	}
+
+	/* prepare to detect if the mux needs more room */
+	sc_ep_clr(sc, SE_FL_WANT_ROOM);
+
+	if ((ic->flags & (CF_STREAMER | CF_STREAMER_FAST)) && !co_data(ic) &&
+	    global.tune.idle_timer &&
+	    (unsigned short)(now_ms - ic->last_read) >= global.tune.idle_timer) {
+		/* The buffer was empty and nothing was transferred for more
+		 * than one second. This was caused by a pause and not by
+		 * congestion. Reset any streaming mode to reduce latency.
+		 */
+		ic->xfer_small = 0;
+		ic->xfer_large = 0;
+		ic->flags &= ~(CF_STREAMER | CF_STREAMER_FAST);
+	}
+
+#if defined(USE_LINUX_SPLICE)
+	/* Detect if the splicing is possible depending on the stream policy */
+	if ((global.tune.options & GTUNE_USE_SPLICE) &&
+	    (ic->to_forward >= MIN_SPLICE_FORWARD) &&
+	    ((!(sc->flags & SC_FL_ISBACK) && ((strm_fe(__sc_strm(sc))->options2|__sc_strm(sc)->be->options2) & PR_O2_SPLIC_REQ)) ||
+	     ((sc->flags & SC_FL_ISBACK) && ((strm_fe(__sc_strm(sc))->options2|__sc_strm(sc)->be->options2) & PR_O2_SPLIC_RTR)) ||
+	     ((ic->flags & CF_STREAMER_FAST) && ((strm_sess(__sc_strm(sc))->fe->options2|__sc_strm(sc)->be->options2) & PR_O2_SPLIC_AUT))))
+		flags |= CO_RFL_MAY_SPLICE;
+#endif
+
+	/* First, let's see if we may fast-forward data from a side to the other
+	 * one without using the channel buffer.
+	 */
+	if (sc_is_fastfwd_supported(sc)) {
+		if (channel_data(ic)) {
+			/* We're embarrassed, there are already data pending in
+			 * the buffer and we don't want to have them at two
+			 * locations at a time. Let's indicate we need some
+			 * place and ask the consumer to hurry.
+			 */
+			flags |= CO_RFL_BUF_FLUSH;
+			goto abort_fastfwd;
+		}
+		ret = conn->mux->fastfwd(sc, ic->to_forward, flags);
+		if (ret < 0)
+			goto abort_fastfwd;
+		else if (ret > 0) {
+			if (ic->to_forward != CHN_INFINITE_FORWARD)
+				ic->to_forward -= ret;
+			ic->total += ret;
+			cur_read += ret;
+			ic->flags |= CF_READ_EVENT;
+		}
+
+		if (sc_ep_test(sc, SE_FL_EOS | SE_FL_ERROR))
+			goto end_recv;
+
+		if (sc_ep_test(sc, SE_FL_WANT_ROOM))
+			sc_need_room(sc, -1);
+
+		if (sc_ep_test(sc, SE_FL_MAY_FASTFWD_PROD) && ic->to_forward)
+			goto done_recv;
+	}
+
+ abort_fastfwd:
+	/* now we'll need a input buffer for the stream */
+	if (!sc_alloc_ibuf(sc, &(__sc_strm(sc)->buffer_wait)))
+		goto end_recv;
+
+	/* For an HTX stream, if the buffer is stuck (no output data with some
+	 * input data) and if the HTX message is fragmented or if its free space
+	 * wraps, we force an HTX deframentation. It is a way to have a
+	 * contiguous free space nad to let the mux to copy as much data as
+	 * possible.
+	 *
+	 * NOTE: A possible optim may be to let the mux decides if defrag is
+	 *       required or not, depending on amount of data to be xferred.
+	 */
+	if (IS_HTX_STRM(__sc_strm(sc)) && !co_data(ic)) {
+		struct htx *htx = htxbuf(&ic->buf);
+
+		if (htx_is_not_empty(htx) && ((htx->flags & HTX_FL_FRAGMENTED) || htx_space_wraps(htx)))
+			htx_defrag(htx, NULL, 0);
+	}
+
+	/* Instruct the mux it must subscribed for read events */
+	if (!(sc->flags & SC_FL_ISBACK) &&                         /* for frontend conns only */
+	    (sc_opposite(sc)->state != SC_ST_INI) &&               /* before backend connection setup */
+	    (__sc_strm(sc)->be->options & PR_O_ABRT_CLOSE))        /* if abortonclose option is set for the current backend */
+		flags |= CO_RFL_KEEP_RECV;
+
+	/* Important note : if we're called with POLL_IN|POLL_HUP, it means the read polling
+	 * was enabled, which implies that the recv buffer was not full. So we have a guarantee
+	 * that if such an event is not handled above in splice, it will be handled here by
+	 * recv().
+	 */
+	while (sc_ep_test(sc, SE_FL_RCV_MORE) ||
+	       (!(conn->flags & CO_FL_HANDSHAKE) &&
+		(!sc_ep_test(sc, SE_FL_ERROR | SE_FL_EOS)) && !(sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE)))) {
+		int cur_flags = flags;
+
+		/* Compute transient CO_RFL_* flags */
+		if (co_data(ic)) {
+			cur_flags |= (CO_RFL_BUF_WET | CO_RFL_BUF_NOT_STUCK);
+		}
+
+		/* <max> may be null. This is the mux responsibility to set
+		 * SE_FL_RCV_MORE on the SC if more space is needed.
+		 */
+		max = channel_recv_max(ic);
+		ret = conn->mux->rcv_buf(sc, &ic->buf, max, cur_flags);
+
+		if (sc_ep_test(sc, SE_FL_WANT_ROOM)) {
+			/* SE_FL_WANT_ROOM must not be reported if the channel's
+			 * buffer is empty.
+			 */
+			BUG_ON(c_empty(ic));
+
+			sc_need_room(sc, channel_recv_max(ic) + 1);
+			/* Add READ_PARTIAL because some data are pending but
+			 * cannot be xferred to the channel
+			 */
+			ic->flags |= CF_READ_EVENT;
+			sc_ep_report_read_activity(sc);
+		}
+
+		if (ret <= 0) {
+			/* if we refrained from reading because we asked for a
+			 * flush to satisfy rcv_pipe(), we must not subscribe
+			 * and instead report that there's not enough room
+			 * here to proceed.
+			 */
+			if (flags & CO_RFL_BUF_FLUSH)
+				sc_need_room(sc, -1);
+			break;
+		}
+
+		cur_read += ret;
+
+		/* if we're allowed to directly forward data, we must update ->o */
+		if (ic->to_forward && !(sc_opposite(sc)->flags & (SC_FL_SHUT_DONE|SC_FL_SHUT_WANTED))) {
+			unsigned long fwd = ret;
+			if (ic->to_forward != CHN_INFINITE_FORWARD) {
+				if (fwd > ic->to_forward)
+					fwd = ic->to_forward;
+				ic->to_forward -= fwd;
+			}
+			c_adv(ic, fwd);
+		}
+
+		ic->flags |= CF_READ_EVENT;
+		ic->total += ret;
+
+		/* End-of-input reached, we can leave. In this case, it is
+		 * important to break the loop to not block the SC because of
+		 * the channel's policies.This way, we are still able to receive
+		 * shutdowns.
+		 */
+		if (sc_ep_test(sc, SE_FL_EOI))
+			break;
+
+		if ((sc->flags & SC_FL_RCV_ONCE) || --read_poll <= 0) {
+			/* we don't expect to read more data */
+			sc_wont_read(sc);
+			break;
+		}
+
+		/* if too many bytes were missing from last read, it means that
+		 * it's pointless trying to read again because the system does
+		 * not have them in buffers.
+		 */
+		if (ret < max) {
+			/* if a streamer has read few data, it may be because we
+			 * have exhausted system buffers. It's not worth trying
+			 * again.
+			 */
+			if (ic->flags & CF_STREAMER) {
+				/* we're stopped by the channel's policy */
+				sc_wont_read(sc);
+				break;
+			}
+
+			/* if we read a large block smaller than what we requested,
+			 * it's almost certain we'll never get anything more.
+			 */
+			if (ret >= global.tune.recv_enough) {
+				/* we're stopped by the channel's policy */
+				sc_wont_read(sc);
+				break;
+			}
+		}
+
+		/* if we are waiting for more space, don't try to read more data
+		 * right now.
+		 */
+		if (sc->flags & (SC_FL_WONT_READ|SC_FL_NEED_BUFF|SC_FL_NEED_ROOM))
+			break;
+	} /* while !flags */
+
+ done_recv:
+	if (!cur_read)
+		se_have_no_more_data(sc->sedesc);
+	else {
+		if ((ic->flags & (CF_STREAMER | CF_STREAMER_FAST)) &&
+		    (cur_read <= ic->buf.size / 2)) {
+			ic->xfer_large = 0;
+			ic->xfer_small++;
+			if (ic->xfer_small >= 3) {
+				/* we have read less than half of the buffer in
+				 * one pass, and this happened at least 3 times.
+				 * This is definitely not a streamer.
+				 */
+				ic->flags &= ~(CF_STREAMER | CF_STREAMER_FAST);
+			}
+			else if (ic->xfer_small >= 2) {
+				/* if the buffer has been at least half full twice,
+				 * we receive faster than we send, so at least it
+				 * is not a "fast streamer".
+				 */
+				ic->flags &= ~CF_STREAMER_FAST;
+			}
+		}
+		else if (!(ic->flags & CF_STREAMER_FAST) && (cur_read >= channel_data_limit(ic))) {
+			/* we read a full buffer at once */
+			ic->xfer_small = 0;
+			ic->xfer_large++;
+			if (ic->xfer_large >= 3) {
+				/* we call this buffer a fast streamer if it manages
+				 * to be filled in one call 3 consecutive times.
+				 */
+				ic->flags |= (CF_STREAMER | CF_STREAMER_FAST);
+			}
+		}
+		else {
+			ic->xfer_small = 0;
+			ic->xfer_large = 0;
+		}
+		ic->last_read = now_ms;
+		sc_ep_report_read_activity(sc);
+	}
+
+ end_recv:
+	ret = (cur_read != 0);
+
+	/* Report EOI on the channel if it was reached from the mux point of
+	 * view. */
+	if (sc_ep_test(sc, SE_FL_EOI) && !(sc->flags & SC_FL_EOI)) {
+		sc_ep_report_read_activity(sc);
+		sc->flags |= SC_FL_EOI;
+		ic->flags |= CF_READ_EVENT;
+		ret = 1;
+	}
+
+	if (sc_ep_test(sc, SE_FL_EOS)) {
+		/* we received a shutdown */
+		if (ic->flags & CF_AUTO_CLOSE)
+			sc_schedule_shutdown(sc_opposite(sc));
+		sc_conn_eos(sc);
+		ret = 1;
+	}
+
+	if (sc_ep_test(sc, SE_FL_ERROR)) {
+		sc->flags |= SC_FL_ERROR;
+		ret = 1;
+	}
+	else if (!cur_read &&
+		 !(sc->flags & (SC_FL_WONT_READ|SC_FL_NEED_BUFF|SC_FL_NEED_ROOM)) &&
+		 !(sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE))) {
+		/* Subscribe to receive events if we're blocking on I/O */
+		conn->mux->subscribe(sc, SUB_RETRY_RECV, &sc->wait_event);
+		se_have_no_more_data(sc->sedesc);
+	}
+	else {
+		se_have_more_data(sc->sedesc);
+		ret = 1;
+	}
+
+	return ret;
+}
+
+/* This tries to perform a synchronous receive on the stream connector to
+ * try to collect last arrived data. In practice it's only implemented on
+ * stconns. Returns 0 if nothing was done, non-zero if new data or a
+ * shutdown were collected. This may result on some delayed receive calls
+ * to be programmed and performed later, though it doesn't provide any
+ * such guarantee.
+ */
+int sc_conn_sync_recv(struct stconn *sc)
+{
+	if (!sc_state_in(sc->state, SC_SB_RDY|SC_SB_EST))
+		return 0;
+
+	if (!sc_mux_ops(sc))
+		return 0; // only stconns are supported
+
+	if (sc->wait_event.events & SUB_RETRY_RECV)
+		return 0; // already subscribed
+
+	if (!sc_is_recv_allowed(sc))
+		return 0; // already failed
+
+	return sc_conn_recv(sc);
+}
+
+/*
+ * This function is called to send buffer data to a stream socket.
+ * It calls the mux layer's snd_buf function. It relies on the
+ * caller to commit polling changes. The caller should check conn->flags
+ * for errors. Please do not statify this function, it's often present in
+ * backtraces, it's useful to recognize it.
+ */
+int sc_conn_send(struct stconn *sc)
+{
+	struct connection *conn = __sc_conn(sc);
+	struct stconn *sco = sc_opposite(sc);
+	struct stream *s = __sc_strm(sc);
+	struct channel *oc = sc_oc(sc);
+	int ret;
+	int did_send = 0;
+
+	if (sc_ep_test(sc, SE_FL_ERROR | SE_FL_ERR_PENDING) || sc_is_conn_error(sc)) {
+		/* We're probably there because the tasklet was woken up,
+		 * but process_stream() ran before, detected there were an
+		 * error and put the SC back to SC_ST_TAR. There's still
+		 * CO_FL_ERROR on the connection but we don't want to add
+		 * SE_FL_ERROR back, so give up
+		 */
+		if (sc->state < SC_ST_CON)
+			return 0;
+		BUG_ON(sc_ep_test(sc, SE_FL_EOS|SE_FL_ERROR|SE_FL_ERR_PENDING) == (SE_FL_EOS|SE_FL_ERR_PENDING));
+		return 1;
+	}
+
+	/* We're already waiting to be able to send, give up */
+	if (sc->wait_event.events & SUB_RETRY_SEND)
+		return 0;
+
+	/* we might have been called just after an asynchronous shutw */
+	if (sc->flags & SC_FL_SHUT_DONE)
+		return 1;
+
+	/* we must wait because the mux is not installed yet */
+	if (!conn->mux)
+		return 0;
+
+	if (sc_ep_have_ff_data(sc)) {
+		unsigned int send_flag = 0;
+
+		if ((!(sc->flags & (SC_FL_SND_ASAP|SC_FL_SND_NEVERWAIT)) &&
+		     ((oc->to_forward && oc->to_forward != CHN_INFINITE_FORWARD) ||
+		      (sc->flags & SC_FL_SND_EXP_MORE) ||
+		      (IS_HTX_STRM(s) &&
+		       (!(sco->flags & (SC_FL_EOI|SC_FL_EOS|SC_FL_ABRT_DONE)) && htx_expect_more(htxbuf(&oc->buf)))))) ||
+		    ((oc->flags & CF_ISRESP) &&
+		     (oc->flags & CF_AUTO_CLOSE) &&
+		     (sc->flags & SC_FL_SHUT_WANTED)))
+			send_flag |= CO_SFL_MSG_MORE;
+
+		if (oc->flags & CF_STREAMER)
+			send_flag |= CO_SFL_STREAMER;
+
+		ret = conn->mux->resume_fastfwd(sc, send_flag);
+		if (ret > 0)
+			did_send = 1;
+
+		if (sc_ep_have_ff_data(sc))
+			goto end;
+	}
+
+	/* At this point, the pipe is empty, but we may still have data pending
+	 * in the normal buffer.
+	 */
+	if (co_data(oc)) {
+		/* when we're here, we already know that there is no spliced
+		 * data left, and that there are sendable buffered data.
+		 */
+
+		/* check if we want to inform the kernel that we're interested in
+		 * sending more data after this call. We want this if :
+		 *  - we're about to close after this last send and want to merge
+		 *    the ongoing FIN with the last segment.
+		 *  - we know we can't send everything at once and must get back
+		 *    here because of unaligned data
+		 *  - there is still a finite amount of data to forward
+		 * The test is arranged so that the most common case does only 2
+		 * tests.
+		 */
+		unsigned int send_flag = 0;
+
+		if ((!(sc->flags & (SC_FL_SND_ASAP|SC_FL_SND_NEVERWAIT)) &&
+		     ((oc->to_forward && oc->to_forward != CHN_INFINITE_FORWARD) ||
+		      (sc->flags & SC_FL_SND_EXP_MORE) ||
+		      (IS_HTX_STRM(s) &&
+		       (!(sco->flags & (SC_FL_EOI|SC_FL_EOS|SC_FL_ABRT_DONE)) && htx_expect_more(htxbuf(&oc->buf)))))) ||
+		    ((oc->flags & CF_ISRESP) &&
+		     (oc->flags & CF_AUTO_CLOSE) &&
+		     (sc->flags & SC_FL_SHUT_WANTED)))
+			send_flag |= CO_SFL_MSG_MORE;
+
+		if (oc->flags & CF_STREAMER)
+			send_flag |= CO_SFL_STREAMER;
+
+		if (s->txn && s->txn->flags & TX_L7_RETRY && !b_data(&s->txn->l7_buffer)) {
+			/* If we want to be able to do L7 retries, copy
+			 * the data we're about to send, so that we are able
+			 * to resend them if needed
+			 */
+			/* Try to allocate a buffer if we had none.
+			 * If it fails, the next test will just
+			 * disable the l7 retries by setting
+			 * l7_conn_retries to 0.
+			 */
+			if (s->txn->req.msg_state != HTTP_MSG_DONE)
+				s->txn->flags &= ~TX_L7_RETRY;
+			else {
+				if (b_alloc(&s->txn->l7_buffer) == NULL)
+					s->txn->flags &= ~TX_L7_RETRY;
+				else {
+					memcpy(b_orig(&s->txn->l7_buffer),
+					       b_orig(&oc->buf),
+					       b_size(&oc->buf));
+					s->txn->l7_buffer.head = co_data(oc);
+					b_add(&s->txn->l7_buffer, co_data(oc));
+				}
+
+			}
+		}
+
+		ret = conn->mux->snd_buf(sc, &oc->buf, co_data(oc), send_flag);
+		if (ret > 0) {
+			did_send = 1;
+			c_rew(oc, ret);
+			c_realign_if_empty(oc);
+
+			if (!co_data(oc)) {
+				/* Always clear both flags once everything has been sent, they're one-shot */
+				sc->flags &= ~(SC_FL_SND_ASAP|SC_FL_SND_EXP_MORE);
+			}
+			/* if some data remain in the buffer, it's only because the
+			 * system buffers are full, we will try next time.
+			 */
+		}
+	}
+
+ end:
+	if (did_send) {
+		oc->flags |= CF_WRITE_EVENT | CF_WROTE_DATA;
+		if (sc->state == SC_ST_CON)
+			sc->state = SC_ST_RDY;
+	}
+
+	if (!sco->room_needed || (did_send && (sco->room_needed < 0 || channel_recv_max(sc_oc(sc)) >= sco->room_needed)))
+		sc_have_room(sco);
+
+	if (sc_ep_test(sc, SE_FL_ERROR | SE_FL_ERR_PENDING)) {
+		oc->flags |= CF_WRITE_EVENT;
+		BUG_ON(sc_ep_test(sc, SE_FL_EOS|SE_FL_ERROR|SE_FL_ERR_PENDING) == (SE_FL_EOS|SE_FL_ERR_PENDING));
+		if (sc_ep_test(sc, SE_FL_ERROR))
+			sc->flags |= SC_FL_ERROR;
+		return 1;
+	}
+
+	/* FIXME: Must be reviewed for FF */
+	if (!co_data(oc) && !sc_ep_have_ff_data(sc)) {
+		if (did_send)
+			sc_ep_report_send_activity(sc);
+		/* If fast-forwarding is blocked, unblock it now to check for
+		 * receive on the other side
+		 */
+		if (sc->sedesc->iobuf.flags & IOBUF_FL_FF_BLOCKED) {
+			sc->sedesc->iobuf.flags &= ~IOBUF_FL_FF_BLOCKED;
+			sc_have_room(sco);
+			did_send = 1;
+		}
+	}
+	else {
+		/* We couldn't send all of our data, let the mux know we'd like to send more */
+		conn->mux->subscribe(sc, SUB_RETRY_SEND, &sc->wait_event);
+		if (sc_state_in(sc->state, SC_SB_EST|SC_SB_DIS|SC_SB_CLO))
+				sc_ep_report_blocked_send(sc, did_send);
+	}
+
+	return did_send;
+}
+
+/* perform a synchronous send() for the stream connector. The CF_WRITE_EVENT
+ * flag are cleared prior to the attempt, and will possibly be updated in case
+ * of success.
+ */
+void sc_conn_sync_send(struct stconn *sc)
+{
+	struct channel *oc = sc_oc(sc);
+
+	oc->flags &= ~CF_WRITE_EVENT;
+
+	if (sc->flags & SC_FL_SHUT_DONE)
+		return;
+
+	if (!co_data(oc))
+		return;
+
+	if (!sc_state_in(sc->state, SC_SB_CON|SC_SB_RDY|SC_SB_EST))
+		return;
+
+	if (!sc_mux_ops(sc))
+		return;
+
+	sc_conn_send(sc);
+}
+
+/* Called by I/O handlers after completion.. It propagates
+ * connection flags to the stream connector, updates the stream (which may or
+ * may not take this opportunity to try to forward data), then update the
+ * connection's polling based on the channels and stream connector's final
+ * states. The function always returns 0. Please do not statify this function,
+ * it's often present in backtraces, it's useful to recognize it.
+ */
+int sc_conn_process(struct stconn *sc)
+{
+	struct connection *conn = __sc_conn(sc);
+	struct channel *ic = sc_ic(sc);
+	struct channel *oc = sc_oc(sc);
+
+	BUG_ON(!conn);
+
+	/* If we have data to send, try it now */
+	if ((co_data(oc) || sc_ep_have_ff_data(sc)) &&
+	    !(sc->wait_event.events & SUB_RETRY_SEND))
+		sc_conn_send(sc);
+
+	/* First step, report to the stream connector what was detected at the
+	 * connection layer : errors and connection establishment.
+	 * Only add SC_FL_ERROR if we're connected, or we're attempting to
+	 * connect, we may get there because we got woken up, but only run
+	 * after process_stream() noticed there were an error, and decided
+	 * to retry to connect, the connection may still have CO_FL_ERROR,
+	 * and we don't want to add SC_FL_ERROR back
+	 *
+	 * Note: This test is only required because sc_conn_process is also the SI
+	 *       wake callback. Otherwise sc_conn_recv()/sc_conn_send() already take
+	 *       care of it.
+	 */
+
+	if (sc->state >= SC_ST_CON) {
+		if (sc_is_conn_error(sc))
+			sc->flags |= SC_FL_ERROR;
+	}
+
+	/* If we had early data, and the handshake ended, then
+	 * we can remove the flag, and attempt to wake the task up,
+	 * in the event there's an analyser waiting for the end of
+	 * the handshake.
+	 */
+	if (!(conn->flags & (CO_FL_WAIT_XPRT | CO_FL_EARLY_SSL_HS)) &&
+	    sc_ep_test(sc, SE_FL_WAIT_FOR_HS)) {
+		sc_ep_clr(sc, SE_FL_WAIT_FOR_HS);
+		task_wakeup(sc_strm_task(sc), TASK_WOKEN_MSG);
+	}
+
+	if (!sc_state_in(sc->state, SC_SB_EST|SC_SB_DIS|SC_SB_CLO) &&
+	    (conn->flags & CO_FL_WAIT_XPRT) == 0) {
+		if (sc->flags & SC_FL_ISBACK)
+			__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+		oc->flags |= CF_WRITE_EVENT;
+		if (sc->state == SC_ST_CON)
+			sc->state = SC_ST_RDY;
+	}
+
+	/* Report EOS on the channel if it was reached from the mux point of
+	 * view.
+	 *
+	 * Note: This test is only required because sc_conn_process is also the SI
+	 *       wake callback. Otherwise sc_conn_recv()/sc_conn_send() already take
+	 *       care of it.
+	 */
+	if (sc_ep_test(sc, SE_FL_EOS) && !(sc->flags & SC_FL_EOS)) {
+		/* we received a shutdown */
+		if (ic->flags & CF_AUTO_CLOSE)
+			sc_schedule_shutdown(sc_opposite(sc));
+		sc_conn_eos(sc);
+	}
+
+	/* Report EOI on the channel if it was reached from the mux point of
+	 * view.
+	 *
+	 * Note: This test is only required because sc_conn_process is also the SI
+	 *       wake callback. Otherwise sc_conn_recv()/sc_conn_send() already take
+	 *       care of it.
+	 */
+	if (sc_ep_test(sc, SE_FL_EOI) && !(sc->flags & SC_FL_EOI)) {
+		sc->flags |= SC_FL_EOI;
+		ic->flags |= CF_READ_EVENT;
+		sc_ep_report_read_activity(sc);
+	}
+
+	if (sc_ep_test(sc, SE_FL_ERROR))
+		sc->flags |= SC_FL_ERROR;
+
+	/* Second step : update the stream connector and channels, try to forward any
+	 * pending data, then possibly wake the stream up based on the new
+	 * stream connector status.
+	 */
+	sc_notify(sc);
+	stream_release_buffers(__sc_strm(sc));
+	return 0;
+}
+
+/* This is the ->process() function for any stream connector's wait_event task.
+ * It's assigned during the stream connector's initialization, for any type of
+ * stream connector. Thus it is always safe to perform a tasklet_wakeup() on a
+ * stream connector, as the presence of the SC is checked there.
+ */
+struct task *sc_conn_io_cb(struct task *t, void *ctx, unsigned int state)
+{
+	struct stconn *sc = ctx;
+	int ret = 0;
+
+	if (!sc_conn(sc))
+		return t;
+
+	if (!(sc->wait_event.events & SUB_RETRY_SEND) && (co_data(sc_oc(sc)) || sc_ep_have_ff_data(sc) || (sc->sedesc->iobuf.flags & IOBUF_FL_FF_BLOCKED)))
+		ret = sc_conn_send(sc);
+	if (!(sc->wait_event.events & SUB_RETRY_RECV))
+		ret |= sc_conn_recv(sc);
+	if (ret != 0)
+		sc_conn_process(sc);
+
+	stream_release_buffers(__sc_strm(sc));
+	return t;
+}
+
+/*
+ * This function propagates an end-of-stream received from an applet. It
+ * updates the stream connector. If it is is already shut, the applet is
+ * released. Otherwise, we try to forward the shutdown, immediately or ASAP.
+ */
+static void sc_applet_eos(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+
+	BUG_ON(!sc_appctx(sc));
+
+	if (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE))
+		return;
+	sc->flags |= SC_FL_EOS;
+	ic->flags |= CF_READ_EVENT;
+	sc_ep_report_read_activity(sc);
+
+	/* Note: on abort, we don't call the applet */
+
+	if (!sc_state_in(sc->state, SC_SB_CON|SC_SB_RDY|SC_SB_EST))
+		return;
+
+	if (sc->flags & SC_FL_SHUT_DONE) {
+		appctx_shut(__sc_appctx(sc));
+		sc->state = SC_ST_DIS;
+		if (sc->flags & SC_FL_ISBACK)
+			__sc_strm(sc)->conn_exp = TICK_ETERNITY;
+	}
+	else if (sc_cond_forward_shut(sc))
+		return sc_app_shut_applet(sc);
+}
+
+/* Callback to be used by applet handlers upon completion. It updates the stream
+ * (which may or may not take this opportunity to try to forward data), then
+ * may re-enable the applet's based on the channels and stream connector's final
+ * states. Please do not statify this function, it's often present in backtraces,
+ * it's useful to recognize it.
+ */
+int sc_applet_process(struct stconn *sc)
+{
+	struct channel *ic = sc_ic(sc);
+
+	BUG_ON(!sc_appctx(sc));
+
+	/* Report EOI on the channel if it was reached from the applet point of
+	 * view. */
+	if (sc_ep_test(sc, SE_FL_EOI) && !(sc->flags & SC_FL_EOI)) {
+		sc_ep_report_read_activity(sc);
+		sc->flags |= SC_FL_EOI;
+		ic->flags |= CF_READ_EVENT;
+	}
+
+	if (sc_ep_test(sc, SE_FL_ERROR))
+		sc->flags |= SC_FL_ERROR;
+
+	if (sc_ep_test(sc, SE_FL_EOS)) {
+		/* we received a shutdown */
+		sc_applet_eos(sc);
+	}
+
+	BUG_ON(sc_ep_test(sc, SE_FL_HAVE_NO_DATA|SE_FL_EOI) == SE_FL_EOI);
+
+	/* If the applet wants to write and the channel is closed, it's a
+	 * broken pipe and it must be reported.
+	 */
+	if (!sc_ep_test(sc, SE_FL_HAVE_NO_DATA) && (sc->flags & (SC_FL_EOS|SC_FL_ABRT_DONE)))
+		sc_ep_set(sc, SE_FL_ERROR);
+
+	/* automatically mark the applet having data available if it reported
+	 * begin blocked by the channel.
+	 */
+	if ((sc->flags & (SC_FL_WONT_READ|SC_FL_NEED_BUFF|SC_FL_NEED_ROOM)) ||
+	    sc_ep_test(sc, SE_FL_APPLET_NEED_CONN))
+		applet_have_more_data(__sc_appctx(sc));
+
+	/* update the stream connector, channels, and possibly wake the stream up */
+	sc_notify(sc);
+	stream_release_buffers(__sc_strm(sc));
+
+	/* sc_notify may have passed through chk_snd and released some blocking
+	 * flags. Process_stream will consider those flags to wake up the
+	 * appctx but in the case the task is not in runqueue we may have to
+	 * wakeup the appctx immediately.
+	 */
+	if (sc_is_recv_allowed(sc) || sc_is_send_allowed(sc))
+		appctx_wakeup(__sc_appctx(sc));
+	return 0;
+}
+
+
+/* Prepares an endpoint upgrade. We don't now at this stage if the upgrade will
+ * succeed or not and if the stconn will be reused by the new endpoint. Thus,
+ * for now, only pretend the stconn is detached.
+ */
+void sc_conn_prepare_endp_upgrade(struct stconn *sc)
+{
+	BUG_ON(!sc_conn(sc) || !sc->app);
+	sc_ep_clr(sc, SE_FL_T_MUX);
+	sc_ep_set(sc, SE_FL_DETACHED);
+}
+
+/* Endpoint upgrade failed. Restore the stconn state. */
+void sc_conn_abort_endp_upgrade(struct stconn *sc)
+{
+	sc_ep_set(sc, SE_FL_T_MUX);
+	sc_ep_clr(sc, SE_FL_DETACHED);
+}
+
+/* Commit the endpoint upgrade. If stconn is attached, it means the new endpoint
+ * use it. So we do nothing. Otherwise, the stconn will be destroy with the
+ * overlying stream. So, it means we must commit the detach.
+*/
+void sc_conn_commit_endp_upgrade(struct stconn *sc)
+{
+	if (!sc_ep_test(sc, SE_FL_DETACHED))
+		return;
+	sc_detach_endp(&sc);
+	/* Because it was already set as detached, the sedesc must be preserved */
+	BUG_ON(!sc);
+	BUG_ON(!sc->sedesc);
+}
+
+/* return the frontend or backend mux stream ID.
+ */
+static int
+smp_fetch_sid(const struct arg *args, struct sample *smp, const char *kw, void *private)
+{
+	struct connection *conn;
+	struct stconn *sc;
+	int64_t sid = 0;
+
+	if (!smp->strm)
+		return 0;
+
+	sc = (kw[0] == 'f' ? smp->strm->scf : smp->strm->scb);
+	conn = sc_conn(sc);
+
+	/* No connection */
+	if (!conn)
+		return 0;
+
+	/* No mux install, this may change */
+	if (!conn->mux) {
+		smp->flags |= SMP_F_MAY_CHANGE;
+		return 0;
+	}
+
+	/* No sctl, report sid=0 in this case */
+	if (conn->mux->sctl) {
+		if (conn->mux->sctl(sc, MUX_SCTL_SID, &sid) == -1)
+			return 0;
+	}
+
+	smp->flags = SMP_F_VOL_TXN;
+	smp->data.type = SMP_T_SINT;
+	smp->data.u.sint = sid;
+
+	return 1;
+}
+
+/* Note: must not be declared <const> as its list will be overwritten.
+ * Note: fetches that may return multiple types should be declared using the
+ * appropriate pseudo-type. If not available it must be declared as the lowest
+ * common denominator, the type that can be casted into all other ones.
+ */
+static struct sample_fetch_kw_list sample_fetch_keywords = {ILH, {
+	{ "bs.id", smp_fetch_sid, 0, NULL, SMP_T_SINT, SMP_USE_L6REQ },
+	{ "fs.id", smp_fetch_sid, 0, NULL, SMP_T_STR, SMP_USE_L6RES },
+	{ /* END */ },
+}};
+
+INITCALL1(STG_REGISTER, sample_register_fetches, &sample_fetch_keywords);