x

src/core/BUILD

@@ -1488,6 +1488,7 @@ grpc_cc_library(
         "context",
         "event_engine_memory_allocator",
         "memory_quota",
+        "resource_quota",
         "//:gpr",
     ],
 )

src/core/client_channel/client_channel.cc

@@ -752,9 +752,9 @@ CallInitiator ClientChannel::CreateCall(
   // Exit IDLE if needed.
   CheckConnectivityState(/*try_to_connect=*/true);
   // Create an initiator/unstarted-handler pair.
-  auto call = MakeCallPair(
-      std::move(client_initial_metadata), event_engine_.get(),
-      call_arena_allocator_->MakeArena(), call_arena_allocator_, nullptr);
+  auto call =
+      MakeCallPair(std::move(client_initial_metadata), event_engine_.get(),
+                   call_arena_allocator_->MakeArena(), nullptr);
   // Spawn a promise to wait for the resolver result.
   // This will eventually start the call.
   call.initiator.SpawnGuarded(

src/core/client_channel/subchannel_stream_client.cc

@@ -59,12 +59,13 @@ SubchannelStreamClient::SubchannelStreamClient(
       connected_subchannel_(std::move(connected_subchannel)),
       interested_parties_(interested_parties),
       tracer_(tracer),
-      call_allocator_(
+      call_allocator_(MakeRefCounted<CallArenaAllocator>(
           connected_subchannel_->args()
               .GetObject<ResourceQuota>()
               ->memory_quota()
               ->CreateMemoryAllocator(
-                  (tracer != nullptr) ? tracer : "SubchannelStreamClient")),
+                  (tracer != nullptr) ? tracer : "SubchannelStreamClient"),
+          1024)),
       event_handler_(std::move(event_handler)),
       retry_backoff_(
           BackOff::Options()
@@ -171,9 +172,7 @@ SubchannelStreamClient::CallState::CallState(
     grpc_pollset_set* interested_parties)
     : subchannel_stream_client_(std::move(health_check_client)),
       pollent_(grpc_polling_entity_create_from_pollset_set(interested_parties)),
-      arena_(Arena::Create(subchannel_stream_client_->connected_subchannel_
-                               ->GetInitialCallSizeEstimate(),
-                           &subchannel_stream_client_->call_allocator_)),
+      arena_(subchannel_stream_client_->call_allocator_->MakeArena()),
       payload_(context_) {}
 
 SubchannelStreamClient::CallState::~CallState() {

src/core/client_channel/subchannel_stream_client.h

@@ -146,7 +146,7 @@ class SubchannelStreamClient final
     RefCountedPtr<SubchannelStreamClient> subchannel_stream_client_;
     grpc_polling_entity pollent_;
 
-    ScopedArenaPtr arena_;
+    RefCountedPtr<Arena> arena_;
     CallCombiner call_combiner_;
     grpc_call_context_element context_[GRPC_CONTEXT_COUNT] = {};
 
@@ -201,7 +201,7 @@ class SubchannelStreamClient final
   RefCountedPtr<ConnectedSubchannel> connected_subchannel_;
   grpc_pollset_set* interested_parties_;  // Do not own.
   const char* tracer_;
-  MemoryAllocator call_allocator_;
+  RefCountedPtr<CallArenaAllocator> call_allocator_;
 
   Mutex mu_;
   std::unique_ptr<CallEventHandler> event_handler_ ABSL_GUARDED_BY(mu_);

src/core/ext/transport/chaotic_good/client/chaotic_good_connector.cc

@@ -333,7 +333,7 @@ void ChaoticGoodConnector::OnHandshakeDone(void* arg, grpc_error_handle error) {
                          status);
           }
         },
-        self->arena_.get(), self->event_engine_.get());
+        self->arena_, self->event_engine_.get());
     MutexLock lock(&self->mu_);
     if (!self->is_shutdown_) {
       self->connect_activity_ = std::move(activity);

src/core/ext/transport/chaotic_good/client/chaotic_good_connector.h

@@ -79,10 +79,7 @@ class ChaoticGoodConnector : public SubchannelConnector {
       RefCountedPtr<ChaoticGoodConnector> self);
   static void OnHandshakeDone(void* arg, grpc_error_handle error);
 
-  grpc_event_engine::experimental::MemoryAllocator memory_allocator_ =
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
-          "connect_activity");
-  ScopedArenaPtr arena_ = MakeScopedArena(1024, &memory_allocator_);
+  RefCountedPtr<Arena> arena_ = SimpleArenaAllocator()->MakeArena();
   Mutex mu_;
   Args args_;
   Result* result_ ABSL_GUARDED_BY(mu_);

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

@@ -146,10 +146,9 @@ auto ChaoticGoodClientTransport::TransportReadLoop(
                 frame_limits);
           } else {
             // Stream not found, skip the frame.
-            auto arena = MakeScopedArena(1024, &allocator_);
-            deserialize_status =
-                transport->DeserializeFrame(frame_header, std::move(buffers),
-                                            arena.get(), frame, frame_limits);
+            deserialize_status = transport->DeserializeFrame(
+                frame_header, std::move(buffers),
+                SimpleArenaAllocator()->MakeArena().get(), frame, frame_limits);
           }
           return If(
               deserialize_status.ok() && call_handler.has_value(),

src/core/ext/transport/chaotic_good/server/chaotic_good_server.cc

@@ -149,8 +149,7 @@ absl::Status ChaoticGoodServerListener::StartListening() {
 ChaoticGoodServerListener::ActiveConnection::ActiveConnection(
     RefCountedPtr<ChaoticGoodServerListener> listener,
     std::unique_ptr<EventEngine::Endpoint> endpoint)
-    : memory_allocator_(listener->memory_allocator_),
-      listener_(std::move(listener)) {
+    : listener_(std::move(listener)) {
   handshaking_state_ = MakeRefCounted<HandshakingState>(Ref());
   handshaking_state_->Start(std::move(endpoint));
 }
@@ -208,8 +207,7 @@ void ChaoticGoodServerListener::ActiveConnection::Done(
 
 ChaoticGoodServerListener::ActiveConnection::HandshakingState::HandshakingState(
     RefCountedPtr<ActiveConnection> connection)
-    : memory_allocator_(connection->memory_allocator_),
-      connection_(std::move(connection)),
+    : connection_(std::move(connection)),
       handshake_mgr_(MakeRefCounted<HandshakeManager>()) {}
 
 void ChaoticGoodServerListener::ActiveConnection::HandshakingState::Start(

src/core/ext/transport/chaotic_good/server/chaotic_good_server.h

@@ -83,7 +83,7 @@ class ChaoticGoodServerListener final : public Server::ListenerInterface {
     class HandshakingState : public RefCounted<HandshakingState> {
      public:
       explicit HandshakingState(RefCountedPtr<ActiveConnection> connection);
-      ~HandshakingState() override{};
+      ~HandshakingState() override {};
       void Start(std::unique_ptr<
                  grpc_event_engine::experimental::EventEngine::Endpoint>
                      endpoint);
@@ -106,8 +106,6 @@ class ChaoticGoodServerListener final : public Server::ListenerInterface {
 
       static void OnHandshakeDone(void* arg, grpc_error_handle error);
       Timestamp GetConnectionDeadline();
-      const std::shared_ptr<grpc_event_engine::experimental::MemoryAllocator>
-          memory_allocator_;
       const RefCountedPtr<ActiveConnection> connection_;
       const RefCountedPtr<HandshakeManager> handshake_mgr_;
     };
@@ -115,9 +113,7 @@ class ChaoticGoodServerListener final : public Server::ListenerInterface {
    private:
     void Done(absl::optional<absl::string_view> error = absl::nullopt);
     void NewConnectionID();
-    const std::shared_ptr<grpc_event_engine::experimental::MemoryAllocator>
-        memory_allocator_;
-    ScopedArenaPtr arena_ = MakeScopedArena(1024, memory_allocator_.get());
+    RefCountedPtr<Arena> arena_ = SimpleArenaAllocator()->MakeArena();
     const RefCountedPtr<ChaoticGoodServerListener> listener_;
     RefCountedPtr<HandshakingState> handshaking_state_;
     Mutex mu_;
@@ -161,11 +157,6 @@ class ChaoticGoodServerListener final : public Server::ListenerInterface {
   absl::AnyInvocable<std::string()> connection_id_generator_
       ABSL_GUARDED_BY(mu_);
   grpc_closure* on_destroy_done_ ABSL_GUARDED_BY(mu_) = nullptr;
-  std::shared_ptr<grpc_event_engine::experimental::MemoryAllocator>
-      memory_allocator_ =
-          std::make_shared<grpc_event_engine::experimental::MemoryAllocator>(
-              ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
-                  "server_connection"));
 };
 
 }  // namespace chaotic_good

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

@@ -235,15 +235,14 @@ auto ChaoticGoodServerTransport::DeserializeAndPushFragmentToNewCall(
     FrameHeader frame_header, BufferPair buffers,
     ChaoticGoodTransport& transport) {
   ClientFragmentFrame fragment_frame;
-  ScopedArenaPtr arena(call_arena_allocator_->MakeArena());
+  RefCountedPtr<Arena> arena(call_arena_allocator_->MakeArena());
   absl::Status status = transport.DeserializeFrame(
       frame_header, std::move(buffers), arena.get(), fragment_frame,
       FrameLimits{1024 * 1024 * 1024, aligned_bytes_ - 1});
   absl::optional<CallInitiator> call_initiator;
   if (status.ok()) {
-    auto call =
-        MakeCallPair(std::move(fragment_frame.headers), event_engine_.get(),
-                     arena.release(), call_arena_allocator_, nullptr);
+    auto call = MakeCallPair(std::move(fragment_frame.headers),
+                             event_engine_.get(), std::move(arena), nullptr);
     call_initiator.emplace(std::move(call.initiator));
     auto add_result = NewStream(frame_header.stream_id, *call_initiator);
     if (add_result.ok()) {

src/core/ext/transport/inproc/inproc_transport.cc

@@ -106,9 +106,9 @@ class InprocServerTransport final : public ServerTransport {
       case ConnectionState::kReady:
         break;
     }
-    auto* arena = call_arena_allocator_->MakeArena();
-    auto server_call = MakeCallPair(std::move(md), event_engine_.get(), arena,
-                                    call_arena_allocator_, nullptr);
+    auto server_call =
+        MakeCallPair(std::move(md), event_engine_.get(),
+                     call_arena_allocator_->MakeArena(), nullptr);
     unstarted_call_handler_->StartCall(std::move(server_call.handler));
     return std::move(server_call.initiator);
   }

src/core/lib/gprpp/ref_counted.h

@@ -355,12 +355,12 @@ class RefCounted : public Impl {
   // friend of this class.
   void Unref() const {
     if (GPR_UNLIKELY(refs_.Unref())) {
-      unref_behavior_(static_cast<const Child*>(this));
+      unref_behavior_(const_cast<Child*>(static_cast<const Child*>(this)));
     }
   }
   void Unref(const DebugLocation& location, const char* reason) const {
     if (GPR_UNLIKELY(refs_.Unref(location, reason))) {
-      unref_behavior_(static_cast<const Child*>(this));
+      unref_behavior_(const_cast<Child*>(static_cast<const Child*>(this)));
     }
   }
 

src/core/lib/promise/activity.h

@@ -289,6 +289,19 @@ class ContextHolder<std::unique_ptr<Context, Deleter>> {
   std::unique_ptr<Context, Deleter> value_;
 };
 
+template <typename Context>
+class ContextHolder<RefCountedPtr<Context>> {
+ public:
+  using ContextType = Context;
+
+  explicit ContextHolder(RefCountedPtr<Context> value)
+      : value_(std::move(value)) {}
+  Context* GetContext() { return value_.get(); }
+
+ private:
+  RefCountedPtr<Context> value_;
+};
+
 template <>
 class Context<Activity> {
  public:
@@ -296,19 +309,23 @@ class Context<Activity> {
 };
 
 template <typename HeldContext>
-using ContextTypeFromHeld = typename ContextHolder<HeldContext>::ContextType;
+using ContextTypeFromHeld = typename ContextHolder<
+    typename std::remove_reference<HeldContext>::type>::ContextType;
 
 template <typename... Contexts>
-class ActivityContexts : public ContextHolder<Contexts>... {
+class ActivityContexts
+    : public ContextHolder<typename std::remove_reference<Contexts>::type>... {
  public:
   explicit ActivityContexts(Contexts&&... contexts)
-      : ContextHolder<Contexts>(std::forward<Contexts>(contexts))... {}
+      : ContextHolder<typename std::remove_reference<Contexts>::type>(
+            std::forward<Contexts>(contexts))... {}
 
   class ScopedContext : public Context<ContextTypeFromHeld<Contexts>>... {
    public:
     explicit ScopedContext(ActivityContexts* contexts)
         : Context<ContextTypeFromHeld<Contexts>>(
-              static_cast<ContextHolder<Contexts>*>(contexts)
+              static_cast<ContextHolder<
+                  typename std::remove_reference<Contexts>::type>*>(contexts)
                   ->GetContext())... {
       // Silence `unused-but-set-parameter` in case of Contexts = {}
       (void)contexts;

src/core/lib/resource_quota/arena.cc

@@ -24,6 +24,7 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/port_platform.h>
 
+#include "src/core/lib/resource_quota/resource_quota.h"
 #include "src/core/util/alloc.h"
 
 namespace {
@@ -58,19 +59,19 @@ Arena::~Arena() {
 #endif
 }
 
-Arena* Arena::Create(size_t initial_size, MemoryAllocator* memory_allocator) {
-  return new (ArenaStorage(initial_size))
-      Arena(initial_size, 0, memory_allocator);
+RefCountedPtr<Arena> Arena::Create(size_t initial_size,
+                                   RefCountedPtr<ArenaFactory> arena_factory) {
+  return RefCountedPtr<Arena>(new (ArenaStorage(initial_size)) Arena(
+      initial_size, 0, std::move(arena_factory)));
 }
 
-std::pair<Arena*, void*> Arena::CreateWithAlloc(
-    size_t initial_size, size_t alloc_size, MemoryAllocator* memory_allocator) {
-  static constexpr size_t base_size =
-      GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(Arena));
-  auto* new_arena = new (ArenaStorage(initial_size))
-      Arena(initial_size, alloc_size, memory_allocator);
-  void* first_alloc = reinterpret_cast<char*>(new_arena) + base_size;
-  return std::make_pair(new_arena, first_alloc);
+Arena::Arena(size_t initial_size, size_t initial_alloc,
+             RefCountedPtr<ArenaFactory> arena_factory)
+    : total_used_(GPR_ROUND_UP_TO_ALIGNMENT_SIZE(initial_alloc)),
+      initial_zone_size_(initial_size),
+      arena_factory_(std::move(arena_factory)) {
+  arena_factory_->allocator().Reserve(
+      GPR_ROUND_UP_TO_ALIGNMENT_SIZE(initial_alloc));
 }
 
 void Arena::DestroyManagedNewObjects() {
@@ -88,7 +89,8 @@ void Arena::DestroyManagedNewObjects() {
 
 void Arena::Destroy() {
   DestroyManagedNewObjects();
-  memory_allocator_->Release(total_allocated_.load(std::memory_order_relaxed));
+  arena_factory_->allocator().Release(
+      total_allocated_.load(std::memory_order_relaxed));
   this->~Arena();
   gpr_free_aligned(this);
 }
@@ -102,7 +104,7 @@ void* Arena::AllocZone(size_t size) {
   static constexpr size_t zone_base_size =
       GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(Zone));
   size_t alloc_size = zone_base_size + size;
-  memory_allocator_->Reserve(alloc_size);
+  arena_factory_->allocator().Reserve(alloc_size);
   total_allocated_.fetch_add(alloc_size, std::memory_order_relaxed);
   Zone* z = new (gpr_malloc_aligned(alloc_size, GPR_MAX_ALIGNMENT)) Zone();
   auto* prev = last_zone_.load(std::memory_order_relaxed);
@@ -120,4 +122,27 @@ void Arena::ManagedNewObject::Link(std::atomic<ManagedNewObject*>* head) {
   }
 }
 
+RefCountedPtr<ArenaFactory> SimpleArenaAllocator(size_t initial_size) {
+  class Allocator : public ArenaFactory {
+   public:
+    explicit Allocator(size_t initial_size)
+        : ArenaFactory(
+              ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
+                  "simple-arena-allocator")),
+          initial_size_(initial_size) {}
+
+    RefCountedPtr<Arena> MakeArena() override {
+      return Arena::Create(initial_size_, Ref());
+    }
+
+    void FinalizeArena(Arena* arena) override {
+      // No-op.
+    }
+
+   private:
+    size_t initial_size_;
+  };
+  return MakeRefCounted<Allocator>(initial_size);
+}
+
 }  // namespace grpc_core

src/core/lib/resource_quota/arena.h

@@ -42,6 +42,8 @@
 
 namespace grpc_core {
 
+class Arena;
+
 namespace arena_detail {
 
 template <typename T, typename A, typename B>
@@ -54,19 +56,35 @@ struct IfArray<T[], A, B> {
   using Result = B;
 };
 
+struct UnrefDestroy {
+  void operator()(Arena* arena) const;
+};
+
 }  // namespace arena_detail
 
-class Arena {
+class ArenaFactory : public RefCounted<ArenaFactory> {
  public:
-  // Create an arena, with \a initial_size bytes in the first allocated buffer.
-  static Arena* Create(size_t initial_size, MemoryAllocator* memory_allocator);
+  virtual RefCountedPtr<Arena> MakeArena() = 0;
+  virtual void FinalizeArena(Arena* arena) = 0;
+
+  MemoryAllocator& allocator() { return allocator_; }
 
-  // Create an arena, with \a initial_size bytes in the first allocated buffer,
-  // and return both a void pointer to the returned arena and a void* with the
-  // first allocation.
-  static std::pair<Arena*, void*> CreateWithAlloc(
-      size_t initial_size, size_t alloc_size,
-      MemoryAllocator* memory_allocator);
+ protected:
+  explicit ArenaFactory(MemoryAllocator allocator)
+      : allocator_(std::move(allocator)) {}
+
+ private:
+  MemoryAllocator allocator_;
+};
+
+RefCountedPtr<ArenaFactory> SimpleArenaAllocator(size_t initial_size = 1024);
+
+class Arena : public RefCounted<Arena, NonPolymorphicRefCount,
+                                arena_detail::UnrefDestroy> {
+ public:
+  // Create an arena, with \a initial_size bytes in the first allocated buffer.
+  static RefCountedPtr<Arena> Create(size_t initial_size,
+                                     RefCountedPtr<ArenaFactory> arena_factory);
 
   // Destroy all `ManagedNew` allocated objects.
   // Allows safe destruction of these objects even if they need context held by
@@ -75,9 +93,6 @@ class Arena {
   // TODO(ctiller): eliminate ManagedNew.
   void DestroyManagedNewObjects();
 
-  // Destroy an arena.
-  void Destroy();
-
   // Return the total amount of memory allocated by this arena.
   size_t TotalUsedBytes() const {
     return total_used_.load(std::memory_order_relaxed);
@@ -201,6 +216,8 @@ class Arena {
   }
 
  private:
+  friend struct arena_detail::UnrefDestroy;
+
   struct Zone {
     Zone* prev;
   };
@@ -231,14 +248,12 @@ class Arena {
   //   where we wish to create an arena and then perform an immediate
   //   allocation.
   explicit Arena(size_t initial_size, size_t initial_alloc,
-                 MemoryAllocator* memory_allocator)
-      : total_used_(GPR_ROUND_UP_TO_ALIGNMENT_SIZE(initial_alloc)),
-        initial_zone_size_(initial_size),
-        memory_allocator_(memory_allocator) {}
+                 RefCountedPtr<ArenaFactory> arena_factory);
 
   ~Arena();
 
   void* AllocZone(size_t size);
+  void Destroy();
 
   // Keep track of the total used size. We use this in our call sizing
   // hysteresis.
@@ -252,24 +267,17 @@ class Arena {
   // last zone; the zone list is reverse-walked during arena destruction only.
   std::atomic<Zone*> last_zone_{nullptr};
   std::atomic<ManagedNewObject*> managed_new_head_{nullptr};
-  // The backing memory quota
-  MemoryAllocator* const memory_allocator_;
+  RefCountedPtr<ArenaFactory> arena_factory_;
 };
 
-// Smart pointer for arenas when the final size is not required.
-struct ScopedArenaDeleter {
-  void operator()(Arena* arena) { arena->Destroy(); }
-};
-using ScopedArenaPtr = std::unique_ptr<Arena, ScopedArenaDeleter>;
-inline ScopedArenaPtr MakeScopedArena(size_t initial_size,
-                                      MemoryAllocator* memory_allocator) {
-  return ScopedArenaPtr(Arena::Create(initial_size, memory_allocator));
-}
-
 // Arenas form a context for activities
 template <>
 struct ContextType<Arena> {};
 
+namespace arena_detail {
+inline void UnrefDestroy::operator()(Arena* arena) const { arena->Destroy(); }
+}  // namespace arena_detail
+
 }  // namespace grpc_core
 
 #endif  // GRPC_SRC_CORE_LIB_RESOURCE_QUOTA_ARENA_H

src/core/lib/surface/call.cc

@@ -380,15 +380,15 @@ void Call::Run() {
 
 class ChannelBasedCall : public Call {
  protected:
-  ChannelBasedCall(Arena* arena, bool is_client, Timestamp send_deadline,
-                   RefCountedPtr<Channel> channel)
+  ChannelBasedCall(RefCountedPtr<Arena> arena, bool is_client,
+                   Timestamp send_deadline, RefCountedPtr<Channel> channel)
       : Call(is_client, send_deadline, channel->event_engine()),
         arena_(arena),
         channel_(std::move(channel)) {
-    DCHECK_NE(arena_, nullptr);
+    DCHECK_NE(arena_.get(), nullptr);
   }
 
-  Arena* arena() final { return arena_; }
+  Arena* arena() final { return arena_.get(); }
 
   char* GetPeer() final {
     Slice peer_slice = GetPeerString();
@@ -415,18 +415,17 @@ class ChannelBasedCall : public Call {
 
   void DeleteThis() {
     RefCountedPtr<Channel> channel = std::move(channel_);
-    Arena* arena = arena_;
+    RefCountedPtr<Arena> arena = arena_;
     this->~ChannelBasedCall();
-    channel->DestroyArena(arena);
   }
 
   Channel* channel() const { return channel_.get(); }
 
   // Non-virtual arena accessor -- needed by PipeBasedCall
-  Arena* GetArena() { return arena_; }
+  Arena* GetArena() { return arena_.get(); }
 
  private:
-  Arena* const arena_;
+  const RefCountedPtr<Arena> arena_;
   RefCountedPtr<Channel> channel_;
 };
 
@@ -597,8 +596,9 @@ class FilterStackCall final : public ChannelBasedCall {
     void FinishBatch(grpc_error_handle error);
   };
 
-  FilterStackCall(Arena* arena, const grpc_call_create_args& args)
-      : ChannelBasedCall(arena, args.server_transport_data == nullptr,
+  FilterStackCall(RefCountedPtr<Arena> arena, const grpc_call_create_args& args)
+      : ChannelBasedCall(std::move(arena),
+                         args.server_transport_data == nullptr,
                          args.send_deadline, args.channel->Ref()),
         cq_(args.cq),
         stream_op_payload_(context_) {
@@ -732,7 +732,7 @@ grpc_error_handle FilterStackCall::Create(grpc_call_create_args* args,
       GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(FilterStackCall)) +
       channel_stack->call_stack_size;
 
-  Arena* arena = channel->CreateArena();
+  RefCountedPtr<Arena> arena = channel->call_arena_allocator()->MakeArena();
   call = new (arena->Alloc(call_alloc_size)) FilterStackCall(arena, *args);
   DCHECK(FromC(call->c_ptr()) == call);
   DCHECK(FromCallStack(call->call_stack()) == call);
@@ -771,7 +771,7 @@ grpc_error_handle FilterStackCall::Create(grpc_call_create_args* args,
         args->server->server_call_tracer_factory() != nullptr) {
       auto* server_call_tracer =
           args->server->server_call_tracer_factory()->CreateNewServerCallTracer(
-              arena, args->server->channel_args());
+              arena.get(), args->server->channel_args());
       if (server_call_tracer != nullptr) {
         // Note that we are setting both
         // GRPC_CONTEXT_CALL_TRACER_ANNOTATION_INTERFACE and
@@ -1906,10 +1906,12 @@ class BasicPromiseBasedCall : public ChannelBasedCall, public Party {
  public:
   using Call::arena;
 
-  BasicPromiseBasedCall(Arena* arena, uint32_t initial_external_refs,
+  BasicPromiseBasedCall(RefCountedPtr<Arena> arena,
+                        uint32_t initial_external_refs,
                         uint32_t initial_internal_refs,
                         const grpc_call_create_args& args)
-      : ChannelBasedCall(arena, args.server_transport_data == nullptr,
+      : ChannelBasedCall(std::move(arena),
+                         args.server_transport_data == nullptr,
                          args.send_deadline, args.channel->Ref()),
         Party(initial_internal_refs),
         external_refs_(initial_external_refs),
@@ -2067,7 +2069,7 @@ class BasicPromiseBasedCall : public ChannelBasedCall, public Party {
 
 class PromiseBasedCall : public BasicPromiseBasedCall {
  public:
-  PromiseBasedCall(Arena* arena, uint32_t initial_external_refs,
+  PromiseBasedCall(RefCountedPtr<Arena> arena, uint32_t initial_external_refs,
                    const grpc_call_create_args& args);
 
   bool Completed() final { return finished_.IsSet(); }
@@ -2349,16 +2351,17 @@ grpc_error_handle MakePromiseBasedCall(grpc_call_create_args* args,
                                        grpc_call** out_call) {
   Channel* channel = args->channel.get();
 
-  auto* arena = channel->CreateArena();
+  auto arena = channel->call_arena_allocator()->MakeArena();
   PromiseBasedCall* call = arena->New<T>(arena, args);
   *out_call = call->c_ptr();
   DCHECK(Call::FromC(*out_call) == call);
   return absl::OkStatus();
 }
 
-PromiseBasedCall::PromiseBasedCall(Arena* arena, uint32_t initial_external_refs,
+PromiseBasedCall::PromiseBasedCall(RefCountedPtr<Arena> arena,
+                                   uint32_t initial_external_refs,
                                    const grpc_call_create_args& args)
-    : BasicPromiseBasedCall(arena, initial_external_refs,
+    : BasicPromiseBasedCall(std::move(arena), initial_external_refs,
                             initial_external_refs != 0 ? 1 : 0, args) {}
 
 static void CToMetadata(grpc_metadata* metadata, size_t count,
@@ -2591,8 +2594,9 @@ void PublishMetadataArray(grpc_metadata_batch* md, grpc_metadata_array* array,
 #ifdef GRPC_EXPERIMENT_IS_INCLUDED_PROMISE_BASED_CLIENT_CALL
 class ClientPromiseBasedCall final : public PromiseBasedCall {
  public:
-  ClientPromiseBasedCall(Arena* arena, grpc_call_create_args* args)
-      : PromiseBasedCall(arena, 1, *args),
+  ClientPromiseBasedCall(RefCountedPtr<Arena> arena,
+                         grpc_call_create_args* args)
+      : PromiseBasedCall(std::move(arena), 1, *args),
         polling_entity_(
             args->cq != nullptr
                 ? grpc_polling_entity_create_from_pollset(

src/core/lib/surface/channel.cc

@@ -66,21 +66,11 @@ Channel::Channel(std::string target, const ChannelArgs& channel_args)
     : target_(std::move(target)),
       channelz_node_(channel_args.GetObjectRef<channelz::ChannelNode>()),
       compression_options_(CompressionOptionsFromChannelArgs(channel_args)),
-      call_size_estimator_(1024),
-      allocator_(channel_args.GetObject<ResourceQuota>()
-                     ->memory_quota()
-                     ->CreateMemoryOwner()) {}
-
-Arena* Channel::CreateArena() {
-  const size_t initial_size = call_size_estimator_.CallSizeEstimate();
-  global_stats().IncrementCallInitialSize(initial_size);
-  return Arena::Create(initial_size, &allocator_);
-}
-
-void Channel::DestroyArena(Arena* arena) {
-  call_size_estimator_.UpdateCallSizeEstimate(arena->TotalUsedBytes());
-  arena->Destroy();
-}
+      call_arena_allocator_(MakeRefCounted<CallArenaAllocator>(
+          channel_args.GetObject<ResourceQuota>()
+              ->memory_quota()
+              ->CreateMemoryOwner(),
+          1024)) {}
 
 Channel::RegisteredCall* Channel::RegisterCall(const char* method,
                                                const char* host) {

src/core/lib/surface/channel.h

@@ -124,12 +124,14 @@ class Channel : public RefCounted<Channel>,
   virtual void Ping(grpc_completion_queue* cq, void* tag) = 0;
 
   // TODO(roth): Remove these methods when LegacyChannel goes away.
-  Arena* CreateArena();
-  void DestroyArena(Arena* arena);
   virtual grpc_channel_stack* channel_stack() const { return nullptr; }
   virtual bool is_client() const { return true; }
   virtual bool is_promising() const { return true; }
 
+  CallArenaAllocator* call_arena_allocator() const {
+    return call_arena_allocator_.get();
+  }
+
  protected:
   Channel(std::string target, const ChannelArgs& channel_args);
 
@@ -138,15 +140,13 @@ class Channel : public RefCounted<Channel>,
   const RefCountedPtr<channelz::ChannelNode> channelz_node_;
   const grpc_compression_options compression_options_;
 
-  CallSizeEstimator call_size_estimator_;
-  MemoryAllocator allocator_;
-
   Mutex mu_;
   // The map key needs to be owned strings rather than unowned char*'s to
   // guarantee that it outlives calls on the core channel (which may outlast
   // the C++ or other wrapped language Channel that registered these calls).
   std::map<std::pair<std::string, std::string>, RegisteredCall>
       registration_table_ ABSL_GUARDED_BY(mu_);
+  const RefCountedPtr<CallArenaAllocator> call_arena_allocator_;
 };
 
 }  // namespace grpc_core

src/core/lib/transport/call_arena_allocator.h

@@ -28,7 +28,7 @@
 
 namespace grpc_core {
 
-class CallSizeEstimator {
+class CallSizeEstimator final {
  public:
   explicit CallSizeEstimator(size_t initial_estimate)
       : call_size_estimate_(initial_estimate) {}
@@ -52,19 +52,21 @@ class CallSizeEstimator {
   std::atomic<size_t> call_size_estimate_;
 };
 
-class CallArenaAllocator : public RefCounted<CallArenaAllocator> {
+class CallArenaAllocator final : public ArenaFactory {
  public:
   CallArenaAllocator(MemoryAllocator allocator, size_t initial_size)
-      : allocator_(std::move(allocator)), call_size_estimator_(initial_size) {}
+      : ArenaFactory(std::move(allocator)),
+        call_size_estimator_(initial_size) {}
 
-  Arena* MakeArena() {
-    return Arena::Create(call_size_estimator_.CallSizeEstimate(), &allocator_);
+  RefCountedPtr<Arena> MakeArena() override {
+    return Arena::Create(call_size_estimator_.CallSizeEstimate(), Ref());
   }
 
-  void Destroy(Arena* arena) { arena->Destroy(); }
+  void FinalizeArena(Arena* arena) override {
+    call_size_estimator_.UpdateCallSizeEstimate(arena->TotalUsedBytes());
+  }
 
  private:
-  MemoryAllocator allocator_;
   CallSizeEstimator call_size_estimator_;
 };
 

src/core/lib/transport/call_spine.cc

@@ -94,12 +94,11 @@ void ForwardCall(CallHandler call_handler, CallInitiator call_initiator) {
 
 CallInitiatorAndHandler MakeCallPair(
     ClientMetadataHandle client_initial_metadata,
-    grpc_event_engine::experimental::EventEngine* event_engine, Arena* arena,
-    RefCountedPtr<CallArenaAllocator> call_arena_allocator_if_arena_is_owned,
-    grpc_call_context_element* legacy_context) {
-  auto spine = CallSpine::Create(
-      std::move(client_initial_metadata), event_engine, arena,
-      std::move(call_arena_allocator_if_arena_is_owned), legacy_context);
+    grpc_event_engine::experimental::EventEngine* event_engine,
+    RefCountedPtr<Arena> arena, grpc_call_context_element* legacy_context) {
+  auto spine =
+      CallSpine::Create(std::move(client_initial_metadata), event_engine,
+                        std::move(arena), legacy_context);
   return {CallInitiator(spine), UnstartedCallHandler(spine)};
 }
 

src/core/lib/transport/call_spine.h

@@ -262,12 +262,11 @@ class CallSpine final : public CallSpineInterface, public Party {
  public:
   static RefCountedPtr<CallSpine> Create(
       ClientMetadataHandle client_initial_metadata,
-      grpc_event_engine::experimental::EventEngine* event_engine, Arena* arena,
-      RefCountedPtr<CallArenaAllocator> call_arena_allocator_if_arena_is_owned,
-      grpc_call_context_element* legacy_context) {
-    return RefCountedPtr<CallSpine>(arena->New<CallSpine>(
-        std::move(client_initial_metadata), event_engine, arena,
-        std::move(call_arena_allocator_if_arena_is_owned), legacy_context));
+      grpc_event_engine::experimental::EventEngine* event_engine,
+      RefCountedPtr<Arena> arena, grpc_call_context_element* legacy_context) {
+    return RefCountedPtr<CallSpine>(
+        arena->New<CallSpine>(std::move(client_initial_metadata), event_engine,
+                              std::move(arena), legacy_context));
   }
 
   ~CallSpine() override {
@@ -283,7 +282,7 @@ class CallSpine final : public CallSpineInterface, public Party {
 
   Party& party() override { return *this; }
 
-  Arena* arena() override { return arena_; }
+  Arena* arena() override { return arena_.get(); }
 
   void IncrementRefCount() override { Party::IncrementRefCount(); }
 
@@ -367,15 +366,12 @@ class CallSpine final : public CallSpineInterface, public Party {
   friend class Arena;
   CallSpine(ClientMetadataHandle client_initial_metadata,
             grpc_event_engine::experimental::EventEngine* event_engine,
-            Arena* arena,
-            RefCountedPtr<CallArenaAllocator> call_arena_allocator,
+            RefCountedPtr<Arena> arena,
             grpc_call_context_element* legacy_context)
       : Party(1),
+        arena_(std::move(arena)),
         call_filters_(std::move(client_initial_metadata)),
-        arena_(arena),
-        event_engine_(event_engine),
-        call_arena_allocator_if_arena_is_owned_(
-            std::move(call_arena_allocator)) {
+        event_engine_(event_engine) {
     if (legacy_context == nullptr) {
       legacy_context_ = static_cast<grpc_call_context_element*>(
           arena->Alloc(sizeof(grpc_call_context_element) * GRPC_CONTEXT_COUNT));
@@ -397,7 +393,7 @@ class CallSpine final : public CallSpineInterface, public Party {
    public:
     explicit ScopedContext(CallSpine* spine)
         : ScopedActivity(spine),
-          Context<Arena>(spine->arena_),
+          Context<Arena>(spine->arena_.get()),
           Context<grpc_event_engine::experimental::EventEngine>(
               spine->event_engine()),
           Context<grpc_call_context_element>(spine->legacy_context_) {}
@@ -409,29 +405,22 @@ class CallSpine final : public CallSpineInterface, public Party {
   }
 
   void PartyOver() override {
-    Arena* a = arena_;
-    RefCountedPtr<CallArenaAllocator> call_arena_allocator_if_arena_is_owned =
-        std::move(call_arena_allocator_if_arena_is_owned_);
     {
       ScopedContext context(this);
       CancelRemainingParticipants();
-      a->DestroyManagedNewObjects();
+      arena_->DestroyManagedNewObjects();
     }
     this->~CallSpine();
-    if (call_arena_allocator_if_arena_is_owned != nullptr) {
-      call_arena_allocator_if_arena_is_owned->Destroy(a);
-    }
   }
 
+  const RefCountedPtr<Arena> arena_;
   // Call filters/pipes part of the spine
   CallFilters call_filters_;
-  Arena* const arena_;
   // Event engine associated with this call
   grpc_event_engine::experimental::EventEngine* const event_engine_;
   // Legacy context
   // TODO(ctiller): remove
   grpc_call_context_element* legacy_context_;
-  RefCountedPtr<CallArenaAllocator> call_arena_allocator_if_arena_is_owned_;
   bool legacy_context_is_owned_;
 };
 
@@ -623,9 +612,8 @@ struct CallInitiatorAndHandler {
 
 CallInitiatorAndHandler MakeCallPair(
     ClientMetadataHandle client_initial_metadata,
-    grpc_event_engine::experimental::EventEngine* event_engine, Arena* arena,
-    RefCountedPtr<CallArenaAllocator> call_arena_allocator_if_arena_is_owned,
-    grpc_call_context_element* legacy_context);
+    grpc_event_engine::experimental::EventEngine* event_engine,
+    RefCountedPtr<Arena> arena, grpc_call_context_element* legacy_context);
 
 template <typename CallHalf>
 auto OutgoingMessages(CallHalf h) {

src/core/lib/transport/interception_chain.cc

@@ -40,7 +40,7 @@ CallInitiator HijackedCall::MakeCall() {
 CallInitiator HijackedCall::MakeCallWithMetadata(
     ClientMetadataHandle metadata) {
   auto call = MakeCallPair(std::move(metadata), call_handler_.event_engine(),
-                           call_handler_.arena(), nullptr,
+                           call_handler_.arena()->Ref(),
                            call_handler_.legacy_context());
   destination_->StartCall(std::move(call.handler));
   return std::move(call.initiator);

test/core/call/yodel/yodel_test.h

@@ -351,10 +351,9 @@ class YodelTest : public ::testing::Test {
   }
 
   auto MakeCall(ClientMetadataHandle client_initial_metadata) {
-    auto* arena = state_->call_arena_allocator->MakeArena();
     return MakeCallPair(std::move(client_initial_metadata),
-                        state_->event_engine.get(), arena,
-                        state_->call_arena_allocator, nullptr);
+                        state_->event_engine.get(),
+                        state_->call_arena_allocator->MakeArena(), nullptr);
   }
 
   void WaitForAllPendingWork();

test/core/channel/call_finalization_test.cc

@@ -30,9 +30,7 @@
 namespace grpc_core {
 
 TEST(CallFinalizationTest, Works) {
-  auto memory_allocator = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   std::string evidence;
   TestContext<Arena> context(arena.get());
   CallFinalization finalization;

test/core/channel/call_tracer_test.cc

@@ -37,20 +37,7 @@ namespace {
 
 class CallTracerTest : public ::testing::Test {
  protected:
-  void SetUp() override {
-    memory_allocator_ = new MemoryAllocator(
-        ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
-            "test"));
-    arena_ = Arena::Create(1024, memory_allocator_);
-  }
-
-  void TearDown() override {
-    arena_->Destroy();
-    delete memory_allocator_;
-  }
-
-  MemoryAllocator* memory_allocator_ = nullptr;
-  Arena* arena_ = nullptr;
+  RefCountedPtr<Arena> arena_ = SimpleArenaAllocator()->MakeArena();
   grpc_call_context_element context_[GRPC_CONTEXT_COUNT] = {};
   std::vector<std::string> annotation_logger_;
 };
@@ -65,7 +52,7 @@ TEST_F(CallTracerTest, BasicClientCallTracer) {
 }
 
 TEST_F(CallTracerTest, MultipleClientCallTracers) {
-  promise_detail::Context<Arena> arena_ctx(arena_);
+  promise_detail::Context<Arena> arena_ctx(arena_.get());
   FakeClientCallTracer client_call_tracer1(&annotation_logger_);
   FakeClientCallTracer client_call_tracer2(&annotation_logger_);
   FakeClientCallTracer client_call_tracer3(&annotation_logger_);
@@ -80,7 +67,7 @@ TEST_F(CallTracerTest, MultipleClientCallTracers) {
 }
 
 TEST_F(CallTracerTest, MultipleClientCallAttemptTracers) {
-  promise_detail::Context<Arena> arena_ctx(arena_);
+  promise_detail::Context<Arena> arena_ctx(arena_.get());
   FakeClientCallTracer client_call_tracer1(&annotation_logger_);
   FakeClientCallTracer client_call_tracer2(&annotation_logger_);
   FakeClientCallTracer client_call_tracer3(&annotation_logger_);
@@ -110,7 +97,7 @@ TEST_F(CallTracerTest, BasicServerCallTracerTest) {
 }
 
 TEST_F(CallTracerTest, MultipleServerCallTracers) {
-  promise_detail::Context<Arena> arena_ctx(arena_);
+  promise_detail::Context<Arena> arena_ctx(arena_.get());
   FakeServerCallTracer server_call_tracer1(&annotation_logger_);
   FakeServerCallTracer server_call_tracer2(&annotation_logger_);
   FakeServerCallTracer server_call_tracer3(&annotation_logger_);

test/core/client_channel/load_balanced_call_destination_test.cc

@@ -49,9 +49,9 @@ class LoadBalancedCallDestinationTest : public YodelTest {
 
   CallInitiatorAndHandler MakeCall(
       ClientMetadataHandle client_initial_metadata) {
-    return MakeCallPair(
-        std::move(client_initial_metadata), event_engine().get(),
-        call_arena_allocator_->MakeArena(), call_arena_allocator_, nullptr);
+    return MakeCallPair(std::move(client_initial_metadata),
+                        event_engine().get(),
+                        call_arena_allocator_->MakeArena(), nullptr);
   }
 
   CallHandler TickUntilCallStarted() {

test/core/filters/filter_test.cc

@@ -83,8 +83,7 @@ class FilterTestBase::Call::Impl
 
   Call* const call_;
   std::shared_ptr<Channel::Impl> const channel_;
-  ScopedArenaPtr arena_{MakeScopedArena(channel_->initial_arena_size,
-                                        &channel_->memory_allocator)};
+  RefCountedPtr<Arena> arena_ = channel_->arena_factory->MakeArena();
   bool run_call_finalization_ = false;
   CallFinalization call_finalization_;
   absl::optional<ArenaPromise<ServerMetadataHandle>> promise_;

test/core/filters/filter_test.h

@@ -101,19 +101,12 @@ class FilterTestBase : public ::testing::Test {
     struct Impl {
       Impl(std::unique_ptr<ChannelFilter> filter, FilterTestBase* test)
           : filter(std::move(filter)), test(test) {}
-      size_t initial_arena_size = 1024;
-      MemoryAllocator memory_allocator =
-          ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
-              "test");
+      RefCountedPtr<ArenaFactory> arena_factory = SimpleArenaAllocator();
       std::unique_ptr<ChannelFilter> filter;
       FilterTestBase* const test;
     };
 
    public:
-    void set_initial_arena_size(size_t size) {
-      impl_->initial_arena_size = size;
-    }
-
     Call MakeCall();
 
    protected:

test/core/gprpp/chunked_vector_fuzzer.cc

@@ -169,7 +169,7 @@ class Fuzzer {
 
   MemoryAllocator memory_allocator_ = MemoryAllocator(
       ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-  ScopedArenaPtr arena_ = MakeScopedArena(128, &memory_allocator_);
+  RefCountedPtr<Arena> arena_ = SimpleArenaAllocator(128)->MakeArena();
   std::map<int, Comparison> vectors_;
 };
 }  // namespace grpc_core

test/core/gprpp/chunked_vector_test.cc

@@ -31,20 +31,14 @@ namespace testing {
 static constexpr size_t kInitialArenaSize = 1024;
 static constexpr size_t kChunkSize = 3;
 
-class ChunkedVectorTest : public ::testing::Test {
- protected:
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-};
-
-TEST_F(ChunkedVectorTest, Noop) {
-  auto arena = MakeScopedArena(kInitialArenaSize, &memory_allocator_);
+TEST(ChunkedVectorTest, Noop) {
+  auto arena = SimpleArenaAllocator(kInitialArenaSize)->MakeArena();
   ChunkedVector<int, kChunkSize> v(arena.get());
   EXPECT_EQ(0, v.size());
 }
 
-TEST_F(ChunkedVectorTest, Stack) {
-  auto arena = MakeScopedArena(kInitialArenaSize, &memory_allocator_);
+TEST(ChunkedVectorTest, Stack) {
+  auto arena = SimpleArenaAllocator(kInitialArenaSize)->MakeArena();
   ChunkedVector<int, kChunkSize> v(arena.get());
 
   // Populate 2 chunks of memory, and 2/3 of a final chunk.
@@ -85,8 +79,8 @@ TEST_F(ChunkedVectorTest, Stack) {
   EXPECT_EQ(0, v.size());
 }
 
-TEST_F(ChunkedVectorTest, Iterate) {
-  auto arena = MakeScopedArena(kInitialArenaSize, &memory_allocator_);
+TEST(ChunkedVectorTest, Iterate) {
+  auto arena = SimpleArenaAllocator(kInitialArenaSize)->MakeArena();
   ChunkedVector<int, kChunkSize> v(arena.get());
   v.EmplaceBack(1);
   v.EmplaceBack(2);
@@ -117,8 +111,8 @@ TEST_F(ChunkedVectorTest, Iterate) {
   EXPECT_EQ(v.end(), it);
 }
 
-TEST_F(ChunkedVectorTest, ConstIterate) {
-  auto arena = MakeScopedArena(kInitialArenaSize, &memory_allocator_);
+TEST(ChunkedVectorTest, ConstIterate) {
+  auto arena = SimpleArenaAllocator(kInitialArenaSize)->MakeArena();
   ChunkedVector<int, kChunkSize> v(arena.get());
   v.EmplaceBack(1);
   v.EmplaceBack(2);
@@ -149,8 +143,8 @@ TEST_F(ChunkedVectorTest, ConstIterate) {
   EXPECT_EQ(v.cend(), it);
 }
 
-TEST_F(ChunkedVectorTest, Clear) {
-  auto arena = MakeScopedArena(kInitialArenaSize, &memory_allocator_);
+TEST(ChunkedVectorTest, Clear) {
+  auto arena = SimpleArenaAllocator(kInitialArenaSize)->MakeArena();
   ChunkedVector<int, kChunkSize> v(arena.get());
   v.EmplaceBack(1);
   EXPECT_EQ(v.size(), 1);
@@ -159,8 +153,8 @@ TEST_F(ChunkedVectorTest, Clear) {
   EXPECT_EQ(v.begin(), v.end());
 }
 
-TEST_F(ChunkedVectorTest, RemoveIf) {
-  auto arena = MakeScopedArena(kInitialArenaSize, &memory_allocator_);
+TEST(ChunkedVectorTest, RemoveIf) {
+  auto arena = SimpleArenaAllocator(kInitialArenaSize)->MakeArena();
   ChunkedVector<int, kChunkSize> v(arena.get());
   v.EmplaceBack(1);
   v.SetEnd(std::remove_if(v.begin(), v.end(), [](int i) { return i == 1; }));

test/core/promise/arena_promise_test.cc

@@ -30,22 +30,16 @@
 
 namespace grpc_core {
 
-class ArenaPromiseTest : public ::testing::Test {
- protected:
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-};
-
-TEST_F(ArenaPromiseTest, DefaultInitializationYieldsNoValue) {
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+TEST(ArenaPromiseTest, DefaultInitializationYieldsNoValue) {
+  auto arena = SimpleArenaAllocator()->MakeArena();
   TestContext<Arena> context(arena.get());
   ArenaPromise<int> p;
   EXPECT_FALSE(p.has_value());
 }
 
-TEST_F(ArenaPromiseTest, AllocatedWorks) {
+TEST(ArenaPromiseTest, AllocatedWorks) {
   ExecCtx exec_ctx;
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   TestContext<Arena> context(arena.get());
   int x = 42;
   ArenaPromise<int> p([x] { return Poll<int>(x); });
@@ -55,9 +49,9 @@ TEST_F(ArenaPromiseTest, AllocatedWorks) {
   EXPECT_EQ(p(), Poll<int>(43));
 }
 
-TEST_F(ArenaPromiseTest, DestructionWorks) {
+TEST(ArenaPromiseTest, DestructionWorks) {
   ExecCtx exec_ctx;
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   TestContext<Arena> context(arena.get());
   auto x = std::make_shared<int>(42);
   auto p = ArenaPromise<int>([x] { return Poll<int>(*x); });
@@ -65,18 +59,18 @@ TEST_F(ArenaPromiseTest, DestructionWorks) {
   EXPECT_EQ(q(), Poll<int>(42));
 }
 
-TEST_F(ArenaPromiseTest, MoveAssignmentWorks) {
+TEST(ArenaPromiseTest, MoveAssignmentWorks) {
   ExecCtx exec_ctx;
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   TestContext<Arena> context(arena.get());
   auto x = std::make_shared<int>(42);
   auto p = ArenaPromise<int>([x] { return Poll<int>(*x); });
   p = ArenaPromise<int>();
 }
 
-TEST_F(ArenaPromiseTest, AllocatedUniquePtrWorks) {
+TEST(ArenaPromiseTest, AllocatedUniquePtrWorks) {
   ExecCtx exec_ctx;
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   TestContext<Arena> context(arena.get());
   std::array<int, 5> garbage = {0, 1, 2, 3, 4};
   auto freer = [garbage](int* p) { free(p + garbage[0]); };

test/core/promise/for_each_test.cc

@@ -40,13 +40,7 @@ using testing::StrictMock;
 
 namespace grpc_core {
 
-class ForEachTest : public ::testing::Test {
- protected:
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-};
-
-TEST_F(ForEachTest, SendThriceWithPipe) {
+TEST(ForEachTest, SendThriceWithPipe) {
   int num_received = 0;
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
@@ -76,12 +70,12 @@ TEST_F(ForEachTest, SendThriceWithPipe) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
   Mock::VerifyAndClearExpectations(&on_done);
   EXPECT_EQ(num_received, 3);
 }
 
-TEST_F(ForEachTest, SendThriceWithInterActivityPipe) {
+TEST(ForEachTest, SendThriceWithInterActivityPipe) {
   int num_received = 0;
   StrictMock<MockFunction<void(absl::Status)>> on_done_sender;
   StrictMock<MockFunction<void(absl::Status)>> on_done_receiver;
@@ -148,7 +142,7 @@ class MoveableUntilPolled {
   int polls_ = 0;
 };
 
-TEST_F(ForEachTest, NoMoveAfterPoll) {
+TEST(ForEachTest, NoMoveAfterPoll) {
   int num_received = 0;
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
@@ -179,12 +173,12 @@ TEST_F(ForEachTest, NoMoveAfterPoll) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
   Mock::VerifyAndClearExpectations(&on_done);
   EXPECT_EQ(num_received, 1);
 }
 
-TEST_F(ForEachTest, NextResultHeldThroughCallback) {
+TEST(ForEachTest, NextResultHeldThroughCallback) {
   int num_received = 0;
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
@@ -230,7 +224,7 @@ TEST_F(ForEachTest, NextResultHeldThroughCallback) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
   Mock::VerifyAndClearExpectations(&on_done);
   EXPECT_EQ(num_received, 1);
 }

test/core/promise/interceptor_list_test.cc

@@ -31,9 +31,7 @@ namespace {
 
 class InterceptorListTest : public ::testing::Test {
  protected:
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-  ScopedArenaPtr arena_ = MakeScopedArena(1024, &memory_allocator_);
+  RefCountedPtr<Arena> arena_ = SimpleArenaAllocator()->MakeArena();
   TestContext<Arena> arena_ctx_{arena_.get()};
 };
 

test/core/promise/map_pipe_test.cc

@@ -58,13 +58,7 @@ class Delayed {
   T x_;
 };
 
-class MapPipeTest : public ::testing::Test {
- protected:
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-};
-
-TEST_F(MapPipeTest, SendThriceWithPipeInterceptingReceive) {
+TEST(MapPipeTest, SendThriceWithPipeInterceptingReceive) {
   int num_received = 0;
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
@@ -99,12 +93,12 @@ TEST_F(MapPipeTest, SendThriceWithPipeInterceptingReceive) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
   Mock::VerifyAndClearExpectations(&on_done);
   EXPECT_EQ(num_received, 3);
 }
 
-TEST_F(MapPipeTest, SendThriceWithPipeInterceptingSend) {
+TEST(MapPipeTest, SendThriceWithPipeInterceptingSend) {
   int num_received = 0;
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
@@ -139,7 +133,7 @@ TEST_F(MapPipeTest, SendThriceWithPipeInterceptingSend) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
   Mock::VerifyAndClearExpectations(&on_done);
   EXPECT_EQ(num_received, 3);
 }

test/core/promise/pipe_test.cc

@@ -41,13 +41,7 @@ using testing::StrictMock;
 
 namespace grpc_core {
 
-class PipeTest : public ::testing::Test {
- protected:
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-};
-
-TEST_F(PipeTest, CanSendAndReceive) {
+TEST(PipeTest, CanSendAndReceive) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -68,10 +62,10 @@ TEST_F(PipeTest, CanSendAndReceive) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanInterceptAndMapAtSender) {
+TEST(PipeTest, CanInterceptAndMapAtSender) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -93,10 +87,10 @@ TEST_F(PipeTest, CanInterceptAndMapAtSender) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanInterceptAndMapAtReceiver) {
+TEST(PipeTest, CanInterceptAndMapAtReceiver) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -118,10 +112,10 @@ TEST_F(PipeTest, CanInterceptAndMapAtReceiver) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, InterceptionOrderingIsCorrect) {
+TEST(PipeTest, InterceptionOrderingIsCorrect) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -154,10 +148,10 @@ TEST_F(PipeTest, InterceptionOrderingIsCorrect) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanReceiveAndSend) {
+TEST(PipeTest, CanReceiveAndSend) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -178,10 +172,10 @@ TEST_F(PipeTest, CanReceiveAndSend) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanSeeClosedOnSend) {
+TEST(PipeTest, CanSeeClosedOnSend) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -208,10 +202,10 @@ TEST_F(PipeTest, CanSeeClosedOnSend) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanSeeClosedOnReceive) {
+TEST(PipeTest, CanSeeClosedOnReceive) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -240,10 +234,10 @@ TEST_F(PipeTest, CanSeeClosedOnReceive) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanCloseSend) {
+TEST(PipeTest, CanCloseSend) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -270,10 +264,10 @@ TEST_F(PipeTest, CanCloseSend) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanCloseWithErrorSend) {
+TEST(PipeTest, CanCloseWithErrorSend) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -300,10 +294,10 @@ TEST_F(PipeTest, CanCloseWithErrorSend) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanCloseWithErrorRecv) {
+TEST(PipeTest, CanCloseWithErrorRecv) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -330,10 +324,10 @@ TEST_F(PipeTest, CanCloseWithErrorRecv) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanCloseSendWithInterceptor) {
+TEST(PipeTest, CanCloseSendWithInterceptor) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -361,10 +355,10 @@ TEST_F(PipeTest, CanCloseSendWithInterceptor) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanCancelSendWithInterceptor) {
+TEST(PipeTest, CanCancelSendWithInterceptor) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -389,10 +383,10 @@ TEST_F(PipeTest, CanCancelSendWithInterceptor) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
-TEST_F(PipeTest, CanFlowControlThroughManyStages) {
+TEST(PipeTest, CanFlowControlThroughManyStages) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   auto done = std::make_shared<bool>(false);
@@ -437,11 +431,11 @@ TEST_F(PipeTest, CanFlowControlThroughManyStages) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
   ASSERT_TRUE(*done);
 }
 
-TEST_F(PipeTest, AwaitClosedWorks) {
+TEST(PipeTest, AwaitClosedWorks) {
   StrictMock<MockFunction<void(absl::Status)>> on_done;
   EXPECT_CALL(on_done, Call(absl::OkStatus()));
   MakeActivity(
@@ -468,7 +462,7 @@ TEST_F(PipeTest, AwaitClosedWorks) {
       },
       NoWakeupScheduler(),
       [&on_done](absl::Status status) { on_done.Call(std::move(status)); },
-      MakeScopedArena(1024, &memory_allocator_));
+      SimpleArenaAllocator()->MakeArena());
 }
 
 class FakeActivity final : public Activity {
@@ -483,7 +477,7 @@ class FakeActivity final : public Activity {
   }
 };
 
-TEST_F(PipeTest, PollAckWaitsForReadyClosed) {
+TEST(PipeTest, PollAckWaitsForReadyClosed) {
   FakeActivity().Run([]() {
     pipe_detail::Center<int> c;
     int i = 1;

test/core/resource_quota/arena_test.cc

@@ -57,9 +57,8 @@ class AllocTest : public ::testing::TestWithParam<AllocShape> {};
 
 TEST_P(AllocTest, Works) {
   ExecCtx exec_ctx;
-  MemoryAllocator memory_allocator = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-  Arena* a = Arena::Create(GetParam().initial_size, &memory_allocator);
+  RefCountedPtr<Arena> a =
+      SimpleArenaAllocator(GetParam().initial_size)->MakeArena();
   std::vector<void*> allocated;
   for (auto alloc : GetParam().allocs) {
     void* p = a->Alloc(alloc);
@@ -73,7 +72,6 @@ TEST_P(AllocTest, Works) {
     memset(p, 1, alloc);
     allocated.push_back(p);
   }
-  a->Destroy();
 }
 
 INSTANTIATE_TEST_SUITE_P(
@@ -92,33 +90,23 @@ size_t concurrent_test_iterations() {
 
 typedef struct {
   gpr_event ev_start;
-  Arena* arena;
+  RefCountedPtr<Arena> arena;
 } concurrent_test_args;
 
-class ArenaTest : public ::testing::Test {
- protected:
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-};
-
-TEST_F(ArenaTest, NoOp) {
-  ExecCtx exec_ctx;
-  Arena::Create(1, &memory_allocator_)->Destroy();
-}
+TEST(ArenaTest, NoOp) { SimpleArenaAllocator()->MakeArena(); }
 
-TEST_F(ArenaTest, ManagedNew) {
+TEST(ArenaTest, ManagedNew) {
   ExecCtx exec_ctx;
-  Arena* arena = Arena::Create(1, &memory_allocator_);
+  auto arena = SimpleArenaAllocator(1)->MakeArena();
   for (int i = 0; i < 100; i++) {
     arena->ManagedNew<std::unique_ptr<int>>(std::make_unique<int>(i));
   }
-  arena->Destroy();
 }
 
-TEST_F(ArenaTest, ConcurrentAlloc) {
+TEST(ArenaTest, ConcurrentAlloc) {
   concurrent_test_args args;
   gpr_event_init(&args.ev_start);
-  args.arena = Arena::Create(1024, &memory_allocator_);
+  args.arena = SimpleArenaAllocator()->MakeArena();
 
   Thread thds[CONCURRENT_TEST_THREADS];
 
@@ -141,14 +129,12 @@ TEST_F(ArenaTest, ConcurrentAlloc) {
   for (auto& th : thds) {
     th.Join();
   }
-
-  args.arena->Destroy();
 }
 
-TEST_F(ArenaTest, ConcurrentManagedNew) {
+TEST(ArenaTest, ConcurrentManagedNew) {
   concurrent_test_args args;
   gpr_event_init(&args.ev_start);
-  args.arena = Arena::Create(1024, &memory_allocator_);
+  args.arena = SimpleArenaAllocator()->MakeArena();
 
   Thread thds[CONCURRENT_TEST_THREADS];
 
@@ -172,8 +158,6 @@ TEST_F(ArenaTest, ConcurrentManagedNew) {
   for (auto& th : thds) {
     th.Join();
   }
-
-  args.arena->Destroy();
 }
 
 template <typename Int>
@@ -192,12 +176,12 @@ bool IsScribbled(Int* ints, int n, int offset) {
 }
 
 #ifndef GRPC_ARENA_POOLED_ALLOCATIONS_USE_MALLOC
-TEST_F(ArenaTest, PooledObjectsArePooled) {
+TEST(ArenaTest, PooledObjectsArePooled) {
   struct TestObj {
     char a[100];
   };
 
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto obj = arena->MakePooled<TestObj>();
   Scribble(obj->a, 100, 1);
   EXPECT_TRUE(IsScribbled(obj->a, 100, 1));
@@ -211,11 +195,11 @@ TEST_F(ArenaTest, PooledObjectsArePooled) {
 }
 #endif
 
-TEST_F(ArenaTest, CreateManyObjects) {
+TEST(ArenaTest, CreateManyObjects) {
   struct TestObj {
     char a[100];
   };
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   std::vector<Arena::PoolPtr<TestObj>> objs;
   objs.reserve(1000);
   for (int i = 0; i < 1000; i++) {
@@ -227,9 +211,9 @@ TEST_F(ArenaTest, CreateManyObjects) {
   }
 }
 
-TEST_F(ArenaTest, CreateManyObjectsWithDestructors) {
+TEST(ArenaTest, CreateManyObjectsWithDestructors) {
   using TestObj = std::unique_ptr<int>;
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   std::vector<Arena::PoolPtr<TestObj>> objs;
   objs.reserve(1000);
   for (int i = 0; i < 1000; i++) {
@@ -237,8 +221,8 @@ TEST_F(ArenaTest, CreateManyObjectsWithDestructors) {
   }
 }
 
-TEST_F(ArenaTest, CreatePoolArray) {
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+TEST(ArenaTest, CreatePoolArray) {
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto p = arena->MakePooledArray<int>(1024);
 #ifndef GRPC_ARENA_POOLED_ALLOCATIONS_USE_MALLOC
   EXPECT_FALSE(p.get_deleter().has_freelist());
@@ -251,10 +235,10 @@ TEST_F(ArenaTest, CreatePoolArray) {
   EXPECT_TRUE(IsScribbled(p.get(), 5, 1));
 }
 
-TEST_F(ArenaTest, ConcurrentMakePooled) {
+TEST(ArenaTest, ConcurrentMakePooled) {
   concurrent_test_args args;
   gpr_event_init(&args.ev_start);
-  args.arena = Arena::Create(1024, &memory_allocator_);
+  args.arena = SimpleArenaAllocator()->MakeArena();
 
   class BaseClass {
    public:
@@ -309,8 +293,6 @@ TEST_F(ArenaTest, ConcurrentMakePooled) {
   for (auto& th : thds2) {
     th.Join();
   }
-
-  args.arena->Destroy();
 }
 
 }  // namespace grpc_core

test/core/security/credentials_test.cc

@@ -517,9 +517,7 @@ class RequestMetadataState : public RefCounted<RequestMetadataState> {
 
   grpc_error_handle expected_error_;
   std::string expected_;
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-  ScopedArenaPtr arena_ = MakeScopedArena(1024, &memory_allocator_);
+  RefCountedPtr<Arena> arena_ = SimpleArenaAllocator()->MakeArena();
   grpc_metadata_batch md_;
   grpc_call_credentials::GetRequestMetadataArgs get_request_metadata_args_;
   grpc_polling_entity pollent_;

test/core/security/oauth2_utils.cc

@@ -50,11 +50,7 @@ char* grpc_test_fetch_oauth2_token_with_credentials(
   auto pops = grpc_polling_entity_create_from_pollset(pollset);
   bool is_done = false;
   grpc_core::Notification done;
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
+  auto arena = grpc_core::SimpleArenaAllocator()->MakeArena();
   grpc_metadata_batch initial_metadata;
   char* token = nullptr;
 

test/core/surface/channel_init_test.cc

@@ -268,7 +268,7 @@ TEST(ChannelInitTest, CanCreateFilterWithCall) {
               "test"),
           1024);
   auto call = MakeCallPair(Arena::MakePooled<ClientMetadata>(), nullptr,
-                           allocator->MakeArena(), allocator, nullptr);
+                           allocator->MakeArena(), nullptr);
   (*stack)->StartCall(std::move(call.handler));
   EXPECT_EQ(p, 1);
   EXPECT_EQ(handled, 1);

test/core/transport/binder/binder_transport_test.cc

@@ -64,7 +64,6 @@ class BinderTransportTest : public ::testing::Test {
       gbs->~grpc_binder_stream();
       gpr_free(gbs);
     }
-    arena_->Destroy();
   }
 
   void PerformStreamOp(grpc_binder_stream* gbs,
@@ -81,7 +80,7 @@ class BinderTransportTest : public ::testing::Test {
     grpc_binder_stream* gbs = static_cast<grpc_binder_stream*>(
         gpr_malloc(transport_->filter_stack_transport()->SizeOfStream()));
     transport_->filter_stack_transport()->InitStream(
-        reinterpret_cast<grpc_stream*>(gbs), &ref_, nullptr, arena_);
+        reinterpret_cast<grpc_stream*>(gbs), &ref_, nullptr, arena_.get());
     stream_buffer_.push_back(gbs);
     return gbs;
   }
@@ -95,12 +94,8 @@ class BinderTransportTest : public ::testing::Test {
   static void TearDownTestSuite() { grpc_shutdown(); }
 
  protected:
-  grpc_core::MemoryAllocator memory_allocator_ =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  grpc_core::Arena* arena_ =
-      grpc_core::Arena::Create(/* initial_size = */ 1, &memory_allocator_);
+  grpc_core::RefCountedPtr<grpc_core::Arena> arena_ =
+      grpc_core::SimpleArenaAllocator()->MakeArena();
   grpc_core::Transport* transport_;
   grpc_stream_refcount ref_;
   std::vector<grpc_binder_stream*> stream_buffer_;
@@ -234,12 +229,6 @@ struct MakeSendInitialMetadata {
   }
   ~MakeSendInitialMetadata() {}
 
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  grpc_core::ScopedArenaPtr arena =
-      grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch grpc_initial_metadata;
 };
 
@@ -269,8 +258,6 @@ struct MakeSendTrailingMetadata {
       grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
                                      ->memory_quota()
                                      ->CreateMemoryAllocator("test"));
-  grpc_core::ScopedArenaPtr arena =
-      grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch grpc_trailing_metadata;
 };
 
@@ -297,8 +284,6 @@ struct MakeRecvInitialMetadata {
       grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
                                      ->memory_quota()
                                      ->CreateMemoryAllocator("test"));
-  grpc_core::ScopedArenaPtr arena =
-      grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch grpc_initial_metadata;
   grpc_core::Notification notification;
 };
@@ -345,8 +330,6 @@ struct MakeRecvTrailingMetadata {
       grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
                                      ->memory_quota()
                                      ->CreateMemoryAllocator("test"));
-  grpc_core::ScopedArenaPtr arena =
-      grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch grpc_trailing_metadata;
   grpc_core::Notification notification;
 };

test/core/transport/call_filters_test.cc

@@ -407,9 +407,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningAbslStatus) {
   // Check promise init
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
@@ -467,9 +465,7 @@ TEST(StackDataTest,
   // Check promise init
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
@@ -526,9 +522,7 @@ TEST(StackDataTest, InstantClientInitialMetadataReturningServerMetadata) {
   // Check promise init
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
@@ -588,9 +582,7 @@ TEST(StackDataTest,
   // Check promise init
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
@@ -648,9 +640,7 @@ TEST(StackDataTest, PromiseClientInitialMetadataReturningAbslStatus) {
   // Check promise init
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
@@ -731,9 +721,7 @@ TEST(StackDataTest,
   // Check promise init
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   d.filter_constructor[0].call_init(call_data, &f1);
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ClientMetadata>();
   promise_detail::Context<Arena> ctx(arena.get());
   // A succeeding call
@@ -801,9 +789,7 @@ TEST(StackDataTest, InstantServerInitialMetadataReturningVoid) {
   EXPECT_EQ(d.server_initial_metadata.ops[0].poll, nullptr);
   EXPECT_EQ(d.server_initial_metadata.ops[0].early_destroy, nullptr);
   // Check promise init
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   char call_data;
@@ -838,9 +824,7 @@ TEST(StackDataTest, InstantClientToServerMessagesReturningVoid) {
   EXPECT_EQ(d.client_to_server_messages.ops[0].poll, nullptr);
   EXPECT_EQ(d.client_to_server_messages.ops[0].early_destroy, nullptr);
   // Check promise init
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto message = Arena::MakePooled<Message>(SliceBuffer(), 0);
   char call_data;
   auto r = d.client_to_server_messages.ops[0].promise_init(
@@ -873,9 +857,7 @@ TEST(StackDataTest, InstantServerToClientMessagesReturningVoid) {
   EXPECT_EQ(d.server_to_client_messages.ops[0].poll, nullptr);
   EXPECT_EQ(d.server_to_client_messages.ops[0].early_destroy, nullptr);
   // Check promise init
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto message = Arena::MakePooled<Message>(SliceBuffer(), 0);
   char call_data;
   auto r = d.server_to_client_messages.ops[0].promise_init(
@@ -908,9 +890,7 @@ TEST(StackDataTest, InstantServerTrailingMetadataReturningVoid) {
   EXPECT_EQ(d.server_trailing_metadata.ops[0].poll, nullptr);
   EXPECT_EQ(d.server_trailing_metadata.ops[0].early_destroy, nullptr);
   // Check promise init
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   char call_data;
@@ -948,9 +928,7 @@ TEST(StackDataTest,
   EXPECT_EQ(d.server_trailing_metadata.ops[0].poll, nullptr);
   EXPECT_EQ(d.server_trailing_metadata.ops[0].early_destroy, nullptr);
   // Check promise init
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   auto md = Arena::MakePooled<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
   char call_data;
@@ -1038,9 +1016,7 @@ TEST(OperationExecutorTest, InstantTwo) {
   d.filter_constructor[0].call_init(call_data1, &f1);
   d.filter_constructor[1].call_init(call_data2, &f2);
   OperationExecutor<ClientMetadataHandle> transformer;
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   promise_detail::Context<Arena> ctx(arena.get());
   // First call succeeds
   auto md = Arena::MakePooled<ServerMetadata>();
@@ -1103,9 +1079,7 @@ TEST(OperationExecutorTest, PromiseTwo) {
   d.filter_constructor[0].call_init(call_data1, &f1);
   d.filter_constructor[1].call_init(call_data2, &f2);
   OperationExecutor<ClientMetadataHandle> transformer;
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  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>();
@@ -1175,9 +1149,7 @@ TEST(InfallibleOperationExecutor, InstantTwo) {
   ASSERT_EQ(d.server_trailing_metadata.ops.size(), 2u);
   void* call_data = gpr_malloc_aligned(d.call_data_size, d.call_data_alignment);
   InfallibleOperationExecutor<ServerMetadataHandle> transformer;
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   promise_detail::Context<Arena> ctx(arena.get());
   auto md = Arena::MakePooled<ServerMetadata>();
   EXPECT_EQ(md->get_pointer(HttpPathMetadata()), nullptr);
@@ -1381,9 +1353,7 @@ TEST(CallFiltersTest, UnaryCall) {
   CallFilters::StackBuilder builder;
   builder.Add(&f1);
   builder.Add(&f2);
-  auto memory_allocator =
-      MakeMemoryQuota("test-quota")->CreateMemoryAllocator("foo");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   CallFilters filters(Arena::MakePooled<ClientMetadata>());
   filters.SetStack(builder.Build());
   promise_detail::Context<Arena> ctx(arena.get());

test/core/transport/chaotic_good/client_transport_error_test.cc

@@ -141,9 +141,8 @@ class ClientTransportTest : public ::testing::Test {
   }
 
   auto MakeCall(ClientMetadataHandle client_initial_metadata) {
-    auto* arena = call_arena_allocator_->MakeArena();
     return MakeCallPair(std::move(client_initial_metadata), event_engine_.get(),
-                        arena, call_arena_allocator_, nullptr);
+                        call_arena_allocator_->MakeArena(), nullptr);
   }
 
  private:

test/core/transport/chaotic_good/frame_fuzzer.cc

@@ -104,9 +104,7 @@ void Run(const frame_fuzzer::Test& test) {
   gpr_log(GPR_INFO, "Read frame header: %s", r->ToString().c_str());
   control_data += 24;
   control_size -= 24;
-  MemoryAllocator memory_allocator = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   TestContext<Arena> ctx(arena.get());
   BufferPair buffers{
       SliceBuffer(Slice::FromCopiedBuffer(control_data, control_size)),

test/core/transport/chaotic_good/frame_test.cc

@@ -50,7 +50,7 @@ void AssertRoundTrips(const T& input, FrameType expected_frame_type) {
   MemoryAllocator allocator = MakeResourceQuota("test-quota")
                                   ->memory_quota()
                                   ->CreateMemoryAllocator("test-allocator");
-  ScopedArenaPtr arena = MakeScopedArena(1024, &allocator);
+  RefCountedPtr<Arena> arena = SimpleArenaAllocator()->MakeArena();
   auto deser =
       output.Deserialize(&hpack_parser, header.value(), absl::BitGenRef(bitgen),
                          arena.get(), std::move(serialized), TestFrameLimits());

test/core/transport/chaotic_good/transport_test.h

@@ -36,16 +36,17 @@ class TransportTest : public ::testing::Test {
     return event_engine_;
   }
 
-  Arena* MakeArena() { return call_arena_allocator_->MakeArena(); }
+  RefCountedPtr<Arena> MakeArena() {
+    return call_arena_allocator_->MakeArena();
+  }
 
   RefCountedPtr<CallArenaAllocator> call_arena_allocator() {
     return call_arena_allocator_;
   }
 
   auto MakeCall(ClientMetadataHandle client_initial_metadata) {
-    auto* arena = call_arena_allocator_->MakeArena();
     return MakeCallPair(std::move(client_initial_metadata), event_engine_.get(),
-                        arena, call_arena_allocator_, nullptr);
+                        MakeArena(), nullptr);
   }
 
  private:

test/core/transport/chttp2/hpack_encoder_test.cc

@@ -153,12 +153,6 @@ grpc_slice EncodeHeaderIntoBytes(
     const std::vector<std::pair<std::string, std::string>>& header_fields) {
   std::unique_ptr<grpc_core::HPackCompressor> compressor =
       std::make_unique<grpc_core::HPackCompressor>();
-
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch b;
 
   for (const auto& field : header_fields) {
@@ -307,7 +301,6 @@ static void verify_continuation_headers(const char* key, const char* value,
       grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
                                      ->memory_quota()
                                      ->CreateMemoryAllocator("test"));
-  auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_slice_buffer output;
   grpc_metadata_batch b;
   b.Append(key, grpc_core::Slice::FromStaticString(value), CrashOnAppendError);
@@ -344,11 +337,6 @@ TEST(HpackEncoderTest, TestContinuationHeaders) {
 }
 
 TEST(HpackEncoderTest, EncodeBinaryAsBase64) {
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch b;
   // Haiku by Bard
   b.Append("grpc-trace-bin",
@@ -374,11 +362,6 @@ TEST(HpackEncoderTest, EncodeBinaryAsBase64) {
 }
 
 TEST(HpackEncoderTest, EncodeBinaryAsTrueBinary) {
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch b;
   // Haiku by Bard
   b.Append("grpc-trace-bin",

test/core/transport/chttp2/hpack_parser_fuzzer_test.cc

@@ -69,7 +69,7 @@ DEFINE_PROTO_FUZZER(const hpack_parser_fuzzer::Msg& msg) {
     bool can_update_max_length = true;
     bool can_add_priority = true;
     for (int i = 0; i < msg.frames_size(); i++) {
-      auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
+      auto arena = grpc_core::SimpleArenaAllocator()->MakeArena();
       grpc_core::ExecCtx exec_ctx;
       grpc_metadata_batch b;
       const auto& frame = msg.frames(i);

test/core/transport/chttp2/hpack_parser_input_size_fuzzer.cc

@@ -82,9 +82,7 @@ bool IsStreamError(const absl::Status& status) {
 
 absl::StatusOr<std::string> TestVector(grpc_slice_split_mode mode,
                                        Slice input) {
-  MemoryAllocator memory_allocator = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-  auto arena = MakeScopedArena(1024, &memory_allocator);
+  auto arena = SimpleArenaAllocator()->MakeArena();
   ExecCtx exec_ctx;
   grpc_slice* slices;
   size_t nslices;

test/core/transport/chttp2/hpack_parser_test.cc

@@ -107,10 +107,6 @@ class ParseTest : public ::testing::TestWithParam<Test> {
                   absl::optional<size_t> max_metadata_size,
                   std::string hexstring, absl::StatusOr<std::string> expect,
                   uint32_t flags) {
-    MemoryAllocator memory_allocator = MemoryAllocator(
-        ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
-            "test"));
-    auto arena = MakeScopedArena(1024, &memory_allocator);
     ExecCtx exec_ctx;
     auto input = ParseHexstring(hexstring);
     grpc_slice* slices;

test/core/transport/chttp2/hpack_sync_fuzzer.cc

@@ -117,10 +117,6 @@ void FuzzOneInput(const hpack_sync_fuzzer::Msg& msg) {
 
   // STAGE 2: Decode the buffer (encode_output) into a list of headers
   HPackParser parser;
-  auto memory_allocator =
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
-          "test-allocator");
-  auto arena = MakeScopedArena(1024, &memory_allocator);
   ExecCtx exec_ctx;
   grpc_metadata_batch read_metadata;
   parser.BeginFrame(

test/core/transport/interception_chain_test.cc

@@ -230,9 +230,8 @@ class InterceptionChainTest : public ::testing::Test {
 
   // Run a call through a UnstartedCallDestination until it's complete.
   FinishedCall RunCall(UnstartedCallDestination* destination) {
-    auto* arena = call_arena_allocator_->MakeArena();
     auto call = MakeCallPair(Arena::MakePooled<ClientMetadata>(), nullptr,
-                             arena, call_arena_allocator_, nullptr);
+                             call_arena_allocator_->MakeArena(), nullptr);
     Poll<ServerMetadataHandle> trailing_md;
     call.initiator.SpawnInfallible(
         "run_call", [destination, &call, &trailing_md]() mutable {

test/core/transport/metadata_map_test.cc

@@ -56,24 +56,11 @@ struct StreamNetworkStateMetadataMap
                     GrpcStreamNetworkState>::MetadataMap;
 };
 
-class MetadataMapTest : public ::testing::Test {
- protected:
-  MemoryAllocator memory_allocator_ = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test"));
-};
-
-TEST_F(MetadataMapTest, Noop) {
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
-  EmptyMetadataMap();
-}
+TEST(MetadataMapTest, Noop) { EmptyMetadataMap(); }
 
-TEST_F(MetadataMapTest, NoopWithDeadline) {
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
-  TimeoutOnlyMetadataMap();
-}
+TEST(MetadataMapTest, NoopWithDeadline) { TimeoutOnlyMetadataMap(); }
 
-TEST_F(MetadataMapTest, SimpleOps) {
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
+TEST(MetadataMapTest, SimpleOps) {
   TimeoutOnlyMetadataMap map;
   EXPECT_EQ(map.get_pointer(GrpcTimeoutMetadata()), nullptr);
   EXPECT_EQ(map.get(GrpcTimeoutMetadata()), absl::nullopt);
@@ -110,17 +97,15 @@ class FakeEncoder {
   std::string output_;
 };
 
-TEST_F(MetadataMapTest, EmptyEncodeTest) {
+TEST(MetadataMapTest, EmptyEncodeTest) {
   FakeEncoder encoder;
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
   TimeoutOnlyMetadataMap map;
   map.Encode(&encoder);
   EXPECT_EQ(encoder.output(), "");
 }
 
-TEST_F(MetadataMapTest, TimeoutEncodeTest) {
+TEST(MetadataMapTest, TimeoutEncodeTest) {
   FakeEncoder encoder;
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
   TimeoutOnlyMetadataMap map;
   map.Set(GrpcTimeoutMetadata(),
           Timestamp::FromMillisecondsAfterProcessEpoch(1234));
@@ -128,13 +113,12 @@ TEST_F(MetadataMapTest, TimeoutEncodeTest) {
   EXPECT_EQ(encoder.output(), "grpc-timeout: deadline=1234\n");
 }
 
-TEST_F(MetadataMapTest, NonEncodableTrait) {
+TEST(MetadataMapTest, NonEncodableTrait) {
   struct EncoderWithNoTraitEncodeFunctions {
     void Encode(const Slice&, const Slice&) {
       abort();  // should not be called
     }
   };
-  auto arena = MakeScopedArena(1024, &memory_allocator_);
   StreamNetworkStateMetadataMap map;
   map.Set(GrpcStreamNetworkState(), GrpcStreamNetworkState::kNotSentOnWire);
   EXPECT_EQ(map.get(GrpcStreamNetworkState()),

test/cpp/microbenchmarks/bm_arena.cc

@@ -29,47 +29,35 @@
 using grpc_core::Arena;
 
 static void BM_Arena_NoOp(benchmark::State& state) {
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
+  auto factory = grpc_core::SimpleArenaAllocator();
   for (auto _ : state) {
-    Arena::Create(state.range(0), &memory_allocator)->Destroy();
+    factory->MakeArena();
   }
 }
 BENCHMARK(BM_Arena_NoOp)->Range(1, 1024 * 1024);
 
 static void BM_Arena_ManyAlloc(benchmark::State& state) {
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  Arena* a = Arena::Create(state.range(0), &memory_allocator);
+  auto allocator = grpc_core::SimpleArenaAllocator(state.range(0));
+  auto a = allocator->MakeArena();
   const size_t realloc_after =
       1024 * 1024 * 1024 / ((state.range(1) + 15) & 0xffffff0u);
   while (state.KeepRunning()) {
     a->Alloc(state.range(1));
     // periodically recreate arena to avoid OOM
     if (state.iterations() % realloc_after == 0) {
-      a->Destroy();
-      a = Arena::Create(state.range(0), &memory_allocator);
+      a = allocator->MakeArena();
     }
   }
-  a->Destroy();
 }
 BENCHMARK(BM_Arena_ManyAlloc)->Ranges({{1, 1024 * 1024}, {1, 32 * 1024}});
 
 static void BM_Arena_Batch(benchmark::State& state) {
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
+  auto allocator = grpc_core::SimpleArenaAllocator(state.range(0));
   for (auto _ : state) {
-    Arena* a = Arena::Create(state.range(0), &memory_allocator);
+    auto a = allocator->MakeArena();
     for (int i = 0; i < state.range(1); i++) {
       a->Alloc(state.range(2));
     }
-    a->Destroy();
   }
 }
 BENCHMARK(BM_Arena_Batch)->Ranges({{1, 64 * 1024}, {1, 64}, {1, 1024}});
@@ -82,30 +70,20 @@ struct TestThingToAllocate {
 };
 
 static void BM_Arena_MakePooled_Small(benchmark::State& state) {
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  Arena* a = Arena::Create(1024, &memory_allocator);
+  auto a = grpc_core::SimpleArenaAllocator()->MakeArena();
   for (auto _ : state) {
     a->MakePooled<TestThingToAllocate>();
   }
-  a->Destroy();
 }
 BENCHMARK(BM_Arena_MakePooled_Small);
 
 static void BM_Arena_MakePooled3_Small(benchmark::State& state) {
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  Arena* a = Arena::Create(1024, &memory_allocator);
+  auto a = grpc_core::SimpleArenaAllocator()->MakeArena();
   for (auto _ : state) {
     auto x = a->MakePooled<TestThingToAllocate>();
     auto y = a->MakePooled<TestThingToAllocate>();
     auto z = a->MakePooled<TestThingToAllocate>();
   }
-  a->Destroy();
 }
 BENCHMARK(BM_Arena_MakePooled3_Small);
 

test/cpp/microbenchmarks/bm_chttp2_hpack.cc

@@ -75,7 +75,6 @@ static void BM_HpackEncoderEncodeDeadline(benchmark::State& state) {
       grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
                                      ->memory_quota()
                                      ->CreateMemoryAllocator("test"));
-  auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch b;
   b.Set(grpc_core::GrpcTimeoutMetadata(),
         saved_now + grpc_core::Duration::Seconds(30));
@@ -107,11 +106,6 @@ static void BM_HpackEncoderEncodeHeader(benchmark::State& state) {
   grpc_core::ExecCtx exec_ctx;
   static bool logged_representative_output = false;
 
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
   grpc_metadata_batch b;
   Fixture::Prepare(&b);
 
@@ -345,12 +339,6 @@ static void BM_HpackParserParseHeader(benchmark::State& state) {
   std::vector<grpc_slice> benchmark_slices = Fixture::GetBenchmarkSlices();
   grpc_core::ExecCtx exec_ctx;
   grpc_core::HPackParser p;
-  const int kArenaSize = 4096 * 4096;
-  grpc_core::MemoryAllocator memory_allocator =
-      grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                     ->memory_quota()
-                                     ->CreateMemoryAllocator("test"));
-  auto* arena = grpc_core::Arena::Create(kArenaSize, &memory_allocator);
   grpc_core::ManualConstructor<grpc_metadata_batch> b;
   b.Init();
   p.BeginFrame(&*b, std::numeric_limits<uint32_t>::max(),
@@ -373,25 +361,11 @@ static void BM_HpackParserParseHeader(benchmark::State& state) {
     b->Clear();
     parse_vec(benchmark_slices);
     grpc_core::ExecCtx::Get()->Flush();
-    // Recreate arena every 4k iterations to avoid oom
-    if (0 == (state.iterations() & 0xfff)) {
-      b.Destroy();
-      arena->Destroy();
-      arena = grpc_core::Arena::Create(kArenaSize, &memory_allocator);
-      b.Init();
-      p.BeginFrame(&*b, std::numeric_limits<uint32_t>::max(),
-                   std::numeric_limits<uint32_t>::max(),
-                   grpc_core::HPackParser::Boundary::None,
-                   grpc_core::HPackParser::Priority::None,
-                   grpc_core::HPackParser::LogInfo{
-                       1, grpc_core::HPackParser::LogInfo::kHeaders, false});
-    }
   }
   // Clean up
   b.Destroy();
   for (auto slice : init_slices) grpc_slice_unref(slice);
   for (auto slice : benchmark_slices) grpc_slice_unref(slice);
-  arena->Destroy();
 }
 
 namespace hpack_parser_fixtures {
@@ -405,12 +379,6 @@ class FromEncoderFixture {
  private:
   static std::vector<grpc_slice> Generate(int iteration) {
     grpc_core::ExecCtx exec_ctx;
-
-    grpc_core::MemoryAllocator memory_allocator =
-        grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
-                                       ->memory_quota()
-                                       ->CreateMemoryAllocator("test"));
-    auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
     grpc_metadata_batch b;
     EncoderFixture::Prepare(&b);