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.

diff --git a/include/haproxy/stconn.h b/include/haproxy/stconn.h
index 1a6b2509379077904fe7d1a7201b4a3dafef06f2..2582ba38d7ec030e618a892957b4d2ec218858d3 100644 (file)
--- a/include/haproxy/stconn.h
+++ b/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 */

diff --git a/src/applet.c b/src/applet.c
index 1d7306a1398cd8de8e0345f8932a22fddc0dd6d0..275e4e65c7dff1bf77073c5299f6a006a9a12197 100644 (file)
--- a/src/applet.c
+++ b/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);

diff --git a/src/stats.c b/src/stats.c
index bf826afe3203cde4cdb0ba7f4bfb9f40d294d65a..50943b9906e8aad3c9868fbb48bafa35ee51ad9e 100644 (file)
--- a/src/stats.c
+++ b/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;
                }
        }