[Placement Group] Support RayPlacementGroupError #10508

java/runtime/src/main/java/io/ray/runtime/object/ObjectSerializer.java

@@ -33,6 +33,8 @@ public class ObjectSerializer {
       .valueOf(ErrorType.OBJECT_UNRECONSTRUCTABLE.getNumber()).getBytes();
   private static final byte[] TASK_EXECUTION_EXCEPTION_META = String
       .valueOf(ErrorType.TASK_EXECUTION_EXCEPTION.getNumber()).getBytes();
+  private static final byte[] PLACEMENT_GROUP_ERROR_META = String
+      .valueOf(ErrorType.PLACEMENT_GROUP_ERROR.getNumber()).getBytes();
 
   public static final byte[] OBJECT_METADATA_TYPE_CROSS_LANGUAGE = "XLANG".getBytes();
   public static final byte[] OBJECT_METADATA_TYPE_JAVA = "JAVA".getBytes();

python/ray/_raylet.pyx

@@ -102,6 +102,7 @@ from ray.exceptions import (
     RayError,
     RaySystemError,
     RayTaskError,
+    RayPlacementGroupError,
     ObjectStoreFullError,
     GetTimeoutError,
     TaskCancelledError

python/ray/exceptions.py

@@ -138,6 +138,13 @@ def __str__(self):
         return "\n".join(out)
 
 
+class RayPlacementGroupError(RayError):
+    """Indicates that the task failed due to placement group errors."""
+
+    def __str__(self):
+        return "The task failed due to placement group error."
+
+
 class WorkerCrashedError(RayError):
     """Indicates that the worker died unexpectedly while executing a task."""
 
@@ -212,6 +219,7 @@ class PlasmaObjectNotAvailable(RayError):
     PlasmaObjectNotAvailable,
     RayError,
     RayTaskError,
+    RayPlacementGroupError,
     WorkerCrashedError,
     RayActorError,
     ObjectStoreFullError,

python/ray/ray_constants.py

@@ -129,7 +129,8 @@ def to_memory_units(memory_bytes, round_up):
 REPORTER_DIED_ERROR = "reporter_died"
 DASHBOARD_AGENT_DIED_ERROR = "dashboard_agent_died"
 DASHBOARD_DIED_ERROR = "dashboard_died"
-RAYLET_CONNECTION_ERROR = "raylet_connection_error"
+RAYLET_CONNECTION_ERROR = "raylet_connection"
+PLACEMENT_GROUP_PUSH_ERROR = "placement_group"
 
 # Used in gpu detection
 RESOURCE_CONSTRAINT_PREFIX = "accelerator_type:"

python/ray/serialization.py

@@ -12,6 +12,7 @@
     RayError,
     PlasmaObjectNotAvailable,
     RayTaskError,
+    RayPlacementGroupError,
     RayActorError,
     TaskCancelledError,
     WorkerCrashedError,
@@ -277,6 +278,8 @@ def _deserialize_object(self, data, metadata, object_ref):
                 return TaskCancelledError()
             elif error_type == ErrorType.Value("OBJECT_UNRECONSTRUCTABLE"):
                 return ObjectLostError(ray.ObjectRef(object_ref.binary()))
+            elif error_type == ErrorType.Value("PLACEMENT_GROUP_ERROR"):
+                return RayPlacementGroupError()
             else:
                 assert error_type != ErrorType.Value("OBJECT_IN_PLASMA"), \
                     "Tried to get object that has been promoted to plasma."

src/ray/gcs/gcs_server/gcs_actor_manager.cc

@@ -596,7 +596,8 @@ void GcsActorManager::CollectStats() const {
 
 void GcsActorManager::OnWorkerDead(const ray::NodeID &node_id,
                                    const ray::WorkerID &worker_id,
-                                   bool intentional_exit) {
+                                   const rpc::ClientDisconnectType disconnect_type) {
+  bool intentional_exit = disconnect_type == rpc::ClientDisconnectType::FINISHED;
   if (intentional_exit) {
     RAY_LOG(INFO) << "Worker " << worker_id << " on node " << node_id
                   << " intentional exit.";

src/ray/gcs/gcs_server/gcs_actor_manager.h

@@ -247,10 +247,10 @@ class GcsActorManager : public rpc::ActorInfoHandler {
   ///
   /// \param node_id ID of the node where the dead worker was located.
   /// \param worker_id ID of the dead worker.
-  /// \param intentional_exit Whether the death was intentional. If yes and the
-  /// worker was an actor, we should not attempt to restart the actor.
+  /// \param disconnect_type the reason why the worker is disconnected.
   void OnWorkerDead(const NodeID &node_id, const WorkerID &worker_id,
-                    bool intentional_exit = false);
+                    rpc::ClientDisconnectType disconnect_type =
+                        rpc::ClientDisconnectType::UNEXPECTED_EXIT);
 
   /// Handle actor creation task failure. This should be called when scheduling
   /// an actor creation task is infeasible.

src/ray/gcs/gcs_server/gcs_server.cc

@@ -295,7 +295,7 @@ void GcsServer::InstallEventListeners() {
         auto worker_id = WorkerID::FromBinary(worker_address.worker_id());
         auto node_id = NodeID::FromBinary(worker_address.raylet_id());
         gcs_actor_manager_->OnWorkerDead(node_id, worker_id,
-                                         worker_failure_data->intentional_disconnect());
+                                         worker_failure_data->disconnect_type());
       });
 
   // Install job event listeners.

src/ray/gcs/gcs_server/gcs_worker_manager.cc

@@ -28,7 +28,7 @@ void GcsWorkerManager::HandleReportWorkerFailure(
   log_stream << "Reporting worker failure, worker id = " << worker_id
              << ", node id = " << node_id
              << ", address = " << worker_address.ip_address();
-  if (request.worker_failure().intentional_disconnect()) {
+  if (request.worker_failure().disconnect_type() == rpc::ClientDisconnectType::FINISHED) {
     RAY_LOG(INFO) << log_stream.str();
   } else {
     RAY_LOG(WARNING) << log_stream.str()

src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc

@@ -725,7 +725,7 @@ TEST_F(GcsActorManagerTest, TestRaceConditionCancelLease) {
   actor->UpdateAddress(address);
   const auto actor_id = actor->GetActorID();
   EXPECT_CALL(*mock_actor_scheduler_, CancelOnLeasing(node_id, actor_id));
-  gcs_actor_manager_->OnWorkerDead(owner_node_id, owner_worker_id, false);
+  gcs_actor_manager_->OnWorkerDead(owner_node_id, owner_worker_id);
 }
 
 TEST_F(GcsActorManagerTest, TestRegisterActor) {
@@ -848,10 +848,10 @@ TEST_F(GcsActorManagerTest, TestOwnerAndChildDiedAtTheSameTimeRaceCondition) {
   const auto child_worker_id = actor->GetWorkerID();
   const auto actor_id = actor->GetActorID();
   // Make worker & owner fail at the same time, but owner's failure comes first.
-  gcs_actor_manager_->OnWorkerDead(owner_node_id, owner_worker_id, false);
+  gcs_actor_manager_->OnWorkerDead(owner_node_id, owner_worker_id);
   EXPECT_CALL(*mock_actor_scheduler_, CancelOnWorker(child_node_id, child_worker_id))
       .WillOnce(Return(actor_id));
-  gcs_actor_manager_->OnWorkerDead(child_node_id, child_worker_id, false);
+  gcs_actor_manager_->OnWorkerDead(child_node_id, child_worker_id);
 }
 
 }  // namespace ray

src/ray/gcs/pb_util.h

@@ -87,14 +87,15 @@ inline std::shared_ptr<ray::rpc::ActorTableData> CreateActorTableData(
 inline std::shared_ptr<ray::rpc::WorkerTableData> CreateWorkerFailureData(
     const NodeID &raylet_id, const WorkerID &worker_id, const std::string &address,
     int32_t port, int64_t timestamp = std::time(nullptr),
-    bool intentional_disconnect = false) {
+    const rpc::ClientDisconnectType disconnect_type =
+        rpc::ClientDisconnectType::UNEXPECTED_EXIT) {
   auto worker_failure_info_ptr = std::make_shared<ray::rpc::WorkerTableData>();
   worker_failure_info_ptr->mutable_worker_address()->set_raylet_id(raylet_id.Binary());
   worker_failure_info_ptr->mutable_worker_address()->set_worker_id(worker_id.Binary());
   worker_failure_info_ptr->mutable_worker_address()->set_ip_address(address);
   worker_failure_info_ptr->mutable_worker_address()->set_port(port);
   worker_failure_info_ptr->set_timestamp(timestamp);
-  worker_failure_info_ptr->set_intentional_disconnect(intentional_disconnect);
+  worker_failure_info_ptr->set_disconnect_type(disconnect_type);
   return worker_failure_info_ptr;
 }
 

src/ray/protobuf/common.proto

@@ -133,6 +133,8 @@ enum ErrorType {
   TASK_CANCELLED = 5;
   // Inidicates that creating the GCS service failed to create the actor.
   ACTOR_CREATION_FAILED = 6;
+  // Indicates failure due to placement group
+  PLACEMENT_GROUP_ERROR = 7;
 }
 
 /// The task exception encapsulates all information about task
@@ -437,3 +439,15 @@ message MetricPoint {
   // [Optional] Unit of the metric.
   string units = 6;
 }
+
+// Type of a worker exit.
+enum ClientDisconnectType {
+  // Unintended worker exit.
+  UNEXPECTED_EXIT = 0;
+  // Intended worker exit.
+  FINISHED = 1;
+  // Worker exit due to resource bundle release.
+  UNUSED_RESOURCE_RELEASED = 2;
+  // Worker exit due to placement group removal.
+  PLACEGROUP_REMOVED = 3;
+}

src/ray/protobuf/gcs.proto

@@ -399,8 +399,8 @@ message WorkerTableData {
   Address worker_address = 2;
   // The UNIX timestamp at which this worker's state was updated.
   int64 timestamp = 3;
-  // Whether it's an intentional disconnect, only applies then `is_alive` is false.
-  bool intentional_disconnect = 4;
+  // The type of worker exit.
+  ClientDisconnectType disconnect_type = 4;
   // Type of this worker.
   WorkerType worker_type = 5;
   // This is for AddWorker.

src/ray/raylet/format/node_manager.fbs

@@ -35,12 +35,9 @@ enum MessageType:int {
   RegisterClientReply,
   // Send the worker's gRPC port to the raylet.
   AnnounceWorkerPort,
-  // Notify the raylet that this client is disconnecting unexpectedly.
+  // Notify the raylet that this client is disconnecting.
   // This is sent from a worker to a raylet.
   DisconnectClient,
-  // Notify the raylet that this client is disconnecting gracefully.
-  // This is sent from a worker to a raylet.
-  IntentionalDisconnectClient,
   // Tell a worker to execute a task. This is sent from a raylet to a
   // worker.
   ExecuteTask,
@@ -111,6 +108,7 @@ table ResourceIdSetInfo {
 
 // This message is sent from a worker to the node manager.
 table DisconnectClient {
+  disconnect_type: int;
 }
 
 table ResourceIdSetInfos {