quic: fixing the disconnect between quiche stream count and Envoy

source/common/conn_pool/conn_pool_base.cc

@@ -175,18 +175,23 @@ void ConnPoolImplBase::attachStreamToClient(Envoy::ConnectionPool::ActiveClient&
   }
   ENVOY_CONN_LOG(debug, "creating stream", client);
 
+  // Latch capacity before updating remaining streams.
+  uint64_t capacity = client.currentUnusedCapacity();
   client.remaining_streams_--;
   if (client.remaining_streams_ == 0) {
     ENVOY_CONN_LOG(debug, "maximum streams per connection, DRAINING", client);
     host_->cluster().stats().upstream_cx_max_requests_.inc();
     transitionActiveClientState(client, Envoy::ConnectionPool::ActiveClient::State::DRAINING);
-  } else if (client.numActiveStreams() + 1 >= client.concurrent_stream_limit_) {
+  } else if (capacity <= 1) {
     // As soon as the new stream is created, the client will be maxed out.
     transitionActiveClientState(client, Envoy::ConnectionPool::ActiveClient::State::BUSY);
   }
 
   // Decrement the capacity, as there's one less stream available for serving.
-  state_.decrConnectingAndConnectedStreamCapacity(1);
+  // For HTTP/3, the capacity is updated in newStreamEncoder.
+  if (!quic()) {
+    state_.decrConnectingAndConnectedStreamCapacity(1);
+  }
   // Track the new active stream.
   state_.incrActiveStreams(1);
   num_active_streams_++;
@@ -213,14 +218,17 @@ void ConnPoolImplBase::onStreamClosed(Envoy::ConnectionPool::ActiveClient& clien
   // If the effective client capacity was limited by concurrency, increase connecting capacity.
   // If the effective client capacity was limited by max total streams, this will not result in an
   // increment as no capacity is freed up.
-  if (client.remaining_streams_ > client.concurrent_stream_limit_ - client.numActiveStreams() - 1 ||
-      had_negative_capacity) {
+  // We don't update the capacity for HTTP/3 as the stream count should only
+  // increase when a MAX_STREAMS frame is received.
+  if (!quic() && (client.remaining_streams_ >
+                      client.concurrent_stream_limit_ - client.numActiveStreams() - 1 ||
+                  had_negative_capacity)) {
     state_.incrConnectingAndConnectedStreamCapacity(1);
   }
   if (client.state() == ActiveClient::State::DRAINING && client.numActiveStreams() == 0) {
     // Close out the draining client if we no longer have active streams.
     client.close();
-  } else if (client.state() == ActiveClient::State::BUSY) {
+  } else if (client.state() == ActiveClient::State::BUSY && client.currentUnusedCapacity() != 0) {
     transitionActiveClientState(client, ActiveClient::State::READY);
     if (!delay_attaching_stream) {
       onUpstreamReady();

source/common/conn_pool/conn_pool_base.h

@@ -56,7 +56,7 @@ class ActiveClient : public LinkedObject<ActiveClient>,
   // Returns the application protocol, or absl::nullopt for TCP.
   virtual absl::optional<Http::Protocol> protocol() const PURE;
 
-  int64_t currentUnusedCapacity() const {
+  virtual int64_t currentUnusedCapacity() const {
     int64_t remaining_concurrent_streams =
         static_cast<int64_t>(concurrent_stream_limit_) - numActiveStreams();
 
@@ -148,6 +148,7 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   virtual ~ConnPoolImplBase();
 
   void deleteIsPendingImpl();
+  virtual bool quic() { return false; }
 
   // A helper function to get the specific context type from the base class context.
   template <class T> T& typedContext(AttachContext& context) {
@@ -234,6 +235,9 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   void decrClusterStreamCapacity(uint32_t delta) {
     state_.decrConnectingAndConnectedStreamCapacity(delta);
   }
+  void incrClusterStreamCapacity(uint32_t delta) {
+    state_.incrConnectingAndConnectedStreamCapacity(delta);
+  }
   void dumpState(std::ostream& os, int indent_level = 0) const {
     const char* spaces = spacesForLevel(indent_level);
     os << spaces << "ConnPoolImplBase " << this << DUMP_MEMBER(ready_clients_.size())

source/common/http/codec_client.h

@@ -131,6 +131,8 @@ class CodecClient : protected Logger::Loggable<Logger::Id::client>,
   // Note this is the L4 stream info, not L7.
   const StreamInfo::StreamInfo& streamInfo() { return connection_->streamInfo(); }
 
+  const Network::ClientConnectionPtr& connection() { return connection_; }
+
 protected:
   /**
    * Create a codec client and connect to a remote host/port.

source/common/http/http3/conn_pool.cc

@@ -28,6 +28,19 @@ ActiveClient::ActiveClient(Envoy::Http::HttpConnPoolImplBase& parent,
                            Upstream::Host::CreateConnectionData& data)
     : MultiplexedActiveClientBase(parent, getMaxStreams(parent.host()->cluster()),
                                   parent.host()->cluster().stats().upstream_cx_http3_total_, data) {
+  auto& connection = dynamic_cast<ActiveClient*>(this)->codec_client_->connection();
+  Quic::EnvoyQuicClientSession* session = const_cast<Quic::EnvoyQuicClientSession*>(
+      dynamic_cast<const Quic::EnvoyQuicClientSession*>(connection.get()));
+  ASSERT(session != nullptr);
+  notifier_ = std::make_unique<Quic::ScopedStreamNotifier>(
+      [this](int32_t s) -> void { onMaxStreamsChanged(s); }, *session);
+}
+
+void ActiveClient::onMaxStreamsChanged(uint32_t num_streams) {
+  updateCapacity(num_streams);
+  if (state() == ActiveClient::State::BUSY && currentUnusedCapacity() != 0) {
+    parent_.transitionActiveClientState(*this, ActiveClient::State::READY);
+  }
 }
 
 void Http3ConnPoolImpl::setQuicConfigFromClusterConfig(const Upstream::ClusterInfo& cluster,

source/common/http/http3/conn_pool.h

@@ -23,6 +23,57 @@ class ActiveClient : public MultiplexedActiveClientBase {
 public:
   ActiveClient(Envoy::Http::HttpConnPoolImplBase& parent,
                Upstream::Host::CreateConnectionData& data);
+
+  // Update quiche_capacity_ when a MAX_STREAMS frame arrives.
+  void onMaxStreamsChanged(uint32_t num_streams);
+
+  RequestEncoder& newStreamEncoder(ResponseDecoder& response_decoder) override {
+    ASSERT(quiche_capacity_ != 0);
+    // Each time a quic stream is allocated the quic capacity needs to get
+    // decremented.  See comments by quiche_capacity_.
+    updateCapacity(quiche_capacity_ - 1);
+    return MultiplexedActiveClientBase::newStreamEncoder(response_decoder);
+  }
+
+  // Overload the default capacity calculations to return the quic capacity
+  // (modified by any stream limits in Envoy config)
+  int64_t currentUnusedCapacity() const override {
+    return std::min<int64_t>(quiche_capacity_, effectiveConcurrentStreamLimit());
+  }
+
+  void updateCapacity(uint64_t new_quiche_capacity) {
+    // Each time we update the capacity make sure to reflect the update in the
+    // connection pool.
+    //
+    // Due to interplay between the max number of concurrent streams Envoy will
+    // allow and the max number of streams per connection this is not as simple
+    // as just updating based on the delta between quiche_capacity_ and
+    // new_quiche_capacity, so we use the delta between the actual calculated
+    // capacity before and after the update.
+    uint64_t old_capacity = currentUnusedCapacity();
+    quiche_capacity_ = new_quiche_capacity;
+    uint64_t new_capacity = currentUnusedCapacity();
+
+    if (new_capacity < old_capacity) {
+      parent_.decrClusterStreamCapacity(old_capacity - new_capacity);
+    } else if (old_capacity < new_capacity) {
+      parent_.incrClusterStreamCapacity(new_capacity - old_capacity);
+    }
+  }
+
+  std::unique_ptr<Quic::ScopedStreamNotifier> notifier_;
+  // Unlike HTTP/2 and HTTP/1, rather than having a cap on the number of active
+  // streams, QUIC has a fixed number of streams available which is updated via
+  // the MAX_STREAMS frame.
+  //
+  // As such each time we create a new stream for QUIC, the capacity goes down
+  // by one, but unlike the other two codecs it is _not_ restored on stream
+  // closure.
+  //
+  // We track the QUIC capacity here, and overload currentUnusedCapacity so the
+  // connection pool can accurately keep track of when it is safe to create new
+  // streams.
+  uint64_t quiche_capacity_ = 100;
 };
 
 // Http3 subclass of FixedHttpConnPoolImpl which exists to store quic data.
@@ -45,6 +96,7 @@ class Http3ConnPoolImpl : public FixedHttpConnPoolImpl {
                                              quic::QuicConfig& quic_config);
 
   Quic::PersistentQuicInfoImpl& quicInfo() { return *quic_info_; }
+  bool quic() override { return true; }
 
 private:
   // Store quic helpers which can be shared between connections and must live

source/common/quic/envoy_quic_client_session.cc

@@ -4,9 +4,31 @@
 
 #include "quic_filter_manager_connection_impl.h"
 
+namespace quic {
+namespace test {
+
+// TODO(alyssawilk) add the necessary accessors to quiche and remove this.
+class QuicSessionPeer {
+public:
+  static quic::QuicStreamIdManager& getStream(Envoy::Quic::EnvoyQuicClientSession* session) {
+    return session->ietf_streamid_manager_.bidirectional_stream_id_manager_;
+  }
+};
+
+} // namespace test
+} // namespace quic
+
 namespace Envoy {
 namespace Quic {
 
+ScopedStreamNotifier::ScopedStreamNotifier(std::function<void(uint32_t)> notify,
+                                           EnvoyQuicClientSession& session)
+    : on_can_create_streams_(notify), session_(session) {
+  session.setNotifier(*this);
+}
+
+ScopedStreamNotifier::~ScopedStreamNotifier() { session_.clearNotifier(); }
+
 EnvoyQuicClientSession::EnvoyQuicClientSession(
     const quic::QuicConfig& config, const quic::ParsedQuicVersionVector& supported_versions,
     std::unique_ptr<EnvoyQuicClientConnection> connection, const quic::QuicServerId& server_id,
@@ -87,6 +109,14 @@ void EnvoyQuicClientSession::SetDefaultEncryptionLevel(quic::EncryptionLevel lev
   }
 }
 
+void EnvoyQuicClientSession::OnCanCreateNewOutgoingStream(bool) {
+  if (notifier_.has_value()) {
+    quic::QuicStreamIdManager& manager = quic::test::QuicSessionPeer::getStream(this);
+    uint32_t streams_available = manager.outgoing_max_streams() - manager.outgoing_stream_count();
+    notifier_->notify(streams_available);
+  }
+}
+
 std::unique_ptr<quic::QuicSpdyClientStream> EnvoyQuicClientSession::CreateClientStream() {
   ASSERT(codec_stats_.has_value() && http3_options_.has_value());
   return std::make_unique<EnvoyQuicClientStream>(GetNextOutgoingBidirectionalStreamId(), this,

source/common/quic/envoy_quic_client_session.h

@@ -11,6 +11,18 @@
 namespace Envoy {
 namespace Quic {
 
+class EnvoyQuicClientSession;
+
+struct ScopedStreamNotifier {
+  ScopedStreamNotifier(std::function<void(uint32_t)> notify, EnvoyQuicClientSession& session);
+  ~ScopedStreamNotifier();
+
+  void notify(uint32_t streams_available) { on_can_create_streams_(streams_available); }
+
+  std::function<void(uint32_t)> on_can_create_streams_;
+  EnvoyQuicClientSession& session_;
+};
+
 // Act as a Network::ClientConnection to ClientCodec.
 // TODO(danzh) This class doesn't need to inherit Network::FilterManager
 // interface but need all other Network::Connection implementation in
@@ -71,6 +83,17 @@ class EnvoyQuicClientSession : public QuicFilterManagerConnectionImpl,
   // QuicFilterManagerConnectionImpl
   void setHttp3Options(const envoy::config::core::v3::Http3ProtocolOptions& http3_options) override;
 
+  bool QuicheShouldCreateOutgoingBidirectionalStream() {
+    return quic::QuicSpdyClientSession::ShouldCreateOutgoingBidirectionalStream();
+  }
+
+  // Notify any registered connection pool when new streams are available.
+  void OnCanCreateNewOutgoingStream(bool) override;
+
+  void setNotifier(ScopedStreamNotifier& notifier) { notifier_ = notifier; }
+
+  void clearNotifier() { notifier_.reset(); }
+
   using quic::QuicSpdyClientSession::PerformActionOnActiveStreams;
 
 protected:
@@ -102,6 +125,7 @@ class EnvoyQuicClientSession : public QuicFilterManagerConnectionImpl,
   EnvoyQuicCryptoClientStreamFactoryInterface& crypto_stream_factory_;
   QuicStatNames& quic_stat_names_;
   Stats::Scope& scope_;
+  OptRef<ScopedStreamNotifier> notifier_;
 };
 
 } // namespace Quic

source/common/quic/envoy_quic_client_stream.cc

@@ -286,7 +286,9 @@ void EnvoyQuicClientStream::ResetWithError(quic::QuicResetStreamError error) {
   stats_.tx_reset_.inc();
   // Upper layers expect calling resetStream() to immediately raise reset callbacks.
   runResetCallbacks(quicRstErrorToEnvoyLocalResetReason(error.internal_code()));
-  quic::QuicSpdyClientStream::ResetWithError(error);
+  if (session()->connection()->connected()) {
+    quic::QuicSpdyClientStream::ResetWithError(error);
+  }
 }
 
 void EnvoyQuicClientStream::OnConnectionClosed(quic::QuicErrorCode error,

test/integration/multiplexed_upstream_integration_test.cc

@@ -205,7 +205,6 @@ TEST_P(Http2UpstreamIntegrationTest, LargeSimultaneousRequestWithBufferLimits) {
 void Http2UpstreamIntegrationTest::manySimultaneousRequests(uint32_t request_bytes,
                                                             uint32_t max_response_bytes,
                                                             uint32_t num_requests) {
-  autonomous_allow_incomplete_streams_ = true;
   TestRandomGenerator rand;
   std::vector<Http::RequestEncoder*> encoders;
   std::vector<IntegrationStreamDecoderPtr> responses;
@@ -236,11 +235,8 @@ void Http2UpstreamIntegrationTest::manySimultaneousRequests(uint32_t request_byt
     ASSERT_TRUE(responses[i]->waitForEndStream());
     if (i % 2 != 0) {
       EXPECT_TRUE(responses[i]->complete());
-      // TODO(18160) remove this if and always check for 200 and body length.
-      if (num_requests <= 100 || upstreamProtocol() != Http::CodecType::HTTP3) {
-        EXPECT_EQ("200", responses[i]->headers().getStatusValue());
-        EXPECT_EQ(response_bytes[i], responses[i]->body().length());
-      }
+      EXPECT_EQ("200", responses[i]->headers().getStatusValue());
+      EXPECT_EQ(response_bytes[i], responses[i]->body().length());
     } else {
       // Upstream stream reset.
       EXPECT_EQ("503", responses[i]->headers().getStatusValue());
@@ -255,13 +251,9 @@ TEST_P(Http2UpstreamIntegrationTest, ManySimultaneousRequest) {
   manySimultaneousRequests(1024, 1024, 100);
 }
 
-#ifdef NDEBUG
-// TODO(alyssawilk) this causes crashes in debug mode for QUIC due to a race
-// condition between Envoy's stream accounting and QUICE's. Debug and fix.
 TEST_P(Http2UpstreamIntegrationTest, TooManySimultaneousRequests) {
   manySimultaneousRequests(1024, 1024, 200);
 }
-#endif
 
 TEST_P(Http2UpstreamIntegrationTest, ManyLargeSimultaneousRequestWithBufferLimits) {
   config_helper_.setBufferLimits(1024, 1024); // Set buffer limits upstream and downstream.