Merge branch 'master' into generic-stub-service-refactor

.github/CODEOWNERS

@@ -1,5 +1,6 @@
 /bazel/** @veblush @gnossen
 /bazel/experiments.yaml
+/bazel/rollouts.yaml
 /cmake/** @veblush @apolcyn
 /src/core/client_channel/** @markdroth
 /src/core/ext/transport/chttp2/transport/** @ctiller

src/core/BUILD

@@ -4870,6 +4870,7 @@ grpc_cc_library(
         "absl/container:inlined_vector",
         "absl/functional:function_ref",
         "absl/log:check",
+        "absl/log:globals",
         "absl/log:log",
         "absl/status",
         "absl/status:statusor",

src/core/lib/event_engine/thready_event_engine/thready_event_engine.cc

@@ -22,6 +22,7 @@
 #include <grpc/support/port_platform.h>
 
 #include "src/core/lib/gprpp/crash.h"
+#include "src/core/lib/gprpp/sync.h"
 #include "src/core/lib/gprpp/thd.h"
 
 namespace grpc_event_engine {
@@ -39,20 +40,45 @@ ThreadyEventEngine::CreateListener(
     absl::AnyInvocable<void(absl::Status)> on_shutdown,
     const EndpointConfig& config,
     std::unique_ptr<MemoryAllocatorFactory> memory_allocator_factory) {
+  struct AcceptState {
+    grpc_core::Mutex mu_;
+    grpc_core::CondVar cv_;
+    int pending_accepts_ ABSL_GUARDED_BY(mu_) = 0;
+  };
+  auto accept_state = std::make_shared<AcceptState>();
   return impl_->CreateListener(
-      [this, on_accept = std::make_shared<Listener::AcceptCallback>(
-                 std::move(on_accept))](std::unique_ptr<Endpoint> endpoint,
-                                        MemoryAllocator memory_allocator) {
+      [this, accept_state,
+       on_accept = std::make_shared<Listener::AcceptCallback>(
+           std::move(on_accept))](std::unique_ptr<Endpoint> endpoint,
+                                  MemoryAllocator memory_allocator) {
+        {
+          grpc_core::MutexLock lock(&accept_state->mu_);
+          ++accept_state->pending_accepts_;
+        }
         Asynchronously(
-            [on_accept, endpoint = std::move(endpoint),
+            [on_accept, accept_state, endpoint = std::move(endpoint),
              memory_allocator = std::move(memory_allocator)]() mutable {
               (*on_accept)(std::move(endpoint), std::move(memory_allocator));
+              {
+                grpc_core::MutexLock lock(&accept_state->mu_);
+                --accept_state->pending_accepts_;
+                if (accept_state->pending_accepts_ == 0) {
+                  accept_state->cv_.Signal();
+                }
+              }
             });
       },
-      [this,
+      [this, accept_state,
        on_shutdown = std::move(on_shutdown)](absl::Status status) mutable {
-        Asynchronously([on_shutdown = std::move(on_shutdown),
+        Asynchronously([accept_state, on_shutdown = std::move(on_shutdown),
                         status = std::move(status)]() mutable {
+          while (true) {
+            grpc_core::MutexLock lock(&accept_state->mu_);
+            if (accept_state->pending_accepts_ == 0) {
+              break;
+            }
+            accept_state->cv_.Wait(&accept_state->mu_);
+          }
           on_shutdown(std::move(status));
         });
       },

src/core/lib/resource_quota/arena.h

@@ -94,8 +94,13 @@ class ArenaContextTraits : public BaseArenaContextTraits {
 };
 
 template <typename T>
-const uint16_t ArenaContextTraits<T>::id_ = BaseArenaContextTraits::MakeId(
-    [](void* ptr) { ArenaContextType<T>::Destroy(static_cast<T*>(ptr)); });
+void DestroyArenaContext(void* p) {
+  ArenaContextType<T>::Destroy(static_cast<T*>(p));
+}
+
+template <typename T>
+const uint16_t ArenaContextTraits<T>::id_ =
+    BaseArenaContextTraits::MakeId(DestroyArenaContext<T>);
 
 template <typename T, typename A, typename B>
 struct IfArray {
@@ -283,6 +288,7 @@ class Arena final : public RefCounted<Arena, NonPolymorphicRefCount,
       ArenaContextType<T>::Destroy(static_cast<T*>(slot));
     }
     slot = context;
+    DCHECK_EQ(GetContext<T>(), context);
   }
 
  private:

src/core/load_balancing/grpclb/grpclb.cc

@@ -73,6 +73,8 @@
 #include "absl/container/inlined_vector.h"
 #include "absl/functional/function_ref.h"
 #include "absl/log/check.h"
+#include "absl/log/globals.h"
+#include "absl/log/log.h"
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/strings/str_cat.h"
@@ -1162,13 +1164,12 @@ void GrpcLb::BalancerCallState::OnBalancerMessageReceivedLocked() {
   upb::Arena arena;
   if (!GrpcLbResponseParse(response_slice, arena.ptr(), &response) ||
       (response.type == response.INITIAL && seen_initial_response_)) {
-    if (gpr_should_log(GPR_LOG_SEVERITY_ERROR)) {
+    if (absl::MinLogLevel() <= absl::LogSeverityAtLeast::kError) {
       char* response_slice_str =
           grpc_dump_slice(response_slice, GPR_DUMP_ASCII | GPR_DUMP_HEX);
-      gpr_log(GPR_ERROR,
-              "[grpclb %p] lb_calld=%p: Invalid LB response received: '%s'. "
-              "Ignoring.",
-              grpclb_policy(), this, response_slice_str);
+      LOG(ERROR) << "[grpclb " << grpclb_policy() << "] lb_calld=" << this
+                 << ": Invalid LB response received: '" << response_slice_str
+                 << "'. Ignoring.";
       gpr_free(response_slice_str);
     }
   } else {

src/cpp/ext/gcp/environment_autodetect.cc

@@ -23,12 +23,12 @@
 
 #include "absl/container/flat_hash_map.h"
 #include "absl/log/check.h"
+#include "absl/log/log.h"
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/types/optional.h"
 
 #include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
 #include <grpc/support/port_platform.h>
 #include <grpc/support/sync.h>
 #include <grpcpp/impl/grpc_library.h>
@@ -64,8 +64,8 @@ std::string GetNamespaceName() {
   auto namespace_name = grpc_core::LoadFile(filename, false);
   if (!namespace_name.ok()) {
     if (GRPC_TRACE_FLAG_ENABLED(environment_autodetect)) {
-      gpr_log(GPR_DEBUG, "Reading file %s failed: %s", filename,
-              grpc_core::StatusToString(namespace_name.status()).c_str());
+      VLOG(2) << "Reading file " << filename << " failed: "
+              << grpc_core::StatusToString(namespace_name.status());
     }
     // Fallback on an environment variable
     return grpc_core::GetEnv("NAMESPACE_NAME").value_or("");
@@ -249,13 +249,12 @@ class EnvironmentAutoDetectHelper
           element.first, &pollent_,
           [this](std::string attribute, absl::StatusOr<std::string> result) {
             if (GRPC_TRACE_FLAG_ENABLED(environment_autodetect)) {
-              gpr_log(
-                  GPR_INFO,
-                  "Environment AutoDetect: Attribute: \"%s\" Result: \"%s\"",
-                  attribute.c_str(),
-                  result.ok()
-                      ? result.value().c_str()
-                      : grpc_core::StatusToString(result.status()).c_str());
+              LOG(INFO) << "Environment AutoDetect: Attribute: \"" << attribute
+                        << "\" Result: \""
+                        << (result.ok()
+                                ? result.value()
+                                : grpc_core::StatusToString(result.status()))
+                        << "\"";
             }
             absl::optional<EnvironmentAutoDetect::ResourceType> resource;
             {
@@ -270,20 +269,18 @@ class EnvironmentAutoDetectHelper
                 // assuming a GCE environment, fallback to "global".
                 else if (assuming_gce_) {
                   if (GRPC_TRACE_FLAG_ENABLED(environment_autodetect)) {
-                    gpr_log(GPR_INFO,
-                            "Environment Autodetect: Falling back to global "
-                            "resource type");
+                    LOG(INFO) << "Environment Autodetect: Falling back to "
+                                 "global resource type";
                   }
                   assuming_gce_ = false;
                   resource_.resource_type = "global";
                 }
                 attributes_to_fetch_.erase(it);
               } else {
                 // This should not happen
-                gpr_log(GPR_ERROR,
-                        "An unexpected attribute was seen from the "
-                        "MetadataServer: %s",
-                        attribute.c_str());
+                LOG(ERROR) << "An unexpected attribute was seen from the "
+                              "MetadataServer: "
+                           << attribute;
               }
               if (attributes_to_fetch_.empty()) {
                 resource = std::move(resource_);