HTTP/[23]: continue upload when state.drain is set

- as reported in #10433, HTTP/2 uploads may stall when a response is
  received before the upload is done. This happens when the
  data->state.drain is set for such a transfer, as the special handling
  in transfer.c from then on only cared about downloads.
- add continuation of uploads, if applicable, in this case.
- add pytest case test_07_12_upload_seq_large to reproduce this scenario
  (although, current nghttp2 implementation is using drain less often)

Reported-by: Lucas Pardue

Fixes #10433
Closes #10443

lib/http2.c

@@ -592,11 +592,12 @@ char *curl_pushheader_byname(struct curl_pushheaders *h, const char *header)
 static void drained_transfer(struct Curl_cfilter *cf,
                              struct Curl_easy *data)
 {
-  struct cf_h2_ctx *ctx = cf->ctx;
-
-  DEBUGASSERT(ctx->drain_total >= data->state.drain);
-  ctx->drain_total -= data->state.drain;
-  data->state.drain = 0;
+  if(data->state.drain) {
+    struct cf_h2_ctx *ctx = cf->ctx;
+    DEBUGASSERT(ctx->drain_total > 0);
+    ctx->drain_total--;
+    data->state.drain = 0;
+  }
 }
 
 /*
@@ -605,11 +606,12 @@ static void drained_transfer(struct Curl_cfilter *cf,
 static void drain_this(struct Curl_cfilter *cf,
                        struct Curl_easy *data)
 {
-  struct cf_h2_ctx *ctx = cf->ctx;
-
-  data->state.drain++;
-  ctx->drain_total++;
-  DEBUGASSERT(ctx->drain_total >= data->state.drain);
+  if(!data->state.drain) {
+    struct cf_h2_ctx *ctx = cf->ctx;
+    data->state.drain = 1;
+    ctx->drain_total++;
+    DEBUGASSERT(ctx->drain_total > 0);
+  }
 }
 
 static struct Curl_easy *h2_duphandle(struct Curl_cfilter *cf,
@@ -1575,8 +1577,6 @@ static ssize_t http2_handle_stream_close(struct Curl_cfilter *cf,
     }
   }
 
-  DEBUGASSERT(data->state.drain == 0);
-
   /* Reset to FALSE to prevent infinite loop in readwrite_data function. */
   stream->closed = FALSE;
   if(stream->error == NGHTTP2_REFUSED_STREAM) {
@@ -1929,8 +1929,9 @@ static ssize_t cf_h2_recv(struct Curl_cfilter *cf, struct Curl_easy *data,
       drain_this(cf, data);
       Curl_expire(data, 0, EXPIRE_RUN_NOW);
     }
-    else
+    else {
       drained_transfer(cf, data);
+    }
 
     nread = retlen;
     DEBUGF(LOG_CF(data, cf, "[h2sid=%u] cf_h2_recv -> %zd",

lib/transfer.c

@@ -1080,6 +1080,8 @@ CURLcode Curl_readwrite(struct connectdata *conn,
   if(data->state.drain) {
     select_res |= CURL_CSELECT_IN;
     DEBUGF(infof(data, "Curl_readwrite: forcibly told to drain data"));
+    if((k->keepon & KEEP_SENDBITS) == KEEP_SEND)
+      select_res |= CURL_CSELECT_OUT;
   }
 #endif
 

tests/tests-httpd/test_07_upload.py

@@ -39,13 +39,11 @@ log = logging.getLogger(__name__)
 class TestUpload:
 
     @pytest.fixture(autouse=True, scope='class')
-    def _class_scope(self, env, nghttpx):
+    def _class_scope(self, env, httpd, nghttpx):
         if env.have_h3():
             nghttpx.start_if_needed()
-        s90 = "01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678\n"
-        with open(os.path.join(env.gen_dir, "data-100k"), 'w') as f:
-            for i in range(1000):
-                f.write(f"{i:09d}-{s90}")
+        env.make_data_file(indir=env.gen_dir, fname="data-100k", fsize=100*1024)
+        env.make_data_file(indir=env.gen_dir, fname="data-10m", fsize=10*1024*1024)
 
     # upload small data, check that this is what was echoed
     @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
@@ -110,6 +108,24 @@ class TestUpload:
             respdata = open(curl.response_file(i)).readlines()
             assert respdata == indata
 
+    # upload very large data sequentially, check that this is what was echoed
+    @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
+    def test_07_12_upload_seq_large(self, env: Env, httpd, nghttpx, repeat, proto):
+        if proto == 'h3' and not env.have_h3():
+            pytest.skip("h3 not supported")
+        fdata = os.path.join(env.gen_dir, 'data-10m')
+        count = 2
+        curl = CurlClient(env=env)
+        url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo?id=[0-{count-1}]'
+        r = curl.http_upload(urls=[url], data=f'@{fdata}', alpn_proto=proto)
+        assert r.exit_code == 0, f'{r}'
+        r.check_stats(count=count, exp_status=200)
+        indata = open(fdata).readlines()
+        r.check_stats(count=count, exp_status=200)
+        for i in range(count):
+            respdata = open(curl.response_file(i)).readlines()
+            assert respdata == indata
+
     # upload data parallel, check that they were echoed
     @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
     def test_07_20_upload_parallel(self, env: Env, httpd, nghttpx, repeat, proto):

tests/tests-httpd/testenv/env.py

@@ -82,14 +82,14 @@ class EnvConfig:
                         lib.lower() for lib in m.group('libs').split(' ')
                     ]
                     self.curl_props['libs'] = [
-                        re.sub(r'/.*', '',lib) for lib in self.curl_props['lib_versions']
+                        re.sub(r'/.*', '', lib) for lib in self.curl_props['lib_versions']
                     ]
             if l.startswith('Features: '):
                 self.curl_props['features'] = [
                     feat.lower() for feat in l[10:].split(' ')
                 ]
             if l.startswith('Protocols: '):
-                self.curl_props['protocols'] =  [
+                self.curl_props['protocols'] = [
                     prot.lower() for prot in l[11:].split(' ')
                 ]
         self.nghttpx_with_h3 = re.match(r'.* nghttp3/.*', p.stdout.strip())
@@ -222,11 +222,11 @@ class Env:
         return 'unknown'
 
     @staticmethod
-    def curl_os() -> bool:
+    def curl_os() -> str:
         return Env.CONFIG.curl_props['os']
 
     @staticmethod
-    def curl_version() -> bool:
+    def curl_version() -> str:
         return Env.CONFIG.curl_props['version']
 
     @staticmethod
@@ -336,3 +336,17 @@ class Env:
         if alpn_proto in ['h3']:
             return f'{domain}:{self.h3_port}'
         return f'{domain}:{self.http_port}'
+
+    def make_data_file(self, indir: str, fname: str, fsize: int) -> str:
+        fpath = os.path.join(indir, fname)
+        s10 = "0123456789"
+        s = (101 * s10) + s10[0:3]
+        with open(fpath, 'w') as fd:
+            for i in range(int(fsize / 1024)):
+                fd.write(f"{i:09d}-{s}\n")
+            remain = int(fsize % 1024)
+            if remain != 0:
+                i = int(fsize / 1024) + 1
+                s = f"{i:09d}-{s}\n"
+                fd.write(s[0:remain])
+        return fpath