hiro/ray
1
0
Fork 0
mirror of https://github.com/vale981/ray synced 2025-03-06 02:21:39 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

[Core] Fix reference counting bug on objects borrowed for a cancelled actor creation. (#27298) (#27595)

Browse source 
This PR fixes a reference counting bug for borrowed objects sent to an actor creation task that is then cancelled.

Before this PR, when actor creation is cancelled before the creation task has been scheduled, the GCS-based actor manager would would destroy the actor without replying to the task submission RPC from the actor creating worker, resulting in the reference counts on that worker to never get cleaned up. This caused us to leak borrowed objects when such cancellation-before-scheduling happened for actors.

This PR fixes this by ensuring that the task submission RPC receives a reply indicating that the actor creation task has been cancelled, at which point the submitting worker will run through the same reference counting cleanup as is done for normal task cancellation.
This commit is contained in:
Clark Zinzow 2022-08-05 23:15:40 -06:00 • committed by GitHub
parent e5a588fe3a
commit e43278a010
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 177 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
python/ray/exceptions.py
View file

@ -293,6 +293,10 @@ class RayActorError(RayError):
if cause.node_ip_address != "":
error_msg_lines.append(f"\tip: {cause.node_ip_address}")
error_msg_lines.append(cause.error_message)
if cause.never_started:
error_msg_lines.append(
"The actor never ran - it was cancelled before it started running."
)
self.error_msg = "\n".join(error_msg_lines)
@property

42
python/ray/tests/test_reference_counting.py
View file

@ -554,6 +554,48 @@ def test_remove_actor_immediately_after_creation(ray_start_regular):
wait_for_condition(_all_actors_dead, timeout=10)
# Test that a reference borrowed by an actor constructor is freed if the actor is
# cancelled before being scheduled.
def test_actor_constructor_borrow_cancellation(ray_start_regular):
# Schedule the actor with a non-existent resource so it's guaranteed to never be
# scheduled.
@ray.remote(resources={"nonexistent_resource": 1})
class Actor:
def __init__(self, obj_containing_ref):
raise ValueError(
"The actor constructor should not be reached; the actor creation task "
"should be cancelled before the actor is scheduled."
)
def should_not_be_run(self):
raise ValueError("This method should never be reached.")
# Test with implicit cancellation by letting the actor handle go out-of-scope.
def test_implicit_cancel():
ref = ray.put(1)
Actor.remote({"foo": ref})
test_implicit_cancel()
# Confirm that the ref object is not leaked.
check_refcounts({})
# Test with explicit cancellation via ray.kill().
ref = ray.put(1)
a = Actor.remote({"foo": ref})
ray.kill(a)
del ref
# Confirm that the ref object is not leaked.
check_refcounts({})
# Check that actor death cause is propagated.
with pytest.raises(
ray.exceptions.RayActorError, match="it was killed by `ray.kill"
) as exc_info:
ray.get(a.should_not_be_run.remote())
print(exc_info._excinfo[1])
if __name__ == "__main__":
import sys

3
src/ray/common/status.cc
View file

@ -53,6 +53,7 @@ namespace ray {
#define STATUS_CODE_UNKNOWN "Unknown"
#define STATUS_CODE_NOT_FOUND "NotFound"
#define STATUS_CODE_DISCONNECTED "Disconnected"
#define STATUS_CODE_SCHEDULING_CANCELLED "SchedulingCancelled"
// object store status
#define STATUS_CODE_OBJECT_EXISTS "ObjectExists"
#define STATUS_CODE_OBJECT_NOT_FOUND "ObjectNotFound"
@ -102,6 +103,7 @@ std::string Status::CodeAsString() const {
{StatusCode::CreationTaskError, STATUS_CODE_CREATION_TASK_ERROR},
{StatusCode::NotFound, STATUS_CODE_NOT_FOUND},
{StatusCode::Disconnected, STATUS_CODE_DISCONNECTED},
{StatusCode::SchedulingCancelled, STATUS_CODE_SCHEDULING_CANCELLED},
{StatusCode::ObjectExists, STATUS_CODE_OBJECT_EXISTS},
{StatusCode::ObjectNotFound, STATUS_CODE_OBJECT_NOT_FOUND},
{StatusCode::ObjectAlreadySealed, STATUS_CODE_OBJECT_STORE_ALREADY_SEALED},
@ -137,6 +139,7 @@ StatusCode Status::StringToCode(const std::string &str) {
{STATUS_CODE_CREATION_TASK_ERROR, StatusCode::CreationTaskError},
{STATUS_CODE_NOT_FOUND, StatusCode::NotFound},
{STATUS_CODE_DISCONNECTED, StatusCode::Disconnected},
{STATUS_CODE_SCHEDULING_CANCELLED, StatusCode::SchedulingCancelled},
{STATUS_CODE_OBJECT_EXISTS, StatusCode::ObjectExists},
{STATUS_CODE_OBJECT_NOT_FOUND, StatusCode::ObjectNotFound},
{STATUS_CODE_OBJECT_STORE_ALREADY_SEALED, StatusCode::ObjectAlreadySealed},

6
src/ray/common/status.h
View file

@ -95,6 +95,7 @@ enum class StatusCode : char {
CreationTaskError = 16,
NotFound = 17,
Disconnected = 18,
SchedulingCancelled = 19,
// object store status
ObjectExists = 21,
ObjectNotFound = 22,
@ -194,6 +195,10 @@ class RAY_EXPORT Status {
return Status(StatusCode::Disconnected, msg);
}
static Status SchedulingCancelled(const std::string &msg) {
return Status(StatusCode::SchedulingCancelled, msg);
}
static Status ObjectExists(const std::string &msg) {
return Status(StatusCode::ObjectExists, msg);
}
@ -256,6 +261,7 @@ class RAY_EXPORT Status {
}
bool IsNotFound() const { return code() == StatusCode::NotFound; }
bool IsDisconnected() const { return code() == StatusCode::Disconnected; }
bool IsSchedulingCancelled() const { return code() == StatusCode::SchedulingCancelled; }
bool IsObjectExists() const { return code() == StatusCode::ObjectExists; }
bool IsObjectNotFound() const { return code() == StatusCode::ObjectNotFound; }
bool IsObjectAlreadySealed() const { return code() == StatusCode::ObjectAlreadySealed; }

19
src/ray/core_worker/transport/direct_task_transport.cc
View file

@ -51,10 +51,23 @@ Status CoreWorkerDirectTaskSubmitter::SubmitTask(TaskSpecification task_spec) {
task_finisher_->CompletePendingTask(
task_id, push_task_reply, reply.actor_address());
} else {
RAY_LOG(INFO) << "Failed to create actor " << actor_id
<< " with status: " << status.ToString();
rpc::RayErrorInfo ray_error_info;
if (status.IsSchedulingCancelled()) {
RAY_LOG(DEBUG) << "Actor creation cancelled, actor id = " << actor_id;
task_finisher_->MarkTaskCanceled(task_id);
if (reply.has_death_cause()) {
ray_error_info.mutable_actor_died_error()->CopyFrom(
reply.death_cause());
}
} else {
RAY_LOG(INFO) << "Failed to create actor " << actor_id
<< " with status: " << status.ToString();
}
RAY_UNUSED(task_finisher_->FailOrRetryPendingTask(
task_id, rpc::ErrorType::ACTOR_CREATION_FAILED, &status));
task_id,
rpc::ErrorType::ACTOR_CREATION_FAILED,
&status,
ray_error_info.has_actor_died_error() ? &ray_error_info : nullptr));
}
}));
return;

59
src/ray/gcs/gcs_server/gcs_actor_manager.cc
View file

@ -39,6 +39,10 @@ void AddActorInfo(const ray::gcs::GcsActor *actor,
mutable_actor_died_error_ctx->set_ray_namespace(actor->GetRayNamespace());
mutable_actor_died_error_ctx->set_class_name(actor->GetActorTableData().class_name());
mutable_actor_died_error_ctx->set_actor_id(actor->GetActorID().Binary());
const auto actor_state = actor->GetState();
mutable_actor_died_error_ctx->set_never_started(
actor_state == ray::rpc::ActorTableData::DEPENDENCIES_UNREADY ||
actor_state == ray::rpc::ActorTableData::PENDING_CREATION);
}
const ray::rpc::ActorDeathCause GenNodeDiedCause(const ray::gcs::GcsActor *actor,
@ -258,12 +262,24 @@ void GcsActorManager::HandleCreateActor(const rpc::CreateActorRequest &request,
Status status = CreateActor(
request,
[reply, send_reply_callback, actor_id](const std::shared_ptr<gcs::GcsActor> &actor,
const rpc::PushTaskReply &task_reply) {
RAY_LOG(INFO) << "Finished creating actor, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id;
reply->mutable_actor_address()->CopyFrom(actor->GetAddress());
reply->mutable_borrowed_refs()->CopyFrom(task_reply.borrowed_refs());
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
const rpc::PushTaskReply &task_reply,
bool creation_cancelled) {
if (creation_cancelled) {
// Actor creation is cancelled.
RAY_LOG(INFO) << "Actor creation was cancelled, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id;
reply->mutable_death_cause()->CopyFrom(
actor->GetActorTableData().death_cause());
GCS_RPC_SEND_REPLY(send_reply_callback,
reply,
Status::SchedulingCancelled("Actor creation cancelled."));
} else {
RAY_LOG(INFO) << "Finished creating actor, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id;
reply->mutable_actor_address()->CopyFrom(actor->GetAddress());
reply->mutable_borrowed_refs()->CopyFrom(task_reply.borrowed_refs());
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
}
});
if (!status.ok()) {
RAY_LOG(WARNING) << "Failed to create actor, job id = " << actor_id.JobId()
@ -578,7 +594,7 @@ Status GcsActorManager::CreateActor(const ray::rpc::CreateActorRequest &request,
// requests to GCS server.
// In this case, we can just reply.
// TODO(swang): Need to pass ref count info.
callback(iter->second, rpc::PushTaskReply());
callback(iter->second, rpc::PushTaskReply(), false);
return Status::OK();
}
@ -590,7 +606,8 @@ Status GcsActorManager::CreateActor(const ray::rpc::CreateActorRequest &request,
return Status::OK();
}
// Mark the callback as pending and invoke it after the actor has been successfully
// created.
// created or if the creation has been cancelled (e.g. via ray.kill() or the actor
// handle going out-of-scope).
actor_to_create_callbacks_[actor_id].emplace_back(std::move(callback));
// If GCS restarts while processing `CreateActor` request, GCS client will resend the
@ -704,7 +721,7 @@ void GcsActorManager::PollOwnerForActorOutOfScope(
<< actor_id.JobId();
} else {
RAY_LOG(INFO) << "Actor " << actor_id
<< " is out of scope, destroying actor child, job id = "
<< " is out of scope, destroying actor, job id = "
<< actor_id.JobId();
}
@ -724,7 +741,13 @@ void GcsActorManager::DestroyActor(const ActorID &actor_id,
RAY_LOG(INFO) << "Destroying actor, actor id = " << actor_id
<< ", job id = " << actor_id.JobId();
actor_to_register_callbacks_.erase(actor_id);
actor_to_create_callbacks_.erase(actor_id);
auto callback_it = actor_to_create_callbacks_.find(actor_id);
const auto creation_callbacks = callback_it != actor_to_create_callbacks_.end()
? std::move(callback_it->second)
: std::vector<CreateActorCallback>{};
if (callback_it != actor_to_create_callbacks_.end()) {
actor_to_create_callbacks_.erase(callback_it);
}
auto it = registered_actors_.find(actor_id);
if (it == registered_actors_.end()) {
RAY_LOG(INFO) << "Tried to destroy actor that does not exist " << actor_id;
@ -750,8 +773,13 @@ void GcsActorManager::DestroyActor(const ActorID &actor_id,
RemoveActorNameFromRegistry(actor);
// The actor is already dead, most likely due to process or node failure.
if (actor->GetState() == rpc::ActorTableData::DEAD) {
RAY_LOG(DEBUG) << "Actor " << actor->GetActorID() << "has been dead,"
<< "skip sending killing request.";
RAY_LOG(DEBUG) << "Actor " << actor->GetActorID() << "is already dead,"
<< "skipping kill request.";
// Inform all creation callbacks that the actor is dead and that actor creation is
// therefore cancelled.
for (auto &callback : creation_callbacks) {
callback(actor, rpc::PushTaskReply(), true);
}
return;
}
if (actor->GetState() == rpc::ActorTableData::DEPENDENCIES_UNREADY) {
@ -801,6 +829,11 @@ void GcsActorManager::DestroyActor(const ActorID &actor_id,
// Destroy placement group owned by this actor.
destroy_owned_placement_group_if_needed_(actor_id);
}));
// Inform all creation callbacks that the actor was cancelled, not created.
for (auto &callback : creation_callbacks) {
callback(actor, rpc::PushTaskReply(), true);
}
}
absl::flat_hash_map<WorkerID, absl::flat_hash_set<ActorID>>
@ -1169,7 +1202,7 @@ void GcsActorManager::OnActorCreationSuccess(const std::shared_ptr<GcsActor> &ac
auto iter = actor_to_create_callbacks_.find(actor_id);
if (iter != actor_to_create_callbacks_.end()) {
for (auto &callback : iter->second) {
callback(actor, reply);
callback(actor, reply, false);
}
actor_to_create_callbacks_.erase(iter);
}

11
src/ray/gcs/gcs_server/gcs_actor_manager.h
View file

@ -164,8 +164,8 @@ class GcsActor {
};
using RegisterActorCallback = std::function<void(std::shared_ptr<GcsActor>)>;
using CreateActorCallback =
std::function<void(std::shared_ptr<GcsActor>, const rpc::PushTaskReply &reply)>;
using CreateActorCallback = std::function<void(
std::shared_ptr<GcsActor>, const rpc::PushTaskReply &reply, bool creation_cancelled)>;
/// GcsActorManager is responsible for managing the lifecycle of all actors.
/// This class is not thread-safe.
@ -273,9 +273,10 @@ class GcsActorManager : public rpc::ActorInfoHandler {
/// Create actor asynchronously.
///
/// \param request Contains the meta info to create the actor.
/// \param callback Will be invoked after the actor is created successfully or be
/// invoked immediately if the actor is already registered to `registered_actors_` and
/// its state is `ALIVE`.
/// \param callback Will be invoked after the actor is created successfully or if the
/// actor creation is cancelled (e.g. due to the actor going out-of-scope or being
/// killed before actor creation has been completed), or will be invoked immediately if
/// the actor is already registered to `registered_actors_` and its state is `ALIVE`.
/// \return Status::Invalid if this is a named actor and an actor with the specified
/// name already exists. The callback will not be called in this case.
Status CreateActor(const rpc::CreateActorRequest &request,

68
src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc
View file

@ -280,7 +280,8 @@ TEST_F(GcsActorManagerTest, TestBasic) {
Status status = gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](const std::shared_ptr<gcs::GcsActor> &actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
});
RAY_CHECK_OK(status);
@ -311,7 +312,8 @@ TEST_F(GcsActorManagerTest, TestSchedulingFailed) {
RAY_CHECK_OK(gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
}));
@ -338,7 +340,8 @@ TEST_F(GcsActorManagerTest, TestWorkerFailure) {
RAY_CHECK_OK(gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
}));
@ -385,7 +388,8 @@ TEST_F(GcsActorManagerTest, TestNodeFailure) {
Status status = gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
});
RAY_CHECK_OK(status);
@ -436,7 +440,8 @@ TEST_F(GcsActorManagerTest, TestActorReconstruction) {
Status status = gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
});
RAY_CHECK_OK(status);
@ -504,7 +509,8 @@ TEST_F(GcsActorManagerTest, TestActorRestartWhenOwnerDead) {
RAY_CHECK_OK(gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
}));
@ -555,7 +561,8 @@ TEST_F(GcsActorManagerTest, TestDetachedActorRestartWhenCreatorDead) {
RAY_CHECK_OK(gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
}));
@ -674,9 +681,10 @@ TEST_F(GcsActorManagerTest, TestNamedActorDeletionWorkerFailure) {
request1.mutable_task_spec()->CopyFrom(
registered_actor_1->GetCreationTaskSpecification().GetMessage());
Status status = gcs_actor_manager_->CreateActor(
request1,
[](std::shared_ptr<gcs::GcsActor> actor, const rpc::PushTaskReply &reply) {});
Status status = gcs_actor_manager_->CreateActor(request1,
[](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply,
bool creation_cancelled) {});
ASSERT_TRUE(status.ok());
ASSERT_EQ(gcs_actor_manager_->GetActorIDByName(actor_name, "test").Binary(),
request1.task_spec().actor_creation_task_spec().actor_id());
@ -711,9 +719,10 @@ TEST_F(GcsActorManagerTest, TestNamedActorDeletionWorkerFailure) {
request2.mutable_task_spec()->CopyFrom(
registered_actor_2->GetCreationTaskSpecification().GetMessage());
status = gcs_actor_manager_->CreateActor(
request2,
[](std::shared_ptr<gcs::GcsActor> actor, const rpc::PushTaskReply &reply) {});
status = gcs_actor_manager_->CreateActor(request2,
[](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply,
bool creation_cancelled) {});
ASSERT_TRUE(status.ok());
ASSERT_EQ(gcs_actor_manager_->GetActorIDByName(actor_name, "test").Binary(),
request2.task_spec().actor_creation_task_spec().actor_id());
@ -730,9 +739,10 @@ TEST_F(GcsActorManagerTest, TestNamedActorDeletionNodeFailure) {
request1.mutable_task_spec()->CopyFrom(
registered_actor_1->GetCreationTaskSpecification().GetMessage());
Status status = gcs_actor_manager_->CreateActor(
request1,
[](std::shared_ptr<gcs::GcsActor> actor, const rpc::PushTaskReply &reply) {});
Status status = gcs_actor_manager_->CreateActor(request1,
[](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply,
bool creation_cancelled) {});
ASSERT_TRUE(status.ok());
ASSERT_EQ(gcs_actor_manager_->GetActorIDByName("actor", "test").Binary(),
request1.task_spec().actor_creation_task_spec().actor_id());
@ -766,9 +776,10 @@ TEST_F(GcsActorManagerTest, TestNamedActorDeletionNodeFailure) {
request2.mutable_task_spec()->CopyFrom(
registered_actor_2->GetCreationTaskSpecification().GetMessage());
status = gcs_actor_manager_->CreateActor(
request2,
[](std::shared_ptr<gcs::GcsActor> actor, const rpc::PushTaskReply &reply) {});
status = gcs_actor_manager_->CreateActor(request2,
[](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply,
bool creation_cancelled) {});
ASSERT_TRUE(status.ok());
ASSERT_EQ(gcs_actor_manager_->GetActorIDByName("actor", "test").Binary(),
request2.task_spec().actor_creation_task_spec().actor_id());
@ -786,9 +797,10 @@ TEST_F(GcsActorManagerTest, TestNamedActorDeletionNotHappendWhenReconstructed) {
request1.mutable_task_spec()->CopyFrom(
registered_actor_1->GetCreationTaskSpecification().GetMessage());
Status status = gcs_actor_manager_->CreateActor(
request1,
[](std::shared_ptr<gcs::GcsActor> actor, const rpc::PushTaskReply &reply) {});
Status status = gcs_actor_manager_->CreateActor(request1,
[](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply,
bool creation_cancelled) {});
ASSERT_TRUE(status.ok());
ASSERT_EQ(gcs_actor_manager_->GetActorIDByName("actor", "test").Binary(),
request1.task_spec().actor_creation_task_spec().actor_id());
@ -835,7 +847,8 @@ TEST_F(GcsActorManagerTest, TestDestroyActorBeforeActorCreationCompletes) {
RAY_CHECK_OK(gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
}));
@ -871,7 +884,8 @@ TEST_F(GcsActorManagerTest, TestRaceConditionCancelLease) {
RAY_CHECK_OK(gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
}));
@ -915,7 +929,8 @@ TEST_F(GcsActorManagerTest, TestRegisterActor) {
RAY_CHECK_OK(gcs_actor_manager_->CreateActor(
request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(std::move(actor));
}));
// Make sure the actor is scheduling.
@ -1035,7 +1050,8 @@ TEST_F(GcsActorManagerTest, TestOwnerAndChildDiedAtTheSameTimeRaceCondition) {
RAY_CHECK_OK(gcs_actor_manager_->CreateActor(
create_actor_request,
[&finished_actors](std::shared_ptr<gcs::GcsActor> actor,
const rpc::PushTaskReply &reply) {
const rpc::PushTaskReply &reply,
bool creation_cancelled) {
finished_actors.emplace_back(actor);
}));
auto actor = mock_actor_scheduler_->actors.back();

3
src/ray/protobuf/common.proto
View file

@ -255,6 +255,9 @@ message ActorDiedErrorContext {
string class_name = 8;
// The id of the actor
bytes actor_id = 9;
// Whether the actor had never started running before it died, i.e. it was cancelled
// before scheduling had completed.
bool never_started = 10;
}
// ---Actor death contexts end----

23
src/ray/protobuf/gcs_service.proto
View file

@ -36,8 +36,7 @@ message MarkJobFinishedReply {
GcsStatus status = 1;
}
message GetAllJobInfoRequest {
}
message GetAllJobInfoRequest {}
message GetAllJobInfoReply {
GcsStatus status = 1;
@ -51,8 +50,7 @@ message ReportJobErrorReply {
GcsStatus status = 1;
}
message GetNextJobIDRequest {
}
message GetNextJobIDRequest {}
message GetNextJobIDReply {
GcsStatus status = 1;
int32 job_id = 2;
@ -172,8 +170,7 @@ message RegisterNodeReply {
GcsStatus status = 1;
}
message GetAllNodeInfoRequest {
}
message GetAllNodeInfoRequest {}
message GetAllNodeInfoReply {
GcsStatus status = 1;
@ -198,8 +195,7 @@ message CheckAliveReply {
repeated bool raylet_alive = 3;
}
message GetInternalConfigRequest {
}
message GetInternalConfigRequest {}
message GetInternalConfigReply {
GcsStatus status = 1;
@ -219,8 +215,7 @@ message DeleteResourcesReply {
GcsStatus status = 1;
}
message GetAllAvailableResourcesRequest {
}
message GetAllAvailableResourcesRequest {}
message GetAllAvailableResourcesReply {
GcsStatus status = 1;
@ -251,8 +246,7 @@ message AddProfileDataReply {
GcsStatus status = 1;
}
message GetAllProfileInfoRequest {
}
message GetAllProfileInfoRequest {}
message GetAllProfileInfoReply {
GcsStatus status = 1;
@ -330,6 +324,8 @@ message CreateActorReply {
Address actor_address = 2;
// Info about any refs that the created actor is borrowing.
repeated ObjectReferenceCount borrowed_refs = 3;
// The cause of this actor's death if creation was cancelled.
optional ActorDeathCause death_cause = 4;
}
message RegisterActorRequest {
@ -582,8 +578,7 @@ service InternalPubSubGcsService {
returns (GcsSubscriberCommandBatchReply);
}
message GetAllResourceUsageRequest {
}
message GetAllResourceUsageRequest {}
message GetAllResourceUsageReply {
GcsStatus status = 1;