A TP>1 decode engine with replicated KV caches needs the same bytes on all of its workers, but a pool-reshard plan allowed exactly one destination.

The changes are for supporting transfer to N destinations. The gist of the changes are:
1. The planner accepts a destination list (all must have identical pool geometry)
2. emits each copy instruction once per destination, computes expected push counts per destination etc.
3. the coordinator arms all receivers concurrently before dispatching each sender, with each arm carrying only that receiver's schedule slice and push count.

This covers caches that are fully replicated on every destination rank: each destination receives the identical byte set. Real resharding to sharded destinations (different bytes per rank, e.g. head-split KV caches) is not supported yet and would build on top of this by mapping each declared span to one destination instead of all of them

PiperOrigin-RevId: 968553686

Author: HarutMov

tpu_sync/kv_cache/reshard/pool_reshard_planner.cc

@@ -143,9 +143,9 @@ absl::StatusOr<PoolReshardPlan> BuildPoolReshardPlan(
     return absl::InvalidArgumentError(
         "uuid must be positive for pool resharding");
   }
-  if (request.dst_units.size() != 1) {
+  if (request.dst_units.empty()) {
     return absl::InvalidArgumentError(
-        "Pool resharding requires exactly one destination unit");
+        "Pool resharding requires at least one destination unit");
   }
   {
     std::set<RaidenId, RequestBlockRegistry::RaidenIdLess> unique_src(
@@ -179,16 +179,18 @@ absl::StatusOr<PoolReshardPlan> BuildPoolReshardPlan(
   auto dst_by_unit_or = MetadataByUnit(request.dst_metadata, request.dst_units);
   if (!dst_by_unit_or.ok()) return dst_by_unit_or.status();
   auto& dst_by_unit = *dst_by_unit_or;
-  const RaidenId& dst_unit = request.dst_units[0];
+  // All destinations must share one local pool geometry (validated below).
   const tpu_sync::rpc::RegisterWorkUnitRequest& dst_meta =
-      *dst_by_unit.at(dst_unit);
+      *dst_by_unit.at(request.dst_units[0]);
 
   std::vector<const tpu_sync::rpc::RegisterWorkUnitRequest*> all_metadata;
-  all_metadata.reserve(request.src_units.size() + 1);
+  all_metadata.reserve(request.src_units.size() + request.dst_units.size());
   for (const RaidenId& unit : request.src_units) {
     all_metadata.push_back(src_by_unit.at(unit));
   }
-  all_metadata.push_back(&dst_meta);
+  for (const RaidenId& unit : request.dst_units) {
+    all_metadata.push_back(dst_by_unit.at(unit));
+  }
   for (const auto* meta : all_metadata) {
     RaidenId unit = RaidenIdFromProto(meta->unit());
     if (meta->layout_fingerprint().empty()) {
@@ -246,6 +248,34 @@ absl::StatusOr<PoolReshardPlan> BuildPoolReshardPlan(
           PythonRepr(src_unit)));
     }
   }
+  {
+    std::vector<std::string> reference_dst_geometry;
+    for (const auto& pool : dst_meta.pools()) {
+      reference_dst_geometry.push_back(GeometrySignature(pool));
+    }
+    for (size_t i = 1; i < request.dst_units.size(); ++i) {
+      const RaidenId& unit = request.dst_units[i];
+      const auto& other_meta = *dst_by_unit.at(unit);
+      std::vector<std::pair<std::string, std::string>> other_identity;
+      for (const auto& pool : other_meta.pools()) {
+        other_identity.emplace_back(pool.tag(), pool.dtype_tag());
+      }
+      if (other_identity != dst_identity) {
+        return absl::InvalidArgumentError(absl::StrCat(
+            "Canonical pool manifest mismatch between destinations at ",
+            PythonRepr(unit)));
+      }
+      std::vector<std::string> other_geometry;
+      for (const auto& pool : other_meta.pools()) {
+        other_geometry.push_back(GeometrySignature(pool));
+      }
+      if (other_geometry != reference_dst_geometry) {
+        return absl::InvalidArgumentError(
+            absl::StrCat("Destination pool geometry differs across units at ",
+                         PythonRepr(unit)));
+      }
+    }
+  }
 
   const tpu_sync::rpc::RegisterWorkUnitRequest& reference_src =
       *src_by_unit.at(request.src_units[0]);
@@ -484,7 +514,22 @@ absl::StatusOr<PoolReshardPlan> BuildPoolReshardPlan(
   // Per-tag planning: each requested tag selects its own pools, owns its
   // own destination block-id space and coverage validation, and emits one
   // entry group.
-  const std::string dst_peer = dst_meta.shards(0);
+  std::map<RaidenId, std::string, RequestBlockRegistry::RaidenIdLess> dst_peers;
+  {
+    std::set<std::string> unique_peers;
+    for (const RaidenId& unit : request.dst_units) {
+      const std::string peer = dst_by_unit.at(unit)->shards(0);
+      unique_peers.insert(peer);
+      dst_peers.emplace(unit, peer);
+    }
+    if (unique_peers.size() != request.dst_units.size()) {
+      std::vector<std::string> sorted_peers(unique_peers.begin(),
+                                            unique_peers.end());
+      return absl::InvalidArgumentError(absl::StrCat(
+          "Destinations must register distinct data-plane endpoints; got ",
+          PyStrListRepr(sorted_peers)));
+    }
+  }
   std::map<RaidenId, std::vector<ScheduleEntry>,
            RequestBlockRegistry::RaidenIdLess>
       schedules;
@@ -736,8 +781,11 @@ absl::StatusOr<PoolReshardPlan> BuildPoolReshardPlan(
           return std::tie(a.span->dst_block_index, a.span->dst_offset_bytes) <
                  std::tie(b.span->dst_block_index, b.span->dst_offset_bytes);
         });
-    std::map<RaidenId, std::set<std::tuple<std::string, int64_t, int64_t>>,
-             RequestBlockRegistry::RaidenIdLess>
+    std::map<
+        RaidenId,
+        std::map<RaidenId, std::set<std::tuple<std::string, int64_t, int64_t>>,
+                 RequestBlockRegistry::RaidenIdLess>,
+        RequestBlockRegistry::RaidenIdLess>
         transfer_pairs_per_sender;
     for (const OrderedSpan& ordered : ordered_spans) {
       const PoolByteSpan& span = *ordered.span;
@@ -754,37 +802,57 @@ absl::StatusOr<PoolReshardPlan> BuildPoolReshardPlan(
             TranslateLiveCopy(precheck.src_segments, precheck.dst_segments,
                               src_offset, dst_offset, span.size_bytes);
         if (!translated.ok()) return translated.status();
-        emitted_chunks += static_cast<int64_t>(translated->size());
+        emitted_chunks += static_cast<int64_t>(translated->size()) *
+                          static_cast<int64_t>(request.dst_units.size());
         if (emitted_chunks > kMaxLiveSegments) {
           return absl::InvalidArgumentError(
               "Byte-span plan exceeds the live-region expansion bound");
         }
+        // Replicated caches: every destination receives the identical
+        // chunk, so emission is the (chunk x destination) cross product and
+        // entries differ only in dst_peer. For supporting sharded
+        // destionations, this needs to be updated.
         for (const LiveCopyChunk& chunk : *translated) {
-          ScheduleEntry schedule_entry;
-          schedule_entry.dst_peer = dst_peer;
-          schedule_entry.dst_shard_idx = 0;
-          schedule_entry.dst_offset_bytes = chunk.dst_physical;
-          schedule_entry.src_offset_bytes = chunk.src_physical;
-          schedule_entry.size_bytes = chunk.size;
-          schedule_entry.src_block_id = src_block_id;
-          schedule_entry.dst_block_id = dst_block_id;
-          schedule_entry.src_stride_bytes = 0;
-          schedule_entry.dst_stride_bytes = 0;
-          schedule_entry.count = 1;
-          schedule_entry.layer_idx = 0;
-          schedule_entry.pool_group = static_cast<int32_t>(group_idx);
-          schedules[src_unit].push_back(std::move(schedule_entry));
+          for (const RaidenId& dst_unit_id : request.dst_units) {
+            ScheduleEntry schedule_entry;
+            schedule_entry.dst_peer = dst_peers.at(dst_unit_id);
+            schedule_entry.dst_shard_idx = 0;
+            schedule_entry.dst_offset_bytes = chunk.dst_physical;
+            schedule_entry.src_offset_bytes = chunk.src_physical;
+            schedule_entry.size_bytes = chunk.size;
+            schedule_entry.src_block_id = src_block_id;
+            schedule_entry.dst_block_id = dst_block_id;
+            schedule_entry.src_stride_bytes = 0;
+            schedule_entry.dst_stride_bytes = 0;
+            schedule_entry.count = 1;
+            schedule_entry.layer_idx = 0;
+            schedule_entry.pool_group = static_cast<int32_t>(group_idx);
+            schedules[src_unit].push_back(std::move(schedule_entry));
+          }
         }
       }
-      transfer_pairs_per_sender[src_unit].insert(
-          std::make_tuple(dst_peer, src_block_id, dst_block_id));
+      auto& sender_pairs = transfer_pairs_per_sender[src_unit];
+      for (const RaidenId& dst_unit_id : request.dst_units) {
+        sender_pairs[dst_unit_id].insert(std::make_tuple(
+            dst_peers.at(dst_unit_id), src_block_id, dst_block_id));
+      }
     }
 
-    int64_t group_expected_pushes = 0;
-    for (const auto& [unit, pairs] : transfer_pairs_per_sender) {
-      group_expected_pushes +=
-          std::min(requested_parallelism, static_cast<int64_t>(pairs.size()));
+    // Computed expected pushes for the receiver.
+    std::map<RaidenId, int32_t, RequestBlockRegistry::RaidenIdLess>
+        expected_pushes_by_dst;
+    for (const RaidenId& dst_unit_id : request.dst_units) {
+      int64_t dst_pushes = 0;
+      for (const auto& [unit, by_dst] : transfer_pairs_per_sender) {
+        auto pairs_it = by_dst.find(dst_unit_id);
+        if (pairs_it == by_dst.end()) continue;
+        dst_pushes += std::min(requested_parallelism,
+                               static_cast<int64_t>(pairs_it->second.size()));
+      }
+      expected_pushes_by_dst[dst_unit_id] = static_cast<int32_t>(dst_pushes);
     }
+    const int64_t group_expected_pushes =
+        expected_pushes_by_dst.at(request.dst_units[0]);
     if (group_expected_pushes <= 0) {
       return absl::InvalidArgumentError(
           absl::StrCat("Pool reshard plan contains no source pushes for tag ",
@@ -793,7 +861,7 @@ absl::StatusOr<PoolReshardPlan> BuildPoolReshardPlan(
     PlanPoolGroup group;
     group.pool_indices = precheck.selected;
     group.dst_device_block_ids = dst_ids_g;
-    group.expected_pushes = static_cast<int32_t>(group_expected_pushes);
+    group.expected_pushes_by_dst = std::move(expected_pushes_by_dst);
     group.dst_expected_extent_bytes = extents;
     // FA (the first requested tag by connector convention) uploads first;
     // state classes land after it on aliased arena pages.
@@ -830,17 +898,18 @@ absl::StatusOr<PoolReshardPlan> BuildPoolReshardPlan(
       plan.src_units.push_back(unit);
     }
   }
-  plan.dst_unit = dst_unit;
+  plan.dst_units = request.dst_units;
   plan.schedules = std::move(schedules);
   for (const auto* meta : all_metadata) {
     plan.worker_rpc_addresses[RaidenIdFromProto(meta->unit())] =
         meta->control_plane_rpc_address();
   }
-  plan.dst_peer = dst_peer;
+  plan.dst_peers = std::move(dst_peers);
   plan.uuid = uuid;
   plan.req_id = req_id;
   plan.expected_block_count = static_cast<int64_t>(dst_ids.size());
-  plan.expected_pushes_per_pool = pool_groups[0].expected_pushes;
+  plan.expected_pushes_per_pool =
+      pool_groups[0].expected_pushes_by_dst.at(request.dst_units[0]);
   plan.transfer_pool_indices = union_selected;
   for (const auto& pool : dst_meta.pools()) {
     plan.pool_dtype_tags.push_back(pool.dtype_tag());

tpu_sync/kv_cache/reshard/pool_reshard_planner.h

@@ -52,17 +52,20 @@ struct ScheduleEntry {
 struct PlanPoolGroup {
   std::vector<int32_t> pool_indices;
   std::vector<int64_t> dst_device_block_ids;
-  int32_t expected_pushes = 0;
+  std::map<RaidenId, int32_t, RequestBlockRegistry::RaidenIdLess>
+      expected_pushes_by_dst;
   std::vector<int64_t> dst_expected_extent_bytes;
   int32_t order_rank = 0;
 };
 
 // The pool-path subset of TransferPlan that the encoder and coordinator
 // consume. Field-for-field mirror of _build_byte_span_plan_claimed's
 // return value.
+// Note: Multi-destination plans replicate one identical byte set
+// to every destination (TP>1 decode engine).
 struct PoolReshardPlan {
   std::vector<RaidenId> src_units;  // active source units, rank order
-  RaidenId dst_unit;
+  std::vector<RaidenId> dst_units;
   // Per source unit: shard 0's entry list (pool planning enforces one
   // endpoint per unit, so the inner Python dict always has the single key
   // 0). Keyed in src_units order.
@@ -71,7 +74,8 @@ struct PoolReshardPlan {
       schedules;
   std::map<RaidenId, std::string, RequestBlockRegistry::RaidenIdLess>
       worker_rpc_addresses;
-  std::string dst_peer;  // worker_data_addresses[dst_unit][0]
+  // worker_data_addresses[unit][0] per destination unit.
+  std::map<RaidenId, std::string, RequestBlockRegistry::RaidenIdLess> dst_peers;
   int64_t uuid = 0;
   std::string req_id;
   int64_t expected_block_count = 0;

tpu_sync/kv_cache/reshard/reshard_coordinator.cc

@@ -110,11 +110,20 @@ tpu_sync::rpc::StartTransferRequest BuildStartTransferForTarget(
     const PoolReshardPlan& plan, const RaidenId& target) {
   const bool is_sender = std::find(plan.src_units.begin(), plan.src_units.end(),
                                    target) != plan.src_units.end();
+  const bool is_receiver =
+      std::find(plan.dst_units.begin(), plan.dst_units.end(), target) !=
+      plan.dst_units.end();
   tpu_sync::rpc::StartTransferRequest start_req;
   for (const RaidenId& unit : plan.src_units) {
     *start_req.add_src_units() = RaidenIdToProto(unit);
   }
-  *start_req.add_dst_units() = RaidenIdToProto(plan.dst_unit);
+  if (is_receiver) {
+    *start_req.add_dst_units() = RaidenIdToProto(target);
+  } else {
+    for (const RaidenId& unit : plan.dst_units) {
+      *start_req.add_dst_units() = RaidenIdToProto(unit);
+    }
+  }
   start_req.set_uuid(plan.uuid);
   start_req.set_is_sender(is_sender);
   start_req.set_dst_mem_type(tpu_sync::rpc::MEMORY_TYPE_HBM);
@@ -141,7 +150,13 @@ tpu_sync::rpc::StartTransferRequest BuildStartTransferForTarget(
     for (int64_t block_id : group.dst_device_block_ids) {
       group_proto->add_dst_device_block_ids(block_id);
     }
-    group_proto->set_expected_pushes(group.expected_pushes);
+    const RaidenId& count_key = is_receiver ? target : plan.dst_units[0];
+    int32_t expected_pushes = 0;
+    auto by_dst_it = group.expected_pushes_by_dst.find(count_key);
+    if (by_dst_it != group.expected_pushes_by_dst.end()) {
+      expected_pushes = by_dst_it->second;
+    }
+    group_proto->set_expected_pushes(expected_pushes);
     for (int64_t extent : group.dst_expected_extent_bytes) {
       group_proto->add_dst_expected_extent_bytes(extent);
     }
@@ -171,16 +186,17 @@ tpu_sync::rpc::StartTransferRequest BuildStartTransferForTarget(
         }
       };
 
-  if (target == plan.dst_unit) {
+  if (is_receiver) {
     // Receiver path: every source's schedule, keyed by source ordinal,
     // filtered to entries targeting this receiver (single-endpoint pool
     // plans always match).
+    const std::string& target_peer = plan.dst_peers.at(target);
     for (const auto& [src_unit, entries] : plan.schedules) {
       auto key_it = plan.src_schedule_keys.find(src_unit);
       if (key_it == plan.src_schedule_keys.end()) continue;
       std::vector<ScheduleEntry> filtered;
       for (const ScheduleEntry& entry : entries) {
-        if (entry.dst_peer == plan.dst_peer) filtered.push_back(entry);
+        if (entry.dst_peer == target_peer) filtered.push_back(entry);
       }
       if (filtered.empty()) continue;
       tpu_sync::rpc::ShardPushScheduleProto schedule_proto;
@@ -205,7 +221,9 @@ std::string EncodeStartTransfer(const PoolReshardPlan& plan,
   tpu_sync::rpc::ControlRequest req;
   req.set_command(tpu_sync::rpc::ControlRequest::COMMAND_START_TRANSFER);
   // peers: the destination units' data endpoints (one dst, one endpoint).
-  req.add_peers(plan.dst_peer);
+  for (const RaidenId& unit : plan.dst_units) {
+    req.add_peers(plan.dst_peers.at(unit));
+  }
   *req.mutable_start_transfer_request() =
       BuildStartTransferForTarget(plan, target);
   return req.SerializeAsString();
@@ -354,20 +372,32 @@ absl::Status ReshardCoordinator::ExecutePoolReshard(
   const int64_t plan_build_end_ns = MonotonicNs();
 
   // Receivers must be armed before any sender can put bytes on the wire.
+  // Multi-destination plans arm every receiver concurrently and join
+  // before dispatching senders.
   const int64_t receiver_arm_start_ns = MonotonicNs();
   {
-    auto addr_it = plan.worker_rpc_addresses.find(plan.dst_unit);
-    if (addr_it == plan.worker_rpc_addresses.end()) {
-      registry_->AbandonClaim(args.req_id, uuid, claim_owner);
-      return absl::InternalError(absl::StrCat(
-          "No control endpoint recorded for ", PythonRepr(plan.dst_unit)));
+    std::vector<absl::Status> arm_status(plan.dst_units.size());
+    std::vector<std::thread> armers;
+    armers.reserve(plan.dst_units.size());
+    for (size_t i = 0; i < plan.dst_units.size(); ++i) {
+      const RaidenId& unit = plan.dst_units[i];
+      armers.emplace_back([this, &plan, &arm_status, i, unit]() {
+        auto addr_it = plan.worker_rpc_addresses.find(unit);
+        if (addr_it == plan.worker_rpc_addresses.end()) {
+          arm_status[i] = absl::InternalError(absl::StrCat(
+              "No control endpoint recorded for ", PythonRepr(unit)));
+          return;
+        }
+        const std::string arm_payload = EncodeStartTransfer(plan, unit);
+        arm_status[i] = SendWorkerRpc(transport_, addr_it->second, arm_payload);
+      });
     }
-    const std::string arm_payload = EncodeStartTransfer(plan, plan.dst_unit);
-    absl::Status arm_status =
-        SendWorkerRpc(transport_, addr_it->second, arm_payload);
-    if (!arm_status.ok()) {
-      registry_->AbandonClaim(args.req_id, uuid, claim_owner);
-      return arm_status;
+    for (std::thread& t : armers) t.join();
+    for (const absl::Status& status : arm_status) {
+      if (!status.ok()) {
+        registry_->AbandonClaim(args.req_id, uuid, claim_owner);
+        return status;
+      }
     }
   }
   const int64_t receiver_arm_ack_ns = MonotonicNs();
@@ -464,6 +494,7 @@ absl::Status ReshardCoordinator::ExecutePoolReshard(
       ", \"controller_total_ms\": ",
       JsonFloat((sender_dispatch_end_ns - controller_start_ns) / 1e6),
       ", \"destination_pages\": ", plan.expected_block_count,
+      ", \"destination_units\": ", plan.dst_units.size(),
       ", \"event\": \"raiden_pool_reshard_senders_dispatched\"",
       ", \"expected_pushes_per_pool\": ", plan.expected_pushes_per_pool,
       ", \"num_tokens\": ", plan.num_tokens, ", \"plan_build_ms\": ",