Plumbed read_buffer through and fixed leftover bytes problem.

src/core/ext/client_config/http_connect_handshaker.c

@@ -56,9 +56,9 @@ typedef struct http_connect_handshaker {
   void* user_data;
 
   // Objects for processing the HTTP CONNECT request and response.
-  gpr_slice_buffer request_buffer;
+  gpr_slice_buffer write_buffer;
+  gpr_slice_buffer* read_buffer;
   grpc_closure request_done_closure;
-  gpr_slice_buffer response_buffer;
   grpc_closure response_read_closure;
   grpc_http_parser http_parser;
   grpc_http_response http_response;
@@ -70,10 +70,11 @@ static void on_write_done(grpc_exec_ctx* exec_ctx, void* arg,
   http_connect_handshaker* h = arg;
   if (error != GRPC_ERROR_NONE) {
     // If the write failed, invoke the callback immediately with the error.
-    h->cb(exec_ctx, h->endpoint, h->args, h->user_data, GRPC_ERROR_REF(error));
+    h->cb(exec_ctx, h->endpoint, h->args, h->read_buffer, h->user_data,
+          GRPC_ERROR_REF(error));
   } else {
     // Otherwise, read the response.
-    grpc_endpoint_read(exec_ctx, h->endpoint, &h->response_buffer,
+    grpc_endpoint_read(exec_ctx, h->endpoint, h->read_buffer,
                        &h->response_read_closure);
   }
 }
@@ -87,12 +88,29 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
     goto done;
   }
   // Add buffer to parser.
-  for (size_t i = 0; i < h->response_buffer.count; ++i) {
-    if (GPR_SLICE_LENGTH(h->response_buffer.slices[i]) > 0) {
+  for (size_t i = 0; i < h->read_buffer->count; ++i) {
+    if (GPR_SLICE_LENGTH(h->read_buffer->slices[i]) > 0) {
+      size_t body_start_offset = 0;
       error = grpc_http_parser_parse(
-          &h->http_parser, h->response_buffer.slices[i]);
+          &h->http_parser, h->read_buffer->slices[i], &body_start_offset);
       if (error != GRPC_ERROR_NONE)
         goto done;
+      if (h->http_parser.state == GRPC_HTTP_BODY) {
+        // Remove the data we've already read from the read buffer,
+        // leaving only the leftover bytes (if any).
+        gpr_slice_buffer tmp_buffer;
+        gpr_slice_buffer_init(&tmp_buffer);
+        if (body_start_offset < GPR_SLICE_LENGTH(h->read_buffer->slices[i])) {
+          gpr_slice_buffer_add(&tmp_buffer,
+                               gpr_slice_split_tail(&h->read_buffer->slices[i],
+                                                    body_start_offset));
+        }
+        gpr_slice_buffer_addn(&tmp_buffer, &h->read_buffer->slices[i + 1],
+                              h->read_buffer->count - i - 1);
+        gpr_slice_buffer_swap(h->read_buffer, &tmp_buffer);
+        gpr_slice_buffer_destroy(&tmp_buffer);
+        break;
+      }
     }
   }
   // If we're not done reading the response, read more data.
@@ -107,8 +125,8 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
   // complete (e.g., handling chunked transfer encoding or looking
   // at the Content-Length: header).
   if (h->http_parser.state != GRPC_HTTP_BODY) {
-    gpr_slice_buffer_reset_and_unref(&h->response_buffer);
-    grpc_endpoint_read(exec_ctx, h->endpoint, &h->response_buffer,
+    gpr_slice_buffer_reset_and_unref(h->read_buffer);
+    grpc_endpoint_read(exec_ctx, h->endpoint, h->read_buffer,
                        &h->response_read_closure);
     return;
   }
@@ -122,7 +140,7 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
   }
  done:
   // Invoke handshake-done callback.
-  h->cb(exec_ctx, h->endpoint, h->args, h->user_data, error);
+  h->cb(exec_ctx, h->endpoint, h->args, h->read_buffer, h->user_data, error);
 }
 
 //
@@ -134,8 +152,7 @@ static void http_connect_handshaker_destroy(grpc_exec_ctx* exec_ctx,
   http_connect_handshaker* h = (http_connect_handshaker*)handshaker;
   gpr_free(h->proxy_server);
   gpr_free(h->server_name);
-  gpr_slice_buffer_destroy(&h->request_buffer);
-  gpr_slice_buffer_destroy(&h->response_buffer);
+  gpr_slice_buffer_destroy(&h->write_buffer);
   grpc_http_parser_destroy(&h->http_parser);
   grpc_http_response_destroy(&h->http_response);
   gpr_free(h);
@@ -148,7 +165,8 @@ static void http_connect_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
 // FIXME BEFORE MERGING: apply deadline
 static void http_connect_handshaker_do_handshake(
     grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker,
-    grpc_endpoint* endpoint, grpc_channel_args* args, gpr_timespec deadline,
+    grpc_endpoint* endpoint, grpc_channel_args* args,
+    gpr_slice_buffer* read_buffer, gpr_timespec deadline,
     grpc_tcp_server_acceptor* acceptor, grpc_handshaker_done_cb cb,
     void* user_data) {
   http_connect_handshaker* h = (http_connect_handshaker*)handshaker;
@@ -158,9 +176,9 @@ static void http_connect_handshaker_do_handshake(
   h->cb = cb;
   h->user_data = user_data;
   // Initialize fields.
-  gpr_slice_buffer_init(&h->request_buffer);
+  gpr_slice_buffer_init(&h->write_buffer);
+  h->read_buffer = read_buffer;
   grpc_closure_init(&h->request_done_closure, on_write_done, h);
-  gpr_slice_buffer_init(&h->response_buffer);
   grpc_closure_init(&h->response_read_closure, on_read_done, h);
   grpc_http_parser_init(&h->http_parser, GRPC_HTTP_RESPONSE,
                         &h->http_response);
@@ -174,8 +192,8 @@ static void http_connect_handshaker_do_handshake(
   request.http.path = h->server_name;
   request.handshaker = &grpc_httpcli_plaintext;
   gpr_slice request_slice = grpc_httpcli_format_connect_request(&request);
-  gpr_slice_buffer_add(&h->request_buffer, request_slice);
-  grpc_endpoint_write(exec_ctx, endpoint, &h->request_buffer,
+  gpr_slice_buffer_add(&h->write_buffer, request_slice);
+  grpc_endpoint_write(exec_ctx, endpoint, &h->write_buffer,
                       &h->request_done_closure);
 }
 

src/core/ext/transport/chttp2/transport/chttp2_transport.c

@@ -1987,7 +1987,8 @@ static grpc_error *try_http_parsing(grpc_exec_ctx *exec_ctx,
 
   grpc_error *parse_error = GRPC_ERROR_NONE;
   for (; i < t->read_buffer.count && parse_error == GRPC_ERROR_NONE; i++) {
-    parse_error = grpc_http_parser_parse(&parser, t->read_buffer.slices[i]);
+    parse_error = grpc_http_parser_parse(&parser, t->read_buffer.slices[i],
+                                         NULL);
   }
   if (parse_error == GRPC_ERROR_NONE &&
       (parse_error = grpc_http_parser_eof(&parser)) == GRPC_ERROR_NONE) {

src/core/lib/http/httpcli.c

@@ -146,7 +146,7 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *user_data,
     if (GPR_SLICE_LENGTH(req->incoming.slices[i])) {
       req->have_read_byte = 1;
       grpc_error *err =
-          grpc_http_parser_parse(&req->parser, req->incoming.slices[i]);
+          grpc_http_parser_parse(&req->parser, req->incoming.slices[i], NULL);
       if (err != GRPC_ERROR_NONE) {
         finish(exec_ctx, req, err);
         return;

src/core/lib/http/parser.c

@@ -33,6 +33,7 @@
 
 #include "src/core/lib/http/parser.h"
 
+#include <stdbool.h>
 #include <string.h>
 
 #include <grpc/support/alloc.h>
@@ -200,7 +201,8 @@ done:
   return error;
 }
 
-static grpc_error *finish_line(grpc_http_parser *parser) {
+static grpc_error *finish_line(grpc_http_parser *parser,
+                               bool *found_body_start) {
   grpc_error *err;
   switch (parser->state) {
     case GRPC_HTTP_FIRST_LINE:
@@ -211,6 +213,7 @@ static grpc_error *finish_line(grpc_http_parser *parser) {
     case GRPC_HTTP_HEADERS:
       if (parser->cur_line_length == parser->cur_line_end_length) {
         parser->state = GRPC_HTTP_BODY;
+        *found_body_start = true;
         break;
       }
       err = add_header(parser);
@@ -274,7 +277,8 @@ static bool check_line(grpc_http_parser *parser) {
   return false;
 }
 
-static grpc_error *addbyte(grpc_http_parser *parser, uint8_t byte) {
+static grpc_error *addbyte(grpc_http_parser *parser, uint8_t byte,
+                           bool *found_body_start) {
   switch (parser->state) {
     case GRPC_HTTP_FIRST_LINE:
     case GRPC_HTTP_HEADERS:
@@ -287,7 +291,7 @@ static grpc_error *addbyte(grpc_http_parser *parser, uint8_t byte) {
       parser->cur_line[parser->cur_line_length] = byte;
       parser->cur_line_length++;
       if (check_line(parser)) {
-        return finish_line(parser);
+        return finish_line(parser, found_body_start);
       }
       return GRPC_ERROR_NONE;
     case GRPC_HTTP_BODY:
@@ -329,14 +333,16 @@ void grpc_http_response_destroy(grpc_http_response *response) {
   gpr_free(response->hdrs);
 }
 
-grpc_error *grpc_http_parser_parse(grpc_http_parser *parser, gpr_slice slice) {
-  size_t i;
-
-  for (i = 0; i < GPR_SLICE_LENGTH(slice); i++) {
-    grpc_error *err = addbyte(parser, GPR_SLICE_START_PTR(slice)[i]);
+grpc_error *grpc_http_parser_parse(grpc_http_parser *parser, gpr_slice slice,
+                                   size_t *start_of_body) {
+  for (size_t i = 0; i < GPR_SLICE_LENGTH(slice); i++) {
+    bool found_body_start = false;
+    grpc_error *err = addbyte(parser, GPR_SLICE_START_PTR(slice)[i],
+                              &found_body_start);
     if (err != GRPC_ERROR_NONE) return err;
+    if (found_body_start && start_of_body != NULL)
+      *start_of_body = i + 1;
   }
-
   return GRPC_ERROR_NONE;
 }
 

src/core/lib/http/parser.h

@@ -113,7 +113,9 @@ void grpc_http_parser_init(grpc_http_parser *parser, grpc_http_type type,
                            void *request_or_response);
 void grpc_http_parser_destroy(grpc_http_parser *parser);
 
-grpc_error *grpc_http_parser_parse(grpc_http_parser *parser, gpr_slice slice);
+/* Sets \a start_of_body to the offset in \a slice of the start of the body. */
+grpc_error *grpc_http_parser_parse(grpc_http_parser *parser, gpr_slice slice,
+                                   size_t *start_of_body);
 grpc_error *grpc_http_parser_eof(grpc_http_parser *parser);
 
 void grpc_http_request_destroy(grpc_http_request *request);

test/core/end2end/fixtures/http_proxy.c

@@ -297,7 +297,8 @@ gpr_log(GPR_INFO, "==> %s()", __func__);
   // We've established a connection, so send back a 200 response code to
   // the client.
   // The write callback inherits our reference to conn.
-  gpr_slice slice = gpr_slice_from_copied_string("200 connected\r\n\r\n");
+  gpr_slice slice =
+      gpr_slice_from_copied_string("HTTP/1.0 200 connected\r\n\r\n");
   gpr_slice_buffer_add(&conn->client_write_buffer, slice);
   grpc_endpoint_write(exec_ctx, conn->client_endpoint,
                       &conn->client_write_buffer,
@@ -323,7 +324,7 @@ gpr_log(GPR_INFO, "==> %s()", __func__);
   for (size_t i = 0; i < conn->client_read_buffer.count; ++i) {
     if (GPR_SLICE_LENGTH(conn->client_read_buffer.slices[i]) > 0) {
       error = grpc_http_parser_parse(
-          &conn->http_parser, conn->client_read_buffer.slices[i]);
+          &conn->http_parser, conn->client_read_buffer.slices[i], NULL);
       if (error != GRPC_ERROR_NONE) {
         proxy_connection_failed(exec_ctx, conn, true /* is_client */,
                                 "HTTP proxy request parse", error);

test/core/http/parser_test.c

@@ -62,7 +62,8 @@ static void test_request_succeeds(grpc_slice_split_mode split_mode,
   grpc_http_parser_init(&parser, GRPC_HTTP_REQUEST, &request);
 
   for (i = 0; i < num_slices; i++) {
-    GPR_ASSERT(grpc_http_parser_parse(&parser, slices[i]) == GRPC_ERROR_NONE);
+    GPR_ASSERT(grpc_http_parser_parse(&parser, slices[i], NULL)
+               == GRPC_ERROR_NONE);
     gpr_slice_unref(slices[i]);
   }
   GPR_ASSERT(grpc_http_parser_eof(&parser) == GRPC_ERROR_NONE);
@@ -118,7 +119,8 @@ static void test_succeeds(grpc_slice_split_mode split_mode, char *response_text,
   grpc_http_parser_init(&parser, GRPC_HTTP_RESPONSE, &response);
 
   for (i = 0; i < num_slices; i++) {
-    GPR_ASSERT(grpc_http_parser_parse(&parser, slices[i]) == GRPC_ERROR_NONE);
+    GPR_ASSERT(grpc_http_parser_parse(&parser, slices[i], NULL)
+               == GRPC_ERROR_NONE);
     gpr_slice_unref(slices[i]);
   }
   GPR_ASSERT(grpc_http_parser_eof(&parser) == GRPC_ERROR_NONE);
@@ -171,7 +173,7 @@ static void test_fails(grpc_slice_split_mode split_mode, char *response_text) {
 
   for (i = 0; i < num_slices; i++) {
     if (GRPC_ERROR_NONE == error) {
-      error = grpc_http_parser_parse(&parser, slices[i]);
+      error = grpc_http_parser_parse(&parser, slices[i], NULL);
     }
     gpr_slice_unref(slices[i]);
   }
@@ -204,7 +206,7 @@ static void test_request_fails(grpc_slice_split_mode split_mode,
 
   for (i = 0; i < num_slices; i++) {
     if (error == GRPC_ERROR_NONE) {
-      error = grpc_http_parser_parse(&parser, slices[i]);
+      error = grpc_http_parser_parse(&parser, slices[i], NULL);
     }
     gpr_slice_unref(slices[i]);
   }

test/core/http/request_fuzzer.c

@@ -48,7 +48,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
   memset(&request, 0, sizeof(request));
   grpc_http_parser_init(&parser, GRPC_HTTP_REQUEST, &request);
   gpr_slice slice = gpr_slice_from_copied_buffer((const char *)data, size);
-  GRPC_ERROR_UNREF(grpc_http_parser_parse(&parser, slice));
+  GRPC_ERROR_UNREF(grpc_http_parser_parse(&parser, slice, NULL));
   GRPC_ERROR_UNREF(grpc_http_parser_eof(&parser));
   gpr_slice_unref(slice);
   grpc_http_parser_destroy(&parser);

test/core/http/response_fuzzer.c

@@ -47,7 +47,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
   memset(&response, 0, sizeof(response));
   grpc_http_parser_init(&parser, GRPC_HTTP_RESPONSE, &response);
   gpr_slice slice = gpr_slice_from_copied_buffer((const char *)data, size);
-  GRPC_ERROR_UNREF(grpc_http_parser_parse(&parser, slice));
+  GRPC_ERROR_UNREF(grpc_http_parser_parse(&parser, slice, NULL));
   GRPC_ERROR_UNREF(grpc_http_parser_eof(&parser));
   gpr_slice_unref(slice);
   grpc_http_parser_destroy(&parser);