BUG/MEDIUM: stconn/applet: Block 0-copy forwarding if producer needs more room

This case does not exist yet with the H1 multiplexer, but applets may decide to
not produce data if there is not enough room in the destination buffer (the
applet's outbuf or the opposite SE buffer). It is true for the stats applets for
instance. However this case is not properly handled when the zero-copy
forwarding is in-use.

To fix the issue, the se_done_ff() function was modified to return the number of
bytes really forwarded and to subs for sends if nothing was forwarded while the
zero-copy forwarding was blocked by the producer. On the applet side, we take
care to block the zero-copy forwarding if the applet requests more room. At the
end, zero-copy forwarding is unblocked if something was forwarded.

This way, it is now possible for the stats applet to report a full buffer and
block the zero-copy forwarding, even if the buffer is not really full, by
requesting more room.

No backport needed.

include/haproxy/stconn.h

@@ -537,21 +537,53 @@ static inline size_t se_nego_ff(struct sedesc *se, struct buffer *input, size_t
 	return ret;
 }
 
-static inline void se_done_ff(struct sedesc *se)
+/* Returns the number of bytes forwarded. May be 0 if nothing is forwarded. It
+ * may also be 0 if there is nothing to forward. Note it is not dependent on
+ * data in the buffer but only on the amount of data to forward.
+ */
+static inline size_t se_done_ff(struct sedesc *se)
 {
+	size_t ret = 0;
+
 	if (se_fl_test(se, SE_FL_T_MUX)) {
 		const struct mux_ops *mux = se->conn->mux;
-		size_t sent, to_send = se_ff_data(se);
+		size_t to_send = se_ff_data(se);
 
 		BUG_ON(!mux->done_fastfwd);
-		sent = mux->done_fastfwd(se->sc);
-		if (to_send) {
-			if (sent == to_send)
+		ret = mux->done_fastfwd(se->sc);
+		if (ret) {
+			/* Something was forwarded, unblock the zero-copy forwarding.
+			 * If all data was sent, report and send activity.
+			 * Otherwise report a conditional blocked send.
+			 */
+			se->iobuf.flags &= ~IOBUF_FL_FF_BLOCKED;
+			if (ret == to_send)
 				sc_ep_report_send_activity(se->sc);
 			else
-				sc_ep_report_blocked_send(se->sc, sent != 0);
+				sc_ep_report_blocked_send(se->sc, 1);
+		}
+		else {
+			/* Nothing was forwarded. If there was something to forward,
+			 * it means the sends are blocked.
+			 * In addition, if the zero-copy forwarding is blocked because the
+			 * producer requests more room, we must subs for sends.
+			 */
+			if (to_send)
+				sc_ep_report_blocked_send(se->sc, 0);
+			if (se->iobuf.flags & IOBUF_FL_FF_BLOCKED) {
+				sc_ep_report_blocked_send(se->sc, 0);
+				
+				if (!(se->sc->wait_event.events & SUB_RETRY_SEND)) {
+					/* The SC must be subs for send to be notify when some
+					 * space is made
+					 */
+					mux->subscribe(se->sc, SUB_RETRY_SEND, &se->sc->wait_event);
+				}
+			}
 		}
 	}
+
+	return ret;
 }
 
 #endif /* _HAPROXY_STCONN_H */

src/applet.c

@@ -690,6 +690,12 @@ int appctx_fastfwd(struct stconn *sc, unsigned int count, unsigned int flags)
 	b_sub(sdo->iobuf.buf, sdo->iobuf.offset);
 	sdo->iobuf.data += ret;
 
+	if (se_fl_test(appctx->sedesc, SE_FL_WANT_ROOM)) {
+		/* The applet request more room, report the info at the iobuf level */
+		sdo->iobuf.flags |= IOBUF_FL_FF_BLOCKED;
+		TRACE_STATE("waiting for more room", APPLET_EV_RECV|APPLET_EV_BLK, appctx);
+	}
+
 	if (applet_fl_test(appctx, APPCTX_FL_EOI)) {
 		se_fl_set(appctx->sedesc, SE_FL_EOI);
 		sdo->iobuf.flags |= IOBUF_FL_EOI; /* TODO: it may be good to have a flag to be sure we can
@@ -708,7 +714,11 @@ int appctx_fastfwd(struct stconn *sc, unsigned int count, unsigned int flags)
 	/* else */
 	/* 	applet_have_more_data(appctx); */
 
-	se_done_ff(sdo);
+	if (se_done_ff(sdo) != 0) {
+		/* Something was forwarding, don't reclaim more room */
+		se_fl_clr(appctx->sedesc, SE_FL_WANT_ROOM);
+		TRACE_STATE("more room available", APPLET_EV_RECV|APPLET_EV_BLK, appctx);
+	}
 
 end:
 	TRACE_LEAVE(APPLET_EV_RECV, appctx);

src/stats.c

@@ -316,8 +316,10 @@ int stats_putchk(struct appctx *appctx, struct buffer *buf, struct htx *htx)
 		chk->data = 0;
 	}
 	else if (buf) {
-		if (b_data(chk) > b_room(buf))
+		if (b_data(chk) > b_room(buf)) {
+			se_fl_set(appctx->sedesc, SE_FL_RCV_MORE | SE_FL_WANT_ROOM);
 			return 0;
+		}
 		b_putblk(buf, b_head(chk), b_data(chk));
 		chk->data = 0;
 	}
@@ -339,6 +341,7 @@ int stats_is_full(struct appctx *appctx, struct buffer *buf, struct htx *htx)
 	}
 	else if (buf) {
 		if (buffer_almost_full(buf)) {
+			se_fl_set(appctx->sedesc, SE_FL_RCV_MORE | SE_FL_WANT_ROOM);
 			goto full;
 		}
 	}