wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-08-04 13:14:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

[Core] Use core worker options to initialize (#10467)

Browse Source 
* fix

* fix

* .
This commit is contained in:
kisuke95
2020-09-07 16:36:43 -07:00
committed by GitHub
parent 8db1f16f25
commit b7003839bd
9 changed files with 149 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/src/ray/util/process_helper.cc
+16 -30
View File
@@ -53,36 +53,22 @@ void ProcessHelper::RayStart(std::shared_ptr<RayConfig> 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);
}
cpp/src/ray/worker/default_worker.cc
+18 -31
View File
@@ -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);
}
src/ray/core_worker/core_worker.h
+26
View File
@@ -64,6 +64,32 @@ struct CoreWorkerOptions {
const std::vector<ObjectID> &return_ids,
std::vector<std::shared_ptr<RayObject>> *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).
src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc
+21 -31
View File
@@ -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<ray::WorkerType>(workerMode), // worker_type
ray::Language::JAVA, // langauge
JavaStringToNativeString(env, storeSocket), // store_socket
JavaStringToNativeString(env, rayletSocket), // raylet_socket
JavaByteArrayToId<ray::JobID>(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<int>(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<int>(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<ray::WorkerType>(workerMode);
options.language = ray::Language::JAVA;
options.store_socket = JavaStringToNativeString(env, storeSocket);
options.raylet_socket = JavaStringToNativeString(env, rayletSocket);
options.job_id = JavaByteArrayToId<ray::JobID>(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<int>(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<int>(numWorkersPerProcess);
options.serialized_job_config = serialized_job_config;
options.metrics_agent_port = -1;
ray::CoreWorkerProcess::Initialize(options);
}
src/ray/core_worker/test/core_worker_test.cc
+16 -30
View File
@@ -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);
}
}
src/ray/core_worker/test/mock_worker.cc
+17 -31
View File
@@ -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);
}
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/channel/ChannelId.java
+2 -2
View File
@@ -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--) {
streaming/src/test/mock_actor.cc
+17 -31
View File
@@ -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";
}
streaming/src/test/queue_tests_base.h
+16 -30
View File
@@ -232,36 +232,22 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
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);