cluster: destroy on main thread

include/envoy/event/dispatcher.h

@@ -260,6 +260,12 @@ class Dispatcher : public DispatcherBase {
    */
   virtual void post(PostCb callback) PURE;
 
+  /**
+   * Similar to `post()` but will destroy passed callback. This simulates posting move only
+   * function.
+   */
+  virtual void movePost(PostCb&& callback) PURE;
+
   /**
    * Runs the event loop. This will not return until exit() is called either from within a callback
    * or from a different thread.

source/common/event/dispatcher_impl.cc

@@ -265,9 +265,23 @@ void DispatcherImpl::post(std::function<void()> callback) {
   }
 }
 
+void DispatcherImpl::movePost(std::function<void()>&& callback) {
+  bool do_post;
+  {
+    Thread::LockGuard lock(post_lock_);
+    do_post = post_callbacks_.empty();
+    post_callbacks_.push_back(std::move(callback));
+    // Poor man's move only function.
+    callback = nullptr;
+  }
+
+  if (do_post) {
+    post_cb_->scheduleCallbackCurrentIteration();
+  }
+}
+
 void DispatcherImpl::run(RunType type) {
   run_tid_ = api_.threadFactory().currentThreadId();
-
   // Flush all post callbacks before we run the event loop. We do this because there are post
   // callbacks that have to get run before the initial event loop starts running. libevent does
   // not guarantee that events are run in any particular order. So even if we post() and call

source/common/event/dispatcher_impl.h

@@ -86,6 +86,7 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
   void exit() override;
   SignalEventPtr listenForSignal(signal_t signal_num, SignalCb cb) override;
   void post(std::function<void()> callback) override;
+  void movePost(std::function<void()>&& callback) override;
   void run(RunType type) override;
   Buffer::WatermarkFactory& getWatermarkFactory() override { return *buffer_factory_; }
   void pushTrackedObject(const ScopeTrackedObject* object) override;

source/common/upstream/upstream_impl.cc

@@ -919,9 +919,19 @@ ClusterImplBase::ClusterImplBase(
 
   auto socket_matcher = std::make_unique<TransportSocketMatcherImpl>(
       cluster.transport_socket_matches(), factory_context, socket_factory, *stats_scope);
-  info_ = std::make_unique<ClusterInfoImpl>(cluster, factory_context.clusterManager().bindConfig(),
-                                            runtime, std::move(socket_matcher),
-                                            std::move(stats_scope), added_via_api, factory_context);
+  auto& dispatcher = factory_context.dispatcher();
+  info_ = std::shared_ptr<const ClusterInfoImpl>(
+      new ClusterInfoImpl(cluster, factory_context.clusterManager().bindConfig(), runtime,
+                          std::move(socket_matcher), std::move(stats_scope), added_via_api,
+                          factory_context),
+      [&dispatcher](const ClusterInfoImpl* self) {
+        const auto name = self->name();
+        ENVOY_LOG(debug, "Schedule destroy cluster info {}", self->name());
+        dispatcher.movePost([raii_cluster = std::shared_ptr<const ClusterInfoImpl>(self)]() {
+          ENVOY_LOG(debug, "Destroying cluster info {}. This thread should be master thread.",
+                    raii_cluster->name());
+        });
+      });
 
   if ((info_->features() & ClusterInfoImpl::Features::USE_ALPN) &&
       !raw_factory_pointer->supportsAlpn()) {

source/extensions/transport_sockets/tls/context_manager_impl.cc

@@ -17,7 +17,8 @@ namespace Tls {
 
 ContextManagerImpl::~ContextManagerImpl() {
   removeEmptyContexts();
-  KNOWN_ISSUE_ASSERT(contexts_.empty(), "https://github.com/envoyproxy/envoy/issues/10030");
+  // Contexts could be referenced by cluster to be destroyed. The cluster destroy is driven by the
+  // master thread dispatcher, but the dispatcher ceases running for good.
 }
 
 void ContextManagerImpl::removeEmptyContexts() {

test/common/upstream/cluster_manager_impl_test.cc

@@ -874,8 +874,10 @@ TEST_F(ClusterManagerImplTest, HttpHealthChecker) {
               createClientConnection_(
                   PointeesEq(Network::Utility::resolveUrl("tcp://127.0.0.1:11001")), _, _, _))
       .WillOnce(Return(connection));
+  EXPECT_CALL(factory_.dispatcher_, movePost(_));
   create(parseBootstrapFromV3Yaml(yaml));
   factory_.tls_.shutdownThread();
+  factory_.dispatcher_.to_delete_.clear();
 }
 
 TEST_F(ClusterManagerImplTest, UnknownCluster) {

test/common/upstream/logical_dns_cluster_test.cc

@@ -204,11 +204,11 @@ class LogicalDnsClusterTest : public Event::TestUsingSimulatedTime, public testi
   Network::DnsResolver::ResolveCb dns_callback_;
   NiceMock<ThreadLocal::MockInstance> tls_;
   Event::MockTimer* resolve_timer_;
-  std::shared_ptr<LogicalDnsCluster> cluster_;
   ReadyWatcher membership_updated_;
   ReadyWatcher initialized_;
   NiceMock<Runtime::MockLoader> runtime_;
   NiceMock<Event::MockDispatcher> dispatcher_;
+  std::shared_ptr<LogicalDnsCluster> cluster_;
   NiceMock<LocalInfo::MockLocalInfo> local_info_;
   NiceMock<Server::MockAdmin> admin_;
   Singleton::ManagerImpl singleton_manager_{Thread::threadFactoryForTest()};

test/common/upstream/original_dst_cluster_test.cc

@@ -94,11 +94,11 @@ class OriginalDstClusterTest : public Event::TestUsingSimulatedTime, public test
 
   Stats::TestUtil::TestStore stats_store_;
   Ssl::MockContextManager ssl_context_manager_;
+  NiceMock<Event::MockDispatcher> dispatcher_;
   OriginalDstClusterSharedPtr cluster_;
   ReadyWatcher membership_updated_;
   ReadyWatcher initialized_;
   NiceMock<Runtime::MockLoader> runtime_;
-  NiceMock<Event::MockDispatcher> dispatcher_;
   Event::MockTimer* cleanup_timer_;
   NiceMock<Random::MockRandomGenerator> random_;
   NiceMock<LocalInfo::MockLocalInfo> local_info_;

test/mocks/event/mocks.h

@@ -146,6 +146,7 @@ class MockDispatcher : public Dispatcher {
   MOCK_METHOD(void, exit, ());
   MOCK_METHOD(SignalEvent*, listenForSignal_, (signal_t signal_num, SignalCb cb));
   MOCK_METHOD(void, post, (std::function<void()> callback));
+  MOCK_METHOD(void, movePost, (std::function<void()> && callback));
   MOCK_METHOD(void, run, (RunType type));
   MOCK_METHOD(void, pushTrackedObject, (const ScopeTrackedObject* object));
   MOCK_METHOD(void, popTrackedObject, (const ScopeTrackedObject* expected_object));

test/mocks/event/wrapped_dispatcher.h

@@ -101,6 +101,8 @@ class WrappedDispatcher : public Dispatcher {
 
   void post(std::function<void()> callback) override { impl_.post(std::move(callback)); }
 
+  void movePost(std::function<void()>&& callback) override { impl_.movePost(std::move(callback)); }
+
   void run(RunType type) override { impl_.run(type); }
 
   Buffer::WatermarkFactory& getWatermarkFactory() override { return impl_.getWatermarkFactory(); }