wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-29 11:26:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

[java] Improve UniqueID code. (#2723)

Browse Source
This commit is contained in:
Wang Qing
2018-08-26 12:32:57 -07:00
committed by Robert Nishihara
parent 4f4bea086a
commit 26d3c0655c
12 changed files with 53 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
java/api/src/main/java/org/ray/api/RayActor.java
+3 -3
View File
@@ -10,14 +10,14 @@ import org.ray.util.Sha1Digestor;
* Ray actor abstraction.
*/
public class RayActor<T> extends RayObject<T> implements Externalizable {
public static final RayActor<?> nil = new RayActor<>(UniqueID.nil, UniqueID.nil);
public static final RayActor<?> NIL = new RayActor<>(UniqueID.NIL, UniqueID.NIL);
private static final long serialVersionUID = 1877485807405645036L;
private int taskCounter = 0;
private UniqueID taskCursor = null;
private UniqueID actorHandleId = UniqueID.nil;
private UniqueID actorHandleId = UniqueID.NIL;
private int forksNum = 0;
@@ -73,7 +73,7 @@ public class RayActor<T> extends RayObject<T> implements Externalizable {
}
public UniqueID computeNextActorHandleId() {
byte[] bytes = Sha1Digestor.digest(actorHandleId.id, ++forksNum);
byte[] bytes = Sha1Digestor.digest(actorHandleId.getBytes(), ++forksNum);
return new UniqueID(bytes);
}
java/api/src/main/java/org/ray/api/UniqueID.java
+20 -39
View File
@@ -2,9 +2,9 @@ package org.ray.api;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.IntBuffer;
import java.util.Arrays;
import java.util.Random;
import javax.xml.bind.DatatypeConverter;
/**
* Unique ID for task, worker, function...
@@ -12,40 +12,25 @@ import java.util.Random;
public class UniqueID implements Serializable {
public static final int LENGTH = 20;
public static final UniqueID nil = genNil();
public static final UniqueID NIL = genNil();
private static final long serialVersionUID = 8588849129675565761L;
byte[] id;
private final byte[] id;
public UniqueID(byte[] id) {
this.id = id;
public static UniqueID fromHexString(String hex) {
byte[] bytes = DatatypeConverter.parseHexBinary(hex);
return new UniqueID(bytes);
}
public UniqueID(ByteBuffer bb) {
assert (bb.remaining() == LENGTH);
id = new byte[bb.remaining()];
public static UniqueID fromByteBuffer(ByteBuffer bb) {
byte[] id = new byte[bb.remaining()];
bb.get(id);
}
public UniqueID(String optionValue) {
assert (optionValue.length() == 2 * LENGTH);
int j = 0;
id = new byte[LENGTH];
for (int i = 0; i < LENGTH; i++) {
char c1 = optionValue.charAt(j++);
char c2 = optionValue.charAt(j++);
int first = c1 <= '9' ? (c1 - '0') : (c1 - 'a' + 0xa);
int second = c2 <= '9' ? (c2 - '0') : (c2 - 'a' + 0xa);
id[i] = (byte) (first * 16 + second);
}
return new UniqueID(id);
}
public static UniqueID genNil() {
byte[] b = new byte[LENGTH];
for (int i = 0; i < b.length; i++) {
b[i] = (byte) 0xFF;
}
Arrays.fill(b, (byte) 0xFF);
return new UniqueID(b);
}
@@ -55,6 +40,14 @@ public class UniqueID implements Serializable {
return new UniqueID(b);
}
public UniqueID(byte[] id) {
if (id.length != LENGTH) {
throw new IllegalArgumentException("Illegal argument: " + id.toString());
}
this.id = id;
}
public byte[] getBytes() {
return id;
}
@@ -70,12 +63,7 @@ public class UniqueID implements Serializable {
@Override
public int hashCode() {
int hash = 0xdeadbeef;
IntBuffer bb = ByteBuffer.wrap(id).asIntBuffer();
while (bb.hasRemaining()) {
hash ^= bb.get();
}
return hash;
return Arrays.hashCode(id);
}
@Override
@@ -94,14 +82,7 @@ public class UniqueID implements Serializable {
@Override
public String toString() {
String s = "";
String hex = "0123456789abcdef";
for (int i = 0; i < LENGTH; i++) {
int val = id[i] & 0xff;
s += hex.charAt(val >> 4);
s += hex.charAt(val & 0xf);
}
return s;
return DatatypeConverter.printHexBinary(id);
}
public boolean isNil() {
java/runtime-common/src/main/java/org/ray/core/RayRuntime.java
+2 -2
View File
@@ -209,7 +209,7 @@ public abstract class RayRuntime implements RayApi {
}
/**
* get the task identity of the currently running task, UniqueID.Nil if not inside any
* get the task identity of the currently running task, UniqueID.NIL if not inside any
*/
public UniqueID getCurrentTaskId() {
return worker.getCurrentTaskId();
@@ -381,7 +381,7 @@ public abstract class RayRuntime implements RayApi {
}
/**
* get the object put identity of the currently running task, UniqueID.Nil if not inside any
* get the object put identity of the currently running task, UniqueID.NIL if not inside any
*/
public UniqueID[] getCurrentTaskReturnIDs() {
return worker.getCurrentTaskReturnIDs();
java/runtime-common/src/main/java/org/ray/core/Worker.java
+2 -3
View File
@@ -67,9 +67,8 @@ public class Worker {
}
private RayObject taskSubmit(UniqueID taskId, MethodId methodId, Object[] args) {
RayInvocation ri = createRemoteInvocation(methodId, args, RayActor.nil);
RayInvocation ri = createRemoteInvocation(methodId, args, RayActor.NIL);
return scheduler.submit(taskId, ri);
}
@@ -95,7 +94,7 @@ public class Worker {
RayFunc func, Object[] args) {
Preconditions.checkNotNull(taskId);
MethodId mid = methodIdOf(func);
RayInvocation ri = createRemoteInvocation(mid, args, RayActor.nil);
RayInvocation ri = createRemoteInvocation(mid, args, RayActor.NIL);
return scheduler.submitActorCreationTask(taskId, createActorId, ri);
}
java/runtime-common/src/main/java/org/ray/core/WorkerContext.java
+3 -3
View File
@@ -35,13 +35,13 @@ public class WorkerContext {
currentWorkerCtx.set(ctx);
TaskSpec dummy = new TaskSpec();
dummy.parentTaskId = UniqueID.nil;
dummy.parentTaskId = UniqueID.NIL;
if (params.worker_mode == WorkerMode.DRIVER) {
dummy.taskId = UniqueID.randomId();
} else {
dummy.taskId = UniqueID.nil;
dummy.taskId = UniqueID.NIL;
}
dummy.actorId = UniqueID.nil;
dummy.actorId = UniqueID.NIL;
dummy.driverId = params.driver_id;
prepare(dummy, null);
java/runtime-common/src/main/java/org/ray/core/model/RayParameters.java
+1 -1
View File
@@ -44,7 +44,7 @@ public class RayParameters {
public int local_scheduler_rpc_port = 34567;
@AConfig(comment = "driver ID when the worker is served as a driver")
public UniqueID driver_id = UniqueID.nil;
public UniqueID driver_id = UniqueID.NIL;
@AConfig(comment = "logging directory")
public String log_dir = "/tmp/raylogs";
java/runtime-common/src/main/java/org/ray/spi/LocalSchedulerProxy.java
+2 -2
View File
@@ -28,14 +28,14 @@ public class LocalSchedulerProxy {
public RayObject submit(UniqueID taskId, RayInvocation invocation) {
UniqueID[] returnIds = genReturnIds(taskId, 1);
this.doSubmit(invocation, taskId, returnIds, UniqueID.nil);
this.doSubmit(invocation, taskId, returnIds, UniqueID.NIL);
return new RayObject(returnIds[0]);
}
public RayObject submitActorTask(UniqueID taskId, RayInvocation invocation) {
// add one for the dummy return ID
UniqueID[] returnIds = genReturnIds(taskId, 2);
this.doSubmit(invocation, taskId, returnIds, UniqueID.nil);
this.doSubmit(invocation, taskId, returnIds, UniqueID.NIL);
return new RayObject(returnIds[0]);
}
java/runtime-common/src/main/java/org/ray/spi/model/RayInvocation.java
+1 -1
View File
@@ -9,7 +9,7 @@ import org.ray.api.UniqueID;
*/
public class RayInvocation {
private static final RayActor<?> nil = new RayActor<>(UniqueID.nil, UniqueID.nil);
private static final RayActor<?> nil = new RayActor<>(UniqueID.NIL, UniqueID.NIL);
public final String className;
/**
* unique id for a method.
java/runtime-native/src/main/java/org/ray/core/impl/RayNativeRuntime.java
+3 -3
View File
@@ -47,7 +47,7 @@ public class RayNativeRuntime extends RayRuntime {
private KeyValueStoreLink kvStore = null;
private RunManager manager = null;
private Object actor = null;
private UniqueID actorId = UniqueID.nil;
private UniqueID actorId = UniqueID.NIL;
protected RayNativeRuntime() {
}
@@ -114,7 +114,7 @@ public class RayNativeRuntime extends RayRuntime {
LocalSchedulerLink slink = new DefaultLocalSchedulerClient(
params.local_scheduler_name,
WorkerContext.currentWorkerId(),
UniqueID.nil,
UniqueID.NIL,
isWorker,
WorkerContext.currentTask().taskId,
0,
@@ -133,7 +133,7 @@ public class RayNativeRuntime extends RayRuntime {
LocalSchedulerLink slink = new DefaultLocalSchedulerClient(
params.raylet_socket_name,
WorkerContext.currentWorkerId(),
UniqueID.nil,
UniqueID.NIL,
isWorker,
WorkerContext.currentTask().taskId,
0,
java/runtime-native/src/main/java/org/ray/runner/RunManager.java
+5 -5
View File
@@ -369,7 +369,7 @@ public class RunManager {
for (int j = 0; j < localNumWorkers[i]; j++) {
startWorker(localStores.storeName, localStores.managerName, localStores.schedulerName,
"/worker" + i + "." + j, params.redis_address,
params.node_ip_address, UniqueID.nil, "", params.redirect, params.cleanup);
params.node_ip_address, UniqueID.NIL, "", params.redirect, params.cleanup);
}
}
}
@@ -570,7 +570,7 @@ public class RunManager {
String workerCmd = null;
workerCmd = buildWorkerCommand(true, info.storeName, info.managerName, name,
UniqueID.nil, "", ip, redisAddress);
UniqueID.NIL, "", ip, redisAddress);
cmd += " -w \"" + workerCmd + "\"";
if (redisAddress.length() > 0) {
@@ -614,7 +614,7 @@ public class RunManager {
//Create the worker command that the raylet will use to start workers.
String workerCommand = buildWorkerCommandRaylet(info.storeName, rayletSocketName,
UniqueID.nil, "", ip, redisAddress);
UniqueID.NIL, "", ip, redisAddress);
int sep = redisAddress.indexOf(':');
assert (sep != -1);
@@ -656,7 +656,7 @@ public class RunManager {
+ ";ray.java.start.raylet_socket_name=" + rayletSocketName
+ ";ray.java.start.worker_mode=WORKER;ray.java.start.use_raylet=true";
workerConfigs += ";ray.java.start.deploy=" + params.deploy;
if (!actorId.equals(UniqueID.nil)) {
if (!actorId.equals(UniqueID.NIL)) {
workerConfigs += ";ray.java.start.actor_id=" + actorId;
}
if (!actorClass.equals("")) {
@@ -688,7 +688,7 @@ public class RunManager {
+ ";ray.java.start.worker_mode=WORKER"
+ ";ray.java.start.local_scheduler_name=" + localSchedulerName;
workerConfigs += ";ray.java.start.deploy=" + params.deploy;
if (!actorId.equals(UniqueID.nil)) {
if (!actorId.equals(UniqueID.NIL)) {
workerConfigs += ";ray.java.start.actor_id=" + actorId;
}
if (!actorClass.equals("")) {
java/runtime-native/src/main/java/org/ray/spi/impl/DefaultLocalSchedulerClient.java
+9 -9
View File
@@ -135,15 +135,15 @@ public class DefaultLocalSchedulerClient implements LocalSchedulerLink {
TaskInfo info = TaskInfo.getRootAsTaskInfo(bb);
TaskSpec spec = new TaskSpec();
spec.driverId = new UniqueID(info.driverIdAsByteBuffer());
spec.taskId = new UniqueID(info.taskIdAsByteBuffer());
spec.parentTaskId = new UniqueID(info.parentTaskIdAsByteBuffer());
spec.driverId = UniqueID.fromByteBuffer(info.driverIdAsByteBuffer());
spec.taskId = UniqueID.fromByteBuffer(info.taskIdAsByteBuffer());
spec.parentTaskId = UniqueID.fromByteBuffer(info.parentTaskIdAsByteBuffer());
spec.parentCounter = info.parentCounter();
spec.actorId = new UniqueID(info.actorIdAsByteBuffer());
spec.actorId = UniqueID.fromByteBuffer(info.actorIdAsByteBuffer());
spec.actorCounter = info.actorCounter();
spec.createActorId = new UniqueID(info.actorCreationIdAsByteBuffer());
spec.createActorId = UniqueID.fromByteBuffer(info.actorCreationIdAsByteBuffer());
spec.functionId = new UniqueID(info.functionIdAsByteBuffer());
spec.functionId = UniqueID.fromByteBuffer(info.functionIdAsByteBuffer());
List<FunctionArg> args = new ArrayList<>();
for (int i = 0; i < info.argsLength(); i++) {
@@ -156,7 +156,7 @@ public class DefaultLocalSchedulerClient implements LocalSchedulerLink {
for (int j = 0; j < idCount; j++) {
ByteBuffer lbb = sarg.objectIdAsByteBuffer(j);
assert (lbb != null && lbb.remaining() > 0);
darg.ids.add(new UniqueID(lbb));
darg.ids.add(UniqueID.fromByteBuffer(lbb));
}
}
@@ -175,7 +175,7 @@ public class DefaultLocalSchedulerClient implements LocalSchedulerLink {
for (int i = 0; i < info.returnsLength(); i++) {
ByteBuffer lbb = info.returnsAsByteBuffer(i);
assert (lbb != null && lbb.remaining() > 0);
rids.add(new UniqueID(lbb));
rids.add(UniqueID.fromByteBuffer(lbb));
}
spec.returnIds = rids.toArray(new UniqueID[0]);
@@ -193,7 +193,7 @@ public class DefaultLocalSchedulerClient implements LocalSchedulerLink {
final int parentTaskIdOffset = fbb.createString(task.parentTaskId.toByteBuffer());
final int parentCounter = task.parentCounter;
final int actorCreateIdOffset = fbb.createString(task.createActorId.toByteBuffer());
final int actorCreateDummyIdOffset = fbb.createString(UniqueID.nil.toByteBuffer());
final int actorCreateDummyIdOffset = fbb.createString(UniqueID.NIL.toByteBuffer());
final int actorIdOffset = fbb.createString(task.actorId.toByteBuffer());
final int actorHandleIdOffset = fbb.createString(task.actorHandleId.toByteBuffer());
final int actorCounter = task.actorCounter;
java/runtime-native/src/main/java/org/ray/spi/impl/NativeRemoteFunctionManager.java
+2 -2
View File
@@ -64,7 +64,7 @@ public class NativeRemoteFunctionManager implements RemoteFunctionManager {
@Override
public UniqueID getAppResourceId(UniqueID driverId) {
return new UniqueID(kvStore.get("App2ResMap", driverId.toString()));
return UniqueID.fromHexString(kvStore.get("App2ResMap", driverId.toString()));
}
@Override
@@ -92,7 +92,7 @@ public class NativeRemoteFunctionManager implements RemoteFunctionManager {
ClassLoader cl = loadedApps.get(driverId);
if (cl == null) {
UniqueID resId = new UniqueID(kvStore.get("App2ResMap", driverId.toString()));
UniqueID resId = UniqueID.fromHexString(kvStore.get("App2ResMap", driverId.toString()));
byte[] res = getResource(resId);
if (res == null) {
throw new RuntimeException("get resource null, the resId " + resId.toString());