http: use a request queue in the http2 conn pool

source/common/http/BUILD

@@ -92,6 +92,16 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "conn_pool_base_lib",
+    srcs = ["conn_pool_base.cc"],
+    hdrs = ["conn_pool_base.h"],
+    deps = [
+        "//include/envoy/http:conn_pool_interface",
+        "//source/common/common:linked_object",
+    ],
+)
+
 envoy_cc_library(
     name = "conn_manager_config_interface",
     hdrs = ["conn_manager_config.h"],

source/common/http/conn_pool_base.cc

@@ -0,0 +1,48 @@
+#include "common/http/conn_pool_base.h"
+
+namespace Envoy {
+namespace Http {
+ConnPoolImplBase::PendingRequest::PendingRequest(ConnPoolImplBase& parent, StreamDecoder& decoder,
+                                                 ConnectionPool::Callbacks& callbacks)
+    : parent_(parent), decoder_(decoder), callbacks_(callbacks) {
+  parent_.host_->cluster().stats().upstream_rq_pending_total_.inc();
+  parent_.host_->cluster().stats().upstream_rq_pending_active_.inc();
+  parent_.host_->cluster().resourceManager(parent_.priority_).pendingRequests().inc();
+}
+
+ConnPoolImplBase::PendingRequest::~PendingRequest() {
+  parent_.host_->cluster().stats().upstream_rq_pending_active_.dec();
+  parent_.host_->cluster().resourceManager(parent_.priority_).pendingRequests().dec();
+}
+
+ConnectionPool::Cancellable*
+ConnPoolImplBase::newPendingRequest(StreamDecoder& decoder, ConnectionPool::Callbacks& callbacks) {
+  ENVOY_LOG(debug, "queueing request due to no available connections");
+  PendingRequestPtr pending_request(new PendingRequest(*this, decoder, callbacks));
+  pending_request->moveIntoList(std::move(pending_request), pending_requests_);
+  return pending_requests_.front().get();
+}
+
+void ConnPoolImplBase::purgePendingRequests(
+    const Upstream::HostDescriptionConstSharedPtr& host_description) {
+  // NOTE: We move the existing pending requests to a temporary list. This is done so that
+  //       if retry logic submits a new request to the pool, we don't fail it inline.
+  std::list<PendingRequestPtr> pending_requests_to_purge(std::move(pending_requests_));
+  while (!pending_requests_to_purge.empty()) {
+    PendingRequestPtr request =
+        pending_requests_to_purge.front()->removeFromList(pending_requests_to_purge);
+    host_->cluster().stats().upstream_rq_pending_failure_eject_.inc();
+    request->callbacks_.onPoolFailure(ConnectionPool::PoolFailureReason::ConnectionFailure,
+                                      host_description);
+  }
+}
+
+void ConnPoolImplBase::onPendingRequestCancel(PendingRequest& request) {
+  ENVOY_LOG(debug, "cancelling pending request");
+  request.removeFromList(pending_requests_);
+  host_->cluster().stats().upstream_rq_cancelled_.inc();
+  checkForDrained();
+}
+
+} // namespace Http
+} // namespace Envoy

source/common/http/conn_pool_base.h

@@ -0,0 +1,50 @@
+#pragma once
+
+#include "envoy/http/conn_pool.h"
+
+#include "common/common/linked_object.h"
+
+namespace Envoy {
+namespace Http {
+
+// Base class that handles request queueing logic shared between connection pool implementations.
+class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
+protected:
+  ConnPoolImplBase(Upstream::HostConstSharedPtr host, Upstream::ResourcePriority priority)
+      : host_(host), priority_(priority) {}
+  virtual ~ConnPoolImplBase() = default;
+
+  struct PendingRequest : LinkedObject<PendingRequest>, public ConnectionPool::Cancellable {
+    PendingRequest(ConnPoolImplBase& parent, StreamDecoder& decoder,
+                   ConnectionPool::Callbacks& callbacks);
+    ~PendingRequest();
+
+    // ConnectionPool::Cancellable
+    void cancel() override { parent_.onPendingRequestCancel(*this); }
+
+    ConnPoolImplBase& parent_;
+    StreamDecoder& decoder_;
+    ConnectionPool::Callbacks& callbacks_;
+  };
+
+  typedef std::unique_ptr<PendingRequest> PendingRequestPtr;
+
+  // Creates a new PendingRequest and enqueues it into the request queue.
+  ConnectionPool::Cancellable* newPendingRequest(StreamDecoder& decoder,
+                                                 ConnectionPool::Callbacks& callbacks);
+  // Removes the PendingRequest from the list of requests. Called when the PendingRequest is
+  // cancelled, e.g. when the stream is reset before a connection has been established.
+  void onPendingRequestCancel(PendingRequest& request);
+
+  // Fails all pending requests, calling onPoolFailure on the associated callbacks.
+  void purgePendingRequests(const Upstream::HostDescriptionConstSharedPtr& host_description);
+
+  // Must be implemented by sub class. Attempts to drain inactive clients.
+  virtual void checkForDrained() PURE;
+
+  const Upstream::HostConstSharedPtr host_;
+  const Upstream::ResourcePriority priority_;
+  std::list<PendingRequestPtr> pending_requests_;
+};
+} // namespace Http
+} // namespace Envoy

source/common/http/http1/BUILD

@@ -52,6 +52,7 @@ envoy_cc_library(
         "//source/common/http:codec_client_lib",
         "//source/common/http:codec_wrappers_lib",
         "//source/common/http:codes_lib",
+        "//source/common/http:conn_pool_base_lib",
         "//source/common/http:headers_lib",
         "//source/common/network:utility_lib",
         "//source/common/upstream:upstream_lib",

source/common/http/http1/conn_pool.cc

@@ -23,7 +23,8 @@ namespace Http1 {
 ConnPoolImpl::ConnPoolImpl(Event::Dispatcher& dispatcher, Upstream::HostConstSharedPtr host,
                            Upstream::ResourcePriority priority,
                            const Network::ConnectionSocket::OptionsSharedPtr& options)
-    : dispatcher_(dispatcher), host_(host), priority_(priority), socket_options_(options),
+    : ConnPoolImplBase(std::move(host), std::move(priority)), dispatcher_(dispatcher),
+      socket_options_(options),
       upstream_ready_timer_(dispatcher_.createTimer([this]() { onUpstreamReady(); })) {}
 
 ConnPoolImpl::~ConnPoolImpl() {
@@ -104,10 +105,7 @@ ConnectionPool::Cancellable* ConnPoolImpl::newStream(StreamDecoder& response_dec
       createNewConnection();
     }
 
-    ENVOY_LOG(debug, "queueing request due to no available connections");
-    PendingRequestPtr pending_request(new PendingRequest(*this, response_decoder, callbacks));
-    pending_request->moveIntoList(std::move(pending_request), pending_requests_);
-    return pending_requests_.front().get();
+    return newPendingRequest(response_decoder, callbacks);
   } else {
     ENVOY_LOG(debug, "max pending requests overflow");
     callbacks.onPoolFailure(ConnectionPool::PoolFailureReason::Overflow, nullptr);
@@ -154,16 +152,7 @@ void ConnPoolImpl::onConnectionEvent(ActiveClient& client, Network::ConnectionEv
       // that is behaving badly, requests can get stuck here in the pending state. If we see a
       // connect failure, we purge all pending requests so that calling code can determine what to
       // do with the request.
-      // NOTE: We move the existing pending requests to a temporary list. This is done so that
-      //       if retry logic submits a new request to the pool, we don't fail it inline.
-      std::list<PendingRequestPtr> pending_requests_to_purge(std::move(pending_requests_));
-      while (!pending_requests_to_purge.empty()) {
-        PendingRequestPtr request =
-            pending_requests_to_purge.front()->removeFromList(pending_requests_to_purge);
-        host_->cluster().stats().upstream_rq_pending_failure_eject_.inc();
-        request->callbacks_.onPoolFailure(ConnectionPool::PoolFailureReason::ConnectionFailure,
-                                          client.real_host_description_);
-      }
+      purgePendingRequests(client.real_host_description_);
     }
 
     dispatcher_.deferredDelete(std::move(removed));
@@ -198,13 +187,6 @@ void ConnPoolImpl::onDownstreamReset(ActiveClient& client) {
   client.codec_client_->close();
 }
 
-void ConnPoolImpl::onPendingRequestCancel(PendingRequest& request) {
-  ENVOY_LOG(debug, "cancelling pending request");
-  request.removeFromList(pending_requests_);
-  host_->cluster().stats().upstream_rq_cancelled_.inc();
-  checkForDrained();
-}
-
 void ConnPoolImpl::onResponseComplete(ActiveClient& client) {
   ENVOY_CONN_LOG(debug, "response complete", *client.codec_client_);
   if (!client.stream_wrapper_->encode_complete_) {
@@ -295,19 +277,6 @@ void ConnPoolImpl::StreamWrapper::onDecodeComplete() {
   parent_.parent_.onResponseComplete(parent_);
 }
 
-ConnPoolImpl::PendingRequest::PendingRequest(ConnPoolImpl& parent, StreamDecoder& decoder,
-                                             ConnectionPool::Callbacks& callbacks)
-    : parent_(parent), decoder_(decoder), callbacks_(callbacks) {
-  parent_.host_->cluster().stats().upstream_rq_pending_total_.inc();
-  parent_.host_->cluster().stats().upstream_rq_pending_active_.inc();
-  parent_.host_->cluster().resourceManager(parent_.priority_).pendingRequests().inc();
-}
-
-ConnPoolImpl::PendingRequest::~PendingRequest() {
-  parent_.host_->cluster().stats().upstream_rq_pending_active_.dec();
-  parent_.host_->cluster().resourceManager(parent_.priority_).pendingRequests().dec();
-}
-
 ConnPoolImpl::ActiveClient::ActiveClient(ConnPoolImpl& parent)
     : parent_(parent),
       connect_timer_(parent_.dispatcher_.createTimer([this]() -> void { onConnectTimeout(); })),

source/common/http/http1/conn_pool.h

@@ -14,6 +14,7 @@
 #include "common/common/linked_object.h"
 #include "common/http/codec_client.h"
 #include "common/http/codec_wrappers.h"
+#include "common/http/conn_pool_base.h"
 
 #include "absl/types/optional.h"
 
@@ -27,7 +28,7 @@ namespace Http1 {
  *       address. Higher layer code should handle resolving DNS on error and creating a new pool
  *       bound to a different IP address.
  */
-class ConnPoolImpl : Logger::Loggable<Logger::Id::pool>, public ConnectionPool::Instance {
+class ConnPoolImpl : public ConnectionPool::Instance, public ConnPoolImplBase {
 public:
   ConnPoolImpl(Event::Dispatcher& dispatcher, Upstream::HostConstSharedPtr host,
                Upstream::ResourcePriority priority,
@@ -42,6 +43,9 @@ class ConnPoolImpl : Logger::Loggable<Logger::Id::pool>, public ConnectionPool::
   ConnectionPool::Cancellable* newStream(StreamDecoder& response_decoder,
                                          ConnectionPool::Callbacks& callbacks) override;
 
+  // ConnPoolImplBase
+  void checkForDrained() override;
+
 protected:
   struct ActiveClient;
 
@@ -98,41 +102,21 @@ class ConnPoolImpl : Logger::Loggable<Logger::Id::pool>, public ConnectionPool::
 
   typedef std::unique_ptr<ActiveClient> ActiveClientPtr;
 
-  struct PendingRequest : LinkedObject<PendingRequest>, public ConnectionPool::Cancellable {
-    PendingRequest(ConnPoolImpl& parent, StreamDecoder& decoder,
-                   ConnectionPool::Callbacks& callbacks);
-    ~PendingRequest();
-
-    // Cancellable
-    void cancel() override { parent_.onPendingRequestCancel(*this); }
-
-    ConnPoolImpl& parent_;
-    StreamDecoder& decoder_;
-    ConnectionPool::Callbacks& callbacks_;
-  };
-
-  typedef std::unique_ptr<PendingRequest> PendingRequestPtr;
-
   void attachRequestToClient(ActiveClient& client, StreamDecoder& response_decoder,
                              ConnectionPool::Callbacks& callbacks);
   virtual CodecClientPtr createCodecClient(Upstream::Host::CreateConnectionData& data) PURE;
-  void checkForDrained();
   void createNewConnection();
   void onConnectionEvent(ActiveClient& client, Network::ConnectionEvent event);
   void onDownstreamReset(ActiveClient& client);
-  void onPendingRequestCancel(PendingRequest& request);
   void onResponseComplete(ActiveClient& client);
   void onUpstreamReady();
   void processIdleClient(ActiveClient& client, bool delay);
 
   Stats::TimespanPtr conn_connect_ms_;
   Event::Dispatcher& dispatcher_;
-  Upstream::HostConstSharedPtr host_;
   std::list<ActiveClientPtr> ready_clients_;
   std::list<ActiveClientPtr> busy_clients_;
-  std::list<PendingRequestPtr> pending_requests_;
   std::list<DrainedCb> drained_callbacks_;
-  Upstream::ResourcePriority priority_;
   const Network::ConnectionSocket::OptionsSharedPtr socket_options_;
   Event::TimerPtr upstream_ready_timer_;
   bool upstream_ready_enabled_{false};

source/common/http/http2/BUILD

@@ -53,6 +53,7 @@ envoy_cc_library(
         "//include/envoy/stats:timespan",
         "//include/envoy/upstream:upstream_interface",
         "//source/common/http:codec_client_lib",
+        "//source/common/http:conn_pool_base_lib",
         "//source/common/network:utility_lib",
         "//source/common/upstream:upstream_lib",
     ],