listener: move active connection collection out of active tcp listener

source/server/active_stream_listener_base.cc

@@ -3,7 +3,6 @@
 #include "envoy/network/filter.h"
 
 #include "source/common/stats/timespan_impl.h"
-#include "source/server/active_tcp_listener.h"
 
 namespace Envoy {
 namespace Server {
@@ -61,7 +60,7 @@ void ActiveStreamListenerBase::newConnection(Network::ConnectionSocketPtr&& sock
   newActiveConnection(*filter_chain, std::move(server_conn_ptr), std::move(stream_info));
 }
 
-ActiveConnections::ActiveConnections(ActiveTcpListener& listener,
+ActiveConnections::ActiveConnections(OwnedActiveStreamListenerBase& listener,
                                      const Network::FilterChain& filter_chain)
     : listener_(listener), filter_chain_(filter_chain) {}
 
@@ -117,5 +116,50 @@ void ActiveTcpConnection::onEvent(Network::ConnectionEvent event) {
   }
 }
 
+void OwnedActiveStreamListenerBase::removeConnection(ActiveTcpConnection& connection) {
+  ENVOY_CONN_LOG(debug, "adding to cleanup list", *connection.connection_);
+  ActiveConnections& active_connections = connection.active_connections_;
+  ActiveConnectionPtr removed = connection.removeFromList(active_connections.connections_);
+  dispatcher().deferredDelete(std::move(removed));
+  // Delete map entry only iff connections becomes empty.
+  if (active_connections.connections_.empty()) {
+    auto iter = connections_by_context_.find(&active_connections.filter_chain_);
+    ASSERT(iter != connections_by_context_.end());
+    // To cover the lifetime of every single connection, Connections need to be deferred deleted
+    // because the previously contained connection is deferred deleted.
+    dispatcher().deferredDelete(std::move(iter->second));
+    // The erase will break the iteration over the connections_by_context_ during the deletion.
+    if (!is_deleting_) {
+      connections_by_context_.erase(iter);
+    }
+  }
+}
+
+ActiveConnections& OwnedActiveStreamListenerBase::getOrCreateActiveConnections(
+    const Network::FilterChain& filter_chain) {
+  ActiveConnectionCollectionPtr& connections = connections_by_context_[&filter_chain];
+  if (connections == nullptr) {
+    connections = std::make_unique<ActiveConnections>(*this, filter_chain);
+  }
+  return *connections;
+}
+
+void OwnedActiveStreamListenerBase::removeFilterChain(const Network::FilterChain* filter_chain) {
+  auto iter = connections_by_context_.find(filter_chain);
+  if (iter == connections_by_context_.end()) {
+    // It is possible when listener is stopping.
+  } else {
+    auto& connections = iter->second->connections_;
+    while (!connections.empty()) {
+      connections.front()->connection_->close(Network::ConnectionCloseType::NoFlush);
+    }
+    // Since is_deleting_ is on, we need to manually remove the map value and drive the
+    // iterator. Defer delete connection container to avoid race condition in destroying
+    // connection.
+    dispatcher().deferredDelete(std::move(iter->second));
+    connections_by_context_.erase(iter);
+  }
+}
+
 } // namespace Server
 } // namespace Envoy

source/server/active_stream_listener_base.h

@@ -24,6 +24,7 @@ namespace Server {
 // After the active socket passes all the listener filters, a server connection is created. The
 // derived listener must override ``newActiveConnection`` to take the ownership of that server
 // connection.
+// TODO(lambdai): Refactor the listener filter test cases to adopt this class.
 class ActiveStreamListenerBase : public ActiveListenerImplBase,
                                  protected Logger::Loggable<Logger::Id::conn_handler> {
 public:
@@ -137,20 +138,21 @@ class ActiveStreamListenerBase : public ActiveListenerImplBase,
 };
 
 struct ActiveTcpConnection;
-class ActiveTcpListener;
+class OwnedActiveStreamListenerBase;
 
 /**
  * Wrapper for a group of active connections which are attached to the same filter chain context.
  */
 class ActiveConnections : public Event::DeferredDeletable {
 public:
-  ActiveConnections(ActiveTcpListener& listener, const Network::FilterChain& filter_chain);
+  ActiveConnections(OwnedActiveStreamListenerBase& listener,
+                    const Network::FilterChain& filter_chain);
   ~ActiveConnections() override;
 
   // listener filter chain pair is the owner of the connections
-  ActiveTcpListener& listener_;
+  OwnedActiveStreamListenerBase& listener_;
   const Network::FilterChain& filter_chain_;
-  // Owned connections
+  // Owned connections.
   std::list<std::unique_ptr<ActiveTcpConnection>> connections_;
 };
 
@@ -179,5 +181,37 @@ struct ActiveTcpConnection : LinkedObject<ActiveTcpConnection>,
 using ActiveConnectionPtr = std::unique_ptr<ActiveTcpConnection>;
 using ActiveConnectionCollectionPtr = std::unique_ptr<ActiveConnections>;
 
+// The mixin that handles the composition type ActiveConnectionCollection. This mixin
+// provides the connection removal helper and the filter chain removal helper.
+// All the prod stream listeners should inherit from this class and leave ActiveStreamListenerBase
+// for unit test only.
+class OwnedActiveStreamListenerBase : public ActiveStreamListenerBase {
+public:
+  OwnedActiveStreamListenerBase(Network::ConnectionHandler& parent, Event::Dispatcher& dispatcher,
+                                Network::ListenerPtr&& listener, Network::ListenerConfig& config)
+      : ActiveStreamListenerBase(parent, dispatcher, std::move(listener), config) {}
+
+  /**
+   * Remove and destroy an active connection.
+   * @param connection supplies the connection to remove.
+   */
+  void removeConnection(ActiveTcpConnection& connection);
+
+protected:
+  /**
+   * Return the active connections container attached to the given filter chain.
+   */
+  ActiveConnections& getOrCreateActiveConnections(const Network::FilterChain& filter_chain);
+
+  /**
+   * Remove an filter chain. All the active connections that are attached to the filter chain will
+   * be destroyed.
+   * @param filter_chain supplies the filter chain to remove.
+   */
+  void removeFilterChain(const Network::FilterChain* filter_chain) override;
+
+  absl::flat_hash_map<const Network::FilterChain*, ActiveConnectionCollectionPtr>
+      connections_by_context_;
+};
 } // namespace Server
 } // namespace Envoy

source/server/active_tcp_listener.cc

@@ -8,26 +8,25 @@
 #include "source/common/common/assert.h"
 #include "source/common/network/connection_impl.h"
 #include "source/common/network/utility.h"
-#include "source/common/stats/timespan_impl.h"
 
 namespace Envoy {
 namespace Server {
 
 ActiveTcpListener::ActiveTcpListener(Network::TcpConnectionHandler& parent,
                                      Network::ListenerConfig& config, uint32_t worker_index)
-    : ActiveStreamListenerBase(parent, parent.dispatcher(),
-                               parent.dispatcher().createListener(
-                                   config.listenSocketFactory().getListenSocket(worker_index),
-                                   *this, config.bindToPort()),
-                               config),
+    : OwnedActiveStreamListenerBase(parent, parent.dispatcher(),
+                                    parent.dispatcher().createListener(
+                                        config.listenSocketFactory().getListenSocket(worker_index),
+                                        *this, config.bindToPort()),
+                                    config),
       tcp_conn_handler_(parent) {
   config.connectionBalancer().registerHandler(*this);
 }
 
 ActiveTcpListener::ActiveTcpListener(Network::TcpConnectionHandler& parent,
                                      Network::ListenerPtr&& listener,
                                      Network::ListenerConfig& config)
-    : ActiveStreamListenerBase(parent, parent.dispatcher(), std::move(listener), config),
+    : OwnedActiveStreamListenerBase(parent, parent.dispatcher(), std::move(listener), config),
       tcp_conn_handler_(parent) {
   config.connectionBalancer().registerHandler(*this);
 }
@@ -62,48 +61,12 @@ ActiveTcpListener::~ActiveTcpListener() {
                                                      config_->name(), numConnections()));
 }
 
-void ActiveTcpListener::removeConnection(ActiveTcpConnection& connection) {
-  ENVOY_CONN_LOG(debug, "adding to cleanup list", *connection.connection_);
-  ActiveConnections& active_connections = connection.active_connections_;
-  auto removed = connection.removeFromList(active_connections.connections_);
-  dispatcher().deferredDelete(std::move(removed));
-  // Delete map entry only iff connections becomes empty.
-  if (active_connections.connections_.empty()) {
-    auto iter = connections_by_context_.find(&active_connections.filter_chain_);
-    ASSERT(iter != connections_by_context_.end());
-    // To cover the lifetime of every single connection, Connections need to be deferred deleted
-    // because the previously contained connection is deferred deleted.
-    dispatcher().deferredDelete(std::move(iter->second));
-    // The erase will break the iteration over the connections_by_context_ during the deletion.
-    if (!is_deleting_) {
-      connections_by_context_.erase(iter);
-    }
-  }
-}
-
 void ActiveTcpListener::updateListenerConfig(Network::ListenerConfig& config) {
   ENVOY_LOG(trace, "replacing listener ", config_->listenerTag(), " by ", config.listenerTag());
   ASSERT(&config_->connectionBalancer() == &config.connectionBalancer());
   config_ = &config;
 }
 
-void ActiveTcpListener::removeFilterChain(const Network::FilterChain* filter_chain) {
-  auto iter = connections_by_context_.find(filter_chain);
-  if (iter == connections_by_context_.end()) {
-    // It is possible when listener is stopping.
-  } else {
-    auto& connections = iter->second->connections_;
-    while (!connections.empty()) {
-      connections.front()->connection_->close(Network::ConnectionCloseType::NoFlush);
-    }
-    // Since is_deleting_ is on, we need to manually remove the map value and drive the
-    // iterator. Defer delete connection container to avoid race condition in destroying
-    // connection.
-    dispatcher().deferredDelete(std::move(iter->second));
-    connections_by_context_.erase(iter);
-  }
-}
-
 void ActiveTcpListener::onAccept(Network::ConnectionSocketPtr&& socket) {
   if (listenerConnectionLimitReached()) {
     RELEASE_ASSERT(socket->addressProvider().remoteAddress() != nullptr, "");
@@ -179,15 +142,6 @@ void ActiveTcpListener::newActiveConnection(const Network::FilterChain& filter_c
   }
 }
 
-ActiveConnections&
-ActiveTcpListener::getOrCreateActiveConnections(const Network::FilterChain& filter_chain) {
-  ActiveConnectionCollectionPtr& connections = connections_by_context_[&filter_chain];
-  if (connections == nullptr) {
-    connections = std::make_unique<ActiveConnections>(*this, filter_chain);
-  }
-  return *connections;
-}
-
 void ActiveTcpListener::post(Network::ConnectionSocketPtr&& socket) {
   // It is not possible to capture a unique_ptr because the post() API copies the lambda, so we must
   // bundle the socket inside a shared_ptr that can be captured.

source/server/active_tcp_listener.h

@@ -1,7 +1,6 @@
 #pragma once
 
 #include "envoy/event/dispatcher.h"
-#include "envoy/stats/timespan.h"
 #include "envoy/stream_info/stream_info.h"
 
 #include "source/common/common/linked_object.h"
@@ -23,7 +22,7 @@ using RebalancedSocketSharedPtr = std::shared_ptr<RebalancedSocket>;
  * Wrapper for an active tcp listener owned by this handler.
  */
 class ActiveTcpListener final : public Network::TcpListenerCallbacks,
-                                public ActiveStreamListenerBase,
+                                public OwnedActiveStreamListenerBase,
                                 public Network::BalancedConnectionHandler {
 public:
   ActiveTcpListener(Network::TcpConnectionHandler& parent, Network::ListenerConfig& config,
@@ -71,28 +70,12 @@ class ActiveTcpListener final : public Network::TcpListenerCallbacks,
                            Network::ServerConnectionPtr server_conn_ptr,
                            std::unique_ptr<StreamInfo::StreamInfo> stream_info) override;
 
-  /**
-   * Return the active connections container attached with the given filter chain.
-   */
-  ActiveConnections& getOrCreateActiveConnections(const Network::FilterChain& filter_chain);
-
   /**
    * Update the listener config. The follow up connections will see the new config. The existing
    * connections are not impacted.
    */
   void updateListenerConfig(Network::ListenerConfig& config);
 
-  void removeFilterChain(const Network::FilterChain* filter_chain) override;
-
-  /**
-   * Remove and destroy an active connection.
-   * @param connection supplies the connection to remove.
-   */
-  void removeConnection(ActiveTcpConnection& connection);
-
-  absl::flat_hash_map<const Network::FilterChain*, std::unique_ptr<ActiveConnections>>
-      connections_by_context_;
-
   Network::TcpConnectionHandler& tcp_conn_handler_;
   // The number of connections currently active on this listener. This is typically used for
   // connection balancing across per-handler listeners.