cf-socket: improvements in socket I/O handling

- Curl_write_plain/Curl_read_plain have been eliminated. Last code use
  now uses Curl_conn_send/recv so that requests use conn->send/revc
  callbacks which defaults to cfilters use.
- Curl_recv_plain/Curl_send_plain have been internalized in cf-socket.c.
- USE_RECV_BEFORE_SEND_WORKAROUND (active on Windows) has been moved
  into cf-socket.c. The pre_recv buffer is held at the socket filter
  context.  `postponed_data` structures have been removed from
  `connectdata`.
- the hanger in HTTP/2 request handling was a result of read buffering
  on all sends and the multi handling is not prepared for this. The
  following happens:

   - multi preforms on a HTTP/2 easy handle
   - h2 reads and processes data
   - this leads to a send of h2 data
   - which receives and buffers before the send
   - h2 returns
   - multi selects on the socket, but no data arrives (its in the buffer already)
   the workaround now receives data in a loop as long as there is something in
   the buffer. The real fix would be for multi to change, so that `data_pending`
   is evaluated before deciding to wait on the socket.

io_buffer, optional, in cf-socket.c, http/2 sets state.drain if lower
filter have pending data.

This io_buffer is only available/used when the
-DUSE_RECV_BEFORE_SEND_WORKAROUND is active, e.g. on Windows
configurations. It also maintains the original checks on protocol
handler being HTTP and conn->send/recv not being replaced.

The HTTP/2 (nghttp2) cfilter now sets data->state.drain when it finds
out that the "lower" filter chain has still pending data at the end of
its IO operation. This prevents the processing from becoming stalled.

Closes #10280

1 files changed, 140 insertions, 141 deletions

diff --git a/lib/socks.c b/lib/socks.c
index b8b4f8cb8..ea1e2c17a 100644
--- a/lib/socks.c
+++ b/lib/socks.c
@@ -89,8 +89,8 @@ struct socks_state {
  *
  * This is STUPID BLOCKING behavior. Only used by the SOCKS GSSAPI functions.
  */
-int Curl_blockread_all(struct Curl_easy *data,   /* transfer */
-                       curl_socket_t sockfd,     /* read from this socket */
+int Curl_blockread_all(struct Curl_cfilter *cf,
+                       struct Curl_easy *data,   /* transfer */
                        char *buf,                /* store read data here */
                        ssize_t buffersize,       /* max amount to read */
                        ssize_t *n)               /* amount bytes read */
@@ -98,6 +98,8 @@ int Curl_blockread_all(struct Curl_easy *data,   /* transfer */
   ssize_t nread = 0;
   ssize_t allread = 0;
   int result;
+  CURLcode err = CURLE_OK;
+
   *n = 0;
   for(;;) {
     timediff_t timeout_ms = Curl_timeleft(data, NULL, TRUE);
@@ -108,15 +110,19 @@ int Curl_blockread_all(struct Curl_easy *data,   /* transfer */
     }
     if(!timeout_ms)
       timeout_ms = TIMEDIFF_T_MAX;
-    if(SOCKET_READABLE(sockfd, timeout_ms) <= 0) {
+    if(SOCKET_READABLE(cf->conn->sock[cf->sockindex], timeout_ms) <= 0) {
       result = ~CURLE_OK;
       break;
     }
-    result = Curl_read_plain(data, sockfd, buf, buffersize, &nread);
-    if(CURLE_AGAIN == result)
-      continue;
-    if(result)
-      break;
+    nread = Curl_conn_cf_recv(cf->next, data, buf, buffersize, &err);
+    if(nread <= 0) {
+      result = err;
+      if(CURLE_AGAIN == err)
+        continue;
+      if(err) {
+        break;
+      }
+    }
 
     if(buffersize == nread) {
       allread += nread;
@@ -192,6 +198,68 @@ static void socksstate(struct socks_state *sx, struct Curl_easy *data,
 #endif
 }
 
+static CURLproxycode socks_state_send(struct Curl_cfilter *cf,
+                                      struct socks_state *sx,
+                                      struct Curl_easy *data,
+                                      CURLproxycode failcode,
+                                      const char *description)
+{
+  ssize_t nwritten;
+  CURLcode result;
+
+  nwritten = Curl_conn_cf_send(cf->next, data, (char *)sx->outp,
+                               sx->outstanding, &result);
+  if(nwritten <= 0) {
+    if(CURLE_AGAIN == result) {
+      return CURLPX_OK;
+    }
+    else if(CURLE_OK == result) {
+      /* connection closed */
+      failf(data, "connection to proxy closed");
+      return CURLPX_CLOSED;
+    }
+    failf(data, "Failed to send %s: %s", description,
+          curl_easy_strerror(result));
+    return failcode;
+  }
+  DEBUGASSERT(sx->outstanding >= nwritten);
+  /* not done, remain in state */
+  sx->outstanding -= nwritten;
+  sx->outp += nwritten;
+  return CURLPX_OK;
+}
+
+static CURLproxycode socks_state_recv(struct Curl_cfilter *cf,
+                                      struct socks_state *sx,
+                                      struct Curl_easy *data,
+                                      CURLproxycode failcode,
+                                      const char *description)
+{
+  ssize_t nread;
+  CURLcode result;
+
+  nread = Curl_conn_cf_recv(cf->next, data, (char *)sx->outp,
+                            sx->outstanding, &result);
+  if(nread <= 0) {
+    if(CURLE_AGAIN == result) {
+      return CURLPX_OK;
+    }
+    else if(CURLE_OK == result) {
+      /* connection closed */
+      failf(data, "connection to proxy closed");
+      return CURLPX_CLOSED;
+    }
+    failf(data, "SOCKS4: Failed receiving %s: %s", description,
+          curl_easy_strerror(result));
+    return failcode;
+  }
+  /* remain in reading state */
+  DEBUGASSERT(sx->outstanding >= nread);
+  sx->outstanding -= nread;
+  sx->outp += nread;
+  return CURLPX_OK;
+}
+
 /*
 * This function logs in to a SOCKS4 proxy and sends the specifics to the final
 * destination server.
@@ -212,10 +280,8 @@ static CURLproxycode do_SOCKS4(struct Curl_cfilter *cf,
     (conn->socks_proxy.proxytype == CURLPROXY_SOCKS4A) ? TRUE : FALSE;
   unsigned char *socksreq = (unsigned char *)data->state.buffer;
   CURLcode result;
-  curl_socket_t sockfd = conn->sock[cf->sockindex];
+  CURLproxycode presult;
   struct Curl_dns_entry *dns = NULL;
-  ssize_t actualread;
-  ssize_t written;
 
   /* make sure that the buffer is at least 600 bytes */
   DEBUGASSERT(READBUFFER_MIN >= 600);
@@ -375,19 +441,14 @@ static CURLproxycode do_SOCKS4(struct Curl_cfilter *cf,
     /* FALLTHROUGH */
   case CONNECT_REQ_SENDING:
     /* Send request */
-    result = Curl_write_plain(data, sockfd, (char *)sx->outp,
-                              sx->outstanding, &written);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Failed to send SOCKS4 connect request.");
-      return CURLPX_SEND_CONNECT;
-    }
-    if(written != sx->outstanding) {
-      /* not done, remain in state */
-      sx->outstanding -= written;
-      sx->outp += written;
+    presult = socks_state_send(cf, sx, data, CURLPX_SEND_CONNECT,
+                               "SOCKS4 connect request");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
+      /* remain in sending state */
       return CURLPX_OK;
     }
-
     /* done sending! */
     sx->outstanding = 8; /* receive data size */
     sx->outp = socksreq;
@@ -396,22 +457,12 @@ static CURLproxycode do_SOCKS4(struct Curl_cfilter *cf,
     /* FALLTHROUGH */
   case CONNECT_SOCKS_READ:
     /* Receive response */
-    result = Curl_read_plain(data, sockfd, (char *)sx->outp,
-                             sx->outstanding, &actualread);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "SOCKS4: Failed receiving connect request ack: %s",
-            curl_easy_strerror(result));
-      return CURLPX_RECV_CONNECT;
-    }
-    else if(!result && !actualread) {
-      /* connection closed */
-      failf(data, "connection to proxy closed");
-      return CURLPX_CLOSED;
-    }
-    else if(actualread != sx->outstanding) {
+    presult = socks_state_recv(cf, sx, data, CURLPX_RECV_CONNECT,
+                               "connect request ack");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
       /* remain in reading state */
-      sx->outstanding -= actualread;
-      sx->outp += actualread;
       return CURLPX_OK;
     }
     sxstate(sx, data, CONNECT_DONE);
@@ -518,10 +569,8 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
   unsigned char *socksreq = (unsigned char *)data->state.buffer;
   char dest[256] = "unknown";  /* printable hostname:port */
   int idx;
-  ssize_t actualread;
-  ssize_t written;
   CURLcode result;
-  curl_socket_t sockfd = conn->sock[cf->sockindex];
+  CURLproxycode presult;
   bool socks5_resolve_local =
     (conn->socks_proxy.proxytype == CURLPROXY_SOCKS5) ? TRUE : FALSE;
   const size_t hostname_len = strlen(sx->hostname);
@@ -567,30 +616,25 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
     /* write the number of authentication methods */
     socksreq[1] = (unsigned char) (idx - 2);
 
-    result = Curl_write_plain(data, sockfd, socksreq, idx, &written);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Unable to send initial SOCKS5 request.");
-      return CURLPX_SEND_CONNECT;
-    }
-    if(written != idx) {
-      sxstate(sx, data, CONNECT_SOCKS_SEND);
-      sx->outstanding = idx - written;
-      sx->outp = &socksreq[written];
+    sx->outp = socksreq;
+    sx->outstanding = idx;
+    presult = socks_state_send(cf, sx, data, CURLPX_SEND_CONNECT,
+                               "initial SOCKS5 request");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
+      /* remain in sending state */
       return CURLPX_OK;
     }
     sxstate(sx, data, CONNECT_SOCKS_READ);
     goto CONNECT_SOCKS_READ_INIT;
   case CONNECT_SOCKS_SEND:
-    result = Curl_write_plain(data, sockfd, (char *)sx->outp,
-                              sx->outstanding, &written);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Unable to send initial SOCKS5 request.");
-      return CURLPX_SEND_CONNECT;
-    }
-    if(written != sx->outstanding) {
-      /* not done, remain in state */
-      sx->outstanding -= written;
-      sx->outp += written;
+    presult = socks_state_send(cf, sx, data, CURLPX_SEND_CONNECT,
+                               "initial SOCKS5 request");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
+      /* remain in sending state */
       return CURLPX_OK;
     }
     /* FALLTHROUGH */
@@ -600,21 +644,12 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
     sx->outp = socksreq; /* store it here */
     /* FALLTHROUGH */
   case CONNECT_SOCKS_READ:
-    result = Curl_read_plain(data, sockfd, (char *)sx->outp,
-                             sx->outstanding, &actualread);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Unable to receive initial SOCKS5 response.");
-      return CURLPX_RECV_CONNECT;
-    }
-    else if(!result && !actualread) {
-      /* connection closed */
-      failf(data, "Connection to proxy closed");
-      return CURLPX_CLOSED;
-    }
-    else if(actualread != sx->outstanding) {
+    presult = socks_state_recv(cf, sx, data, CURLPX_RECV_CONNECT,
+                               "initial SOCKS5 response");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
       /* remain in reading state */
-      sx->outstanding -= actualread;
-      sx->outp += actualread;
       return CURLPX_OK;
     }
     else if(socksreq[0] != 5) {
@@ -634,7 +669,7 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
 #if defined(HAVE_GSSAPI) || defined(USE_WINDOWS_SSPI)
     else if(allow_gssapi && (socksreq[1] == 1)) {
       sxstate(sx, data, CONNECT_GSSAPI_INIT);
-      result = Curl_SOCKS5_gssapi_negotiate(cf->sockindex, data);
+      result = Curl_SOCKS5_gssapi_negotiate(cf, data);
       if(result) {
         failf(data, "Unable to negotiate SOCKS5 GSS-API context.");
         return CURLPX_GSSAPI;
@@ -713,16 +748,12 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
   }
     /* FALLTHROUGH */
   case CONNECT_AUTH_SEND:
-    result = Curl_write_plain(data, sockfd, sx->outp,
-                              sx->outstanding, &written);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Failed to send SOCKS5 sub-negotiation request.");
-      return CURLPX_SEND_AUTH;
-    }
-    if(sx->outstanding != written) {
-      /* remain in state */
-      sx->outstanding -= written;
-      sx->outp += written;
+    presult = socks_state_send(cf, sx, data, CURLPX_SEND_AUTH,
+                               "SOCKS5 sub-negotiation request");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
+      /* remain in sending state */
       return CURLPX_OK;
     }
     sx->outp = socksreq;
@@ -730,21 +761,12 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
     sxstate(sx, data, CONNECT_AUTH_READ);
     /* FALLTHROUGH */
   case CONNECT_AUTH_READ:
-    result = Curl_read_plain(data, sockfd, (char *)sx->outp,
-                             sx->outstanding, &actualread);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Unable to receive SOCKS5 sub-negotiation response.");
-      return CURLPX_RECV_AUTH;
-    }
-    else if(!result && !actualread) {
-      /* connection closed */
-      failf(data, "connection to proxy closed");
-      return CURLPX_CLOSED;
-    }
-    else if(actualread != sx->outstanding) {
-      /* remain in state */
-      sx->outstanding -= actualread;
-      sx->outp += actualread;
+    presult = socks_state_recv(cf, sx, data, CURLPX_RECV_AUTH,
+                               "SOCKS5 sub-negotiation response");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
+      /* remain in reading state */
       return CURLPX_OK;
     }
     /* ignore the first (VER) byte */
@@ -909,16 +931,12 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
     sxstate(sx, data, CONNECT_REQ_SENDING);
     /* FALLTHROUGH */
   case CONNECT_REQ_SENDING:
-    result = Curl_write_plain(data, sockfd, (char *)sx->outp,
-                              sx->outstanding, &written);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Failed to send SOCKS5 connect request.");
-      return CURLPX_SEND_REQUEST;
-    }
-    if(sx->outstanding != written) {
-      /* remain in state */
-      sx->outstanding -= written;
-      sx->outp += written;
+    presult = socks_state_send(cf, sx, data, CURLPX_SEND_REQUEST,
+                               "SOCKS5 connect request");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
+      /* remain in send state */
       return CURLPX_OK;
     }
 #if defined(HAVE_GSSAPI) || defined(USE_WINDOWS_SSPI)
@@ -932,25 +950,15 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
     sxstate(sx, data, CONNECT_REQ_READ);
     /* FALLTHROUGH */
   case CONNECT_REQ_READ:
-    result = Curl_read_plain(data, sockfd, (char *)sx->outp,
-                             sx->outstanding, &actualread);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Failed to receive SOCKS5 connect request ack.");
-      return CURLPX_RECV_REQACK;
-    }
-    else if(!result && !actualread) {
-      /* connection closed */
-      failf(data, "connection to proxy closed");
-      return CURLPX_CLOSED;
-    }
-    else if(actualread != sx->outstanding) {
-      /* remain in state */
-      sx->outstanding -= actualread;
-      sx->outp += actualread;
+    presult = socks_state_recv(cf, sx, data, CURLPX_RECV_REQACK,
+                               "SOCKS5 connect request ack");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
+      /* remain in reading state */
       return CURLPX_OK;
     }
-
-    if(socksreq[0] != 5) { /* version */
+    else if(socksreq[0] != 5) { /* version */
       failf(data,
             "SOCKS5 reply has wrong version, version should be 5.");
       return CURLPX_BAD_VERSION;
@@ -1031,21 +1039,12 @@ static CURLproxycode do_SOCKS5(struct Curl_cfilter *cf,
 #endif
     /* FALLTHROUGH */
   case CONNECT_REQ_READ_MORE:
-    result = Curl_read_plain(data, sockfd, (char *)sx->outp,
-                             sx->outstanding, &actualread);
-    if(result && (CURLE_AGAIN != result)) {
-      failf(data, "Failed to receive SOCKS5 connect request ack.");
-      return CURLPX_RECV_ADDRESS;
-    }
-    else if(!result && !actualread) {
-      /* connection closed */
-      failf(data, "connection to proxy closed");
-      return CURLPX_CLOSED;
-    }
-    else if(actualread != sx->outstanding) {
-      /* remain in state */
-      sx->outstanding -= actualread;
-      sx->outp += actualread;
+    presult = socks_state_recv(cf, sx, data, CURLPX_RECV_ADDRESS,
+                               "SOCKS5 connect request address");
+    if(CURLPX_OK != presult)
+      return presult;
+    else if(sx->outstanding) {
+      /* remain in reading state */
       return CURLPX_OK;
     }
     sxstate(sx, data, CONNECT_DONE);