From 72755563652ea153c0dc60c95e233f31a4c3082a Mon Sep 17 00:00:00 2001 From: fangfengbin <869218239a@zju.edu.cn> Date: Sat, 30 Nov 2019 18:03:48 +0800 Subject: [PATCH] Reconstruct local dead actors immediately instead of waiting for initial_reconstruction_timeout_ms (#6243) --- src/ray/core_worker/test/core_worker_test.cc | 53 ++++++++++++++++---- src/ray/core_worker/test/mock_worker.cc | 31 +++++++++++- src/ray/raylet/node_manager.cc | 10 ++++ 3 files changed, 83 insertions(+), 11 deletions(-) diff --git a/src/ray/core_worker/test/core_worker_test.cc b/src/ray/core_worker/test/core_worker_test.cc index 1f2d35a15..f9c1c1793 100644 --- a/src/ray/core_worker/test/core_worker_test.cc +++ b/src/ray/core_worker/test/core_worker_test.cc @@ -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 &resources, + bool is_direct_call); + std::vector raylet_socket_names_; std::vector 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 &resources, + bool is_direct_call) { + std::vector args; + TaskOptions options{1, is_direct_call, resources}; + std::vector return_ids; + RayFunction func{Language::PYTHON, {"GetWorkerPid"}}; + + RAY_CHECK_OK(worker.SubmitActorTask(actor_id, func, args, options, &return_ids)); + + std::vector> 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(results[0]->GetData()->Data()); + std::string pid_string(data, results[0]->GetData()->Size()); + return std::stoi(pid_string); +} + void CoreWorkerTest::TestNormalTask(std::unordered_map &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 &res TaskArg::PassByValue(std::make_shared(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 &reso TaskOptions options{1, false, resources}; std::vector 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 &reso TaskOptions options{1, false, resources}; std::vector 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 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 &r TaskOptions options{1, false, resources}; std::vector 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 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); diff --git a/src/ray/core_worker/test/mock_worker.cc b/src/ray/core_worker/test/mock_worker.cc index 096e9f372..754d165b0 100644 --- a/src/ray/core_worker/test/mock_worker.cc +++ b/src/ray/core_worker/test/mock_worker.cc @@ -35,8 +35,37 @@ class MockWorker { const std::vector &return_ids, std::vector> *results) { // Note that this doesn't include dummy object id. - RAY_CHECK(return_ids.size() >= 0); + const std::vector &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> *results) { + // Save the pid of current process to the return object. + std::string pid_string = std::to_string(static_cast(getpid())); + auto data = + const_cast(reinterpret_cast(pid_string.data())); + auto memory_buffer = + std::make_shared(data, pid_string.size(), true); + results->push_back(std::make_shared(memory_buffer, nullptr)); + return Status::OK(); + } + + Status MergeInputArgsAsOutput(const std::vector> &args, + const std::vector &return_ids, + std::vector> *results) { // Merge all the content from input args. std::vector buffer; for (const auto &arg : args) { diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 0e39e35b9..c450f2007 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -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(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(