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

Code before:
  };

  using ClusterInitializationObjectConstSharedPtr =
      std::shared_ptr<const ClusterInitializationObject>;
  using ClusterInitializationMap =
      absl::flat_hash_map<std::string, ClusterInitializationObjectConstSharedPtr>;

  /**
   * 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_;
  };


Review comment (suggestion):
It seems that the underlying issue is that the subscription should not be even accessed directly by the worker thread.
Would it make sense to separate the handle and the subscription s.t. the subscription lives only in the main thread and it will handle its lifecycle? The separation will keep the ownership cleaner, and will avoid other races, as the worker thread shouldn't need to access the subscription directly.

Code after:
    const std::string drop_category_;
  };

  using ClusterInitializationObjectConstSharedPtr =
      std::shared_ptr<const ClusterInitializationObject>;
  using ClusterInitializationMap =
      absl::flat_hash_map<std::string, ClusterInitializationObjectConstSharedPtr>;

  /**
   * 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.
   *