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

Reconstruct local dead actors immediately instead of waiting for initial_reconstruction_timeout_ms (#6243)

Browse Source
This commit is contained in:
fangfengbin
2019-11-30 18:03:48 +08:00
committed by Hao Chen
parent 4033d65e4f
commit 7275556365
3 changed files with 83 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/core_worker/test/core_worker_test.cc
+43 -10
View File
@@ -190,6 +190,11 @@ class CoreWorkerTest : public ::testing::Test {
bool WaitForDirectCallActorState(CoreWorker &worker, const ActorID &actor_id,
bool wait_alive, int timeout_ms);
// Get the pid for the worker process that runs the actor.
int GetActorPid(CoreWorker &worker, const ActorID &actor_id,
std::unordered_map<std::string, double> &resources,
bool is_direct_call);
std::vector<std::string> raylet_socket_names_;
std::vector<std::string> raylet_store_socket_names_;
gcs::GcsClientOptions gcs_options_;
@@ -206,6 +211,29 @@ bool CoreWorkerTest::WaitForDirectCallActorState(CoreWorker &worker,
return WaitForCondition(condition_func, timeout_ms);
}
int CoreWorkerTest::GetActorPid(CoreWorker &worker, const ActorID &actor_id,
std::unordered_map<std::string, double> &resources,
bool is_direct_call) {
std::vector<TaskArg> args;
TaskOptions options{1, is_direct_call, resources};
std::vector<ObjectID> return_ids;
RayFunction func{Language::PYTHON, {"GetWorkerPid"}};
RAY_CHECK_OK(worker.SubmitActorTask(actor_id, func, args, options, &return_ids));
std::vector<std::shared_ptr<ray::RayObject>> results;
RAY_CHECK_OK(worker.Get(return_ids, -1, &results));
if (nullptr == results[0]->GetData()) {
// If failed to get actor process pid, return -1
return -1;
}
auto data = reinterpret_cast<char *>(results[0]->GetData()->Data());
std::string pid_string(data, results[0]->GetData()->Size());
return std::stoi(pid_string);
}
void CoreWorkerTest::TestNormalTask(std::unordered_map<std::string, double> &resources) {
CoreWorker driver(WorkerType::DRIVER, Language::PYTHON, raylet_store_socket_names_[0],
raylet_socket_names_[0], NextJobId(), gcs_options_, "", "127.0.0.1",
@@ -226,7 +254,7 @@ void CoreWorkerTest::TestNormalTask(std::unordered_map<std::string, double> &res
TaskArg::PassByValue(std::make_shared<RayObject>(buffer1, nullptr)));
args.emplace_back(TaskArg::PassByReference(object_id));
RayFunction func(ray::Language::PYTHON, {});
RayFunction func(ray::Language::PYTHON, {"MergeInputArgsAsOutput"});
TaskOptions options;
options.is_direct_call = true;
@@ -273,7 +301,7 @@ void CoreWorkerTest::TestActorTask(std::unordered_map<std::string, double> &reso
TaskOptions options{1, false, resources};
std::vector<ObjectID> return_ids;
RayFunction func(ray::Language::PYTHON, {});
RayFunction func(ray::Language::PYTHON, {"MergeInputArgsAsOutput"});
RAY_CHECK_OK(driver.SubmitActorTask(actor_id, func, args, options, &return_ids));
ASSERT_EQ(return_ids.size(), 1);
@@ -313,7 +341,7 @@ void CoreWorkerTest::TestActorTask(std::unordered_map<std::string, double> &reso
TaskOptions options{1, false, resources};
std::vector<ObjectID> return_ids;
RayFunction func(ray::Language::PYTHON, {});
RayFunction func(ray::Language::PYTHON, {"MergeInputArgsAsOutput"});
auto status = driver.SubmitActorTask(actor_id, func, args, options, &return_ids);
ASSERT_TRUE(status.ok());
@@ -344,6 +372,9 @@ void CoreWorkerTest::TestActorReconstruction(
ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_id, true, 30 * 1000 /* 30s */));
RAY_LOG(INFO) << "actor has been created";
auto pid = GetActorPid(driver, actor_id, resources, is_direct_call);
RAY_CHECK(pid != -1);
// Test submitting some tasks with by-value args for that actor.
{
const int num_tasks = 100;
@@ -355,10 +386,12 @@ void CoreWorkerTest::TestActorReconstruction(
ASSERT_EQ(system("pkill mock_worker"), 0);
// Wait for actor restruction event, and then for alive event.
ASSERT_TRUE(
WaitForDirectCallActorState(driver, actor_id, false, 30 * 1000 /* 30s */));
ASSERT_TRUE(
WaitForDirectCallActorState(driver, actor_id, true, 30 * 1000 /* 30s */));
auto check_actor_restart_func = [this, pid, &driver, &actor_id, &resources,
is_direct_call]() -> bool {
auto new_pid = GetActorPid(driver, actor_id, resources, is_direct_call);
return new_pid != -1 && new_pid != pid;
};
ASSERT_TRUE(WaitForCondition(check_actor_restart_func, 30 * 1000 /* 30s */));
RAY_LOG(INFO) << "actor has been reconstructed";
}
@@ -373,7 +406,7 @@ void CoreWorkerTest::TestActorReconstruction(
TaskOptions options{1, false, resources};
std::vector<ObjectID> return_ids;
RayFunction func(ray::Language::PYTHON, {});
RayFunction func(ray::Language::PYTHON, {"MergeInputArgsAsOutput"});
RAY_CHECK_OK(driver.SubmitActorTask(actor_id, func, args, options, &return_ids));
ASSERT_EQ(return_ids.size(), 1);
@@ -418,7 +451,7 @@ void CoreWorkerTest::TestActorFailure(std::unordered_map<std::string, double> &r
TaskOptions options{1, false, resources};
std::vector<ObjectID> return_ids;
RayFunction func(ray::Language::PYTHON, {});
RayFunction func(ray::Language::PYTHON, {"MergeInputArgsAsOutput"});
RAY_CHECK_OK(driver.SubmitActorTask(actor_id, func, args, options, &return_ids));
@@ -566,7 +599,7 @@ TEST_F(SingleNodeTest, TestDirectActorTaskSubmissionPerf) {
TaskOptions options{1, false, resources};
std::vector<ObjectID> return_ids;
RayFunction func(ray::Language::PYTHON, {});
RayFunction func(ray::Language::PYTHON, {"MergeInputArgsAsOutput"});
RAY_CHECK_OK(driver.SubmitActorTask(actor_id, func, args, options, &return_ids));
ASSERT_EQ(return_ids.size(), 1);
src/ray/core_worker/test/mock_worker.cc
+30 -1
View File
@@ -35,8 +35,37 @@ class MockWorker {
const std::vector<ObjectID> &return_ids,
std::vector<std::shared_ptr<RayObject>> *results) {
// Note that this doesn't include dummy object id.
RAY_CHECK(return_ids.size() >= 0);
const std::vector<std::string> &function_descriptor =
ray_function.GetFunctionDescriptor();
RAY_CHECK(return_ids.size() >= 0 && 1 == function_descriptor.size());
if ("actor creation task" == function_descriptor[0]) {
return Status::OK();
} else if ("GetWorkerPid" == function_descriptor[0]) {
// Get mock worker pid
return GetWorkerPid(results);
} else if ("MergeInputArgsAsOutput" == function_descriptor[0]) {
// Merge input args and write the merged content to each of return ids
return MergeInputArgsAsOutput(args, return_ids, results);
} else {
return Status::TypeError("Unknown function descriptor: " + function_descriptor[0]);
}
}
Status GetWorkerPid(std::vector<std::shared_ptr<RayObject>> *results) {
// Save the pid of current process to the return object.
std::string pid_string = std::to_string(static_cast<int>(getpid()));
auto data =
const_cast<uint8_t *>(reinterpret_cast<const uint8_t *>(pid_string.data()));
auto memory_buffer =
std::make_shared<LocalMemoryBuffer>(data, pid_string.size(), true);
results->push_back(std::make_shared<RayObject>(memory_buffer, nullptr));
return Status::OK();
}
Status MergeInputArgsAsOutput(const std::vector<std::shared_ptr<RayObject>> &args,
const std::vector<ObjectID> &return_ids,
std::vector<std::shared_ptr<RayObject>> *results) {
// Merge all the content from input args.
std::vector<uint8_t> buffer;
for (const auto &arg : args) {
src/ray/raylet/node_manager.cc
+10
View File
@@ -783,6 +783,7 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id,
// creation task.
reconstruction_policy_.ListenAndMaybeReconstruct(
actor_registration.GetActorCreationDependency());
// When an actor fails but can be reconstructed, resubmit all of the queued
// tasks for that actor. This will mark the tasks as waiting for actor
// creation.
@@ -1060,6 +1061,15 @@ void NodeManager::HandleDisconnectedActor(const ActorID &actor_id, bool was_loca
};
auto actor_notification = std::make_shared<ActorTableData>(new_actor_info);
RAY_CHECK_OK(gcs_client_->Actors().AsyncUpdate(actor_id, actor_notification, done));
if (was_local && new_state == ActorTableData::RECONSTRUCTING) {
RAY_LOG(INFO) << "A local actor (id = " << actor_id
<< " ) is dead, reconstructing it.";
const ObjectID &actor_creation_dummy_object_id =
actor_registration.GetActorCreationDependency();
HandleTaskReconstruction(actor_creation_dummy_object_id.TaskId(),
actor_creation_dummy_object_id);
}
}
void NodeManager::HandleWorkerAvailable(