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,49 @@
+### 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. QUIC codec is a very thin layer between QUIC session and HCM. It doesn't do de-multiplexing stream management, but only provides interfaces for HCM to communicate with QUIC session.
+
+QUIC's Http::StreamEncoder and Http::StreamDecoder implementation is decoupled, encoder is implemented by EnvoyQuicStream which is a QUIC stream and owned by session. HCM owns the decoder which can be accessed by QUIC stream instances. And decoder also knows about stream encoder.
+
+### Request piepline
+
+QUIC stream calls decodeHeaders() to deliver request headers. Request body needs
+to be delivered after headers are delieverd as some QUICHE implementation allows
+body to arrive earlier than headers. If not read disabled, always deliever available data via decodeData(). Stream
+doesn't buffer any readable data in QUICHE stream buffers.
+
+### Response pipeline
+
+HCM will call encoder's encodeHeaders() to write response headers, and then encodeData() and encodeTrailers(). encodeData() calls WriteBodySlices() to write response body. Quic stream in QUICHE can config it's send buffer threshold QuicStream::buffered_data_threshold_ to be high enough to take all the data passed in, so stream can have unlimited buffer.
+
+### Flow Control
+
+## Receive buffer
+
+All the arrived out-of-order data is buffered in QUICH stream. And this buffered
+in capped by max stream flow control window in QUICHE which is 64MB. Once they are put
+in sequence and ready to be used, OnBodyDataAvailable() is called. Our stream
+implementation overrides this call and call StreamDecoder::decodeData() in it.
+Request and response body is buffered in each L7 filter if desired, and the
+stream itself doesn't buffer any of them if not set to be blocked.
+
+When upstream or any L7 filter reaches its buffer limit, it call
+Http::Stream::readDisable() with false to set QUIC stream in block state. In
+this state, even more request/reponse body is available to be delivered,
+OnBodyDataAvailable() will not be called. As a result, downstream flow contol
+will not shift as no data will be consumed. As both filters and upstream buffer
+can call readDisable(), each stream has a counter of how many
+times the HCM wants to block the stream. Until the counter is cleared, stream
+will resume its state to be unblocked and thus delivering any new and existing
+available data in QUICHE stream.
+
+## Send buffer
+
+We use the unlimited stream send buffer in QUICHE along with a book keeping data structure `EnvoyQuicSimulatedWatermarkBffer` to server the purpose 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(). And 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 book keeping once at the end if buffered bytes are changed.
+
+QUICHE doesn't buffer data in connection. And all the data is buffered in stream.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 changes upon each stream write.
+
+When the aggregated buffered bytes goes above high watermark, its registered network callbacks will call Network::ConnectionCallbacks::onAboveWriteBufferHighWatermark(). HCM as a such callback will 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. Same way, any QuicStream::OnCanWrite() may change the aggregated buffered bytes in the connection level book keeping as well. If the buffered bytes goes down below low watermark, same calls will be triggered to propergate onBelowWriteBufferLowWatermark() to each stream callbacks.

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,13 @@ envoy_cc_library(
     hdrs = ["envoy_quic_stream.h"],
     tags = ["nofips"],
     deps = [
+        ":envoy_quic_simulated_watermark_buffer_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 +123,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 +209,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

@@ -44,7 +44,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};

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};