conn pool: add extra details to ASSERT, clarify comments (#39286)

Also added a test for a case that wasn't covered for negative capacity,
and removed a function that wasn't used.

Signed-off-by: Greg Greenway <[email protected]>

Author: ggreenway

source/common/conn_pool/conn_pool_base.cc

@@ -19,10 +19,13 @@ int64_t currentUnusedCapacity(const std::list<ActiveClientPtr>& connecting_clien
 }
 } // namespace
 
+std::string ConnPoolImplBase::dumpState() const { return fmt::format("State: {}", *this); }
+
 void ConnPoolImplBase::assertCapacityCountsAreCorrect() {
   SLOW_ASSERT(static_cast<int64_t>(connecting_stream_capacity_) ==
-              currentUnusedCapacity(connecting_clients_) +
-                  currentUnusedCapacity(early_data_clients_));
+                  currentUnusedCapacity(connecting_clients_) +
+                      currentUnusedCapacity(early_data_clients_),
+              dumpState());
 
   // Note: must include `busy_clients_` because they can have negative current unused capacity,
   // which is included in `connecting_and_connected_stream_capacity_`.
@@ -31,10 +34,18 @@ void ConnPoolImplBase::assertCapacityCountsAreCorrect() {
           (static_cast<int64_t>(connecting_stream_capacity_) +
            currentUnusedCapacity(ready_clients_) + currentUnusedCapacity(busy_clients_)),
       fmt::format(
-          "connecting_and_connected_stream_capacity_ {}, connecting_stream_capacity_ {}, "
-          "currentUnusedCapacity(ready_clients_) {}, currentUnusedCapacity(busy_clients_) {}",
-          connecting_and_connected_stream_capacity_, connecting_stream_capacity_,
-          currentUnusedCapacity(ready_clients_), currentUnusedCapacity(busy_clients_)));
+          "{} currentUnusedCapacity(ready_clients_) {}, currentUnusedCapacity(busy_clients_) {}",
+          *this, currentUnusedCapacity(ready_clients_), currentUnusedCapacity(busy_clients_)));
+
+  SLOW_ASSERT(currentUnusedCapacity(busy_clients_) <= 0, dumpState());
+
+  if (ready_clients_.empty()) {
+    ASSERT((connecting_and_connected_stream_capacity_ - connecting_stream_capacity_) <= 0,
+           dumpState());
+  } else {
+    ASSERT((connecting_and_connected_stream_capacity_ - connecting_stream_capacity_) > 0,
+           dumpState());
+  }
 }
 
 ConnPoolImplBase::ConnPoolImplBase(
@@ -47,16 +58,16 @@ ConnPoolImplBase::ConnPoolImplBase(
       upstream_ready_cb_(dispatcher_.createSchedulableCallback([this]() { onUpstreamReady(); })) {}
 
 ConnPoolImplBase::~ConnPoolImplBase() {
-  ASSERT(isIdleImpl());
-  ASSERT(connecting_stream_capacity_ == 0);
-  ASSERT(connecting_and_connected_stream_capacity_ == 0);
+  ASSERT(isIdleImpl(), dumpState());
+  ASSERT(connecting_stream_capacity_ == 0, dumpState());
+  ASSERT(connecting_and_connected_stream_capacity_ == 0, dumpState());
 }
 
 void ConnPoolImplBase::deleteIsPendingImpl() {
   deferred_deleting_ = true;
-  ASSERT(isIdleImpl());
-  ASSERT(connecting_stream_capacity_ == 0);
-  ASSERT(connecting_and_connected_stream_capacity_ == 0);
+  ASSERT(isIdleImpl(), dumpState());
+  ASSERT(connecting_stream_capacity_ == 0, dumpState());
+  ASSERT(connecting_and_connected_stream_capacity_ == 0, dumpState());
 }
 
 void ConnPoolImplBase::destructAllConnections() {
@@ -156,7 +167,8 @@ ConnPoolImplBase::ConnectionResult ConnPoolImplBase::tryCreateNewConnections() {
       break;
     }
   }
-  ASSERT(!is_draining_for_deletion_ || result != ConnectionResult::CreatedNewConnection);
+  ASSERT(!is_draining_for_deletion_ || result != ConnectionResult::CreatedNewConnection,
+         dumpState());
   return result;
 }
 
@@ -184,9 +196,10 @@ ConnPoolImplBase::tryCreateNewConnection(float global_preconnect_ratio) {
       ENVOY_LOG(trace, "connection creation failed");
       return ConnectionResult::FailedToCreateConnection;
     }
-    ASSERT(client->state() == ActiveClient::State::Connecting);
+    ASSERT(client->state() == ActiveClient::State::Connecting, dumpState());
     ASSERT(std::numeric_limits<uint64_t>::max() - connecting_stream_capacity_ >=
-           static_cast<uint64_t>(client->currentUnusedCapacity()));
+               static_cast<uint64_t>(client->currentUnusedCapacity()),
+           dumpState());
     ASSERT(client->real_host_description_);
     // Increase the connecting capacity to reflect the streams this connection can serve.
     incrConnectingAndConnectedStreamCapacity(client->currentUnusedCapacity(), *client);
@@ -202,7 +215,7 @@ ConnPoolImplBase::tryCreateNewConnection(float global_preconnect_ratio) {
 
 void ConnPoolImplBase::attachStreamToClient(Envoy::ConnectionPool::ActiveClient& client,
                                             AttachContext& context) {
-  ASSERT(client.readyForStream());
+  ASSERT(client.readyForStream(), dumpState());
 
   Upstream::ClusterTrafficStats& traffic_stats = *host_->cluster().trafficStats();
   if (client.state() == Envoy::ConnectionPool::ActiveClient::State::ReadyForEarlyData) {
@@ -252,7 +265,7 @@ void ConnPoolImplBase::onStreamClosed(Envoy::ConnectionPool::ActiveClient& clien
   ENVOY_CONN_LOG(
       debug, "destroying stream: {} active remaining, readyForStream {}, currentUnusedCapacity {}",
       client, client.numActiveStreams(), client.readyForStream(), client.currentUnusedCapacity());
-  ASSERT(num_active_streams_ > 0);
+  ASSERT(num_active_streams_ > 0, dumpState());
   state_.decrActiveStreams(1);
   num_active_streams_--;
   host_->stats().rq_active_.dec();
@@ -265,9 +278,9 @@ void ConnPoolImplBase::onStreamClosed(Envoy::ConnectionPool::ActiveClient& clien
     bool limited_by_concurrency =
         client.remaining_streams_ > client.concurrent_stream_limit_ - client.numActiveStreams() - 1;
     // The capacity calculated by concurrency could be negative if a SETTINGS frame lowered the
-    // number of allowed streams. In this case, effective client capacity was still limited by
-    // concurrency, compare client.concurrent_stream_limit_ and client.numActiveStreams() directly
-    // to avoid overflow.
+    // number of allowed streams. In this case, connecting_and_connected_stream_capacity_ can be
+    // negative, and effective client capacity was still limited by concurrency. Compare
+    // client.concurrent_stream_limit_ and client.numActiveStreams() directly to avoid overflow.
     bool negative_capacity = client.concurrent_stream_limit_ < client.numActiveStreams() + 1;
     if (negative_capacity || limited_by_concurrency) {
       incrConnectingAndConnectedStreamCapacity(1, client);
@@ -293,8 +306,8 @@ void ConnPoolImplBase::onStreamClosed(Envoy::ConnectionPool::ActiveClient& clien
 
 ConnectionPool::Cancellable* ConnPoolImplBase::newStreamImpl(AttachContext& context,
                                                              bool can_send_early_data) {
-  ASSERT(!is_draining_for_deletion_);
-  ASSERT(!deferred_deleting_);
+  ASSERT(!is_draining_for_deletion_, dumpState());
+  ASSERT(!deferred_deleting_, dumpState());
   assertCapacityCountsAreCorrect();
 
   if (!ready_clients_.empty()) {
@@ -351,7 +364,7 @@ ConnectionPool::Cancellable* ConnPoolImplBase::newStreamImpl(AttachContext& cont
 }
 
 bool ConnPoolImplBase::maybePreconnectImpl(float global_preconnect_ratio) {
-  ASSERT(!deferred_deleting_);
+  ASSERT(!deferred_deleting_, dumpState());
   return tryCreateNewConnection(global_preconnect_ratio) == ConnectionResult::CreatedNewConnection;
 }
 
@@ -440,7 +453,7 @@ void ConnPoolImplBase::closeIdleConnectionsForDrainingPool() {
 
 void ConnPoolImplBase::drainClients(std::list<ActiveClientPtr>& clients) {
   while (!clients.empty()) {
-    ASSERT(clients.front()->numActiveStreams() > 0u);
+    ASSERT(clients.front()->numActiveStreams() > 0u, dumpState());
     ENVOY_LOG_EVENT(
         debug, "draining_non_idle_client", "draining {} client {} for cluster {}",
         (clients.front()->state() == ActiveClient::State::Ready ? "ready" : "early data"),
@@ -469,7 +482,7 @@ void ConnPoolImplBase::drainConnectionsImpl(DrainBehavior drain_behavior) {
 
   // Changing busy_clients_ to Draining does not move them between lists,
   // so use a for-loop since the list is not mutated.
-  ASSERT(&owningList(ActiveClient::State::Draining) == &busy_clients_);
+  ASSERT(&owningList(ActiveClient::State::Draining) == &busy_clients_, dumpState());
   for (auto& busy_client : busy_clients_) {
     if (busy_client->state() == ActiveClient::State::Draining) {
       continue;
@@ -609,7 +622,7 @@ void ConnPoolImplBase::onConnectionEvent(ActiveClient& client, absl::string_view
     client.connect_timer_->disableTimer();
     client.connect_timer_.reset();
 
-    ASSERT(connecting_stream_capacity_ >= client.currentUnusedCapacity());
+    ASSERT(connecting_stream_capacity_ >= client.currentUnusedCapacity(), dumpState());
     connecting_stream_capacity_ -= client.currentUnusedCapacity();
     client.has_handshake_completed_ = true;
     client.conn_connect_ms_->complete();
@@ -690,7 +703,7 @@ void ConnPoolImplBase::purgePendingStreams(
 
 bool ConnPoolImplBase::connectingConnectionIsExcess(const ActiveClient& client) const {
   ASSERT(!client.hasHandshakeCompleted());
-  ASSERT(connecting_stream_capacity_ >= client.currentUnusedCapacity());
+  ASSERT(connecting_stream_capacity_ >= client.currentUnusedCapacity(), dumpState());
   // If perUpstreamPreconnectRatio is one, this simplifies to checking if there would still be
   // sufficient connecting stream capacity to serve all pending streams if the most recent client
   // were removed from the picture.
@@ -747,7 +760,7 @@ void ConnPoolImplBase::decrConnectingAndConnectedStreamCapacity(uint32_t delta,
   if (!client.hasHandshakeCompleted()) {
     // If still doing handshake, it is contributing to the local connecting stream capacity. Update
     // the capacity as well.
-    ASSERT(connecting_stream_capacity_ >= delta);
+    ASSERT(connecting_stream_capacity_ >= delta, dumpState());
     connecting_stream_capacity_ -= delta;
   }
 }

source/common/conn_pool/conn_pool_base.h

@@ -85,12 +85,6 @@ class ActiveClient : public LinkedObject<ActiveClient>,
     return state_ == State::Ready;
   }
 
-  // This function is called onStreamClosed to see if there was a negative delta
-  // and (if necessary) update associated bookkeeping.
-  // HTTP/1 and TCP pools can not have negative delta so the default implementation simply returns
-  // false. The HTTP/2 connection pool can have this state, so overrides this function.
-  virtual bool hadNegativeDeltaOnStreamClosed() { return false; }
-
   enum class State {
     Connecting,        // Connection is not yet established.
     ReadyForEarlyData, // Any additional early data stream can be immediately dispatched to this
@@ -296,7 +290,8 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
     const char* spaces = spacesForLevel(indent_level);
     os << spaces << "ConnPoolImplBase " << this << DUMP_MEMBER(ready_clients_.size())
        << DUMP_MEMBER(busy_clients_.size()) << DUMP_MEMBER(connecting_clients_.size())
-       << DUMP_MEMBER(connecting_stream_capacity_) << DUMP_MEMBER(num_active_streams_)
+       << DUMP_MEMBER(connecting_stream_capacity_)
+       << DUMP_MEMBER(connecting_and_connected_stream_capacity_) << DUMP_MEMBER(num_active_streams_)
        << DUMP_MEMBER(pending_streams_.size())
        << " per upstream preconnect ratio: " << perUpstreamPreconnectRatio();
   }
@@ -305,6 +300,10 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
     s.dumpState(os);
     return os;
   }
+
+  // Helper for use as the 2nd argument to ASSERT.
+  std::string dumpState() const;
+
   Upstream::ClusterConnectivityState& state() { return state_; }
 
   void decrConnectingAndConnectedStreamCapacity(uint32_t delta, ActiveClient& client);

test/common/conn_pool/conn_pool_base_test.cc

@@ -250,6 +250,7 @@ TEST_F(ConnPoolImplBaseTest, DumpState) {
   std::string state = out.str();
   EXPECT_THAT(state, HasSubstr("ready_clients_.size(): 0, busy_clients_.size(): 0, "
                                "connecting_clients_.size(): 0, connecting_stream_capacity_: 0, "
+                               "connecting_and_connected_stream_capacity_: 0, "
                                "num_active_streams_: 0"));
 }
 

test/common/http/http2/conn_pool_test.cc

@@ -1805,6 +1805,49 @@ TEST_F(Http2ConnPoolImplTest, TestUnusedCapacity) {
   CHECK_STATE(0 /*active*/, 0 /*pending*/, 0 /*capacity*/);
 }
 
+TEST_F(Http2ConnPoolImplTest, TestNegativeUnusedCapacity) {
+  cluster_->http2_options_.mutable_max_concurrent_streams()->set_value(4);
+
+  expectClientsCreate(2);
+  ActiveTestRequest r1(*this, 0, false);
+  CHECK_STATE(0 /*active*/, 1 /*pending*/, 4 /*capacity*/);
+  expectClientConnect(0, r1);
+  CHECK_STATE(1 /*active*/, 0 /*pending*/, 3 /*capacity*/);
+
+  ActiveTestRequest r2(*this, 0, true);
+  ActiveTestRequest r3(*this, 0, true);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, 1 /*capacity*/);
+
+  // Settings frame results in negative unused capacity.
+  NiceMock<MockReceivedSettings> settings;
+  settings.max_concurrent_streams_ = 1;
+  test_clients_[0].codec_client_->onSettings(settings);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, -2 /*capacity*/);
+
+  completeRequest(r1);
+  CHECK_STATE(2 /*active*/, 0 /*pending*/, -1 /*capacity*/);
+
+  // With negative capacity, verify that a new request creates a new connection.
+  ActiveTestRequest r4(*this, 1, false);
+  CHECK_STATE(2 /*active*/, 1 /*pending*/, 3 /*capacity*/);
+  expectClientConnect(1, r4);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, 2 /*capacity*/);
+
+  completeRequest(r2);
+  CHECK_STATE(2 /*active*/, 0 /*pending*/, 3 /*capacity*/);
+
+  completeRequest(r3);
+  CHECK_STATE(1 /*active*/, 0 /*pending*/, 4 /*capacity*/);
+
+  completeRequest(r4);
+  CHECK_STATE(0 /*active*/, 0 /*pending*/, 5 /*capacity*/);
+
+  // Clean up with an outstanding stream.
+  pool_->drainConnections(Envoy::ConnectionPool::DrainBehavior::DrainExistingConnections);
+  closeAllClients();
+  CHECK_STATE(0 /*active*/, 0 /*pending*/, 0 /*capacity*/);
+}
+
 TEST_F(Http2ConnPoolImplTest, TestStateWithMultiplexing) {
   cluster_->http2_options_.mutable_max_concurrent_streams()->set_value(2);
   cluster_->max_requests_per_connection_ = 4;