Repository: envoyproxy/envoy
File: source/common/upstream/cluster_manager_impl.h (C/C++)

Code before:
  /**
   * An implementation of an on-demand CDS handle. It forwards the discovery request to the cluster
   * manager that created the handle.
   *
   * It's a protected type, so unit tests can use it.
   */
  class OdCdsApiHandleImpl : public OdCdsApiHandle {
  public:
    static OdCdsApiHandlePtr create(ClusterManagerImpl& parent, OdCdsApiSharedPtr odcds) {
      return std::make_unique<OdCdsApiHandleImpl>(parent, std::move(odcds));
    }

    OdCdsApiHandleImpl(ClusterManagerImpl& parent, OdCdsApiSharedPtr odcds)
        : parent_(parent), odcds_(std::move(odcds)) {
      ASSERT(odcds_ != nullptr);
    }

    ~OdCdsApiHandleImpl() override {
      // The OdCdsApiSharedPtr contains a subscription that must be destroyed on the main thread.
      // If we're being destroyed on a worker thread, we need to dispatch the destruction to the
      // main thread dispatcher. Otherwise, the subscription cleanup will trigger an assertion
      // failure or SIGABRT.
      if (!Thread::MainThread::isMainThread() && odcds_ != nullptr) {
        // Move the shared_ptr into a lambda that will be executed on the main thread.
        // This ensures the subscription is destroyed on the main thread where it was created.
        parent_.dispatcher_.post([odcds = std::move(odcds_)]() {
          // The shared_ptr will be destroyed here on the main thread, ensuring proper cleanup
          // of the subscription.
        });
      }
    }

    ClusterDiscoveryCallbackHandlePtr
    requestOnDemandClusterDiscovery(absl::string_view name, ClusterDiscoveryCallbackPtr callback,
                                    std::chrono::milliseconds timeout) override {
      return parent_.requestOnDemandClusterDiscovery(odcds_, std::string(name), std::move(callback),
                                                     timeout);
    }

  private:
    ClusterManagerImpl& parent_;
    OdCdsApiSharedPtr odcds_;
  };

  virtual void postThreadLocalClusterUpdate(ClusterManagerCluster& cm_cluster,
                                            ThreadLocalClusterUpdateParams&& params);

  /**
   * Notifies cluster discovery managers in each worker thread that the discovery process for the
   * cluster with a passed name has timed out.
   *

Review comment (suggestion):
While this pattern is used in very cross-thread specific code, it should not leak to the rest of the code base, if possible.

Code after:

  /**
   * An implementation of an on-demand CDS handle. It forwards the discovery request to the cluster
   * manager that created the handle.
   *
   * It's a protected type, so unit tests can use it.
   */
  class OdCdsApiHandleImpl : public OdCdsApiHandle {
  public:
    static OdCdsApiHandlePtr create(ClusterManagerImpl& parent, uint64_t config_source_key) {
      return std::make_unique<OdCdsApiHandleImpl>(parent, config_source_key);
    }

    OdCdsApiHandleImpl(ClusterManagerImpl& parent, uint64_t config_source_key)
        : parent_(parent), config_source_key_(config_source_key) {}

    ClusterDiscoveryCallbackHandlePtr
    requestOnDemandClusterDiscovery(absl::string_view name, ClusterDiscoveryCallbackPtr callback,
                                    std::chrono::milliseconds timeout) override {
      return parent_.requestOnDemandClusterDiscovery(config_source_key_, std::string(name),
                                                     std::move(callback), timeout);
    }

  private:
    ClusterManagerImpl& parent_;
    uint64_t config_source_key_;
  };

  virtual void postThreadLocalClusterUpdate(ClusterManagerCluster& cm_cluster,
                                            ThreadLocalClusterUpdateParams&& params);

  /**
   * Notifies cluster discovery managers in each worker thread that the discovery process for the
   * cluster with a passed name has timed out.
   *
   * It's protected, so the tests can use it.
   */
  void notifyExpiredDiscovery(absl::string_view name);

  /**
   * Creates a new discovery manager in current thread and swaps it with the one in thread local
   * cluster manager. This could be used to simulate requesting a cluster from a different
   * thread. Used for tests only.
   *
   * Protected, so tests can use it.
   *
   * @return the previous cluster discovery manager.
   */
  ClusterDiscoveryManager createAndSwapClusterDiscoveryManager(std::string thread_name);

private: