[http1] Buffer pending http/1 body before dispatching to the filter chain

source/common/http/http1/codec_impl.cc

@@ -388,15 +388,19 @@ http_parser_settings ConnectionImpl::settings_{
       return static_cast<ConnectionImpl*>(parser->data)->onHeadersCompleteBase();
     },
     [](http_parser* parser, const char* at, size_t length) -> int {
-      static_cast<ConnectionImpl*>(parser->data)->onBody(at, length);
+      static_cast<ConnectionImpl*>(parser->data)->bufferBody(at, length);
       return 0;
     },
     [](http_parser* parser) -> int {
       static_cast<ConnectionImpl*>(parser->data)->onMessageCompleteBase();
       return 0;
     },
-    nullptr, // on_chunk_header
-    nullptr  // on_chunk_complete
+    [](http_parser* parser) -> int {
+      const bool is_final_chunk = (parser->content_length == 0);
+      static_cast<ConnectionImpl*>(parser->data)->onChunkHeader(is_final_chunk);
+      return 0;
+    },
+    nullptr // on_chunk_complete
 };
 
 ConnectionImpl::ConnectionImpl(Network::Connection& connection, Stats::Scope& stats,
@@ -450,18 +454,15 @@ bool ConnectionImpl::maybeDirectDispatch(Buffer::Instance& data) {
     return false;
   }
 
-  ssize_t total_parsed = 0;
-  for (const Buffer::RawSlice& slice : data.getRawSlices()) {
-    total_parsed += slice.len_;
-    onBody(static_cast<const char*>(slice.mem_), slice.len_);
-  }
-  ENVOY_CONN_LOG(trace, "direct-dispatched {} bytes", connection_, total_parsed);
-  data.drain(total_parsed);
+  ENVOY_CONN_LOG(trace, "direct-dispatched {} bytes", connection_, data.length());
+  onBody(data);
+  data.drain(data.length());
   return true;
 }
 
 void ConnectionImpl::dispatch(Buffer::Instance& data) {
   ENVOY_CONN_LOG(trace, "parsing {} bytes", connection_, data.length());
+  ASSERT(buffered_body_.length() == 0);
 
   if (maybeDirectDispatch(data)) {
     return;
@@ -474,10 +475,15 @@ void ConnectionImpl::dispatch(Buffer::Instance& data) {
   if (data.length() > 0) {
     for (const Buffer::RawSlice& slice : data.getRawSlices()) {
       total_parsed += dispatchSlice(static_cast<const char*>(slice.mem_), slice.len_);
+      if (HTTP_PARSER_ERRNO(&parser_) != HPE_OK) {
+        break;
+      }
     }
+    dispatchBufferedBody();
   } else {
     dispatchSlice(nullptr, 0);
   }
+  ASSERT(buffered_body_.length() == 0);
 
   ENVOY_CONN_LOG(trace, "parsed {} bytes", connection_, total_parsed);
   data.drain(total_parsed);
@@ -618,9 +624,31 @@ int ConnectionImpl::onHeadersCompleteBase() {
   return handling_upgrade_ ? 2 : rc;
 }
 
+void ConnectionImpl::bufferBody(const char* data, size_t length) {
+  buffered_body_.add(data, length);
+}
+
+void ConnectionImpl::dispatchBufferedBody() {
+  ASSERT(HTTP_PARSER_ERRNO(&parser_) == HPE_OK || HTTP_PARSER_ERRNO(&parser_) == HPE_PAUSED);
+  if (buffered_body_.length() > 0) {
+    onBody(buffered_body_);
+    buffered_body_.drain(buffered_body_.length());
+  }
+}
+
+void ConnectionImpl::onChunkHeader(bool is_final_chunk) {
+  if (is_final_chunk) {
+    // Dispatch body before parsing trailers, so body ends up dispatched even if an error is found
+    // while processing trailers.
+    dispatchBufferedBody();
+  }
+}
+
 void ConnectionImpl::onMessageCompleteBase() {
   ENVOY_CONN_LOG(trace, "message complete", connection_);
 
+  dispatchBufferedBody();
+
   if (handling_upgrade_) {
     // If this is an upgrade request, swallow the onMessageComplete. The
     // upgrade payload will be treated as stream body.
@@ -809,12 +837,11 @@ void ServerConnectionImpl::onUrl(const char* data, size_t length) {
   }
 }
 
-void ServerConnectionImpl::onBody(const char* data, size_t length) {
+void ServerConnectionImpl::onBody(Buffer::Instance& data) {
   ASSERT(!deferred_end_stream_headers_);
   if (active_request_.has_value()) {
-    ENVOY_CONN_LOG(trace, "body size={}", connection_, length);
-    Buffer::OwnedImpl buffer(data, length);
-    active_request_.value().request_decoder_->decodeData(buffer, false);
+    ENVOY_CONN_LOG(trace, "body size={}", connection_, data.length());
+    active_request_.value().request_decoder_->decodeData(data, false);
   }
 }
 
@@ -947,12 +974,10 @@ int ClientConnectionImpl::onHeadersComplete() {
   return cannotHaveBody() ? 1 : 0;
 }
 
-void ClientConnectionImpl::onBody(const char* data, size_t length) {
+void ClientConnectionImpl::onBody(Buffer::Instance& data) {
   ASSERT(!deferred_end_stream_headers_);
   if (pending_response_.has_value()) {
-    Buffer::OwnedImpl buffer;
-    buffer.add(data, length);
-    pending_response_.value().decoder_->decodeData(buffer, false);
+    pending_response_.value().decoder_->decodeData(data, false);
   }
 }
 

source/common/http/http1/codec_impl.h

@@ -243,6 +243,18 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
    */
   size_t dispatchSlice(const char* slice, size_t len);
 
+  /**
+   * Called by the http_parser when body data is received.
+   * @param data supplies the start address.
+   * @param length supplies the length.
+   */
+  void bufferBody(const char* data, size_t length);
+
+  /**
+   * Push the accumulated body through the filter pipeline.
+   */
+  void dispatchBufferedBody();
+
   /**
    * Called when a request/response is beginning. A base routine happens first then a virtual
    * dispatch is invoked.
@@ -281,18 +293,22 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
   virtual int onHeadersComplete() PURE;
 
   /**
-   * Called when body data is received.
-   * @param data supplies the start address.
-   * @param length supplies the length.
+   * Called when body data is available for processing.
+   * @param data supplies the body data
    */
-  virtual void onBody(const char* data, size_t length) PURE;
+  virtual void onBody(Buffer::Instance& data) PURE;
 
   /**
    * Called when the request/response is complete.
    */
   void onMessageCompleteBase();
   virtual void onMessageComplete() PURE;
 
+  /**
+   * Called when accepting a chunk header.
+   */
+  void onChunkHeader(bool is_final_chunk);
+
   /**
    * @see onResetStreamBase().
    */
@@ -320,6 +336,7 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
   HeaderParsingState header_parsing_state_{HeaderParsingState::Field};
   HeaderString current_header_field_;
   HeaderString current_header_value_;
+  Buffer::OwnedImpl buffered_body_;
   Buffer::WatermarkBuffer output_buffer_;
   Protocol protocol_{Protocol::Http11};
   const uint32_t max_headers_kb_;
@@ -368,7 +385,7 @@ class ServerConnectionImpl : public ServerConnection, public ConnectionImpl {
   void onMessageBegin() override;
   void onUrl(const char* data, size_t length) override;
   int onHeadersComplete() override;
-  void onBody(const char* data, size_t length) override;
+  void onBody(Buffer::Instance& data) override;
   void onMessageComplete() override;
   void onResetStream(StreamResetReason reason) override;
   void sendProtocolError(absl::string_view details) override;
@@ -447,7 +464,7 @@ class ClientConnectionImpl : public ClientConnection, public ConnectionImpl {
   void onMessageBegin() override {}
   void onUrl(const char*, size_t) override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
   int onHeadersComplete() override;
-  void onBody(const char* data, size_t length) override;
+  void onBody(Buffer::Instance& data) override;
   void onMessageComplete() override;
   void onResetStream(StreamResetReason reason) override;
   void sendProtocolError(absl::string_view details) override;

test/benchmark/BUILD

@@ -15,8 +15,9 @@ envoy_cc_test_library(
         "abseil_symbolize",
         "benchmark",
     ],
-    deps = select({
-        "//bazel:disable_signal_trace": [],
-        "//conditions:default": ["//source/common/signal:sigaction_lib"],
-    }),
+    deps = ["//test/test_common:environment_lib"] +
+           select({
+               "//bazel:disable_signal_trace": [],
+               "//conditions:default": ["//source/common/signal:sigaction_lib"],
+           }),
 )

test/benchmark/main.cc

@@ -7,6 +7,8 @@
 #include "common/signal/signal_action.h"
 #endif
 
+#include "test/test_common/environment.h"
+
 #include "absl/debugging/symbolize.h"
 
 // Boilerplate main(), which discovers benchmarks and runs them.
@@ -23,5 +25,6 @@ int main(int argc, char** argv) {
   if (benchmark::ReportUnrecognizedArguments(argc, argv)) {
     return 1;
   }
+  Envoy::TestEnvironment::initializeOptions(argc, argv);
   benchmark::RunSpecifiedBenchmarks();
 }