Pass Arena ptr from transport to frame deserialization.

src/core/BUILD

@@ -5838,9 +5838,7 @@ grpc_cc_library(
         "arena",
         "bitset",
         "chaotic_good_frame_header",
-        "memory_quota",
         "no_destruct",
-        "resource_quota",
         "slice",
         "slice_buffer",
         "status_helper",
@@ -5849,7 +5847,6 @@ grpc_cc_library(
         "//:grpc_base",
         "//:hpack_encoder",
         "//:hpack_parser",
-        "//:ref_counted_ptr",
     ],
 )
 

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

@@ -151,6 +151,7 @@ ClientTransport::ClientTransport(
           ServerFragmentFrame frame;
           // Initialized to get this_cpu() info in global_stat().
           ExecCtx exec_ctx;
+          frame.SetArena(arena_);
           // Deserialize frame from read buffer.
           auto status = frame.Deserialize(hpack_parser_.get(), *frame_header_,
                                           control_endpoint_read_buffer_);

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

@@ -188,7 +188,7 @@ class ClientTransport {
   std::unique_ptr<HPackParser> hpack_parser_;
   std::shared_ptr<FrameHeader> frame_header_;
   MemoryAllocator memory_allocator_;
-  ScopedArenaPtr arena_;
+  std::shared_ptr<Arena> arena_;  // Shared ownership with segment frames.
   // Use to synchronize writer_ and reader_ activity with outside activities;
   std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_;
 };

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

@@ -25,17 +25,13 @@
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 
-#include <grpc/event_engine/memory_allocator.h>
 #include <grpc/slice.h>
 #include <grpc/support/log.h>
 
 #include "src/core/lib/gprpp/bitset.h"
 #include "src/core/lib/gprpp/no_destruct.h"
-#include "src/core/lib/gprpp/ref_counted_ptr.h"
 #include "src/core/lib/gprpp/status_helper.h"
 #include "src/core/lib/resource_quota/arena.h"
-#include "src/core/lib/resource_quota/memory_quota.h"
-#include "src/core/lib/resource_quota/resource_quota.h"
 #include "src/core/lib/slice/slice.h"
 #include "src/core/lib/slice/slice_buffer.h"
 
@@ -117,13 +113,10 @@ class FrameDeserializer {
 template <typename Metadata>
 absl::StatusOr<Arena::PoolPtr<Metadata>> ReadMetadata(
     HPackParser* parser, absl::StatusOr<SliceBuffer> maybe_slices,
-    uint32_t stream_id, bool is_header, bool is_client) {
+    uint32_t stream_id, bool is_header, bool is_client,
+    std::shared_ptr<Arena> arena) {
   if (!maybe_slices.ok()) return maybe_slices.status();
   auto& slices = *maybe_slices;
-  MemoryAllocator memory_allocator = MemoryAllocator(
-      ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
-          "read_metadata"));
-  auto arena = MakeScopedArena(1024, &memory_allocator);
   Arena::PoolPtr<Metadata> metadata = arena->MakePooled<Metadata>(arena.get());
   parser->BeginFrame(
       metadata.get(), std::numeric_limits<uint32_t>::max(),
@@ -175,7 +168,7 @@ absl::Status ClientFragmentFrame::Deserialize(HPackParser* parser,
   FrameDeserializer deserializer(header, slice_buffer);
   if (header.flags.is_set(0)) {
     auto r = ReadMetadata<ClientMetadata>(parser, deserializer.ReceiveHeaders(),
-                                          header.stream_id, true, true);
+                                          header.stream_id, true, true, arena_);
     if (!r.ok()) return r.status();
   }
   if (header.flags.is_set(1)) {
@@ -213,15 +206,17 @@ absl::Status ServerFragmentFrame::Deserialize(HPackParser* parser,
   }
   FrameDeserializer deserializer(header, slice_buffer);
   if (header.flags.is_set(0)) {
-    auto r = ReadMetadata<ServerMetadata>(parser, deserializer.ReceiveHeaders(),
-                                          header.stream_id, true, false);
+    auto r =
+        ReadMetadata<ServerMetadata>(parser, deserializer.ReceiveHeaders(),
+                                     header.stream_id, true, false, arena_);
     if (!r.ok()) return r.status();
     GPR_ASSERT(r.value() != nullptr);
     headers = std::move(r.value());
   }
   if (header.flags.is_set(1)) {
-    auto r = ReadMetadata<ServerMetadata>(
-        parser, deserializer.ReceiveTrailers(), header.stream_id, false, false);
+    auto r =
+        ReadMetadata<ServerMetadata>(parser, deserializer.ReceiveTrailers(),
+                                     header.stream_id, false, false, arena_);
     if (!r.ok()) return r.status();
     GPR_ASSERT(r.value() != nullptr);
     trailers = std::move(r.value());

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

@@ -82,6 +82,9 @@ struct ClientFragmentFrame final : public FrameInterface {
     return stream_id == other.stream_id && EqHdl(headers, other.headers) &&
            end_of_stream == other.end_of_stream;
   }
+  // Shared ownership with transport.
+  std::shared_ptr<Arena> arena_;
+  void SetArena(std::shared_ptr<Arena> arena) { arena_ = arena; }
 };
 
 struct ServerFragmentFrame final : public FrameInterface {
@@ -98,6 +101,10 @@ struct ServerFragmentFrame final : public FrameInterface {
     return stream_id == other.stream_id && EqHdl(headers, other.headers) &&
            EqHdl(trailers, other.trailers);
   }
+
+  // Shared ownership with transport.
+  std::shared_ptr<Arena> arena_;
+  void SetArena(std::shared_ptr<Arena> arena) { arena_ = arena; }
 };
 
 struct CancelFrame final : public FrameInterface {