grpc · ctiller · May 29, 2024 · May 29, 2024 · May 29, 2024 · May 29, 2024
diff --git a/src/core/BUILD b/src/core/BUILD
@@ -685,6 +685,8 @@ grpc_cc_library(
         "lib/promise/cancel_callback.h",
     ],
     deps = [
+        "arena",
+        "context",
         "promise_like",
         "//:gpr_platform",
     ],
@@ -3593,7 +3595,6 @@ grpc_cc_library(
     hdrs = ["load_balancing/subchannel_interface.h"],
     external_deps = ["absl/status"],
     deps = [
-        "call_destination",
         "dual_ref_counted",
         "iomgr_fwd",
         "//:event_engine_base_hdrs",

diff --git a/src/core/client_channel/client_channel.cc b/src/core/client_channel/client_channel.cc
@@ -79,6 +79,7 @@
 #include "src/core/lib/promise/context.h"
 #include "src/core/lib/promise/exec_ctx_wakeup_scheduler.h"
 #include "src/core/lib/promise/latch.h"
+#include "src/core/lib/promise/loop.h"
 #include "src/core/lib/promise/map.h"
 #include "src/core/lib/promise/pipe.h"
 #include "src/core/lib/promise/poll.h"
@@ -122,6 +123,73 @@ extern TraceFlag grpc_client_channel_trace;
 extern TraceFlag grpc_client_channel_call_trace;
 extern TraceFlag grpc_client_channel_lb_call_trace;
 
+//
+// ClientChannel::SubchannelWrapper
+//
+
+class ClientChannel::SubchannelWrapper
+    : public SubchannelInterfaceWithCallDestination {
+ public:
+  SubchannelWrapper(RefCountedPtr<ClientChannel> client_channel,
+                    RefCountedPtr<Subchannel> subchannel);
+  ~SubchannelWrapper() override;
+
+  void Orphaned() override;
+  void WatchConnectivityState(
+      std::unique_ptr<ConnectivityStateWatcherInterface> watcher) override
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(*client_channel_->work_serializer_);
+  void CancelConnectivityStateWatch(
+      ConnectivityStateWatcherInterface* watcher) override
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(*client_channel_->work_serializer_);
+
+  RefCountedPtr<UnstartedCallDestination> call_destination() override {
+    return subchannel_->call_destination();
+  }
+
+  void RequestConnection() override { subchannel_->RequestConnection(); }
+
+  void ResetBackoff() override { subchannel_->ResetBackoff(); }
+
+  void AddDataWatcher(std::unique_ptr<DataWatcherInterface> watcher) override
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(*client_channel_->work_serializer_);
+  void CancelDataWatcher(DataWatcherInterface* watcher) override
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(*client_channel_->work_serializer_);
+  void ThrottleKeepaliveTime(int new_keepalive_time);
+
+ private:
+  class WatcherWrapper;
+
+  // A heterogenous lookup comparator for data watchers that allows
+  // unique_ptr keys to be looked up as raw pointers.
+  struct DataWatcherLessThan {
+    using is_transparent = void;
+    bool operator()(const std::unique_ptr<DataWatcherInterface>& p1,
+                    const std::unique_ptr<DataWatcherInterface>& p2) const {
+      return p1 < p2;
+    }
+    bool operator()(const std::unique_ptr<DataWatcherInterface>& p1,
+                    const DataWatcherInterface* p2) const {
+      return p1.get() < p2;
+    }
+    bool operator()(const DataWatcherInterface* p1,
+                    const std::unique_ptr<DataWatcherInterface>& p2) const {
+      return p1 < p2.get();
+    }
+  };
+
+  RefCountedPtr<ClientChannel> client_channel_;
+  RefCountedPtr<Subchannel> subchannel_;
+  // Maps from the address of the watcher passed to us by the LB policy
+  // to the address of the WrapperWatcher that we passed to the underlying
+  // subchannel.  This is needed so that when the LB policy calls
+  // CancelConnectivityStateWatch() with its watcher, we know the
+  // corresponding WrapperWatcher to cancel on the underlying subchannel.
+  std::map<ConnectivityStateWatcherInterface*, WatcherWrapper*> watcher_map_
+      ABSL_GUARDED_BY(*client_channel_->work_serializer_);
+  std::set<std::unique_ptr<DataWatcherInterface>, DataWatcherLessThan>
+      data_watchers_ ABSL_GUARDED_BY(*client_channel_->work_serializer_);
+};
+
 //
 // ClientChannel::ResolverResultHandler
 //
@@ -581,11 +649,6 @@ ClientChannel::ClientChannel(
       default_authority_(
           GetDefaultAuthorityFromChannelArgs(channel_args_, this->target())),
       channelz_node_(channel_args_.GetObject<channelz::ChannelNode>()),
-      call_arena_allocator_(MakeRefCounted<CallArenaAllocator>(
-          channel_args_.GetObject<ResourceQuota>()
-              ->memory_quota()
-              ->CreateMemoryAllocator("client_channel"),
-          1024)),
       idle_timeout_(GetClientIdleTimeout(channel_args_)),
       resolver_data_for_calls_(ResolverDataForCalls{}),
       picker_(nullptr),
@@ -655,7 +718,7 @@ grpc_connectivity_state ClientChannel::CheckConnectivityState(
 
 void ClientChannel::WatchConnectivityState(grpc_connectivity_state, Timestamp,
                                            grpc_completion_queue*, void*) {
-  // FIXME: implement
+  // TODO(ctiller): implement
   Crash("not implemented");
 }
 
@@ -730,6 +793,7 @@ class PingRequest {
 }  // namespace
 
 void ClientChannel::Ping(grpc_completion_queue*, void*) {
+  // TODO(ctiller): implement
   Crash("not implemented");
 }
 
@@ -753,7 +817,7 @@ CallInitiator ClientChannel::CreateCall(
   // Create an initiator/unstarted-handler pair.
   auto call =
       MakeCallPair(std::move(client_initial_metadata), event_engine_.get(),
-                   call_arena_allocator_->MakeArena(), nullptr);
+                   call_arena_allocator()->MakeArena(), nullptr);
   // Spawn a promise to wait for the resolver result.
   // This will eventually start the call.
   call.initiator.SpawnGuardedUntilCallCompletes(
@@ -1179,7 +1243,6 @@ void ClientChannel::UpdateServiceConfigInDataPlaneLocked() {
         MakeRefCounted<DefaultConfigSelector>(saved_service_config_);
   }
   // Construct filter stack.
-  // TODO(roth): Add service_config to channel_args_.
   InterceptionChainBuilder builder(channel_args_.SetObject(this));
   if (idle_timeout_ != Duration::Zero()) {
     builder.AddOnServerTrailingMetadata([this](ServerMetadata&) {
@@ -1205,14 +1268,14 @@ void ClientChannel::UpdateServiceConfigInDataPlaneLocked() {
   if (enable_retries) {
     Crash("call v3 stack does not yet support retries");
   }
-  auto filter_stack = builder.Build(call_destination_);
+  auto top_of_stack_call_destination = builder.Build(call_destination_);
   // Send result to data plane.
-  if (!filter_stack.ok()) {
+  if (!top_of_stack_call_destination.ok()) {
     resolver_data_for_calls_.Set(MaybeRewriteIllegalStatusCode(
-        filter_stack.status(), "channel construction"));
+        top_of_stack_call_destination.status(), "channel construction"));
   } else {
     resolver_data_for_calls_.Set(ResolverDataForCalls{
-        std::move(config_selector), std::move(*filter_stack)});
+        std::move(config_selector), std::move(*top_of_stack_call_destination)});
   }
 }
 

diff --git a/src/core/client_channel/client_channel.h b/src/core/client_channel/client_channel.h
@@ -28,24 +28,15 @@
 #include "src/core/client_channel/subchannel.h"
 #include "src/core/ext/filters/channel_idle/idle_filter_state.h"
 #include "src/core/lib/gprpp/single_set_ptr.h"
-#include "src/core/lib/promise/loop.h"
 #include "src/core/lib/promise/observable.h"
 #include "src/core/lib/surface/channel.h"
-#include "src/core/lib/transport/call_filters.h"
 #include "src/core/lib/transport/metadata.h"
 #include "src/core/load_balancing/lb_policy.h"
 #include "src/core/resolver/resolver.h"
 #include "src/core/service_config/service_config.h"
 
 namespace grpc_core {
 
-class SubchannelInterfaceWithCallDestination : public SubchannelInterface {
- public:
-  using SubchannelInterface::SubchannelInterface;
-  // Obtain the call destination for this subchannel.
-  virtual RefCountedPtr<UnstartedCallDestination> call_destination() = 0;
-};
-
 class ClientChannel : public Channel {
  public:
   // This class is a wrapper for Subchannel that hides details of the
@@ -56,67 +47,7 @@ class ClientChannel : public Channel {
   // underlying subchannel is shared between channels, this wrapper will only
   // be used within one channel, so it will always be synchronized by the
   // control plane work_serializer.
-  class SubchannelWrapper : public SubchannelInterfaceWithCallDestination {
-   public:
-    SubchannelWrapper(RefCountedPtr<ClientChannel> client_channel,
-                      RefCountedPtr<Subchannel> subchannel);
-    ~SubchannelWrapper() override;
-
-    void Orphaned() override;
-    void WatchConnectivityState(
-        std::unique_ptr<ConnectivityStateWatcherInterface> watcher) override
-        ABSL_EXCLUSIVE_LOCKS_REQUIRED(*client_channel_->work_serializer_);
-    void CancelConnectivityStateWatch(
-        ConnectivityStateWatcherInterface* watcher) override
-        ABSL_EXCLUSIVE_LOCKS_REQUIRED(*client_channel_->work_serializer_);
-
-    RefCountedPtr<UnstartedCallDestination> call_destination() override {
-      return subchannel_->connected_subchannel()->unstarted_call_destination();
-    }
-
-    void RequestConnection() override { subchannel_->RequestConnection(); }
-
-    void ResetBackoff() override { subchannel_->ResetBackoff(); }
-
-    void AddDataWatcher(std::unique_ptr<DataWatcherInterface> watcher) override
-        ABSL_EXCLUSIVE_LOCKS_REQUIRED(*client_channel_->work_serializer_);
-    void CancelDataWatcher(DataWatcherInterface* watcher) override
-        ABSL_EXCLUSIVE_LOCKS_REQUIRED(*client_channel_->work_serializer_);
-    void ThrottleKeepaliveTime(int new_keepalive_time);
-
-   private:
-    class WatcherWrapper;
-
-    // A heterogenous lookup comparator for data watchers that allows
-    // unique_ptr keys to be looked up as raw pointers.
-    struct DataWatcherLessThan {
-      using is_transparent = void;
-      bool operator()(const std::unique_ptr<DataWatcherInterface>& p1,
-                      const std::unique_ptr<DataWatcherInterface>& p2) const {
-        return p1 < p2;
-      }
-      bool operator()(const std::unique_ptr<DataWatcherInterface>& p1,
-                      const DataWatcherInterface* p2) const {
-        return p1.get() < p2;
-      }
-      bool operator()(const DataWatcherInterface* p1,
-                      const std::unique_ptr<DataWatcherInterface>& p2) const {
-        return p1 < p2.get();
-      }
-    };
-
-    RefCountedPtr<ClientChannel> client_channel_;
-    RefCountedPtr<Subchannel> subchannel_;
-    // Maps from the address of the watcher passed to us by the LB policy
-    // to the address of the WrapperWatcher that we passed to the underlying
-    // subchannel.  This is needed so that when the LB policy calls
-    // CancelConnectivityStateWatch() with its watcher, we know the
-    // corresponding WrapperWatcher to cancel on the underlying subchannel.
-    std::map<ConnectivityStateWatcherInterface*, WatcherWrapper*> watcher_map_
-        ABSL_GUARDED_BY(*client_channel_->work_serializer_);
-    std::set<std::unique_ptr<DataWatcherInterface>, DataWatcherLessThan>
-        data_watchers_ ABSL_GUARDED_BY(*client_channel_->work_serializer_);
-  };
+  class SubchannelWrapper;
 
   using PickerObservable =
       Observable<RefCountedPtr<LoadBalancingPolicy::SubchannelPicker>>;
@@ -257,7 +188,6 @@ class ClientChannel : public Channel {
   ClientChannelFactory* const client_channel_factory_;
   const std::string default_authority_;
   channelz::ChannelNode* const channelz_node_;
-  const RefCountedPtr<CallArenaAllocator> call_arena_allocator_;
   GlobalStatsPluginRegistry::StatsPluginGroup stats_plugin_group_;
 
   //

diff --git a/src/core/client_channel/client_channel_filter.cc b/src/core/client_channel/client_channel_filter.cc
@@ -3454,9 +3454,8 @@ void ClientChannelFilter::FilterBasedLoadBalancedCall::CreateSubchannelCall() {
   Slice* path = send_initial_metadata()->get_pointer(HttpPathMetadata());
   CHECK_NE(path, nullptr);
   SubchannelCall::Args call_args = {
-      connected_subchannel()->RefAsSubclass<ConnectedSubchannel>(), pollent_,
-      path->Ref(), /*start_time=*/0, arena_->GetContext<Call>()->deadline(),
-      arena_,
+      connected_subchannel()->Ref(), pollent_, path->Ref(), /*start_time=*/0,
+      arena_->GetContext<Call>()->deadline(), arena_,
       // TODO(roth): When we implement hedging support, we will probably
       // need to use a separate call context for each subchannel call.
       call_context(), call_combiner_};

diff --git a/src/core/client_channel/client_channel_internal.h b/src/core/client_channel/client_channel_internal.h
@@ -29,6 +29,7 @@
 #include "src/core/lib/channel/context.h"
 #include "src/core/lib/gprpp/unique_type_name.h"
 #include "src/core/lib/resource_quota/arena.h"
+#include "src/core/lib/transport/call_destination.h"
 #include "src/core/load_balancing/lb_policy.h"
 #include "src/core/service_config/service_config_call_data.h"
 #include "src/core/telemetry/call_tracer.h"
@@ -79,6 +80,13 @@ struct ContextSubclass<ClientChannelServiceConfigCallData> {
   using Base = ServiceConfigCallData;
 };
 
+class SubchannelInterfaceWithCallDestination : public SubchannelInterface {
+ public:
+  using SubchannelInterface::SubchannelInterface;
+  // Obtain the call destination for this subchannel.
+  virtual RefCountedPtr<UnstartedCallDestination> call_destination() = 0;
+};
+
 }  // namespace grpc_core
 
 #endif  // GRPC_SRC_CORE_CLIENT_CHANNEL_CLIENT_CHANNEL_INTERNAL_H
diff --git a/src/core/client_channel/load_balanced_call_destination.cc b/src/core/client_channel/load_balanced_call_destination.cc
@@ -18,6 +18,7 @@
 #include "src/core/client_channel/client_channel_internal.h"
 #include "src/core/client_channel/subchannel.h"
 #include "src/core/lib/channel/status_util.h"
+#include "src/core/lib/promise/loop.h"
 #include "src/core/telemetry/call_tracer.h"
 
 namespace grpc_core {
@@ -157,7 +158,7 @@ T HandlePickResult(
 // Does an LB pick for a call.  Returns one of the following things:
 // - Continue{}, meaning to queue the pick
 // - a non-OK status, meaning to fail the call
-// - a connected subchannel, meaning that the pick is complete
+// - a call destination, meaning that the pick is complete
 // When the pick is complete, pushes client_initial_metadata onto
 // call_initiator.  Also adds the subchannel call tracker (if any) to
 // context.
@@ -190,13 +191,13 @@ LoopCtl<absl::StatusOr<RefCountedPtr<UnstartedCallDestination>>> PickSubchannel(
                   complete_pick->subchannel.get());
         }
         CHECK(complete_pick->subchannel != nullptr);
-        // Grab a ref to the connected subchannel while we're still
+        // Grab a ref to the call destination while we're still
         // holding the data plane mutex.
         auto call_destination =
             DownCast<SubchannelInterfaceWithCallDestination*>(
                 complete_pick->subchannel.get())
                 ->call_destination();
-        // If the subchannel has no connected subchannel (e.g., if the
+        // If the subchannel has no call destination (e.g., if the
         // subchannel has moved out of state READY but the LB policy hasn't
         // yet seen that change and given us a new picker), then just
         // queue the pick.  We'll try again as soon as we get a new picker.
@@ -302,10 +303,10 @@ void LoadBalancedCallDestination::StartCall(
                     absl::StatusOr<RefCountedPtr<UnstartedCallDestination>>,
                     bool>
                     pick_result) {
-              auto& connected_subchannel = std::get<0>(pick_result);
+              auto& call_destination = std::get<0>(pick_result);
               const bool was_queued = std::get<1>(pick_result);
-              if (!connected_subchannel.ok()) {
-                return connected_subchannel.status();
+              if (!call_destination.ok()) {
+                return call_destination.status();
               }
               // LB pick is done, so indicate that we've committed.
               auto* on_commit = MaybeGetContext<LbOnCommit>();
@@ -323,7 +324,7 @@ void LoadBalancedCallDestination::StartCall(
               // Delegate to connected subchannel.
               // FIXME: need to insert LbCallTracingFilter at the top of the
               // stack
-              (*connected_subchannel)->StartCall(unstarted_handler);
+              (*call_destination)->StartCall(unstarted_handler);
               return absl::OkStatus();
             });
       });

diff --git a/src/core/client_channel/subchannel.h b/src/core/client_channel/subchannel.h
@@ -248,6 +248,12 @@ class Subchannel final : public DualRefCounted<Subchannel> {
     return connected_subchannel_;
   }
 
+  RefCountedPtr<UnstartedCallDestination> call_destination() {
+    MutexLock lock(&mu_);
+    if (connected_subchannel_ == nullptr) return nullptr;
+    return connected_subchannel_->unstarted_call_destination();
+  }
+
   // Attempt to connect to the backend.  Has no effect if already connected.
   void RequestConnection() ABSL_LOCKS_EXCLUDED(mu_);
 

diff --git a/src/core/ext/filters/http/message_compress/compression_filter.cc b/src/core/ext/filters/http/message_compress/compression_filter.cc
@@ -119,7 +119,7 @@ MessageHandle ChannelCompression::CompressMessage(
     gpr_log(GPR_INFO, "CompressMessage: len=%" PRIdPTR " alg=%d flags=%d",
             message->payload()->Length(), algorithm, message->flags());
   }
-  auto* call_tracer = GetContext<CallTracerInterface>();
+  auto* call_tracer = MaybeGetContext<CallTracerInterface>();
   if (call_tracer != nullptr) {
     call_tracer->RecordSendMessage(*message->payload());
   }
@@ -176,7 +176,7 @@ absl::StatusOr<MessageHandle> ChannelCompression::DecompressMessage(
             message->payload()->Length(),
             args.max_recv_message_length.value_or(-1), args.algorithm);
   }
-  auto* call_tracer = GetContext<CallTracerInterface>();
+  auto* call_tracer = MaybeGetContext<CallTracerInterface>();
   if (call_tracer != nullptr) {
     call_tracer->RecordReceivedMessage(*message->payload());
   }