http2: better support for --limit-rate

- leave transfer loop when --limit-rate is in effect and has
  been received
- adjust stream window size to --limit-rate plus some slack
  to make the server observe the pacing we want
- add test case to confirm behaviour

Closes #11115

lib/http2.c

@@ -183,6 +183,7 @@ struct stream_ctx {
 
   int status_code; /* HTTP response status code */
   uint32_t error; /* stream error code */
+  uint32_t local_window_size; /* the local recv window size */
   bool closed; /* TRUE on stream close */
   bool reset;  /* TRUE on stream reset */
   bool close_handled; /* TRUE if stream closure is handled by libcurl */
@@ -252,6 +253,7 @@ static CURLcode http2_data_setup(struct Curl_cfilter *cf,
   stream->closed = FALSE;
   stream->close_handled = FALSE;
   stream->error = NGHTTP2_NO_ERROR;
+  stream->local_window_size = H2_STREAM_WINDOW_SIZE;
   stream->upload_left = 0;
 
   H2_STREAM_LCTX(data) = stream;
@@ -964,6 +966,7 @@ static CURLcode on_stream_frame(struct Curl_cfilter *cf,
   struct stream_ctx *stream = H2_STREAM_CTX(data);
   int32_t stream_id = frame->hd.stream_id;
   CURLcode result;
+  size_t rbuflen;
   int rv;
 
   if(!stream) {
@@ -973,10 +976,10 @@ static CURLcode on_stream_frame(struct Curl_cfilter *cf,
 
   switch(frame->hd.type) {
   case NGHTTP2_DATA:
+    rbuflen = Curl_bufq_len(&stream->recvbuf);
     DEBUGF(LOG_CF(data, cf, "[h2sid=%d] FRAME[DATA len=%zu pad=%zu], "
                   "buffered=%zu, window=%d/%d",
-                  stream_id, frame->hd.length, frame->data.padlen,
-                  Curl_bufq_len(&stream->recvbuf),
+                  stream_id, frame->hd.length, frame->data.padlen, rbuflen,
                   nghttp2_session_get_stream_effective_recv_data_length(
                     ctx->h2, stream->id),
                   nghttp2_session_get_stream_effective_local_window_size(
@@ -993,6 +996,20 @@ static CURLcode on_stream_frame(struct Curl_cfilter *cf,
     if(frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
       drain_stream(cf, data, stream);
     }
+    else if(rbuflen > stream->local_window_size) {
+      int32_t wsize = nghttp2_session_get_stream_local_window_size(
+                        ctx->h2, stream->id);
+      if(wsize > 0 && (uint32_t)wsize != stream->local_window_size) {
+        /* H2 flow control is not absolute, as the server might not have the
+         * same view, yet. When we recieve more than we want, we enforce
+         * the local window size again to make nghttp2 send WINDOW_UPATEs
+         * accordingly. */
+        nghttp2_session_set_local_window_size(ctx->h2,
+                                              NGHTTP2_FLAG_NONE,
+                                              stream->id,
+                                              stream->local_window_size);
+      }
+    }
     break;
   case NGHTTP2_HEADERS:
     DEBUGF(LOG_CF(data, cf, "[h2sid=%d] FRAME[HEADERS]", stream_id));
@@ -1969,6 +1986,19 @@ static ssize_t h2_submit(struct stream_ctx **pstream,
   infof(data, "Using Stream ID: %u (easy handle %p)",
         stream_id, (void *)data);
   stream->id = stream_id;
+  stream->local_window_size = H2_STREAM_WINDOW_SIZE;
+  if(data->set.max_recv_speed) {
+    /* We are asked to only receive `max_recv_speed` bytes per second.
+     * Let's limit our stream window size around that, otherwise the server
+     * will send in large bursts only. We make the window 50% larger to
+     * allow for data in flight and avoid stalling. */
+    size_t n = (((data->set.max_recv_speed - 1) / H2_CHUNK_SIZE) + 1);
+    n += CURLMAX((n/2), 1);
+    if(n < (H2_STREAM_WINDOW_SIZE / H2_CHUNK_SIZE) &&
+       n < (UINT_MAX / H2_CHUNK_SIZE)) {
+      stream->local_window_size = (uint32_t)n * H2_CHUNK_SIZE;
+    }
+  }
 
 out:
   DEBUGF(LOG_CF(data, cf, "[h2sid=%d] submit -> %zd, %d",
@@ -2104,6 +2134,7 @@ static ssize_t cf_h2_send(struct Curl_cfilter *cf, struct Curl_easy *data,
       }
       goto out;
     }
+
   }
 
 out:
@@ -2241,7 +2272,7 @@ static CURLcode http2_data_pause(struct Curl_cfilter *cf,
 
   DEBUGASSERT(data);
   if(ctx && ctx->h2 && stream) {
-    uint32_t window = !pause * H2_STREAM_WINDOW_SIZE;
+    uint32_t window = pause? 0 : stream->local_window_size;
     CURLcode result;
 
     int rv = nghttp2_session_set_local_window_size(ctx->h2,

lib/transfer.c

@@ -428,6 +428,8 @@ static CURLcode readwrite_data(struct Curl_easy *data,
   size_t excess = 0; /* excess bytes read */
   bool readmore = FALSE; /* used by RTP to signal for more data */
   int maxloops = 100;
+  curl_off_t max_recv = data->set.max_recv_speed?
+                        data->set.max_recv_speed : CURL_OFF_T_MAX;
   char *buf = data->state.buffer;
   DEBUGASSERT(buf);
 
@@ -666,6 +668,7 @@ static CURLcode readwrite_data(struct Curl_easy *data,
       }
 
       k->bytecount += nread;
+      max_recv -= nread;
 
       Curl_pgrsSetDownloadCounter(data, k->bytecount);
 
@@ -749,9 +752,9 @@ static CURLcode readwrite_data(struct Curl_easy *data,
       break;
     }
 
-  } while(data_pending(data) && maxloops--);
+  } while((max_recv > 0) && data_pending(data) && maxloops--);
 
-  if(maxloops <= 0) {
+  if(maxloops <= 0 || max_recv <= 0) {
     /* we mark it as read-again-please */
     data->state.dselect_bits = CURL_CSELECT_IN;
     *comeback = TRUE;

tests/http/test_02_download.py

@@ -28,6 +28,7 @@ import difflib
 import filecmp
 import logging
 import os
+from datetime import timedelta
 import pytest
 
 from testenv import Env, CurlClient, LocalClient
@@ -334,6 +335,21 @@ class TestDownload:
         # downloads should be there, but not necessarily complete
         self.check_downloads(client, srcfile, count, complete=False)
 
+    # speed limited download
+    @pytest.mark.parametrize("proto", ['h2', 'h3'])
+    def test_02_24_speed_limit(self, env: Env, httpd, nghttpx, proto, repeat):
+        if proto == 'h3' and not env.have_h3():
+            pytest.skip("h3 not supported")
+        count = 1
+        url = f'https://{env.authority_for(env.domain1, proto)}/data-1m'
+        curl = CurlClient(env=env)
+        r = curl.http_download(urls=[url], alpn_proto=proto, extra_args=[
+            '--limit-rate', f'{196 * 1024}'
+        ])
+        r.check_response(count=count, http_status=200)
+        assert r.duration > timedelta(seconds=4), \
+            f'rate limited transfer should take more than 4s, not {r.duration}'
+
     def check_downloads(self, client, srcfile: str, count: int,
                         complete: bool = True):
         for i in range(count):