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

Implement named actors using the GCS service (#8328)

Browse Source
This commit is contained in:
Edward Oakes
2020-05-09 08:58:10 -05:00
committed by GitHub
parent 93138e617a
commit 2677b71003
37 changed files with 483 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/_raylet.pxd
+2 -1
View File
@@ -17,7 +17,7 @@ from ray.includes.common cimport (
CBuffer,
CRayObject
)
from ray.includes.libcoreworker cimport CFiberEvent
from ray.includes.libcoreworker cimport CActorHandle, CFiberEvent
from ray.includes.unique_ids cimport (
CObjectID,
CActorID
@@ -86,6 +86,7 @@ cdef class CoreWorker:
self, worker, outputs, const c_vector[CObjectID] return_ids,
c_vector[shared_ptr[CRayObject]] *returns)
cdef yield_current_fiber(self, CFiberEvent &fiber_event)
cdef make_actor_handle(self, CActorHandle *c_actor_handle)
cdef class FunctionDescriptor:
cdef:
python/ray/_raylet.pyx
+37 -15
View File
@@ -112,6 +112,12 @@ include "includes/libcoreworker.pxi"
logger = logging.getLogger(__name__)
def gcs_actor_service_enabled():
return (
RayConfig.instance().gcs_service_enabled() and
RayConfig.instance().gcs_actor_service_enabled())
cdef int check_status(const CRayStatus& status) nogil except -1:
if status.ok():
return 0
@@ -125,6 +131,8 @@ cdef int check_status(const CRayStatus& status) nogil except -1:
raise KeyboardInterrupt()
elif status.IsTimedOut():
raise RayTimeoutError(message)
elif status.IsNotFound():
raise ValueError(message)
else:
raise RayletError(message)
@@ -899,6 +907,7 @@ cdef class CoreWorker:
placement_resources,
int32_t max_concurrency,
c_bool is_detached,
c_string name,
c_bool is_asyncio,
c_string extension_data):
cdef:
@@ -922,7 +931,7 @@ cdef class CoreWorker:
CActorCreationOptions(
max_reconstructions, max_concurrency,
c_resources, c_placement_resources,
dynamic_worker_options, is_detached, is_asyncio),
dynamic_worker_options, is_detached, name, is_asyncio),
extension_data,
&c_actor_id))
@@ -1013,23 +1022,12 @@ cdef class CoreWorker:
CCoreWorkerProcess.GetCoreWorker().RemoveActorHandleReference(
c_actor_id)
def deserialize_and_register_actor_handle(self, const c_string &bytes,
ObjectID
outer_object_id):
cdef:
CActorHandle* c_actor_handle
CObjectID c_outer_object_id = (outer_object_id.native() if
outer_object_id else
CObjectID.Nil())
cdef make_actor_handle(self, CActorHandle *c_actor_handle):
worker = ray.worker.global_worker
worker.check_connected()
manager = worker.function_actor_manager
c_actor_id = (CCoreWorkerProcess.GetCoreWorker()
.DeserializeAndRegisterActorHandle(
bytes, c_outer_object_id))
check_status(CCoreWorkerProcess.GetCoreWorker().GetActorHandle(
c_actor_id, &c_actor_handle))
actor_id = ActorID(c_actor_id.Binary())
actor_id = ActorID(c_actor_handle.GetActorID().Binary())
job_id = JobID(c_actor_handle.CreationJobID().Binary())
language = Language.from_native(c_actor_handle.ActorLanguage())
actor_creation_function_descriptor = \
@@ -1064,6 +1062,30 @@ cdef class CoreWorker:
actor_creation_function_descriptor,
worker.current_session_and_job)
def deserialize_and_register_actor_handle(self, const c_string &bytes,
ObjectID
outer_object_id):
cdef:
CActorHandle* c_actor_handle
CObjectID c_outer_object_id = (outer_object_id.native() if
outer_object_id else
CObjectID.Nil())
c_actor_id = (CCoreWorkerProcess.GetCoreWorker()
.DeserializeAndRegisterActorHandle(
bytes, c_outer_object_id))
check_status(CCoreWorkerProcess.GetCoreWorker().GetActorHandle(
c_actor_id, &c_actor_handle))
return self.make_actor_handle(c_actor_handle)
def get_named_actor_handle(self, const c_string &name):
cdef:
CActorHandle* c_actor_handle
with nogil:
check_status(CCoreWorkerProcess.GetCoreWorker()
.GetNamedActorHandle(name, &c_actor_handle))
return self.make_actor_handle(c_actor_handle)
def serialize_actor_handle(self, ActorID actor_id):
cdef:
c_string output
python/ray/actor.py
+16 -3
View File
@@ -10,7 +10,7 @@ import ray._raylet
import ray.signature as signature
import ray.worker
from ray import ActorClassID, Language
from ray._raylet import PythonFunctionDescriptor
from ray._raylet import PythonFunctionDescriptor, gcs_actor_service_enabled
from ray import cross_language
logger = logging.getLogger(__name__)
@@ -472,11 +472,23 @@ class ActorClass:
"Please use Actor._remote(name='some_name') "
"to associate the name.")
if name and not detached:
raise ValueError("Only detached actors can be named. "
"Please use Actor._remote(detached=True, "
"name='some_name').")
if name == "":
raise ValueError("Actor name cannot be an empty string.")
# Check whether the name is already taken.
# TODO(edoakes): this check has a race condition because two drivers
# could pass the check and then create the same named actor. We should
# instead check this when we create the actor, but that's currently an
# async call.
if name is not None:
try:
ray.util.get_actor(name)
except ValueError: # name is not taken, expected.
except ValueError: # Name is not taken.
pass
else:
raise ValueError(
@@ -551,6 +563,7 @@ class ActorClass:
actor_placement_resources,
max_concurrency,
detached,
name if name is not None else "",
is_asyncio,
# Store actor_method_cpu in actor handle's extension data.
extension_data=str(actor_method_cpu))
@@ -566,7 +579,7 @@ class ActorClass:
worker.current_session_and_job,
original_handle=True)
if name is not None:
if name is not None and not gcs_actor_service_enabled():
ray.util.register_actor(name, actor_handle)
return actor_handle
python/ray/includes/common.pxd
+5 -1
View File
@@ -88,6 +88,9 @@ cdef extern from "ray/common/status.h" namespace "ray" nogil:
@staticmethod
CRayStatus UnexpectedSystemExit()
@staticmethod
CRayStatus NotFound()
c_bool ok()
c_bool IsOutOfMemory()
c_bool IsKeyError()
@@ -101,6 +104,7 @@ cdef extern from "ray/common/status.h" namespace "ray" nogil:
c_bool IsTimedOut()
c_bool IsInterrupted()
c_bool IsSystemExit()
c_bool IsNotFound()
c_string ToString()
c_string CodeAsString()
@@ -231,7 +235,7 @@ cdef extern from "ray/core_worker/common.h" nogil:
const unordered_map[c_string, double] &resources,
const unordered_map[c_string, double] &placement_resources,
const c_vector[c_string] &dynamic_worker_options,
c_bool is_detached, c_bool is_asyncio)
c_bool is_detached, c_string &name, c_bool is_asyncio)
cdef extern from "ray/gcs/gcs_client.h" nogil:
cdef cppclass CGcsClientOptions "ray::gcs::GcsClientOptions":
python/ray/includes/libcoreworker.pxd
+2
View File
@@ -123,6 +123,8 @@ cdef extern from "ray/core_worker/core_worker.h" nogil:
CObjectID *c_actor_handle_id)
CRayStatus GetActorHandle(const CActorID &actor_id,
CActorHandle **actor_handle) const
CRayStatus GetNamedActorHandle(const c_string &name,
CActorHandle **actor_handle)
void AddLocalReference(const CObjectID &object_id)
void RemoveLocalReference(const CObjectID &object_id)
void PromoteObjectToPlasma(const CObjectID &object_id)
python/ray/includes/ray_config.pxd
+5
View File
@@ -1,3 +1,4 @@
from libcpp cimport bool as c_bool
from libc.stdint cimport int64_t, uint64_t, uint32_t
from libcpp.string cimport string as c_string
from libcpp.unordered_map cimport unordered_map
@@ -83,3 +84,7 @@ cdef extern from "ray/common/ray_config.h" nogil:
uint32_t maximum_gcs_deletion_batch_size() const
int64_t max_direct_call_object_size() const
c_bool gcs_service_enabled() const
c_bool gcs_actor_service_enabled() const
python/ray/tests/test_actor_advanced.py
+17 -6
View File
@@ -658,20 +658,31 @@ def test_detached_actor(ray_start_regular):
def ping(self):
return "pong"
with pytest.raises(Exception, match="Detached actors must be named"):
with pytest.raises(
ValueError, match="Actor name cannot be an empty string"):
DetachedActor._remote(detached=True, name="")
with pytest.raises(ValueError, match="Detached actors must be named"):
DetachedActor._remote(detached=True)
with pytest.raises(ValueError, match="Please use a different name"):
_ = DetachedActor._remote(name="d_actor")
with pytest.raises(ValueError, match="Only detached actors can be named"):
DetachedActor._remote(name="d_actor")
DetachedActor._remote(detached=True, name="d_actor")
with pytest.raises(ValueError, match="Please use a different name"):
DetachedActor._remote(detached=True, name="d_actor")
redis_address = ray_start_regular["redis_address"]
actor_name = "DetachedActor"
get_actor_name = "d_actor"
create_actor_name = "DetachedActor"
driver_script = """
import ray
ray.init(address="{}")
existing_actor = ray.util.get_actor("{}")
assert ray.get(existing_actor.ping.remote()) == "pong"
@ray.remote
class DetachedActor:
def ping(self):
@@ -679,10 +690,10 @@ class DetachedActor:
actor = DetachedActor._remote(name="{}", detached=True)
ray.get(actor.ping.remote())
""".format(redis_address, actor_name)
""".format(redis_address, get_actor_name, create_actor_name)
run_string_as_driver(driver_script)
detached_actor = ray.util.get_actor(actor_name)
detached_actor = ray.util.get_actor(create_actor_name)
assert ray.get(detached_actor.ping.remote()) == "pong"
python/ray/util/named_actors.py
+11 -5
View File
@@ -26,11 +26,17 @@ def get_actor(name):
Returns:
The ActorHandle object corresponding to the name.
"""
actor_name = _calculate_key(name)
pickled_state = _internal_kv_get(actor_name)
if pickled_state is None:
raise ValueError("The actor with name={} doesn't exist".format(name))
handle = pickle.loads(pickled_state)
if ray._raylet.gcs_actor_service_enabled():
worker = ray.worker.global_worker
handle = worker.core_worker.get_named_actor_handle(name)
else:
actor_name = _calculate_key(name)
pickled_state = _internal_kv_get(actor_name)
if pickled_state is None:
raise ValueError(
"The actor with name={} doesn't exist".format(name))
handle = pickle.loads(pickled_state)
return handle
src/ray/common/status.cc
+3 -1
View File
@@ -51,6 +51,7 @@ namespace ray {
#define STATUS_CODE_INTENTIONAL_SYSTEM_EXIT "IntentionalSystemExit"
#define STATUS_CODE_UNEXPECTED_SYSTEM_EXIT "UnexpectedSystemExit"
#define STATUS_CODE_UNKNOWN "Unknown"
#define STATUS_CODE_NOT_FOUND "NotFound"
Status::Status(StatusCode code, const std::string &msg) {
assert(code != StatusCode::OK);
@@ -88,7 +89,8 @@ std::string Status::CodeAsString() const {
{StatusCode::TimedOut, STATUS_CODE_TIMED_OUT},
{StatusCode::Interrupted, STATUS_CODE_INTERRUPTED},
{StatusCode::IntentionalSystemExit, STATUS_CODE_INTENTIONAL_SYSTEM_EXIT},
{StatusCode::UnexpectedSystemExit, STATUS_CODE_UNEXPECTED_SYSTEM_EXIT}};
{StatusCode::UnexpectedSystemExit, STATUS_CODE_UNEXPECTED_SYSTEM_EXIT},
{StatusCode::NotFound, STATUS_CODE_NOT_FOUND}};
if (!code_to_str.count(code())) {
return STATUS_CODE_UNKNOWN;
src/ray/common/status.h
+6
View File
@@ -107,6 +107,7 @@ enum class StatusCode : char {
Interrupted = 13,
IntentionalSystemExit = 14,
UnexpectedSystemExit = 15,
NotFound = 16,
};
#if defined(__clang__)
@@ -186,6 +187,10 @@ class RAY_EXPORT Status {
return Status(StatusCode::UnexpectedSystemExit, "user code caused exit");
}
static Status NotFound(const std::string &msg) {
return Status(StatusCode::NotFound, msg);
}
// Returns true iff the status indicates success.
bool ok() const { return (state_ == NULL); }
@@ -208,6 +213,7 @@ class RAY_EXPORT Status {
bool IsIntentionalSystemExit() const {
return code() == StatusCode::IntentionalSystemExit;
}
bool IsNotFound() const { return code() == StatusCode::NotFound; }
// Return a string representation of this status suitable for printing.
// Returns the string "OK" for success.
src/ray/common/task/task_util.h
+5 -2
View File
@@ -103,7 +103,8 @@ class TaskSpecBuilder {
TaskSpecBuilder &SetActorCreationTaskSpec(
const ActorID &actor_id, uint64_t max_reconstructions = 0,
const std::vector<std::string> &dynamic_worker_options = {},
int max_concurrency = 1, bool is_detached = false, bool is_asyncio = false) {
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());
@@ -112,8 +113,10 @@ class TaskSpecBuilder {
actor_creation_spec->add_dynamic_worker_options(option);
}
actor_creation_spec->set_max_concurrency(max_concurrency);
actor_creation_spec->set_is_asyncio(is_asyncio);
actor_creation_spec->set_is_detached(is_detached);
actor_creation_spec->set_name(name);
actor_creation_spec->set_is_asyncio(is_asyncio);
actor_creation_spec->set_extension_data(extension_data);
return *this;
}
src/ray/core_worker/actor_handle.cc
+20
View File
@@ -43,6 +43,23 @@ ray::rpc::ActorHandle CreateInnerActorHandleFromString(const std::string &serial
return inner;
}
ray::rpc::ActorHandle CreateInnerActorHandleFromActorTableData(
const ray::gcs::ActorTableData &actor_table_data) {
ray::rpc::ActorHandle inner;
inner.set_actor_id(actor_table_data.actor_id());
inner.set_owner_id(actor_table_data.parent_id());
inner.mutable_owner_address()->CopyFrom(actor_table_data.owner_address());
inner.set_creation_job_id(actor_table_data.job_id());
inner.set_actor_language(actor_table_data.task_spec().language());
inner.mutable_actor_creation_task_function_descriptor()->CopyFrom(
actor_table_data.task_spec().function_descriptor());
ray::TaskSpecification task_spec(actor_table_data.task_spec());
inner.set_actor_cursor(task_spec.ReturnId(0, ray::TaskTransportType::DIRECT).Binary());
inner.set_extension_data(
actor_table_data.task_spec().actor_creation_task_spec().extension_data());
return inner;
}
} // namespace
namespace ray {
@@ -60,6 +77,9 @@ ActorHandle::ActorHandle(
ActorHandle::ActorHandle(const std::string &serialized)
: ActorHandle(CreateInnerActorHandleFromString(serialized)) {}
ActorHandle::ActorHandle(const gcs::ActorTableData &actor_table_data)
: ActorHandle(CreateInnerActorHandleFromActorTableData(actor_table_data)) {}
void ActorHandle::SetActorTaskSpec(TaskSpecBuilder &builder, const ObjectID new_cursor) {
absl::MutexLock guard(&mutex_);
// Build actor task spec.
src/ray/core_worker/actor_handle.h
+4
View File
@@ -21,6 +21,7 @@
#include "ray/common/task/task_util.h"
#include "ray/core_worker/common.h"
#include "ray/core_worker/context.h"
#include "ray/gcs/redis_gcs_client.h"
#include "ray/protobuf/core_worker.pb.h"
#include "ray/protobuf/gcs.pb.h"
@@ -41,6 +42,9 @@ class ActorHandle {
/// Constructs an ActorHandle from a serialized string.
ActorHandle(const std::string &serialized);
/// Constructs an ActorHandle from a gcs::ActorTableData message.
ActorHandle(const gcs::ActorTableData &actor_table_data);
ActorID GetActorID() const { return ActorID::FromBinary(inner_.actor_id()); };
TaskID GetOwnerId() const { return TaskID::FromBinary(inner_.owner_id()); }
src/ray/core_worker/common.h
+6 -1
View File
@@ -115,13 +115,14 @@ struct ActorCreationOptions {
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, bool is_asyncio)
bool is_detached, std::string &name, bool is_asyncio)
: max_reconstructions(max_reconstructions),
max_concurrency(max_concurrency),
resources(resources),
placement_resources(placement_resources),
dynamic_worker_options(dynamic_worker_options),
is_detached(is_detached),
name(name),
is_asyncio(is_asyncio){};
/// Maximum number of times that the actor should be reconstructed when it dies
@@ -139,6 +140,10 @@ struct ActorCreationOptions {
/// Whether to keep the actor persistent after driver exit. If true, this will set
/// the worker to not be destroyed after the driver shutdown.
const bool is_detached = false;
/// The name to give this detached actor that can be used to get a handle to it from
/// other drivers. This must be globally unique across the cluster.
/// This should set if and only if is_detached is true.
const std::string name;
/// Whether to use async mode of direct actor call.
const bool is_asyncio = false;
};
src/ray/core_worker/core_worker.cc
+62 -5
View File
@@ -1145,11 +1145,11 @@ Status CoreWorker::CreateActor(const RayFunction &function,
worker_context_.GetCurrentTaskID(), next_task_index, GetCallerId(),
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_creation_options.dynamic_worker_options,
actor_creation_options.max_concurrency,
actor_creation_options.is_detached,
actor_creation_options.is_asyncio);
builder.SetActorCreationTaskSpec(
actor_id, actor_creation_options.max_reconstructions,
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);
*return_actor_id = actor_id;
TaskSpecification task_spec = builder.Build();
@@ -1362,6 +1362,63 @@ Status CoreWorker::GetActorHandle(const ActorID &actor_id,
return Status::OK();
}
Status CoreWorker::GetNamedActorHandle(const std::string &name,
ActorHandle **actor_handle) {
RAY_CHECK(RayConfig::instance().gcs_service_enabled());
RAY_CHECK(RayConfig::instance().gcs_actor_service_enabled());
RAY_CHECK(!name.empty());
// This call needs to be blocking because we can't return until the actor
// handle is created, which requires the response from the RPC. This is
// implemented using a condition variable that's captured in the RPC
// callback. There should be no risk of deadlock because we don't hold any
// locks during the call and the RPCs run on a separate thread.
ActorID actor_id;
std::shared_ptr<bool> ready = std::make_shared<bool>(false);
std::shared_ptr<std::mutex> m = std::make_shared<std::mutex>();
std::shared_ptr<std::condition_variable> cv =
std::make_shared<std::condition_variable>();
std::unique_lock<std::mutex> lk(*m);
RAY_CHECK_OK(gcs_client_->Actors().AsyncGetByName(
name, [this, &actor_id, name, ready, m, cv](
Status status, const boost::optional<gcs::ActorTableData> &result) {
if (!status.ok()) {
RAY_LOG(INFO) << "Failed to look up actor with name: " << name;
// Use a NIL actor ID to signal that the actor wasn't found.
actor_id = ActorID::Nil();
} else {
RAY_CHECK(result);
auto actor_handle = std::unique_ptr<ActorHandle>(new ActorHandle(*result));
actor_id = actor_handle->GetActorID();
AddActorHandle(std::move(actor_handle), /*is_owner_handle=*/false);
}
// Notify the main thread that the RPC has finished.
{
std::unique_lock<std::mutex> lk(*m);
*ready = true;
}
cv->notify_one();
}));
// Block until the RPC completes. Set a timeout to avoid hangs if the
// GCS service crashes.
cv->wait_for(lk, std::chrono::seconds(5), [ready] { return *ready; });
if (!*ready) {
return Status::TimedOut("Timed out trying to get named actor.");
}
Status status;
if (actor_id.IsNil()) {
std::stringstream stream;
stream << "Failed to look up actor with name '" << name << "'.";
status = Status::NotFound(stream.str());
} else {
status = GetActorHandle(actor_id, actor_handle);
}
return status;
}
const ResourceMappingType CoreWorker::GetResourceIDs() const {
absl::MutexLock lock(&mutex_);
return *resource_ids_;
src/ray/core_worker/core_worker.h
+7
View File
@@ -664,6 +664,13 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
/// \return Status::Invalid if we don't have this actor handle.
Status GetActorHandle(const ActorID &actor_id, ActorHandle **actor_handle) const;
/// Get a handle to a named actor.
///
/// \param[in] name The name of the actor whose handle to get.
/// \param[out] actor_handle A handle to the requested actor.
/// \return Status::NotFound if an actor with the specified name wasn't found.
Status GetNamedActorHandle(const std::string &name, ActorHandle **actor_handle);
///
/// The following methods are handlers for the core worker's gRPC server, which follow
/// a macro-generated call convention. These are executed on the io_service_ and
src/ray/core_worker/lib/java/io_ray_runtime_task_NativeTaskSubmitter.cc
+2
View File
@@ -107,6 +107,7 @@ inline ray::ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
actorCreationOptions, java_actor_creation_options_max_concurrency));
}
std::string name = "";
ray::ActorCreationOptions actor_creation_options{
static_cast<uint64_t>(max_reconstructions),
static_cast<int>(max_concurrency),
@@ -114,6 +115,7 @@ inline ray::ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
resources,
dynamic_worker_options,
/*is_detached=*/false,
name,
/*is_asyncio=*/false};
return actor_creation_options;
}
src/ray/core_worker/test/core_worker_test.cc
+9 -6
View File
@@ -70,10 +70,11 @@ ActorID CreateActorHelper(std::unordered_map<std::string, double> &resources,
args.emplace_back(TaskArg::PassByValue(
std::make_shared<RayObject>(buffer, nullptr, std::vector<ObjectID>())));
ActorCreationOptions actor_options{max_reconstructions,
/*max_concurrency*/ 1, resources, resources, {},
/*is_detached*/ false,
/*is_asyncio*/ false};
std::string name = "";
ActorCreationOptions actor_options{
max_reconstructions,
/*max_concurrency*/ 1, resources, resources, {},
/*is_detached=*/false, name, /*is_asyncio=*/false};
// Create an actor.
ActorID actor_id;
@@ -641,13 +642,15 @@ TEST_F(ZeroNodeTest, TestTaskSpecPerf) {
std::make_shared<RayObject>(buffer, nullptr, std::vector<ObjectID>())));
std::unordered_map<std::string, double> resources;
std::string name = "";
ActorCreationOptions actor_options{0,
1,
resources,
resources,
{},
/*is_detached*/ false,
/*is_asyncio*/ false};
/*is_detached=*/false,
name,
/*is_asyncio=*/false};
const auto job_id = NextJobId();
ActorHandle actor_handle(ActorID::Of(job_id, TaskID::ForDriverTask(job_id), 1),
TaskID::Nil(), rpc::Address(), job_id, ObjectID::FromRandom(),
src/ray/core_worker/transport/direct_task_transport.cc
+12 -12
View File
@@ -30,19 +30,19 @@ Status CoreWorkerDirectTaskSubmitter::SubmitTask(TaskSpecification task_spec) {
auto actor_id = task_spec.ActorCreationId();
auto task_id = task_spec.TaskId();
RAY_LOG(INFO) << "Submitting actor creation task to GCS: " << actor_id;
auto status =
RAY_CHECK_OK(
actor_create_callback_(task_spec, [this, actor_id, task_id](Status status) {
// If GCS is failed, GcsRpcClient may receive IOError status but it will
// not trigger this callback, because GcsRpcClient has retry logic at the
// bottom. So if this callback is invoked with an error there must be
// something wrong with the protocol of gcs-based actor management.
// So just check `status.ok()` here.
RAY_CHECK_OK(status);
RAY_LOG(INFO) << "Actor creation task submitted to GCS: " << actor_id;
task_finisher_->CompletePendingTask(task_id, rpc::PushTaskReply(),
rpc::Address());
});
RAY_CHECK_OK(status);
if (status.ok()) {
RAY_LOG(INFO) << "Actor creation task submitted to GCS: " << actor_id;
task_finisher_->CompletePendingTask(task_id, rpc::PushTaskReply(),
rpc::Address());
} else {
RAY_LOG(ERROR) << "Failed to create actor " << actor_id
<< " with: " << status.ToString();
task_finisher_->PendingTaskFailed(
task_id, rpc::ErrorType::ACTOR_CREATION_FAILED, &status);
}
}));
return;
}
src/ray/gcs/accessor.h
+9
View File
@@ -47,6 +47,15 @@ class ActorInfoAccessor {
virtual Status AsyncGet(const ActorID &actor_id,
const OptionalItemCallback<rpc::ActorTableData> &callback) = 0;
/// Get actor specification for a named actor from GCS asynchronously.
///
/// \param name The name of the detached actor to look up in the GCS.
/// \param callback Callback that will be called after lookup finishes.
/// \return Status
virtual Status AsyncGetByName(
const std::string &name,
const OptionalItemCallback<rpc::ActorTableData> &callback) = 0;
/// Create an actor to GCS asynchronously.
///
/// \param task_spec The specification for the actor creation task.
src/ray/gcs/gcs_client/service_based_accessor.cc
+20
View File
@@ -107,6 +107,26 @@ Status ServiceBasedActorInfoAccessor::AsyncGet(
return Status::OK();
}
Status ServiceBasedActorInfoAccessor::AsyncGetByName(
const std::string &name, const OptionalItemCallback<rpc::ActorTableData> &callback) {
RAY_LOG(DEBUG) << "Getting actor info, name = " << name;
rpc::GetNamedActorInfoRequest request;
request.set_name(name);
client_impl_->GetGcsRpcClient().GetNamedActorInfo(
request,
[name, callback](const Status &status, const rpc::GetNamedActorInfoReply &reply) {
if (reply.has_actor_table_data()) {
rpc::ActorTableData actor_table_data(reply.actor_table_data());
callback(status, actor_table_data);
} else {
callback(status, boost::none);
}
RAY_LOG(DEBUG) << "Finished getting actor info, status = " << status
<< ", name = " << name;
});
return Status::OK();
}
Status ServiceBasedActorInfoAccessor::AsyncCreateActor(
const ray::TaskSpecification &task_spec, const ray::gcs::StatusCallback &callback) {
RAY_CHECK(task_spec.IsActorCreationTask() && callback);
src/ray/gcs/gcs_client/service_based_accessor.h
+4
View File
@@ -61,6 +61,10 @@ class ServiceBasedActorInfoAccessor : public ActorInfoAccessor {
Status AsyncGet(const ActorID &actor_id,
const OptionalItemCallback<rpc::ActorTableData> &callback) override;
Status AsyncGetByName(
const std::string &name,
const OptionalItemCallback<rpc::ActorTableData> &callback) override;
Status AsyncCreateActor(const TaskSpecification &task_spec,
const StatusCallback &callback) override;
src/ray/gcs/gcs_server/actor_info_handler_impl.cc
+52 -7
View File
@@ -26,19 +26,24 @@ void DefaultActorInfoHandler::HandleCreateActor(
ActorID::FromBinary(request.task_spec().actor_creation_task_spec().actor_id());
RAY_LOG(INFO) << "Registering actor, actor id = " << actor_id;
gcs_actor_manager_.RegisterActor(request, [reply, send_reply_callback, actor_id](
std::shared_ptr<gcs::GcsActor> actor) {
RAY_LOG(INFO) << "Registered actor, actor id = " << actor_id;
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
});
Status status = gcs_actor_manager_.RegisterActor(
request,
[reply, send_reply_callback, actor_id](std::shared_ptr<gcs::GcsActor> actor) {
RAY_LOG(INFO) << "Registered actor, actor id = " << actor_id;
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
});
if (!status.ok()) {
RAY_LOG(ERROR) << "Failed to create actor: " << status.ToString();
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
}
}
void DefaultActorInfoHandler::HandleGetActorInfo(
const rpc::GetActorInfoRequest &request, rpc::GetActorInfoReply *reply,
rpc::SendReplyCallback send_reply_callback) {
ActorID actor_id = ActorID::FromBinary(request.actor_id());
RAY_LOG(DEBUG) << "Getting actor info, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id;
RAY_LOG(DEBUG) << "Getting actor info"
<< ", job id = " << actor_id.JobId() << ", actor id = " << actor_id;
auto on_done = [actor_id, reply, send_reply_callback](
Status status, const boost::optional<ActorTableData> &result) {
@@ -53,14 +58,54 @@ void DefaultActorInfoHandler::HandleGetActorInfo(
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
};
// Look up the actor_id in the GCS.
Status status = gcs_client_.Actors().AsyncGet(actor_id, on_done);
if (!status.ok()) {
on_done(status, boost::none);
}
RAY_LOG(DEBUG) << "Finished getting actor info, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id;
}
void DefaultActorInfoHandler::HandleGetNamedActorInfo(
const rpc::GetNamedActorInfoRequest &request, rpc::GetNamedActorInfoReply *reply,
rpc::SendReplyCallback send_reply_callback) {
const std::string &name = request.name();
RAY_LOG(DEBUG) << "Getting actor info"
<< ", name = " << name;
auto on_done = [name, reply, send_reply_callback](
Status status, const boost::optional<ActorTableData> &result) {
if (status.ok()) {
if (result) {
reply->mutable_actor_table_data()->CopyFrom(*result);
}
} else {
RAY_LOG(ERROR) << "Failed to get actor info: " << status.ToString()
<< ", name = " << name;
}
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
};
// Try to look up the actor ID for the named actor.
ActorID actor_id = gcs_actor_manager_.GetActorIDByName(name);
if (actor_id.IsNil()) {
// The named actor was not found.
std::stringstream stream;
stream << "Actor with name '" << name << "' was not found.";
on_done(Status::NotFound(stream.str()), boost::none);
} else {
// Look up the actor_id in the GCS.
Status status = gcs_client_.Actors().AsyncGet(actor_id, on_done);
if (!status.ok()) {
on_done(status, boost::none);
}
RAY_LOG(DEBUG) << "Finished getting actor info, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id;
}
}
void DefaultActorInfoHandler::HandleRegisterActorInfo(
const rpc::RegisterActorInfoRequest &request, rpc::RegisterActorInfoReply *reply,
rpc::SendReplyCallback send_reply_callback) {
src/ray/gcs/gcs_server/actor_info_handler_impl.h
+4
View File
@@ -35,6 +35,10 @@ class DefaultActorInfoHandler : public rpc::ActorInfoHandler {
void HandleGetActorInfo(const GetActorInfoRequest &request, GetActorInfoReply *reply,
SendReplyCallback send_reply_callback) override;
void HandleGetNamedActorInfo(const GetNamedActorInfoRequest &request,
GetNamedActorInfoReply *reply,
SendReplyCallback send_reply_callback) override;
void HandleRegisterActorInfo(const RegisterActorInfoRequest &request,
RegisterActorInfoReply *reply,
SendReplyCallback send_reply_callback) override;
src/ray/gcs/gcs_server/gcs_actor_manager.cc
+33 -5
View File
@@ -54,6 +54,14 @@ ActorID GcsActor::GetActorID() const {
return ActorID::FromBinary(actor_table_data_.actor_id());
}
bool GcsActor::IsDetached() const { return actor_table_data_.is_detached(); }
std::string GcsActor::GetName() const {
RAY_CHECK(actor_table_data_.is_detached())
<< "Actor names are only valid for detached actors.";
return actor_table_data_.name();
}
TaskSpecification GcsActor::GetCreationTaskSpecification() const {
const auto &task_spec = actor_table_data_.task_spec();
return TaskSpecification(task_spec);
@@ -71,7 +79,7 @@ GcsActorManager::GcsActorManager(std::shared_ptr<GcsActorSchedulerInterface> sch
: gcs_actor_scheduler_(std::move(scheduler)),
actor_info_accessor_(actor_info_accessor) {}
void GcsActorManager::RegisterActor(
Status GcsActorManager::RegisterActor(
const ray::rpc::CreateActorRequest &request,
std::function<void(std::shared_ptr<GcsActor>)> callback) {
RAY_CHECK(callback);
@@ -86,7 +94,7 @@ void GcsActorManager::RegisterActor(
// receiving the reply from GcsServer. If the actor has been created successfully then
// just reply to the caller.
callback(iter->second);
return;
return Status::OK();
}
auto pending_register_iter = actor_to_register_callbacks_.find(actor_id);
@@ -94,16 +102,36 @@ void GcsActorManager::RegisterActor(
// It is a duplicate message, just mark the callback as pending and invoke it after
// the actor has been successfully created.
pending_register_iter->second.emplace_back(std::move(callback));
return;
return Status::OK();
}
auto actor = std::make_shared<GcsActor>(request);
if (actor->IsDetached()) {
auto it = named_actors_.find(actor->GetName());
if (it == named_actors_.end()) {
named_actors_.emplace(actor->GetName(), actor->GetActorID());
} else {
std::stringstream stream;
stream << "Actor with name '" << actor->GetName() << "' already exists.";
return Status::Invalid(stream.str());
}
}
// Mark the callback as pending and invoke it after the actor has been successfully
// created.
actor_to_register_callbacks_[actor_id].emplace_back(std::move(callback));
auto actor = std::make_shared<GcsActor>(request);
RAY_CHECK(registered_actors_.emplace(actor->GetActorID(), actor).second);
gcs_actor_scheduler_->Schedule(actor);
return Status::OK();
}
ActorID GcsActorManager::GetActorIDByName(const std::string &name) {
ActorID actor_id = ActorID::Nil();
auto it = named_actors_.find(name);
if (it != named_actors_.end()) {
actor_id = it->second;
}
return actor_id;
}
void GcsActorManager::ReconstructActorOnWorker(const ray::ClientID &node_id,
src/ray/gcs/gcs_server/gcs_actor_manager.h
+16 -2
View File
@@ -56,6 +56,7 @@ class GcsActor {
actor_table_data_.set_actor_creation_dummy_object_id(dummy_object);
actor_table_data_.set_is_detached(actor_creation_task_spec.is_detached());
actor_table_data_.set_name(actor_creation_task_spec.name());
actor_table_data_.mutable_owner_address()->CopyFrom(
request.task_spec().caller_address());
@@ -83,6 +84,10 @@ class GcsActor {
/// Get the id of this actor.
ActorID GetActorID() const;
/// Returns whether or not this is a detached actor.
bool IsDetached() const;
/// Get the name of this actor (only set if it's a detached actor).
std::string GetName() const;
/// Get the task specification of this actor.
TaskSpecification GetCreationTaskSpecification() const;
@@ -117,8 +122,15 @@ class GcsActorManager {
/// \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`.
void RegisterActor(const rpc::CreateActorRequest &request,
RegisterActorCallback callback);
/// \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 RegisterActor(const rpc::CreateActorRequest &request,
RegisterActorCallback callback);
/// Get the actor ID for the named actor. Returns nil if the actor was not found.
/// \param name The name of the detached actor to look up.
/// \returns ActorID The ID of the actor. Nil if the actor was not found.
ActorID GetActorIDByName(const std::string &name);
/// Schedule actors in the `pending_actors_` queue.
/// This method should be called when new nodes are registered or resources
@@ -172,6 +184,8 @@ class GcsActorManager {
/// All registered actors (pending actors are also included).
/// TODO(swang): Use unique_ptr instead of shared_ptr.
absl::flat_hash_map<ActorID, std::shared_ptr<GcsActor>> registered_actors_;
/// Maps detached actor names to their actor ID for lookups by name.
absl::flat_hash_map<std::string, ActorID> named_actors_;
/// The pending actors which will not be scheduled until there's a resource change.
std::vector<std::shared_ptr<GcsActor>> pending_actors_;
/// Map contains the relationship of node and created actors. Each node ID
src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc
+48 -3
View File
@@ -50,10 +50,11 @@ TEST_F(GcsActorManagerTest, TestBasic) {
auto job_id = JobID::FromInt(1);
auto create_actor_request = Mocker::GenCreateActorRequest(job_id);
std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
gcs_actor_manager_.RegisterActor(
Status status = gcs_actor_manager_.RegisterActor(
create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
finished_actors.emplace_back(actor);
});
RAY_CHECK_OK(status);
ASSERT_EQ(finished_actors.size(), 0);
ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
@@ -93,10 +94,11 @@ TEST_F(GcsActorManagerTest, TestBasicNodeFailure) {
auto job_id = JobID::FromInt(1);
auto create_actor_request = Mocker::GenCreateActorRequest(job_id);
std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
gcs_actor_manager_.RegisterActor(
Status status = gcs_actor_manager_.RegisterActor(
create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
finished_actors.emplace_back(actor);
});
RAY_CHECK_OK(status);
ASSERT_EQ(finished_actors.size(), 0);
ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
@@ -138,10 +140,11 @@ TEST_F(GcsActorManagerTest, TestActorReconstruction) {
auto create_actor_request = Mocker::GenCreateActorRequest(
job_id, /*max_reconstructions=*/1, /*detached=*/false);
std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
gcs_actor_manager_.RegisterActor(
Status status = gcs_actor_manager_.RegisterActor(
create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
finished_actors.emplace_back(actor);
});
RAY_CHECK_OK(status);
ASSERT_EQ(finished_actors.size(), 0);
ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
@@ -190,6 +193,48 @@ TEST_F(GcsActorManagerTest, TestActorReconstruction) {
ASSERT_EQ(mock_actor_scheduler_->actors.size(), 0);
}
TEST_F(GcsActorManagerTest, TestNamedActors) {
auto job_id_1 = JobID::FromInt(1);
auto job_id_2 = JobID::FromInt(2);
auto request1 =
Mocker::GenCreateActorRequest(job_id_1, 0, /*is_detached=*/true, /*name=*/"actor1");
Status status = gcs_actor_manager_.RegisterActor(
request1, [](std::shared_ptr<gcs::GcsActor> actor) {});
ASSERT_TRUE(status.ok());
ASSERT_EQ(gcs_actor_manager_.GetActorIDByName("actor1").Binary(),
request1.task_spec().actor_creation_task_spec().actor_id());
auto request2 =
Mocker::GenCreateActorRequest(job_id_1, 0, /*is_detached=*/true, /*name=*/"actor2");
status = gcs_actor_manager_.RegisterActor(request2,
[](std::shared_ptr<gcs::GcsActor> actor) {});
ASSERT_TRUE(status.ok());
ASSERT_EQ(gcs_actor_manager_.GetActorIDByName("actor2").Binary(),
request2.task_spec().actor_creation_task_spec().actor_id());
// Check that looking up a non-existent name returns ActorID::Nil();
ASSERT_EQ(gcs_actor_manager_.GetActorIDByName("actor3"), ActorID::Nil());
// Check that naming collisions return Status::Invalid.
auto request3 =
Mocker::GenCreateActorRequest(job_id_1, 0, /*is_detached=*/true, /*name=*/"actor2");
status = gcs_actor_manager_.RegisterActor(request3,
[](std::shared_ptr<gcs::GcsActor> actor) {});
ASSERT_TRUE(status.IsInvalid());
ASSERT_EQ(gcs_actor_manager_.GetActorIDByName("actor2").Binary(),
request2.task_spec().actor_creation_task_spec().actor_id());
// Check that naming collisions are enforced across JobIDs.
auto request4 =
Mocker::GenCreateActorRequest(job_id_2, 0, /*is_detached=*/true, /*name=*/"actor2");
status = gcs_actor_manager_.RegisterActor(request4,
[](std::shared_ptr<gcs::GcsActor> actor) {});
ASSERT_TRUE(status.IsInvalid());
ASSERT_EQ(gcs_actor_manager_.GetActorIDByName("actor2").Binary(),
request2.task_spec().actor_creation_task_spec().actor_id());
}
} // namespace ray
int main(int argc, char **argv) {
src/ray/gcs/gcs_server/test/gcs_server_test_util.h
+6
View File
@@ -179,6 +179,12 @@ struct GcsServerMocker {
return Status::NotImplemented("");
}
Status AsyncGetByName(
const std::string &name,
const gcs::OptionalItemCallback<rpc::ActorTableData> &callback) override {
return Status::NotImplemented("");
}
Status AsyncCreateActor(const TaskSpecification &task_spec,
const gcs::StatusCallback &callback) override {
return Status::NotImplemented("");
src/ray/gcs/redis_accessor.h
+12
View File
@@ -42,6 +42,12 @@ class RedisLogBasedActorInfoAccessor : public ActorInfoAccessor {
Status AsyncGet(const ActorID &actor_id,
const OptionalItemCallback<ActorTableData> &callback) override;
Status AsyncGetByName(const std::string &name,
const OptionalItemCallback<ActorTableData> &callback) override {
return Status::NotImplemented(
"RedisLogBasedActorInfoAccessor does not support named detached actors.");
}
Status AsyncCreateActor(const TaskSpecification &task_spec,
const StatusCallback &callback) override;
@@ -114,6 +120,12 @@ class RedisActorInfoAccessor : public RedisLogBasedActorInfoAccessor {
Status AsyncGet(const ActorID &actor_id,
const OptionalItemCallback<ActorTableData> &callback) override;
Status AsyncGetByName(const std::string &name,
const OptionalItemCallback<ActorTableData> &callback) override {
return Status::NotImplemented(
"RedisActorInfoAccessor does not support named detached actors.");
}
Status AsyncRegister(const std::shared_ptr<ActorTableData> &data_ptr,
const StatusCallback &callback) override;
src/ray/gcs/test/gcs_test_util.h
+6 -3
View File
@@ -31,6 +31,7 @@ namespace ray {
struct Mocker {
static TaskSpecification GenActorCreationTask(const JobID &job_id,
int max_reconstructions, bool detached,
const std::string &name,
const rpc::Address &owner_address) {
TaskSpecBuilder builder;
rpc::Address empty_address;
@@ -40,13 +41,15 @@ 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);
builder.SetActorCreationTaskSpec(actor_id, max_reconstructions, {}, 1, detached,
name);
return builder.Build();
}
static rpc::CreateActorRequest GenCreateActorRequest(const JobID &job_id,
int max_reconstructions = 0,
bool detached = false) {
bool detached = false,
const std::string name = "") {
rpc::CreateActorRequest request;
rpc::Address owner_address;
if (owner_address.raylet_id().empty()) {
@@ -56,7 +59,7 @@ struct Mocker {
owner_address.set_worker_id(WorkerID::FromRandom().Binary());
}
auto actor_creation_task_spec =
GenActorCreationTask(job_id, max_reconstructions, detached, owner_address);
GenActorCreationTask(job_id, max_reconstructions, detached, name, owner_address);
request.mutable_task_spec()->CopyFrom(actor_creation_task_spec.GetMessage());
return request;
}
src/ray/protobuf/common.proto
+5 -1
View File
@@ -159,8 +159,12 @@ message ActorCreationTaskSpec {
int32 max_concurrency = 5;
// Whether the actor is persistent
bool is_detached = 6;
// Globally-unique name of the actor. Should only be populated when is_detached is true.
string name = 7;
// Whether the actor use async actor calls
bool is_asyncio = 7;
bool is_asyncio = 8;
// Field used for storing application-level extensions to the actor definition.
string extension_data = 9;
}
// Task spec of an actor task.
src/ray/protobuf/gcs.proto
+6 -3
View File
@@ -131,10 +131,12 @@ message ActorTableData {
Address owner_address = 10;
// Whether the actor is persistent.
bool is_detached = 11;
// Name of the actor. Only populated if is_detached is true.
string name = 12;
// Timestamp that the actor is created or reconstructed.
double timestamp = 12;
double timestamp = 13;
// The task specification of this actor's creation task.
TaskSpec task_spec = 13;
TaskSpec task_spec = 14;
}
message ErrorTableData {
@@ -356,7 +358,8 @@ enum ErrorType {
// exposed to user code; it is only used internally to indicate the result of a direct
// call has been placed in plasma.
OBJECT_IN_PLASMA = 4;
// Indicates that an object has been cancelled.
TASK_CANCELLED = 5;
// Inidicates that creating the GCS service failed to create the actor.
ACTOR_CREATION_FAILED = 6;
}
src/ray/protobuf/gcs_service.proto
+18 -2
View File
@@ -49,8 +49,11 @@ service JobInfoGcsService {
}
message GetActorInfoRequest {
// ID of this actor.
// ID of this actor. If actor_id is set, name will not be set.
bytes actor_id = 1;
// Name of the actor. This is only used for detached actors. If name is set,
// actor_id will not be set.
string name = 2;
}
message GetActorInfoReply {
@@ -59,6 +62,17 @@ message GetActorInfoReply {
ActorTableData actor_table_data = 2;
}
message GetNamedActorInfoRequest {
// Name of the actor.
string name = 2;
}
message GetNamedActorInfoReply {
GcsStatus status = 1;
// Data of actor.
ActorTableData actor_table_data = 2;
}
message RegisterActorInfoRequest {
// Data of actor.
ActorTableData actor_table_data = 1;
@@ -110,8 +124,10 @@ message GetActorCheckpointIDReply {
service ActorInfoGcsService {
// Create actor via gcs service
rpc CreateActor(CreateActorRequest) returns (CreateActorReply);
// Get actor data from GCS Service.
// Get actor data from GCS Service by actor id.
rpc GetActorInfo(GetActorInfoRequest) returns (GetActorInfoReply);
// Get actor data from GCS Service by name.
rpc GetNamedActorInfo(GetNamedActorInfoRequest) returns (GetNamedActorInfoReply);
// Register an actor to GCS Service.
rpc RegisterActorInfo(RegisterActorInfoRequest) returns (RegisterActorInfoReply);
// Update actor info in GCS Service.
src/ray/raylet/task_dependency_manager_test.cc
+1 -1
View File
@@ -111,7 +111,7 @@ static inline Task ExampleTask(const std::vector<ObjectID> &arguments,
FunctionDescriptorBuilder::BuildPython("", "", "", ""),
JobID::Nil(), RandomTaskId(), 0, RandomTaskId(), address,
num_returns, {}, {});
builder.SetActorCreationTaskSpec(ActorID::Nil(), 1, {}, 1, false, false);
builder.SetActorCreationTaskSpec(ActorID::Nil(), 1, {}, 1, false, "", false);
for (const auto &arg : arguments) {
builder.AddByRefArg(arg);
}
src/ray/rpc/gcs_server/gcs_rpc_client.h
+4
View File
@@ -102,6 +102,10 @@ class GcsRpcClient {
/// Get actor data from GCS Service.
VOID_GCS_RPC_CLIENT_METHOD(ActorInfoGcsService, GetActorInfo, actor_info_grpc_client_, )
/// Get actor data from GCS Service by name.
VOID_GCS_RPC_CLIENT_METHOD(ActorInfoGcsService, GetNamedActorInfo,
actor_info_grpc_client_, )
/// Register an actor to GCS Service.
VOID_GCS_RPC_CLIENT_METHOD(ActorInfoGcsService, RegisterActorInfo,
actor_info_grpc_client_, )
src/ray/rpc/gcs_server/gcs_rpc_server.h
+5
View File
@@ -102,6 +102,10 @@ class ActorInfoGcsServiceHandler {
GetActorInfoReply *reply,
SendReplyCallback send_reply_callback) = 0;
virtual void HandleGetNamedActorInfo(const GetNamedActorInfoRequest &request,
GetNamedActorInfoReply *reply,
SendReplyCallback send_reply_callback) = 0;
virtual void HandleRegisterActorInfo(const RegisterActorInfoRequest &request,
RegisterActorInfoReply *reply,
SendReplyCallback send_reply_callback) = 0;
@@ -141,6 +145,7 @@ class ActorInfoGrpcService : public GrpcService {
std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories) override {
ACTOR_INFO_SERVICE_RPC_HANDLER(CreateActor);
ACTOR_INFO_SERVICE_RPC_HANDLER(GetActorInfo);
ACTOR_INFO_SERVICE_RPC_HANDLER(GetNamedActorInfo);
ACTOR_INFO_SERVICE_RPC_HANDLER(RegisterActorInfo);
ACTOR_INFO_SERVICE_RPC_HANDLER(UpdateActorInfo);
ACTOR_INFO_SERVICE_RPC_HANDLER(AddActorCheckpoint);
streaming/src/test/queue_tests_base.h
+3 -2
View File
@@ -272,10 +272,11 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
args.emplace_back(TaskArg::PassByValue(
std::make_shared<RayObject>(buffer, nullptr, std::vector<ObjectID>())));
std::string name = "";
ActorCreationOptions actor_options{
max_reconstructions,
/*max_concurrency*/ 1, resources, resources, {},
/*is_detached*/ false, /*is_asyncio*/ false};
/*max_concurrency=*/1, resources, resources, {},
/*is_detached=*/false, name, /*is_asyncio=*/false};
// Create an actor.
ActorID actor_id;