Use protobuf Arena for fanout response deserialization (valkey-io#1040)

Using protobuf Arena for GRPC fanout response deserialization

### Problem

`SearchIndexPartitionResponse` is heap-allocated in the gRPC client.
Each response contains N `NeighborEntry` sub-messages, each individually
malloc'd during deserialization and individually freed on destruction.
At large result sets, this creates thousands of malloc/free calls per
query per remote shard.

### Fix

Allocate `SearchIndexPartitionResponse` on a `google::protobuf::Arena`.
All sub-messages use bulk arena allocation and are freed in one shot
when the arena is destroyed.


### Benchmark

3-shard CME cluster, 10K docs, TEXT index, 500 concurrent clients
querying one node (fanout to 2 remote shards).
| Workload | Before | After | Improvement |
|---|---|---|---|
| LIMIT 100 CONTENT | 2,496 req/s | 2,572 req/s | +3% throughput |
| LIMIT 500 CONTENT | 306 req/s | 510 req/s | **+67% throughput, -25%
p50** |
| LIMIT 1000 CONTENT | 88 req/s | 272 req/s | **+209% throughput, -44%
p50** |
| NOCONTENT LIMIT 100 | 2,943 req/s | 6,969 req/s | **+137% throughput,
-65% p50** |

Signed-off-by: Karthik Subbarao <karthikrs2021@gmail.com>

Author: KarthikSubbarao

src/coordinator/client.cc

@@ -16,6 +16,7 @@
 #include "absl/strings/string_view.h"
 #include "absl/time/clock.h"
 #include "absl/time/time.h"
+#include "google/protobuf/arena.h"
 #include "grpc/grpc.h"
 #include "grpcpp/channel.h"
 #include "grpcpp/client_context.h"
@@ -142,11 +143,15 @@ void ClientImpl::SearchIndexPartition(
   struct SearchIndexPartitionArgs {
     ::grpc::ClientContext context;
     std::unique_ptr<SearchIndexPartitionRequest> request;
-    SearchIndexPartitionResponse response;
+    google::protobuf::Arena arena;
+    SearchIndexPartitionResponse* response;
     SearchIndexPartitionCallback callback;
     std::unique_ptr<vmsdk::StopWatch> latency_sample;
   };
   auto args = std::make_unique<SearchIndexPartitionArgs>();
+  args->response =
+      google::protobuf::Arena::Create<SearchIndexPartitionResponse>(
+          &args->arena);
   args->context.set_deadline(absl::ToChronoTime(
       absl::Now() + absl::Seconds(query_connection_timeout->GetValue())));
   args->callback = std::move(done);
@@ -156,20 +161,20 @@ void ClientImpl::SearchIndexPartition(
   Metrics::GetStats().coordinator_bytes_out.fetch_add(
       args_raw->request->ByteSizeLong(), std::memory_order_relaxed);
   stub_->async()->SearchIndexPartition(
-      &args_raw->context, args_raw->request.get(), &args_raw->response,
+      &args_raw->context, args_raw->request.get(), args_raw->response,
       // std::function is not move-only.
       [args_raw](grpc::Status s) mutable {
         GRPCSuspensionGuard guard(GRPCSuspender::Instance());
         auto args = std::unique_ptr<SearchIndexPartitionArgs>(args_raw);
-        args->callback(s, args->response);
+        args->callback(s, *args->response);
         if (s.ok()) {
           Metrics::GetStats()
               .coordinator_client_search_index_partition_success_cnt++;
           Metrics::GetStats()
               .coordinator_client_search_index_partition_success_latency
               .SubmitSample(std::move(args->latency_sample));
           Metrics::GetStats().coordinator_bytes_in.fetch_add(
-              args->response.ByteSizeLong(), std::memory_order_relaxed);
+              args->response->ByteSizeLong(), std::memory_order_relaxed);
         } else {
           Metrics::GetStats()
               .coordinator_client_search_index_partition_failure_cnt++;