From 1305f5d4e5534c864c795abc8429a801ee89dfab Mon Sep 17 00:00:00 2001
From: fangfengbin <869218239a@zju.edu.cn>
Date: Thu, 10 Dec 2020 03:54:23 +0800
Subject: [PATCH] [GCS]GCS based Actor Scheduling support actor colocation
 (#12707)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* [GCS]GCS based Actor Scheduling support actor colocation

* fix review comment

Co-authored-by: 灵洵 <fengbin.ffb@antgroup.com>
---
 src/ray/gcs/gcs_server/gcs_actor_scheduler.cc | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc b/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc
index 8e2329c7b..7c788275a 100644
--- a/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc
+++ b/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc
@@ -45,7 +45,18 @@ void GcsActorScheduler::Schedule(std::shared_ptr<GcsActor> actor) {
   RAY_CHECK(actor->GetNodeID().IsNil() && actor->GetWorkerID().IsNil());
 
   // Select a node to lease worker for the actor.
-  auto node = SelectNodeRandomly();
+  std::shared_ptr<rpc::GcsNodeInfo> node;
+
+  // If an actor is non-detached and has resource requirements, We will try to schedule it
+  // on the same node as the owner if possible.
+  const auto &task_spec = actor->GetCreationTaskSpecification();
+  if (!actor->IsDetached() && !task_spec.GetRequiredResources().IsEmpty()) {
+    auto maybe_node = gcs_node_manager_.GetNode(actor->GetOwnerNodeID());
+    node = maybe_node.has_value() ? maybe_node.value() : SelectNodeRandomly();
+  } else {
+    node = SelectNodeRandomly();
+  }
+
   if (node == nullptr) {
     // There are no available nodes to schedule the actor, so just trigger the failed
     // handler.