Merge remote-tracking branch 'upstream/master' into reorg_tests

src/core/lib/promise/party.cc

@@ -357,6 +357,7 @@ void Party::RunPartyAndUnref(uint64_t prev_state) {
 }
 
 void Party::AddParticipants(Participant** participants, size_t count) {
+  GRPC_LATENT_SEE_INNER_SCOPE("Party::AddParticipants");
   uint64_t state = state_.load(std::memory_order_acquire);
   uint64_t allocated;
 
@@ -400,6 +401,7 @@ void Party::AddParticipants(Participant** participants, size_t count) {
 }
 
 void Party::AddParticipant(Participant* participant) {
+  GRPC_LATENT_SEE_INNER_SCOPE("Party::AddParticipant");
   uint64_t state = state_.load(std::memory_order_acquire);
   uint64_t allocated;
   size_t slot;
@@ -468,6 +470,7 @@ void Party::WakeupAsync(WakeupMask wakeup_mask) {
         wakeup_mask_ |= wakeup_mask;
         arena_->GetContext<grpc_event_engine::experimental::EventEngine>()->Run(
             [this, prev_state]() {
+              GRPC_LATENT_SEE_PARENT_SCOPE("Party::WakeupAsync");
               ApplicationCallbackExecCtx app_exec_ctx;
               ExecCtx exec_ctx;
               RunLockedAndUnref(this, prev_state);

src/core/lib/promise/party.h

@@ -342,6 +342,7 @@ class Party : public Activity, private Wakeable {
 
   // Wakeable implementation
   void Wakeup(WakeupMask wakeup_mask) final {
+    GRPC_LATENT_SEE_INNER_SCOPE("Party::Wakeup");
     if (Activity::current() == this) {
       wakeup_mask_ |= wakeup_mask;
       Unref();
@@ -352,6 +353,7 @@ class Party : public Activity, private Wakeable {
 
   GPR_ATTRIBUTE_ALWAYS_INLINE_FUNCTION void WakeupFromState(
       uint64_t cur_state, WakeupMask wakeup_mask) {
+    GRPC_LATENT_SEE_INNER_SCOPE("Party::WakeupFromState");
     DCHECK_NE(wakeup_mask & kWakeupMask, 0u)
         << "Wakeup mask must be non-zero: " << wakeup_mask;
     while (true) {

src/core/lib/surface/client_call.cc

@@ -184,6 +184,7 @@ void ClientCall::CancelWithError(grpc_error_handle error) {
 
 template <typename Batch>
 void ClientCall::ScheduleCommittedBatch(Batch batch) {
+  GRPC_LATENT_SEE_INNER_SCOPE("ClientCall::ScheduleCommittedBatch");
   auto cur_state = call_state_.load(std::memory_order_acquire);
   while (true) {
     switch (cur_state) {
@@ -225,6 +226,7 @@ void ClientCall::ScheduleCommittedBatch(Batch batch) {
 }
 
 void ClientCall::StartCall(const grpc_op& send_initial_metadata_op) {
+  GRPC_LATENT_SEE_INNER_SCOPE("ClientCall::StartCall");
   auto cur_state = call_state_.load(std::memory_order_acquire);
   CToMetadata(send_initial_metadata_op.data.send_initial_metadata.metadata,
               send_initial_metadata_op.data.send_initial_metadata.count,
@@ -271,6 +273,7 @@ void ClientCall::StartCall(const grpc_op& send_initial_metadata_op) {
 
 void ClientCall::CommitBatch(const grpc_op* ops, size_t nops, void* notify_tag,
                              bool is_notify_tag_closure) {
+  GRPC_LATENT_SEE_INNER_SCOPE("ClientCall::CommitBatch");
   if (nops == 1 && ops[0].op == GRPC_OP_SEND_INITIAL_METADATA) {
     StartCall(ops[0]);
     EndOpImmediately(cq_, notify_tag, is_notify_tag_closure);

src/core/lib/transport/metadata_batch.cc

@@ -83,6 +83,7 @@ bool DebugStringBuilder::IsAllowListed(const absl::string_view key) const {
         allow_list.insert(std::string(LbTokenMetadata::key()));
         allow_list.insert(std::string(TeMetadata::key()));
         allow_list.insert(std::string(UserAgentMetadata::key()));
+        allow_list.insert(std::string(W3CTraceParentMetadata::key()));
         allow_list.insert(std::string(XEnvoyPeerMetadata::key()));
         // go/keep-sorted end
         // go/keep-sorted start

src/core/lib/transport/metadata_batch.h

@@ -468,6 +468,14 @@ struct LbCostBinMetadata {
                                   MetadataParseErrorFn on_error);
 };
 
+// traceparent metadata
+struct W3CTraceParentMetadata : public SimpleSliceBasedMetadata {
+  static constexpr bool kRepeatable = false;
+  static constexpr bool kTransferOnTrailersOnly = false;
+  using CompressionTraits = FrequentKeyWithNoValueCompressionCompressor;
+  static absl::string_view key() { return "traceparent"; }
+};
+
 // Annotation added by a transport to note whether a failed request was never
 // placed on the wire, or never seen by a server.
 struct GrpcStreamNetworkState {
@@ -1582,7 +1590,7 @@ using grpc_metadata_batch_base = grpc_core::MetadataMap<
     grpc_core::GrpcServerStatsBinMetadata, grpc_core::GrpcTraceBinMetadata,
     grpc_core::GrpcTagsBinMetadata, grpc_core::GrpcLbClientStatsMetadata,
     grpc_core::LbCostBinMetadata, grpc_core::LbTokenMetadata,
-    grpc_core::XEnvoyPeerMetadata,
+    grpc_core::XEnvoyPeerMetadata, grpc_core::W3CTraceParentMetadata,
     // Non-encodable things
     grpc_core::GrpcStreamNetworkState, grpc_core::PeerString,
     grpc_core::GrpcStatusContext, grpc_core::GrpcStatusFromWire,

src/core/telemetry/metrics.cc

@@ -120,42 +120,45 @@ void GlobalStatsPluginRegistry::StatsPluginGroup::AddServerCallTracers(
   }
 }
 
-NoDestruct<Mutex> GlobalStatsPluginRegistry::mutex_;
-NoDestruct<std::vector<std::shared_ptr<StatsPlugin>>>
+std::atomic<GlobalStatsPluginRegistry::GlobalStatsPluginNode*>
     GlobalStatsPluginRegistry::plugins_;
 
 void GlobalStatsPluginRegistry::RegisterStatsPlugin(
     std::shared_ptr<StatsPlugin> plugin) {
-  MutexLock lock(&*mutex_);
-  plugins_->push_back(std::move(plugin));
+  GlobalStatsPluginNode* node = new GlobalStatsPluginNode();
+  node->plugin = std::move(plugin);
+  node->next = plugins_.load(std::memory_order_relaxed);
+  while (!plugins_.compare_exchange_weak(
+      node->next, node, std::memory_order_acq_rel, std::memory_order_relaxed)) {
+  }
 }
 
 GlobalStatsPluginRegistry::StatsPluginGroup
 GlobalStatsPluginRegistry::GetStatsPluginsForChannel(
     const experimental::StatsPluginChannelScope& scope) {
-  MutexLock lock(&*mutex_);
   StatsPluginGroup group;
-  for (const auto& plugin : *plugins_) {
+  for (GlobalStatsPluginNode* node = plugins_.load(std::memory_order_acquire);
+       node != nullptr; node = node->next) {
     bool is_enabled = false;
     std::shared_ptr<StatsPlugin::ScopeConfig> config;
-    std::tie(is_enabled, config) = plugin->IsEnabledForChannel(scope);
+    std::tie(is_enabled, config) = node->plugin->IsEnabledForChannel(scope);
     if (is_enabled) {
-      group.AddStatsPlugin(plugin, std::move(config));
+      group.AddStatsPlugin(node->plugin, std::move(config));
     }
   }
   return group;
 }
 
 GlobalStatsPluginRegistry::StatsPluginGroup
 GlobalStatsPluginRegistry::GetStatsPluginsForServer(const ChannelArgs& args) {
-  MutexLock lock(&*mutex_);
   StatsPluginGroup group;
-  for (const auto& plugin : *plugins_) {
+  for (GlobalStatsPluginNode* node = plugins_.load(std::memory_order_acquire);
+       node != nullptr; node = node->next) {
     bool is_enabled = false;
     std::shared_ptr<StatsPlugin::ScopeConfig> config;
-    std::tie(is_enabled, config) = plugin->IsEnabledForServer(args);
+    std::tie(is_enabled, config) = node->plugin->IsEnabledForServer(args);
     if (is_enabled) {
-      group.AddStatsPlugin(plugin, std::move(config));
+      group.AddStatsPlugin(node->plugin, std::move(config));
     }
   }
   return group;

src/core/telemetry/metrics.h

@@ -446,6 +446,8 @@ class GlobalStatsPluginRegistry {
       return false;
     }
 
+    size_t size() const { return plugins_state_.size(); }
+
     // Registers a callback to be used to populate callback metrics.
     // The callback will update the specified metrics.  The callback
     // will be invoked no more often than min_interval.  Multiple callbacks may
@@ -508,13 +510,15 @@ class GlobalStatsPluginRegistry {
   static StatsPluginGroup GetStatsPluginsForServer(const ChannelArgs& args);
 
  private:
+  struct GlobalStatsPluginNode {
+    std::shared_ptr<StatsPlugin> plugin;
+    GlobalStatsPluginNode* next = nullptr;
+  };
   friend class GlobalStatsPluginRegistryTestPeer;
 
   GlobalStatsPluginRegistry() = default;
 
-  static NoDestruct<Mutex> mutex_;
-  static NoDestruct<std::vector<std::shared_ptr<StatsPlugin>>> plugins_
-      ABSL_GUARDED_BY(mutex_);
+  static std::atomic<GlobalStatsPluginNode*> plugins_;
 };
 
 // A metric callback that is registered with a stats plugin group.

src/core/util/latent_see.h

@@ -35,6 +35,7 @@
 #include "absl/functional/function_ref.h"
 #include "absl/log/log.h"
 #include "absl/strings/string_view.h"
+#include "absl/types/optional.h"
 #include "src/core/util/per_cpu.h"
 #include "src/core/util/sync.h"
 

src/core/xds/grpc/xds_route_config.cc

@@ -199,6 +199,7 @@ std::string XdsRouteConfigResource::Route::RouteAction::ToString() const {
   if (max_stream_duration.has_value()) {
     contents.push_back(max_stream_duration->ToString());
   }
+  if (auto_host_rewrite) contents.push_back("auto_host_rewrite=true");
   return absl::StrCat("{", absl::StrJoin(contents, ", "), "}");
 }
 

test/core/end2end/tests/disappearing_server.cc

@@ -26,7 +26,6 @@
 #include "src/core/util/time.h"
 #include "test/core/end2end/end2end_tests.h"
 
-#ifndef GPR_WINDOWS  // b/148110727 for more details
 namespace grpc_core {
 
 static void OneRequestAndShutdownServer(CoreEnd2endTest& test) {
@@ -77,4 +76,3 @@ CORE_END2END_TEST(CoreClientChannelTest, DisappearingServer) {
 }
 
 }  // namespace grpc_core
-#endif  // GPR_WINDOWS

test/core/telemetry/metrics_test.cc

@@ -15,6 +15,7 @@
 #include "src/core/telemetry/metrics.h"
 
 #include <memory>
+#include <thread>
 
 #include "absl/log/log.h"
 #include "gmock/gmock.h"
@@ -648,6 +649,38 @@ TEST_F(MetricsTest, FindInstrumentByName) {
                   ::testing::Eq(uint64_counter_handle.index))));
 }
 
+TEST_F(MetricsTest, ParallelStatsPluginRegistrationAndLookup) {
+  std::vector<std::thread> register_threads;
+  std::vector<std::thread> lookup_threads;
+  register_threads.reserve(100);
+  lookup_threads.reserve(100);
+  // 100 threads that register 100 stats plugins each
+  for (int i = 0; i < 100; ++i) {
+    register_threads.emplace_back([] {
+      for (int j = 0; j < 100; ++j) {
+        FakeStatsPluginBuilder().BuildAndRegister();
+      }
+    });
+  }
+  // 100 threads that keep looking up stats plugins till they see 10000 stats
+  // plugins
+  for (int i = 0; i < 100; ++i) {
+    lookup_threads.emplace_back([this] {
+      while (GlobalStatsPluginRegistry::GetStatsPluginsForChannel(
+                 StatsPluginChannelScope("", "", endpoint_config_))
+                 .size() < 10000) {
+      };
+    });
+  }
+  for (int i = 0; i < 100; ++i) {
+    register_threads[i].join();
+    lookup_threads[i].join();
+  }
+  EXPECT_THAT(GlobalStatsPluginRegistry::GetStatsPluginsForChannel(
+                  StatsPluginChannelScope("", "", endpoint_config_)),
+              ::testing::SizeIs(10000));
+}
+
 using MetricsDeathTest = MetricsTest;
 
 TEST_F(MetricsDeathTest, RegisterTheSameMetricNameWouldCrash) {

test/core/test_util/BUILD

@@ -263,11 +263,6 @@ grpc_cc_test(
     ],
 )
 
-sh_library(
-    name = "fuzzer_one_entry_runner",
-    srcs = ["fuzzer_one_entry_runner.sh"],
-)
-
 grpc_cc_library(
     name = "stack_tracer",
     srcs = [

test/core/test_util/fake_stats_plugin.h

@@ -696,8 +696,14 @@ class GlobalInstrumentsRegistryTestPeer {
 class GlobalStatsPluginRegistryTestPeer {
  public:
   static void ResetGlobalStatsPluginRegistry() {
-    MutexLock lock(&*GlobalStatsPluginRegistry::mutex_);
-    GlobalStatsPluginRegistry::plugins_->clear();
+    GlobalStatsPluginRegistry::GlobalStatsPluginNode* node =
+        GlobalStatsPluginRegistry::plugins_.exchange(nullptr,
+                                                     std::memory_order_acq_rel);
+    while (node != nullptr) {
+      GlobalStatsPluginRegistry::GlobalStatsPluginNode* next = node->next;
+      delete node;
+      node = next;
+    }
   }
 };