conn pool: state variable cleanup and more ENVOY_BUGs (envoyproxy#39381)

Rename `state_` to a more descriptive name and make it private. Convert
some ASSERTs to ENVOY_BUGs for better detection of accounting bugs.

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

Author: ggreenway

source/common/conn_pool/conn_pool_base.cc

@@ -40,11 +40,11 @@ void ConnPoolImplBase::assertCapacityCountsAreCorrect() {
   SLOW_ASSERT(currentUnusedCapacity(busy_clients_) <= 0, dumpState());
 
   if (ready_clients_.empty()) {
-    ASSERT((connecting_and_connected_stream_capacity_ - connecting_stream_capacity_) <= 0,
-           dumpState());
+    ENVOY_BUG((connecting_and_connected_stream_capacity_ - connecting_stream_capacity_) <= 0,
+              dumpState());
   } else {
-    ASSERT((connecting_and_connected_stream_capacity_ - connecting_stream_capacity_) > 0,
-           dumpState());
+    ENVOY_BUG((connecting_and_connected_stream_capacity_ - connecting_stream_capacity_) > 0,
+              dumpState());
   }
 }
 
@@ -53,21 +53,21 @@ ConnPoolImplBase::ConnPoolImplBase(
     Event::Dispatcher& dispatcher, const Network::ConnectionSocket::OptionsSharedPtr& options,
     const Network::TransportSocketOptionsConstSharedPtr& transport_socket_options,
     Upstream::ClusterConnectivityState& state)
-    : state_(state), host_(host), priority_(priority), dispatcher_(dispatcher),
-      socket_options_(options), transport_socket_options_(transport_socket_options),
+    : host_(host), priority_(priority), dispatcher_(dispatcher), socket_options_(options),
+      transport_socket_options_(transport_socket_options), cluster_connectivity_state_(state),
       upstream_ready_cb_(dispatcher_.createSchedulableCallback([this]() { onUpstreamReady(); })) {}
 
 ConnPoolImplBase::~ConnPoolImplBase() {
-  ASSERT(isIdleImpl(), dumpState());
-  ASSERT(connecting_stream_capacity_ == 0, dumpState());
-  ASSERT(connecting_and_connected_stream_capacity_ == 0, dumpState());
+  ENVOY_BUG(isIdleImpl(), dumpState());
+  ENVOY_BUG(connecting_stream_capacity_ == 0, dumpState());
+  ENVOY_BUG(connecting_and_connected_stream_capacity_ == 0, dumpState());
 }
 
 void ConnPoolImplBase::deleteIsPendingImpl() {
   deferred_deleting_ = true;
-  ASSERT(isIdleImpl(), dumpState());
-  ASSERT(connecting_stream_capacity_ == 0, dumpState());
-  ASSERT(connecting_and_connected_stream_capacity_ == 0, dumpState());
+  ENVOY_BUG(isIdleImpl(), dumpState());
+  ENVOY_BUG(connecting_stream_capacity_ == 0, dumpState());
+  ENVOY_BUG(connecting_and_connected_stream_capacity_ == 0, dumpState());
 }
 
 void ConnPoolImplBase::destructAllConnections() {
@@ -197,9 +197,9 @@ ConnPoolImplBase::tryCreateNewConnection(float global_preconnect_ratio) {
       return ConnectionResult::FailedToCreateConnection;
     }
     ASSERT(client->state() == ActiveClient::State::Connecting, dumpState());
-    ASSERT(std::numeric_limits<uint64_t>::max() - connecting_stream_capacity_ >=
-               static_cast<uint64_t>(client->currentUnusedCapacity()),
-           dumpState());
+    ENVOY_BUG(std::numeric_limits<uint64_t>::max() - connecting_stream_capacity_ >=
+                  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);
@@ -249,7 +249,7 @@ void ConnPoolImplBase::attachStreamToClient(Envoy::ConnectionPool::ActiveClient&
     decrConnectingAndConnectedStreamCapacity(1, client);
   }
   // Track the new active stream.
-  state_.incrActiveStreams(1);
+  cluster_connectivity_state_.incrActiveStreams(1);
   num_active_streams_++;
   host_->stats().rq_total_.inc();
   host_->stats().rq_active_.inc();
@@ -266,7 +266,7 @@ void ConnPoolImplBase::onStreamClosed(Envoy::ConnectionPool::ActiveClient& clien
       debug, "destroying stream: {} active remaining, readyForStream {}, currentUnusedCapacity {}",
       client, client.numActiveStreams(), client.readyForStream(), client.currentUnusedCapacity());
   ASSERT(num_active_streams_ > 0, dumpState());
-  state_.decrActiveStreams(1);
+  cluster_connectivity_state_.decrActiveStreams(1);
   num_active_streams_--;
   host_->stats().rq_active_.dec();
   host_->cluster().trafficStats()->upstream_rq_active_.dec();
@@ -378,7 +378,7 @@ void ConnPoolImplBase::onUpstreamReady() {
     ENVOY_CONN_LOG(debug, "attaching to next stream", *client);
     // Pending streams are pushed onto the front, so pull from the back.
     attachStreamToClient(*client, pending_streams_.back()->context());
-    state_.decrPendingStreams(1);
+    cluster_connectivity_state_.decrPendingStreams(1);
     pending_streams_.pop_back();
   }
   if (!pending_streams_.empty()) {
@@ -622,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(), dumpState());
+    ENVOY_BUG(connecting_stream_capacity_ >= client.currentUnusedCapacity(), dumpState());
     connecting_stream_capacity_ -= client.currentUnusedCapacity();
     client.has_handshake_completed_ = true;
     client.conn_connect_ms_->complete();
@@ -691,7 +691,7 @@ void ConnPoolImplBase::purgePendingStreams(
     absl::string_view failure_reason, ConnectionPool::PoolFailureReason reason) {
   // NOTE: We move the existing pending streams to a temporary list. This is done so that
   //       if retry logic submits a new stream to the pool, we don't fail it inline.
-  state_.decrPendingStreams(pending_streams_.size());
+  cluster_connectivity_state_.decrPendingStreams(pending_streams_.size());
   pending_streams_to_purge_ = std::move(pending_streams_);
   while (!pending_streams_to_purge_.empty()) {
     PendingStreamPtr stream =
@@ -703,7 +703,7 @@ void ConnPoolImplBase::purgePendingStreams(
 
 bool ConnPoolImplBase::connectingConnectionIsExcess(const ActiveClient& client) const {
   ASSERT(!client.hasHandshakeCompleted());
-  ASSERT(connecting_stream_capacity_ >= client.currentUnusedCapacity(), dumpState());
+  ENVOY_BUG(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.
@@ -725,7 +725,7 @@ void ConnPoolImplBase::onPendingStreamCancel(PendingStream& stream,
     // and there is no need to call its onPoolFailure callback.
     stream.removeFromList(pending_streams_to_purge_);
   } else {
-    state_.decrPendingStreams(1);
+    cluster_connectivity_state_.decrPendingStreams(1);
     stream.removeFromList(pending_streams_);
   }
   if (policy == Envoy::ConnectionPool::CancelPolicy::CloseExcess) {
@@ -760,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, dumpState());
+    ENVOY_BUG(connecting_stream_capacity_ >= delta, dumpState());
     connecting_stream_capacity_ -= delta;
   }
 }
@@ -796,7 +796,7 @@ void ConnPoolImplBase::onUpstreamReadyForEarlyData(ActiveClient& client) {
     if (stream.can_send_early_data_) {
       ENVOY_CONN_LOG(debug, "creating stream for early data.", client);
       attachStreamToClient(client, stream.context());
-      state_.decrPendingStreams(1);
+      cluster_connectivity_state_.decrPendingStreams(1);
       stream.removeFromList(pending_streams_);
     }
     if (stop_iteration) {

source/common/conn_pool/conn_pool_base.h

@@ -279,11 +279,11 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   bool hasPendingStreams() const { return !pending_streams_.empty(); }
 
   void decrClusterStreamCapacity(uint32_t delta) {
-    state_.decrConnectingAndConnectedStreamCapacity(delta);
+    cluster_connectivity_state_.decrConnectingAndConnectedStreamCapacity(delta);
     connecting_and_connected_stream_capacity_ -= delta;
   }
   void incrClusterStreamCapacity(uint32_t delta) {
-    state_.incrConnectingAndConnectedStreamCapacity(delta);
+    cluster_connectivity_state_.incrConnectingAndConnectedStreamCapacity(delta);
     connecting_and_connected_stream_capacity_ += delta;
   }
   void dumpState(std::ostream& os, int indent_level = 0) const {
@@ -304,8 +304,6 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   // 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);
   void incrConnectingAndConnectedStreamCapacity(uint32_t delta, ActiveClient& client);
 
@@ -349,14 +347,12 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   ConnectionPool::Cancellable*
   addPendingStream(Envoy::ConnectionPool::PendingStreamPtr&& pending_stream) {
     LinkedList::moveIntoList(std::move(pending_stream), pending_streams_);
-    state_.incrPendingStreams(1);
+    cluster_connectivity_state_.incrPendingStreams(1);
     return pending_streams_.front().get();
   }
 
   bool hasActiveStreams() const { return num_active_streams_ > 0; }
 
-  Upstream::ClusterConnectivityState& state_;
-
   const Upstream::HostConstSharedPtr host_;
   const Upstream::ResourcePriority priority_;
 
@@ -399,6 +395,8 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
 
   void assertCapacityCountsAreCorrect();
 
+  Upstream::ClusterConnectivityState& cluster_connectivity_state_;
+
   std::list<PendingStreamPtr> pending_streams_;
 
   // The number of streams that can be immediately dispatched from the current