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

[Multi-tenancy] Delete flag enable_multi_tenancy and remove old code path (#10573)

Browse Source
This commit is contained in:
Kai Yang
2020-12-10 19:01:40 +08:00
committed by GitHub
parent d681991773
commit e3b5deb741
47 changed files with 279 additions and 530 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/common/ray_config_def.h
-14
View File
@@ -196,15 +196,6 @@ RAY_CONFIG(uint64_t, object_manager_default_chunk_size, 5 * 1024 * 1024)
/// excessive memory usage during object broadcast to many receivers.
RAY_CONFIG(uint64_t, object_manager_max_bytes_in_flight, 2L * 1024 * 1024 * 1024)
/// Number of workers per Python worker process
RAY_CONFIG(int, num_workers_per_process_python, 1)
/// Number of workers per Java worker process
RAY_CONFIG(int, num_workers_per_process_java, 1)
/// Number of workers per CPP worker process
RAY_CONFIG(int, num_workers_per_process_cpp, 1)
/// Maximum number of ids in one batch to send to GCS to delete keys.
RAY_CONFIG(uint32_t, maximum_gcs_deletion_batch_size, 1000)
@@ -301,11 +292,6 @@ RAY_CONFIG(bool, report_worker_backlog, true)
/// The timeout for synchronous GCS requests in seconds.
RAY_CONFIG(int64_t, gcs_server_request_timeout_seconds, 5)
/// Whether to enable multi tenancy features.
RAY_CONFIG(bool, enable_multi_tenancy,
getenv("RAY_ENABLE_MULTI_TENANCY") == nullptr ||
getenv("RAY_ENABLE_MULTI_TENANCY") == std::string("1"))
/// Whether to enable worker prestarting: https://github.com/ray-project/ray/issues/12052
RAY_CONFIG(bool, enable_worker_prestart,
getenv("RAY_ENABLE_WORKER_PRESTART") == nullptr ||
src/ray/common/test_util.cc
+1 -2
View File
@@ -148,8 +148,7 @@ std::string TestSetupUtil::StartRaylet(const std::string &store_socket_name,
"--node_manager_port=" + std::to_string(port),
"--node_ip_address=" + node_ip_address, "--redis_address=" + redis_address,
"--redis_port=6379", "--min-worker-port=0", "--max-worker-port=0",
"--num_initial_workers=1", "--maximum_startup_concurrency=10",
"--static_resource_list=" + resource,
"--maximum_startup_concurrency=10", "--static_resource_list=" + resource,
"--python_worker_command=" +
CreateCommandLine({TEST_MOCK_WORKER_EXEC_PATH, store_socket_name,
raylet_socket_name, std::to_string(port)}),
src/ray/core_worker/context.cc
+3 -25
View File
@@ -111,9 +111,7 @@ WorkerContext::WorkerContext(WorkerType worker_type, const WorkerID &worker_id,
const JobID &job_id)
: worker_type_(worker_type),
worker_id_(worker_id),
current_job_id_(RayConfig::instance().enable_multi_tenancy()
? job_id
: (worker_type_ == WorkerType::DRIVER ? job_id : JobID::Nil())),
current_job_id_(job_id),
current_actor_id_(ActorID::Nil()),
current_actor_placement_group_id_(PlacementGroupID::Nil()),
placement_group_capture_child_tasks_(true),
@@ -163,29 +161,17 @@ const std::unordered_map<std::string, std::string>
return override_environment_variables_;
}
void WorkerContext::SetCurrentJobId(const JobID &job_id) {
RAY_CHECK(!RayConfig::instance().enable_multi_tenancy());
current_job_id_ = job_id;
}
void WorkerContext::SetCurrentTaskId(const TaskID &task_id) {
GetThreadContext().SetCurrentTaskId(task_id);
}
void WorkerContext::SetCurrentTask(const TaskSpecification &task_spec) {
GetThreadContext().SetCurrentTask(task_spec);
RAY_CHECK(current_job_id_ == task_spec.JobId());
if (task_spec.IsNormalTask()) {
if (!RayConfig::instance().enable_multi_tenancy()) {
RAY_CHECK(current_job_id_.IsNil());
SetCurrentJobId(task_spec.JobId());
}
current_task_is_direct_call_ = true;
override_environment_variables_ = task_spec.OverrideEnvironmentVariables();
} else if (task_spec.IsActorCreationTask()) {
if (!RayConfig::instance().enable_multi_tenancy()) {
RAY_CHECK(current_job_id_.IsNil());
SetCurrentJobId(task_spec.JobId());
}
RAY_CHECK(current_actor_id_.IsNil());
current_actor_id_ = task_spec.ActorCreationId();
current_actor_is_direct_call_ = true;
@@ -196,21 +182,13 @@ void WorkerContext::SetCurrentTask(const TaskSpecification &task_spec) {
placement_group_capture_child_tasks_ = task_spec.PlacementGroupCaptureChildTasks();
override_environment_variables_ = task_spec.OverrideEnvironmentVariables();
} else if (task_spec.IsActorTask()) {
if (!RayConfig::instance().enable_multi_tenancy()) {
RAY_CHECK(current_job_id_ == task_spec.JobId());
}
RAY_CHECK(current_actor_id_ == task_spec.ActorId());
} else {
RAY_CHECK(false);
}
}
void WorkerContext::ResetCurrentTask(const TaskSpecification &task_spec) {
GetThreadContext().ResetCurrentTask();
if (!RayConfig::instance().enable_multi_tenancy() && task_spec.IsNormalTask()) {
SetCurrentJobId(JobID::Nil());
}
}
void WorkerContext::ResetCurrentTask() { GetThreadContext().ResetCurrentTask(); }
std::shared_ptr<const TaskSpecification> WorkerContext::GetCurrentTask() const {
return GetThreadContext().GetCurrentTask();
src/ray/core_worker/context.h
+1 -5
View File
@@ -42,16 +42,12 @@ class WorkerContext {
const std::unordered_map<std::string, std::string>
&GetCurrentOverrideEnvironmentVariables() const;
// TODO(kfstorm): Remove this once `enable_multi_tenancy` is deleted.
// TODO(edoakes): remove this once Python core worker uses the task interfaces.
void SetCurrentJobId(const JobID &job_id);
// TODO(edoakes): remove this once Python core worker uses the task interfaces.
void SetCurrentTaskId(const TaskID &task_id);
void SetCurrentTask(const TaskSpecification &task_spec);
void ResetCurrentTask(const TaskSpecification &task_spec);
void ResetCurrentTask();
std::shared_ptr<const TaskSpecification> GetCurrentTask() const;
src/ray/core_worker/core_worker.cc
+2 -10
View File
@@ -69,15 +69,7 @@ ray::JobID GetProcessJobID(const ray::CoreWorkerOptions &options) {
if (options.worker_type == ray::WorkerType::WORKER) {
// For workers, the job ID is assigned by Raylet via an environment variable.
const char *job_id_env = std::getenv(kEnvVarKeyJobId);
// TODO(kfstorm): Use `RAY_CHECK` instead once the `enable_multi_tenancy` flag is
// removed.
// RAY_CHECK(job_id_env);
if (!job_id_env) {
// Multi-tenancy is disabled.
// NOTE(kfstorm): We can't read `RayConfig::instance().enable_multi_tenancy()` here
// because `RayConfig` is not initialized yet.
return ray::JobID::Nil();
}
RAY_CHECK(job_id_env);
return ray::JobID::FromHex(job_id_env);
}
return options.job_id;
@@ -1859,7 +1851,7 @@ Status CoreWorker::ExecuteTask(const TaskSpecification &task_spec,
if (!options_.is_local_mode) {
SetCurrentTaskId(TaskID::Nil());
worker_context_.ResetCurrentTask(task_spec);
worker_context_.ResetCurrentTask();
}
{
absl::MutexLock lock(&mutex_);
src/ray/core_worker/core_worker.h
+1 -1
View File
@@ -198,7 +198,7 @@ struct CoreWorkerOptions {
/// thread with a worker. You can obtain the worker ID via
/// `CoreWorkerProcess::GetCoreWorker()->GetWorkerID()`. Currently a Java worker process
/// starts multiple workers by default, but can be configured to start only 1 worker by
/// overwriting the internal config `num_workers_per_process_java`.
/// speicifying `num_java_workers_per_process` in the job config.
///
/// If only 1 worker is started (either because the worker type is driver, or the
/// `num_workers` in `CoreWorkerOptions` is set to 1), all threads will be automatically
src/ray/core_worker/test/core_worker_test.cc
+2 -1
View File
@@ -620,8 +620,9 @@ TEST_F(ZeroNodeTest, TestWorkerContext) {
TaskSpecification task_spec;
size_t num_returns = 3;
task_spec.GetMutableMessage().set_job_id(job_id.Binary());
task_spec.GetMutableMessage().set_num_returns(num_returns);
context.ResetCurrentTask(task_spec);
context.ResetCurrentTask();
context.SetCurrentTask(task_spec);
ASSERT_EQ(context.GetCurrentTaskID(), task_spec.TaskId());
src/ray/raylet/main.cc
-3
View File
@@ -38,7 +38,6 @@ DEFINE_int32(max_worker_port, 0,
"The highest port that workers' gRPC servers will bind on.");
DEFINE_string(worker_port_list, "",
"An explicit list of ports that workers' gRPC servers will bind on.");
DEFINE_int32(num_initial_workers, 0, "Number of initial workers.");
DEFINE_int32(num_initial_python_workers_for_first_job, 0,
"Number of initial Python workers for the first job.");
DEFINE_int32(maximum_startup_concurrency, 1, "Maximum startup concurrency");
@@ -78,7 +77,6 @@ int main(int argc, char *argv[]) {
const int min_worker_port = static_cast<int>(FLAGS_min_worker_port);
const int max_worker_port = static_cast<int>(FLAGS_max_worker_port);
const std::string worker_port_list = FLAGS_worker_port_list;
const int num_initial_workers = static_cast<int>(FLAGS_num_initial_workers);
const int num_initial_python_workers_for_first_job =
static_cast<int>(FLAGS_num_initial_python_workers_for_first_job);
const int maximum_startup_concurrency =
@@ -183,7 +181,6 @@ int main(int argc, char *argv[]) {
<< node_manager_config.resource_config.ToString();
node_manager_config.node_manager_address = node_ip_address;
node_manager_config.node_manager_port = node_manager_port;
node_manager_config.num_initial_workers = num_initial_workers;
node_manager_config.num_workers_soft_limit = num_cpus;
node_manager_config.num_initial_python_workers_for_first_job =
num_initial_python_workers_for_first_job;
src/ray/raylet/node_manager.cc
+11 -24
View File
@@ -133,7 +133,7 @@ NodeManager::NodeManager(boost::asio::io_service &io_service, const NodeID &self
initial_config_(config),
local_available_resources_(config.resource_config),
worker_pool_(
io_service, config.num_initial_workers, config.num_workers_soft_limit,
io_service, config.num_workers_soft_limit,
config.num_initial_python_workers_for_first_job,
config.maximum_startup_concurrency, config.min_worker_port,
config.max_worker_port, config.worker_ports, gcs_client_,
@@ -345,15 +345,13 @@ void NodeManager::HandleJobStarted(const JobID &job_id, const JobTableData &job_
RAY_CHECK(!job_data.is_dead());
worker_pool_.HandleJobStarted(job_id, job_data.config());
if (RayConfig::instance().enable_multi_tenancy()) {
// Tasks of this job may already arrived but failed to pop a worker because the job
// config is not local yet. So we trigger dispatching again here to try to
// reschedule these tasks.
if (new_scheduler_enabled_) {
ScheduleAndDispatch();
} else {
DispatchTasks(local_queues_.GetReadyTasksByClass());
}
// Tasks of this job may already arrived but failed to pop a worker because the job
// config is not local yet. So we trigger dispatching again here to try to
// reschedule these tasks.
if (new_scheduler_enabled_) {
ScheduleAndDispatch();
} else {
DispatchTasks(local_queues_.GetReadyTasksByClass());
}
}
@@ -1193,8 +1191,7 @@ void NodeManager::ProcessRegisterClientRequestMessage(
std::string worker_ip_address = string_from_flatbuf(*message->ip_address());
// TODO(suquark): Use `WorkerType` in `common.proto` without type converting.
rpc::WorkerType worker_type = static_cast<rpc::WorkerType>(message->worker_type());
if ((RayConfig::instance().enable_multi_tenancy() &&
(worker_type != rpc::WorkerType::SPILL_WORKER &&
if (((worker_type != rpc::WorkerType::SPILL_WORKER &&
worker_type != rpc::WorkerType::RESTORE_WORKER)) ||
worker_type == rpc::WorkerType::DRIVER) {
RAY_CHECK(!job_id.IsNil());
@@ -1627,9 +1624,7 @@ void NodeManager::HandleRequestWorkerLease(const rpc::RequestWorkerLeaseRequest
RAY_CHECK_OK(gcs_client_->Tasks().AsyncAdd(data, nullptr));
}
// Prestart optimization is only needed when multi-tenancy is on.
if (RayConfig::instance().enable_multi_tenancy() &&
RayConfig::instance().enable_worker_prestart()) {
if (RayConfig::instance().enable_worker_prestart()) {
auto task_spec = task.GetTaskSpecification();
worker_pool_.PrestartWorkers(task_spec, request.backlog_size());
}
@@ -2522,12 +2517,6 @@ bool NodeManager::FinishAssignedTask(const std::shared_ptr<WorkerInterface> &wor
task_dependency_manager_.UnsubscribeGetDependencies(spec.TaskId());
task_dependency_manager_.TaskCanceled(task_id);
if (!RayConfig::instance().enable_multi_tenancy()) {
// Unset the worker's assigned job Id if this is not an actor.
if (!spec.IsActorCreationTask()) {
worker.AssignJobId(JobID::Nil());
}
}
if (!spec.IsActorCreationTask()) {
// Unset the worker's assigned task. We keep the assigned task ID for
// direct actor creation calls because this ID is used later if the actor
@@ -2776,9 +2765,7 @@ void NodeManager::FinishAssignTask(const std::shared_ptr<WorkerInterface> &worke
// We successfully assigned the task to the worker.
worker->AssignTaskId(spec.TaskId());
worker->SetOwnerAddress(spec.CallerAddress());
if (!RayConfig::instance().enable_multi_tenancy()) {
worker->AssignJobId(spec.JobId());
}
RAY_CHECK(worker->GetAssignedJobId() == spec.JobId());
// TODO(swang): For actors with multiple actor handles, to
// guarantee that tasks are replayed in the same order after a
// failure, we must update the task's execution dependency to be
src/ray/raylet/node_manager.h
-2
View File
@@ -70,8 +70,6 @@ struct NodeManagerConfig {
/// An explicit list of open ports that workers started will bind
/// on. This takes precedence over min_worker_port and max_worker_port.
std::vector<int> worker_ports;
/// The initial number of workers to create.
int num_initial_workers;
/// The soft limit of the number of workers.
int num_workers_soft_limit;
/// Number of initial Python workers for the first job.
src/ray/raylet/object_manager_integration_test.cc
-1
View File
@@ -40,7 +40,6 @@ class TestObjectManagerBase : public ::testing::Test {
static_resource_conf = {{"CPU", 1}, {"GPU", 1}};
node_manager_config.resource_config =
ray::raylet::ResourceSet(std::move(static_resource_conf));
node_manager_config.num_initial_workers = 0;
// Use a default worker that can execute empty tasks with dependencies.
std::vector<std::string> py_worker_command;
py_worker_command.push_back("python");
src/ray/raylet/scheduling/cluster_task_manager.cc
-3
View File
@@ -186,9 +186,6 @@ bool ClusterTaskManager::AttemptDispatchWork(const Work &work,
worker->SetAllocatedInstances(allocated_instances);
}
worker->AssignTaskId(spec.TaskId());
if (!RayConfig::instance().enable_multi_tenancy()) {
worker->AssignJobId(spec.JobId());
}
worker->SetAssignedTask(task);
*worker_leased = true;
dispatched = true;
src/ray/raylet/test/util.h
-3
View File
@@ -33,9 +33,6 @@ class MockWorker : public WorkerInterface {
void AssignTaskId(const TaskID &task_id) {}
// TODO(kfstorm): Remove this once `enable_multi_tenancy` is deleted.
void AssignJobId(const JobID &job_id) {}
void SetAssignedTask(const Task &assigned_task) {}
const std::string IpAddress() const { return address_.ip_address(); }
src/ray/raylet/worker.cc
-5
View File
@@ -111,11 +111,6 @@ const std::unordered_set<TaskID> &Worker::GetBlockedTaskIds() const {
return blocked_task_ids_;
}
void Worker::AssignJobId(const JobID &job_id) {
RAY_CHECK(!RayConfig::instance().enable_multi_tenancy());
assigned_job_id_ = job_id;
}
const JobID &Worker::GetAssignedJobId() const { return assigned_job_id_; }
void Worker::AssignActorId(const ActorID &actor_id) {
src/ray/raylet/worker.h
-4
View File
@@ -60,8 +60,6 @@ class WorkerInterface {
virtual bool AddBlockedTaskId(const TaskID &task_id) = 0;
virtual bool RemoveBlockedTaskId(const TaskID &task_id) = 0;
virtual const std::unordered_set<TaskID> &GetBlockedTaskIds() const = 0;
// TODO(kfstorm): Remove this once `enable_multi_tenancy` is deleted.
virtual void AssignJobId(const JobID &job_id) = 0;
virtual const JobID &GetAssignedJobId() const = 0;
virtual void AssignActorId(const ActorID &actor_id) = 0;
virtual const ActorID &GetActorId() const = 0;
@@ -151,8 +149,6 @@ class Worker : public WorkerInterface {
bool AddBlockedTaskId(const TaskID &task_id);
bool RemoveBlockedTaskId(const TaskID &task_id);
const std::unordered_set<TaskID> &GetBlockedTaskIds() const;
// TODO(kfstorm): Remove this once `enable_multi_tenancy` is deleted.
void AssignJobId(const JobID &job_id);
const JobID &GetAssignedJobId() const;
void AssignActorId(const ActorID &actor_id);
const ActorID &GetActorId() const;
src/ray/raylet/worker_pool.cc
+54 -135
View File
@@ -54,8 +54,7 @@ namespace ray {
namespace raylet {
WorkerPool::WorkerPool(boost::asio::io_service &io_service, int num_workers,
int num_workers_soft_limit,
WorkerPool::WorkerPool(boost::asio::io_service &io_service, int num_workers_soft_limit,
int num_initial_python_workers_for_first_job,
int maximum_startup_concurrency, int min_worker_port,
int max_worker_port, const std::vector<int> &worker_ports,
@@ -83,33 +82,7 @@ WorkerPool::WorkerPool(boost::asio::io_service &io_service, int num_workers,
for (const auto &entry : worker_commands) {
// Initialize the pool state for this language.
auto &state = states_by_lang_[entry.first];
if (!RayConfig::instance().enable_multi_tenancy()) {
switch (entry.first) {
case Language::PYTHON:
state.num_workers_per_process =
RayConfig::instance().num_workers_per_process_python();
break;
case Language::JAVA:
state.num_workers_per_process =
RayConfig::instance().num_workers_per_process_java();
break;
case Language::CPP:
state.num_workers_per_process =
RayConfig::instance().num_workers_per_process_cpp();
break;
default:
RAY_LOG(FATAL) << "The number of workers per process for "
<< Language_Name(entry.first) << " worker is not set.";
}
RAY_CHECK(state.num_workers_per_process > 0)
<< "Number of workers per process of language " << Language_Name(entry.first)
<< " must be positive.";
state.multiple_for_warning =
std::max(state.num_workers_per_process,
std::max(num_workers, maximum_startup_concurrency));
} else {
state.multiple_for_warning = maximum_startup_concurrency;
}
state.multiple_for_warning = maximum_startup_concurrency;
// Set worker command for this language.
state.worker_command = entry.second;
RAY_CHECK(!state.worker_command.empty()) << "Worker command must not be empty.";
@@ -131,27 +104,7 @@ WorkerPool::WorkerPool(boost::asio::io_service &io_service, int num_workers,
free_ports_->push(port);
}
}
if (!RayConfig::instance().enable_multi_tenancy()) {
Start(num_workers);
} else {
ScheduleIdleWorkerKilling();
}
}
void WorkerPool::Start(int num_workers) {
RAY_CHECK(!RayConfig::instance().enable_multi_tenancy());
for (auto &entry : states_by_lang_) {
if (entry.first == Language::JAVA) {
// Disable initial workers for Java.
continue;
}
auto &state = entry.second;
int num_worker_processes = static_cast<int>(
std::ceil(static_cast<double>(num_workers) / state.num_workers_per_process));
for (int i = 0; i < num_worker_processes; i++) {
StartWorkerProcess(entry.first, rpc::WorkerType::WORKER, JobID::Nil());
}
}
ScheduleIdleWorkerKilling();
}
WorkerPool::~WorkerPool() {
@@ -178,7 +131,7 @@ Process WorkerPool::StartWorkerProcess(
std::vector<std::string> dynamic_options,
std::unordered_map<std::string, std::string> override_environment_variables) {
rpc::JobConfig *job_config = nullptr;
if (RayConfig::instance().enable_multi_tenancy() && !IsIOWorkerType(worker_type)) {
if (!IsIOWorkerType(worker_type)) {
RAY_CHECK(!job_id.IsNil());
auto it = all_jobs_.find(job_id);
if (it == all_jobs_.end()) {
@@ -212,15 +165,12 @@ Process WorkerPool::StartWorkerProcess(
int workers_to_start = 1;
if (dynamic_options.empty()) {
if (!RayConfig::instance().enable_multi_tenancy()) {
workers_to_start = state.num_workers_per_process;
} else if (language == Language::JAVA) {
if (language == Language::JAVA) {
workers_to_start = job_config->num_java_workers_per_process();
}
}
// For non-multi-tenancy mode, job code search path is embedded in worker_command.
if (RayConfig::instance().enable_multi_tenancy() && job_config) {
if (job_config) {
// Note that we push the item to the front of the vector to make
// sure this is the freshest option than others.
if (!job_config->jvm_options().empty()) {
@@ -272,9 +222,6 @@ Process WorkerPool::StartWorkerProcess(
switch (language) {
case Language::JAVA:
for (auto &entry : raylet_config_) {
if (entry.first == "num_workers_per_process_java") {
continue;
}
std::string arg;
arg.append("-Dray.raylet.config.");
arg.append(entry.first);
@@ -282,17 +229,8 @@ Process WorkerPool::StartWorkerProcess(
arg.append(entry.second);
worker_command_args.push_back(arg);
}
if (!RayConfig::instance().enable_multi_tenancy()) {
// The value of `num_workers_per_process_java` may change depends on whether
// dynamic options is empty, so we can't use the value in `RayConfig`. We always
// overwrite the value here.
worker_command_args.push_back(
"-Dray.raylet.config.num_workers_per_process_java=" +
std::to_string(workers_to_start));
} else {
worker_command_args.push_back("-Dray.job.num-java-workers-per-process=" +
std::to_string(workers_to_start));
}
worker_command_args.push_back("-Dray.job.num-java-workers-per-process=" +
std::to_string(workers_to_start));
break;
default:
RAY_LOG(FATAL)
@@ -327,12 +265,12 @@ Process WorkerPool::StartWorkerProcess(
}
ProcessEnvironment env;
if (RayConfig::instance().enable_multi_tenancy() && !IsIOWorkerType(worker_type)) {
if (!IsIOWorkerType(worker_type)) {
// We pass the job ID to worker processes via an environment variable, so we don't
// need to add a new CLI parameter for both Python and Java workers.
env.emplace(kEnvVarKeyJobId, job_id.Hex());
}
if (RayConfig::instance().enable_multi_tenancy() && job_config) {
if (job_config) {
env.insert(job_config->worker_env().begin(), job_config->worker_env().end());
}
@@ -516,18 +454,16 @@ void WorkerPool::OnWorkerStarted(const std::shared_ptr<WorkerInterface> &worker)
io_worker_state.num_starting_io_workers--;
}
if (RayConfig::instance().enable_multi_tenancy()) {
// This is a workaround to finish driver registration after all initial workers are
// registered to Raylet if and only if Raylet is started by a Python driver and the
// job config is not set in `ray.init(...)`.
if (first_job_ == worker->GetAssignedJobId() &&
worker->GetLanguage() == Language::PYTHON) {
if (++first_job_registered_python_worker_count_ ==
first_job_driver_wait_num_python_workers_) {
if (first_job_send_register_client_reply_to_driver_) {
first_job_send_register_client_reply_to_driver_();
first_job_send_register_client_reply_to_driver_ = nullptr;
}
// This is a workaround to finish driver registration after all initial workers are
// registered to Raylet if and only if Raylet is started by a Python driver and the
// job config is not set in `ray.init(...)`.
if (first_job_ == worker->GetAssignedJobId() &&
worker->GetLanguage() == Language::PYTHON) {
if (++first_job_registered_python_worker_count_ ==
first_job_driver_wait_num_python_workers_) {
if (first_job_send_register_client_reply_to_driver_) {
first_job_send_register_client_reply_to_driver_();
first_job_send_register_client_reply_to_driver_ = nullptr;
}
}
}
@@ -554,18 +490,15 @@ Status WorkerPool::RegisterDriver(const std::shared_ptr<WorkerInterface> &driver
// Invoke the `send_reply_callback` later to only finish driver
// registration after all initial workers are registered to Raylet.
bool delay_callback = false;
// Multi-tenancy is enabled.
if (RayConfig().instance().enable_multi_tenancy()) {
// If this is the first job.
if (first_job_.IsNil()) {
first_job_ = job_id;
// If the number of Python workers we need to wait is positive.
if (num_initial_python_workers_for_first_job_ > 0) {
delay_callback = true;
// Start initial Python workers for the first job.
for (int i = 0; i < num_initial_python_workers_for_first_job_; i++) {
StartWorkerProcess(Language::PYTHON, rpc::WorkerType::WORKER, job_id);
}
// If this is the first job.
if (first_job_.IsNil()) {
first_job_ = job_id;
// If the number of Python workers we need to wait is positive.
if (num_initial_python_workers_for_first_job_ > 0) {
delay_callback = true;
// Start initial Python workers for the first job.
for (int i = 0; i < num_initial_python_workers_for_first_job_; i++) {
StartWorkerProcess(Language::PYTHON, rpc::WorkerType::WORKER, job_id);
}
}
}
@@ -695,11 +628,9 @@ void WorkerPool::PushWorker(const std::shared_ptr<WorkerInterface> &worker) {
// Put the worker to the corresponding idle pool.
if (worker->GetActorId().IsNil()) {
state.idle.insert(worker);
if (RayConfig::instance().enable_multi_tenancy()) {
int64_t now = current_time_ms();
idle_of_all_languages_.emplace_back(worker, now);
idle_of_all_languages_map_[worker] = now;
}
int64_t now = current_time_ms();
idle_of_all_languages_.emplace_back(worker, now);
idle_of_all_languages_map_[worker] = now;
} else {
state.idle_actor[worker->GetActorId()] = worker;
}
@@ -858,40 +789,28 @@ std::shared_ptr<WorkerInterface> WorkerPool::PopWorker(
}
} else if (!task_spec.IsActorTask()) {
// Code path of normal task or actor creation task without dynamic worker options.
if (!RayConfig::instance().enable_multi_tenancy()) {
if (!state.idle.empty()) {
worker = std::move(*state.idle.begin());
state.idle.erase(state.idle.begin());
} else {
// There are no more non-actor workers available to execute this task.
// Start a new worker process.
proc = StartWorkerProcess(task_spec.GetLanguage(), rpc::WorkerType::WORKER,
JobID::Nil());
}
} else {
// Find an available worker which is already assigned to this job.
// Try to pop the most recently pushed worker.
for (auto it = idle_of_all_languages_.rbegin(); it != idle_of_all_languages_.rend();
it++) {
if (task_spec.GetLanguage() != it->first->GetLanguage() ||
it->first->GetAssignedJobId() != task_spec.JobId()) {
continue;
}
state.idle.erase(it->first);
// We can't erase a reverse_iterator.
auto lit = it.base();
lit--;
worker = std::move(lit->first);
idle_of_all_languages_.erase(lit);
idle_of_all_languages_map_.erase(worker);
break;
}
if (worker == nullptr) {
// There are no more non-actor workers available to execute this task.
// Start a new worker process.
proc = StartWorkerProcess(task_spec.GetLanguage(), rpc::WorkerType::WORKER,
task_spec.JobId());
// Find an available worker which is already assigned to this job.
// Try to pop the most recently pushed worker.
for (auto it = idle_of_all_languages_.rbegin(); it != idle_of_all_languages_.rend();
it++) {
if (task_spec.GetLanguage() != it->first->GetLanguage() ||
it->first->GetAssignedJobId() != task_spec.JobId()) {
continue;
}
state.idle.erase(it->first);
// We can't erase a reverse_iterator.
auto lit = it.base();
lit--;
worker = std::move(lit->first);
idle_of_all_languages_.erase(lit);
idle_of_all_languages_map_.erase(worker);
break;
}
if (worker == nullptr) {
// There are no more non-actor workers available to execute this task.
// Start a new worker process.
proc = StartWorkerProcess(task_spec.GetLanguage(), rpc::WorkerType::WORKER,
task_spec.JobId());
}
} else {
// Code path of actor task.
@@ -907,7 +826,7 @@ std::shared_ptr<WorkerInterface> WorkerPool::PopWorker(
WarnAboutSize();
}
if (RayConfig::instance().enable_multi_tenancy() && worker) {
if (worker) {
RAY_CHECK(worker->GetAssignedJobId() == task_spec.JobId());
}
return worker;
src/ray/raylet/worker_pool.h
+4 -13
View File
@@ -95,7 +95,6 @@ class WorkerPool : public WorkerPoolInterface, public IOWorkerPoolInterface {
/// process should create and register the specified number of workers, and add them to
/// the pool.
///
/// \param num_workers The number of workers to start, per language.
/// \param num_workers_soft_limit The soft limit of the number of workers.
/// \param num_initial_python_workers_for_first_job The number of initial Python
/// workers for the first job.
@@ -113,8 +112,8 @@ class WorkerPool : public WorkerPoolInterface, public IOWorkerPoolInterface {
/// \param raylet_config The raylet config list of this node.
/// \param starting_worker_timeout_callback The callback that will be triggered once
/// it times out to start a worker.
WorkerPool(boost::asio::io_service &io_service, int num_workers,
int num_workers_soft_limit, int num_initial_python_workers_for_first_job,
WorkerPool(boost::asio::io_service &io_service, int num_workers_soft_limit,
int num_initial_python_workers_for_first_job,
int maximum_startup_concurrency, int min_worker_port, int max_worker_port,
const std::vector<int> &worker_ports,
std::shared_ptr<gcs::GcsClient> gcs_client,
@@ -251,7 +250,7 @@ class WorkerPool : public WorkerPoolInterface, public IOWorkerPoolInterface {
/// Try to prestart a number of workers suitable the given task spec. Prestarting
/// is needed since core workers request one lease at a time, if starting is slow,
/// then it means it takes a long time to scale up when multi-tenancy is on.
/// then it means it takes a long time to scale up.
///
/// \param task_spec The returned worker must be able to execute this task.
/// \param backlog_size The number of tasks in the client backlog of this shape.
@@ -306,7 +305,7 @@ class WorkerPool : public WorkerPoolInterface, public IOWorkerPoolInterface {
protected:
/// Asynchronously start a new worker process. Once the worker process has
/// registered with an external server, the process should create and
/// register num_workers_per_process workers, then add them to the pool.
/// register N workers, then add them to the pool.
/// Failure to start the worker process is a fatal error. If too many workers
/// are already being started, then this function will return without starting
/// any workers.
@@ -354,8 +353,6 @@ class WorkerPool : public WorkerPoolInterface, public IOWorkerPoolInterface {
struct State {
/// The commands and arguments used to start the worker process
std::vector<std::string> worker_command;
/// The number of workers per process.
int num_workers_per_process;
/// The pool of dedicated workers for actor creation tasks
/// with prefix or suffix worker command.
std::unordered_map<TaskID, std::shared_ptr<WorkerInterface>> idle_dedicated_workers;
@@ -392,12 +389,6 @@ class WorkerPool : public WorkerPoolInterface, public IOWorkerPoolInterface {
std::unordered_map<Language, State, std::hash<int>> states_by_lang_;
private:
/// Force-start at least num_workers workers for this language. Used for internal and
/// test purpose only.
///
/// \param num_workers The number of workers to start, per language.
void Start(int num_workers);
/// A helper function that returns the reference of the pool state
/// for a given language.
State &GetStateForLanguage(const Language &language);
src/ray/raylet/worker_pool_test.cc
+48 -83
View File
@@ -36,12 +36,9 @@ class WorkerPoolMock : public WorkerPool {
public:
explicit WorkerPoolMock(boost::asio::io_service &io_service,
const WorkerCommandMap &worker_commands)
: WorkerPool(io_service, 0, POOL_SIZE_SOFT_LIMIT, 0, MAXIMUM_STARTUP_CONCURRENCY, 0,
0, {}, nullptr, worker_commands, {}, []() {}),
last_worker_process_() {
states_by_lang_[ray::Language::JAVA].num_workers_per_process =
NUM_WORKERS_PER_PROCESS_JAVA;
}
: WorkerPool(io_service, POOL_SIZE_SOFT_LIMIT, 0, MAXIMUM_STARTUP_CONCURRENCY, 0, 0,
{}, nullptr, worker_commands, {}, []() {}),
last_worker_process_() {}
~WorkerPoolMock() {
// Avoid killing real processes
@@ -103,14 +100,11 @@ class WorkerPoolMock : public WorkerPool {
std::unordered_map<Process, std::vector<std::string>> worker_commands_by_proc_;
};
class WorkerPoolTest : public ::testing::TestWithParam<bool> {
class WorkerPoolTest : public ::testing::Test {
public:
WorkerPoolTest() : error_message_type_(1), client_call_manager_(io_service_) {
bool enable_multi_tenancy = GetParam();
RayConfig::instance().initialize(
{{"enable_multi_tenancy", std::to_string(enable_multi_tenancy)},
{"num_workers_per_process_java", std::to_string(NUM_WORKERS_PER_PROCESS_JAVA)},
{"object_spilling_config", "mock_config"},
{{"object_spilling_config", "mock_config"},
{"max_io_workers", std::to_string(MAX_IO_WORKER_SIZE)}});
SetWorkerCommands(
{{Language::PYTHON, {"dummy_py_worker_command"}},
@@ -236,16 +230,10 @@ static inline TaskSpecification ExampleTaskSpec(
}
static inline std::string GetNumJavaWorkersPerProcessSystemProperty(int num) {
std::string key;
if (RayConfig::instance().enable_multi_tenancy()) {
key = "ray.job.num-java-workers-per-process";
} else {
key = "ray.raylet.config.num_workers_per_process_java";
}
return std::string("-D") + key + "=" + std::to_string(num);
return std::string("-Dray.job.num-java-workers-per-process=") + std::to_string(num);
}
TEST_P(WorkerPoolTest, CompareWorkerProcessObjects) {
TEST_F(WorkerPoolTest, CompareWorkerProcessObjects) {
typedef Process T;
T a(T::CreateNewDummy()), b(T::CreateNewDummy()), empty = T();
ASSERT_TRUE(empty.IsNull());
@@ -259,7 +247,7 @@ TEST_P(WorkerPoolTest, CompareWorkerProcessObjects) {
ASSERT_TRUE(!std::equal_to<T>()(a, empty));
}
TEST_P(WorkerPoolTest, HandleWorkerRegistration) {
TEST_F(WorkerPoolTest, HandleWorkerRegistration) {
Process proc =
worker_pool_->StartWorkerProcess(Language::JAVA, rpc::WorkerType::WORKER, JOB_ID);
std::vector<std::shared_ptr<WorkerInterface>> workers;
@@ -286,59 +274,49 @@ TEST_P(WorkerPoolTest, HandleWorkerRegistration) {
}
}
TEST_P(WorkerPoolTest, HandleUnknownWorkerRegistration) {
TEST_F(WorkerPoolTest, HandleUnknownWorkerRegistration) {
auto worker = CreateWorker(Process(), Language::PYTHON);
auto status = worker_pool_->RegisterWorker(worker, 1234, [](Status, int) {});
ASSERT_FALSE(status.ok());
}
TEST_P(WorkerPoolTest, StartupPythonWorkerProcessCount) {
TEST_F(WorkerPoolTest, StartupPythonWorkerProcessCount) {
TestStartupWorkerProcessCount(Language::PYTHON, 1, {"dummy_py_worker_command"});
}
TEST_P(WorkerPoolTest, StartupJavaWorkerProcessCount) {
TEST_F(WorkerPoolTest, StartupJavaWorkerProcessCount) {
TestStartupWorkerProcessCount(
Language::JAVA, NUM_WORKERS_PER_PROCESS_JAVA,
{"dummy_java_worker_command",
GetNumJavaWorkersPerProcessSystemProperty(NUM_WORKERS_PER_PROCESS_JAVA)});
}
TEST_P(WorkerPoolTest, InitialWorkerProcessCount) {
if (!RayConfig::instance().enable_multi_tenancy()) {
worker_pool_->Start(1);
// Here we try to start only 1 worker for each worker language. But since we disabled
// initial workers for Java, we expect to see only 1 worker which is a Python worker.
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 1);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 1);
} else {
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 0);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 0);
}
TEST_F(WorkerPoolTest, InitialWorkerProcessCount) {
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 0);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 0);
}
TEST_P(WorkerPoolTest, TestPrestartingWorkers) {
if (RayConfig::instance().enable_multi_tenancy()) {
const auto task_spec = ExampleTaskSpec();
// Prestarts 2 workers.
worker_pool_->PrestartWorkers(task_spec, 2);
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 2);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 2);
// Prestarts 1 more worker.
worker_pool_->PrestartWorkers(task_spec, 3);
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 3);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 3);
// No more needed.
worker_pool_->PrestartWorkers(task_spec, 1);
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 3);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 3);
// Capped by soft limit of 5.
worker_pool_->PrestartWorkers(task_spec, 20);
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 5);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 5);
}
TEST_F(WorkerPoolTest, TestPrestartingWorkers) {
const auto task_spec = ExampleTaskSpec();
// Prestarts 2 workers.
worker_pool_->PrestartWorkers(task_spec, 2);
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 2);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 2);
// Prestarts 1 more worker.
worker_pool_->PrestartWorkers(task_spec, 3);
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 3);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 3);
// No more needed.
worker_pool_->PrestartWorkers(task_spec, 1);
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 3);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 3);
// Capped by soft limit of 5.
worker_pool_->PrestartWorkers(task_spec, 20);
ASSERT_EQ(worker_pool_->NumWorkersStarting(), 5);
ASSERT_EQ(worker_pool_->NumWorkerProcessesStarting(), 5);
}
TEST_P(WorkerPoolTest, HandleWorkerPushPop) {
TEST_F(WorkerPoolTest, HandleWorkerPushPop) {
// Try to pop a worker from the empty pool and make sure we don't get one.
std::shared_ptr<WorkerInterface> popped_worker;
const auto task_spec = ExampleTaskSpec();
@@ -365,7 +343,7 @@ TEST_P(WorkerPoolTest, HandleWorkerPushPop) {
ASSERT_EQ(popped_worker, nullptr);
}
TEST_P(WorkerPoolTest, PopActorWorker) {
TEST_F(WorkerPoolTest, PopActorWorker) {
// Create a worker.
auto worker = CreateWorker(Process::CreateNewDummy());
// Add the worker to the pool.
@@ -387,7 +365,7 @@ TEST_P(WorkerPoolTest, PopActorWorker) {
ASSERT_EQ(actor->GetActorId(), actor_id);
}
TEST_P(WorkerPoolTest, PopWorkersOfMultipleLanguages) {
TEST_F(WorkerPoolTest, PopWorkersOfMultipleLanguages) {
// Create a Python Worker, and add it to the pool
auto py_worker = CreateWorker(Process::CreateNewDummy(), Language::PYTHON);
worker_pool_->PushWorker(py_worker);
@@ -405,7 +383,7 @@ TEST_P(WorkerPoolTest, PopWorkersOfMultipleLanguages) {
ASSERT_NE(worker_pool_->PopWorker(java_task_spec), nullptr);
}
TEST_P(WorkerPoolTest, StartWorkerWithDynamicOptionsCommand) {
TEST_F(WorkerPoolTest, StartWorkerWithDynamicOptionsCommand) {
const std::vector<std::string> java_worker_command = {
"RAY_WORKER_DYNAMIC_OPTION_PLACEHOLDER", "dummy_java_worker_command",
"RAY_WORKER_RAYLET_CONFIG_PLACEHOLDER"};
@@ -426,25 +404,15 @@ TEST_P(WorkerPoolTest, StartWorkerWithDynamicOptionsCommand) {
const auto real_command =
worker_pool_->GetWorkerCommand(worker_pool_->LastStartedWorkerProcess());
if (RayConfig::instance().enable_multi_tenancy()) {
ASSERT_EQ(
real_command,
std::vector<std::string>(
{"test_op_0", "test_op_1", "-Dray.job.code-search-path=/test/code_serch_path",
"dummy_java_worker_command", GetNumJavaWorkersPerProcessSystemProperty(1)}));
} else {
ASSERT_EQ(real_command, std::vector<std::string>(
{"test_op_0", "test_op_1", "dummy_java_worker_command",
GetNumJavaWorkersPerProcessSystemProperty(1)}));
}
ASSERT_EQ(
real_command,
std::vector<std::string>(
{"test_op_0", "test_op_1", "-Dray.job.code-search-path=/test/code_serch_path",
"dummy_java_worker_command", GetNumJavaWorkersPerProcessSystemProperty(1)}));
worker_pool_->HandleJobFinished(JOB_ID);
}
TEST_P(WorkerPoolTest, PopWorkerMultiTenancy) {
if (!RayConfig::instance().enable_multi_tenancy()) {
return;
}
TEST_F(WorkerPoolTest, PopWorkerMultiTenancy) {
auto job_id1 = JOB_ID;
auto job_id2 = JobID::FromInt(2);
ASSERT_NE(job_id1, job_id2);
@@ -505,7 +473,7 @@ TEST_P(WorkerPoolTest, PopWorkerMultiTenancy) {
}
}
TEST_P(WorkerPoolTest, MaximumStartupConcurrency) {
TEST_F(WorkerPoolTest, MaximumStartupConcurrency) {
auto task_spec = ExampleTaskSpec();
std::vector<Process> started_processes;
@@ -550,7 +518,7 @@ TEST_P(WorkerPoolTest, MaximumStartupConcurrency) {
ASSERT_EQ(0, worker_pool_->NumWorkerProcessesStarting());
}
TEST_P(WorkerPoolTest, HandleIOWorkersPushPop) {
TEST_F(WorkerPoolTest, HandleIOWorkersPushPop) {
std::unordered_set<std::shared_ptr<WorkerInterface>> spill_pushed_worker;
std::unordered_set<std::shared_ptr<WorkerInterface>> restore_pushed_worker;
auto spill_worker_callback =
@@ -603,7 +571,7 @@ TEST_P(WorkerPoolTest, HandleIOWorkersPushPop) {
ASSERT_EQ(restore_pushed_worker.size(), 1);
}
TEST_P(WorkerPoolTest, MaxIOWorkerSimpleTest) {
TEST_F(WorkerPoolTest, MaxIOWorkerSimpleTest) {
// Make sure max number of spill workers are respected.
auto callback = [](std::shared_ptr<WorkerInterface> worker) {};
std::vector<Process> started_processes;
@@ -633,7 +601,7 @@ TEST_P(WorkerPoolTest, MaxIOWorkerSimpleTest) {
ASSERT_EQ(worker_pool_->NumSpillWorkerStarting(), 0);
}
TEST_P(WorkerPoolTest, MaxIOWorkerComplicateTest) {
TEST_F(WorkerPoolTest, MaxIOWorkerComplicateTest) {
// Make sure max number of restore workers are respected.
// This test will test a little more complicated scneario.
// For example, it tests scenarios where there are
@@ -685,7 +653,7 @@ TEST_P(WorkerPoolTest, MaxIOWorkerComplicateTest) {
ASSERT_EQ(worker_pool_->NumSpillWorkerStarting(), 0);
}
TEST_P(WorkerPoolTest, MaxSpillRestoreWorkersIntegrationTest) {
TEST_F(WorkerPoolTest, MaxSpillRestoreWorkersIntegrationTest) {
auto callback = [](std::shared_ptr<WorkerInterface> worker) {};
// Run many pop spill/restore workers and make sure the max worker size doesn't exceed.
std::vector<Process> started_restore_processes;
@@ -732,7 +700,7 @@ TEST_P(WorkerPoolTest, MaxSpillRestoreWorkersIntegrationTest) {
ASSERT_EQ(worker_pool_->GetProcessSize(), 2 * MAX_IO_WORKER_SIZE);
}
TEST_P(WorkerPoolTest, DeleteWorkerPushPop) {
TEST_F(WorkerPoolTest, DeleteWorkerPushPop) {
/// Make sure delete workers always pop an I/O worker that has more idle worker in their
/// pools.
// 2 spill worker and 1 restore worker.
@@ -770,9 +738,6 @@ TEST_P(WorkerPoolTest, DeleteWorkerPushPop) {
});
}
INSTANTIATE_TEST_CASE_P(WorkerPoolMultiTenancyTest, WorkerPoolTest,
::testing::Values(true, false));
} // namespace raylet
} // namespace ray