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

Rename max_reconstructions to max_restarts and use -1 for infinite (#8274)

Browse Source 
Co-authored-by: Edward Oakes <ed.nmi.oakes@gmail.com>
This commit is contained in:
Max Fitton
2020-05-14 10:30:29 -05:00
committed by GitHub
co-authored by Edward Oakes
parent 5f4c196fed
commit 00325eb2b2
71 changed files with 403 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/common/ray_config_def.h
+1 -1
View File
@@ -117,7 +117,7 @@ RAY_CONFIG(int64_t, max_direct_call_object_size, 100 * 1024)
RAY_CONFIG(int64_t, max_grpc_message_size, 100 * 1024 * 1024)
// The min number of retries for direct actor creation tasks. The actual number
// of creation retries will be MAX(actor_creation_min_retries, max_reconstructions).
// of creation retries will be MAX(actor_creation_min_retries, max_restarts).
RAY_CONFIG(uint64_t, actor_creation_min_retries, 3)
/// The initial period for a task execution lease. The lease will expire this
src/ray/common/task/task_spec.cc
+5 -4
View File
@@ -1,6 +1,7 @@
#include "ray/common/task/task_spec.h"
#include <sstream>
#include "ray/common/task/task_spec.h"
#include "ray/util/logging.h"
namespace ray {
@@ -189,9 +190,9 @@ ActorID TaskSpecification::ActorCreationId() const {
return ActorID::FromBinary(message_->actor_creation_task_spec().actor_id());
}
uint64_t TaskSpecification::MaxActorReconstructions() const {
int64_t TaskSpecification::MaxActorRestarts() const {
RAY_CHECK(IsActorCreationTask());
return message_->actor_creation_task_spec().max_actor_reconstructions();
return message_->actor_creation_task_spec().max_actor_restarts();
}
std::vector<std::string> TaskSpecification::DynamicWorkerOptions() const {
@@ -266,7 +267,7 @@ std::string TaskSpecification::DebugString() const {
if (IsActorCreationTask()) {
// Print actor creation task spec.
stream << ", actor_creation_task_spec={actor_id=" << ActorCreationId()
<< ", max_reconstructions=" << MaxActorReconstructions()
<< ", max_restarts=" << MaxActorRestarts()
<< ", max_concurrency=" << MaxActorConcurrency()
<< ", is_asyncio_actor=" << IsAsyncioActor()
<< ", is_detached=" << IsDetachedActor() << "}";
src/ray/common/task/task_spec.h
+1 -1
View File
@@ -139,7 +139,7 @@ class TaskSpecification : public MessageWrapper<rpc::TaskSpec> {
ActorID ActorCreationId() const;
uint64_t MaxActorReconstructions() const;
int64_t MaxActorRestarts() const;
std::vector<std::string> DynamicWorkerOptions() const;
src/ray/common/task/task_util.h
+2 -2
View File
@@ -101,14 +101,14 @@ class TaskSpecBuilder {
///
/// \return Reference to the builder object itself.
TaskSpecBuilder &SetActorCreationTaskSpec(
const ActorID &actor_id, uint64_t max_reconstructions = 0,
const ActorID &actor_id, int64_t max_restarts = 0,
const std::vector<std::string> &dynamic_worker_options = {},
int max_concurrency = 1, bool is_detached = false, std::string name = "",
bool is_asyncio = false, const std::string &extension_data = "") {
message_->set_type(TaskType::ACTOR_CREATION_TASK);
auto actor_creation_spec = message_->mutable_actor_creation_task_spec();
actor_creation_spec->set_actor_id(actor_id.Binary());
actor_creation_spec->set_max_actor_reconstructions(max_reconstructions);
actor_creation_spec->set_max_actor_restarts(max_restarts);
for (const auto &option : dynamic_worker_options) {
actor_creation_spec->add_dynamic_worker_options(option);
}
src/ray/core_worker/common.h
+5 -4
View File
@@ -111,12 +111,12 @@ struct TaskOptions {
/// Options for actor creation tasks.
struct ActorCreationOptions {
ActorCreationOptions() {}
ActorCreationOptions(uint64_t max_reconstructions, int max_concurrency,
ActorCreationOptions(int64_t max_restarts, int max_concurrency,
const std::unordered_map<std::string, double> &resources,
const std::unordered_map<std::string, double> &placement_resources,
const std::vector<std::string> &dynamic_worker_options,
bool is_detached, std::string &name, bool is_asyncio)
: max_reconstructions(max_reconstructions),
: max_restarts(max_restarts),
max_concurrency(max_concurrency),
resources(resources),
placement_resources(placement_resources),
@@ -126,8 +126,9 @@ struct ActorCreationOptions {
is_asyncio(is_asyncio){};
/// Maximum number of times that the actor should be reconstructed when it dies
/// unexpectedly. It must be non-negative. If it's 0, the actor won't be reconstructed.
const uint64_t max_reconstructions = 0;
/// unexpectedly. A value of -1 indicates infinite restarts.
/// If it's 0, the actor won't be restarted.
const int64_t max_restarts = 0;
/// The max number of concurrent tasks to run on this direct call actor.
const int max_concurrency = 1;
/// Resources required by the whole lifetime of this actor.
src/ray/core_worker/core_worker.cc
+15 -11
View File
@@ -1146,7 +1146,7 @@ Status CoreWorker::CreateActor(const RayFunction &function,
rpc_address_, function, args, 1, actor_creation_options.resources,
actor_creation_options.placement_resources, &return_ids);
builder.SetActorCreationTaskSpec(
actor_id, actor_creation_options.max_reconstructions,
actor_id, actor_creation_options.max_restarts,
actor_creation_options.dynamic_worker_options,
actor_creation_options.max_concurrency, actor_creation_options.is_detached,
actor_creation_options.name, actor_creation_options.is_asyncio, extension_data);
@@ -1167,10 +1167,15 @@ Status CoreWorker::CreateActor(const RayFunction &function,
if (options_.is_local_mode) {
ExecuteTaskLocalMode(task_spec);
} else {
task_manager_->AddPendingTask(
GetCallerId(), rpc_address_, task_spec, CurrentCallSite(),
std::max(RayConfig::instance().actor_creation_min_retries(),
actor_creation_options.max_reconstructions));
int max_retries;
if (actor_creation_options.max_restarts == -1) {
max_retries = -1;
} else {
max_retries = std::max((int64_t)RayConfig::instance().actor_creation_min_retries(),
actor_creation_options.max_restarts);
}
task_manager_->AddPendingTask(GetCallerId(), rpc_address_, task_spec,
CurrentCallSite(), max_retries);
status = direct_task_submitter_->SubmitTask(task_spec);
}
return status;
@@ -1243,11 +1248,10 @@ Status CoreWorker::CancelTask(const ObjectID &object_id, bool force_kill) {
return Status::OK();
}
Status CoreWorker::KillActor(const ActorID &actor_id, bool force_kill,
bool no_reconstruction) {
Status CoreWorker::KillActor(const ActorID &actor_id, bool force_kill, bool no_restart) {
ActorHandle *actor_handle = nullptr;
RAY_RETURN_NOT_OK(GetActorHandle(actor_id, &actor_handle));
direct_actor_submitter_->KillActor(actor_id, force_kill, no_reconstruction);
direct_actor_submitter_->KillActor(actor_id, force_kill, no_restart);
return Status::OK();
}
@@ -1308,7 +1312,7 @@ bool CoreWorker::AddActorHandle(std::unique_ptr<ActorHandle> actor_handle,
const gcs::ActorTableData &actor_data) {
if (actor_data.state() == gcs::ActorTableData::PENDING) {
// The actor is being created and not yet ready, just ignore!
} else if (actor_data.state() == gcs::ActorTableData::RECONSTRUCTING) {
} else if (actor_data.state() == gcs::ActorTableData::RESTARTING) {
absl::MutexLock lock(&actor_handles_mutex_);
auto it = actor_handles_.find(actor_id);
RAY_CHECK(it != actor_handles_.end());
@@ -1355,7 +1359,7 @@ bool CoreWorker::AddActorHandle(std::unique_ptr<ActorHandle> actor_handle,
<< " has gone out of scope, sending message to actor "
<< actor_id << " to do a clean exit.";
RAY_CHECK_OK(
KillActor(actor_id, /*force_kill=*/false, /*no_reconstruction=*/false));
KillActor(actor_id, /*force_kill=*/false, /*no_restart=*/false));
}
}
@@ -1941,7 +1945,7 @@ void CoreWorker::HandleKillActor(const rpc::KillActorRequest &request,
if (request.force_kill()) {
RAY_LOG(INFO) << "Got KillActor, exiting immediately...";
if (request.no_reconstruction()) {
if (request.no_restart()) {
RAY_IGNORE_EXPR(local_raylet_client_->Disconnect());
}
if (options_.num_workers > 1) {
src/ray/core_worker/core_worker.h
+3 -3
View File
@@ -585,10 +585,10 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
/// Tell an actor to exit immediately, without completing outstanding work.
///
/// \param[in] actor_id ID of the actor to kill.
/// \param[in] no_reconstruction If set to true, the killed actor will not be
/// reconstructed anymore.
/// \param[in] no_restart If set to true, the killed actor will not be
/// restarted anymore.
/// \param[out] Status
Status KillActor(const ActorID &actor_id, bool force_kill, bool no_reconstruction);
Status KillActor(const ActorID &actor_id, bool force_kill, bool no_restart);
/// Stops the task associated with the given Object ID.
///
src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc
+2 -2
View File
@@ -155,10 +155,10 @@ JNIEXPORT void JNICALL Java_io_ray_runtime_RayNativeRuntime_nativeSetResource(
}
JNIEXPORT void JNICALL Java_io_ray_runtime_RayNativeRuntime_nativeKillActor(
JNIEnv *env, jclass, jbyteArray actorId, jboolean noReconstruction) {
JNIEnv *env, jclass, jbyteArray actorId, jboolean noRestart) {
auto status = ray::CoreWorkerProcess::GetCoreWorker().KillActor(
JavaByteArrayToId<ActorID>(env, actorId),
/*force_kill=*/true, noReconstruction);
/*force_kill=*/true, noRestart);
THROW_EXCEPTION_AND_RETURN_IF_NOT_OK(env, status, (void)0);
}
src/ray/core_worker/lib/java/io_ray_runtime_task_NativeTaskSubmitter.cc
+13 -12
View File
@@ -13,7 +13,9 @@
// limitations under the License.
#include "ray/core_worker/lib/java/io_ray_runtime_task_NativeTaskSubmitter.h"
#include <jni.h>
#include "ray/common/id.h"
#include "ray/core_worker/common.h"
#include "ray/core_worker/core_worker.h"
@@ -87,13 +89,13 @@ inline ray::TaskOptions ToTaskOptions(JNIEnv *env, jint numReturns, jobject call
inline ray::ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
jobject actorCreationOptions) {
uint64_t max_reconstructions = 0;
int64_t max_restarts = 0;
std::unordered_map<std::string, double> resources;
std::vector<std::string> dynamic_worker_options;
uint64_t max_concurrency = 1;
if (actorCreationOptions) {
max_reconstructions = static_cast<uint64_t>(env->GetIntField(
actorCreationOptions, java_actor_creation_options_max_reconstructions));
max_restarts =
env->GetIntField(actorCreationOptions, java_actor_creation_options_max_restarts);
jobject java_resources =
env->GetObjectField(actorCreationOptions, java_base_task_options_resources);
resources = ToResources(env, java_resources);
@@ -108,15 +110,14 @@ inline ray::ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
}
std::string name = "";
ray::ActorCreationOptions actor_creation_options{
static_cast<uint64_t>(max_reconstructions),
static_cast<int>(max_concurrency),
resources,
resources,
dynamic_worker_options,
/*is_detached=*/false,
name,
/*is_asyncio=*/false};
ray::ActorCreationOptions actor_creation_options{max_restarts,
static_cast<int>(max_concurrency),
resources,
resources,
dynamic_worker_options,
/*is_detached=*/false,
name,
/*is_asyncio=*/false};
return actor_creation_options;
}
src/ray/core_worker/lib/java/jni_init.cc
+3 -3
View File
@@ -66,7 +66,7 @@ jclass java_base_task_options_class;
jfieldID java_base_task_options_resources;
jclass java_actor_creation_options_class;
jfieldID java_actor_creation_options_max_reconstructions;
jfieldID java_actor_creation_options_max_restarts;
jfieldID java_actor_creation_options_jvm_options;
jfieldID java_actor_creation_options_max_concurrency;
@@ -169,8 +169,8 @@ jint JNI_OnLoad(JavaVM *vm, void *reserved) {
java_actor_creation_options_class =
LoadClass(env, "io/ray/api/options/ActorCreationOptions");
java_actor_creation_options_max_reconstructions =
env->GetFieldID(java_actor_creation_options_class, "maxReconstructions", "I");
java_actor_creation_options_max_restarts =
env->GetFieldID(java_actor_creation_options_class, "maxRestarts", "I");
java_actor_creation_options_jvm_options = env->GetFieldID(
java_actor_creation_options_class, "jvmOptions", "Ljava/lang/String;");
java_actor_creation_options_max_concurrency =
src/ray/core_worker/lib/java/jni_utils.h
+3 -2
View File
@@ -16,6 +16,7 @@
#define RAY_COMMON_JAVA_JNI_UTILS_H
#include <jni.h>
#include "ray/common/buffer.h"
#include "ray/common/function_descriptor.h"
#include "ray/common/id.h"
@@ -111,8 +112,8 @@ extern jfieldID java_base_task_options_resources;
/// ActorCreationOptions class
extern jclass java_actor_creation_options_class;
/// maxReconstructions field of ActorCreationOptions class
extern jfieldID java_actor_creation_options_max_reconstructions;
/// maxRestarts field of ActorCreationOptions class
extern jfieldID java_actor_creation_options_max_restarts;
/// jvmOptions field of ActorCreationOptions class
extern jfieldID java_actor_creation_options_jvm_options;
/// maxConcurrency field of ActorCreationOptions class
src/ray/core_worker/object_recovery_manager.cc
+1 -1
View File
@@ -33,7 +33,7 @@ Status ObjectRecoveryManager::RecoverObject(const ObjectID &object_id) {
{
absl::MutexLock lock(&mu_);
// Mark that we are attempting recovery for this object to prevent
// duplicate reconstructions of the same object.
// duplicate restarts of the same object.
already_pending_recovery = !objects_pending_recovery_.insert(object_id).second;
}
}
src/ray/core_worker/task_manager.cc
+10 -6
View File
@@ -94,7 +94,7 @@ Status TaskManager::ResubmitTask(const TaskID &task_id,
if (!it->second.pending) {
resubmit = true;
it->second.pending = true;
RAY_CHECK(it->second.num_retries_left > 0);
RAY_CHECK(it->second.num_retries_left != 0);
it->second.num_retries_left--;
spec = it->second.spec;
}
@@ -241,8 +241,8 @@ void TaskManager::CompletePendingTask(const TaskID &task_id,
// A finished task can be only be re-executed if it has some number of
// retries left and returned at least one object that is still in use and
// stored in plasma.
bool task_retryable =
it->second.num_retries_left > 0 && !it->second.reconstructable_return_ids.empty();
bool task_retryable = it->second.num_retries_left != 0 &&
!it->second.reconstructable_return_ids.empty();
if (task_retryable) {
// Pin the task spec if it may be retried again.
release_lineage = false;
@@ -277,8 +277,10 @@ void TaskManager::PendingTaskFailed(const TaskID &task_id, rpc::ErrorType error_
if (num_retries_left == 0) {
submissible_tasks_.erase(it);
num_pending_tasks_--;
} else if (num_retries_left == -1) {
release_lineage = false;
} else {
RAY_CHECK(it->second.num_retries_left > 0);
RAY_CHECK(num_retries_left > 0);
it->second.num_retries_left--;
release_lineage = false;
}
@@ -286,8 +288,10 @@ void TaskManager::PendingTaskFailed(const TaskID &task_id, rpc::ErrorType error_
// We should not hold the lock during these calls because they may trigger
// callbacks in this or other classes.
if (num_retries_left > 0) {
RAY_LOG(ERROR) << num_retries_left << " retries left for task " << spec.TaskId()
if (num_retries_left != 0) {
auto retries_str =
num_retries_left == -1 ? "infinite" : std::to_string(num_retries_left);
RAY_LOG(ERROR) << retries_str << " retries left for task " << spec.TaskId()
<< ", attempting to resubmit.";
retry_task_callback_(spec, /*delay=*/true);
} else {
src/ray/core_worker/test/core_worker_test.cc
+10 -10
View File
@@ -58,7 +58,7 @@ static void flushall_redis(void) {
}
ActorID CreateActorHelper(std::unordered_map<std::string, double> &resources,
uint64_t max_reconstructions) {
int64_t max_restarts) {
std::unique_ptr<ActorHandle> actor_handle;
uint8_t array[] = {1, 2, 3};
@@ -72,7 +72,7 @@ ActorID CreateActorHelper(std::unordered_map<std::string, double> &resources,
std::string name = "";
ActorCreationOptions actor_options{
max_reconstructions,
max_restarts,
/*max_concurrency*/ 1, resources, resources, {},
/*is_detached=*/false, name, /*is_asyncio=*/false};
@@ -301,7 +301,7 @@ class CoreWorkerTest : public ::testing::Test {
// Test actor failover case. Verify that actor can be reconstructed successfully,
// and as long as we wait for actor reconstruction before submitting new tasks,
// it is guaranteed that all tasks are successfully completed.
void TestActorReconstruction(std::unordered_map<std::string, double> &resources);
void TestActorRestart(std::unordered_map<std::string, double> &resources);
protected:
bool WaitForDirectCallActorState(const ActorID &actor_id, bool wait_alive,
@@ -481,7 +481,7 @@ void CoreWorkerTest::TestActorTask(std::unordered_map<std::string, double> &reso
}
}
void CoreWorkerTest::TestActorReconstruction(
void CoreWorkerTest::TestActorRestart(
std::unordered_map<std::string, double> &resources) {
auto &driver = CoreWorkerProcess::GetCoreWorker();
@@ -512,10 +512,10 @@ void CoreWorkerTest::TestActorReconstruction(
};
ASSERT_TRUE(WaitForCondition(check_actor_restart_func, 30 * 1000 /* 30s */));
RAY_LOG(INFO) << "actor has been reconstructed";
RAY_LOG(INFO) << "actor has been restarted";
}
// wait for actor being reconstructed.
// wait for actor being restarted.
auto buffer1 = GenerateRandomBuffer();
// Create arguments with PassByValue.
@@ -558,7 +558,7 @@ void CoreWorkerTest::TestActorFailure(
ASSERT_EQ(system("pkill mock_worker"), 0);
}
// wait for actor being reconstructed.
// wait for actor being restarted.
auto buffer1 = GenerateRandomBuffer();
// Create arguments with PassByRef and PassByValue.
@@ -699,7 +699,7 @@ TEST_F(SingleNodeTest, TestDirectActorTaskSubmissionPerf) {
// Create an actor.
std::unordered_map<std::string, double> resources;
auto actor_id = CreateActorHelper(resources,
/*max_reconstructions=*/0);
/*max_restarts=*/0);
// wait for actor creation finish.
ASSERT_TRUE(WaitForDirectCallActorState(actor_id, true, 30 * 1000 /* 30s */));
// Test submitting some tasks with by-value args for that actor.
@@ -1002,13 +1002,13 @@ TEST_F(TwoNodeTest, TestActorTaskCrossNodes) {
TEST_F(SingleNodeTest, TestActorTaskLocalReconstruction) {
std::unordered_map<std::string, double> resources;
TestActorReconstruction(resources);
TestActorRestart(resources);
}
TEST_F(TwoNodeTest, TestActorTaskCrossNodesReconstruction) {
std::unordered_map<std::string, double> resources;
resources.emplace("resource1", 1);
TestActorReconstruction(resources);
TestActorRestart(resources);
}
TEST_F(SingleNodeTest, TestActorTaskLocalFailure) {
src/ray/core_worker/test/object_recovery_manager_test.cc
+2 -2
View File
@@ -12,14 +12,14 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#include "ray/core_worker/transport/direct_task_transport.h"
#include "ray/core_worker/object_recovery_manager.h"
#include "gtest/gtest.h"
#include "ray/common/task/task_spec.h"
#include "ray/common/task/task_util.h"
#include "ray/common/test_util.h"
#include "ray/core_worker/object_recovery_manager.h"
#include "ray/core_worker/store_provider/memory_store/memory_store.h"
#include "ray/core_worker/transport/direct_task_transport.h"
#include "ray/raylet/raylet_client.h"
namespace ray {
src/ray/core_worker/test/task_manager_test.cc
+1 -1
View File
@@ -142,7 +142,7 @@ TEST_F(TaskManagerTest, TestTaskFailure) {
ASSERT_EQ(reference_counter_->NumObjectIDsInScope(), 0);
}
TEST_F(TaskManagerTest, TestTaskRetry) {
TEST_F(TaskManagerTest, TestTaskReconstruction) {
TaskID caller_id = TaskID::Nil();
rpc::Address caller_address;
ObjectID dep1 = ObjectID::FromRandom();
src/ray/core_worker/transport/direct_actor_transport.cc
+9 -10
View File
@@ -23,27 +23,26 @@ using ray::rpc::ActorTableData;
namespace ray {
void CoreWorkerDirectActorTaskSubmitter::KillActor(const ActorID &actor_id,
bool force_kill,
bool no_reconstruction) {
bool force_kill, bool no_restart) {
absl::MutexLock lock(&mu_);
rpc::KillActorRequest request;
request.set_intended_actor_id(actor_id.Binary());
request.set_force_kill(force_kill);
request.set_no_reconstruction(no_reconstruction);
request.set_no_restart(no_restart);
auto inserted = pending_force_kills_.emplace(actor_id, request);
if (!inserted.second && force_kill) {
// Overwrite the previous request to kill the actor if the new request is a
// force kill.
inserted.first->second.set_force_kill(true);
if (no_reconstruction) {
if (no_restart) {
// Overwrite the previous request to disable reconstruction if the new request's
// no_reconstruction flag is set to true.
inserted.first->second.set_no_reconstruction(true);
// no_restart flag is set to true.
inserted.first->second.set_no_restart(true);
}
}
auto it = rpc_clients_.find(actor_id);
if (it == rpc_clients_.end()) {
// Actor is not yet created, or is being reconstructed, cache the request
// Actor is not yet created, or is being restarted, cache the request
// and submit after actor is alive.
// TODO(zhijunfu): it might be possible for a user to specify an invalid
// actor handle (e.g. from unpickling), in that case it might be desirable
@@ -85,7 +84,7 @@ Status CoreWorkerDirectActorTaskSubmitter::SubmitTask(TaskSpecification task_spe
auto it = rpc_clients_.find(actor_id);
if (it == rpc_clients_.end()) {
// Actor is not yet created, or is being reconstructed, cache the request
// Actor is not yet created, or is being restarted, cache the request
// and submit after actor is alive.
// TODO(zhijunfu): it might be possible for a user to specify an invalid
// actor handle (e.g. from unpickling), in that case it might be desirable
@@ -120,7 +119,7 @@ void CoreWorkerDirectActorTaskSubmitter::DisconnectActor(const ActorID &actor_id
bool dead) {
absl::MutexLock lock(&mu_);
if (!dead) {
// We're reconstructing the actor, so erase the client for now. The new client
// We're restarting the actor, so erase the client for now. The new client
// will be inserted once actor reconstruction completes. We don't erase the
// client when the actor is DEAD, so that all further tasks will be failed.
rpc_clients_.erase(actor_id);
@@ -322,7 +321,7 @@ void CoreWorkerDirectTaskReceiver::HandlePushTask(
if (it != scheduling_queue_.end()) {
if (it->second.first.caller_worker_id != caller_worker_id) {
// We received a request with the same caller ID, but from a different worker,
// this indicates the caller (actor) is reconstructed.
// this indicates the caller (actor) is restarted.
if (it->second.first.caller_creation_timestamp_ms < caller_version) {
// The new request has a newer caller version, then remove the old entry
// from scheduling queue since it's invalid now.
src/ray/core_worker/transport/direct_actor_transport.h
+3 -3
View File
@@ -70,9 +70,9 @@ class CoreWorkerDirectActorTaskSubmitter {
/// \param[in] actor_id The actor_id of the actor to kill.
/// \param[in] force_kill Whether to force kill the actor, or let the actor
/// try a clean exit.
/// \param[in] no_reconstruction If set to true, the killed actor will not be
/// reconstructed anymore.
void KillActor(const ActorID &actor_id, bool force_kill, bool no_reconstruction);
/// \param[in] no_restart If set to true, the killed actor will not be
/// restarted anymore.
void KillActor(const ActorID &actor_id, bool force_kill, bool no_restart);
/// Create connection to actor and send all pending tasks.
///
src/ray/gcs/gcs_server/gcs_actor_manager.cc
+19 -10
View File
@@ -13,6 +13,7 @@
// limitations under the License.
#include "gcs_actor_manager.h"
#include <ray/common/ray_config.h>
#include <utility>
@@ -236,7 +237,7 @@ void GcsActorManager::DestroyActor(const ActorID &actor_id) {
rpc::KillActorRequest request;
request.set_intended_actor_id(actor_id.Binary());
request.set_force_kill(true);
request.set_no_reconstruction(true);
request.set_no_restart(true);
RAY_UNUSED(actor_client->KillActor(request, nullptr));
RAY_CHECK(node_it->second.erase(actor->GetWorkerID()));
@@ -311,7 +312,7 @@ void GcsActorManager::OnWorkerDead(const ray::ClientID &node_id,
if (!actor_id.IsNil()) {
RAY_LOG(INFO) << "Worker " << worker_id << " on node " << node_id
<< " failed, reconstructing actor " << actor_id;
<< " failed, restarting actor " << actor_id;
// Reconstruct the actor.
ReconstructActor(actor_id, /*need_reschedule=*/!intentional_exit);
}
@@ -360,17 +361,25 @@ void GcsActorManager::ReconstructActor(const ActorID &actor_id, bool need_resche
auto worker_id = actor->GetWorkerID();
actor->UpdateAddress(rpc::Address());
auto mutable_actor_table_data = actor->GetMutableActorTableData();
// If the need_reschedule is set to false, then set the `remaining_reconstructions` to 0
// If the need_reschedule is set to false, then set the `remaining_restarts` to 0
// so that the actor will never be rescheduled.
auto remaining_reconstructions =
need_reschedule ? mutable_actor_table_data->remaining_reconstructions() : 0;
int64_t max_restarts = mutable_actor_table_data->max_restarts();
uint64_t num_restarts = mutable_actor_table_data->num_restarts();
int64_t remaining_restarts;
if (!need_reschedule) {
remaining_restarts = 0;
} else if (max_restarts == -1) {
remaining_restarts = -1;
} else {
int64_t remaining = max_restarts - num_restarts;
remaining_restarts = std::max(remaining, static_cast<int64_t>(0));
}
RAY_LOG(WARNING) << "Actor is failed " << actor_id << " on worker " << worker_id
<< " at node " << node_id << ", need_reschedule = " << need_reschedule
<< ", remaining_reconstructions = " << remaining_reconstructions;
if (remaining_reconstructions > 0) {
mutable_actor_table_data->set_remaining_reconstructions(--remaining_reconstructions);
mutable_actor_table_data->set_state(rpc::ActorTableData::RECONSTRUCTING);
<< ", remaining_restarts = " << remaining_restarts;
if (remaining_restarts != 0) {
mutable_actor_table_data->set_num_restarts(++num_restarts);
mutable_actor_table_data->set_state(rpc::ActorTableData::RESTARTING);
auto actor_table_data =
std::make_shared<rpc::ActorTableData>(*mutable_actor_table_data);
// The backend storage is reliable in the future, so the status must be ok.
src/ray/gcs/gcs_server/gcs_actor_manager.h
+2 -4
View File
@@ -48,10 +48,8 @@ class GcsActor {
const auto &actor_creation_task_spec = request.task_spec().actor_creation_task_spec();
actor_table_data_.set_actor_id(actor_creation_task_spec.actor_id());
actor_table_data_.set_job_id(request.task_spec().job_id());
actor_table_data_.set_max_reconstructions(
actor_creation_task_spec.max_actor_reconstructions());
actor_table_data_.set_remaining_reconstructions(
actor_creation_task_spec.max_actor_reconstructions());
actor_table_data_.set_max_restarts(actor_creation_task_spec.max_actor_restarts());
actor_table_data_.set_num_restarts(0);
auto dummy_object =
TaskSpecification(request.task_spec()).ActorDummyObject().Binary();
src/ray/gcs/gcs_server/gcs_server.cc
+1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
#include "gcs_server.h"
#include "actor_info_handler_impl.h"
#include "error_info_handler_impl.h"
#include "gcs_actor_manager.h"
src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc
+7 -6
View File
@@ -16,6 +16,7 @@
#include <ray/gcs/test/gcs_test_util.h>
#include <memory>
#include "gtest/gtest.h"
namespace ray {
@@ -227,8 +228,8 @@ TEST_F(GcsActorManagerTest, TestNodeFailure) {
TEST_F(GcsActorManagerTest, TestActorReconstruction) {
auto job_id = JobID::FromInt(1);
auto create_actor_request = Mocker::GenCreateActorRequest(
job_id, /*max_reconstructions=*/1, /*detached=*/false);
auto create_actor_request =
Mocker::GenCreateActorRequest(job_id, /*max_restarts=*/1, /*detached=*/false);
std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
Status status = gcs_actor_manager_->RegisterActor(
create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
@@ -254,7 +255,7 @@ TEST_F(GcsActorManagerTest, TestActorReconstruction) {
// Remove worker and then check that the actor is being restarted.
EXPECT_CALL(*mock_actor_scheduler_, CancelOnNode(node_id));
gcs_actor_manager_->OnNodeDead(node_id);
ASSERT_EQ(actor->GetState(), rpc::ActorTableData::RECONSTRUCTING);
ASSERT_EQ(actor->GetState(), rpc::ActorTableData::RESTARTING);
// Add node and check that the actor is restarted.
gcs_actor_manager_->SchedulePendingActors();
@@ -287,8 +288,8 @@ TEST_F(GcsActorManagerTest, TestActorReconstruction) {
TEST_F(GcsActorManagerTest, TestActorRestartWhenOwnerDead) {
auto job_id = JobID::FromInt(1);
auto create_actor_request = Mocker::GenCreateActorRequest(
job_id, /*max_reconstructions=*/1, /*detached=*/false);
auto create_actor_request =
Mocker::GenCreateActorRequest(job_id, /*max_restarts=*/1, /*detached=*/false);
std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
RAY_CHECK_OK(gcs_actor_manager_->RegisterActor(
create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
@@ -331,7 +332,7 @@ TEST_F(GcsActorManagerTest, TestActorRestartWhenOwnerDead) {
TEST_F(GcsActorManagerTest, TestDetachedActorRestartWhenCreatorDead) {
auto job_id = JobID::FromInt(1);
auto create_actor_request =
Mocker::GenCreateActorRequest(job_id, /*max_reconstructions=*/1, /*detached=*/true);
Mocker::GenCreateActorRequest(job_id, /*max_restarts=*/1, /*detached=*/true);
std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
RAY_CHECK_OK(gcs_actor_manager_->RegisterActor(
create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
src/ray/gcs/pb_util.h
+3 -3
View File
@@ -60,7 +60,7 @@ inline std::shared_ptr<ray::rpc::ErrorTableData> CreateErrorTableData(
/// Helper function to produce actor table data.
inline std::shared_ptr<ray::rpc::ActorTableData> CreateActorTableData(
const TaskSpecification &task_spec, const ray::rpc::Address &address,
ray::rpc::ActorTableData::ActorState state, uint64_t remaining_reconstructions) {
ray::rpc::ActorTableData::ActorState state, uint64_t num_restarts) {
RAY_CHECK(task_spec.IsActorCreationTask());
auto actor_id = task_spec.ActorCreationId();
auto actor_info_ptr = std::make_shared<ray::rpc::ActorTableData>();
@@ -71,10 +71,10 @@ inline std::shared_ptr<ray::rpc::ActorTableData> CreateActorTableData(
actor_info_ptr->set_actor_creation_dummy_object_id(
task_spec.ActorDummyObject().Binary());
actor_info_ptr->set_job_id(task_spec.JobId().Binary());
actor_info_ptr->set_max_reconstructions(task_spec.MaxActorReconstructions());
actor_info_ptr->set_max_restarts(task_spec.MaxActorRestarts());
actor_info_ptr->set_is_detached(task_spec.IsDetachedActor());
// Set the fields that change when the actor is restarted.
actor_info_ptr->set_remaining_reconstructions(remaining_reconstructions);
actor_info_ptr->set_num_restarts(num_restarts);
actor_info_ptr->mutable_address()->CopyFrom(address);
actor_info_ptr->mutable_owner_address()->CopyFrom(
task_spec.GetMessage().caller_address());
src/ray/gcs/redis_accessor.cc
+3 -4
View File
@@ -99,12 +99,11 @@ Status RedisLogBasedActorInfoAccessor::AsyncUpdate(
const ActorID &actor_id, const std::shared_ptr<ActorTableData> &data_ptr,
const StatusCallback &callback) {
// The actor log starts with an ALIVE entry. This is followed by 0 to N pairs
// of (RECONSTRUCTING, ALIVE) entries, where N is the maximum number of
// of (RESTARTING, ALIVE) entries, where N is the maximum number of
// reconstructions. This is followed optionally by a DEAD entry.
int log_length =
2 * (data_ptr->max_reconstructions() - data_ptr->remaining_reconstructions());
int log_length = 2 * (data_ptr->num_restarts());
if (data_ptr->state() != ActorTableData::ALIVE) {
// RECONSTRUCTING or DEAD entries have an odd index.
// RESTARTING or DEAD entries have an odd index.
log_length += 1;
}
RAY_LOG(DEBUG) << "AsyncUpdate actor state to " << data_ptr->state()
src/ray/gcs/store_client/test/store_client_test_base.h
+3 -2
View File
@@ -19,6 +19,7 @@
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include "ray/common/id.h"
#include "ray/common/test_util.h"
#include "ray/gcs/store_client/store_client.h"
@@ -221,8 +222,8 @@ class StoreClientTestBase : public ::testing::Test {
void GenTestData() {
for (size_t i = 0; i < key_count_; i++) {
rpc::ActorTableData actor;
actor.set_max_reconstructions(1);
actor.set_remaining_reconstructions(1);
actor.set_max_restarts(1);
actor.set_num_restarts(0);
JobID job_id = JobID::FromInt(i % index_count_);
actor.set_job_id(job_id.Binary());
actor.set_state(rpc::ActorTableData::ALIVE);
src/ray/gcs/tables.h
+3 -4
View File
@@ -23,12 +23,11 @@
#include "ray/common/constants.h"
#include "ray/common/id.h"
#include "ray/common/status.h"
#include "ray/util/logging.h"
#include "ray/gcs/callback.h"
#include "ray/gcs/entry_change_notification.h"
#include "ray/gcs/redis_context.h"
#include "ray/protobuf/gcs.pb.h"
#include "ray/util/logging.h"
struct redisAsyncContext;
@@ -717,8 +716,8 @@ class JobTable : public Log<JobID, JobTableData> {
};
/// Log-based Actor table starts with an ALIVE entry, which represents the first time the
/// actor is created. This may be followed by 0 or more pairs of RECONSTRUCTING, ALIVE
/// entries, which represent each time the actor fails (RECONSTRUCTING) and gets recreated
/// actor is created. This may be followed by 0 or more pairs of RESTARTING, ALIVE
/// entries, which represent each time the actor fails (RESTARTING) and gets recreated
/// (ALIVE). These may be followed by a DEAD entry, which means that the actor has failed
/// and will not be reconstructed.
class LogBasedActorTable : public Log<ActorID, ActorTableData> {
src/ray/gcs/test/gcs_test_util.h
+9 -12
View File
@@ -17,21 +17,19 @@
#include <memory>
#include <utility>
#include "gmock/gmock.h"
#include "gmock/gmock.h"
#include "src/ray/common/task/task.h"
#include "src/ray/common/task/task_util.h"
#include "src/ray/common/test_util.h"
#include "src/ray/util/asio_util.h"
#include "src/ray/protobuf/gcs_service.grpc.pb.h"
#include "src/ray/util/asio_util.h"
namespace ray {
struct Mocker {
static TaskSpecification GenActorCreationTask(const JobID &job_id,
int max_reconstructions, bool detached,
const std::string &name,
static TaskSpecification GenActorCreationTask(const JobID &job_id, int max_restarts,
bool detached, const std::string &name,
const rpc::Address &owner_address) {
TaskSpecBuilder builder;
rpc::Address empty_address;
@@ -41,13 +39,12 @@ struct Mocker {
auto task_id = TaskID::ForActorCreationTask(actor_id);
builder.SetCommonTaskSpec(task_id, Language::PYTHON, empty_descriptor, job_id,
TaskID::Nil(), 0, TaskID::Nil(), owner_address, 1, {}, {});
builder.SetActorCreationTaskSpec(actor_id, max_reconstructions, {}, 1, detached,
name);
builder.SetActorCreationTaskSpec(actor_id, max_restarts, {}, 1, detached, name);
return builder.Build();
}
static rpc::CreateActorRequest GenCreateActorRequest(const JobID &job_id,
int max_reconstructions = 0,
int max_restarts = 0,
bool detached = false,
const std::string name = "") {
rpc::CreateActorRequest request;
@@ -59,7 +56,7 @@ struct Mocker {
owner_address.set_worker_id(WorkerID::FromRandom().Binary());
}
auto actor_creation_task_spec =
GenActorCreationTask(job_id, max_reconstructions, detached, name, owner_address);
GenActorCreationTask(job_id, max_restarts, detached, name, owner_address);
request.mutable_task_spec()->CopyFrom(actor_creation_task_spec.GetMessage());
return request;
}
@@ -89,8 +86,8 @@ struct Mocker {
actor_table_data->set_job_id(job_id.Binary());
actor_table_data->set_state(
rpc::ActorTableData_ActorState::ActorTableData_ActorState_ALIVE);
actor_table_data->set_max_reconstructions(1);
actor_table_data->set_remaining_reconstructions(1);
actor_table_data->set_max_restarts(1);
actor_table_data->set_num_restarts(0);
return actor_table_data;
}
src/ray/gcs/test/redis_actor_info_accessor_test.cc
+2 -2
View File
@@ -32,8 +32,8 @@ class ActorInfoAccessorTest : public AccessorTestBase<ActorID, ActorTableData> {
virtual void GenTestData() {
for (size_t i = 0; i < 100; ++i) {
std::shared_ptr<ActorTableData> actor = std::make_shared<ActorTableData>();
actor->set_max_reconstructions(1);
actor->set_remaining_reconstructions(1);
actor->set_max_restarts(1);
actor->set_num_restarts(0);
JobID job_id = JobID::FromInt(i);
actor->set_job_id(job_id.Binary());
actor->set_state(ActorTableData::ALIVE);
src/ray/gcs/test/subscription_executor_test.cc
+3 -2
View File
@@ -13,6 +13,7 @@
// limitations under the License.
#include "ray/gcs/subscription_executor.h"
#include "gtest/gtest.h"
#include "ray/gcs/callback.h"
#include "ray/gcs/entry_change_notification.h"
@@ -61,8 +62,8 @@ class SubscriptionExecutorTest : public AccessorTestBase<ActorID, ActorTableData
virtual void GenTestData() {
for (size_t i = 0; i < 100; ++i) {
std::shared_ptr<ActorTableData> actor = std::make_shared<ActorTableData>();
actor->set_max_reconstructions(1);
actor->set_remaining_reconstructions(1);
actor->set_max_restarts(1);
actor->set_num_restarts(0);
JobID job_id = JobID::FromInt(i);
actor->set_job_id(job_id.Binary());
actor->set_state(ActorTableData::ALIVE);
src/ray/protobuf/common.proto
+3 -2
View File
@@ -148,8 +148,9 @@ message ActorCreationTaskSpec {
// ID of the actor that will be created by this task.
bytes actor_id = 2;
// The max number of times this actor should be recontructed.
// If this number of 0 or negative, the actor won't be reconstructed on failure.
uint64 max_actor_reconstructions = 3;
// If this number is 0 the actor won't be restarted.
// If this number is -1 the actor will be restarted indefinitely.
int64 max_actor_restarts = 3;
// The dynamic options used in the worker command when starting a worker process for
// an actor creation task. If the list isn't empty, the options will be used to replace
// the placeholder strings (`RAY_WORKER_DYNAMIC_OPTION_PLACEHOLDER_0`,
src/ray/protobuf/core_worker.proto
+5 -6
View File
@@ -107,12 +107,11 @@ message PushTaskRequest {
// Resource mapping ids assigned to the worker executing the task.
repeated ResourceMapEntry resource_mapping = 6;
// The version of the caller. This is used to distinguish on-the-fly
// requests from a caller before it die, and requests from the reconstructed
// requests from a caller before it die, and requests from the restarted
// caller, which might happen theoretically when network has issues.
// - For an actor, this is set to the timestamp when the actor is created,
// so it can be used to differentiate which is the new reconstructed actor.
// - For a non-actor task, it's set to the timestamp the task starts
// execution.
// so it can be used to differentiate which is the newly restarted actor.
// - For a non-actor task, it's set to the timestamp the task starts execution.
int64 caller_version = 7;
}
@@ -188,8 +187,8 @@ message KillActorRequest {
bytes intended_actor_id = 1;
// Whether to force kill the actor.
bool force_kill = 2;
// If set to true, the killed actor will not be reconstructed anymore.
bool no_reconstruction = 3;
// If set to true, the killed actor will not be restarted anymore.
bool no_restart = 3;
}
message KillActorReply {
src/ray/protobuf/gcs.proto
+10 -9
View File
@@ -103,10 +103,10 @@ message ActorTableData {
PENDING = 0;
// Actor is alive.
ALIVE = 1;
// Actor is dead, now being reconstructed.
// Actor is dead, now being restarted.
// After reconstruction finishes, the state will become alive again.
RECONSTRUCTING = 2;
// Actor is already dead and won't be reconstructed.
RESTARTING = 2;
// Actor is already dead and won't be restarted.
DEAD = 3;
}
// The ID of the actor that was created.
@@ -114,17 +114,18 @@ message ActorTableData {
// The ID of the caller of the actor creation task.
bytes parent_id = 2;
// The dummy object ID returned by the actor creation task. If the actor
// dies, then this is the object that should be reconstructed for the actor
// dies, then this is the object that should be restarted for the actor
// to be recreated.
bytes actor_creation_dummy_object_id = 3;
// The ID of the job that created the actor.
bytes job_id = 4;
// Current state of this actor.
ActorState state = 6;
// Max number of times this actor should be reconstructed.
uint64 max_reconstructions = 7;
// Remaining number of reconstructions.
uint64 remaining_reconstructions = 8;
// Max number of times this actor should be restarted,
// a value of -1 indicates an infinite number of reconstruction attempts.
int64 max_restarts = 7;
// Number of restarts that have already been performed on this actor.
uint64 num_restarts = 8;
// The address of the the actor.
Address address = 9;
// The address of the the actor's owner (parent).
@@ -344,7 +345,7 @@ enum ErrorType {
WORKER_DIED = 0;
// Indicates that a task failed because the actor died unexpectedly before finishing it.
ACTOR_DIED = 1;
// Indicates that an object is lost and cannot be reconstructed.
// Indicates that an object is lost and cannot be restarted.
// Note, this currently only happens to actor objects. When the actor's state is already
// after the object's creating task, the actor cannot re-run the task.
// TODO(hchen): we may want to reuse this error type for more cases. E.g.,
src/ray/raylet/actor_registration.cc
+11 -4
View File
@@ -65,12 +65,19 @@ const JobID ActorRegistration::GetJobId() const {
return JobID::FromBinary(actor_table_data_.job_id());
}
const int64_t ActorRegistration::GetMaxReconstructions() const {
return actor_table_data_.max_reconstructions();
const int64_t ActorRegistration::GetMaxRestarts() const {
return actor_table_data_.max_restarts();
}
const int64_t ActorRegistration::GetRemainingReconstructions() const {
return actor_table_data_.remaining_reconstructions();
const int64_t ActorRegistration::GetRemainingRestarts() const {
if (actor_table_data_.max_restarts() == -1) {
return -1;
}
return actor_table_data_.max_restarts() - actor_table_data_.num_restarts();
}
const uint64_t ActorRegistration::GetNumRestarts() const {
return actor_table_data_.num_restarts();
}
const std::unordered_map<TaskID, ActorRegistration::FrontierLeaf>
src/ray/raylet/actor_registration.h
+8 -5
View File
@@ -82,7 +82,7 @@ class ActorRegistration {
/// Get the object that represents the actor's initial state. This is the
/// execution dependency returned by this actor's creation task. If
/// reconstructed, this will recreate the actor.
/// restarted, this will recreate the actor.
///
/// \return The execution dependency returned by the actor's creation task.
const ObjectID GetActorCreationDependency() const;
@@ -90,11 +90,14 @@ class ActorRegistration {
/// Get actor's job ID.
const JobID GetJobId() const;
/// Get the max number of times this actor should be reconstructed.
const int64_t GetMaxReconstructions() const;
/// Get the max number of times this actor should be restarted.
const int64_t GetMaxRestarts() const;
/// Get the remaining number of times this actor should be reconstructed.
const int64_t GetRemainingReconstructions() const;
/// Get the remaining number of times this actor should be restarted.
const int64_t GetRemainingRestarts() const;
/// Get the number of times this actor has already been restarted
const uint64_t GetNumRestarts() const;
/// Get the object that represents the actor's current state. This is the
/// execution dependency returned by the task most recently executed on the
src/ray/raylet/node_manager.cc
+40 -46
View File
@@ -55,7 +55,7 @@ int64_t GetExpectedTaskCounter(
struct ActorStats {
int live_actors = 0;
int dead_actors = 0;
int reconstructing_actors = 0;
int restarting_actors = 0;
int max_num_handles = 0;
};
@@ -66,8 +66,8 @@ ActorStats GetActorStatisticalData(
for (auto &pair : actor_registry) {
if (pair.second.GetState() == ray::rpc::ActorTableData::ALIVE) {
item.live_actors += 1;
} else if (pair.second.GetState() == ray::rpc::ActorTableData::RECONSTRUCTING) {
item.reconstructing_actors += 1;
} else if (pair.second.GetState() == ray::rpc::ActorTableData::RESTARTING) {
item.restarting_actors += 1;
} else {
item.dead_actors += 1;
}
@@ -588,7 +588,7 @@ void NodeManager::NodeRemoved(const GcsNodeInfo &node_info) {
actor_entry.second.GetState() == ActorTableData::ALIVE) {
RAY_LOG(INFO) << "Actor " << actor_entry.first
<< " is disconnected, because its node " << node_id
<< " is removed from cluster. It may be reconstructed.";
<< " is removed from cluster. It may be restarted.";
HandleDisconnectedActor(actor_entry.first, /*was_local=*/false,
/*intentional_disconnect=*/false);
}
@@ -815,13 +815,11 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id,
} else {
// Only process the state transition if it is to a later state than ours.
if (actor_registration.GetState() > it->second.GetState() &&
actor_registration.GetRemainingReconstructions() ==
it->second.GetRemainingReconstructions()) {
actor_registration.GetNumRestarts() == it->second.GetNumRestarts()) {
// The new state is later than ours if it is about the same lifetime, but
// a greater state.
it->second = actor_registration;
} else if (actor_registration.GetRemainingReconstructions() <
it->second.GetRemainingReconstructions()) {
} else if (actor_registration.GetNumRestarts() > it->second.GetNumRestarts()) {
// The new state is also later than ours it is about a later lifetime of
// the actor.
it->second = actor_registration;
@@ -835,11 +833,11 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id,
<< ", node_manager_id = " << actor_registration.GetNodeManagerId()
<< ", state = "
<< ActorTableData::ActorState_Name(actor_registration.GetState())
<< ", remaining_reconstructions = "
<< actor_registration.GetRemainingReconstructions();
<< ", remaining_restarts = "
<< actor_registration.GetRemainingRestarts();
if (actor_registration.GetState() == ActorTableData::ALIVE) {
// The actor is now alive (created for the first time or reconstructed). We can
// The actor is now alive (created for the first time or restarted). We can
// stop listening for the actor creation task. This is needed because we use
// `ListenAndMaybeReconstruct` to reconstruct the actor.
reconstruction_policy_.Cancel(actor_registration.GetActorCreationDependency());
@@ -877,8 +875,8 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id,
for (auto const &task : removed_tasks) {
TreatTaskAsFailed(task, ErrorType::ACTOR_DIED);
}
} else if (actor_registration.GetState() == ActorTableData::RECONSTRUCTING) {
RAY_LOG(DEBUG) << "Actor is being reconstructed: " << actor_id;
} else if (actor_registration.GetState() == ActorTableData::RESTARTING) {
RAY_LOG(DEBUG) << "Actor is being restarted: " << actor_id;
if (!(RayConfig::instance().gcs_service_enabled() &&
RayConfig::instance().gcs_actor_service_enabled())) {
// The actor is dead and needs reconstruction. Attempting to reconstruct its
@@ -887,7 +885,7 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id,
actor_registration.GetActorCreationDependency());
}
// When an actor fails but can be reconstructed, resubmit all of the queued
// When an actor fails but can be restarted, resubmit all of the queued
// tasks for that actor. This will mark the tasks as waiting for actor
// creation.
auto tasks_to_remove = local_queues_.GetTaskIdsForActor(actor_id);
@@ -1150,15 +1148,15 @@ void NodeManager::HandleDisconnectedActor(const ActorID &actor_id, bool was_loca
auto actor_entry = actor_registry_.find(actor_id);
RAY_CHECK(actor_entry != actor_registry_.end());
auto &actor_registration = actor_entry->second;
auto remainingRestarts = actor_registration.GetRemainingRestarts();
RAY_LOG(DEBUG) << "The actor with ID " << actor_id << " died "
<< (intentional_disconnect ? "intentionally" : "unintentionally")
<< ", remaining reconstructions = "
<< actor_registration.GetRemainingReconstructions();
<< ", remaining restarts = " << remainingRestarts;
// Check if this actor needs to be reconstructed.
// Check if this actor needs to be restarted.
ActorState new_state =
actor_registration.GetRemainingReconstructions() > 0 && !intentional_disconnect
? ActorTableData::RECONSTRUCTING
(remainingRestarts == -1 || remainingRestarts > 0) && !intentional_disconnect
? ActorTableData::RESTARTING
: ActorTableData::DEAD;
if (was_local) {
// Clean up the dummy objects from this actor.
@@ -1189,7 +1187,7 @@ 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) {
if (was_local && new_state == ActorTableData::RESTARTING) {
RAY_LOG(INFO) << "A local actor (id = " << actor_id
<< " ) is dead, reconstructing it.";
const ObjectID &actor_creation_dummy_object_id =
@@ -1385,7 +1383,7 @@ void NodeManager::ProcessFetchOrReconstructMessage(
} else {
// If reconstruction is also required, then add any requested objects to
// the list to subscribe to in the task dependency manager. These objects
// will be pulled from remote node managers and reconstructed if
// will be pulled from remote node managers and restarted if
// necessary.
required_object_ids.push_back(object_id);
}
@@ -1412,7 +1410,7 @@ void NodeManager::ProcessWaitRequestMessage(
if (!task_dependency_manager_.CheckObjectLocal(object_id)) {
// Add any missing objects to the list to subscribe to in the task
// dependency manager. These objects will be pulled from remote node
// managers and reconstructed if necessary.
// managers and restarted if necessary.
required_object_ids.push_back(object_id);
}
}
@@ -1463,7 +1461,7 @@ void NodeManager::ProcessWaitForDirectActorCallArgsRequestMessage(
if (!task_dependency_manager_.CheckObjectLocal(object_id)) {
// Add any missing objects to the list to subscribe to in the task
// dependency manager. These objects will be pulled from remote node
// managers and reconstructed if necessary.
// managers and restarted if necessary.
required_object_ids.push_back(object_id);
}
}
@@ -2142,7 +2140,7 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag
if (local_queues_.HasTask(task_id)) {
RAY_LOG(WARNING) << "Submitted task " << task_id
<< " is already queued and will not be reconstructed. This is most "
<< " is already queued and will not be restarted. This is most "
"likely due to spurious reconstruction.";
return;
}
@@ -2151,10 +2149,10 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag
// Check whether we know the location of the actor.
const auto actor_entry = actor_registry_.find(spec.ActorId());
bool seen = actor_entry != actor_registry_.end();
// If we have already seen this actor and this actor is not being reconstructed,
// If we have already seen this actor and this actor is not being restarted,
// its location is known.
bool location_known =
seen && actor_entry->second.GetState() != ActorTableData::RECONSTRUCTING;
seen && actor_entry->second.GetState() != ActorTableData::RESTARTING;
if (location_known) {
if (actor_entry->second.GetState() == ActorTableData::DEAD) {
// If this actor is dead, either because the actor process is dead
@@ -2369,7 +2367,7 @@ void NodeManager::AsyncResolveObjects(const std::shared_ptr<ClientConnection> &c
}
// Subscribe to the objects required by the task. These objects will be
// fetched and/or reconstructed as necessary, until the objects become local
// fetched and/or restarted as necessary, until the objects become local
// or are unsubscribed.
if (ray_get) {
// TODO(ekl) using the assigned task id is a hack to handle unsubscription for
@@ -2617,42 +2615,38 @@ std::shared_ptr<ActorTableData> NodeManager::CreateActorTableDataFromCreationTas
auto actor_id = task_spec.ActorCreationId();
auto actor_entry = actor_registry_.find(actor_id);
std::shared_ptr<ActorTableData> actor_info_ptr;
// TODO(swang): If this is an actor that was reconstructed, and previous
// TODO(swang): If this is an actor that was restarted, and previous
// actor notifications were delayed, then this node may not have an entry for
// the actor in actor_regisry_. Then, the fields for the number of
// reconstructions will be wrong.
// restarts will be wrong.
if (actor_entry == actor_registry_.end()) {
actor_info_ptr.reset(new ActorTableData());
// Set all of the static fields for the actor. These fields will not
// change even if the actor fails or is reconstructed.
// change even if the actor fails or is restarted.
actor_info_ptr->set_actor_id(actor_id.Binary());
actor_info_ptr->set_actor_creation_dummy_object_id(
task_spec.ActorDummyObject().Binary());
actor_info_ptr->set_job_id(task_spec.JobId().Binary());
actor_info_ptr->set_max_reconstructions(task_spec.MaxActorReconstructions());
// This is the first time that the actor has been created, so the number
// of remaining reconstructions is the max.
actor_info_ptr->set_remaining_reconstructions(task_spec.MaxActorReconstructions());
actor_info_ptr->set_max_restarts(task_spec.MaxActorRestarts());
actor_info_ptr->set_num_restarts(0);
actor_info_ptr->set_is_detached(task_spec.IsDetachedActor());
actor_info_ptr->mutable_owner_address()->CopyFrom(
task_spec.GetMessage().caller_address());
} else {
// If we've already seen this actor, it means that this actor was reconstructed.
// Thus, its previous state must be RECONSTRUCTING.
// If we've already seen this actor, it means that this actor was restarted.
// Thus, its previous state must be RESTARTING.
// TODO: The following is a workaround for the issue described in
// https://github.com/ray-project/ray/issues/5524, please see the issue
// description for more information.
if (actor_entry->second.GetState() != ActorTableData::RECONSTRUCTING) {
RAY_LOG(WARNING) << "Actor not in reconstructing state, most likely it "
if (actor_entry->second.GetState() != ActorTableData::RESTARTING) {
RAY_LOG(WARNING) << "Actor not in restarting state, most likely it "
<< "died before creation handler could run. Actor state is "
<< actor_entry->second.GetState();
}
// Copy the static fields from the current actor entry.
actor_info_ptr.reset(new ActorTableData(actor_entry->second.GetTableData()));
// We are reconstructing the actor, so subtract its
// remaining_reconstructions by 1.
actor_info_ptr->set_remaining_reconstructions(
actor_info_ptr->remaining_reconstructions() - 1);
// We are restarting the actor, so increment its num_restarts
actor_info_ptr->set_num_restarts(actor_info_ptr->num_restarts() + 1);
}
// Set the new fields for the actor's state to indicate that the actor is
@@ -2768,7 +2762,7 @@ void NodeManager::FinishAssignedActorTask(Worker &worker, const Task &task) {
// NOTE(swang): The dummy objects must be marked as local whenever
// ExtendFrontier is called, and vice versa, so that we can clean up the
// dummy objects properly in case the actor fails and needs to be
// reconstructed.
// restarted.
HandleObjectLocal(task_spec.ActorDummyObject());
}
}
@@ -3324,7 +3318,7 @@ std::string NodeManager::DebugString() const {
auto statistical_data = GetActorStatisticalData(actor_registry_);
result << "\n- num live actors: " << statistical_data.live_actors;
result << "\n- num reconstructing actors: " << statistical_data.reconstructing_actors;
result << "\n- num restarting actors: " << statistical_data.restarting_actors;
result << "\n- num dead actors: " << statistical_data.dead_actors;
result << "\n- max num handles: " << statistical_data.max_num_handles;
@@ -3725,8 +3719,8 @@ void NodeManager::RecordMetrics() {
auto statistical_data = GetActorStatisticalData(actor_registry_);
stats::ActorStats().Record(statistical_data.live_actors,
{{stats::ValueTypeKey, "live_actors"}});
stats::ActorStats().Record(statistical_data.reconstructing_actors,
{{stats::ValueTypeKey, "reconstructing_actors"}});
stats::ActorStats().Record(statistical_data.restarting_actors,
{{stats::ValueTypeKey, "restarting_actors"}});
stats::ActorStats().Record(statistical_data.dead_actors,
{{stats::ValueTypeKey, "dead_actors"}});
stats::ActorStats().Record(statistical_data.max_num_handles,
src/ray/raylet/node_manager.h
+1 -1
View File
@@ -533,7 +533,7 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
const uint8_t *message_data);
/// Handle the case where an actor is disconnected, determine whether this
/// actor needs to be reconstructed and then update actor table.
/// actor needs to be restarted and then update actor table.
/// This function needs to be called either when actor process dies or when
/// a node dies.
///
src/ray/raylet/task_dependency_manager.cc
+3 -4
View File
@@ -15,7 +15,6 @@
#include "task_dependency_manager.h"
#include "absl/time/clock.h"
#include "ray/stats/stats.h"
namespace ray {
@@ -331,8 +330,8 @@ void TaskDependencyManager::TaskPending(const Task &task) {
// thus it doesn't need task lease. And actually if we
// acquire a lease in this case and forget to cancel it,
// the lease would never expire which will prevent the
// actor from being reconstructed;
// - When a direct actor is reconstructed, raylet resubmits
// actor from being restarted;
// - When a direct actor is restarted, raylet resubmits
// the task, and the task can be forwarded to another raylet,
// and eventually assigned to a worker. In this case we need
// the task lease to make sure there's only one raylet can
@@ -347,7 +346,7 @@ void TaskDependencyManager::TaskPending(const Task &task) {
// - when it's resubmitted by raylet because of reconstruction,
// `OnDispatch` will not be overriden and thus is nullptr.
if (task.GetTaskSpecification().IsActorCreationTask() && task.OnDispatch() == nullptr) {
// This is an actor creation task, and it's being reconstructed,
// This is an actor creation task, and it's being restarted,
// in this case we still need the task lease. Note that we don't
// require task lease for direct actor creation task.
} else {