wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-12 10:59:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

[Release] Revert Enable GCS Server by Default (#7840)

Browse Source
This commit is contained in:
Simon Mo
2020-04-01 10:05:07 -07:00
committed by GitHub
parent c23e56ce9a
commit 1ab98155eb
13 changed files with 26 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+4 -4
View File
@@ -49,11 +49,11 @@ matrix:
- os: linux
env:
- TESTSUITE=gcs_service_disabled
- TESTSUITE=gcs_service
- JDK='Oracle JDK 8'
- RAY_GCS_SERVICE_ENABLED=true
- PYTHON=3.6 PYTHONWARNINGS=ignore
- RAY_INSTALL_JAVA=1
- RAY_GCS_SERVICE_ENABLED=false
install:
- eval `python $TRAVIS_BUILD_DIR/ci/travis/determine_tests_to_run.py`
- ./ci/travis/install-bazel.sh
@@ -69,11 +69,11 @@ matrix:
- os: linux
env:
- TESTSUITE=gcs_service_disabled_python_testcase
- TESTSUITE=gcs_service_python_testcase
- JDK='Oracle JDK 8'
- RAY_GCS_SERVICE_ENABLED=true
- PYTHON=3.6 PYTHONWARNINGS=ignore
- RAY_INSTALL_JAVA=1
- RAY_GCS_SERVICE_ENABLED=false
install:
- eval `python $TRAVIS_BUILD_DIR/ci/travis/determine_tests_to_run.py`
- ./ci/travis/install-bazel.sh
java/runtime/src/main/java/org/ray/runtime/config/RayConfig.java
-5
View File
@@ -72,8 +72,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() {
@@ -225,9 +223,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();
LOGGER.debug("Created config: {}", this);
java/runtime/src/main/java/org/ray/runtime/runner/RunManager.java
+1 -1
View File
@@ -226,7 +226,7 @@ public class RunManager {
}
// start gcs server
if (rayConfig.gcsServiceEnabled) {
if (System.getenv("RAY_GCS_SERVICE_ENABLED") != null) {
String redisPasswordOption = "";
if (!Strings.isNullOrEmpty(rayConfig.headRedisPassword)) {
redisPasswordOption = rayConfig.headRedisPassword;
python/ray/node.py
+1 -1
View File
@@ -626,7 +626,7 @@ class Node:
# If this is the head node, start the relevant head node processes.
self.start_redis()
if ray_constants.GCS_SERVICE_ENABLED:
if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
self.start_gcs_server()
else:
self.start_raylet_monitor()
python/ray/ray_constants.py
+1 -7
View File
@@ -13,12 +13,6 @@ def env_integer(key, default):
return default
def env_bool(key, default):
if key in os.environ:
return True if os.environ[key].lower() == "true" else False
return default
ID_SIZE = 20
# The default maximum number of bytes to allocate to the object store unless
@@ -203,4 +197,4 @@ 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)
RAY_GCS_SERVICE_ENABLED = "RAY_GCS_SERVICE_ENABLED"
python/ray/tests/test_component_failures_3.py
+2 -2
View File
@@ -82,7 +82,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:
if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
ray.worker._global_node.kill_gcs_server()
else:
ray.worker._global_node.kill_raylet_monitor()
@@ -96,7 +96,7 @@ 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:
if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
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] +
python/ray/tests/test_multinode_failures_2.py
+2 -2
View File
@@ -132,7 +132,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:
if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
ray.worker._global_node.kill_gcs_server()
else:
ray.worker._global_node.kill_raylet_monitor()
@@ -145,7 +145,7 @@ 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:
if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
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] +
python/ray/tests/test_tempfile.py
+1 -1
View File
@@ -145,7 +145,7 @@ def test_raylet_tempfiles(shutdown_only):
"raylet.err"
}
if ray_constants.GCS_SERVICE_ENABLED:
if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
log_files_expected.update({"gcs_server.out", "gcs_server.err"})
else:
log_files_expected.update({"raylet_monitor.out", "raylet_monitor.err"})
src/ray/common/ray_config_def.h
-9
View File
@@ -273,12 +273,3 @@ RAY_CONFIG(uint32_t, object_store_full_initial_delay_ms, 1000)
/// Duration to wait between retries for failed tasks.
RAY_CONFIG(uint32_t, task_retry_delay_ms, 5000)
/// 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"))
src/ray/core_worker/core_worker.cc
+1 -1
View File
@@ -115,7 +115,7 @@ CoreWorker::CoreWorker(const WorkerType worker_type, const Language language,
RayLog::InstallFailureSignalHandler();
}
// Initialize gcs client.
if (RayConfig::instance().gcs_service_enabled()) {
if (getenv("RAY_GCS_SERVICE_ENABLED") != nullptr) {
gcs_client_ = std::make_shared<ray::gcs::ServiceBasedGcsClient>(gcs_options);
} else {
gcs_client_ = std::make_shared<ray::gcs::RedisGcsClient>(gcs_options);
src/ray/core_worker/test/core_worker_test.cc
+1 -1
View File
@@ -109,7 +109,7 @@ class CoreWorkerTest : public ::testing::Test {
}
// start gcs server
if (RayConfig::instance().gcs_service_enabled()) {
if (getenv("RAY_GCS_SERVICE_ENABLED") != nullptr) {
gcs_server_pid_ = StartGcsServer("127.0.0.1");
} else {
// core worker test relies on node resources. It's important that one raylet can
src/ray/gcs/gcs_server/gcs_server.cc
+10 -19
View File
@@ -145,27 +145,18 @@ std::unique_ptr<rpc::ObjectInfoHandler> GcsServer::InitObjectInfoHandler() {
void GcsServer::StoreGcsServerAddressInRedis() {
boost::asio::ip::detail::endpoint primary_endpoint;
boost::asio::ip::tcp::resolver resolver(main_service_);
boost::asio::ip::tcp::resolver::query query(
boost::asio::ip::host_name(), "",
boost::asio::ip::resolver_query_base::flags::v4_mapped);
boost::system::error_code error_code;
boost::asio::ip::tcp::resolver::iterator iter = resolver.resolve(query, error_code);
boost::asio::ip::tcp::resolver::query query(boost::asio::ip::host_name(), "");
boost::asio::ip::tcp::resolver::iterator iter = resolver.resolve(query);
boost::asio::ip::tcp::resolver::iterator end; // End marker.
if (!error_code) {
while (iter != end) {
boost::asio::ip::tcp::endpoint ep = *iter;
if (ep.address().is_v4() && !ep.address().is_loopback() &&
!ep.address().is_multicast()) {
primary_endpoint.address(ep.address());
primary_endpoint.port(ep.port());
break;
}
iter++;
while (iter != end) {
boost::asio::ip::tcp::endpoint ep = *iter;
if (ep.address().is_v4() && !ep.address().is_loopback() &&
!ep.address().is_multicast()) {
primary_endpoint.address(ep.address());
primary_endpoint.port(ep.port());
break;
}
} else {
RAY_LOG(WARNING) << "Failed to resolve ip address, error = "
<< strerror(error_code.value());
iter = end;
iter++;
}
std::string address;
src/ray/raylet/main.cc
+2 -1
View File
@@ -177,7 +177,8 @@ int main(int argc, char *argv[]) {
ray::gcs::GcsClientOptions client_options(redis_address, redis_port, redis_password);
std::shared_ptr<ray::gcs::GcsClient> gcs_client;
if (RayConfig::instance().gcs_service_enabled()) {
// RAY_GCS_SERVICE_ENABLED only set in ci job, so we just check if it is null.
if (getenv("RAY_GCS_SERVICE_ENABLED") != nullptr) {
gcs_client = std::make_shared<ray::gcs::ServiceBasedGcsClient>(client_options);
} else {
gcs_client = std::make_shared<ray::gcs::RedisGcsClient>(client_options);