add getActor(name, namespace) api

cpp/include/ray/api.h

@@ -120,6 +120,18 @@ ray::internal::ActorCreator<JavaActorClass> Actor(JavaActorClass func);
 template <typename T>
 boost::optional<ActorHandle<T>> GetActor(const std::string &actor_name);
 
+/// Get a handle to a named actor in the given namespace.
+/// Gets a handle to a named actor with the given name of the given namespace. The actor
+/// must have been created with name specified.
+///
+/// \param[in] actor_name The name of the named actor.
+/// \param[in] namespace The namespace of the actor.
+/// \return An ActorHandle to the actor if the actor of specified name exists in
+/// specifiled namespace or an empty optional object.
+template <typename T>
+boost::optional<ActorHandle<T>> GetActor(const std::string &actor_name,
+ const std::string &ray_namespace);
+
 /// Intentionally exit the current actor.
 /// It is used to disconnect an actor and exit the worker.
 /// \Throws RayException if the current process is a driver or the current worker is not
@@ -271,11 +283,17 @@ inline ray::internal::ActorCreator<F> Actor(F create_func) {
 
 template <typename T>
 boost::optional<ActorHandle<T>> GetActor(const std::string &actor_name) {
+ return GetActor<T>(actor_name, "");
+}
+
+template <typename T>
+boost::optional<ActorHandle<T>> GetActor(const std::string &actor_name,
+ const std::string &ray_namespace) {
  if (actor_name.empty()) {
  return {};
  }
 
- auto actor_id = ray::internal::GetRayRuntime()->GetActorId(actor_name);
+ auto actor_id = ray::internal::GetRayRuntime()->GetActorId(actor_name, ray_namespace);
  if (actor_id.empty()) {
  return {};
  }

cpp/include/ray/api/ray_runtime.h

@@ -78,7 +78,8 @@ class RayRuntime {
  const CallOptions &call_options) = 0;
  virtual void AddLocalReference(const std::string &id) = 0;
  virtual void RemoveLocalReference(const std::string &id) = 0;
- virtual std::string GetActorId(const std::string &actor_name) = 0;
+ virtual std::string GetActorId(const std::string &actor_name,
+ const std::string &ray_namespace) = 0;
  virtual void KillActor(const std::string &str_actor_id, bool no_restart) = 0;
  virtual void ExitActor() = 0;
  virtual ray::PlacementGroup CreatePlacementGroup(

cpp/src/ray/runtime/abstract_ray_runtime.cc

@@ -242,8 +242,9 @@ void AbstractRayRuntime::RemoveLocalReference(const std::string &id) {
  }
 }
 
-std::string AbstractRayRuntime::GetActorId(const std::string &actor_name) {
- auto actor_id = task_submitter_->GetActor(actor_name);
+std::string AbstractRayRuntime::GetActorId(const std::string &actor_name,
+ const std::string &ray_namespace) {
+ auto actor_id = task_submitter_->GetActor(actor_name, ray_namespace);
  if (actor_id.IsNil()) {
  return "";
  }

cpp/src/ray/runtime/abstract_ray_runtime.h

@@ -75,7 +75,7 @@ class AbstractRayRuntime : public RayRuntime {
 
  void RemoveLocalReference(const std::string &id);
 
- std::string GetActorId(const std::string &actor_name);
+ std::string GetActorId(const std::string &actor_name, const std::string &ray_namespace);
 
  void KillActor(const std::string &str_actor_id, bool no_restart);
 

cpp/src/ray/runtime/task/local_mode_task_submitter.cc

@@ -145,7 +145,8 @@ ObjectID LocalModeTaskSubmitter::SubmitActorTask(InvocationSpec &invocation,
  return Submit(invocation, {});
 }
 
-ActorID LocalModeTaskSubmitter::GetActor(const std::string &actor_name) const {
+ActorID LocalModeTaskSubmitter::GetActor(const std::string &actor_name,
+ const std::string &ray_namespace) const {
  absl::MutexLock lock(&named_actors_mutex_);
  auto it = named_actors_.find(actor_name);
  if (it == named_actors_.end()) {

cpp/src/ray/runtime/task/local_mode_task_submitter.h

@@ -38,7 +38,7 @@ class LocalModeTaskSubmitter : public TaskSubmitter {
 
  ObjectID SubmitActorTask(InvocationSpec &invocation, const CallOptions &call_options);
 
- ActorID GetActor(const std::string &actor_name) const;
+ ActorID GetActor(const std::string &actor_name, const std::string &ray_namespace) const;
 
  ray::PlacementGroup CreatePlacementGroup(
  const ray::PlacementGroupCreationOptions &create_options);

cpp/src/ray/runtime/task/native_task_submitter.cc

@@ -145,9 +145,12 @@ ObjectID NativeTaskSubmitter::SubmitActorTask(InvocationSpec &invocation,
  return Submit(invocation, task_options);
 }
 
-ActorID NativeTaskSubmitter::GetActor(const std::string &actor_name) const {
+ActorID NativeTaskSubmitter::GetActor(const std::string &actor_name,
+ const std::string &ray_namespace) const {
  auto &core_worker = CoreWorkerProcess::GetCoreWorker();
- auto pair = core_worker.GetNamedActorHandle(actor_name, "");
+ const std::string ns =
+ ray_namespace.empty() ? core_worker.GetJobConfig().ray_namespace() : ray_namespace;
+ auto pair = core_worker.GetNamedActorHandle(actor_name, ns);
  if (!pair.second.ok()) {
  RAY_LOG(WARNING) << pair.second.message();
  return ActorID::Nil();

cpp/src/ray/runtime/task/native_task_submitter.h

@@ -30,7 +30,7 @@ class NativeTaskSubmitter : public TaskSubmitter {
 
  ObjectID SubmitActorTask(InvocationSpec &invocation, const CallOptions &call_options);
 
- ActorID GetActor(const std::string &actor_name) const;
+ ActorID GetActor(const std::string &actor_name, const std::string &ray_namespace) const;
 
  ray::PlacementGroup CreatePlacementGroup(
  const ray::PlacementGroupCreationOptions &create_options);

cpp/src/ray/runtime/task/task_submitter.h

@@ -38,7 +38,8 @@ class TaskSubmitter {
  virtual ObjectID SubmitActorTask(InvocationSpec &invocation,
  const CallOptions &call_options) = 0;
 
- virtual ActorID GetActor(const std::string &actor_name) const = 0;
+ virtual ActorID GetActor(const std::string &actor_name,
+ const std::string &ray_namespace) const = 0;
 
  virtual ray::PlacementGroup CreatePlacementGroup(
  const ray::PlacementGroupCreationOptions &create_options) = 0;

cpp/src/ray/test/cluster/cluster_mode_test.cc

@@ -480,14 +480,22 @@ TEST(RayClusterModeTest, TaskWithPlacementGroup) {
 }
 
 TEST(RayClusterModeTest, NamespaceTest) {
+ // Create a named actor in namespace `isolated_ns`.
  std::string actor_name = "named_actor";
- ray::ActorHandle<Counter> actor = ray::Actor(RAY_FUNC(Counter::FactoryCreate))
-  .SetName(actor_name, "isolated_ns")
-  .Remote();
+ std::string ns_name = "isolated_ns";
+ ray::ActorHandle<Counter> actor =
+ ray::Actor(RAY_FUNC(Counter::FactoryCreate)).SetName(actor_name, ns_name).Remote();
  auto initialized_obj = actor.Task(&Counter::Initialized).Remote();
  EXPECT_TRUE(*initialized_obj.Get());
+ // It is invisible to job default namespace.
  auto actor_optional = ray::GetActor<Counter>(actor_name);
  EXPECT_TRUE(!actor_optional);
+ // It is visible to the namespace it belongs.
+ actor_optional = ray::GetActor<Counter>(actor_name, ns_name);
+ EXPECT_TRUE(actor_optional);
+ // It is invisible to any other namespaces.
+ actor_optional = ray::GetActor<Counter>(actor_name, "other_ns");
+ EXPECT_TRUE(!actor_optional);
 }
 
 int main(int argc, char **argv) {