[xray] Implement actor reconstruction.

.travis.yml

@@ -50,6 +50,7 @@ matrix:
         - cd doc
         - pip install -q -r requirements-doc.txt
         - pip install yapf==0.23.0
+        - pip install pytest-timeout
         - sphinx-build -W -b html -d _build/doctrees source _build/html
         - cd ..
         # Run Python linting, ignore dict vs {} (C408), others are defaults

java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java

@@ -1,18 +1,31 @@
 package org.ray.api.options;
 
+import java.util.HashMap;
 import java.util.Map;
 
 /**
  * The options for creating actor.
  */
 public class ActorCreationOptions extends BaseTaskOptions {
 
+  public static final int NO_RECONSTRUCTION = 0;
+  public static final int INFINITE_RECONSTRUCTIONS = (int) Math.pow(2, 30);
+
+  public final int maxReconstructions;
+
   public ActorCreationOptions() {
     super();
+    this.maxReconstructions = NO_RECONSTRUCTION;
   }
 
   public ActorCreationOptions(Map<String, Double> resources) {
     super(resources);
+    this.maxReconstructions = NO_RECONSTRUCTION;
   }
 
+
+  public ActorCreationOptions(Map<String, Double> resources, int maxReconstructions) {
+    super(resources);
+    this.maxReconstructions = maxReconstructions;
+  }
 }

java/runtime/src/main/java/org/ray/runtime/AbstractRayRuntime.java

@@ -220,6 +220,7 @@ public <T> RayActor<T> createActor(RayFunc actorFactoryFunc,
     TaskSpec spec = createTaskSpec(actorFactoryFunc, RayActorImpl.NIL,
         args, true, options);
     RayActorImpl<?> actor = new RayActorImpl(spec.returnIds[0]);
+    actor.setCreationDummyObjectId(spec.actorCreationId);
     actor.increaseTaskCounter();
     actor.setTaskCursor(spec.returnIds[0]);
     rayletClient.submitTask(spec);
@@ -258,6 +259,11 @@ private TaskSpec createTaskSpec(RayFunc func, RayActorImpl actor, Object[] args,
       actorCreationId = returnIds[0];
     }
 
+    UniqueId actorCreationDummyObjectId = actor.getCreationDummyObjectId();
+    if (isActorCreationTask) {
+      actorCreationDummyObjectId = actorCreationId;
+    }
+
     Map<String, Double> resources;
     if (null == taskOptions) {
       resources = new HashMap<>();
@@ -270,13 +276,19 @@ private TaskSpec createTaskSpec(RayFunc func, RayActorImpl actor, Object[] args,
       resources.put(ResourceUtil.CPU_LITERAL, 0.0);
     }
 
+    int maxActorReconstruction = 0;
+    if (taskOptions instanceof ActorCreationOptions) {
+      maxActorReconstruction = ((ActorCreationOptions) taskOptions).maxReconstructions;
+    }
     RayFunction rayFunction = functionManager.getFunction(current.driverId, func);
     return new TaskSpec(
         current.driverId,
         taskId,
         current.taskId,
         -1,
         actorCreationId,
+        actorCreationDummyObjectId,
+        maxActorReconstruction,
         actor.getId(),
         actor.getHandleId(),
         actor.increaseTaskCounter(),

java/runtime/src/main/java/org/ray/runtime/RayActorImpl.java

@@ -28,6 +28,10 @@ public final class RayActorImpl<T> implements RayActor<T>, Externalizable {
    * It's used to make sure ids of actor handles are unique.
    */
   private int numForks;
+  /**
+   * Dummy object ID of the actor creation task.
+   */
+  private UniqueId creationDummyObjectId;
 
   public RayActorImpl() {
     this(UniqueId.NIL, UniqueId.NIL);
@@ -43,6 +47,7 @@ public RayActorImpl(UniqueId id, UniqueId handleId) {
     this.taskCounter = 0;
     this.taskCursor = null;
     numForks = 0;
+    creationDummyObjectId = UniqueId.NIL;
   }
 
   @Override
@@ -67,6 +72,13 @@ public int increaseTaskCounter() {
     return taskCounter++;
   }
 
+  public UniqueId getCreationDummyObjectId() {
+    return creationDummyObjectId;
+  }
+
+  public void setCreationDummyObjectId(UniqueId creationDummyObjectId) {
+    this.creationDummyObjectId = creationDummyObjectId;
+  }
 
   private UniqueId computeNextActorHandleId() {
     byte[] bytes = Sha1Digestor.digest(handleId.getBytes(), ++numForks);
@@ -78,12 +90,14 @@ public void writeExternal(ObjectOutput out) throws IOException {
     out.writeObject(this.id);
     out.writeObject(this.computeNextActorHandleId());
     out.writeObject(this.taskCursor);
+    out.writeObject(this.creationDummyObjectId);
   }
 
   @Override
   public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
     this.id = (UniqueId) in.readObject();
     this.handleId = (UniqueId) in.readObject();
     this.taskCursor = (UniqueId) in.readObject();
+    this.creationDummyObjectId = (UniqueId) in.readObject();
   }
 }

java/runtime/src/main/java/org/ray/runtime/WorkerContext.java

@@ -80,6 +80,8 @@ private TaskSpec createDummyTask(WorkerMode workerMode, UniqueId driverId) {
         0,
         UniqueId.NIL,
         UniqueId.NIL,
+        0,
+        UniqueId.NIL,
         UniqueId.NIL,
         0,
         null,

java/runtime/src/main/java/org/ray/runtime/generated/TaskInfo.java

@@ -29,31 +29,32 @@ public final class TaskInfo extends Table {
   public String actorCreationDummyObjectId() { int o = __offset(14); return o != 0 ? __string(o + bb_pos) : null; }
   public ByteBuffer actorCreationDummyObjectIdAsByteBuffer() { return __vector_as_bytebuffer(14, 1); }
   public ByteBuffer actorCreationDummyObjectIdInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb, 14, 1); }
-  public String actorId() { int o = __offset(16); return o != 0 ? __string(o + bb_pos) : null; }
-  public ByteBuffer actorIdAsByteBuffer() { return __vector_as_bytebuffer(16, 1); }
-  public ByteBuffer actorIdInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb, 16, 1); }
-  public String actorHandleId() { int o = __offset(18); return o != 0 ? __string(o + bb_pos) : null; }
-  public ByteBuffer actorHandleIdAsByteBuffer() { return __vector_as_bytebuffer(18, 1); }
-  public ByteBuffer actorHandleIdInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb, 18, 1); }
-  public int actorCounter() { int o = __offset(20); return o != 0 ? bb.getInt(o + bb_pos) : 0; }
-  public boolean isActorCheckpointMethod() { int o = __offset(22); return o != 0 ? 0!=bb.get(o + bb_pos) : false; }
-  public String functionId() { int o = __offset(24); return o != 0 ? __string(o + bb_pos) : null; }
-  public ByteBuffer functionIdAsByteBuffer() { return __vector_as_bytebuffer(24, 1); }
-  public ByteBuffer functionIdInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb, 24, 1); }
+  public int maxActorReconstructions() { int o = __offset(16); return o != 0 ? bb.getInt(o + bb_pos) : 0; }
+  public String actorId() { int o = __offset(18); return o != 0 ? __string(o + bb_pos) : null; }
+  public ByteBuffer actorIdAsByteBuffer() { return __vector_as_bytebuffer(18, 1); }
+  public ByteBuffer actorIdInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb, 18, 1); }
+  public String actorHandleId() { int o = __offset(20); return o != 0 ? __string(o + bb_pos) : null; }
+  public ByteBuffer actorHandleIdAsByteBuffer() { return __vector_as_bytebuffer(20, 1); }
+  public ByteBuffer actorHandleIdInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb, 20, 1); }
+  public int actorCounter() { int o = __offset(22); return o != 0 ? bb.getInt(o + bb_pos) : 0; }
+  public boolean isActorCheckpointMethod() { int o = __offset(24); return o != 0 ? 0!=bb.get(o + bb_pos) : false; }
+  public String functionId() { int o = __offset(26); return o != 0 ? __string(o + bb_pos) : null; }
+  public ByteBuffer functionIdAsByteBuffer() { return __vector_as_bytebuffer(26, 1); }
+  public ByteBuffer functionIdInByteBuffer(ByteBuffer _bb) { return __vector_in_bytebuffer(_bb, 26, 1); }
   public Arg args(int j) { return args(new Arg(), j); }
-  public Arg args(Arg obj, int j) { int o = __offset(26); return o != 0 ? obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
-  public int argsLength() { int o = __offset(26); return o != 0 ? __vector_len(o) : 0; }
-  public String returns(int j) { int o = __offset(28); return o != 0 ? __string(__vector(o) + j * 4) : null; }
-  public int returnsLength() { int o = __offset(28); return o != 0 ? __vector_len(o) : 0; }
+  public Arg args(Arg obj, int j) { int o = __offset(28); return o != 0 ? obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
+  public int argsLength() { int o = __offset(28); return o != 0 ? __vector_len(o) : 0; }
+  public String returns(int j) { int o = __offset(30); return o != 0 ? __string(__vector(o) + j * 4) : null; }
+  public int returnsLength() { int o = __offset(30); return o != 0 ? __vector_len(o) : 0; }
   public ResourcePair requiredResources(int j) { return requiredResources(new ResourcePair(), j); }
-  public ResourcePair requiredResources(ResourcePair obj, int j) { int o = __offset(30); return o != 0 ? obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
-  public int requiredResourcesLength() { int o = __offset(30); return o != 0 ? __vector_len(o) : 0; }
+  public ResourcePair requiredResources(ResourcePair obj, int j) { int o = __offset(32); return o != 0 ? obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
+  public int requiredResourcesLength() { int o = __offset(32); return o != 0 ? __vector_len(o) : 0; }
   public ResourcePair requiredPlacementResources(int j) { return requiredPlacementResources(new ResourcePair(), j); }
-  public ResourcePair requiredPlacementResources(ResourcePair obj, int j) { int o = __offset(32); return o != 0 ? obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
-  public int requiredPlacementResourcesLength() { int o = __offset(32); return o != 0 ? __vector_len(o) : 0; }
-  public int language() { int o = __offset(34); return o != 0 ? bb.getInt(o + bb_pos) : 0; }
-  public String functionDescriptor(int j) { int o = __offset(36); return o != 0 ? __string(__vector(o) + j * 4) : null; }
-  public int functionDescriptorLength() { int o = __offset(36); return o != 0 ? __vector_len(o) : 0; }
+  public ResourcePair requiredPlacementResources(ResourcePair obj, int j) { int o = __offset(34); return o != 0 ? obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
+  public int requiredPlacementResourcesLength() { int o = __offset(34); return o != 0 ? __vector_len(o) : 0; }
+  public int language() { int o = __offset(36); return o != 0 ? bb.getInt(o + bb_pos) : 0; }
+  public String functionDescriptor(int j) { int o = __offset(38); return o != 0 ? __string(__vector(o) + j * 4) : null; }
+  public int functionDescriptorLength() { int o = __offset(38); return o != 0 ? __vector_len(o) : 0; }
 
   public static int createTaskInfo(FlatBufferBuilder builder,
       int driver_idOffset,
@@ -62,6 +63,7 @@ public static int createTaskInfo(FlatBufferBuilder builder,
       int parent_counter,
       int actor_creation_idOffset,
       int actor_creation_dummy_object_idOffset,
+      int max_actor_reconstructions,
       int actor_idOffset,
       int actor_handle_idOffset,
       int actor_counter,
@@ -73,7 +75,7 @@ public static int createTaskInfo(FlatBufferBuilder builder,
       int required_placement_resourcesOffset,
       int language,
       int function_descriptorOffset) {
-    builder.startObject(17);
+    builder.startObject(18);
     TaskInfo.addFunctionDescriptor(builder, function_descriptorOffset);
     TaskInfo.addLanguage(builder, language);
     TaskInfo.addRequiredPlacementResources(builder, required_placement_resourcesOffset);
@@ -84,6 +86,7 @@ public static int createTaskInfo(FlatBufferBuilder builder,
     TaskInfo.addActorCounter(builder, actor_counter);
     TaskInfo.addActorHandleId(builder, actor_handle_idOffset);
     TaskInfo.addActorId(builder, actor_idOffset);
+    TaskInfo.addMaxActorReconstructions(builder, max_actor_reconstructions);
     TaskInfo.addActorCreationDummyObjectId(builder, actor_creation_dummy_object_idOffset);
     TaskInfo.addActorCreationId(builder, actor_creation_idOffset);
     TaskInfo.addParentCounter(builder, parent_counter);
@@ -94,32 +97,33 @@ public static int createTaskInfo(FlatBufferBuilder builder,
     return TaskInfo.endTaskInfo(builder);
   }
 
-  public static void startTaskInfo(FlatBufferBuilder builder) { builder.startObject(17); }
+  public static void startTaskInfo(FlatBufferBuilder builder) { builder.startObject(18); }
   public static void addDriverId(FlatBufferBuilder builder, int driverIdOffset) { builder.addOffset(0, driverIdOffset, 0); }
   public static void addTaskId(FlatBufferBuilder builder, int taskIdOffset) { builder.addOffset(1, taskIdOffset, 0); }
   public static void addParentTaskId(FlatBufferBuilder builder, int parentTaskIdOffset) { builder.addOffset(2, parentTaskIdOffset, 0); }
   public static void addParentCounter(FlatBufferBuilder builder, int parentCounter) { builder.addInt(3, parentCounter, 0); }
   public static void addActorCreationId(FlatBufferBuilder builder, int actorCreationIdOffset) { builder.addOffset(4, actorCreationIdOffset, 0); }
   public static void addActorCreationDummyObjectId(FlatBufferBuilder builder, int actorCreationDummyObjectIdOffset) { builder.addOffset(5, actorCreationDummyObjectIdOffset, 0); }
-  public static void addActorId(FlatBufferBuilder builder, int actorIdOffset) { builder.addOffset(6, actorIdOffset, 0); }
-  public static void addActorHandleId(FlatBufferBuilder builder, int actorHandleIdOffset) { builder.addOffset(7, actorHandleIdOffset, 0); }
-  public static void addActorCounter(FlatBufferBuilder builder, int actorCounter) { builder.addInt(8, actorCounter, 0); }
-  public static void addIsActorCheckpointMethod(FlatBufferBuilder builder, boolean isActorCheckpointMethod) { builder.addBoolean(9, isActorCheckpointMethod, false); }
-  public static void addFunctionId(FlatBufferBuilder builder, int functionIdOffset) { builder.addOffset(10, functionIdOffset, 0); }
-  public static void addArgs(FlatBufferBuilder builder, int argsOffset) { builder.addOffset(11, argsOffset, 0); }
+  public static void addMaxActorReconstructions(FlatBufferBuilder builder, int maxActorReconstructions) { builder.addInt(6, maxActorReconstructions, 0); }
+  public static void addActorId(FlatBufferBuilder builder, int actorIdOffset) { builder.addOffset(7, actorIdOffset, 0); }
+  public static void addActorHandleId(FlatBufferBuilder builder, int actorHandleIdOffset) { builder.addOffset(8, actorHandleIdOffset, 0); }
+  public static void addActorCounter(FlatBufferBuilder builder, int actorCounter) { builder.addInt(9, actorCounter, 0); }
+  public static void addIsActorCheckpointMethod(FlatBufferBuilder builder, boolean isActorCheckpointMethod) { builder.addBoolean(10, isActorCheckpointMethod, false); }
+  public static void addFunctionId(FlatBufferBuilder builder, int functionIdOffset) { builder.addOffset(11, functionIdOffset, 0); }
+  public static void addArgs(FlatBufferBuilder builder, int argsOffset) { builder.addOffset(12, argsOffset, 0); }
   public static int createArgsVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4, data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]); return builder.endVector(); }
   public static void startArgsVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4, numElems, 4); }
-  public static void addReturns(FlatBufferBuilder builder, int returnsOffset) { builder.addOffset(12, returnsOffset, 0); }
+  public static void addReturns(FlatBufferBuilder builder, int returnsOffset) { builder.addOffset(13, returnsOffset, 0); }
   public static int createReturnsVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4, data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]); return builder.endVector(); }
   public static void startReturnsVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4, numElems, 4); }
-  public static void addRequiredResources(FlatBufferBuilder builder, int requiredResourcesOffset) { builder.addOffset(13, requiredResourcesOffset, 0); }
+  public static void addRequiredResources(FlatBufferBuilder builder, int requiredResourcesOffset) { builder.addOffset(14, requiredResourcesOffset, 0); }
   public static int createRequiredResourcesVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4, data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]); return builder.endVector(); }
   public static void startRequiredResourcesVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4, numElems, 4); }
-  public static void addRequiredPlacementResources(FlatBufferBuilder builder, int requiredPlacementResourcesOffset) { builder.addOffset(14, requiredPlacementResourcesOffset, 0); }
+  public static void addRequiredPlacementResources(FlatBufferBuilder builder, int requiredPlacementResourcesOffset) { builder.addOffset(15, requiredPlacementResourcesOffset, 0); }
   public static int createRequiredPlacementResourcesVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4, data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]); return builder.endVector(); }
   public static void startRequiredPlacementResourcesVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4, numElems, 4); }
-  public static void addLanguage(FlatBufferBuilder builder, int language) { builder.addInt(15, language, 0); }
-  public static void addFunctionDescriptor(FlatBufferBuilder builder, int functionDescriptorOffset) { builder.addOffset(16, functionDescriptorOffset, 0); }
+  public static void addLanguage(FlatBufferBuilder builder, int language) { builder.addInt(16, language, 0); }
+  public static void addFunctionDescriptor(FlatBufferBuilder builder, int functionDescriptorOffset) { builder.addOffset(17, functionDescriptorOffset, 0); }
   public static int createFunctionDescriptorVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4, data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]); return builder.endVector(); }
   public static void startFunctionDescriptorVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4, numElems, 4); }
   public static int endTaskInfo(FlatBufferBuilder builder) {
@@ -130,7 +134,7 @@ public static int endTaskInfo(FlatBufferBuilder builder) {
   //this is manually added to avoid encoding/decoding cost as our object
   //id is a byte array instead of a string
   public ByteBuffer returnsAsByteBuffer(int j) {
-    int o = __offset(28);
+    int o = __offset(30);
     if (o == 0) {
       return null;
     }