diff --git a/java/api/src/main/java/org/ray/api/RayActor.java b/java/api/src/main/java/org/ray/api/RayActor.java index f6dda3039..c9eb9df82 100644 --- a/java/api/src/main/java/org/ray/api/RayActor.java +++ b/java/api/src/main/java/org/ray/api/RayActor.java @@ -10,14 +10,14 @@ import org.ray.util.Sha1Digestor; * Ray actor abstraction. */ public class RayActor extends RayObject 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 extends RayObject implements Externalizable { } public UniqueID computeNextActorHandleId() { - byte[] bytes = Sha1Digestor.digest(actorHandleId.id, ++forksNum); + byte[] bytes = Sha1Digestor.digest(actorHandleId.getBytes(), ++forksNum); return new UniqueID(bytes); } diff --git a/java/api/src/main/java/org/ray/api/UniqueID.java b/java/api/src/main/java/org/ray/api/UniqueID.java index 08940b991..09323f264 100644 --- a/java/api/src/main/java/org/ray/api/UniqueID.java +++ b/java/api/src/main/java/org/ray/api/UniqueID.java @@ -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() { diff --git a/java/runtime-common/src/main/java/org/ray/core/RayRuntime.java b/java/runtime-common/src/main/java/org/ray/core/RayRuntime.java index fa4cc5859..b15664888 100644 --- a/java/runtime-common/src/main/java/org/ray/core/RayRuntime.java +++ b/java/runtime-common/src/main/java/org/ray/core/RayRuntime.java @@ -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(); diff --git a/java/runtime-common/src/main/java/org/ray/core/Worker.java b/java/runtime-common/src/main/java/org/ray/core/Worker.java index e8966441e..2254b5b5d 100644 --- a/java/runtime-common/src/main/java/org/ray/core/Worker.java +++ b/java/runtime-common/src/main/java/org/ray/core/Worker.java @@ -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); } diff --git a/java/runtime-common/src/main/java/org/ray/core/WorkerContext.java b/java/runtime-common/src/main/java/org/ray/core/WorkerContext.java index a9ca12561..0b19fb31d 100644 --- a/java/runtime-common/src/main/java/org/ray/core/WorkerContext.java +++ b/java/runtime-common/src/main/java/org/ray/core/WorkerContext.java @@ -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); diff --git a/java/runtime-common/src/main/java/org/ray/core/model/RayParameters.java b/java/runtime-common/src/main/java/org/ray/core/model/RayParameters.java index b8dd42dd8..a240b7c4f 100644 --- a/java/runtime-common/src/main/java/org/ray/core/model/RayParameters.java +++ b/java/runtime-common/src/main/java/org/ray/core/model/RayParameters.java @@ -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"; diff --git a/java/runtime-common/src/main/java/org/ray/spi/LocalSchedulerProxy.java b/java/runtime-common/src/main/java/org/ray/spi/LocalSchedulerProxy.java index ced01280a..d2baa0181 100644 --- a/java/runtime-common/src/main/java/org/ray/spi/LocalSchedulerProxy.java +++ b/java/runtime-common/src/main/java/org/ray/spi/LocalSchedulerProxy.java @@ -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]); } diff --git a/java/runtime-common/src/main/java/org/ray/spi/model/RayInvocation.java b/java/runtime-common/src/main/java/org/ray/spi/model/RayInvocation.java index 4ad2da95f..40dbe40cf 100644 --- a/java/runtime-common/src/main/java/org/ray/spi/model/RayInvocation.java +++ b/java/runtime-common/src/main/java/org/ray/spi/model/RayInvocation.java @@ -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. diff --git a/java/runtime-native/src/main/java/org/ray/core/impl/RayNativeRuntime.java b/java/runtime-native/src/main/java/org/ray/core/impl/RayNativeRuntime.java index 78abc9104..45d26ac21 100644 --- a/java/runtime-native/src/main/java/org/ray/core/impl/RayNativeRuntime.java +++ b/java/runtime-native/src/main/java/org/ray/core/impl/RayNativeRuntime.java @@ -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, diff --git a/java/runtime-native/src/main/java/org/ray/runner/RunManager.java b/java/runtime-native/src/main/java/org/ray/runner/RunManager.java index 48febe0dc..ebf4bfea8 100644 --- a/java/runtime-native/src/main/java/org/ray/runner/RunManager.java +++ b/java/runtime-native/src/main/java/org/ray/runner/RunManager.java @@ -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("")) { diff --git a/java/runtime-native/src/main/java/org/ray/spi/impl/DefaultLocalSchedulerClient.java b/java/runtime-native/src/main/java/org/ray/spi/impl/DefaultLocalSchedulerClient.java index 50e6ae013..690783812 100644 --- a/java/runtime-native/src/main/java/org/ray/spi/impl/DefaultLocalSchedulerClient.java +++ b/java/runtime-native/src/main/java/org/ray/spi/impl/DefaultLocalSchedulerClient.java @@ -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 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; diff --git a/java/runtime-native/src/main/java/org/ray/spi/impl/NativeRemoteFunctionManager.java b/java/runtime-native/src/main/java/org/ray/spi/impl/NativeRemoteFunctionManager.java index 1fab8918d..4a04414f2 100644 --- a/java/runtime-native/src/main/java/org/ray/spi/impl/NativeRemoteFunctionManager.java +++ b/java/runtime-native/src/main/java/org/ray/spi/impl/NativeRemoteFunctionManager.java @@ -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());