From b7003839bdeec51108497c3914c28319cef2f591 Mon Sep 17 00:00:00 2001 From: kisuke95 <2522134184@qq.com> Date: Mon, 7 Sep 2020 18:36:43 -0500 Subject: [PATCH] [Core] Use core worker options to initialize (#10467) * fix * fix * . --- cpp/src/ray/util/process_helper.cc | 46 ++++++---------- cpp/src/ray/worker/default_worker.cc | 49 +++++++---------- src/ray/core_worker/core_worker.h | 26 ++++++++++ .../java/io_ray_runtime_RayNativeRuntime.cc | 52 ++++++++----------- src/ray/core_worker/test/core_worker_test.cc | 46 ++++++---------- src/ray/core_worker/test/mock_worker.cc | 48 ++++++----------- .../runtime/transfer/channel/ChannelId.java | 4 +- streaming/src/test/mock_actor.cc | 48 ++++++----------- streaming/src/test/queue_tests_base.h | 46 ++++++---------- 9 files changed, 149 insertions(+), 216 deletions(-) diff --git a/cpp/src/ray/util/process_helper.cc b/cpp/src/ray/util/process_helper.cc index 7fd2ab1e8..737700865 100644 --- a/cpp/src/ray/util/process_helper.cc +++ b/cpp/src/ray/util/process_helper.cc @@ -53,36 +53,22 @@ void ProcessHelper::RayStart(std::shared_ptr config) { gcs::GcsClientOptions gcs_options = gcs::GcsClientOptions(redis_ip, config->redis_port, config->redis_password); - CoreWorkerOptions options = { - config->worker_type, // worker_type - Language::CPP, // langauge - session_dir + "/sockets/plasma_store", // store_socket - session_dir + "/sockets/raylet", // raylet_socket - JobID::FromInt(1), // job_id - gcs_options, // gcs_options - true, // enable_logging - "", // log_dir - true, // install_failure_signal_handler - "127.0.0.1", // node_ip_address - config->node_manager_port, // node_manager_port - "127.0.0.1", // raylet_ip_address - "cpp_worker", // driver_name - "", // stdout_file - "", // stderr_file - nullptr, // task_execution_callback - nullptr, // check_signals - nullptr, // gc_collect - nullptr, // spill_objects - nullptr, // restore_spilled_objects - nullptr, // get_lang_stack - nullptr, // kill_main - true, // ref_counting_enabled - false, // is_local_mode - 1, // num_workers - nullptr, // terminate_asyncio_thread - "", // serialized_job_config - -1, // metrics_agent_port - }; + CoreWorkerOptions options; + options.worker_type = config->worker_type; + options.language = Language::CPP; + options.store_socket = session_dir + "/sockets/plasma_store"; + options.raylet_socket = session_dir + "/sockets/raylet"; + options.job_id = JobID::FromInt(1); + options.gcs_options = gcs_options; + options.enable_logging = true; + options.install_failure_signal_handler = true; + options.node_ip_address = "127.0.0.1"; + options.node_manager_port = config->node_manager_port; + options.raylet_ip_address = "127.0.0.1"; + options.driver_name = "cpp_worker"; + options.ref_counting_enabled = true; + options.num_workers = 1; + options.metrics_agent_port = -1; CoreWorkerProcess::Initialize(options); } diff --git a/cpp/src/ray/worker/default_worker.cc b/cpp/src/ray/worker/default_worker.cc index 81104509d..1c3157bd4 100644 --- a/cpp/src/ray/worker/default_worker.cc +++ b/cpp/src/ray/worker/default_worker.cc @@ -13,37 +13,24 @@ class DefaultWorker { DefaultWorker(const std::string &store_socket, const std::string &raylet_socket, int node_manager_port, const gcs::GcsClientOptions &gcs_options, const std::string &session_dir) { - CoreWorkerOptions options = { - WorkerType::WORKER, // worker_type - Language::CPP, // langauge - store_socket, // store_socket - raylet_socket, // raylet_socket - JobID::FromInt(1), // job_id - gcs_options, // gcs_options - true, // enable_logging - session_dir + "/logs", // log_dir - true, // install_failure_signal_handler - "127.0.0.1", // node_ip_address - node_manager_port, // node_manager_port - "127.0.0.1", // raylet_ip_address - "", // driver_name - "", // stdout_file - "", // stderr_file - std::bind(&DefaultWorker::ExecuteTask, this, _1, _2, _3, _4, _5, _6, _7, - _8), // task_execution_callback - nullptr, // check_signals - nullptr, // gc_collect - nullptr, // spill_objects - nullptr, // restore_spilled_objects - nullptr, // get_lang_stack - nullptr, // kill_main - true, // ref_counting_enabled - false, // is_local_mode - 1, // num_workers - nullptr, // terminate_asyncio_thread - "", // serialized_job_config - -1, // metrics_agent_port - }; + CoreWorkerOptions options; + options.worker_type = WorkerType::WORKER; + options.language = Language::CPP; + options.store_socket = store_socket; + options.raylet_socket = raylet_socket; + options.job_id = JobID::FromInt(1); + options.gcs_options = gcs_options; + options.enable_logging = true; + options.log_dir = session_dir + "/logs"; + options.install_failure_signal_handler = true; + options.node_ip_address = "127.0.0.1"; + options.node_manager_port = node_manager_port; + options.raylet_ip_address = "127.0.0.1"; + options.task_execution_callback = + std::bind(&DefaultWorker::ExecuteTask, this, _1, _2, _3, _4, _5, _6, _7, _8); + options.ref_counting_enabled = true; + options.num_workers = 1; + options.metrics_agent_port = -1; CoreWorkerProcess::Initialize(options); } diff --git a/src/ray/core_worker/core_worker.h b/src/ray/core_worker/core_worker.h index 5d4c68134..3fa81bedb 100644 --- a/src/ray/core_worker/core_worker.h +++ b/src/ray/core_worker/core_worker.h @@ -64,6 +64,32 @@ struct CoreWorkerOptions { const std::vector &return_ids, std::vector> *results)>; + CoreWorkerOptions() + : store_socket(""), + raylet_socket(""), + enable_logging(false), + log_dir(""), + install_failure_signal_handler(false), + node_ip_address(""), + node_manager_port(0), + raylet_ip_address(""), + driver_name(""), + stdout_file(""), + stderr_file(""), + task_execution_callback(nullptr), + check_signals(nullptr), + gc_collect(nullptr), + spill_objects(nullptr), + restore_spilled_objects(nullptr), + get_lang_stack(nullptr), + kill_main(nullptr), + ref_counting_enabled(false), + is_local_mode(false), + num_workers(0), + terminate_asyncio_thread(nullptr), + serialized_job_config(""), + metrics_agent_port(-1) {} + /// Type of this worker (i.e., DRIVER or WORKER). WorkerType worker_type; /// Application language of this worker (i.e., PYTHON or JAVA). diff --git a/src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc b/src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc index 78ca14b41..bf5050c50 100644 --- a/src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc +++ b/src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc @@ -213,37 +213,27 @@ JNIEXPORT void JNICALL Java_io_ray_runtime_RayNativeRuntime_nativeInitialize( std::string serialized_job_config = (jobConfig == nullptr ? "" : JavaByteArrayToNativeString(env, jobConfig)); - ray::CoreWorkerOptions options = { - static_cast(workerMode), // worker_type - ray::Language::JAVA, // langauge - JavaStringToNativeString(env, storeSocket), // store_socket - JavaStringToNativeString(env, rayletSocket), // raylet_socket - JavaByteArrayToId(env, jobId), // job_id - ToGcsClientOptions(env, gcsClientOptions), // gcs_options - true, // enable_logging - JavaStringToNativeString(env, logDir), // log_dir - // TODO (kfstorm): JVM would crash if install_failure_signal_handler was set to true - false, // install_failure_signal_handler - JavaStringToNativeString(env, nodeIpAddress), // node_ip_address - static_cast(nodeManagerPort), // node_manager_port - JavaStringToNativeString(env, nodeIpAddress), // raylet_ip_address - JavaStringToNativeString(env, driverName), // driver_name - "", // stdout_file - "", // stderr_file - task_execution_callback, // task_execution_callback - nullptr, // check_signals - gc_collect, // gc_collect - nullptr, // spill_objects - nullptr, // restore_spilled_objects - nullptr, // get_lang_stack - nullptr, // kill_main - true, // ref_counting_enabled - false, // is_local_mode - static_cast(numWorkersPerProcess), // num_workers - nullptr, // terminate_asyncio_thread - serialized_job_config, // serialized_job_config - -1, // metrics_agent_port - }; + ray::CoreWorkerOptions options; + options.worker_type = static_cast(workerMode); + options.language = ray::Language::JAVA; + options.store_socket = JavaStringToNativeString(env, storeSocket); + options.raylet_socket = JavaStringToNativeString(env, rayletSocket); + options.job_id = JavaByteArrayToId(env, jobId); + options.gcs_options = ToGcsClientOptions(env, gcsClientOptions); + options.enable_logging = true; + options.log_dir = JavaStringToNativeString(env, logDir); + // TODO (kfstorm): JVM would crash if install_failure_signal_handler was set to true + options.install_failure_signal_handler = false; + options.node_ip_address = JavaStringToNativeString(env, nodeIpAddress); + options.node_manager_port = static_cast(nodeManagerPort); + options.raylet_ip_address = JavaStringToNativeString(env, nodeIpAddress); + options.driver_name = JavaStringToNativeString(env, driverName); + options.task_execution_callback = task_execution_callback; + options.gc_collect = gc_collect; + options.ref_counting_enabled = true; + options.num_workers = static_cast(numWorkersPerProcess); + options.serialized_job_config = serialized_job_config; + options.metrics_agent_port = -1; ray::CoreWorkerProcess::Initialize(options); } diff --git a/src/ray/core_worker/test/core_worker_test.cc b/src/ray/core_worker/test/core_worker_test.cc index b03218270..c55bc30ba 100644 --- a/src/ray/core_worker/test/core_worker_test.cc +++ b/src/ray/core_worker/test/core_worker_test.cc @@ -140,36 +140,22 @@ class CoreWorkerTest : public ::testing::Test { void SetUp() { if (num_nodes_ > 0) { - CoreWorkerOptions options = { - WorkerType::DRIVER, // worker_type - Language::PYTHON, // langauge - raylet_store_socket_names_[0], // store_socket - raylet_socket_names_[0], // raylet_socket - NextJobId(), // job_id - gcs_options_, // gcs_options - true, // enable_logging - "", // log_dir - true, // install_failure_signal_handler - "127.0.0.1", // node_ip_address - node_manager_port, // node_manager_port - "127.0.0.1", // raylet_ip_address - "core_worker_test", // driver_name - "", // stdout_file - "", // stderr_file - nullptr, // task_execution_callback - nullptr, // check_signals - nullptr, // gc_collect - nullptr, // spill_objects - nullptr, // restore_spilled_objects - nullptr, // get_lang_stack - nullptr, // kill_main - true, // ref_counting_enabled - false, // is_local_mode - 1, // num_workers - nullptr, // terminate_asyncio_thread - "", // serialized_job_config - -1, // metrics_agent_port - }; + CoreWorkerOptions options; + options.worker_type = WorkerType::DRIVER; + options.language = Language::PYTHON; + options.store_socket = raylet_store_socket_names_[0]; + options.raylet_socket = raylet_socket_names_[0]; + options.job_id = NextJobId(); + options.gcs_options = gcs_options_; + options.enable_logging = true; + options.install_failure_signal_handler = true; + options.node_ip_address = "127.0.0.1"; + options.node_manager_port = node_manager_port; + options.raylet_ip_address = "127.0.0.1"; + options.driver_name = "core_worker_test"; + options.ref_counting_enabled = true; + options.num_workers = 1; + options.metrics_agent_port = -1; CoreWorkerProcess::Initialize(options); } } diff --git a/src/ray/core_worker/test/mock_worker.cc b/src/ray/core_worker/test/mock_worker.cc index 29b6a26f0..200833703 100644 --- a/src/ray/core_worker/test/mock_worker.cc +++ b/src/ray/core_worker/test/mock_worker.cc @@ -34,37 +34,23 @@ class MockWorker { public: MockWorker(const std::string &store_socket, const std::string &raylet_socket, int node_manager_port, const gcs::GcsClientOptions &gcs_options) { - CoreWorkerOptions options = { - WorkerType::WORKER, // worker_type - Language::PYTHON, // langauge - store_socket, // store_socket - raylet_socket, // raylet_socket - JobID::FromInt(1), // job_id - gcs_options, // gcs_options - true, // enable_logging - "", // log_dir - true, // install_failure_signal_handler - "127.0.0.1", // node_ip_address - node_manager_port, // node_manager_port - "127.0.0.1", // raylet_ip_address - "", // driver_name - "", // stdout_file - "", // stderr_file - std::bind(&MockWorker::ExecuteTask, this, _1, _2, _3, _4, _5, _6, _7, - _8), // task_execution_callback - nullptr, // check_signals - nullptr, // gc_collect - nullptr, // spill_objects - nullptr, // restore_spilled_objects - nullptr, // get_lang_stack - nullptr, // kill_main - true, // ref_counting_enabled - false, // is_local_mode - 1, // num_workers - nullptr, // terminate_asyncio_thread - "", // serialized_job_config - -1, // metrics_agent_port - }; + CoreWorkerOptions options; + options.worker_type = WorkerType::WORKER; + options.language = Language::PYTHON; + options.store_socket = store_socket; + options.raylet_socket = raylet_socket; + options.job_id = JobID::FromInt(1); + options.gcs_options = gcs_options; + options.enable_logging = true; + options.install_failure_signal_handler = true; + options.node_ip_address = "127.0.0.1"; + options.node_manager_port = node_manager_port; + options.raylet_ip_address = "127.0.0.1"; + options.task_execution_callback = + std::bind(&MockWorker::ExecuteTask, this, _1, _2, _3, _4, _5, _6, _7, _8); + options.ref_counting_enabled = true; + options.num_workers = 1; + options.metrics_agent_port = -1; CoreWorkerProcess::Initialize(options); } diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java index 07e98ae3f..dbd8fef73 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java @@ -101,9 +101,9 @@ public class ChannelId { | 8 bytes | 4bytes | 4bytes| 2bytes| 2bytes | */ Preconditions.checkArgument(fromTaskId < Short.MAX_VALUE, - "fromTaskId %d is larger than %d", fromTaskId, Short.MAX_VALUE); + "fromTaskId %s is larger than %s", fromTaskId, Short.MAX_VALUE); Preconditions.checkArgument(toTaskId < Short.MAX_VALUE, - "toTaskId %d is larger than %d", fromTaskId, Short.MAX_VALUE); + "toTaskId %s is larger than %s", fromTaskId, Short.MAX_VALUE); byte[] channelName = new byte[20]; for (int i = 11; i >= 8; i--) { diff --git a/streaming/src/test/mock_actor.cc b/streaming/src/test/mock_actor.cc index 4d3c652a8..4f15fafd3 100644 --- a/streaming/src/test/mock_actor.cc +++ b/streaming/src/test/mock_actor.cc @@ -478,37 +478,23 @@ class StreamingWorker { int node_manager_port, const gcs::GcsClientOptions &gcs_options) : test_suite_(nullptr), peer_actor_handle_(nullptr) { // You must keep it same with `src/ray/core_worker/core_worker.h:CoreWorkerOptions` - CoreWorkerOptions options = { - WorkerType::WORKER, // worker_type - Language::PYTHON, // langauge - store_socket, // store_socket - raylet_socket, // raylet_socket - JobID::FromInt(1), // job_id - gcs_options, // gcs_options - true, // enable_logging - "", // log_dir - true, // install_failure_signal_handler - "127.0.0.1", // node_ip_address - node_manager_port, // node_manager_port - "127.0.0.1", // raylet_ip_address - "", // driver_name - "", // stdout_file - "", // stderr_file - std::bind(&StreamingWorker::ExecuteTask, this, _1, _2, _3, _4, _5, _6, _7, - _8), // task_execution_callback - nullptr, // check_signals - nullptr, // gc_collect - nullptr, // spill_objects - nullptr, // restore_spilled_objects - nullptr, // get_lang_stack - nullptr, // kill_main - true, // ref_counting_enabled - false, // is_local_mode - 1, // num_workers - nullptr, // terminate_asyncio_thread - "", // serialized_job_config - -1, // metrics_agent_port - }; + CoreWorkerOptions options; + options.worker_type = WorkerType::WORKER; + options.language = Language::PYTHON; + options.store_socket = store_socket; + options.raylet_socket = raylet_socket; + options.job_id = JobID::FromInt(1); + options.gcs_options = gcs_options; + options.enable_logging = true; + options.install_failure_signal_handler = true; + options.node_ip_address = "127.0.0.1"; + options.node_manager_port = node_manager_port; + options.raylet_ip_address = "127.0.0.1"; + options.task_execution_callback = + std::bind(&StreamingWorker::ExecuteTask, this, _1, _2, _3, _4, _5, _6, _7, _8); + options.ref_counting_enabled = true; + options.num_workers = 1; + options.metrics_agent_port = -1; CoreWorkerProcess::Initialize(options); STREAMING_LOG(INFO) << "StreamingWorker constructor"; } diff --git a/streaming/src/test/queue_tests_base.h b/streaming/src/test/queue_tests_base.h index fc9e81d18..28be67981 100644 --- a/streaming/src/test/queue_tests_base.h +++ b/streaming/src/test/queue_tests_base.h @@ -232,36 +232,22 @@ class StreamingQueueTestBase : public ::testing::TestWithParam { STREAMING_LOG(INFO) << "Sub process: writer."; // You must keep it same with `src/ray/core_worker/core_worker.h:CoreWorkerOptions` - CoreWorkerOptions options = { - WorkerType::DRIVER, // worker_type - Language::PYTHON, // langauge - raylet_store_socket_names_[0], // store_socket - raylet_socket_names_[0], // raylet_socket - NextJobId(), // job_id - gcs_options_, // gcs_options - true, // enable_logging - "", // log_dir - true, // install_failure_signal_handler - "127.0.0.1", // node_ip_address - node_manager_port_, // node_manager_port - "127.0.0.1", // raylet_ip_address - "queue_tests", // driver_name - "", // stdout_file - "", // stderr_file - nullptr, // task_execution_callback - nullptr, // check_signals - nullptr, // gc_collect - nullptr, // spill_objects - nullptr, // restore_spilled_objects - nullptr, // get_lang_stack - nullptr, // kill_main - true, // ref_counting_enabled - false, // is_local_mode - 1, // num_workers - nullptr, // terminate_asyncio_thread - "", // serialized_job_config - -1, // metrics_agent_port - }; + CoreWorkerOptions options; + options.worker_type = WorkerType::DRIVER; + options.language = Language::PYTHON; + options.store_socket = raylet_store_socket_names_[0]; + options.raylet_socket = raylet_socket_names_[0]; + options.job_id = NextJobId(); + options.gcs_options = gcs_options_; + options.enable_logging = true; + options.install_failure_signal_handler = true; + options.node_ip_address = "127.0.0.1"; + options.node_manager_port = node_manager_port_; + options.raylet_ip_address = "127.0.0.1"; + options.driver_name = "queue_tests"; + options.ref_counting_enabled = true; + options.num_workers = 1; + options.metrics_agent_port = -1; InitShutdownRAII core_worker_raii(CoreWorkerProcess::Initialize, CoreWorkerProcess::Shutdown, options);