From 2677b7100306f2e5689ab6bebdc264a0e37930ae Mon Sep 17 00:00:00 2001 From: Edward Oakes Date: Sat, 9 May 2020 08:58:10 -0500 Subject: [PATCH] Implement named actors using the GCS service (#8328) --- python/ray/_raylet.pxd | 3 +- python/ray/_raylet.pyx | 52 +++++++++----- python/ray/actor.py | 19 +++++- python/ray/includes/common.pxd | 6 +- python/ray/includes/libcoreworker.pxd | 2 + python/ray/includes/ray_config.pxd | 5 ++ python/ray/tests/test_actor_advanced.py | 23 +++++-- python/ray/util/named_actors.py | 16 +++-- src/ray/common/status.cc | 4 +- src/ray/common/status.h | 6 ++ src/ray/common/task/task_util.h | 7 +- src/ray/core_worker/actor_handle.cc | 20 ++++++ src/ray/core_worker/actor_handle.h | 4 ++ src/ray/core_worker/common.h | 7 +- src/ray/core_worker/core_worker.cc | 67 +++++++++++++++++-- src/ray/core_worker/core_worker.h | 7 ++ ...io_ray_runtime_task_NativeTaskSubmitter.cc | 2 + src/ray/core_worker/test/core_worker_test.cc | 15 +++-- .../transport/direct_task_transport.cc | 24 +++---- src/ray/gcs/accessor.h | 9 +++ .../gcs/gcs_client/service_based_accessor.cc | 20 ++++++ .../gcs/gcs_client/service_based_accessor.h | 4 ++ .../gcs/gcs_server/actor_info_handler_impl.cc | 59 ++++++++++++++-- .../gcs/gcs_server/actor_info_handler_impl.h | 4 ++ src/ray/gcs/gcs_server/gcs_actor_manager.cc | 38 +++++++++-- src/ray/gcs/gcs_server/gcs_actor_manager.h | 18 ++++- .../gcs_server/test/gcs_actor_manager_test.cc | 51 +++++++++++++- .../gcs_server/test/gcs_server_test_util.h | 6 ++ src/ray/gcs/redis_accessor.h | 12 ++++ src/ray/gcs/test/gcs_test_util.h | 9 ++- src/ray/protobuf/common.proto | 6 +- src/ray/protobuf/gcs.proto | 9 ++- src/ray/protobuf/gcs_service.proto | 20 +++++- .../raylet/task_dependency_manager_test.cc | 2 +- src/ray/rpc/gcs_server/gcs_rpc_client.h | 4 ++ src/ray/rpc/gcs_server/gcs_rpc_server.h | 5 ++ streaming/src/test/queue_tests_base.h | 5 +- 37 files changed, 483 insertions(+), 87 deletions(-) diff --git a/python/ray/_raylet.pxd b/python/ray/_raylet.pxd index 05b0dfca7..e69b5c452 100644 --- a/python/ray/_raylet.pxd +++ b/python/ray/_raylet.pxd @@ -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: diff --git a/python/ray/_raylet.pyx b/python/ray/_raylet.pyx index 7173860ec..065153ee1 100644 --- a/python/ray/_raylet.pyx +++ b/python/ray/_raylet.pyx @@ -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 diff --git a/python/ray/actor.py b/python/ray/actor.py index 6bc878c4e..4031a1cbf 100644 --- a/python/ray/actor.py +++ b/python/ray/actor.py @@ -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 diff --git a/python/ray/includes/common.pxd b/python/ray/includes/common.pxd index 01542d756..0cd72bfdb 100644 --- a/python/ray/includes/common.pxd +++ b/python/ray/includes/common.pxd @@ -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": diff --git a/python/ray/includes/libcoreworker.pxd b/python/ray/includes/libcoreworker.pxd index a2698ee51..7a85cf4a7 100644 --- a/python/ray/includes/libcoreworker.pxd +++ b/python/ray/includes/libcoreworker.pxd @@ -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) diff --git a/python/ray/includes/ray_config.pxd b/python/ray/includes/ray_config.pxd index e4be33b41..0684e6374 100644 --- a/python/ray/includes/ray_config.pxd +++ b/python/ray/includes/ray_config.pxd @@ -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 diff --git a/python/ray/tests/test_actor_advanced.py b/python/ray/tests/test_actor_advanced.py index 7db745ba3..7a18fab59 100644 --- a/python/ray/tests/test_actor_advanced.py +++ b/python/ray/tests/test_actor_advanced.py @@ -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" diff --git a/python/ray/util/named_actors.py b/python/ray/util/named_actors.py index e74e31bba..f02839171 100644 --- a/python/ray/util/named_actors.py +++ b/python/ray/util/named_actors.py @@ -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 diff --git a/src/ray/common/status.cc b/src/ray/common/status.cc index e3e954546..cba252966 100644 --- a/src/ray/common/status.cc +++ b/src/ray/common/status.cc @@ -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; diff --git a/src/ray/common/status.h b/src/ray/common/status.h index f119e7d9c..348b4e5e3 100644 --- a/src/ray/common/status.h +++ b/src/ray/common/status.h @@ -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. diff --git a/src/ray/common/task/task_util.h b/src/ray/common/task/task_util.h index b1100e6bc..bc76634ce 100644 --- a/src/ray/common/task/task_util.h +++ b/src/ray/common/task/task_util.h @@ -103,7 +103,8 @@ class TaskSpecBuilder { TaskSpecBuilder &SetActorCreationTaskSpec( const ActorID &actor_id, uint64_t max_reconstructions = 0, const std::vector &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; } diff --git a/src/ray/core_worker/actor_handle.cc b/src/ray/core_worker/actor_handle.cc index 1907bfef2..67495df4a 100644 --- a/src/ray/core_worker/actor_handle.cc +++ b/src/ray/core_worker/actor_handle.cc @@ -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. diff --git a/src/ray/core_worker/actor_handle.h b/src/ray/core_worker/actor_handle.h index 5989aff6e..078d2ee21 100644 --- a/src/ray/core_worker/actor_handle.h +++ b/src/ray/core_worker/actor_handle.h @@ -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()); } diff --git a/src/ray/core_worker/common.h b/src/ray/core_worker/common.h index 40fc0278b..be23179ae 100644 --- a/src/ray/core_worker/common.h +++ b/src/ray/core_worker/common.h @@ -115,13 +115,14 @@ struct ActorCreationOptions { const std::unordered_map &resources, const std::unordered_map &placement_resources, const std::vector &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; }; diff --git a/src/ray/core_worker/core_worker.cc b/src/ray/core_worker/core_worker.cc index 7739ee867..13a095949 100644 --- a/src/ray/core_worker/core_worker.cc +++ b/src/ray/core_worker/core_worker.cc @@ -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 ready = std::make_shared(false); + std::shared_ptr m = std::make_shared(); + std::shared_ptr cv = + std::make_shared(); + std::unique_lock lk(*m); + RAY_CHECK_OK(gcs_client_->Actors().AsyncGetByName( + name, [this, &actor_id, name, ready, m, cv]( + Status status, const boost::optional &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(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 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_; diff --git a/src/ray/core_worker/core_worker.h b/src/ray/core_worker/core_worker.h index 5b6ea601b..afed1c355 100644 --- a/src/ray/core_worker/core_worker.h +++ b/src/ray/core_worker/core_worker.h @@ -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 diff --git a/src/ray/core_worker/lib/java/io_ray_runtime_task_NativeTaskSubmitter.cc b/src/ray/core_worker/lib/java/io_ray_runtime_task_NativeTaskSubmitter.cc index bb94d153e..8749519ba 100644 --- a/src/ray/core_worker/lib/java/io_ray_runtime_task_NativeTaskSubmitter.cc +++ b/src/ray/core_worker/lib/java/io_ray_runtime_task_NativeTaskSubmitter.cc @@ -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(max_reconstructions), static_cast(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; } diff --git a/src/ray/core_worker/test/core_worker_test.cc b/src/ray/core_worker/test/core_worker_test.cc index 0d95aabe7..01f54e93f 100644 --- a/src/ray/core_worker/test/core_worker_test.cc +++ b/src/ray/core_worker/test/core_worker_test.cc @@ -70,10 +70,11 @@ ActorID CreateActorHelper(std::unordered_map &resources, args.emplace_back(TaskArg::PassByValue( std::make_shared(buffer, nullptr, std::vector()))); - 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(buffer, nullptr, std::vector()))); std::unordered_map 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(), diff --git a/src/ray/core_worker/transport/direct_task_transport.cc b/src/ray/core_worker/transport/direct_task_transport.cc index ba2104160..0c29d99af 100644 --- a/src/ray/core_worker/transport/direct_task_transport.cc +++ b/src/ray/core_worker/transport/direct_task_transport.cc @@ -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; } diff --git a/src/ray/gcs/accessor.h b/src/ray/gcs/accessor.h index e4988f729..72f11a870 100644 --- a/src/ray/gcs/accessor.h +++ b/src/ray/gcs/accessor.h @@ -47,6 +47,15 @@ class ActorInfoAccessor { virtual Status AsyncGet(const ActorID &actor_id, const OptionalItemCallback &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 &callback) = 0; + /// Create an actor to GCS asynchronously. /// /// \param task_spec The specification for the actor creation task. diff --git a/src/ray/gcs/gcs_client/service_based_accessor.cc b/src/ray/gcs/gcs_client/service_based_accessor.cc index 1e55b2a2c..eabee5b96 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.cc +++ b/src/ray/gcs/gcs_client/service_based_accessor.cc @@ -107,6 +107,26 @@ Status ServiceBasedActorInfoAccessor::AsyncGet( return Status::OK(); } +Status ServiceBasedActorInfoAccessor::AsyncGetByName( + const std::string &name, const OptionalItemCallback &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); diff --git a/src/ray/gcs/gcs_client/service_based_accessor.h b/src/ray/gcs/gcs_client/service_based_accessor.h index 117ebc30f..0863a0799 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.h +++ b/src/ray/gcs/gcs_client/service_based_accessor.h @@ -61,6 +61,10 @@ class ServiceBasedActorInfoAccessor : public ActorInfoAccessor { Status AsyncGet(const ActorID &actor_id, const OptionalItemCallback &callback) override; + Status AsyncGetByName( + const std::string &name, + const OptionalItemCallback &callback) override; + Status AsyncCreateActor(const TaskSpecification &task_spec, const StatusCallback &callback) override; diff --git a/src/ray/gcs/gcs_server/actor_info_handler_impl.cc b/src/ray/gcs/gcs_server/actor_info_handler_impl.cc index 68866f4b6..e7e8499bf 100644 --- a/src/ray/gcs/gcs_server/actor_info_handler_impl.cc +++ b/src/ray/gcs/gcs_server/actor_info_handler_impl.cc @@ -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 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 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 &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 &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) { diff --git a/src/ray/gcs/gcs_server/actor_info_handler_impl.h b/src/ray/gcs/gcs_server/actor_info_handler_impl.h index 4cbefd4af..0c47b97a9 100644 --- a/src/ray/gcs/gcs_server/actor_info_handler_impl.h +++ b/src/ray/gcs/gcs_server/actor_info_handler_impl.h @@ -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; diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.cc b/src/ray/gcs/gcs_server/gcs_actor_manager.cc index 8c7a21e49..008330045 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_actor_manager.cc @@ -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 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)> 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(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(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, diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.h b/src/ray/gcs/gcs_server/gcs_actor_manager.h index c16803979..b9a8ad897 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.h +++ b/src/ray/gcs/gcs_server/gcs_actor_manager.h @@ -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> registered_actors_; + /// Maps detached actor names to their actor ID for lookups by name. + absl::flat_hash_map named_actors_; /// The pending actors which will not be scheduled until there's a resource change. std::vector> pending_actors_; /// Map contains the relationship of node and created actors. Each node ID diff --git a/src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc b/src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc index 8ec69a203..6c0de63ce 100644 --- a/src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc +++ b/src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc @@ -50,10 +50,11 @@ TEST_F(GcsActorManagerTest, TestBasic) { auto job_id = JobID::FromInt(1); auto create_actor_request = Mocker::GenCreateActorRequest(job_id); std::vector> finished_actors; - gcs_actor_manager_.RegisterActor( + Status status = gcs_actor_manager_.RegisterActor( create_actor_request, [&finished_actors](std::shared_ptr 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> finished_actors; - gcs_actor_manager_.RegisterActor( + Status status = gcs_actor_manager_.RegisterActor( create_actor_request, [&finished_actors](std::shared_ptr 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> finished_actors; - gcs_actor_manager_.RegisterActor( + Status status = gcs_actor_manager_.RegisterActor( create_actor_request, [&finished_actors](std::shared_ptr 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 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 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 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 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) { diff --git a/src/ray/gcs/gcs_server/test/gcs_server_test_util.h b/src/ray/gcs/gcs_server/test/gcs_server_test_util.h index 53ad39373..e426eed6f 100644 --- a/src/ray/gcs/gcs_server/test/gcs_server_test_util.h +++ b/src/ray/gcs/gcs_server/test/gcs_server_test_util.h @@ -179,6 +179,12 @@ struct GcsServerMocker { return Status::NotImplemented(""); } + Status AsyncGetByName( + const std::string &name, + const gcs::OptionalItemCallback &callback) override { + return Status::NotImplemented(""); + } + Status AsyncCreateActor(const TaskSpecification &task_spec, const gcs::StatusCallback &callback) override { return Status::NotImplemented(""); diff --git a/src/ray/gcs/redis_accessor.h b/src/ray/gcs/redis_accessor.h index 7561c2f0c..348e85fe1 100644 --- a/src/ray/gcs/redis_accessor.h +++ b/src/ray/gcs/redis_accessor.h @@ -42,6 +42,12 @@ class RedisLogBasedActorInfoAccessor : public ActorInfoAccessor { Status AsyncGet(const ActorID &actor_id, const OptionalItemCallback &callback) override; + Status AsyncGetByName(const std::string &name, + const OptionalItemCallback &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 &callback) override; + Status AsyncGetByName(const std::string &name, + const OptionalItemCallback &callback) override { + return Status::NotImplemented( + "RedisActorInfoAccessor does not support named detached actors."); + } + Status AsyncRegister(const std::shared_ptr &data_ptr, const StatusCallback &callback) override; diff --git a/src/ray/gcs/test/gcs_test_util.h b/src/ray/gcs/test/gcs_test_util.h index 13b0d18d4..572b5b4ea 100644 --- a/src/ray/gcs/test/gcs_test_util.h +++ b/src/ray/gcs/test/gcs_test_util.h @@ -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; } diff --git a/src/ray/protobuf/common.proto b/src/ray/protobuf/common.proto index 02f907402..cedeeaff6 100644 --- a/src/ray/protobuf/common.proto +++ b/src/ray/protobuf/common.proto @@ -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. diff --git a/src/ray/protobuf/gcs.proto b/src/ray/protobuf/gcs.proto index bbdf50039..78fc792ad 100644 --- a/src/ray/protobuf/gcs.proto +++ b/src/ray/protobuf/gcs.proto @@ -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; } diff --git a/src/ray/protobuf/gcs_service.proto b/src/ray/protobuf/gcs_service.proto index dd356b4cf..716168713 100644 --- a/src/ray/protobuf/gcs_service.proto +++ b/src/ray/protobuf/gcs_service.proto @@ -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. diff --git a/src/ray/raylet/task_dependency_manager_test.cc b/src/ray/raylet/task_dependency_manager_test.cc index 1b3c67bbe..572f1ae9e 100644 --- a/src/ray/raylet/task_dependency_manager_test.cc +++ b/src/ray/raylet/task_dependency_manager_test.cc @@ -111,7 +111,7 @@ static inline Task ExampleTask(const std::vector &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); } diff --git a/src/ray/rpc/gcs_server/gcs_rpc_client.h b/src/ray/rpc/gcs_server/gcs_rpc_client.h index 84427c48f..c90ed6aa8 100644 --- a/src/ray/rpc/gcs_server/gcs_rpc_client.h +++ b/src/ray/rpc/gcs_server/gcs_rpc_client.h @@ -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_, ) diff --git a/src/ray/rpc/gcs_server/gcs_rpc_server.h b/src/ray/rpc/gcs_server/gcs_rpc_server.h index e30828169..02f3738dd 100644 --- a/src/ray/rpc/gcs_server/gcs_rpc_server.h +++ b/src/ray/rpc/gcs_server/gcs_rpc_server.h @@ -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> *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); diff --git a/streaming/src/test/queue_tests_base.h b/streaming/src/test/queue_tests_base.h index dd03f1c1a..5445ba690 100644 --- a/streaming/src/test/queue_tests_base.h +++ b/streaming/src/test/queue_tests_base.h @@ -272,10 +272,11 @@ class StreamingQueueTestBase : public ::testing::TestWithParam { args.emplace_back(TaskArg::PassByValue( std::make_shared(buffer, nullptr, std::vector()))); + 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;