hiro/ray
1
0
Fork 0
mirror of https://github.com/vale981/ray synced 2025-03-04 17:41:43 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

[Java] Remove global named actor and global pg (#20135)

Browse source 
This PR removes global named actor and global PGs.

I believe these APIs are not used widely in OSS.
CPP part is not included in this PR.
@kfstorm @clay4444 @raulchen Please take a look if this change is reasonable.


IMPORTANT NOTE: This is a Java API change and will lead backward incompatibility in Java global named actor and global PG usage.

CPP part is not included in this PR.
INCLUDES:

 Remove setGlobalName() and getGlobalActor() APIs.
 Remove getGlobalPlacementGroup() and setGlobalPG
 Add getActor(name, namespace) API
 Add getPlacementGroup(name, namespace) API
 Update doc pages.
This commit is contained in:
Qing Wang 2021-11-15 16:28:53 +08:00 • committed by GitHub
parent a4f72c6606
commit 1172195571
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
29 changed files with 127 additions and 340 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cpp/src/ray/runtime/task/task_submitter.h
View file

@ -49,12 +49,14 @@ class TaskSubmitter {
return true;
}
/// TODO(qwang): Remove this.
std::string GetFullName(bool global, const std::string &name) const {
if (name.empty()) {
return "";
}
return global ? name : GetCurrentJobID().Hex() + "-" + name;
}
virtual JobID GetCurrentJobID() const = 0;
};
} // namespace internal

52
doc/source/actors.rst
View file

@ -529,26 +529,13 @@ exist. See :ref:`actor-lifetimes` for more details.
.. code-block:: java
// Create an actor with a globally unique name
ActorHandle<Counter> counter = Ray.actor(Counter::new).setGlobalName("some_name").remote();
// Create an actor with a name.
ActorHandle<Counter> counter = Ray.actor(Counter::new).setName("some_name").remote();
...
// Retrieve the actor later somewhere
Optional<ActorHandle<Counter>> counter = Ray.getGlobalActor("some_name");
Assert.assertTrue(counter.isPresent());
We also support non-global named actors in Java, which means that the actor name is only valid within the job and the actor cannot be accessed from another job.
.. code-block:: java
// Create an actor with a job-scope-unique name
ActorHandle<Counter> counter = Ray.actor(Counter::new).setName("some_name_in_job").remote();
...
// Retrieve the actor later somewhere in the same job
Optional<ActorHandle<Counter>> counter = Ray.getActor("some_name_in_job");
Optional<ActorHandle<Counter>> counter = Ray.getActor("some_name");
Assert.assertTrue(counter.isPresent());
.. group-tab:: C++
@ -579,7 +566,8 @@ exist. See :ref:`actor-lifetimes` for more details.
Named actors are only accessible in the same namespace.
.. code-block:: python
.. tabs::
.. code-tab:: python
import ray
@ -602,6 +590,36 @@ exist. See :ref:`actor-lifetimes` for more details.
# Job 3 connects to the original "colors" namespace
ray.init(address="auto", namespace="colors")
# This returns the "orange" actor we created in the first job.
ray.get_actor("orange")
.. code-tab:: java
import ray
class Actor {
}
// Driver1.java
// Job 1 creates an actor, "orange" in the "colors" namespace.
System.setProperty("ray.job.namespace", "colors");
Ray.init();
Ray.actor(Actor::new).setName("orange").remote();
// Driver2.java
// Job 2 is now connecting to a different namespace.
System.setProperty("ray.job.namespace", "fruits");
Ray.init();
// This fails because "orange" was defined in the "colors" namespace.
Optional<ActorHandle<Actor>> actor = Ray.getActor("orange");
Assert.assertFalse(actor.isPresent()); // actor.isPresent() is false.
// Driver3.java
System.setProperty("ray.job.namespace", "colors");
Ray.init();
// This returns the "orange" actor we created in the first job.
Optional<ActorHandle<Actor>> actor = Ray.getActor("orange");
Assert.assertTrue(actor.isPresent()); // actor.isPresent() is true.
.. _actor-lifetimes:

30
doc/source/placement-group.rst
View file

@ -610,7 +610,7 @@ See :ref:`placement-group-lifetimes` for more details.
.. code-block:: java
// Create a placement group with a globally unique name.
// Create a placement group with a unique name.
Map<String, Double> bundle = ImmutableMap.of("CPU", 1.0);
List<Map<String, Double>> bundles = ImmutableList.of(bundle);
@ -618,7 +618,7 @@ See :ref:`placement-group-lifetimes` for more details.
new PlacementGroupCreationOptions.Builder()
.setBundles(bundles)
.setStrategy(PlacementStrategy.STRICT_SPREAD)
.setGlobalName("global_name")
.setName("global_name")
.build();
PlacementGroup pg = PlacementGroups.createPlacementGroup(options);
@ -627,31 +627,7 @@ See :ref:`placement-group-lifetimes` for more details.
...
// Retrieve the placement group later somewhere.
PlacementGroup group = PlacementGroups.getGlobalPlacementGroup("global_name");
Assert.assertNotNull(group);
We also support non-global named placement group in Java, which means that the placement group name is only valid within the job and cannot be accessed from another job.
.. code-block:: java
// Create a placement group with a job-scope-unique name.
Map<String, Double> bundle = ImmutableMap.of("CPU", 1.0);
List<Map<String, Double>> bundles = ImmutableList.of(bundle);
PlacementGroupCreationOptions options =
new PlacementGroupCreationOptions.Builder()
.setBundles(bundles)
.setStrategy(PlacementStrategy.STRICT_SPREAD)
.setName("non_global_name")
.build();
PlacementGroup pg = PlacementGroups.createPlacementGroup(options);
pg.wait(60);
...
// Retrieve the placement group later somewhere in the same job.
PlacementGroup group = PlacementGroups.getPlacementGroup("non_global_name");
PlacementGroup group = PlacementGroups.getPlacementGroup("global_name");
Assert.assertNotNull(group);
.. group-tab:: C++

11
java/api/src/main/java/io/ray/api/PlacementGroups.java
View file

@ -32,23 +32,24 @@ public class PlacementGroups {
}
/**
* Get a placement group by placement group name from current job.
* Get a placement group by placement group name from the current namespace.
*
* @param name The placement group name.
* @return The placement group.
*/
public static PlacementGroup getPlacementGroup(String name) {
return Ray.internal().getPlacementGroup(name, false);
return Ray.internal().getPlacementGroup(name, null);
}
/**
* Get a placement group by placement group name from all jobs.
* Get a placement group by placement group name from the given namespace.
*
* @param name The placement group name.
* @param namespace The namespace of the placement group.
* @return The placement group.
*/
public static PlacementGroup getGlobalPlacementGroup(String name) {
return Ray.internal().getPlacementGroup(name, true);
public static PlacementGroup getPlacementGroup(String name, String namespace) {
return Ray.internal().getPlacementGroup(name, namespace);
}
/**

29
java/api/src/main/java/io/ray/api/Ray.java
View file

@ -151,33 +151,34 @@ public final class Ray extends RayCall {
}
/**
* Get a handle to a named actor of current job.
* Get a handle to a named actor in current namespace.
*
* <p>Gets a handle to a named actor with the given name. The actor must have been created with
* name specified.
* <p>Gets a handle to a named actor with the given name of current namespace. The actor must have
* been created with name specified.
*
* @param name The name of the named actor.
* @return an ActorHandle to the actor if the actor of specified name exists or an
* Optional.empty()
* @return an ActorHandle to the actor if the actor of specified name exists in current namespace
* or an Optional.empty()
* @throws RayException An exception is raised if timed out.
*/
public static <T extends BaseActorHandle> Optional<T> getActor(String name) {
return internal().getActor(name, false);
return internal().getActor(name, null);
}
/**
* Get a handle to a global named actor.
* Get a handle to a named actor in the given namespace.
*
* <p>Gets a handle to a global named actor with the given name. The actor must have been created
* with global name specified.
* <p>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 name The global name of the named actor.
* @return an ActorHandle to the actor if the actor of specified name exists or an
* Optional.empty()
* @param name The name of the named actor.
* @param namespace The namespace of the actor.
* @return an ActorHandle to the actor if the actor of specified name exists in current namespace
* or an Optional.empty()
* @throws RayException An exception is raised if timed out.
*/
public static <T extends BaseActorHandle> Optional<T> getGlobalActor(String name) {
return internal().getActor(name, true);
public static <T extends BaseActorHandle> Optional<T> getActor(String name, String namespace) {
return internal().getActor(name, namespace);
}
/**

14
java/api/src/main/java/io/ray/api/call/BaseActorCreator.java
View file

@ -27,20 +27,6 @@ public class BaseActorCreator<T extends BaseActorCreator> {
return self();
}
/**
* Set the name of this actor. This actor will be accessible from all jobs by this name via {@link
* Ray#getGlobalActor(java.lang.String)}. If you want to create a named actor that is only
* accessible from this job, use {@link BaseActorCreator#setName(java.lang.String)} instead.
*
* @param name The name of the named actor.
* @return self
* @see io.ray.api.options.ActorCreationOptions.Builder#setGlobalName(String)
*/
public T setGlobalName(String name) {
builder.setGlobalName(name);
return self();
}
/**
* Set a custom resource requirement to reserve for the lifetime of this actor. This method can be
* called multiple times. If the same resource is set multiple times, the latest quantity will be

26
java/api/src/main/java/io/ray/api/options/ActorCreationOptions.java
View file

@ -10,7 +10,6 @@ import java.util.Map;
/** The options for creating actor. */
public class ActorCreationOptions extends BaseTaskOptions {
public final boolean global;
public final String name;
public final int maxRestarts;
public final List<String> jvmOptions;
@ -20,7 +19,6 @@ public class ActorCreationOptions extends BaseTaskOptions {
public final List<ConcurrencyGroup> concurrencyGroups;
private ActorCreationOptions(
boolean global,
String name,
Map<String, Double> resources,
int maxRestarts,
@ -30,7 +28,6 @@ public class ActorCreationOptions extends BaseTaskOptions {
int bundleIndex,
List<ConcurrencyGroup> concurrencyGroups) {
super(resources);
this.global = global;
this.name = name;
this.maxRestarts = maxRestarts;
this.jvmOptions = jvmOptions;
@ -42,7 +39,6 @@ public class ActorCreationOptions extends BaseTaskOptions {
/** The inner class for building ActorCreationOptions. */
public static class Builder {
private boolean global;
private String name;
private Map<String, Double> resources = new HashMap<>();
private int maxRestarts = 0;
@ -53,30 +49,15 @@ public class ActorCreationOptions extends BaseTaskOptions {
private List<ConcurrencyGroup> concurrencyGroups = new ArrayList<>();
/**
* Set the actor name of a named actor. This named actor is only accessible from this job by
* this name via {@link Ray#getActor(java.lang.String)}. If you want create a named actor that
* is accessible from all jobs, use {@link Builder#setGlobalName(java.lang.String)} instead.
* Set the actor name of a named actor. This named actor is accessible in this namespace by this
* name via {@link Ray#getActor(java.lang.String)} and in other namespaces via {@link
* Ray#getActor(java.lang.String, java.lang.String)}.
*
* @param name The name of the named actor.
* @return self
*/
public Builder setName(String name) {
this.name = name;
this.global = false;
return this;
}
/**
* Set the name of this actor. This actor will be accessible from all jobs by this name via
* {@link Ray#getGlobalActor(java.lang.String)}. If you want to create a named actor that is
* only accessible from this job, use {@link Builder#setName(java.lang.String)} instead.
*
* @param name The name of the named actor.
* @return self
*/
public Builder setGlobalName(String name) {
this.name = name;
this.global = true;
return this;
}
@ -166,7 +147,6 @@ public class ActorCreationOptions extends BaseTaskOptions {
public ActorCreationOptions build() {
return new ActorCreationOptions(
global,
name,
resources,
maxRestarts,

33
java/api/src/main/java/io/ray/api/options/PlacementGroupCreationOptions.java
View file

@ -7,13 +7,12 @@ import java.util.Map;
/** The options for creating placement group. */
public class PlacementGroupCreationOptions {
public final boolean global;
public final String name;
public final List<Map<String, Double>> bundles;
public final PlacementStrategy strategy;
public PlacementGroupCreationOptions(
boolean global, String name, List<Map<String, Double>> bundles, PlacementStrategy strategy) {
String name, List<Map<String, Double>> bundles, PlacementStrategy strategy) {
if (bundles == null || bundles.isEmpty()) {
throw new IllegalArgumentException(
"`Bundles` must be specified when creating a new placement group.");
@ -30,7 +29,6 @@ public class PlacementGroupCreationOptions {
throw new IllegalArgumentException(
"`PlacementStrategy` must be specified when creating a new placement group.");
}
this.global = global;
this.name = name;
this.bundles = bundles;
this.strategy = strategy;
@ -38,16 +36,14 @@ public class PlacementGroupCreationOptions {
/** The inner class for building PlacementGroupCreationOptions. */
public static class Builder {
private boolean global;
private String name;
private List<Map<String, Double>> bundles;
private PlacementStrategy strategy;
/**
* Set the name of a named placement group. This named placement group is only accessible from
* this job by this name via {@link Ray#getPlacementGroup(java.lang.String)}. If you want to
* create a named placement group that is accessible from all jobs, use {@link
* Builder#setGlobalName(java.lang.String)} instead.
* Set the name of a named placement group. This named placement group is accessible in this
* namespace by this name via {@link Ray#getPlacementGroup(java.lang.String)} or in other
* namespaces via {@link PlacementGroups#getPlacementGroup(java.lang.String, java.lang.String)}.
*
* @param name The name of the named placement group.
* @return self
@ -57,25 +53,6 @@ public class PlacementGroupCreationOptions {
throw new IllegalArgumentException("Repeated assignment of the name is not allowed!");
}
this.name = name;
this.global = false;
return this;
}
/**
* Set the name of a named placement group. This placement group can be accessed by all jobs
* with this name via {@link Ray#getGlobalPlacementGroup(java.lang.String)}. If you want to
* create a named placement group that is only accessible from this job, use {@link
* Builder#setName(java.lang.String)} instead.
*
* @param name The name of the named placement group.
* @return self
*/
public Builder setGlobalName(String name) {
if (this.name != null) {
throw new IllegalArgumentException("Repeated assignment of the name is not allowed!");
}
this.name = name;
this.global = true;
return this;
}
@ -104,7 +81,7 @@ public class PlacementGroupCreationOptions {
}
public PlacementGroupCreationOptions build() {
return new PlacementGroupCreationOptions(global, name, bundles, strategy);
return new PlacementGroupCreationOptions(name, bundles, strategy);
}
}
}

8
java/api/src/main/java/io/ray/api/runtime/RayRuntime.java
View file

@ -99,10 +99,10 @@ public interface RayRuntime {
* name specified.
*
* @param name The name of the named actor.
* @param global Whether the named actor is global.
* @param namespace The namespace of the actor.
* @return ActorHandle to the actor.
*/
<T extends BaseActorHandle> Optional<T> getActor(String name, boolean global);
<T extends BaseActorHandle> Optional<T> getActor(String name, String namespace);
/**
* Kill the actor immediately.
@ -229,10 +229,10 @@ public interface RayRuntime {
* Get a placement group by name.
*
* @param name The name of the placement group.
* @param global Whether the named placement group is global.
* @param namespace The namespace of the placement group.
* @return The placement group.
*/
PlacementGroup getPlacementGroup(String name, boolean global);
PlacementGroup getPlacementGroup(String name, String namespace);
/**
* Get all placement groups in this cluster.

6
java/runtime/src/main/java/io/ray/runtime/AbstractRayRuntime.java
View file

@ -211,8 +211,10 @@ public abstract class AbstractRayRuntime implements RayRuntimeInternal {
}
@Override
public PlacementGroup getPlacementGroup(String name, boolean global) {
return gcsClient.getPlacementGroupInfo(name, global);
public PlacementGroup getPlacementGroup(String name, String namespace) {
return namespace == null
? gcsClient.getPlacementGroupInfo(name, runtimeContext.getNamespace())
: gcsClient.getPlacementGroupInfo(name, namespace);
}
@Override

4
java/runtime/src/main/java/io/ray/runtime/RayDevRuntime.java
View file

@ -78,8 +78,8 @@ public class RayDevRuntime extends AbstractRayRuntime {
@SuppressWarnings("unchecked")
@Override
public <T extends BaseActorHandle> Optional<T> getActor(String name, boolean global) {
return (Optional<T>) ((LocalModeTaskSubmitter) taskSubmitter).getActor(name, global);
public <T extends BaseActorHandle> Optional<T> getActor(String name, String namespace) {
return (Optional<T>) ((LocalModeTaskSubmitter) taskSubmitter).getActor(name);
}
@Override

6
java/runtime/src/main/java/io/ray/runtime/RayNativeRuntime.java
View file

@ -187,11 +187,11 @@ public final class RayNativeRuntime extends AbstractRayRuntime {
@SuppressWarnings("unchecked")
@Override
public <T extends BaseActorHandle> Optional<T> getActor(String name, boolean global) {
public <T extends BaseActorHandle> Optional<T> getActor(String name, String namespace) {
if (name.isEmpty()) {
return Optional.empty();
}
byte[] actorIdBytes = nativeGetActorIdOfNamedActor(name, global);
byte[] actorIdBytes = nativeGetActorIdOfNamedActor(name, namespace);
ActorId actorId = ActorId.fromBytes(actorIdBytes);
if (actorId.isNil()) {
return Optional.empty();
@ -264,7 +264,7 @@ public final class RayNativeRuntime extends AbstractRayRuntime {
private static native void nativeKillActor(byte[] actorId, boolean noRestart);
private static native byte[] nativeGetActorIdOfNamedActor(String actorName, boolean global);
private static native byte[] nativeGetActorIdOfNamedActor(String actorName, String namespace);
private static native void nativeSetCoreWorker(byte[] workerId);

6
java/runtime/src/main/java/io/ray/runtime/gcs/GcsClient.java
View file

@ -43,11 +43,11 @@ public class GcsClient {
* Get a placement group by name.
*
* @param name Name of the placement group.
* @param global Whether the named placement group is global.
* @param namespace The namespace of the placement group.
* @return The placement group.
*/
public PlacementGroup getPlacementGroupInfo(String name, boolean global) {
byte[] result = globalStateAccessor.getPlacementGroupInfo(name, global);
public PlacementGroup getPlacementGroupInfo(String name, String namespace) {
byte[] result = globalStateAccessor.getPlacementGroupInfo(name, namespace);
return result == null ? null : PlacementGroupUtils.generatePlacementGroupFromByteArray(result);
}

10
java/runtime/src/main/java/io/ray/runtime/gcs/GlobalStateAccessor.java
View file

@ -1,11 +1,9 @@
package io.ray.runtime.gcs;
import com.google.common.base.Preconditions;
import io.ray.api.Ray;
import io.ray.api.id.ActorId;
import io.ray.api.id.PlacementGroupId;
import io.ray.api.id.UniqueId;
import io.ray.runtime.RayRuntimeInternal;
import java.util.List;
/** `GlobalStateAccessor` is used for accessing information from GCS. */
@ -94,12 +92,10 @@ public class GlobalStateAccessor {
}
}
public byte[] getPlacementGroupInfo(String name, boolean global) {
public byte[] getPlacementGroupInfo(String name, String namespace) {
synchronized (GlobalStateAccessor.class) {
validateGlobalStateAccessorPointer();
RayRuntimeInternal runtime = (RayRuntimeInternal) Ray.internal();
return nativeGetPlacementGroupInfoByName(
globalStateAccessorNativePointer, name, runtime.getRayConfig().namespace, global);
return nativeGetPlacementGroupInfoByName(globalStateAccessorNativePointer, name, namespace);
}
}
@ -175,7 +171,7 @@ public class GlobalStateAccessor {
private native byte[] nativeGetPlacementGroupInfo(long nativePtr, byte[] placementGroupId);
private native byte[] nativeGetPlacementGroupInfoByName(
long nativePtr, String name, String namespace, boolean global);
long nativePtr, String name, String namespace);
private native List<byte[]> nativeGetAllPlacementGroupInfo(long nativePtr);

16
java/runtime/src/main/java/io/ray/runtime/task/LocalModeTaskSubmitter.java
View file

@ -5,7 +5,6 @@ import com.google.common.collect.ImmutableList;
import com.google.protobuf.ByteString;
import io.ray.api.ActorHandle;
import io.ray.api.BaseActorHandle;
import io.ray.api.Ray;
import io.ray.api.id.ActorId;
import io.ray.api.id.ObjectId;
import io.ray.api.id.PlacementGroupId;
@ -308,13 +307,10 @@ public class LocalModeTaskSubmitter implements TaskSubmitter {
new LocalModeActorHandle(actorId, getReturnIds(taskSpec).get(0));
actorHandles.put(actorId, actorHandle.copy());
if (StringUtils.isNotBlank(options.name)) {
String fullName =
options.global
? options.name
: String.format("%s-%s", Ray.getRuntimeContext().getCurrentJobId(), options.name);
Preconditions.checkArgument(
!namedActors.containsKey(fullName), String.format("Actor of name %s exists", fullName));
namedActors.put(fullName, actorHandle);
!namedActors.containsKey(options.name),
String.format("Actor of name %s exists", options.name));
namedActors.put(options.name, actorHandle);
}
return actorHandle;
}
@ -381,10 +377,8 @@ public class LocalModeTaskSubmitter implements TaskSubmitter {
return actorHandles.get(actorId).copy();
}
public Optional<BaseActorHandle> getActor(String name, boolean global) {
String fullName =
global ? name : String.format("%s-%s", Ray.getRuntimeContext().getCurrentJobId(), name);
ActorHandle actorHandle = namedActors.get(fullName);
public Optional<BaseActorHandle> getActor(String name) {
ActorHandle actorHandle = namedActors.get(name);
if (null == actorHandle) {
return Optional.empty();
}

9
java/runtime/src/main/java/io/ray/runtime/task/NativeTaskSubmitter.java
View file

@ -51,8 +51,7 @@ public class NativeTaskSubmitter implements TaskSubmitter {
}
if (StringUtils.isNotBlank(options.name)) {
Optional<BaseActorHandle> actor =
options.global ? Ray.getGlobalActor(options.name) : Ray.getActor(options.name);
Optional<BaseActorHandle> actor = Ray.getActor(options.name);
Preconditions.checkArgument(
!actor.isPresent(), String.format("Actor of name %s exists", options.name));
}
@ -92,11 +91,7 @@ public class NativeTaskSubmitter implements TaskSubmitter {
@Override
public PlacementGroup createPlacementGroup(PlacementGroupCreationOptions creationOptions) {
if (StringUtils.isNotBlank(creationOptions.name)) {
PlacementGroup placementGroup =
creationOptions.global
? PlacementGroups.getGlobalPlacementGroup(creationOptions.name)
: PlacementGroups.getPlacementGroup(creationOptions.name);
PlacementGroup placementGroup = PlacementGroups.getPlacementGroup(creationOptions.name);
Preconditions.checkArgument(
placementGroup == null,
String.format("Placement group with name %s exists!", creationOptions.name));

27
java/test/src/main/java/io/ray/docdemo/PlacementGroupDemo.java
View file

@ -95,31 +95,6 @@ public class PlacementGroupDemo {
Assert.assertEquals(removedPlacementGroup.getState(), PlacementGroupState.REMOVED);
}
public static void createGlobalNamedPlacementGroup() {
// Create a placement group with a globally unique name.
Map<String, Double> bundle = ImmutableMap.of("CPU", 1.0);
List<Map<String, Double>> bundles = ImmutableList.of(bundle);
PlacementGroupCreationOptions options =
new PlacementGroupCreationOptions.Builder()
.setBundles(bundles)
.setStrategy(PlacementStrategy.STRICT_SPREAD)
.setGlobalName("global_name")
.build();
PlacementGroup pg = PlacementGroups.createPlacementGroup(options);
pg.wait(60);
// Retrieve the placement group later somewhere.
PlacementGroup group = PlacementGroups.getGlobalPlacementGroup("global_name");
Assert.assertNotNull(group);
PlacementGroups.removePlacementGroup(pg.getId());
PlacementGroup removedPlacementGroup = PlacementGroups.getPlacementGroup(pg.getId());
Assert.assertEquals(removedPlacementGroup.getState(), PlacementGroupState.REMOVED);
}
public static void createNonGlobalNamedPlacementGroup() {
// Create a placement group with a job-scope-unique name.
Map<String, Double> bundle = ImmutableMap.of("CPU", 1.0);
@ -187,8 +162,6 @@ public class PlacementGroupDemo {
runNormalTaskWithPlacementGroup();
createGlobalNamedPlacementGroup();
createNonGlobalNamedPlacementGroup();
strictPackExample();

10
java/test/src/main/java/io/ray/docdemo/UsingActorsDemo.java
View file

@ -121,16 +121,6 @@ public class UsingActorsDemo {
actorHandle.kill();
}
{
// Create an actor with a globally unique name
ActorHandle<Counter> counter = Ray.actor(Counter::new).setGlobalName("some_name").remote();
}
{
// Retrieve the actor later somewhere
Optional<ActorHandle<Counter>> counter = Ray.getGlobalActor("some_name");
Assert.assertTrue(counter.isPresent());
}
{
// Create an actor with a job-scope-unique name
ActorHandle<Counter> counter = Ray.actor(Counter::new).setName("some_name_in_job").remote();

44
java/test/src/main/java/io/ray/test/NamedActorTest.java
View file

@ -2,9 +2,7 @@ package io.ray.test;
import io.ray.api.ActorHandle;
import io.ray.api.Ray;
import java.io.IOException;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@ -41,48 +39,6 @@ public class NamedActorTest extends BaseTest {
namedActor.get().task(Counter::increment).remote().get(), Integer.valueOf(2));
}
@Test
public void testGlobalActor() throws IOException, InterruptedException {
String name = "global-actor-counter";
// Create an actor.
ActorHandle<Counter> actor = Ray.actor(Counter::new).setGlobalName(name).remote();
Assert.assertEquals(actor.task(Counter::increment).remote().get(), Integer.valueOf(1));
Assert.assertFalse(Ray.getActor(name).isPresent());
// Get the global actor.
Optional<ActorHandle<Counter>> namedActor = Ray.getGlobalActor(name);
Assert.assertTrue(namedActor.isPresent());
// Verify that this handle is correct.
Assert.assertEquals(
namedActor.get().task(Counter::increment).remote().get(), Integer.valueOf(2));
if (!TestUtils.isSingleProcessMode()) {
// Get the global actor from another driver.
ProcessBuilder builder = TestUtils.buildDriver(NamedActorTest.class, new String[] {name});
builder.redirectError(ProcessBuilder.Redirect.INHERIT);
Process driver = builder.start();
Assert.assertTrue(driver.waitFor(60, TimeUnit.SECONDS));
Assert.assertEquals(
driver.exitValue(), 0, "The driver exited with code " + driver.exitValue());
Assert.assertEquals(
namedActor.get().task(Counter::increment).remote().get(), Integer.valueOf(4));
}
}
public static void main(String[] args) {
System.setProperty("ray.job.namespace", "named_actor_test");
Ray.init();
String actorName = args[0];
// Get the global actor.
Optional<ActorHandle<Counter>> namedActor = Ray.getGlobalActor(actorName);
Assert.assertTrue(namedActor.isPresent());
// Verify that this handle is correct.
Assert.assertEquals(
namedActor.get().task(Counter::increment).remote().get(), Integer.valueOf(3));
}
@Test(expectedExceptions = IllegalArgumentException.class)
public void testActorDuplicatedName() {
String name = "named-actor-counter";

14
java/test/src/main/java/io/ray/test/NamespaceTest.java
View file

@ -26,7 +26,7 @@ public class NamespaceTest {
Assert.assertThrows(
NoSuchElementException.class,
() -> {
Ray.getGlobalActor("a").get();
Ray.getActor("a").get();
}));
}
@ -36,7 +36,7 @@ public class NamespaceTest {
testIsolation(
MainClassForNamespaceTest.class,
() -> {
ActorHandle<A> a = (ActorHandle<A>) Ray.getGlobalActor("a").get();
ActorHandle<A> a = (ActorHandle<A>) Ray.getActor("a").get();
Assert.assertEquals("hello", a.task(A::hello).remote().get());
});
}
@ -48,7 +48,7 @@ public class NamespaceTest {
Assert.assertThrows(
NoSuchElementException.class,
() -> {
Ray.getGlobalActor("a").get();
Ray.getActor("a").get();
}));
}
@ -82,19 +82,19 @@ public class NamespaceTest {
public static class MainClassForNamespaceTest {
public static void main(String[] args) throws IOException, InterruptedException {
System.setProperty("ray.job.namespace", "test1");
startDriverWithGlobalActor();
startDriver();
}
}
public static class MainClassForAnonymousNamespaceTest {
public static void main(String[] args) throws IOException, InterruptedException {
startDriverWithGlobalActor();
startDriver();
}
}
private static void startDriverWithGlobalActor() throws InterruptedException {
private static void startDriver() throws InterruptedException {
Ray.init();
ActorHandle<A> a = Ray.actor(A::new).setGlobalName("a").remote();
ActorHandle<A> a = Ray.actor(A::new).setName("a").remote();
Assert.assertEquals("hello", a.task(A::hello).remote().get());
/// Because we don't support long running job yet, so sleep to don't destroy
/// it for a while. Otherwise the actor created in this job will be destroyed

44
java/test/src/main/java/io/ray/test/PlacementGroupTest.java
View file

@ -61,11 +61,11 @@ public class PlacementGroupTest extends BaseTest {
public void testGetPlacementGroup() {
PlacementGroup firstPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, "first_placement_group", false);
"CPU", 1, PlacementStrategy.PACK, 1.0, "first_placement_group");
PlacementGroup secondPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, "second_placement_group", false);
"CPU", 1, PlacementStrategy.PACK, 1.0, "second_placement_group");
Assert.assertTrue(firstPlacementGroup.wait(60));
Assert.assertTrue(secondPlacementGroup.wait(60));
@ -100,11 +100,11 @@ public class PlacementGroupTest extends BaseTest {
public void testRemovePlacementGroup() {
PlacementGroup firstPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, "first_placement_group", false);
"CPU", 1, PlacementStrategy.PACK, 1.0, "first_placement_group");
PlacementGroup secondPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, "second_placement_group", false);
"CPU", 1, PlacementStrategy.PACK, 1.0, "second_placement_group");
Assert.assertTrue(firstPlacementGroup.wait(60));
Assert.assertTrue(secondPlacementGroup.wait(60));
@ -164,23 +164,12 @@ public class PlacementGroupTest extends BaseTest {
String pgName = "named_placement_group";
PlacementGroup firstPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, pgName, false);
"CPU", 1, PlacementStrategy.PACK, 1.0, pgName);
Assert.assertTrue(firstPlacementGroup.wait(60));
// Make sure we can get it by name successfully.
PlacementGroup placementGroup = PlacementGroups.getPlacementGroup(pgName);
Assert.assertNotNull(placementGroup);
Assert.assertEquals(placementGroup.getBundles().size(), 1);
// Test global placement group.
String pgGlobalName = "global_placement_group";
PlacementGroup secondPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, pgGlobalName, true);
Assert.assertTrue(secondPlacementGroup.wait(60));
// Make sure we can get it by name successfully.
placementGroup = PlacementGroups.getGlobalPlacementGroup(pgGlobalName);
Assert.assertNotNull(placementGroup);
Assert.assertEquals(placementGroup.getBundles().size(), 1);
}
@Test(groups = {"cluster"})
@ -188,29 +177,12 @@ public class PlacementGroupTest extends BaseTest {
String pgName = "named_placement_group";
PlacementGroup firstPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, pgName, false);
"CPU", 1, PlacementStrategy.PACK, 1.0, pgName);
Assert.assertTrue(firstPlacementGroup.wait(60));
int exceptionCount = 0;
try {
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, pgName, false);
} catch (IllegalArgumentException e) {
++exceptionCount;
}
Assert.assertEquals(exceptionCount, 1);
}
@Test(groups = {"cluster"})
public void testCreateGlobalPlacementGroupWithSameName() {
String pgGlobalName = "global_placement_group";
PlacementGroup firstPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, pgGlobalName, true);
Assert.assertTrue(firstPlacementGroup.wait(60));
int exceptionCount = 0;
try {
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"CPU", 1, PlacementStrategy.PACK, 1.0, pgGlobalName, true);
"CPU", 1, PlacementStrategy.PACK, 1.0, pgName);
} catch (IllegalArgumentException e) {
++exceptionCount;
}
@ -223,7 +195,7 @@ public class PlacementGroupTest extends BaseTest {
String pgName = "named_placement_group";
PlacementGroup nonExistPlacementGroup =
PlacementGroupTestUtils.createNameSpecifiedSimpleGroup(
"non-exist-resource", 1, PlacementStrategy.PACK, 1.0, pgName, false);
"non-exist-resource", 1, PlacementStrategy.PACK, 1.0, pgName);
// Make sure its creation will failed.
Assert.assertFalse(nonExistPlacementGroup.wait(60));

11
java/test/src/main/java/io/ray/test/PlacementGroupTestUtils.java
View file

@ -17,8 +17,7 @@ public class PlacementGroupTestUtils {
int bundleSize,
PlacementStrategy strategy,
Double resourceSize,
String groupName,
boolean isGlobal) {
String groupName) {
List<Map<String, Double>> bundles = new ArrayList<>();
for (int i = 0; i < bundleSize; i++) {
@ -28,11 +27,7 @@ public class PlacementGroupTestUtils {
}
PlacementGroupCreationOptions.Builder builder =
new PlacementGroupCreationOptions.Builder().setBundles(bundles).setStrategy(strategy);
if (isGlobal) {
builder.setGlobalName(groupName);
} else {
builder.setName(groupName);
}
builder.setName(groupName);
return PlacementGroups.createPlacementGroup(builder.build());
}
@ -44,7 +39,7 @@ public class PlacementGroupTestUtils {
Double resourceSize,
boolean isGlobal) {
return createNameSpecifiedSimpleGroup(
resourceName, bundleSize, strategy, resourceSize, "unnamed_group", isGlobal);
resourceName, bundleSize, strategy, resourceSize, "unnamed_group");
}
public static PlacementGroup createSimpleGroup() {

10
src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc
View file

@ -288,12 +288,14 @@ JNIEXPORT void JNICALL Java_io_ray_runtime_RayNativeRuntime_nativeShutdown(JNIEn
JNIEXPORT jbyteArray JNICALL
Java_io_ray_runtime_RayNativeRuntime_nativeGetActorIdOfNamedActor(JNIEnv *env, jclass,
jstring actor_name,
jboolean global) {
jstring ray_namespace) {
const char *native_actor_name = env->GetStringUTFChars(actor_name, JNI_FALSE);
auto full_name = GetFullName(global, native_actor_name);
const char *native_ray_namespace =
ray_namespace == nullptr
? CoreWorkerProcess::GetCoreWorker().GetJobConfig().ray_namespace().c_str()
: env->GetStringUTFChars(ray_namespace, JNI_FALSE);
const auto pair = CoreWorkerProcess::GetCoreWorker().GetNamedActorHandle(
full_name, /*ray_namespace=*/"");
native_actor_name, /*ray_namespace=*/native_ray_namespace);
const auto status = pair.second;
if (status.IsNotFound()) {
return IdToJavaByteArray<ActorID>(env, ActorID::Nil());

4
src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.h
View file

@ -60,11 +60,11 @@ JNIEXPORT void JNICALL Java_io_ray_runtime_RayNativeRuntime_nativeKillActor(JNIE
/*
* Class: io_ray_runtime_RayNativeRuntime
* Method: nativeGetActorIdOfNamedActor
* Signature: (Ljava/lang/String;Z)[B
* Signature: (Ljava/lang/String;Ljava/lang/String;)[B
*/
JNIEXPORT jbyteArray JNICALL
Java_io_ray_runtime_RayNativeRuntime_nativeGetActorIdOfNamedActor(JNIEnv *, jclass,
jstring, jboolean);
jstring, jstring);
/*
* Class: io_ray_runtime_RayNativeRuntime

6
src/ray/core_worker/lib/java/io_ray_runtime_gcs_GlobalStateAccessor.cc
View file

@ -123,13 +123,11 @@ Java_io_ray_runtime_gcs_GlobalStateAccessor_nativeGetPlacementGroupInfo(
JNIEXPORT jbyteArray JNICALL
Java_io_ray_runtime_gcs_GlobalStateAccessor_nativeGetPlacementGroupInfoByName(
JNIEnv *env, jobject o, jlong gcs_accessor_ptr, jstring name, jstring ray_namespace,
jboolean global) {
JNIEnv *env, jobject o, jlong gcs_accessor_ptr, jstring name, jstring ray_namespace) {
std::string placement_group_name = JavaStringToNativeString(env, name);
auto full_name = GetFullName(global, placement_group_name);
auto *gcs_accessor = reinterpret_cast<gcs::GlobalStateAccessor *>(gcs_accessor_ptr);
auto placement_group = gcs_accessor->GetPlacementGroupByName(
full_name, JavaStringToNativeString(env, ray_namespace));
placement_group_name, JavaStringToNativeString(env, ray_namespace));
if (placement_group) {
return NativeStringToJavaByteArray(env, *placement_group);
}

4
src/ray/core_worker/lib/java/io_ray_runtime_gcs_GlobalStateAccessor.h
View file

@ -115,11 +115,11 @@ Java_io_ray_runtime_gcs_GlobalStateAccessor_nativeGetPlacementGroupInfo(JNIEnv *
/*
* Class: io_ray_runtime_gcs_GlobalStateAccessor
* Method: nativeGetPlacementGroupInfoByName
* Signature: (JLjava/lang/String;Ljava/lang/String;Z)[B
* Signature: (JLjava/lang/String;Ljava/lang/String;)[B
*/
JNIEXPORT jbyteArray JNICALL
Java_io_ray_runtime_gcs_GlobalStateAccessor_nativeGetPlacementGroupInfoByName(
JNIEnv *, jobject, jlong, jstring, jstring, jboolean);
JNIEnv *, jobject, jlong, jstring, jstring);
/*
* Class: io_ray_runtime_gcs_GlobalStateAccessor

11
src/ray/core_worker/lib/java/io_ray_runtime_task_NativeTaskSubmitter.cc
View file

@ -143,7 +143,6 @@ inline TaskOptions ToTaskOptions(JNIEnv *env, jint numReturns, jobject callOptio
inline ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
jobject actorCreationOptions) {
bool global = false;
std::string name = "";
int64_t max_restarts = 0;
std::unordered_map<std::string, double> resources;
@ -153,8 +152,6 @@ inline ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
std::vector<ConcurrencyGroup> concurrency_groups;
if (actorCreationOptions) {
global =
env->GetBooleanField(actorCreationOptions, java_actor_creation_options_global);
auto java_name = (jstring)env->GetObjectField(actorCreationOptions,
java_actor_creation_options_name);
if (java_name) {
@ -219,7 +216,6 @@ inline ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
});
}
auto full_name = GetFullName(global, name);
// TODO(suquark): support passing namespace for Java. Currently
// there is no use case.
std::string ray_namespace = "";
@ -231,7 +227,7 @@ inline ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
resources,
dynamic_worker_options,
/*is_detached=*/false,
full_name,
name,
ray_namespace,
/*is_asyncio=*/false,
placement_options,
@ -258,8 +254,6 @@ inline PlacementStrategy ConvertStrategy(jint java_strategy) {
inline PlacementGroupCreationOptions ToPlacementGroupCreationOptions(
JNIEnv *env, jobject placementGroupCreationOptions) {
// We have make sure the placementGroupCreationOptions is not null in java api.
bool global = env->GetBooleanField(placementGroupCreationOptions,
java_placement_group_creation_options_global);
std::string name = "";
jstring java_name = (jstring)env->GetObjectField(
placementGroupCreationOptions, java_placement_group_creation_options_name);
@ -286,8 +280,7 @@ inline PlacementGroupCreationOptions ToPlacementGroupCreationOptions(
return value;
});
});
auto full_name = GetFullName(global, name);
return PlacementGroupCreationOptions(full_name, ConvertStrategy(java_strategy), bundles,
return PlacementGroupCreationOptions(name, ConvertStrategy(java_strategy), bundles,
/*is_detached=*/false);
}

6
src/ray/core_worker/lib/java/jni_init.cc
View file

@ -94,7 +94,6 @@ jfieldID java_task_creation_options_bundle_index;
jfieldID java_call_options_concurrency_group_name;
jclass java_actor_creation_options_class;
jfieldID java_actor_creation_options_global;
jfieldID java_actor_creation_options_name;
jfieldID java_actor_creation_options_max_restarts;
jfieldID java_actor_creation_options_jvm_options;
@ -105,7 +104,6 @@ jfieldID java_actor_creation_options_concurrency_groups;
jclass java_placement_group_creation_options_class;
jclass java_placement_group_creation_options_strategy_class;
jfieldID java_placement_group_creation_options_global;
jfieldID java_placement_group_creation_options_name;
jfieldID java_placement_group_creation_options_bundles;
jfieldID java_placement_group_creation_options_strategy;
@ -275,8 +273,6 @@ jint JNI_OnLoad(JavaVM *vm, void *reserved) {
LoadClass(env, "io/ray/api/options/PlacementGroupCreationOptions");
java_placement_group_creation_options_strategy_class =
LoadClass(env, "io/ray/api/placementgroup/PlacementStrategy");
java_placement_group_creation_options_global =
env->GetFieldID(java_placement_group_creation_options_class, "global", "Z");
java_placement_group_creation_options_name = env->GetFieldID(
java_placement_group_creation_options_class, "name", "Ljava/lang/String;");
java_placement_group_creation_options_bundles = env->GetFieldID(
@ -289,8 +285,6 @@ jint JNI_OnLoad(JavaVM *vm, void *reserved) {
java_actor_creation_options_class =
LoadClass(env, "io/ray/api/options/ActorCreationOptions");
java_actor_creation_options_global =
env->GetFieldID(java_actor_creation_options_class, "global", "Z");
java_actor_creation_options_name =
env->GetFieldID(java_actor_creation_options_class, "name", "Ljava/lang/String;");
java_actor_creation_options_max_restarts =

14
src/ray/core_worker/lib/java/jni_utils.h
View file

@ -167,8 +167,6 @@ extern jfieldID java_call_options_concurrency_group_name;
/// ActorCreationOptions class
extern jclass java_actor_creation_options_class;
/// global field of ActorCreationOptions class
extern jfieldID java_actor_creation_options_global;
/// name field of ActorCreationOptions class
extern jfieldID java_actor_creation_options_name;
/// maxRestarts field of ActorCreationOptions class
@ -197,8 +195,6 @@ extern jfieldID java_concurrency_group_impl_max_concurrency;
extern jclass java_placement_group_creation_options_class;
/// PlacementStrategy class
extern jclass java_placement_group_creation_options_strategy_class;
/// global field of PlacementGroupCreationOptions class
extern jfieldID java_placement_group_creation_options_global;
/// name field of PlacementGroupCreationOptions class
extern jfieldID java_placement_group_creation_options_name;
/// bundles field of PlacementGroupCreationOptions class
@ -615,16 +611,6 @@ inline NativeT JavaProtobufObjectToNativeProtobufObject(JNIEnv *env, jobject jav
return native_obj;
}
// Return an actor or a placement group fullname with job id prepended if this is a global
// actor or placement group.
inline std::string GetFullName(bool global, std::string name) {
if (name.empty()) {
return "";
}
return global ? name
: CoreWorkerProcess::GetCoreWorker().GetCurrentJobId().Hex() + "-" + name;
}
inline std::shared_ptr<LocalMemoryBuffer> SerializeActorCreationException(
JNIEnv *env, jthrowable creation_exception) {
jbyteArray exception_jbyte_array = static_cast<jbyteArray>(