quiche: implement http stream interfaces

bazel/external/quiche.BUILD

@@ -3112,6 +3112,20 @@ envoy_cc_test_library(
     ],
 )
 
+envoy_cc_test_library(
+    name = "quic_test_tools_server_session_base_peer",
+    hdrs = [
+        "quiche/quic/test_tools/quic_server_session_base_peer.h",
+    ],
+    copts = quiche_copt,
+    repository = "@envoy",
+    tags = ["nofips"],
+    deps = [
+        ":quic_core_http_spdy_session_lib",
+        ":quic_core_utils_lib",
+    ],
+)
+
 envoy_cc_test_library(
     name = "quic_test_tools_simple_quic_framer_lib",
     srcs = ["quiche/quic/test_tools/simple_quic_framer.cc"],

docs/quiche_integration.md

@@ -0,0 +1,33 @@
+### Overview
+
+QUICHE is integrated in the way described below:
+
+A combination of QUIC session and connection serves as a Network::Connection instance. More than that, QUIC session manages all the QUIC streams. The QUIC codec is a very thin layer between QUIC session and HCM. It doesn't do de-multiplexing or stream management, but only provides interfaces for the HCM to communicate with the QUIC session.
+
+QUIC's Http::StreamEncoder and Http::StreamDecoder implementation is decoupled. The encoder is implemented by EnvoyQuicStream which is a QUIC stream and owned by session. The HCM owns the decoder which can be accessed by QUIC stream instances. And the decoder also knows about stream encoder.
+
+### Request pipeline
+
+The QUIC stream calls decodeHeaders() to deliver request headers. The request body needs to be delivered after headers are delivered as some QUICHE implementations allow the body to arrive earlier than headers. If not read disabled, it always deliver available data via decodeData(). The stream doesn't buffer any readable data in QUICHE stream buffers.
+
+### Response pipeline
+
+The HCM will call encoder's encodeHeaders() to write response headers, and then encodeData() and encodeTrailers(). encodeData() calls WriteBodySlices() to write out response body. The quic stream in QUICHE configures its send buffer threshold QuicStream::buffered_data_threshold_ to be high enough to take all the data passed in, so stream functionally has unlimited buffer.
+
+### Flow Control
+
+#### Receive buffer
+
+All arrived out-of-order data is buffered in QUICHE stream. This buffered is capped by max stream flow control window in QUICHE which is 64MB. Once bytes are put in sequence and ready to be used, OnBodyDataAvailable() is called. The stream implementation overrides this call and calls StreamDecoder::decodeData() in it.Request and response body are buffered in each L7 filter if desired, and the stream itself doesn't buffer any of them unless set as read blocked.
+
+When upstream or any L7 filter reaches its buffer limit, it will call Http::Stream::readDisable() with false to set QUIC stream to be read blocked. In this state, even if more request/response body is available to be delivered, OnBodyDataAvailable() will not be called. As a result, downstream flow control will not shift as no data will be consumed. As both filters and upstream buffers can call readDisable(), each stream has a counter of how many times the HCM blocks the stream. When the counter is cleared, the stream will set its state to unblocked and thus deliver any new and existing available data buttered in the QUICHE stream.
+
+#### Send buffer
+
+We use the unlimited stream send buffer in QUICHE along with a book keeping data structure `EnvoyQuicSimulatedWatermarkBffer` to serve the function of WatermarkBuffer in Envoy to prevent buffering too much in QUICHE.
+
+When the bytes buffered in a stream's send buffer exceeds its high watermark, its inherited method StreamCallbackHelper::runHighWatermarkCallbacks() is called. The buffered bytes will go below stream's low watermark as the stream writes out data gradually via QuicStream::OnCanWrite(). In this case StreamCallbackHelper::runLowWatermarkCallbacks() will be called. QUICHE buffers all the data upon QuicSpdyStream::WriteBodySlices(), assuming `buffered_data_threshold_` is set high enough, and then writes all or part of them according to flow control window and congestion control window. To prevent transient changes in buffered bytes from triggering these two watermark callbacks one immediately after the other, encodeData() and OnCanWrite() only update the watermark bookkeeping once at the end if buffered bytes are changed.
+
+QUICHE doesn't buffer data at the local connection layer. All the data is buffered in the respective streams.To prevent the case where all streams collectively buffers a lot of data, there is also a simulated watermark buffer for each QUIC connection which is updated upon each stream write.
+
+When the aggregated buffered bytes goes above high watermark, its registered network callbacks will call Network::ConnectionCallbacks::onAboveWriteBufferHighWatermark(). The HCMwill notify each stream via QUIC codec Http::Connection::onUnderlyingConnectionAboveWriteBufferHighWatermark() which will call each stream's StreamCallbackHelper::runHighWatermarkCallbacks(). There might be a way to simply the call stack as Quic connection already knows about all the stream, there is no need to call to HCM and notify each stream via codec. But here we just follow the same logic as HTTP2 codec does. In the same way, any QuicStream::OnCanWrite() may change the aggregated buffered bytes in the connection level bookkeeping as well. If the buffered bytes goes down below the low watermark, the same calls will be triggered to propergate onBelowWriteBufferLowWatermark() to each stream.

source/common/common/logger.h

@@ -49,6 +49,7 @@ namespace Logger {
   FUNCTION(misc)                 \
   FUNCTION(mongo)                \
   FUNCTION(quic)                 \
+  FUNCTION(quic_stream)          \
   FUNCTION(pool)                 \
   FUNCTION(rbac)                 \
   FUNCTION(redis)                \

source/extensions/quic_listeners/quiche/BUILD

@@ -97,26 +97,14 @@ envoy_cc_library(
     hdrs = ["envoy_quic_stream.h"],
     tags = ["nofips"],
     deps = [
+        ":envoy_quic_simulated_watermark_buffer_lib",
+        ":quic_filter_manager_connection_lib",
+        "//include/envoy/event:dispatcher_interface",
         "//include/envoy/http:codec_interface",
         "//source/common/http:codec_helper_lib",
     ],
 )
 
-envoy_cc_library(
-    name = "envoy_quic_server_stream_lib",
-    srcs = ["envoy_quic_server_stream.cc"],
-    hdrs = ["envoy_quic_server_stream.h"],
-    tags = ["nofips"],
-    deps = [
-        ":envoy_quic_stream_lib",
-        ":envoy_quic_utils_lib",
-        "//source/common/buffer:buffer_lib",
-        "//source/common/common:assert_lib",
-        "//source/common/http:header_map_lib",
-        "@com_googlesource_quiche//:quic_core_http_spdy_session_lib",
-    ],
-)
-
 envoy_cc_library(
     name = "codec_lib",
     srcs = ["codec_impl.cc"],
@@ -136,22 +124,37 @@ envoy_cc_library(
     tags = ["nofips"],
     deps = [
         ":envoy_quic_connection_lib",
+        ":envoy_quic_simulated_watermark_buffer_lib",
         "//include/envoy/event:dispatcher_interface",
         "//include/envoy/network:connection_interface",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/common:assert_lib",
         "//source/common/common:empty_string",
+        "//source/common/http:header_map_lib",
         "//source/common/network:filter_manager_lib",
         "//source/common/stream_info:stream_info_lib",
     ],
 )
 
 envoy_cc_library(
     name = "envoy_quic_server_session_lib",
-    srcs = ["envoy_quic_server_session.cc"],
-    hdrs = ["envoy_quic_server_session.h"],
+    srcs = [
+        "envoy_quic_server_session.cc",
+        "envoy_quic_server_stream.cc",
+    ],
+    hdrs = [
+        "envoy_quic_server_session.h",
+        "envoy_quic_server_stream.h",
+    ],
     tags = ["nofips"],
     deps = [
-        ":envoy_quic_server_stream_lib",
+        ":envoy_quic_stream_lib",
+        ":envoy_quic_utils_lib",
         ":quic_filter_manager_connection_lib",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/common:assert_lib",
+        "//source/common/http:header_map_lib",
+        "//source/extensions/quic_listeners/quiche/platform:quic_platform_mem_slice_storage_impl_lib",
         "@com_googlesource_quiche//:quic_core_http_spdy_session_lib",
     ],
 )
@@ -207,6 +210,12 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "envoy_quic_simulated_watermark_buffer_lib",
+    hdrs = ["envoy_quic_simulated_watermark_buffer.h"],
+    deps = ["//source/common/common:assert_lib"],
+)
+
 envoy_cc_library(
     name = "active_quic_listener_lib",
     srcs = ["active_quic_listener.cc"],

source/extensions/quic_listeners/quiche/codec_impl.cc

@@ -1,18 +1,34 @@
 #include "extensions/quic_listeners/quiche/codec_impl.h"
 
+#include "extensions/quic_listeners/quiche/envoy_quic_server_stream.h"
+
 namespace Envoy {
 namespace Quic {
 
 bool QuicHttpConnectionImplBase::wantsToWrite() { return quic_session_.HasDataToWrite(); }
 
-// TODO(danzh): modify QUIC stack to react based on aggregated bytes across all
-// the streams. And call StreamCallbackHelper::runHighWatermarkCallbacks() for each stream.
-void QuicHttpConnectionImplBase::onUnderlyingConnectionAboveWriteBufferHighWatermark() {
-  NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+QuicHttpServerConnectionImpl::QuicHttpServerConnectionImpl(
+    EnvoyQuicServerSession& quic_session, Http::ServerConnectionCallbacks& callbacks)
+    : QuicHttpConnectionImplBase(quic_session), quic_server_session_(quic_session) {
+  quic_session.setHttpConnectionCallbacks(callbacks);
+}
+
+void QuicHttpServerConnectionImpl::onUnderlyingConnectionAboveWriteBufferHighWatermark() {
+  for (auto& it : quic_server_session_.stream_map()) {
+    if (!it.second->is_static()) {
+      ENVOY_LOG(debug, "runHighWatermarkCallbacks on stream {}", it.first);
+      dynamic_cast<EnvoyQuicServerStream*>(it.second.get())->runHighWatermarkCallbacks();
+    }
+  }
 }
 
-void QuicHttpConnectionImplBase::onUnderlyingConnectionBelowWriteBufferLowWatermark() {
-  NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+void QuicHttpServerConnectionImpl::onUnderlyingConnectionBelowWriteBufferLowWatermark() {
+  for (const auto& it : quic_server_session_.stream_map()) {
+    if (!it.second->is_static()) {
+      ENVOY_LOG(debug, "runLowWatermarkCallbacks on stream {}", it.first);
+      dynamic_cast<EnvoyQuicServerStream*>(it.second.get())->runLowWatermarkCallbacks();
+    }
+  }
 }
 
 void QuicHttpServerConnectionImpl::goAway() {

source/extensions/quic_listeners/quiche/codec_impl.h

@@ -27,11 +27,10 @@ class QuicHttpConnectionImplBase : public virtual Http::Connection,
     // TODO(danzh) add Http3 enum value for QUIC.
     return Http::Protocol::Http2;
   }
+
   // Returns true if the session has data to send but queued in connection or
   // stream send buffer.
   bool wantsToWrite() override;
-  void onUnderlyingConnectionAboveWriteBufferHighWatermark() override;
-  void onUnderlyingConnectionBelowWriteBufferLowWatermark() override;
 
 protected:
   quic::QuicSpdySession& quic_session_;
@@ -41,17 +40,16 @@ class QuicHttpServerConnectionImpl : public QuicHttpConnectionImplBase,
                                      public Http::ServerConnection {
 public:
   QuicHttpServerConnectionImpl(EnvoyQuicServerSession& quic_session,
-                               Http::ServerConnectionCallbacks& callbacks)
-      : QuicHttpConnectionImplBase(quic_session), quic_server_session_(quic_session) {
-    quic_session.setHttpConnectionCallbacks(callbacks);
-  }
+                               Http::ServerConnectionCallbacks& callbacks);
 
   // Http::Connection
   void goAway() override;
   void shutdownNotice() override {
     // TODO(danzh): Add double-GOAWAY support in QUIC.
     ENVOY_CONN_LOG(error, "Shutdown notice is not propagated to QUIC.", quic_server_session_);
   }
+  void onUnderlyingConnectionAboveWriteBufferHighWatermark() override;
+  void onUnderlyingConnectionBelowWriteBufferLowWatermark() override;
 
 private:
   EnvoyQuicServerSession& quic_server_session_;

source/extensions/quic_listeners/quiche/envoy_quic_dispatcher.cc

@@ -25,6 +25,13 @@ EnvoyQuicDispatcher::EnvoyQuicDispatcher(
   // Network::UdpListenerCallbacks which should be called at the beginning
   // of HandleReadEvent(). And this callback should call quic::Dispatcher::ProcessBufferedChlos().
   SetQuicFlag(FLAGS_quic_allow_chlo_buffering, false);
+  // Set send buffer twice of max flow control window to ensure that stream send
+  // buffer always takes all the data.
+  // The max amount of data buffered is the per-stream high watermark + the max
+  // flow control window of upstream. The per-stream high watermark should be
+  // smaller than max flow control window to make sure upper stream can be flow
+  // control blocked early enough not to send more than the threshold allows.
+  SetQuicFlag(FLAGS_quic_buffered_data_threshold, 2 * 16 * 1024 * 1024); // 32MB
 }
 
 void EnvoyQuicDispatcher::OnConnectionClosed(quic::QuicConnectionId connection_id,
@@ -44,7 +51,8 @@ quic::QuicSession* EnvoyQuicDispatcher::CreateQuicSession(
       listener_stats_);
   auto quic_session = new EnvoyQuicServerSession(
       config(), quic::ParsedQuicVersionVector{version}, std::move(quic_connection), this,
-      session_helper(), crypto_config(), compressed_certs_cache(), dispatcher_);
+      session_helper(), crypto_config(), compressed_certs_cache(), dispatcher_,
+      listener_config_.perConnectionBufferLimitBytes());
   quic_session->Initialize();
   // Filter chain can't be retrieved here as self address is unknown at this
   // point.

source/extensions/quic_listeners/quiche/envoy_quic_server_session.cc

@@ -9,6 +9,7 @@
 #include "quiche/quic/core/quic_crypto_server_stream.h"
 #pragma GCC diagnostic pop
 
+#include "common/common/assert.h"
 #include "extensions/quic_listeners/quiche/envoy_quic_server_stream.h"
 
 namespace Envoy {
@@ -18,10 +19,16 @@ EnvoyQuicServerSession::EnvoyQuicServerSession(
     const quic::QuicConfig& config, const quic::ParsedQuicVersionVector& supported_versions,
     std::unique_ptr<EnvoyQuicConnection> connection, quic::QuicSession::Visitor* visitor,
     quic::QuicCryptoServerStream::Helper* helper, const quic::QuicCryptoServerConfig* crypto_config,
-    quic::QuicCompressedCertsCache* compressed_certs_cache, Event::Dispatcher& dispatcher)
+    quic::QuicCompressedCertsCache* compressed_certs_cache, Event::Dispatcher& dispatcher,
+    uint32_t send_buffer_limit)
     : quic::QuicServerSessionBase(config, supported_versions, connection.get(), visitor, helper,
                                   crypto_config, compressed_certs_cache),
-      QuicFilterManagerConnectionImpl(std::move(connection), dispatcher) {}
+      QuicFilterManagerConnectionImpl(connection.get(), dispatcher, send_buffer_limit),
+      quic_connection_(std::move(connection)) {}
+
+EnvoyQuicServerSession::~EnvoyQuicServerSession() {
+  QuicFilterManagerConnectionImpl::quic_connection_ = nullptr;
+}
 
 absl::string_view EnvoyQuicServerSession::requestedServerName() const {
   return {GetCryptoStream()->crypto_negotiated_params().sni};
@@ -41,6 +48,9 @@ quic::QuicSpdyStream* EnvoyQuicServerSession::CreateIncomingStream(quic::QuicStr
   auto stream = new EnvoyQuicServerStream(id, this, quic::BIDIRECTIONAL);
   ActivateStream(absl::WrapUnique(stream));
   setUpRequestDecoder(*stream);
+  if (aboveHighWatermark()) {
+    stream->runHighWatermarkCallbacks();
+  }
   return stream;
 }
 

source/extensions/quic_listeners/quiche/envoy_quic_server_session.h

@@ -30,7 +30,9 @@ class EnvoyQuicServerSession : public quic::QuicServerSessionBase,
                          quic::QuicCryptoServerStream::Helper* helper,
                          const quic::QuicCryptoServerConfig* crypto_config,
                          quic::QuicCompressedCertsCache* compressed_certs_cache,
-                         Event::Dispatcher& dispatcher);
+                         Event::Dispatcher& dispatcher, uint32_t send_buffer_limit);
+
+  ~EnvoyQuicServerSession() override;
 
   // Network::Connection
   absl::string_view requestedServerName() const override;
@@ -48,6 +50,8 @@ class EnvoyQuicServerSession : public quic::QuicServerSessionBase,
   // quic::QuicSpdySession
   void OnCryptoHandshakeEvent(CryptoHandshakeEvent event) override;
 
+  using quic::QuicSession::stream_map;
+
 protected:
   // quic::QuicServerSessionBase
   quic::QuicCryptoServerStreamBase*
@@ -64,6 +68,7 @@ class EnvoyQuicServerSession : public quic::QuicServerSessionBase,
 private:
   void setUpRequestDecoder(EnvoyQuicStream& stream);
 
+  std::unique_ptr<EnvoyQuicConnection> quic_connection_;
   // These callbacks are owned by network filters and quic session should out live
   // them.
   Http::ServerConnectionCallbacks* http_connection_callbacks_{nullptr};