diff --git a/.travis.yml b/.travis.yml index 6ea7ac3b3..5b70be81a 100644 --- a/.travis.yml +++ b/.travis.yml @@ -88,38 +88,6 @@ matrix: - if [ $RAY_CI_STREAMING_PYTHON_AFFECTED == "1" ]; then python -m pytest -v --durations=5 --timeout=300 streaming/python/tests/; fi - if [ $RAY_CI_STREAMING_JAVA_AFFECTED == "1" ]; then ./streaming/java/test.sh; fi - - os: linux - env: - - TESTSUITE=gcs_service_disabled - - JDK='Oracle JDK 8' - - PYTHON=3.6 PYTHONWARNINGS=ignore - - RAY_INSTALL_JAVA=1 - - RAY_GCS_SERVICE_ENABLED=false - - RAY_CYTHON_EXAMPLES=1 - install: - - . ./ci/travis/ci.sh init RAY_CI_ONLY_RLLIB_AFFECTED - before_script: - - . ./ci/travis/ci.sh build - script: - - ./ci/suppress_output bash streaming/src/test/run_streaming_queue_test.sh - - ./java/test.sh - - - os: linux - env: - - TESTSUITE=gcs_service_disabled_python_testcase - - JDK='Oracle JDK 8' - - PYTHON=3.6 PYTHONWARNINGS=ignore - - RAY_INSTALL_JAVA=1 - - RAY_GCS_SERVICE_ENABLED=false - - RAY_CYTHON_EXAMPLES=1 - - RAY_USE_RANDOM_PORTS=1 - install: - - . ./ci/travis/ci.sh init RAY_CI_ONLY_RLLIB_AFFECTED - before_script: - - . ./ci/travis/ci.sh build - script: - - ./ci/keep_alive bazel test --config=ci --test_tag_filters=-jenkins_only python/ray/tests/... - - os: linux env: - LINT=1 diff --git a/java/runtime/src/main/java/io/ray/runtime/config/RayConfig.java b/java/runtime/src/main/java/io/ray/runtime/config/RayConfig.java index 48be365ca..0f9f9f96b 100644 --- a/java/runtime/src/main/java/io/ray/runtime/config/RayConfig.java +++ b/java/runtime/src/main/java/io/ray/runtime/config/RayConfig.java @@ -68,8 +68,6 @@ public class RayConfig { public final String jobResourcePath; public final String pythonWorkerCommand; - public final boolean gcsServiceEnabled; - private static volatile RayConfig instance = null; public static RayConfig getInstance() { @@ -210,9 +208,6 @@ public class RayConfig { numWorkersPerProcess = config.getInt("ray.raylet.config.num_workers_per_process_java"); - gcsServiceEnabled = System.getenv("RAY_GCS_SERVICE_ENABLED") == null || - System.getenv("RAY_GCS_SERVICE_ENABLED").toLowerCase().equals("true"); - // Validate config. validate(); } diff --git a/java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java b/java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java index 58d27fe6c..294646f06 100644 --- a/java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java +++ b/java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java @@ -9,7 +9,6 @@ import io.ray.api.id.JobId; import io.ray.api.id.TaskId; import io.ray.api.id.UniqueId; import io.ray.api.runtimecontext.NodeInfo; -import io.ray.runtime.config.RayConfig; import io.ray.runtime.generated.Gcs; import io.ray.runtime.generated.Gcs.ActorCheckpointIdData; import io.ray.runtime.generated.Gcs.GcsNodeInfo; @@ -119,9 +118,6 @@ public class GcsClient { public boolean wasCurrentActorRestarted(ActorId actorId) { byte[] key = ArrayUtils.addAll(TablePrefix.ACTOR.toString().getBytes(), actorId.getBytes()); - if (!RayConfig.getInstance().gcsServiceEnabled) { - return primary.exists(key); - } // TODO(ZhuSenlin): Get the actor table data from CoreWorker later. byte[] value = globalStateAccessor.getActorInfo(actorId); diff --git a/java/runtime/src/main/java/io/ray/runtime/runner/RunManager.java b/java/runtime/src/main/java/io/ray/runtime/runner/RunManager.java index 7290bedaa..0beadb003 100644 --- a/java/runtime/src/main/java/io/ray/runtime/runner/RunManager.java +++ b/java/runtime/src/main/java/io/ray/runtime/runner/RunManager.java @@ -229,28 +229,26 @@ public class RunManager { } // start gcs server - if (rayConfig.gcsServiceEnabled) { - String redisPasswordOption = ""; - if (!Strings.isNullOrEmpty(rayConfig.headRedisPassword)) { - redisPasswordOption = rayConfig.headRedisPassword; - } - - // See `src/ray/gcs/gcs_server/gcs_server_main.cc` for the meaning of each parameter. - final File gcsServerFile = BinaryFileUtil.getFile( - rayConfig.sessionDir, BinaryFileUtil.GCS_SERVER_BINARY_NAME); - Preconditions.checkState(gcsServerFile.setExecutable(true)); - List command = ImmutableList.of( - gcsServerFile.getAbsolutePath(), - String.format("--redis_address=%s", rayConfig.getRedisIp()), - String.format("--redis_port=%d", rayConfig.getRedisPort()), - String.format("--config_list=%s", - rayConfig.rayletConfigParameters.entrySet().stream() - .map(entry -> entry.getKey() + "," + entry.getValue()).collect(Collectors - .joining(","))), - String.format("--redis_password=%s", redisPasswordOption) - ); - startProcess(command, null, "gcs_server"); + String redisPasswordOption = ""; + if (!Strings.isNullOrEmpty(rayConfig.headRedisPassword)) { + redisPasswordOption = rayConfig.headRedisPassword; } + + // See `src/ray/gcs/gcs_server/gcs_server_main.cc` for the meaning of each parameter. + final File gcsServerFile = BinaryFileUtil.getFile( + rayConfig.sessionDir, BinaryFileUtil.GCS_SERVER_BINARY_NAME); + Preconditions.checkState(gcsServerFile.setExecutable(true)); + List command = ImmutableList.of( + gcsServerFile.getAbsolutePath(), + String.format("--redis_address=%s", rayConfig.getRedisIp()), + String.format("--redis_port=%d", rayConfig.getRedisPort()), + String.format("--config_list=%s", + rayConfig.rayletConfigParameters.entrySet().stream() + .map(entry -> entry.getKey() + "," + entry.getValue()).collect(Collectors + .joining(","))), + String.format("--redis_password=%s", redisPasswordOption) + ); + startProcess(command, null, "gcs_server"); } private String startRedisInstance(String ip, int port, String password, Integer shard) { diff --git a/python/ray/_raylet.pyx b/python/ray/_raylet.pyx index d3a65bd0b..5be509a7d 100644 --- a/python/ray/_raylet.pyx +++ b/python/ray/_raylet.pyx @@ -117,7 +117,6 @@ logger = logging.getLogger(__name__) def gcs_actor_service_enabled(): return ( - RayConfig.instance().gcs_service_enabled() and RayConfig.instance().gcs_actor_service_enabled()) diff --git a/python/ray/includes/ray_config.pxd b/python/ray/includes/ray_config.pxd index 12c0ae4a2..1837cd1de 100644 --- a/python/ray/includes/ray_config.pxd +++ b/python/ray/includes/ray_config.pxd @@ -85,6 +85,4 @@ cdef extern from "ray/common/ray_config.h" nogil: int64_t max_direct_call_object_size() const - c_bool gcs_service_enabled() const - c_bool gcs_actor_service_enabled() const diff --git a/python/ray/node.py b/python/ray/node.py index a4f373cd8..51e4997b0 100644 --- a/python/ray/node.py +++ b/python/ray/node.py @@ -662,22 +662,6 @@ class Node: assert ray_constants.PROCESS_TYPE_MONITOR not in self.all_processes self.all_processes[ray_constants.PROCESS_TYPE_MONITOR] = [process_info] - def start_raylet_monitor(self): - """Start the raylet monitor.""" - stdout_file, stderr_file = self.new_log_files("raylet_monitor") - process_info = ray.services.start_raylet_monitor( - self._redis_address, - stdout_file=stdout_file, - stderr_file=stderr_file, - redis_password=self._ray_params.redis_password, - config=self._config, - fate_share=self.kernel_fate_share) - assert (ray_constants.PROCESS_TYPE_RAYLET_MONITOR not in - self.all_processes) - self.all_processes[ray_constants.PROCESS_TYPE_RAYLET_MONITOR] = [ - process_info, - ] - def start_head_processes(self): """Start head processes on the node.""" logger.debug( @@ -687,10 +671,7 @@ class Node: # If this is the head node, start the relevant head node processes. self.start_redis() - if ray_constants.GCS_SERVICE_ENABLED: - self.start_gcs_server() - else: - self.start_raylet_monitor() + self.start_gcs_server() self.start_monitor() @@ -876,16 +857,6 @@ class Node: self._kill_process_type( ray_constants.PROCESS_TYPE_GCS_SERVER, check_alive=check_alive) - def kill_raylet_monitor(self, check_alive=True): - """Kill the raylet monitor. - - Args: - check_alive (bool): Raise an exception if the process was already - dead. - """ - self._kill_process_type( - ray_constants.PROCESS_TYPE_RAYLET_MONITOR, check_alive=check_alive) - def kill_reaper(self, check_alive=True): """Kill the reaper process. diff --git a/python/ray/ray_constants.py b/python/ray/ray_constants.py index 744409051..721e6d024 100644 --- a/python/ray/ray_constants.py +++ b/python/ray/ray_constants.py @@ -197,11 +197,6 @@ NODE_DEFAULT_IP = "127.0.0.1" # The Mach kernel page size in bytes. MACH_PAGE_SIZE_BYTES = 4096 -# RAY_GCS_SERVICE_ENABLED only set in ci job. -# TODO(ffbin): Once we entirely migrate to service-based GCS, we should -# remove it. -GCS_SERVICE_ENABLED = env_bool("RAY_GCS_SERVICE_ENABLED", True) - # Max 64 bit integer value, which is needed to ensure against overflow # in C++ when passing integer values cross-language. MAX_INT64_VALUE = 9223372036854775807 diff --git a/python/ray/tests/test_component_failures_3.py b/python/ray/tests/test_component_failures_3.py index da3678731..b910f22d3 100644 --- a/python/ray/tests/test_component_failures_3.py +++ b/python/ray/tests/test_component_failures_3.py @@ -83,10 +83,7 @@ def test_driver_lives_sequential(ray_start_regular): ray.worker._global_node.kill_plasma_store() ray.worker._global_node.kill_log_monitor() ray.worker._global_node.kill_monitor() - if ray_constants.GCS_SERVICE_ENABLED: - ray.worker._global_node.kill_gcs_server() - else: - ray.worker._global_node.kill_raylet_monitor() + ray.worker._global_node.kill_gcs_server() # If the driver can reach the tearDown method, then it is still alive. @@ -97,19 +94,11 @@ def test_driver_lives_sequential(ray_start_regular): def test_driver_lives_parallel(ray_start_regular): all_processes = ray.worker._global_node.all_processes - if ray_constants.GCS_SERVICE_ENABLED: - process_infos = (all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] - + all_processes[ray_constants.PROCESS_TYPE_GCS_SERVER] - + all_processes[ray_constants.PROCESS_TYPE_RAYLET] + - all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] - + all_processes[ray_constants.PROCESS_TYPE_MONITOR]) - else: - process_infos = ( - all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] + - all_processes[ray_constants.PROCESS_TYPE_RAYLET] + - all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] + - all_processes[ray_constants.PROCESS_TYPE_MONITOR] + - all_processes[ray_constants.PROCESS_TYPE_RAYLET_MONITOR]) + process_infos = (all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] + + all_processes[ray_constants.PROCESS_TYPE_GCS_SERVER] + + all_processes[ray_constants.PROCESS_TYPE_RAYLET] + + all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] + + all_processes[ray_constants.PROCESS_TYPE_MONITOR]) assert len(process_infos) == 5 # Kill all the components in parallel. diff --git a/python/ray/tests/test_multinode_failures_2.py b/python/ray/tests/test_multinode_failures_2.py index 3a86d89a3..6b05a017d 100644 --- a/python/ray/tests/test_multinode_failures_2.py +++ b/python/ray/tests/test_multinode_failures_2.py @@ -133,10 +133,7 @@ def test_driver_lives_sequential(ray_start_regular): ray.worker._global_node.kill_plasma_store() ray.worker._global_node.kill_log_monitor() ray.worker._global_node.kill_monitor() - if ray_constants.GCS_SERVICE_ENABLED: - ray.worker._global_node.kill_gcs_server() - else: - ray.worker._global_node.kill_raylet_monitor() + ray.worker._global_node.kill_gcs_server() # If the driver can reach the tearDown method, then it is still alive. @@ -146,19 +143,12 @@ def test_driver_lives_sequential(ray_start_regular): reason="Hanging with new GCS API.") def test_driver_lives_parallel(ray_start_regular): all_processes = ray.worker._global_node.all_processes - if ray_constants.GCS_SERVICE_ENABLED: - process_infos = (all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] - + all_processes[ray_constants.PROCESS_TYPE_GCS_SERVER] - + all_processes[ray_constants.PROCESS_TYPE_RAYLET] + - all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] - + all_processes[ray_constants.PROCESS_TYPE_MONITOR]) - else: - process_infos = ( - all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] + - all_processes[ray_constants.PROCESS_TYPE_RAYLET] + - all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] + - all_processes[ray_constants.PROCESS_TYPE_MONITOR] + - all_processes[ray_constants.PROCESS_TYPE_RAYLET_MONITOR]) + + process_infos = (all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] + + all_processes[ray_constants.PROCESS_TYPE_GCS_SERVER] + + all_processes[ray_constants.PROCESS_TYPE_RAYLET] + + all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] + + all_processes[ray_constants.PROCESS_TYPE_MONITOR]) assert len(process_infos) == 5 # Kill all the components in parallel. diff --git a/python/ray/tests/test_tempfile.py b/python/ray/tests/test_tempfile.py index 160cffbff..17f533e0d 100644 --- a/python/ray/tests/test_tempfile.py +++ b/python/ray/tests/test_tempfile.py @@ -6,7 +6,6 @@ import time import pytest import ray -import ray.ray_constants as ray_constants from ray.cluster_utils import Cluster @@ -154,10 +153,7 @@ def test_raylet_tempfiles(shutdown_only): "raylet.err" } - if ray_constants.GCS_SERVICE_ENABLED: - log_files_expected.update({"gcs_server.out", "gcs_server.err"}) - else: - log_files_expected.update({"raylet_monitor.out", "raylet_monitor.err"}) + log_files_expected.update({"gcs_server.out", "gcs_server.err"}) assert log_files.issuperset(log_files_expected) diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index 4fe255a9a..4627a7f29 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -300,15 +300,6 @@ RAY_CONFIG(int32_t, ping_gcs_rpc_server_max_retries, 600) // Whether start the Plasma Store as a Raylet thread. RAY_CONFIG(bool, plasma_store_as_thread, false) -/// Whether to enable gcs service. -/// RAY_GCS_SERVICE_ENABLED is an env variable which only set in ci job. -/// If the value of RAY_GCS_SERVICE_ENABLED is false, we will disable gcs service, -/// otherwise gcs service is enabled. -/// TODO(ffbin): Once we entirely migrate to service-based GCS, we should remove it. -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")) diff --git a/src/ray/common/test_util.cc b/src/ray/common/test_util.cc index cde9d3de5..ab6b97e3f 100644 --- a/src/ray/common/test_util.cc +++ b/src/ray/common/test_util.cc @@ -157,22 +157,6 @@ void TestSetupUtil::StopRaylet(const std::string &raylet_socket_name) { KillProcessBySocketName(raylet_socket_name); } -std::string TestSetupUtil::StartRayletMonitor(const std::string &redis_address) { - std::string raylet_monitor_socket_name = ray::JoinPaths( - ray::GetUserTempDir(), "raylet_monitor" + ObjectID::FromRandom().Hex() + ".pid"); - std::vector cmdargs({TEST_RAYLET_MONITOR_EXEC_PATH, - "--redis_address=" + redis_address, - "--redis_port=6379"}); - RAY_LOG(DEBUG) << "Raylet monitor Start command: " << CreateCommandLine(cmdargs); - RAY_CHECK(!Process::Spawn(cmdargs, true, raylet_monitor_socket_name).second); - usleep(200 * 1000); - return raylet_monitor_socket_name; -} - -void TestSetupUtil::StopRayletMonitor(const std::string &raylet_monitor_socket_name) { - KillProcessBySocketName(raylet_monitor_socket_name); -} - bool WaitForCondition(std::function condition, int timeout_ms) { int wait_time = 0; while (true) { diff --git a/src/ray/common/test_util.h b/src/ray/common/test_util.h index b48ca586c..4fb2e9332 100644 --- a/src/ray/common/test_util.h +++ b/src/ray/common/test_util.h @@ -106,9 +106,6 @@ class TestSetupUtil { const std::string &resource); static void StopRaylet(const std::string &raylet_socket_name); - static std::string StartRayletMonitor(const std::string &redis_address); - static void StopRayletMonitor(const std::string &raylet_monitor_socket_name); - private: static int StartUpRedisServer(const int &port); static void ShutDownRedisServer(const int &port); diff --git a/src/ray/core_worker/core_worker.cc b/src/ray/core_worker/core_worker.cc index e39aed379..bb6e119b7 100644 --- a/src/ray/core_worker/core_worker.cc +++ b/src/ray/core_worker/core_worker.cc @@ -320,11 +320,8 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_ << ", raylet " << local_raylet_id; // Initialize gcs client. - if (RayConfig::instance().gcs_service_enabled()) { - gcs_client_ = std::make_shared(options_.gcs_options); - } else { - gcs_client_ = std::make_shared(options_.gcs_options); - } + gcs_client_ = std::make_shared(options_.gcs_options); + RAY_CHECK_OK(gcs_client_->Connect(io_service_)); RegisterToGcs(); @@ -434,8 +431,7 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_ std::function actor_create_callback = nullptr; - if (RayConfig::instance().gcs_service_enabled() && - RayConfig::instance().gcs_actor_service_enabled()) { + if (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); @@ -1332,8 +1328,7 @@ bool CoreWorker::AddActorHandle(std::unique_ptr actor_handle, // If we own the actor and the actor handle is no longer in scope, // terminate the actor. We do not do this if the GCS service is // enabled since then the GCS will terminate the actor for us. - if (!(RayConfig::instance().gcs_service_enabled() && - RayConfig::instance().gcs_actor_service_enabled())) { + if (!RayConfig::instance().gcs_actor_service_enabled()) { RAY_LOG(INFO) << "Owner's handle and creation ID " << object_id << " has gone out of scope, sending message to actor " << actor_id << " to do a clean exit."; @@ -1373,7 +1368,6 @@ Status CoreWorker::GetActorHandle(const ActorID &actor_id, Status CoreWorker::GetNamedActorHandle(const std::string &name, ActorHandle **actor_handle) { - RAY_CHECK(RayConfig::instance().gcs_service_enabled()); RAY_CHECK(RayConfig::instance().gcs_actor_service_enabled()); RAY_CHECK(!name.empty()); diff --git a/src/ray/core_worker/test/core_worker_test.cc b/src/ray/core_worker/test/core_worker_test.cc index 54af54774..2242bfcfe 100644 --- a/src/ray/core_worker/test/core_worker_test.cc +++ b/src/ray/core_worker/test/core_worker_test.cc @@ -105,13 +105,7 @@ class CoreWorkerTest : public ::testing::Test { } // start gcs server - if (RayConfig::instance().gcs_service_enabled()) { - gcs_server_socket_name_ = TestSetupUtil::StartGcsServer("127.0.0.1"); - } else { - // core worker test relies on node resources. It's important that one raylet can - // receive the heartbeat from another. So starting raylet monitor is required here. - raylet_monitor_socket_name_ = TestSetupUtil::StartRayletMonitor("127.0.0.1"); - } + gcs_server_socket_name_ = TestSetupUtil::StartGcsServer("127.0.0.1"); // start raylet on each node. Assign each node with different resources so that // a task can be scheduled to the desired node. @@ -131,10 +125,6 @@ class CoreWorkerTest : public ::testing::Test { TestSetupUtil::StopObjectStore(store_socket_name); } - if (!raylet_monitor_socket_name_.empty()) { - TestSetupUtil::StopRayletMonitor(raylet_monitor_socket_name_); - } - if (!gcs_server_socket_name_.empty()) { TestSetupUtil::StopGcsServer(gcs_server_socket_name_); } @@ -211,7 +201,6 @@ class CoreWorkerTest : public ::testing::Test { int num_nodes_; std::vector raylet_socket_names_; std::vector raylet_store_socket_names_; - std::string raylet_monitor_socket_name_; gcs::GcsClientOptions gcs_options_; std::string gcs_server_socket_name_; }; diff --git a/src/ray/gcs/redis_gcs_client.cc b/src/ray/gcs/redis_gcs_client.cc index ed82a0595..ef5c507cf 100644 --- a/src/ray/gcs/redis_gcs_client.cc +++ b/src/ray/gcs/redis_gcs_client.cc @@ -69,12 +69,7 @@ Status RedisGcsClient::Connect(boost::asio::io_service &io_service) { actor_checkpoint_id_table_.reset(new ActorCheckpointIdTable(shard_contexts, this)); resource_table_.reset(new DynamicResourceTable({primary_context}, this)); worker_failure_table_.reset(new WorkerFailureTable(shard_contexts, this)); - - if (RayConfig::instance().gcs_service_enabled()) { - actor_accessor_.reset(new RedisActorInfoAccessor(this)); - } else { - actor_accessor_.reset(new RedisLogBasedActorInfoAccessor(this)); - } + actor_accessor_.reset(new RedisActorInfoAccessor(this)); job_accessor_.reset(new RedisJobInfoAccessor(this)); object_accessor_.reset(new RedisObjectInfoAccessor(this)); node_accessor_.reset(new RedisNodeInfoAccessor(this)); diff --git a/src/ray/raylet/main.cc b/src/ray/raylet/main.cc index 83c445f00..91a0592dd 100644 --- a/src/ray/raylet/main.cc +++ b/src/ray/raylet/main.cc @@ -195,11 +195,8 @@ int main(int argc, char *argv[]) { ray::gcs::GcsClientOptions client_options(redis_address, redis_port, redis_password); std::shared_ptr gcs_client; - if (RayConfig::instance().gcs_service_enabled()) { - gcs_client = std::make_shared(client_options); - } else { - gcs_client = std::make_shared(client_options); - } + gcs_client = std::make_shared(client_options); + RAY_CHECK_OK(gcs_client->Connect(main_service)); std::unique_ptr server(new ray::raylet::Raylet( diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index ff3878651..db16a062d 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -810,8 +810,7 @@ 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() && - RayConfig::instance().gcs_actor_service_enabled()) { + if (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. @@ -878,8 +877,7 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id, } } else if (actor_registration.GetState() == ActorTableData::RESTARTING) { RAY_LOG(DEBUG) << "Actor is being restarted: " << actor_id; - if (!(RayConfig::instance().gcs_service_enabled() && - RayConfig::instance().gcs_actor_service_enabled())) { + if (!RayConfig::instance().gcs_actor_service_enabled()) { // The actor is dead and needs reconstruction. Attempting to reconstruct its // creation task. reconstruction_policy_.ListenAndMaybeReconstruct( @@ -1169,8 +1167,7 @@ void NodeManager::ProcessAnnounceWorkerPortMessage( void NodeManager::HandleDisconnectedActor(const ActorID &actor_id, bool was_local, bool intentional_disconnect) { - if (RayConfig::instance().gcs_service_enabled() && - RayConfig::instance().gcs_actor_service_enabled()) { + if (RayConfig::instance().gcs_actor_service_enabled()) { // If gcs actor management is enabled, the gcs will take over the status change of all // actors. return; @@ -2716,8 +2713,7 @@ void NodeManager::FinishAssignedActorTask(Worker &worker, const Task &task) { worker.MarkDetachedActor(); } - if (RayConfig::instance().gcs_service_enabled() && - RayConfig::instance().gcs_actor_service_enabled()) { + if (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`