MEDIUM: h2: process streams pending for sending

The send() callback calls h2_process_mux() which iterates over the list
of flow controlled streams first, then streams waiting for room in the
send_list. If a stream from the send_list ends up being flow controlled,
it is then moved to the fctl_list. This way we can maintain the most
accurate fairness by ensuring that flows are always processed in order
of arrival except when they're blocked by flow control, in which case
only the other ones may pass in front of them.

It's a bit tricky as we want to remove a stream from the active lists
if it doesn't block (ie it has no reason for staying there).

src/mux_h2.c

@@ -701,6 +701,84 @@ static void h2_process_demux(struct h2c *h2c)
  */
 static int h2_process_mux(struct h2c *h2c)
 {
+	struct h2s *h2s, *h2s_back;
+
+	/* First we always process the flow control list because the streams
+	 * waiting there were already elected for immediate emission but were
+	 * blocked just on this.
+	 */
+
+	list_for_each_entry_safe(h2s, h2s_back, &h2c->fctl_list, list) {
+		if (h2c->mws <= 0 || h2c->flags & H2_CF_MUX_BLOCK_ANY ||
+		    h2c->st0 >= H2_CS_ERROR)
+			break;
+
+		/* In theory it's possible that h2s->cs == NULL here :
+		 *  - client sends crap that causes a parse error
+		 *  - RST_STREAM is produced and CS_FL_ERROR at the same time
+		 *  - RST_STREAM cannot be emitted because mux is busy/full
+		 *  - stream gets notified, detaches and quits
+		 *  - mux buffer gets ready and wakes pending streams up
+		 *  - bam!
+		 */
+		h2s->flags &= ~H2_SF_BLK_ANY;
+
+		if (h2s->cs) {
+			h2s->cs->data_cb->send(h2s->cs);
+			h2s->cs->data_cb->wake(h2s->cs);
+		}
+
+		/* depending on callee's blocking reasons, we may queue in send
+		 * list or completely dequeue.
+		 */
+		if ((h2s->flags & H2_SF_BLK_MFCTL) == 0) {
+			if (h2s->flags & H2_SF_BLK_ANY) {
+				LIST_DEL(&h2s->list);
+				LIST_ADDQ(&h2c->send_list, &h2s->list);
+			}
+			else {
+				LIST_DEL(&h2s->list);
+				LIST_INIT(&h2s->list);
+				if (h2s->cs)
+					h2s->cs->flags &= ~CS_FL_DATA_WR_ENA;
+			}
+		}
+	}
+
+	list_for_each_entry_safe(h2s, h2s_back, &h2c->send_list, list) {
+		if (h2c->st0 >= H2_CS_ERROR || h2c->flags & H2_CF_MUX_BLOCK_ANY)
+			break;
+
+		/* In theory it's possible that h2s->cs == NULL here :
+		 *  - client sends crap that causes a parse error
+		 *  - RST_STREAM is produced and CS_FL_ERROR at the same time
+		 *  - RST_STREAM cannot be emitted because mux is busy/full
+		 *  - stream gets notified, detaches and quits
+		 *  - mux buffer gets ready and wakes pending streams up
+		 *  - bam!
+		 */
+		h2s->flags &= ~H2_SF_BLK_ANY;
+
+		if (h2s->cs) {
+			h2s->cs->data_cb->send(h2s->cs);
+			h2s->cs->data_cb->wake(h2s->cs);
+		}
+		/* depending on callee's blocking reasons, we may queue in fctl
+		 * list or completely dequeue.
+		 */
+		if (h2s->flags & H2_SF_BLK_MFCTL) {
+			/* stream hit the connection's flow control */
+			LIST_DEL(&h2s->list);
+			LIST_ADDQ(&h2c->fctl_list, &h2s->list);
+		}
+		else if (!(h2s->flags & H2_SF_BLK_ANY)) {
+			LIST_DEL(&h2s->list);
+			LIST_INIT(&h2s->list);
+			if (h2s->cs)
+				h2s->cs->flags &= ~CS_FL_DATA_WR_ENA;
+		}
+	}
+
 	if (unlikely(h2c->st0 > H2_CS_ERROR)) {
 		if (h2c->st0 == H2_CS_ERROR) {
 			if (h2c->max_id >= 0) {
@@ -713,7 +791,7 @@ static int h2_process_mux(struct h2c *h2c)
 		}
 		return 1;
 	}
-	return 1;
+	return (h2c->mws <= 0 || LIST_ISEMPTY(&h2c->fctl_list)) && LIST_ISEMPTY(&h2c->send_list);
 }