h2/h3: replace state.drain counter with state.dselect_bits

- `drain` was used by http/2 and http/3 implementations to indicate
  that the transfer requires send/recv independant from its socket
  poll state. Intended as a counter, it was used as bool flag only.
- a similar mechanism exists on `connectdata->cselect_bits` where
  specific protocols can indicate something similar, only for the
  whole connection.
- `cselect_bits` are cleard in transfer.c on use and, importantly,
  also set when the transfer loop expended its `maxloops` tries.
  `drain` was not cleared by transfer and the http2/3 implementations
  had to take care of that.
- `dselect_bits` is cleared *and* set by the transfer loop. http2/3
  does no longer clear it, only set when new events happen.

This change unifies the handling of socket poll overrides, extending
`cselect_bits` by a easy handle specific value and a common treatment in
transfers.

Closes #11005

lib/vquic/curl_msh3.c

@@ -189,12 +189,33 @@ static void h3_data_done(struct Curl_cfilter *cf, struct Curl_easy *data)
   }
 }
 
-static void notify_drain(struct Curl_cfilter *cf,
+static void drain_stream_from_other_thread(struct Curl_easy *data,
+                                           struct stream_ctx *stream)
+{
+  int bits;
+
+  /* risky */
+  bits = CURL_CSELECT_IN;
+  if(stream && !stream->upload_done)
+    bits |= CURL_CSELECT_OUT;
+  if(data->state.dselect_bits != bits) {
+    data->state.dselect_bits = bits;
+    /* cannot expire from other thread */
+  }
+}
+
+static void drain_stream(struct Curl_cfilter *cf,
                          struct Curl_easy *data)
 {
+  struct stream_ctx *stream = H3_STREAM_CTX(data);
+  int bits;
+
   (void)cf;
-  if(!data->state.drain) {
-    data->state.drain = 1;
+  bits = CURL_CSELECT_IN;
+  if(stream && !stream->upload_done)
+    bits |= CURL_CSELECT_OUT;
+  if(data->state.dselect_bits != bits) {
+    data->state.dselect_bits = bits;
     Curl_expire(data, 0, EXPIRE_RUN_NOW);
   }
 }
@@ -350,7 +371,7 @@ static void MSH3_CALL msh3_header_received(MSH3_REQUEST *Request,
     }
   }
 
-  data->state.drain = 1;
+  drain_stream_from_other_thread(data, stream);
   msh3_lock_release(&stream->recv_lock);
 }
 
@@ -469,7 +490,6 @@ static ssize_t recv_closed_stream(struct Curl_cfilter *cf,
   nread = 0;
 
 out:
-  data->state.drain = 0;
   return nread;
 }
 
@@ -508,7 +528,6 @@ static ssize_t cf_msh3_recv(struct Curl_cfilter *cf, struct Curl_easy *data,
 
   if(stream->recv_error) {
     failf(data, "request aborted");
-    data->state.drain = 0;
     *err = stream->recv_error;
     goto out;
   }
@@ -522,10 +541,8 @@ static ssize_t cf_msh3_recv(struct Curl_cfilter *cf, struct Curl_easy *data,
                   len, nread, *err));
     if(nread < 0)
       goto out;
-    if(!Curl_bufq_is_empty(&stream->recvbuf) ||
-       stream->closed) {
-       notify_drain(cf, data);
-    }
+    if(stream->closed)
+       drain_stream(cf, data);
   }
   else if(stream->closed) {
     nread = recv_closed_stream(cf, data, err);
@@ -669,15 +686,14 @@ static int cf_msh3_get_select_socks(struct Curl_cfilter *cf,
 
     if(stream->recv_error) {
       bitmap |= GETSOCK_READSOCK(0);
-      notify_drain(cf, data);
+      drain_stream(cf, data);
     }
     else if(stream->req) {
       bitmap |= GETSOCK_READSOCK(0);
-      notify_drain(cf, data);
+      drain_stream(cf, data);
     }
   }
-  DEBUGF(LOG_CF(data, cf, "select_sock %u -> %d",
-                (uint32_t)data->state.drain, bitmap));
+  DEBUGF(LOG_CF(data, cf, "select_sock -> %d", bitmap));
   CF_DATA_RESTORE(cf, save);
   return bitmap;
 }
@@ -698,6 +714,8 @@ static bool cf_msh3_data_pending(struct Curl_cfilter *cf,
                   Curl_bufq_len(&stream->recvbuf)));
     pending = !Curl_bufq_is_empty(&stream->recvbuf);
     msh3_lock_release(&stream->recv_lock);
+    if(pending)
+      drain_stream(cf, (struct Curl_easy *)data);
   }
 
   CF_DATA_RESTORE(cf, save);

lib/vquic/curl_ngtcp2.c

@@ -709,11 +709,6 @@ static void report_consumed_data(struct Curl_cfilter *cf,
                                          consumed);
     ngtcp2_conn_extend_max_offset(ctx->qconn, consumed);
   }
-  if(!stream->closed && data->state.drain &&
-     Curl_bufq_is_empty(&stream->recvbuf)) {
-     /* nothing buffered any more */
-     data->state.drain = 0;
-  }
 }
 
 static int cb_recv_stream_data(ngtcp2_conn *tconn, uint32_t flags,
@@ -995,12 +990,18 @@ static int cf_ngtcp2_get_select_socks(struct Curl_cfilter *cf,
   return rv;
 }
 
-static void notify_drain(struct Curl_cfilter *cf,
+static void drain_stream(struct Curl_cfilter *cf,
                          struct Curl_easy *data)
 {
+  struct stream_ctx *stream = H3_STREAM_CTX(data);
+  int bits;
+
   (void)cf;
-  if(!data->state.drain) {
-    data->state.drain = 1;
+  bits = CURL_CSELECT_IN;
+  if(stream && !stream->upload_done)
+    bits |= CURL_CSELECT_OUT;
+  if(data->state.dselect_bits != bits) {
+    data->state.dselect_bits = bits;
     Curl_expire(data, 0, EXPIRE_RUN_NOW);
   }
 }
@@ -1028,7 +1029,7 @@ static int cb_h3_stream_close(nghttp3_conn *conn, int64_t stream_id,
   if(app_error_code == NGHTTP3_H3_INTERNAL_ERROR) {
     stream->reset = TRUE;
   }
-  notify_drain(cf, data);
+  drain_stream(cf, data);
   return 0;
 }
 
@@ -1082,9 +1083,7 @@ static int cb_h3_recv_data(nghttp3_conn *conn, int64_t stream3_id,
   (void)stream3_id;
 
   result = write_resp_raw(cf, data, buf, buflen, TRUE);
-  if(CF_DATA_CURRENT(cf) != data) {
-    notify_drain(cf, data);
-  }
+  drain_stream(cf, data);
   return result? -1 : 0;
 }
 
@@ -1129,9 +1128,7 @@ static int cb_h3_end_headers(nghttp3_conn *conn, int64_t stream_id,
   if(stream->status_code / 100 != 1) {
     stream->resp_hds_complete = TRUE;
   }
-  if(CF_DATA_CURRENT(cf) != data) {
-    notify_drain(cf, data);
-  }
+  drain_stream(cf, data);
   return 0;
 }
 
@@ -1358,7 +1355,6 @@ static ssize_t recv_closed_stream(struct Curl_cfilter *cf,
   nread = 0;
 
 out:
-  data->state.drain = 0;
   return nread;
 }
 
@@ -1413,16 +1409,13 @@ static ssize_t cf_ngtcp2_recv(struct Curl_cfilter *cf, struct Curl_easy *data,
   }
 
   if(nread > 0) {
-    if(1 || !Curl_bufq_is_empty(&stream->recvbuf)) {
-      notify_drain(cf, data);
-    }
+    drain_stream(cf, data);
   }
   else {
     if(stream->closed) {
       nread = recv_closed_stream(cf, data, err);
       goto out;
     }
-    data->state.drain = FALSE;
     *err = CURLE_AGAIN;
     nread = -1;
   }
@@ -1468,7 +1461,7 @@ static int cb_h3_acked_req_body(nghttp3_conn *conn, int64_t stream_id,
     if((data->req.keepon & KEEP_SEND_HOLD) &&
        (data->req.keepon & KEEP_SEND)) {
       data->req.keepon &= ~KEEP_SEND_HOLD;
-      notify_drain(cf, data);
+      drain_stream(cf, data);
       DEBUGF(LOG_CF(data, cf, "[h3sid=%" PRId64 "] unpausing acks",
                     stream_id));
     }

lib/vquic/curl_quiche.c

@@ -299,11 +299,18 @@ static void h3_data_done(struct Curl_cfilter *cf, struct Curl_easy *data)
   }
 }
 
-static void notify_drain(struct Curl_cfilter *cf, struct Curl_easy *data)
+static void drain_stream(struct Curl_cfilter *cf,
+                         struct Curl_easy *data)
 {
+  struct stream_ctx *stream = H3_STREAM_CTX(data);
+  int bits;
+
   (void)cf;
-  if(!data->state.drain) {
-    data->state.drain = 1;
+  bits = CURL_CSELECT_IN;
+  if(stream && !stream->upload_done)
+    bits |= CURL_CSELECT_OUT;
+  if(data->state.dselect_bits != bits) {
+    data->state.dselect_bits = bits;
     Curl_expire(data, 0, EXPIRE_RUN_NOW);
   }
 }
@@ -579,9 +586,7 @@ static CURLcode cf_poll_events(struct Curl_cfilter *cf,
     }
     else {
       result = h3_process_event(cf, sdata, stream3_id, ev);
-      if(sdata != data) {
-        notify_drain(cf, sdata);
-      }
+      drain_stream(cf, sdata);
       if(result) {
         DEBUGF(LOG_CF(data, cf, "[h3sid=%"PRId64"] error processing event %s "
                       "for [h3sid=%"PRId64"] -> %d",
@@ -848,15 +853,20 @@ static ssize_t cf_quiche_recv(struct Curl_cfilter *cf, struct Curl_easy *data,
   }
 
   if(nread > 0) {
-    data->state.drain = (!Curl_bufq_is_empty(&stream->recvbuf) ||
-                         stream->closed);
+    if(stream->closed)
+      drain_stream(cf, data);
   }
   else {
-    data->state.drain = FALSE;
     if(stream->closed) {
       nread = recv_closed_stream(cf, data, err);
       goto out;
     }
+    else if(quiche_conn_is_draining(ctx->qconn)) {
+      failf(data, "QUIC connection is draining");
+      *err = CURLE_HTTP3;
+      nread = -1;
+      goto out;
+    }
     *err = CURLE_AGAIN;
     nread = -1;
   }
@@ -1065,24 +1075,9 @@ static bool stream_is_writeable(struct Curl_cfilter *cf,
 {
   struct cf_quiche_ctx *ctx = cf->ctx;
   struct stream_ctx *stream = H3_STREAM_CTX(data);
-  quiche_stream_iter *qiter;
-  bool is_writable = FALSE;
 
-  if(!stream)
-    return FALSE;
-  /* surely, there must be a better way */
-  qiter = quiche_conn_writable(ctx->qconn);
-  if(qiter) {
-    uint64_t stream_id;
-    while(quiche_stream_iter_next(qiter, &stream_id)) {
-      if(stream_id == (uint64_t)stream->id) {
-        is_writable = TRUE;
-        break;
-      }
-    }
-    quiche_stream_iter_free(qiter);
-  }
-  return is_writable;
+  return stream &&
+         quiche_conn_stream_writable(ctx->qconn, (uint64_t)stream->id, 1);
 }
 
 static int cf_quiche_get_select_socks(struct Curl_cfilter *cf,
@@ -1152,7 +1147,8 @@ static CURLcode cf_quiche_data_event(struct Curl_cfilter *cf,
   }
   case CF_CTRL_DATA_IDLE:
     result = cf_flush_egress(cf, data);
-    DEBUGF(LOG_CF(data, cf, "data idle, flush egress -> %d", result));
+    if(result)
+      DEBUGF(LOG_CF(data, cf, "data idle, flush egress -> %d", result));
     break;
   default:
     break;

lib/vquic/vquic.c

@@ -398,7 +398,6 @@ static CURLcode recvmsg_packets(struct Curl_cfilter *cf,
       ;
     if(nread == -1) {
       if(SOCKERRNO == EAGAIN || SOCKERRNO == EWOULDBLOCK) {
-        DEBUGF(LOG_CF(data, cf, "ingress, recvmsg -> EAGAIN"));
         goto out;
       }
       if(!cf->connected && SOCKERRNO == ECONNREFUSED) {