[core] Disable GCS actor management (#8271)

.travis.yml

@@ -32,6 +32,7 @@ matrix:
         - JDK='Oracle JDK 8'
         - PYTHON=3.6 PYTHONWARNINGS=ignore
         - RAY_INSTALL_JAVA=1
+        - RAY_GCS_ACTOR_SERVICE_ENABLED=true
       before_install:
         - . ./ci/travis/ci.sh init RAY_CI_JAVA_AFFECTED
       install:
@@ -45,6 +46,7 @@ matrix:
         - TESTSUITE=streaming
         - JDK='Oracle JDK 8'
         - RAY_INSTALL_JAVA=1
+        - RAY_GCS_ACTOR_SERVICE_ENABLED=true
         - PYTHON=3.6 PYTHONWARNINGS=ignore
       before_install:
         - . ./ci/travis/ci.sh init RAY_CI_STREAMING_PYTHON_AFFECTED,RAY_CI_STREAMING_JAVA_AFFECTED
@@ -144,6 +146,7 @@ matrix:
         - LINUX_WHEELS=1
         - PYTHONWARNINGS=ignore
         - RAY_INSTALL_JAVA=1
+        - RAY_GCS_ACTOR_SERVICE_ENABLED=true
       before_install:
         - . ./ci/travis/ci.sh init RAY_CI_LINUX_WHEELS_AFFECTED
       install:
@@ -161,6 +164,7 @@ matrix:
         - MAC_WHEELS=1
         - PYTHONWARNINGS=ignore
         - RAY_INSTALL_JAVA=1
+        - RAY_GCS_ACTOR_SERVICE_ENABLED=true
       before_install:
         - . ./ci/travis/ci.sh init RAY_CI_MACOS_WHEELS_AFFECTED
       install:

src/ray/common/ray_config_def.h

@@ -289,3 +289,7 @@ RAY_CONFIG(uint32_t, cancellation_retry_ms, 2000)
 RAY_CONFIG(bool, gcs_service_enabled,
            getenv("RAY_GCS_SERVICE_ENABLED") == nullptr ||
                getenv("RAY_GCS_SERVICE_ENABLED") == std::string("true"))
+
+RAY_CONFIG(bool, gcs_actor_service_enabled,
+           getenv("RAY_GCS_ACTOR_SERVICE_ENABLED") != nullptr &&
+               getenv("RAY_GCS_ACTOR_SERVICE_ENABLED") == std::string("true"))

src/ray/core_worker/core_worker.cc

@@ -413,7 +413,8 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_
 
   std::function<Status(const TaskSpecification &, const gcs::StatusCallback &)>
       actor_create_callback = nullptr;
-  if (RayConfig::instance().gcs_service_enabled()) {
+  if (RayConfig::instance().gcs_service_enabled() &&
+      RayConfig::instance().gcs_actor_service_enabled()) {
     actor_create_callback = [this](const TaskSpecification &task_spec,
                                    const gcs::StatusCallback &callback) {
       return gcs_client_->Actors().AsyncCreateActor(task_spec, callback);

src/ray/raylet/node_manager.cc

@@ -805,7 +805,8 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id,
   if (it == actor_registry_.end()) {
     it = actor_registry_.emplace(actor_id, actor_registration).first;
   } else {
-    if (RayConfig::instance().gcs_service_enabled()) {
+    if (RayConfig::instance().gcs_service_enabled() &&
+        RayConfig::instance().gcs_actor_service_enabled()) {
       it->second = actor_registration;
     } else {
       // Only process the state transition if it is to a later state than ours.
@@ -874,7 +875,8 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id,
     }
   } else if (actor_registration.GetState() == ActorTableData::RECONSTRUCTING) {
     RAY_LOG(DEBUG) << "Actor is being reconstructed: " << actor_id;
-    if (!RayConfig::instance().gcs_service_enabled()) {
+    if (!(RayConfig::instance().gcs_service_enabled() &&
+          RayConfig::instance().gcs_actor_service_enabled())) {
       // The actor is dead and needs reconstruction. Attempting to reconstruct its
       // creation task.
       reconstruction_policy_.ListenAndMaybeReconstruct(
@@ -1127,7 +1129,8 @@ void NodeManager::ProcessRegisterClientRequestMessage(
 
 void NodeManager::HandleDisconnectedActor(const ActorID &actor_id, bool was_local,
                                           bool intentional_disconnect) {
-  if (RayConfig::instance().gcs_service_enabled()) {
+  if (RayConfig::instance().gcs_service_enabled() &&
+      RayConfig::instance().gcs_actor_service_enabled()) {
     // If gcs actor management is enabled, the gcs will take over the status change of all
     // actors.
     return;
@@ -2677,7 +2680,8 @@ void NodeManager::FinishAssignedActorTask(Worker &worker, const Task &task) {
       worker.MarkDetachedActor();
     }
 
-    if (RayConfig::instance().gcs_service_enabled()) {
+    if (RayConfig::instance().gcs_service_enabled() &&
+        RayConfig::instance().gcs_actor_service_enabled()) {
       // Gcs server is responsible for notifying other nodes of the changes of actor
       // status, and thus raylet doesn't need to handle this anymore.
       // And if `new_scheduler_enabled_` is true, this function `FinishAssignedActorTask`