[metadata] Optimize metadata creation (#37111)

- add a benchmark for various metadata creation styles
- add factory functions for status + message - these are 3-10x faster than going via absl::Status
- add a `MakePooledForOverwrite` function to Arena, use it everywhere -- this naming matches `std::make_unique_for_overwrite` in C++20, and avoids some language mandated initialization in `Table` (underlying `MetadataMap<>`) - speeding creation of metadata handles by 30%

For `bm_call_spine` we see before:
```
BM_UnaryWithSpawnPerEnd<CallSpineFixture>_median             745 ns          745 ns
```
and after:
```
BM_UnaryWithSpawnPerEnd<CallSpineFixture>_median             699 ns          699 ns
```

Closes #37111

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/37111 from ctiller:meta-magic-2 100464fc7e
PiperOrigin-RevId: 652900726

CMakeLists.txt

@@ -907,6 +907,9 @@ if(gRPC_BUILD_TESTS)
   if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_POSIX)
     add_dependencies(buildtests_c bm_inproc)
   endif()
+  if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_POSIX)
+    add_dependencies(buildtests_c bm_metadata)
+  endif()
   if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_POSIX)
     add_dependencies(buildtests_c bm_party)
   endif()
@@ -6211,6 +6214,45 @@ if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_POSIX)
   )
 
 
+endif()
+endif()
+if(gRPC_BUILD_TESTS)
+if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_POSIX)
+
+  add_executable(bm_metadata
+    test/core/transport/bm_metadata.cc
+  )
+  if(WIN32 AND MSVC)
+    if(BUILD_SHARED_LIBS)
+      target_compile_definitions(bm_metadata
+      PRIVATE
+        "GPR_DLL_IMPORTS"
+        "GRPC_DLL_IMPORTS"
+      )
+    endif()
+  endif()
+  target_compile_features(bm_metadata PUBLIC cxx_std_14)
+  target_include_directories(bm_metadata
+    PRIVATE
+      ${CMAKE_CURRENT_SOURCE_DIR}
+      ${CMAKE_CURRENT_SOURCE_DIR}/include
+      ${_gRPC_ADDRESS_SORTING_INCLUDE_DIR}
+      ${_gRPC_RE2_INCLUDE_DIR}
+      ${_gRPC_SSL_INCLUDE_DIR}
+      ${_gRPC_UPB_GENERATED_DIR}
+      ${_gRPC_UPB_GRPC_GENERATED_DIR}
+      ${_gRPC_UPB_INCLUDE_DIR}
+      ${_gRPC_XXHASH_INCLUDE_DIR}
+      ${_gRPC_ZLIB_INCLUDE_DIR}
+  )
+
+  target_link_libraries(bm_metadata
+    ${_gRPC_ALLTARGETS_LIBRARIES}
+    ${_gRPC_BENCHMARK_LIBRARIES}
+    grpc
+  )
+
+
 endif()
 endif()
 if(gRPC_BUILD_TESTS)

build_autogenerated.yaml

@@ -5279,6 +5279,24 @@ targets:
   - linux
   - posix
   uses_polling: false
+- name: bm_metadata
+  build: test
+  language: c
+  headers:
+  - test/core/transport/call_spine_benchmarks.h
+  src:
+  - test/core/transport/bm_metadata.cc
+  deps:
+  - benchmark
+  - grpc
+  args:
+  - --benchmark_min_time=0.001s
+  benchmark: true
+  defaults: benchmark
+  platforms:
+  - linux
+  - posix
+  uses_polling: false
 - name: bm_party
   build: test
   language: c

src/core/ext/filters/message_size/message_size_filter.cc

@@ -165,14 +165,12 @@ ServerMetadataHandle CheckPayload(const Message& msg,
               << " len:" << msg.payload()->Length() << " max:" << *max_length;
   }
   if (msg.payload()->Length() <= *max_length) return nullptr;
-  auto r = Arena::MakePooled<ServerMetadata>();
-  r->Set(GrpcStatusMetadata(), GRPC_STATUS_RESOURCE_EXHAUSTED);
-  r->Set(GrpcMessageMetadata(),
-         Slice::FromCopiedString(absl::StrFormat(
-             "%s: %s message larger than max (%u vs. %d)",
-             is_client ? "CLIENT" : "SERVER", is_send ? "Sent" : "Received",
-             msg.payload()->Length(), *max_length)));
-  return r;
+  return ServerMetadataFromStatus(
+      GRPC_STATUS_RESOURCE_EXHAUSTED,
+      absl::StrFormat("%s: %s message larger than max (%u vs. %d)",
+                      is_client ? "CLIENT" : "SERVER",
+                      is_send ? "Sent" : "Received", msg.payload()->Length(),
+                      *max_length));
 }
 }  // namespace
 

src/core/ext/transport/chaotic_good/frame.cc

@@ -154,7 +154,7 @@ absl::StatusOr<Arena::PoolPtr<Metadata>> ReadMetadata(
   if (!maybe_slices.ok()) return maybe_slices.status();
   auto& slices = *maybe_slices;
   CHECK_NE(arena, nullptr);
-  Arena::PoolPtr<Metadata> metadata = Arena::MakePooled<Metadata>();
+  Arena::PoolPtr<Metadata> metadata = Arena::MakePooledForOverwrite<Metadata>();
   parser->BeginFrame(
       metadata.get(), std::numeric_limits<uint32_t>::max(),
       std::numeric_limits<uint32_t>::max(),

src/core/ext/transport/chaotic_good/settings_metadata.cc

@@ -24,7 +24,7 @@ namespace grpc_core {
 namespace chaotic_good {
 
 Arena::PoolPtr<grpc_metadata_batch> SettingsMetadata::ToMetadataBatch() {
-  auto md = Arena::MakePooled<grpc_metadata_batch>();
+  auto md = Arena::MakePooledForOverwrite<grpc_metadata_batch>();
   auto add = [&md](absl::string_view key, std::string value) {
     md->Append(key, Slice::FromCopiedString(value),
                [key, value](absl::string_view error, const Slice&) {

src/core/lib/gprpp/table.h

@@ -35,20 +35,28 @@ namespace grpc_core {
 namespace table_detail {
 
 // A tuple-like type that contains manually constructed elements.
-template <typename... Ts>
+template <typename, typename... Ts>
 struct Elements;
 
 template <typename T, typename... Ts>
-struct Elements<T, Ts...> : Elements<Ts...> {
-  union U {
-    U() {}
-    ~U() {}
-    GPR_NO_UNIQUE_ADDRESS T x;
+struct Elements<absl::enable_if_t<!std::is_empty<T>::value>, T, Ts...>
+    : Elements<void, Ts...> {
+  struct alignas(T) Data {
+    unsigned char bytes[sizeof(T)];
   };
-  U u;
+  Data x;
+  Elements() {}
+  T* ptr() { return reinterpret_cast<T*>(x.bytes); }
+  const T* ptr() const { return reinterpret_cast<const T*>(x.bytes); }
+};
+template <typename T, typename... Ts>
+struct Elements<absl::enable_if_t<std::is_empty<T>::value>, T, Ts...>
+    : Elements<void, Ts...> {
+  T* ptr() { return reinterpret_cast<T*>(this); }
+  const T* ptr() const { return reinterpret_cast<const T*>(this); }
 };
 template <>
-struct Elements<> {};
+struct Elements<void> {};
 
 // Element accessor for Elements<>
 // Provides a static method f that returns a pointer to the value of element I
@@ -58,17 +66,17 @@ struct GetElem;
 
 template <typename T, typename... Ts>
 struct GetElem<0, T, Ts...> {
-  static T* f(Elements<T, Ts...>* e) { return &e->u.x; }
-  static const T* f(const Elements<T, Ts...>* e) { return &e->u.x; }
+  static T* f(Elements<void, T, Ts...>* e) { return e->ptr(); }
+  static const T* f(const Elements<void, T, Ts...>* e) { return e->ptr(); }
 };
 
 template <size_t I, typename T, typename... Ts>
 struct GetElem<I, T, Ts...> {
-  static auto f(Elements<T, Ts...>* e)
+  static auto f(Elements<void, T, Ts...>* e)
       -> decltype(GetElem<I - 1, Ts...>::f(e)) {
     return GetElem<I - 1, Ts...>::f(e);
   }
-  static auto f(const Elements<T, Ts...>* e)
+  static auto f(const Elements<void, T, Ts...>* e)
       -> decltype(GetElem<I - 1, Ts...>::f(e)) {
     return GetElem<I - 1, Ts...>::f(e);
   }
@@ -340,7 +348,7 @@ class Table {
   // the default) -- one bit for each type in Ts.
   using PresentBits = BitSet<sizeof...(Ts)>;
   // The tuple-like backing structure for Table.
-  using Elements = table_detail::Elements<Ts...>;
+  using Elements = table_detail::Elements<void, Ts...>;
   // Extractor from Elements
   template <size_t I>
   using GetElem = table_detail::GetElem<I, Ts...>;
@@ -442,9 +450,9 @@ class Table {
   }
 
   // Bit field indicating which elements are set.
-  GPR_NO_UNIQUE_ADDRESS PresentBits present_bits_;
+  PresentBits present_bits_;
   // The memory to store the elements themselves.
-  GPR_NO_UNIQUE_ADDRESS Elements elements_;
+  Elements elements_;
 };
 
 }  // namespace grpc_core

src/core/lib/resource_quota/arena.h

@@ -247,6 +247,11 @@ class Arena final : public RefCounted<Arena, NonPolymorphicRefCount,
     return PoolPtr<T>(new T(std::forward<Args>(args)...), PooledDeleter());
   }
 
+  template <typename T>
+  static PoolPtr<T> MakePooledForOverwrite() {
+    return PoolPtr<T>(new T, PooledDeleter());
+  }
+
   // Make a unique_ptr to an array of T that is allocated from the arena.
   // When the pointer is released, the memory may be reused for other
   // MakePooled(.*) calls.

src/core/lib/surface/client_call.cc

@@ -312,7 +312,7 @@ void ClientCall::CommitBatch(const grpc_op* ops, size_t nops, void* notify_tag,
                 ServerMetadataHandle metadata;
                 if (!md.ok() || !md->has_value()) {
                   is_trailers_only_ = true;
-                  metadata = Arena::MakePooled<ServerMetadata>();
+                  metadata = Arena::MakePooledForOverwrite<ServerMetadata>();
                 } else {
                   metadata = std::move(md->value());
                   is_trailers_only_ =

src/core/lib/surface/client_call.h

@@ -150,7 +150,7 @@ class ClientCall final
   };
   std::atomic<uintptr_t> call_state_{kUnstarted};
   ClientMetadataHandle send_initial_metadata_{
-      Arena::MakePooled<ClientMetadata>()};
+      Arena::MakePooledForOverwrite<ClientMetadata>()};
   CallInitiator started_call_initiator_;
   // Status passed to CancelWithError;
   // if call_state_ == kCancelled then this is the authoritative status,

src/core/lib/transport/call_filters.cc

@@ -193,7 +193,7 @@ void CallFilters::CancelDueToFailedPipeOperation(SourceLocation but_where) {
         << "Cancelling due to failed pipe operation: " << DebugString();
   }
   auto status =
-      ServerMetadataFromStatus(absl::CancelledError("Failed pipe operation"));
+      ServerMetadataFromStatus(GRPC_STATUS_CANCELLED, "Failed pipe operation");
   status->Set(GrpcCallWasCancelled(), true);
   PushServerTrailingMetadata(std::move(status));
 }

src/core/lib/transport/interception_chain.cc

@@ -32,7 +32,7 @@ std::atomic<size_t> InterceptionChainBuilder::next_filter_id_{0};
 // HijackedCall
 
 CallInitiator HijackedCall::MakeCall() {
-  auto metadata = Arena::MakePooled<ClientMetadata>();
+  auto metadata = Arena::MakePooledForOverwrite<ClientMetadata>();
   *metadata = metadata_->Copy();
   return MakeCallWithMetadata(std::move(metadata));
 }

src/core/lib/transport/metadata.cc

@@ -22,7 +22,7 @@
 namespace grpc_core {
 
 ServerMetadataHandle ServerMetadataFromStatus(const absl::Status& status) {
-  auto hdl = Arena::MakePooled<ServerMetadata>();
+  auto hdl = Arena::MakePooledForOverwrite<ServerMetadata>();
   grpc_status_code code;
   std::string message;
   grpc_error_get_status(status, Timestamp::InfFuture(), &code, &message,
@@ -41,4 +41,21 @@ ServerMetadataHandle CancelledServerMetadataFromStatus(
   return hdl;
 }
 
+ServerMetadataHandle ServerMetadataFromStatus(grpc_status_code code,
+                                              absl::string_view message) {
+  auto hdl = Arena::MakePooledForOverwrite<ServerMetadata>();
+  hdl->Set(GrpcStatusMetadata(), code);
+  hdl->Set(GrpcMessageMetadata(), Slice::FromCopiedString(message));
+  return hdl;
+}
+
+ServerMetadataHandle CancelledServerMetadataFromStatus(
+    grpc_status_code code, absl::string_view message) {
+  auto hdl = Arena::MakePooledForOverwrite<ServerMetadata>();
+  hdl->Set(GrpcStatusMetadata(), code);
+  hdl->Set(GrpcMessageMetadata(), Slice::FromCopiedString(message));
+  hdl->Set(GrpcCallWasCancelled(), true);
+  return hdl;
+}
+
 }  // namespace grpc_core

src/core/lib/transport/metadata.h

@@ -51,9 +51,30 @@ inline bool IsStatusOk(const ServerMetadataHandle& m) {
          GRPC_STATUS_OK;
 }
 
+// Convert absl::Status to ServerMetadata
 ServerMetadataHandle ServerMetadataFromStatus(const absl::Status& status);
+// Convert absl::Status to ServerMetadata, and set GrpcCallWasCancelled() to
+// true
 ServerMetadataHandle CancelledServerMetadataFromStatus(
     const absl::Status& status);
+// Server metadata with status code set
+inline ServerMetadataHandle ServerMetadataFromStatus(grpc_status_code code) {
+  auto hdl = Arena::MakePooledForOverwrite<ServerMetadata>();
+  hdl->Set(GrpcStatusMetadata(), code);
+  return hdl;
+}
+inline ServerMetadataHandle CancelledServerMetadataFromStatus(
+    grpc_status_code code) {
+  auto hdl = Arena::MakePooledForOverwrite<ServerMetadata>();
+  hdl->Set(GrpcStatusMetadata(), code);
+  hdl->Set(GrpcCallWasCancelled(), true);
+  return hdl;
+}
+// The same, but with an error string
+ServerMetadataHandle ServerMetadataFromStatus(grpc_status_code code,
+                                              absl::string_view message);
+ServerMetadataHandle CancelledServerMetadataFromStatus(
+    grpc_status_code code, absl::string_view message);
 
 template <>
 struct StatusCastImpl<ServerMetadataHandle, absl::Status> {

test/core/call/bm_client_call.cc

@@ -140,7 +140,8 @@ void BM_Unary(benchmark::State& state) {
     auto unstarted_handler = helper.TakeHandler();
     unstarted_handler.SpawnInfallible("run_handler", [&]() mutable {
       auto handler = unstarted_handler.StartCall();
-      handler.PushServerInitialMetadata(Arena::MakePooled<ServerMetadata>());
+      handler.PushServerInitialMetadata(
+          Arena::MakePooledForOverwrite<ServerMetadata>());
       auto response =
           Arena::MakePooled<Message>(SliceBuffer(response_payload.Copy()), 0);
       return Map(
@@ -156,7 +157,8 @@ void BM_Unary(benchmark::State& state) {
               handler.PushMessage(std::move(response))),
           [handler](StatusFlag status) mutable {
             CHECK(status.ok());
-            auto trailing_metadata = Arena::MakePooled<ServerMetadata>();
+            auto trailing_metadata =
+                Arena::MakePooledForOverwrite<ServerMetadata>();
             trailing_metadata->Set(GrpcStatusMetadata(), GRPC_STATUS_OK);
             handler.PushServerTrailingMetadata(std::move(trailing_metadata));
             return Empty{};

test/core/call/client_call_test.cc

@@ -196,7 +196,7 @@ CLIENT_CALL_TEST(SendInitialMetadataAndReceiveStatusAfterCancellation) {
         EXPECT_EQ((*md)->get_pointer(HttpPathMetadata())->as_string_view(),
                   kDefaultPath);
         handler().PushServerTrailingMetadata(
-            ServerMetadataFromStatus(absl::InternalError("test error")));
+            ServerMetadataFromStatus(GRPC_STATUS_INTERNAL, "test error"));
         return Immediate(Empty{});
       });
   Expect(1, true);

test/core/call/server_call_test.cc

@@ -73,7 +73,8 @@ class ServerCallTest : public YodelTest {
   ClientMetadataHandle MakeClientInitialMetadata(
       std::initializer_list<std::pair<absl::string_view, absl::string_view>>
           md) {
-    auto client_initial_metadata = Arena::MakePooled<ClientMetadata>();
+    auto client_initial_metadata =
+        Arena::MakePooledForOverwrite<ClientMetadata>();
     client_initial_metadata->Set(HttpPathMetadata(),
                                  Slice::FromCopiedString(kDefaultPath));
     for (const auto& pair : md) {

test/core/client_channel/bm_client_channel.cc

@@ -50,19 +50,19 @@ class ClientChannelTraits {
   }
 
   ClientMetadataHandle MakeClientInitialMetadata() {
-    auto md = Arena::MakePooled<ClientMetadata>();
+    auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
     md->Set(HttpPathMetadata(), kTestPath.Copy());
     return md;
   }
 
   ServerMetadataHandle MakeServerInitialMetadata() {
-    return Arena::MakePooled<ServerMetadata>();
+    return Arena::MakePooledForOverwrite<ServerMetadata>();
   }
 
   MessageHandle MakePayload() { return Arena::MakePooled<Message>(); }
 
   ServerMetadataHandle MakeServerTrailingMetadata() {
-    auto md = Arena::MakePooled<ServerMetadata>();
+    auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
     return md;
   }
 

test/core/client_channel/client_channel_test.cc

@@ -57,7 +57,8 @@ class ClientChannelTest : public YodelTest {
   ClientChannel& channel() { return *channel_; }
 
   ClientMetadataHandle MakeClientInitialMetadata() {
-    auto client_initial_metadata = Arena::MakePooled<ClientMetadata>();
+    auto client_initial_metadata =
+        Arena::MakePooledForOverwrite<ClientMetadata>();
     client_initial_metadata->Set(HttpPathMetadata(),
                                  Slice::FromCopiedString(kTestPath));
     return client_initial_metadata;

test/core/client_channel/connected_subchannel_test.cc

@@ -59,7 +59,8 @@ class ConnectedSubchannelTest : public YodelTest {
   }
 
   ClientMetadataHandle MakeClientInitialMetadata() {
-    auto client_initial_metadata = Arena::MakePooled<ClientMetadata>();
+    auto client_initial_metadata =
+        Arena::MakePooledForOverwrite<ClientMetadata>();
     client_initial_metadata->Set(HttpPathMetadata(),
                                  Slice::FromCopiedString(kTestPath));
     return client_initial_metadata;

test/core/client_channel/load_balanced_call_destination_test.cc

@@ -41,7 +41,8 @@ class LoadBalancedCallDestinationTest : public YodelTest {
   using YodelTest::YodelTest;
 
   ClientMetadataHandle MakeClientInitialMetadata() {
-    auto client_initial_metadata = Arena::MakePooled<ClientMetadata>();
+    auto client_initial_metadata =
+        Arena::MakePooledForOverwrite<ClientMetadata>();
     client_initial_metadata->Set(HttpPathMetadata(),
                                  Slice::FromCopiedString(kTestPath));
     return client_initial_metadata;

test/core/filters/bm_http_client_filter.cc

@@ -34,17 +34,17 @@ class HttpClientFilterTraits {
   ChannelArgs MakeChannelArgs() { return ChannelArgs().SetObject(&transport_); }
 
   ClientMetadataHandle MakeClientInitialMetadata() {
-    return Arena::MakePooled<ClientMetadata>();
+    return Arena::MakePooledForOverwrite<ClientMetadata>();
   }
 
   ServerMetadataHandle MakeServerInitialMetadata() {
-    return Arena::MakePooled<ServerMetadata>();
+    return Arena::MakePooledForOverwrite<ServerMetadata>();
   }
 
   MessageHandle MakePayload() { return Arena::MakePooled<Message>(); }
 
   ServerMetadataHandle MakeServerTrailingMetadata() {
-    auto md = Arena::MakePooled<ServerMetadata>();
+    auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
     md->Set(HttpStatusMetadata(), 200);
     return md;
   }

test/core/surface/channel_init_test.cc

@@ -338,8 +338,8 @@ TEST(ChannelInitTest, CanCreateFilterWithCall) {
   auto arena = allocator->MakeArena();
   arena->SetContext<grpc_event_engine::experimental::EventEngine>(
       event_engine.get());
-  auto call =
-      MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena));
+  auto call = MakeCallPair(Arena::MakePooledForOverwrite<ClientMetadata>(),
+                           std::move(arena));
   (*stack)->StartCall(std::move(call.handler));
   EXPECT_EQ(p, 1);
   EXPECT_EQ(handled, 1);

test/core/transport/BUILD

@@ -250,3 +250,13 @@ grpc_cc_benchmark(
         "//src/core:default_event_engine",
     ],
 )
+
+grpc_cc_benchmark(
+    name = "bm_metadata",
+    srcs = ["bm_metadata.cc"],
+    deps = [
+        ":call_spine_benchmarks",
+        "//:grpc",
+        "//src/core:default_event_engine",
+    ],
+)

test/core/transport/benchmarks/bm_chaotic_good.cc

@@ -54,19 +54,19 @@ class ChaoticGoodTraits {
   }
 
   ClientMetadataHandle MakeClientInitialMetadata() {
-    auto md = Arena::MakePooled<ClientMetadata>();
+    auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
     md->Set(HttpPathMetadata(), kTestPath.Copy());
     return md;
   }
 
   ServerMetadataHandle MakeServerInitialMetadata() {
-    return Arena::MakePooled<ServerMetadata>();
+    return Arena::MakePooledForOverwrite<ServerMetadata>();
   }
 
   MessageHandle MakePayload() { return Arena::MakePooled<Message>(); }
 
   ServerMetadataHandle MakeServerTrailingMetadata() {
-    auto md = Arena::MakePooled<ServerMetadata>();
+    auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
     return md;
   }
 };

test/core/transport/benchmarks/bm_inproc.cc

@@ -44,19 +44,19 @@ class InprocTraits {
   }
 
   ClientMetadataHandle MakeClientInitialMetadata() {
-    auto md = Arena::MakePooled<ClientMetadata>();
+    auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
     md->Set(HttpPathMetadata(), kTestPath.Copy());
     return md;
   }
 
   ServerMetadataHandle MakeServerInitialMetadata() {
-    return Arena::MakePooled<ServerMetadata>();
+    return Arena::MakePooledForOverwrite<ServerMetadata>();
   }
 
   MessageHandle MakePayload() { return Arena::MakePooled<Message>(); }
 
   ServerMetadataHandle MakeServerTrailingMetadata() {
-    auto md = Arena::MakePooled<ServerMetadata>();
+    auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
     return md;
   }
 };

test/core/transport/bm_call_spine.cc

@@ -28,19 +28,19 @@ class CallSpineFixture {
     auto arena = arena_allocator_->MakeArena();
     arena->SetContext<grpc_event_engine::experimental::EventEngine>(
         event_engine_.get());
-    auto p =
-        MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena));
+    auto p = MakeCallPair(Arena::MakePooledForOverwrite<ClientMetadata>(),
+                          std::move(arena));
     return {std::move(p.initiator), p.handler.StartCall()};
   }
 
   ServerMetadataHandle MakeServerInitialMetadata() {
-    return Arena::MakePooled<ServerMetadata>();
+    return Arena::MakePooledForOverwrite<ServerMetadata>();
   }
 
   MessageHandle MakePayload() { return Arena::MakePooled<Message>(); }
 
   ServerMetadataHandle MakeServerTrailingMetadata() {
-    return Arena::MakePooled<ServerMetadata>();
+    return Arena::MakePooledForOverwrite<ServerMetadata>();
   }
 
  private:
@@ -63,10 +63,10 @@ class ForwardCallFixture {
         event_engine_.get());
     arena2->SetContext<grpc_event_engine::experimental::EventEngine>(
         event_engine_.get());
-    auto p1 =
-        MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena1));
-    auto p2 =
-        MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena2));
+    auto p1 = MakeCallPair(Arena::MakePooledForOverwrite<ClientMetadata>(),
+                           std::move(arena1));
+    auto p2 = MakeCallPair(Arena::MakePooledForOverwrite<ClientMetadata>(),
+                           std::move(arena2));
     p1.handler.SpawnInfallible("initial_metadata", [&]() {
       auto p1_handler = p1.handler.StartCall();
       return Map(
@@ -86,13 +86,13 @@ class ForwardCallFixture {
   }
 
   ServerMetadataHandle MakeServerInitialMetadata() {
-    return Arena::MakePooled<ServerMetadata>();
+    return Arena::MakePooledForOverwrite<ServerMetadata>();
   }
 
   MessageHandle MakePayload() { return Arena::MakePooled<Message>(); }
 
   ServerMetadataHandle MakeServerTrailingMetadata() {
-    return Arena::MakePooled<ServerMetadata>();
+    return Arena::MakePooledForOverwrite<ServerMetadata>();
   }
 
  private:

test/core/transport/bm_metadata.cc

@@ -0,0 +1,86 @@
+//
+//
+// Copyright 2017 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+//
+
+#include <benchmark/benchmark.h>
+
+#include "src/core/lib/transport/metadata.h"
+
+namespace grpc_core {
+namespace {
+
+void BM_MetadataMapCreateDestroy(benchmark::State& state) {
+  for (auto _ : state) {
+    auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
+  }
+}
+BENCHMARK(BM_MetadataMapCreateDestroy);
+
+void BM_MetadataMapCreateDestroyOnStack(benchmark::State& state) {
+  for (auto _ : state) {
+    ServerMetadata md;
+  }
+}
+BENCHMARK(BM_MetadataMapCreateDestroyOnStack);
+
+void BM_MetadataMapCreateDestroySetStatus(benchmark::State& state) {
+  auto message = Slice::FromExternalString("message");
+  for (auto _ : state) {
+    auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
+    md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNKNOWN);
+    md->Set(GrpcMessageMetadata(), message.Copy());
+  }
+}
+BENCHMARK(BM_MetadataMapCreateDestroySetStatus);
+
+void BM_MetadataMapCreateDestroySetStatusCancelled(benchmark::State& state) {
+  auto message = Slice::FromExternalString("message");
+  for (auto _ : state) {
+    auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
+    md->Set(GrpcStatusMetadata(), GRPC_STATUS_CANCELLED);
+  }
+}
+BENCHMARK(BM_MetadataMapCreateDestroySetStatusCancelled);
+
+void BM_MetadataMapFromAbslStatusCancelled(benchmark::State& state) {
+  for (auto _ : state) {
+    ServerMetadataFromStatus(absl::CancelledError());
+  }
+}
+BENCHMARK(BM_MetadataMapFromAbslStatusCancelled);
+
+void BM_MetadataMapFromAbslStatusOk(benchmark::State& state) {
+  for (auto _ : state) {
+    ServerMetadataFromStatus(absl::OkStatus());
+  }
+}
+BENCHMARK(BM_MetadataMapFromAbslStatusOk);
+
+}  // namespace
+}  // namespace grpc_core
+
+// Some distros have RunSpecifiedBenchmarks under the benchmark namespace,
+// and others do not. This allows us to support both modes.
+namespace benchmark {
+void RunTheBenchmarksNamespaced() { RunSpecifiedBenchmarks(); }
+}  // namespace benchmark
+
+int main(int argc, char** argv) {
+  ::benchmark::Initialize(&argc, argv);
+  benchmark::RunTheBenchmarksNamespaced();
+  return 0;
+}

test/core/transport/call_filters_test.cc

@@ -329,7 +329,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningVoid) {
   EXPECT_EQ(d.client_initial_metadata.ops[0].poll, nullptr);
   EXPECT_EQ(d.client_initial_metadata.ops[0].early_destroy, nullptr);
   // Check promise init
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   char call_data;
   auto r = d.client_initial_metadata.ops[0].promise_init(
@@ -364,7 +364,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningVoidTakingChannelPtr) {
   EXPECT_EQ(d.client_initial_metadata.ops[0].poll, nullptr);
   EXPECT_EQ(d.client_initial_metadata.ops[0].early_destroy, nullptr);
   // Check promise init
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   char call_data;
   auto r = d.client_initial_metadata.ops[0].promise_init(
@@ -409,7 +409,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningAbslStatus) {
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
@@ -420,7 +420,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningAbslStatus) {
   EXPECT_EQ(r.value().ok->get_pointer(HttpPathMetadata())->as_string_view(),
             "hello");
   // A failing call
-  md = Arena::MakePooled<ClientMetadata>();
+  md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = d.client_initial_metadata.ops[0].promise_init(
       nullptr, call_data, d.client_initial_metadata.ops[0].channel_data,
@@ -467,7 +467,7 @@ TEST(StackDataTest,
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
@@ -478,7 +478,7 @@ TEST(StackDataTest,
   EXPECT_EQ(r.value().ok->get_pointer(HttpPathMetadata())->as_string_view(),
             "hello");
   // A failing call
-  md = Arena::MakePooled<ClientMetadata>();
+  md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = d.client_initial_metadata.ops[0].promise_init(
       nullptr, call_data, d.client_initial_metadata.ops[0].channel_data,
@@ -498,7 +498,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningServerMetadata) {
         md.Set(HttpPathMetadata(), Slice::FromStaticString("hello"));
         bool first = std::exchange(first_, false);
         return first ? nullptr
-                     : ServerMetadataFromStatus(absl::CancelledError());
+                     : ServerMetadataFromStatus(GRPC_STATUS_CANCELLED);
       }
 
      private:
@@ -524,7 +524,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningServerMetadata) {
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
@@ -535,7 +535,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningServerMetadata) {
   EXPECT_EQ(r.value().ok->get_pointer(HttpPathMetadata())->as_string_view(),
             "hello");
   // A failing call
-  md = Arena::MakePooled<ClientMetadata>();
+  md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = d.client_initial_metadata.ops[0].promise_init(
       nullptr, call_data, d.client_initial_metadata.ops[0].channel_data,
@@ -557,7 +557,7 @@ TEST(StackDataTest,
         const bool first = std::exchange(first_, false);
         p->v.push_back(first ? 11 : 22);
         return first ? nullptr
-                     : ServerMetadataFromStatus(absl::CancelledError());
+                     : ServerMetadataFromStatus(GRPC_STATUS_CANCELLED);
       }
 
      private:
@@ -584,7 +584,7 @@ TEST(StackDataTest,
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
@@ -595,7 +595,7 @@ TEST(StackDataTest,
   EXPECT_EQ(r.value().ok->get_pointer(HttpPathMetadata())->as_string_view(),
             "hello");
   // A failing call
-  md = Arena::MakePooled<ClientMetadata>();
+  md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = d.client_initial_metadata.ops[0].promise_init(
       nullptr, call_data, d.client_initial_metadata.ops[0].channel_data,
@@ -642,7 +642,7 @@ TEST(StackDataTest, PromiseClientInitialMetadataReturningAbslStatus) {
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
@@ -660,7 +660,7 @@ TEST(StackDataTest, PromiseClientInitialMetadataReturningAbslStatus) {
   EXPECT_EQ(r.value().ok->get_pointer(HttpPathMetadata())->as_string_view(),
             "hello");
   // A failing call
-  md = Arena::MakePooled<ClientMetadata>();
+  md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = d.client_initial_metadata.ops[0].promise_init(
       promise_data, call_data, d.client_initial_metadata.ops[0].channel_data,
@@ -673,7 +673,7 @@ TEST(StackDataTest, PromiseClientInitialMetadataReturningAbslStatus) {
   EXPECT_EQ(r.value().ok, nullptr);
   EXPECT_EQ(r.value().error->get(GrpcStatusMetadata()), GRPC_STATUS_CANCELLED);
   // A cancelled call
-  md = Arena::MakePooled<ClientMetadata>();
+  md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = d.client_initial_metadata.ops[0].promise_init(
       promise_data, call_data, d.client_initial_metadata.ops[0].channel_data,
@@ -723,7 +723,7 @@ TEST(StackDataTest,
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
@@ -741,7 +741,7 @@ TEST(StackDataTest,
   EXPECT_EQ(r.value().ok->get_pointer(HttpPathMetadata())->as_string_view(),
             "hello");
   // A failing call
-  md = Arena::MakePooled<ClientMetadata>();
+  md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = d.client_initial_metadata.ops[0].promise_init(
       promise_data, call_data, d.client_initial_metadata.ops[0].channel_data,
@@ -754,7 +754,7 @@ TEST(StackDataTest,
   EXPECT_EQ(r.value().ok, nullptr);
   EXPECT_EQ(r.value().error->get(GrpcStatusMetadata()), GRPC_STATUS_CANCELLED);
   // A cancelled call
-  md = Arena::MakePooled<ClientMetadata>();
+  md = Arena::MakePooledForOverwrite<ClientMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = d.client_initial_metadata.ops[0].promise_init(
       promise_data, call_data, d.client_initial_metadata.ops[0].channel_data,
@@ -791,7 +791,7 @@ TEST(StackDataTest, InstantServerInitialMetadataReturningVoid) {
   EXPECT_EQ(d.server_initial_metadata.ops[0].early_destroy, nullptr);
   // Check promise init
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ServerMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   char call_data;
   auto r = d.server_initial_metadata.ops[0].promise_init(
@@ -889,7 +889,7 @@ TEST(StackDataTest, ServerTrailingMetadataReturningVoid) {
   EXPECT_EQ(d.server_trailing_metadata[0].channel_data, &f1);
   // Check operation
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ServerMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   char call_data;
   auto r = d.server_trailing_metadata[0].server_trailing_metadata(
@@ -920,7 +920,7 @@ TEST(StackDataTest, ServerTrailingMetadataReturningVoidTakingChannelPtr) {
   EXPECT_EQ(d.server_trailing_metadata[0].channel_data, &f1);
   // Check operation
   auto arena = SimpleArenaAllocator()->MakeArena();
-  auto md = Arena::MakePooled<ServerMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   char call_data;
   auto r = d.server_trailing_metadata[0].server_trailing_metadata(
@@ -1007,7 +1007,7 @@ TEST(OperationExecutorTest, InstantTwo) {
   auto arena = SimpleArenaAllocator()->MakeArena();
   promise_detail::Context<Arena> ctx(arena.get());
   // First call succeeds
-  auto md = Arena::MakePooled<ServerMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   auto r =
       transformer.Start(&d.client_initial_metadata, std::move(md), call_data1);
@@ -1015,7 +1015,7 @@ TEST(OperationExecutorTest, InstantTwo) {
   EXPECT_EQ(r.value().ok->get_pointer(HttpPathMetadata())->as_string_view(),
             "world");
   // Second fails
-  md = Arena::MakePooled<ServerMetadata>();
+  md = Arena::MakePooledForOverwrite<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = transformer.Start(&d.client_initial_metadata, std::move(md), call_data1);
   EXPECT_TRUE(r.ready());
@@ -1070,7 +1070,7 @@ TEST(OperationExecutorTest, PromiseTwo) {
   auto arena = SimpleArenaAllocator()->MakeArena();
   promise_detail::Context<Arena> ctx(arena.get());
   // First call succeeds after two sets of two step delays.
-  auto md = Arena::MakePooled<ServerMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   auto r =
       transformer.Start(&d.client_initial_metadata, std::move(md), call_data1);
@@ -1086,7 +1086,7 @@ TEST(OperationExecutorTest, PromiseTwo) {
   EXPECT_EQ(r.value().ok->get_pointer(HttpPathMetadata())->as_string_view(),
             "world");
   // Second fails after one set of two step delays.
-  md = Arena::MakePooled<ServerMetadata>();
+  md = Arena::MakePooledForOverwrite<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   r = transformer.Start(&d.client_initial_metadata, std::move(md), call_data1);
   EXPECT_FALSE(r.ready());
@@ -1161,7 +1161,7 @@ TEST(CallFiltersTest, UnaryCall) {
   builder.Add(&f1);
   builder.Add(&f2);
   auto arena = SimpleArenaAllocator()->MakeArena();
-  CallFilters filters(Arena::MakePooled<ClientMetadata>());
+  CallFilters filters(Arena::MakePooledForOverwrite<ClientMetadata>());
   filters.AddStack(builder.Build());
   filters.Start();
   promise_detail::Context<Arena> ctx(arena.get());
@@ -1182,7 +1182,8 @@ TEST(CallFiltersTest, UnaryCall) {
   // Push should be done
   EXPECT_THAT(push_client_to_server_message(), IsReady(Success{}));
   // Push server initial metadata
-  filters.PushServerInitialMetadata(Arena::MakePooled<ServerMetadata>());
+  filters.PushServerInitialMetadata(
+      Arena::MakePooledForOverwrite<ServerMetadata>());
   auto pull_server_initial_metadata = filters.PullServerInitialMetadata();
   // Pull server initial metadata
   EXPECT_THAT(pull_server_initial_metadata(), IsReady());
@@ -1198,7 +1199,8 @@ TEST(CallFiltersTest, UnaryCall) {
   // Push should be done
   EXPECT_THAT(push_server_to_client_message(), IsReady(Success{}));
   // Push server trailing metadata
-  filters.PushServerTrailingMetadata(Arena::MakePooled<ServerMetadata>());
+  filters.PushServerTrailingMetadata(
+      Arena::MakePooledForOverwrite<ServerMetadata>());
   // Pull server trailing metadata
   auto pull_server_trailing_metadata = filters.PullServerTrailingMetadata();
   // Should be done
@@ -1253,7 +1255,7 @@ TEST(CallFiltersTest, UnaryCallWithMultiStack) {
   builder1.Add(&f1);
   builder2.Add(&f2);
   auto arena = SimpleArenaAllocator()->MakeArena();
-  CallFilters filters(Arena::MakePooled<ClientMetadata>());
+  CallFilters filters(Arena::MakePooledForOverwrite<ClientMetadata>());
   filters.AddStack(builder1.Build());
   filters.AddStack(builder2.Build());
   filters.Start();
@@ -1275,7 +1277,8 @@ TEST(CallFiltersTest, UnaryCallWithMultiStack) {
   // Push should be done
   EXPECT_THAT(push_client_to_server_message(), IsReady(Success{}));
   // Push server initial metadata
-  filters.PushServerInitialMetadata(Arena::MakePooled<ServerMetadata>());
+  filters.PushServerInitialMetadata(
+      Arena::MakePooledForOverwrite<ServerMetadata>());
   auto pull_server_initial_metadata = filters.PullServerInitialMetadata();
   // Pull server initial metadata
   EXPECT_THAT(pull_server_initial_metadata(), IsReady());
@@ -1291,7 +1294,8 @@ TEST(CallFiltersTest, UnaryCallWithMultiStack) {
   // Push should be done
   EXPECT_THAT(push_server_to_client_message(), IsReady(Success{}));
   // Push server trailing metadata
-  filters.PushServerTrailingMetadata(Arena::MakePooled<ServerMetadata>());
+  filters.PushServerTrailingMetadata(
+      Arena::MakePooledForOverwrite<ServerMetadata>());
   // Pull server trailing metadata
   auto pull_server_trailing_metadata = filters.PullServerTrailingMetadata();
   // Should be done

test/core/transport/call_spine_benchmarks.h

@@ -227,7 +227,7 @@ void BM_ClientToServerStreaming(benchmark::State& state) {
                                  });
   call.handler.SpawnInfallible("done", [handler = call.handler]() mutable {
     handler.PushServerTrailingMetadata(
-        CancelledServerMetadataFromStatus(absl::CancelledError()));
+        CancelledServerMetadataFromStatus(GRPC_STATUS_CANCELLED));
     return Empty{};
   });
 }

test/core/transport/call_spine_test.cc

@@ -41,7 +41,8 @@ class CallSpineTest : public YodelTest {
   using YodelTest::YodelTest;
 
   ClientMetadataHandle MakeClientInitialMetadata() {
-    auto client_initial_metadata = Arena::MakePooled<ClientMetadata>();
+    auto client_initial_metadata =
+        Arena::MakePooledForOverwrite<ClientMetadata>();
     client_initial_metadata->Set(HttpPathMetadata(),
                                  Slice::FromCopiedString(kTestPath));
     return client_initial_metadata;
@@ -126,7 +127,7 @@ void CallSpineTest::UnaryRequest(CallInitiator initiator, CallHandler handler) {
       [handler](ValueOrFailure<absl::optional<MessageHandle>> msg) mutable {
         EXPECT_TRUE(msg.ok());
         EXPECT_FALSE(msg.value().has_value());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
         return handler.PushServerInitialMetadata(std::move(md));
       },
@@ -137,7 +138,7 @@ void CallSpineTest::UnaryRequest(CallInitiator initiator, CallHandler handler) {
       },
       [handler](StatusFlag result) mutable {
         EXPECT_TRUE(result.ok());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};

test/core/transport/chaotic_good/client_transport_error_test.cc

@@ -123,7 +123,7 @@ auto SendClientToServerMessages(CallInitiator initiator, int num_messages) {
 }
 
 ClientMetadataHandle TestInitialMetadata() {
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromStaticString("/test"));
   return md;
 }

test/core/transport/chaotic_good/client_transport_test.cc

@@ -68,7 +68,7 @@ const uint8_t kGrpcStatus0[] = {0x10, 0x0b, 0x67, 0x72, 0x70, 0x63, 0x2d, 0x73,
                                 0x74, 0x61, 0x74, 0x75, 0x73, 0x01, 0x30};
 
 ClientMetadataHandle TestInitialMetadata() {
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromStaticString("/demo.Service/Step"));
   return md;
 }

test/core/transport/chaotic_good/server_transport_test.cc

@@ -71,13 +71,13 @@ const uint8_t kGrpcStatus0[] = {0x40, 0x0b, 0x67, 0x72, 0x70, 0x63, 0x2d, 0x73,
                                 0x74, 0x61, 0x74, 0x75, 0x73, 0x01, 0x30};
 
 ServerMetadataHandle TestInitialMetadata() {
-  auto md = Arena::MakePooled<ServerMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromStaticString("/demo.Service/Step"));
   return md;
 }
 
 ServerMetadataHandle TestTrailingMetadata() {
-  auto md = Arena::MakePooled<ServerMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
   md->Set(GrpcStatusMetadata(), GRPC_STATUS_OK);
   return md;
 }

test/core/transport/interception_chain_test.cc

@@ -161,7 +161,7 @@ class TestConsumingInterceptor final : public Interceptor {
   void InterceptCall(UnstartedCallHandler unstarted_call_handler) override {
     Consume(std::move(unstarted_call_handler))
         .PushServerTrailingMetadata(
-            ServerMetadataFromStatus(absl::InternalError("👊 consumed")));
+            ServerMetadataFromStatus(GRPC_STATUS_INTERNAL, "👊 consumed"));
   }
   void Orphaned() override {}
   static absl::StatusOr<RefCountedPtr<TestConsumingInterceptor<I>>> Create(
@@ -252,8 +252,8 @@ class InterceptionChainTest : public ::testing::Test {
     auto arena = call_arena_allocator_->MakeArena();
     arena->SetContext<grpc_event_engine::experimental::EventEngine>(
         event_engine_.get());
-    auto call =
-        MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena));
+    auto call = MakeCallPair(Arena::MakePooledForOverwrite<ClientMetadata>(),
+                             std::move(arena));
     Poll<ServerMetadataHandle> trailing_md;
     call.initiator.SpawnInfallible(
         "run_call", [destination, &call, &trailing_md]() mutable {
@@ -278,11 +278,11 @@ class InterceptionChainTest : public ::testing::Test {
                 << unstarted_call_handler.UnprocessedClientInitialMetadata()
                        .DebugString();
       EXPECT_EQ(metadata_.get(), nullptr);
-      metadata_ = Arena::MakePooled<ClientMetadata>();
+      metadata_ = Arena::MakePooledForOverwrite<ClientMetadata>();
       *metadata_ =
           unstarted_call_handler.UnprocessedClientInitialMetadata().Copy();
       unstarted_call_handler.PushServerTrailingMetadata(
-          ServerMetadataFromStatus(absl::InternalError("👊 cancelled")));
+          ServerMetadataFromStatus(GRPC_STATUS_INTERNAL, "👊 cancelled"));
     }
 
     void Orphaned() override {}

test/core/transport/test_suite/call_content.cc

@@ -68,7 +68,7 @@ TRANSPORT_TEST(UnaryWithSomeContent) {
   const auto server_trailing_metadata = RandomMetadata();
   const auto client_payload = RandomMessage();
   const auto server_payload = RandomMessage();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   FillMetadata(client_initial_metadata, *md);
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -126,7 +126,7 @@ TRANSPORT_TEST(UnaryWithSomeContent) {
       [&](ValueOrFailure<absl::optional<MessageHandle>> msg) {
         EXPECT_TRUE(msg.ok());
         EXPECT_FALSE(msg.value().has_value());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         FillMetadata(server_initial_metadata, *md);
         return handler.PushServerInitialMetadata(std::move(md));
       },
@@ -137,7 +137,7 @@ TRANSPORT_TEST(UnaryWithSomeContent) {
       },
       [&](StatusFlag result) mutable {
         EXPECT_TRUE(result.ok());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         FillMetadata(server_trailing_metadata, *md);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};

test/core/transport/test_suite/call_shapes.cc

@@ -18,7 +18,7 @@ namespace grpc_core {
 
 TRANSPORT_TEST(MetadataOnlyRequest) {
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -52,13 +52,13 @@ TRANSPORT_TEST(MetadataOnlyRequest) {
       [&](ValueOrFailure<absl::optional<MessageHandle>> msg) {
         EXPECT_TRUE(msg.ok());
         EXPECT_FALSE(msg.value().has_value());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
         return handler.PushServerInitialMetadata(std::move(md));
       },
       [&](StatusFlag result) mutable {
         EXPECT_TRUE(result.ok());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};
@@ -73,7 +73,7 @@ TRANSPORT_TEST(MetadataOnlyRequestServerAbortsAfterInitialMetadata) {
                   "rolling out soon, so leaving this disabled.";
 
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -105,13 +105,13 @@ TRANSPORT_TEST(MetadataOnlyRequestServerAbortsAfterInitialMetadata) {
             "/foo/bar");
         // Don't wait for end of stream for client->server messages, just
         // publish initial then trailing metadata.
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
         return handler.PushServerInitialMetadata(std::move(md));
       },
       [&](StatusFlag result) mutable {
         EXPECT_TRUE(result.ok());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};
@@ -126,7 +126,7 @@ TRANSPORT_TEST(MetadataOnlyRequestServerAbortsImmediately) {
                   "rolling out soon, so leaving this disabled.";
 
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -156,7 +156,7 @@ TRANSPORT_TEST(MetadataOnlyRequestServerAbortsImmediately) {
             "/foo/bar");
         // Don't wait for end of stream for client->server messages, just
         // and don't send initial metadata - just trailing metadata.
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};
@@ -166,7 +166,7 @@ TRANSPORT_TEST(MetadataOnlyRequestServerAbortsImmediately) {
 
 TRANSPORT_TEST(CanCreateCallThenAbandonIt) {
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   auto handler = TickUntilServerCall();
@@ -179,7 +179,7 @@ TRANSPORT_TEST(CanCreateCallThenAbandonIt) {
 
 TRANSPORT_TEST(UnaryRequest) {
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -237,7 +237,7 @@ TRANSPORT_TEST(UnaryRequest) {
       [&](ValueOrFailure<absl::optional<MessageHandle>> msg) {
         EXPECT_TRUE(msg.ok());
         EXPECT_FALSE(msg.value().has_value());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
         return handler.PushServerInitialMetadata(std::move(md));
       },
@@ -248,7 +248,7 @@ TRANSPORT_TEST(UnaryRequest) {
       },
       [&](StatusFlag result) mutable {
         EXPECT_TRUE(result.ok());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};
@@ -258,7 +258,7 @@ TRANSPORT_TEST(UnaryRequest) {
 
 TRANSPORT_TEST(UnaryRequestOmitCheckEndOfStream) {
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -306,7 +306,7 @@ TRANSPORT_TEST(UnaryRequestOmitCheckEndOfStream) {
         EXPECT_TRUE(msg.value().has_value());
         EXPECT_EQ(msg.value().value()->payload()->JoinIntoString(),
                   "hello world");
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
         return handler.PushServerInitialMetadata(std::move(md));
       },
@@ -317,7 +317,7 @@ TRANSPORT_TEST(UnaryRequestOmitCheckEndOfStream) {
       },
       [&](StatusFlag result) mutable {
         EXPECT_TRUE(result.ok());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};
@@ -327,7 +327,7 @@ TRANSPORT_TEST(UnaryRequestOmitCheckEndOfStream) {
 
 TRANSPORT_TEST(UnaryRequestWaitForServerInitialMetadataBeforeSendingPayload) {
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -371,7 +371,7 @@ TRANSPORT_TEST(UnaryRequestWaitForServerInitialMetadataBeforeSendingPayload) {
         EXPECT_TRUE(md.ok());
         EXPECT_EQ(md.value()->get_pointer(HttpPathMetadata())->as_string_view(),
                   "/foo/bar");
-        auto md_out = Arena::MakePooled<ServerMetadata>();
+        auto md_out = Arena::MakePooledForOverwrite<ServerMetadata>();
         md_out->Set(ContentTypeMetadata(),
                     ContentTypeMetadata::kApplicationGrpc);
         return handler.PushServerInitialMetadata(std::move(md_out));
@@ -395,7 +395,7 @@ TRANSPORT_TEST(UnaryRequestWaitForServerInitialMetadataBeforeSendingPayload) {
       },
       [&](StatusFlag result) mutable {
         EXPECT_TRUE(result.ok());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};
@@ -405,7 +405,7 @@ TRANSPORT_TEST(UnaryRequestWaitForServerInitialMetadataBeforeSendingPayload) {
 
 TRANSPORT_TEST(ClientStreamingRequest) {
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -462,7 +462,7 @@ TRANSPORT_TEST(ClientStreamingRequest) {
         EXPECT_TRUE(md.ok());
         EXPECT_EQ(md.value()->get_pointer(HttpPathMetadata())->as_string_view(),
                   "/foo/bar");
-        auto md_out = Arena::MakePooled<ServerMetadata>();
+        auto md_out = Arena::MakePooledForOverwrite<ServerMetadata>();
         md_out->Set(ContentTypeMetadata(),
                     ContentTypeMetadata::kApplicationGrpc);
         return handler.PushServerInitialMetadata(std::move(md_out));
@@ -509,7 +509,7 @@ TRANSPORT_TEST(ClientStreamingRequest) {
       [&](ValueOrFailure<absl::optional<MessageHandle>> msg) {
         EXPECT_TRUE(msg.ok());
         EXPECT_FALSE(msg.value().has_value());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};
@@ -519,7 +519,7 @@ TRANSPORT_TEST(ClientStreamingRequest) {
 
 TRANSPORT_TEST(ServerStreamingRequest) {
   SetServerCallDestination();
-  auto md = Arena::MakePooled<ClientMetadata>();
+  auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
   md->Set(HttpPathMetadata(), Slice::FromExternalString("/foo/bar"));
   auto initiator = CreateCall(std::move(md));
   SpawnTestSeq(
@@ -593,7 +593,7 @@ TRANSPORT_TEST(ServerStreamingRequest) {
         EXPECT_TRUE(md.ok());
         EXPECT_EQ(md.value()->get_pointer(HttpPathMetadata())->as_string_view(),
                   "/foo/bar");
-        auto md_out = Arena::MakePooled<ServerMetadata>();
+        auto md_out = Arena::MakePooledForOverwrite<ServerMetadata>();
         md_out->Set(ContentTypeMetadata(),
                     ContentTypeMetadata::kApplicationGrpc);
         return handler.PushServerInitialMetadata(std::move(md_out));
@@ -635,7 +635,7 @@ TRANSPORT_TEST(ServerStreamingRequest) {
       },
       [&](StatusFlag result) mutable {
         EXPECT_TRUE(result.ok());
-        auto md = Arena::MakePooled<ServerMetadata>();
+        auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
         md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
         handler.PushServerTrailingMetadata(std::move(md));
         return Empty{};

test/core/transport/test_suite/stress.cc

@@ -30,7 +30,7 @@ TRANSPORT_TEST(ManyUnaryRequests) {
   std::map<int, std::string> client_messages;
   std::map<int, std::string> server_messages;
   for (int i = 0; i < kNumRequests; i++) {
-    auto md = Arena::MakePooled<ClientMetadata>();
+    auto md = Arena::MakePooledForOverwrite<ClientMetadata>();
     md->Set(HttpPathMetadata(), Slice::FromCopiedString(std::to_string(i)));
     auto initiator = CreateCall(std::move(md));
     client_messages[i] = RandomMessage();
@@ -99,7 +99,7 @@ TRANSPORT_TEST(ManyUnaryRequests) {
         [handler](ValueOrFailure<absl::optional<MessageHandle>> msg) mutable {
           EXPECT_TRUE(msg.ok());
           EXPECT_FALSE(msg.value().has_value());
-          auto md = Arena::MakePooled<ServerMetadata>();
+          auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
           md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
           return handler.PushServerInitialMetadata(std::move(md));
         },
@@ -113,7 +113,7 @@ TRANSPORT_TEST(ManyUnaryRequests) {
         },
         [handler](StatusFlag result) mutable {
           EXPECT_TRUE(result.ok());
-          auto md = Arena::MakePooled<ServerMetadata>();
+          auto md = Arena::MakePooledForOverwrite<ServerMetadata>();
           md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED);
           handler.PushServerTrailingMetadata(std::move(md));
           return Empty{};

tools/run_tests/generated/tests.json

@@ -155,6 +155,28 @@
     ],
     "uses_polling": false
   },
+  {
+    "args": [
+      "--benchmark_min_time=0.001s"
+    ],
+    "benchmark": true,
+    "ci_platforms": [
+      "linux",
+      "posix"
+    ],
+    "cpu_cost": 1.0,
+    "exclude_configs": [],
+    "exclude_iomgrs": [],
+    "flaky": false,
+    "gtest": false,
+    "language": "c",
+    "name": "bm_metadata",
+    "platforms": [
+      "linux",
+      "posix"
+    ],
+    "uses_polling": false
+  },
   {
     "args": [
       "--benchmark_min_time=0.001s"