Move planning hints from global coordinator state to per-operation

Instead of setting hints globally on the Coordinator (which requires
coordinating hint state across operations), hints are now passed with
each copy/pull submission. This makes the API stateless and allows
different operations to use different routing strategies.

Key changes:
- Add hints parameter to SubmitCopyRequest and SubmitPullRequest
- Add RoutingHint with serializable topology path
- Update Client.copy() and Client.pull() to accept hints
- Remove global add_hint/clear_hints from Coordinator pybind
- Add Topology serialization support for hint transport

Author: 1ntEgr8

csrc/setu/client/Client.cpp

@@ -150,7 +150,8 @@ std::optional<TensorShardRef> Client::RegisterTensorShard(
   return response.shard_ref;
 }
 
-std::optional<CopyOperationId> Client::SubmitCopy(const CopySpec& copy_spec) {
+std::optional<CopyOperationId> Client::SubmitCopy(
+    const CopySpec& copy_spec, const std::vector<CompilerHint>& hints) {
   // Find all shards owned by this client that are involved in the copy
   // (either as source or destination)
   std::vector<ShardId> involved_shards;
@@ -171,10 +172,14 @@ std::optional<CopyOperationId> Client::SubmitCopy(const CopySpec& copy_spec) {
                        "Client has no shards for src {} or dst {}",
                        copy_spec.src_name, copy_spec.dst_name);
 
+  // Compute fingerprint once for all shard submissions
+  const auto fingerprint = setu::planner::hints::Fingerprint(hints);
+
   // Submit a request for each involved shard
   std::optional<CopyOperationId> copy_op_id;
   for (const auto& shard_id : involved_shards) {
-    ClientRequest request = SubmitCopyRequest(shard_id, copy_spec);
+    ClientRequest request =
+        SubmitCopyRequest(shard_id, copy_spec, hints, fingerprint);
     Comm::Send(request_socket_, request);
 
     auto response = Comm::Recv<SubmitCopyResponse>(request_socket_);
@@ -192,18 +197,23 @@ std::optional<CopyOperationId> Client::SubmitCopy(const CopySpec& copy_spec) {
   return copy_op_id;
 }
 
-std::optional<CopyOperationId> Client::SubmitPull(const CopySpec& copy_spec) {
+std::optional<CopyOperationId> Client::SubmitPull(
+    const CopySpec& copy_spec, const std::vector<CompilerHint>& hints) {
   // For Pull: only destination shards submit (one-sided operation)
   auto it = tensor_shards_.find(copy_spec.dst_name);
   ASSERT_VALID_RUNTIME(it != tensor_shards_.end(),
                        "Client has no shards for dst {}", copy_spec.dst_name);
 
+  // Compute fingerprint once for all shard submissions
+  const auto fingerprint = setu::planner::hints::Fingerprint(hints);
+
   // Submit a request for each destination shard
   std::optional<CopyOperationId> copy_op_id;
   for (const auto& shard_ref : it->second) {
     const auto shard_id = shard_ref->shard_id;
 
-    ClientRequest request = SubmitPullRequest(shard_id, copy_spec);
+    ClientRequest request =
+        SubmitPullRequest(shard_id, copy_spec, hints, fingerprint);
     Comm::Send(request_socket_, request);
 
     auto response = Comm::Recv<SubmitCopyResponse>(request_socket_);

csrc/setu/client/Client.h

@@ -26,6 +26,7 @@
 #include "commons/utils/TorchTensorIPC.h"
 #include "commons/utils/ZmqHelper.h"
 #include "messaging/GetTensorHandleResponse.h"
+#include "planner/hints/Hint.h"
 
 namespace setu::client {
 using setu::commons::CopyOperationId;
@@ -41,6 +42,7 @@ using setu::commons::messages::GetTensorHandleResponse;
 using setu::commons::utils::TensorIPCSpec;
 using setu::commons::utils::ZmqContextPtr;
 using setu::commons::utils::ZmqSocketPtr;
+using setu::planner::hints::CompilerHint;
 
 class Client {
  public:
@@ -58,9 +60,11 @@ class Client {
   std::optional<TensorShardRef> RegisterTensorShard(
       const TensorShardSpec& shard_spec);
 
-  std::optional<CopyOperationId> SubmitCopy(const CopySpec& copy_spec);
+  std::optional<CopyOperationId> SubmitCopy(
+      const CopySpec& copy_spec, const std::vector<CompilerHint>& hints = {});
 
-  std::optional<CopyOperationId> SubmitPull(const CopySpec& copy_spec);
+  std::optional<CopyOperationId> SubmitPull(
+      const CopySpec& copy_spec, const std::vector<CompilerHint>& hints = {});
 
   void WaitForCopy(CopyOperationId copy_op_id);
 

csrc/setu/client/Pybind.cpp

@@ -23,13 +23,16 @@
 #include "commons/datatypes/CopySpec.h"
 #include "commons/datatypes/TensorShardSpec.h"
 #include "commons/enums/Enums.h"
+#include "planner/hints/Hint.h"
 //==============================================================================
 namespace setu::client {
 //==============================================================================
 using setu::commons::CopyOperationId;
 using setu::commons::datatypes::CopySpec;
 using setu::commons::datatypes::TensorShardSpec;
 using setu::commons::enums::ErrorCode;
+using setu::planner::hints::CompilerHint;
+using setu::planner::hints::RoutingHint;
 //==============================================================================
 void InitClientPybindClass(py::module_& m) {
   py::class_<Client, std::shared_ptr<Client>>(m, "Client")
@@ -45,10 +48,26 @@ void InitClientPybindClass(py::module_& m) {
       .def("register_tensor_shard", &Client::RegisterTensorShard,
            py::arg("shard_spec"),
            "Register a tensor shard and return a reference to it")
-      .def("submit_copy", &Client::SubmitCopy, py::arg("copy_spec"),
-           "Submit a copy operation and return an operation ID")
-      .def("submit_pull", &Client::SubmitPull, py::arg("copy_spec"),
-           "Submit a pull operation and return an operation ID")
+      .def(
+          "submit_copy",
+          [](Client& self, const CopySpec& copy_spec,
+             const std::vector<RoutingHint>& hints) {
+            std::vector<CompilerHint> compiler_hints(hints.begin(),
+                                                     hints.end());
+            return self.SubmitCopy(copy_spec, compiler_hints);
+          },
+          py::arg("copy_spec"), py::arg("hints") = std::vector<RoutingHint>{},
+          "Submit a copy operation and return an operation ID")
+      .def(
+          "submit_pull",
+          [](Client& self, const CopySpec& copy_spec,
+             const std::vector<RoutingHint>& hints) {
+            std::vector<CompilerHint> compiler_hints(hints.begin(),
+                                                     hints.end());
+            return self.SubmitPull(copy_spec, compiler_hints);
+          },
+          py::arg("copy_spec"), py::arg("hints") = std::vector<RoutingHint>{},
+          "Submit a pull operation and return an operation ID")
       .def("wait_for_copy", &Client::WaitForCopy, py::arg("copy_op_id"),
            "Wait for a copy operation to complete")
       .def("wait_for_shard_allocation", &Client::WaitForShardAllocation,

csrc/setu/coordinator/Coordinator.cpp

@@ -60,9 +60,8 @@ Coordinator::Coordinator(std::size_t port, PlannerPtr planner)
 
   handler_ = std::make_unique<Handler>(inbox_queue_, outbox_queue_, metastore_,
                                        planner_queue_, outbox_notify);
-  executor_ =
-      std::make_unique<Executor>(planner_queue_, outbox_queue_, metastore_,
-                                 *planner_, hint_store_, outbox_notify);
+  executor_ = std::make_unique<Executor>(planner_queue_, outbox_queue_,
+                                         metastore_, *planner_, outbox_notify);
 }
 
 Coordinator::~Coordinator() {
@@ -108,12 +107,6 @@ std::optional<CopyOperationId> Coordinator::SubmitCopy(
   return std::nullopt;
 }
 
-void Coordinator::AddHint(setu::planner::hints::CompilerHint hint) {
-  hint_store_.AddHint(std::move(hint));
-}
-
-void Coordinator::ClearHints() { hint_store_.Clear(); }
-
 void Coordinator::PlanExecuted(CopyOperationId copy_op_id) {
   LOG_DEBUG("Plan executed for copy operation ID: {}", copy_op_id);
 
@@ -387,7 +380,8 @@ void Coordinator::Handler::HandleSubmitCopyRequest(
       metastore_.GetNumShardsForTensor(request.copy_spec.dst_name);
 
   HandleShardSubmission(node_agent_identity, request.request_id,
-                        request.shard_id, request.copy_spec, expected_shards);
+                        request.shard_id, request.copy_spec, expected_shards,
+                        std::vector(request.hints), request.hints_fingerprint);
 }
 
 void Coordinator::Handler::HandleSubmitPullRequest(
@@ -413,17 +407,38 @@ void Coordinator::Handler::HandleSubmitPullRequest(
       metastore_.GetNumShardsForTensor(request.copy_spec.dst_name);
 
   HandleShardSubmission(node_agent_identity, request.request_id,
-                        request.shard_id, request.copy_spec, expected_shards);
+                        request.shard_id, request.copy_spec, expected_shards,
+                        std::vector(request.hints), request.hints_fingerprint);
 }
 
 void Coordinator::Handler::HandleShardSubmission(
     const Identity& node_agent_identity, const RequestId& request_id,
     const ShardId& shard_id, const CopySpec& copy_spec,
-    std::size_t expected_shards) {
+    std::size_t expected_shards,
+    std::vector<setu::planner::hints::CompilerHint> hints,
+    std::uint64_t hints_fingerprint) {
   using setu::commons::utils::AggregationParticipant;
 
   CopyKey copy_key{copy_spec.src_name, copy_spec.dst_name};
 
+  // First-writer-wins hint storage: the first shard submission's hints
+  // become authoritative for this operation.
+  if (operation_hints_.find(copy_key) == operation_hints_.end()) {
+    // First shard for this operation — store its hints
+    operation_hints_[copy_key] = std::move(hints);
+    operation_fingerprints_[copy_key] = hints_fingerprint;
+  } else {
+    // Subsequent shard — verify fingerprint in debug mode
+    if (setu::commons::Logger::log_level <= setu::commons::LogLevel::kDebug) {
+      ASSERT_VALID_RUNTIME(
+          hints_fingerprint == operation_fingerprints_[copy_key],
+          "SPMD hint mismatch for {} -> {}: shard {} sent fingerprint {} but "
+          "first submission had {}",
+          copy_spec.src_name, copy_spec.dst_name, shard_id, hints_fingerprint,
+          operation_fingerprints_[copy_key]);
+    }
+  }
+
   auto result = shard_aggregator_.Submit(
       copy_key, shard_id, copy_spec,
       AggregationParticipant{node_agent_identity, request_id}, expected_shards,
@@ -465,8 +480,14 @@ void Coordinator::Handler::HandleShardSubmission(
   // Store the shared state (will be accessed by HandleExecuteResponse)
   copy_operations_.emplace(copy_op_id, state);
 
-  // Add to planner queue with copy_op_id and shared state
-  planner_queue_.push(PlannerTask{copy_op_id, result->payload, state});
+  // Extract per-operation hints and clean up tracking maps
+  auto op_hints = std::move(operation_hints_[copy_key]);
+  operation_hints_.erase(copy_key);
+  operation_fingerprints_.erase(copy_key);
+
+  // Add to planner queue with copy_op_id, shared state, and per-op hints
+  planner_queue_.push(PlannerTask{copy_op_id, result->payload, state,
+                                  HintStore(std::move(op_hints))});
 
   // Send responses to all waiting participants with copy_op_id
   for (const auto& participant : result->participants) {
@@ -583,6 +604,16 @@ void Coordinator::Handler::HandleDeregisterShardsRequest(
                tensor_names.contains(key.dst_name);
       });
 
+  // Clean up per-operation hint tracking for cancelled operations
+  std::erase_if(operation_hints_, [&tensor_names](const auto& entry) {
+    return tensor_names.contains(entry.first.src_name) ||
+           tensor_names.contains(entry.first.dst_name);
+  });
+  std::erase_if(operation_fingerprints_, [&tensor_names](const auto& entry) {
+    return tensor_names.contains(entry.first.src_name) ||
+           tensor_names.contains(entry.first.dst_name);
+  });
+
   // Send error responses to cancelled participants
   for (const auto& participant : cancelled_participants) {
     LOG_INFO(
@@ -637,13 +668,11 @@ void Coordinator::Handler::HandleDeregisterShardsRequest(
 Coordinator::Executor::Executor(Queue<PlannerTask>& planner_queue,
                                 Queue<OutboxMessage>& outbox_queue,
                                 MetaStore& metastore, Planner& planner,
-                                HintStore& hint_store,
                                 OutboxNotifyFn outbox_notify)
     : planner_queue_(planner_queue),
       outbox_queue_(outbox_queue),
       metastore_(metastore),
       planner_(planner),
-      hint_store_(hint_store),
       outbox_notify_(std::move(outbox_notify)) {}
 
 void Coordinator::Executor::PushOutbox(OutboxMessage msg) {
@@ -676,9 +705,8 @@ void Coordinator::Executor::Loop() {
 
       LOG_DEBUG("Executor received task for copy_op_id: {}", task.copy_op_id);
 
-      auto hints = hint_store_.Snapshot();
       auto t_compile_start = std::chrono::steady_clock::now();
-      Plan plan = planner_.Compile(task.copy_spec, metastore_, hints);
+      Plan plan = planner_.Compile(task.copy_spec, metastore_, task.hints);
       auto t_compile_end = std::chrono::steady_clock::now();
 
       LOG_DEBUG("Compiled plan:\n{}", plan);

csrc/setu/coordinator/Coordinator.h

@@ -106,9 +106,6 @@ class Coordinator {
 
   void PlanExecuted(CopyOperationId copy_op_id);
 
-  void AddHint(setu::planner::hints::CompilerHint hint);
-  void ClearHints();
-
   void Start();
   void Stop();
 
@@ -149,12 +146,13 @@ class Coordinator {
     CoordinatorMessage message;
   };
 
-  /// @brief Task for the planner containing CopyOperationId, CopySpec, and
-  /// shared state
+  /// @brief Task for the planner containing CopyOperationId, CopySpec,
+  /// shared state, and per-operation hints from the first shard submission.
   struct PlannerTask {
     CopyOperationId copy_op_id;
     CopySpec copy_spec;
     CopyOperationStatePtr state;  // Shared with Handler's copy_operations_ map
+    HintStore hints;              // Per-operation hints (first-writer-wins)
   };
 
   //============================================================================
@@ -233,11 +231,12 @@ class Coordinator {
         const setu::commons::messages::DeregisterShardsRequest& request);
 
     /// @brief Unified shard submission logic for both Copy and Pull.
-    void HandleShardSubmission(const Identity& node_agent_identity,
-                               const RequestId& request_id,
-                               const ShardId& shard_id,
-                               const CopySpec& copy_spec,
-                               std::size_t expected_shards);
+    void HandleShardSubmission(
+        const Identity& node_agent_identity, const RequestId& request_id,
+        const ShardId& shard_id, const CopySpec& copy_spec,
+        std::size_t expected_shards,
+        std::vector<setu::planner::hints::CompilerHint> hints,
+        std::uint64_t hints_fingerprint);
 
     /// Key for tracking copy operations by (src, dst) tensor pair
     struct CopyKey {
@@ -262,6 +261,13 @@ class Coordinator {
     /// address that
     setu::commons::utils::ShardAggregator<CopyKey, CopySpec> shard_aggregator_;
 
+    /// Per-operation hint storage: first shard's hints become authoritative
+    /// (first-writer-wins). Keyed by CopyKey, cleared when aggregation
+    /// completes.
+    std::map<CopyKey, std::vector<setu::planner::hints::CompilerHint>>
+        operation_hints_;
+    std::map<CopyKey, std::uint64_t> operation_fingerprints_;
+
     /// Maps CopyOperationId to shared CopyOperationState (includes submitters
     /// and completion tracking)
     std::map<CopyOperationId, CopyOperationStatePtr> copy_operations_;
@@ -285,8 +291,7 @@ class Coordinator {
   struct Executor {
     Executor(Queue<PlannerTask>& planner_queue,
              Queue<OutboxMessage>& outbox_queue, MetaStore& metastore,
-             Planner& planner, HintStore& hint_store,
-             OutboxNotifyFn outbox_notify);
+             Planner& planner, OutboxNotifyFn outbox_notify);
 
     void Start();
     void Stop();
@@ -300,7 +305,6 @@ class Coordinator {
     Queue<OutboxMessage>& outbox_queue_;
     MetaStore& metastore_;
     Planner& planner_;
-    HintStore& hint_store_;
     OutboxNotifyFn outbox_notify_;
 
     std::thread thread_;
@@ -313,7 +317,6 @@ class Coordinator {
 
   MetaStore metastore_;
   PlannerPtr planner_;
-  HintStore hint_store_;
 
   // Internal message queues
   Queue<InboxMessage> inbox_queue_;

csrc/setu/coordinator/Pybind.cpp

@@ -21,37 +21,18 @@
 #include "commons/TorchCommon.h"
 #include "coordinator/Coordinator.h"
 #include "metastore/Pybind.h"
-#include "planner/hints/Hint.h"
 //==============================================================================
 namespace setu::coordinator {
 //==============================================================================
 using setu::planner::PlannerPtr;
-using setu::planner::hints::CompilerHint;
-using setu::planner::hints::RoutingHint;
 //==============================================================================
 void InitCoordinatorPybindClass(py::module_& m) {
   py::class_<Coordinator, std::shared_ptr<Coordinator>>(m, "Coordinator")
       .def(py::init<std::size_t, PlannerPtr>(), py::arg("port"),
            py::arg("planner"),
            "Create a Coordinator with specified port and planner")
       .def("start", &Coordinator::Start, "Start the Coordinator loops")
-      .def("stop", &Coordinator::Stop, "Stop the Coordinator loops")
-      // TODO: Ideally we'd bind AddHint directly:
-      //   .def("add_hint", &Coordinator::AddHint)
-      // and let pybind11/stl.h auto-cast RoutingHint -> CompilerHint
-      // (std::variant). However, this fails at compile time —
-      // pybind11 can't default-construct the type_caster tuple for
-      // the member function pointer's arguments. Root cause is
-      // unclear, investigate later as this is not a blocking concern.
-      // Using a per-type lambda as a workaround; needs a new overload
-      // for each hint type added to CompilerHint.
-      .def(
-          "add_hint",
-          [](Coordinator& self, const RoutingHint& hint) {
-            self.AddHint(CompilerHint{hint});
-          },
-          py::arg("hint"), "Add a compiler hint (e.g. RoutingHint)")
-      .def("clear_hints", &Coordinator::ClearHints, "Clear all compiler hints");
+      .def("stop", &Coordinator::Stop, "Stop the Coordinator loops");
 }
 //==============================================================================
 }  // namespace setu::coordinator