diff --git a/java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java b/java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java index 20db30944..267b7a354 100644 --- a/java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java +++ b/java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java @@ -1,5 +1,6 @@ package org.ray.api.options; +import java.util.HashMap; import java.util.Map; /** @@ -7,12 +8,24 @@ import java.util.Map; */ 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 resources) { super(resources); + this.maxReconstructions = NO_RECONSTRUCTION; } + + public ActorCreationOptions(Map resources, int maxReconstructions) { + super(resources); + this.maxReconstructions = maxReconstructions; + } } diff --git a/java/runtime/src/main/java/org/ray/runtime/AbstractRayRuntime.java b/java/runtime/src/main/java/org/ray/runtime/AbstractRayRuntime.java index 10dc172fd..2a178a3d9 100644 --- a/java/runtime/src/main/java/org/ray/runtime/AbstractRayRuntime.java +++ b/java/runtime/src/main/java/org/ray/runtime/AbstractRayRuntime.java @@ -270,6 +270,10 @@ public abstract class AbstractRayRuntime implements RayRuntime { 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, @@ -277,6 +281,7 @@ public abstract class AbstractRayRuntime implements RayRuntime { current.taskId, -1, actorCreationId, + maxActorReconstruction, actor.getId(), actor.getHandleId(), actor.increaseTaskCounter(), diff --git a/java/runtime/src/main/java/org/ray/runtime/RayActorImpl.java b/java/runtime/src/main/java/org/ray/runtime/RayActorImpl.java index fd303fa93..5f0ae13ea 100644 --- a/java/runtime/src/main/java/org/ray/runtime/RayActorImpl.java +++ b/java/runtime/src/main/java/org/ray/runtime/RayActorImpl.java @@ -67,7 +67,6 @@ public final class RayActorImpl implements RayActor, Externalizable { return taskCounter++; } - private UniqueId computeNextActorHandleId() { byte[] bytes = Sha1Digestor.digest(handleId.getBytes(), ++numForks); return new UniqueId(bytes); diff --git a/java/runtime/src/main/java/org/ray/runtime/WorkerContext.java b/java/runtime/src/main/java/org/ray/runtime/WorkerContext.java index 7708466ed..fdb507689 100644 --- a/java/runtime/src/main/java/org/ray/runtime/WorkerContext.java +++ b/java/runtime/src/main/java/org/ray/runtime/WorkerContext.java @@ -79,6 +79,7 @@ public class WorkerContext { UniqueId.NIL, 0, UniqueId.NIL, + 0, UniqueId.NIL, UniqueId.NIL, 0, diff --git a/java/runtime/src/main/java/org/ray/runtime/generated/TaskInfo.java b/java/runtime/src/main/java/org/ray/runtime/generated/TaskInfo.java index 011130960..f06992ea8 100644 --- a/java/runtime/src/main/java/org/ray/runtime/generated/TaskInfo.java +++ b/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 final class TaskInfo extends Table { 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 final class TaskInfo extends Table { 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 final class TaskInfo extends Table { 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 final class TaskInfo extends Table { 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 final class TaskInfo extends Table { //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; } diff --git a/java/runtime/src/main/java/org/ray/runtime/raylet/RayletClientImpl.java b/java/runtime/src/main/java/org/ray/runtime/raylet/RayletClientImpl.java index cd4f3fd31..b1932c08a 100644 --- a/java/runtime/src/main/java/org/ray/runtime/raylet/RayletClientImpl.java +++ b/java/runtime/src/main/java/org/ray/runtime/raylet/RayletClientImpl.java @@ -127,6 +127,7 @@ public class RayletClientImpl implements RayletClient { UniqueId parentTaskId = UniqueId.fromByteBuffer(info.parentTaskIdAsByteBuffer()); int parentCounter = info.parentCounter(); UniqueId actorCreationId = UniqueId.fromByteBuffer(info.actorCreationIdAsByteBuffer()); + int maxActorReconstructions = info.maxActorReconstructions(); UniqueId actorId = UniqueId.fromByteBuffer(info.actorIdAsByteBuffer()); UniqueId actorHandleId = UniqueId.fromByteBuffer(info.actorHandleIdAsByteBuffer()); int actorCounter = info.actorCounter(); @@ -162,8 +163,9 @@ public class RayletClientImpl implements RayletClient { FunctionDescriptor functionDescriptor = new FunctionDescriptor( info.functionDescriptor(0), info.functionDescriptor(1), info.functionDescriptor(2) ); - return new TaskSpec(driverId, taskId, parentTaskId, parentCounter, actorCreationId, actorId, - actorHandleId, actorCounter, args, returnIds, resources, functionDescriptor); + return new TaskSpec(driverId, taskId, parentTaskId, parentCounter, actorCreationId, + maxActorReconstructions, actorId, actorHandleId, actorCounter, args, returnIds, resources, + functionDescriptor); } private static ByteBuffer convertTaskSpecToFlatbuffer(TaskSpec task) { @@ -177,10 +179,11 @@ public class RayletClientImpl implements RayletClient { final int parentCounter = task.parentCounter; final int actorCreateIdOffset = fbb.createString(task.actorCreationId.toByteBuffer()); final int actorCreateDummyIdOffset = fbb.createString(task.actorId.toByteBuffer()); + final int maxActorReconstructions = task.maxActorReconstructions; final int actorIdOffset = fbb.createString(task.actorId.toByteBuffer()); final int actorHandleIdOffset = fbb.createString(task.actorHandleId.toByteBuffer()); final int actorCounter = task.actorCounter; - final int functionIdOffset = fbb.createString(UniqueId.NIL.toByteBuffer()); + final int functionIdOffset = fbb.createString(UniqueId.randomId().toByteBuffer()); // Serialize args int[] argsOffsets = new int[task.args.length]; for (int i = 0; i < argsOffsets.length; i++) { @@ -230,13 +233,24 @@ public class RayletClientImpl implements RayletClient { int functionDescriptorOffset = fbb.createVectorOfTables(functionDescriptorOffsets); int root = TaskInfo.createTaskInfo( - fbb, driverIdOffset, taskIdOffset, - parentTaskIdOffset, parentCounter, - actorCreateIdOffset, actorCreateDummyIdOffset, - actorIdOffset, actorHandleIdOffset, actorCounter, - false, functionIdOffset, - argsOffset, returnsOffset, requiredResourcesOffset, - requiredPlacementResourcesOffset, Language.JAVA, + fbb, + driverIdOffset, + taskIdOffset, + parentTaskIdOffset, + parentCounter, + actorCreateIdOffset, + actorCreateDummyIdOffset, + maxActorReconstructions, + actorIdOffset, + actorHandleIdOffset, + actorCounter, + false, + functionIdOffset, + argsOffset, + returnsOffset, + requiredResourcesOffset, + requiredPlacementResourcesOffset, + Language.JAVA, functionDescriptorOffset); fbb.finish(root); ByteBuffer buffer = fbb.dataBuffer(); diff --git a/java/runtime/src/main/java/org/ray/runtime/task/TaskSpec.java b/java/runtime/src/main/java/org/ray/runtime/task/TaskSpec.java index 864be3754..8988c933f 100644 --- a/java/runtime/src/main/java/org/ray/runtime/task/TaskSpec.java +++ b/java/runtime/src/main/java/org/ray/runtime/task/TaskSpec.java @@ -28,6 +28,8 @@ public class TaskSpec { // Id for createActor a target actor public final UniqueId actorCreationId; + public final int maxActorReconstructions; + // Actor ID of the task. This is the actor that this task is executed on // or NIL_ACTOR_ID if the task is just a normal task. public final UniqueId actorId; @@ -62,14 +64,15 @@ public class TaskSpec { } public TaskSpec(UniqueId driverId, UniqueId taskId, UniqueId parentTaskId, int parentCounter, - UniqueId actorCreationId, UniqueId actorId, UniqueId actorHandleId, int actorCounter, - FunctionArg[] args, UniqueId[] returnIds, + UniqueId actorCreationId, int maxActorReconstructions, UniqueId actorId, + UniqueId actorHandleId, int actorCounter, FunctionArg[] args, UniqueId[] returnIds, Map resources, FunctionDescriptor functionDescriptor) { this.driverId = driverId; this.taskId = taskId; this.parentTaskId = parentTaskId; this.parentCounter = parentCounter; this.actorCreationId = actorCreationId; + this.maxActorReconstructions = maxActorReconstructions; this.actorId = actorId; this.actorHandleId = actorHandleId; this.actorCounter = actorCounter; diff --git a/java/test/src/main/java/org/ray/api/test/ActorReconstructionTest.java b/java/test/src/main/java/org/ray/api/test/ActorReconstructionTest.java new file mode 100644 index 000000000..d78518162 --- /dev/null +++ b/java/test/src/main/java/org/ray/api/test/ActorReconstructionTest.java @@ -0,0 +1,69 @@ +package org.ray.api.test; + +import static org.ray.runtime.util.SystemUtil.pid; + +import java.io.IOException; +import java.util.HashMap; +import java.util.concurrent.TimeUnit; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.ray.api.Ray; +import org.ray.api.RayActor; +import org.ray.api.annotation.RayRemote; +import org.ray.api.options.ActorCreationOptions; + +@RunWith(MyRunner.class) +public class ActorReconstructionTest { + + @RayRemote() + public static class Counter { + + private int value = 0; + + public int increase(int delta) { + value += delta; + return value; + } + + public int getPid() { + return pid(); + } + } + + @Test + public void testActorReconstruction() throws InterruptedException, IOException { + ActorCreationOptions options = new ActorCreationOptions(new HashMap<>(), 1); + RayActor actor = Ray.createActor(Counter::new, options); + // Call increase 3 times. + for (int i = 0; i < 3; i++) { + Ray.call(Counter::increase, actor, 1).get(); + } + + // Kill the actor process. + int pid = Ray.call(Counter::getPid, actor).get(); + Runtime.getRuntime().exec("kill -9 " + pid); + // Wait for the actor to be killed. + TimeUnit.SECONDS.sleep(1); + + // Try calling increase on this actor again and check the value is now 4. + int value = Ray.call(Counter::increase, actor, 1).get(); + Assert.assertEquals(value, 4); + + // Kill the actor process again. + pid = Ray.call(Counter::getPid, actor).get(); + Runtime.getRuntime().exec("kill -9 " + pid); + TimeUnit.SECONDS.sleep(1); + + // Try calling increase on this actor again and this should fail. + try { + Ray.call(Counter::increase, actor, 1).get(); + Assert.fail("The above task didn't fail."); + } catch (StringIndexOutOfBoundsException e) { + // Raylet backend will put invalid data in task's result to indicate the task has failed. + // Thus, Java deserialization will fail and throw `StringIndexOutOfBoundsException`. + // TODO(hchen): we should use object's metadata to indicate task failure, + // instead of throwing this exception. + } + } +} diff --git a/python/ray/actor.py b/python/ray/actor.py index 66fecb118..1167f064a 100644 --- a/python/ray/actor.py +++ b/python/ray/actor.py @@ -271,12 +271,14 @@ class ActorClass(object): each actor method. """ - def __init__(self, modified_class, class_id, checkpoint_interval, num_cpus, - num_gpus, resources, actor_method_cpus): + def __init__(self, modified_class, class_id, checkpoint_interval, + max_reconstructions, num_cpus, num_gpus, resources, + actor_method_cpus): self._modified_class = modified_class self._class_id = class_id self._class_name = modified_class.__name__ self._checkpoint_interval = checkpoint_interval + self._max_reconstructions = max_reconstructions self._num_cpus = num_cpus self._num_gpus = num_gpus self._resources = resources @@ -413,6 +415,7 @@ class ActorClass(object): function_id, creation_args, actor_creation_id=actor_id, + max_actor_reconstructions=self._max_reconstructions, num_return_vals=1, resources=resources, placement_resources=actor_placement_resources) @@ -775,12 +778,19 @@ class ActorHandle(object): def make_actor(cls, num_cpus, num_gpus, resources, actor_method_cpus, - checkpoint_interval): + checkpoint_interval, max_reconstructions): if checkpoint_interval is None: checkpoint_interval = -1 + if max_reconstructions is None: + max_reconstructions = 0 if checkpoint_interval == 0: raise Exception("checkpoint_interval must be greater than 0.") + if not (ray_constants.NO_RECONSTRUCTION <= max_reconstructions <= + ray_constants.INFINITE_RECONSTRUCTION): + raise Exception("max_reconstructions must be in range [%d, %d]." % + (ray_constants.NO_RECONSTRUCTION, + ray_constants.INFINITE_RECONSTRUCTION)) # Modify the class to have an additional method that will be used for # terminating the worker. @@ -872,8 +882,9 @@ def make_actor(cls, num_cpus, num_gpus, resources, actor_method_cpus, class_id = _random_string() - return ActorClass(Class, class_id, checkpoint_interval, num_cpus, num_gpus, - resources, actor_method_cpus) + return ActorClass(Class, class_id, checkpoint_interval, + max_reconstructions, num_cpus, num_gpus, resources, + actor_method_cpus) ray.worker.global_worker.make_actor = make_actor diff --git a/python/ray/ray_constants.py b/python/ray/ray_constants.py index a1d5e1a76..82aa1617d 100644 --- a/python/ray/ray_constants.py +++ b/python/ray/ray_constants.py @@ -76,3 +76,8 @@ LOGGER_LEVEL = "info" LOGGER_LEVEL_CHOICES = ['debug', 'info', 'warning', 'error', 'critical'] LOGGER_LEVEL_HELP = ("The logging level threshold, choices=['debug', 'info'," " 'warning', 'error', 'critical'], default='info'") + +# A constant indicating that an actor doesn't need reconstructions. +NO_RECONSTRUCTION = 0 +# A constant indicating that an actor should be reconstructed infinite times. +INFINITE_RECONSTRUCTION = 2**30 diff --git a/python/ray/test/cluster_utils.py b/python/ray/test/cluster_utils.py index 41dc3b6cd..aff302efc 100644 --- a/python/ray/test/cluster_utils.py +++ b/python/ray/test/cluster_utils.py @@ -34,6 +34,7 @@ class Cluster(object): self.head_node = None self.worker_nodes = {} self.redis_address = None + self.connected = False if not initialize_head and connect: raise RuntimeError("Cannot connect to uninitialized cluster.") @@ -41,14 +42,19 @@ class Cluster(object): head_node_args = head_node_args or {} self.add_node(**head_node_args) if connect: - redis_password = head_node_args.get("redis_password") - output_info = ray.init( - redis_address=self.redis_address, - redis_password=redis_password) - logger.info(output_info) + self.connect(head_node_args) if shutdown_at_exit: atexit.register(self.shutdown) + def connect(self, head_node_args): + assert self.redis_address is not None + assert not self.connected + redis_password = head_node_args.get("redis_password") + output_info = ray.init( + redis_address=self.redis_address, redis_password=redis_password) + logger.info(output_info) + self.connected = True + def add_node(self, **override_kwargs): """Adds a node to the local Ray Cluster. @@ -83,7 +89,7 @@ class Cluster(object): process_dict_copy = services.all_processes.copy() for key in services.all_processes: services.all_processes[key] = [] - node = Node(process_dict_copy) + node = Node(address_info, process_dict_copy) self.head_node = node else: address_info = services.start_ray_node( @@ -93,7 +99,7 @@ class Cluster(object): process_dict_copy = services.all_processes.copy() for key in services.all_processes: services.all_processes[key] = [] - node = Node(process_dict_copy) + node = Node(address_info, process_dict_copy) self.worker_nodes[node] = address_info logger.info("Starting Node with raylet socket {}".format( address_info["raylet_socket_names"])) @@ -182,8 +188,9 @@ class Cluster(object): class Node(object): """Abstraction for a Ray node.""" - def __init__(self, process_dict): + def __init__(self, address_info, process_dict): # TODO(rliaw): Is there a unique identifier for a node? + self.address_info = address_info self.process_dict = process_dict def kill_plasma_store(self): @@ -224,3 +231,11 @@ class Node(object): def all_processes_alive(self): return not any(self.dead_processes()) + + def get_plasma_store_name(self): + """Return the plasma store name. + + Assuming one plasma store per raylet, this may be used as a unique + identifier for a node. + """ + return self.address_info['object_store_addresses'][0] diff --git a/python/ray/worker.py b/python/ray/worker.py index e9a6aac4e..06f7f03d2 100644 --- a/python/ray/worker.py +++ b/python/ray/worker.py @@ -525,6 +525,7 @@ class Worker(object): is_actor_checkpoint_method=False, actor_creation_id=None, actor_creation_dummy_object_id=None, + max_actor_reconstructions=0, execution_dependencies=None, num_return_vals=None, resources=None, @@ -622,12 +623,12 @@ class Worker(object): assert not self.current_task_id.is_nil() # Submit the task to local scheduler. task = ray.raylet.Task( - driver_id, ray.ObjectID( - function_id.id()), args_for_local_scheduler, - num_return_vals, self.current_task_id, task_index, - actor_creation_id, actor_creation_dummy_object_id, actor_id, - actor_handle_id, actor_counter, execution_dependencies, - resources, placement_resources) + driver_id, ray.ObjectID(function_id.id()), + args_for_local_scheduler, num_return_vals, + self.current_task_id, task_index, actor_creation_id, + actor_creation_dummy_object_id, max_actor_reconstructions, + actor_id, actor_handle_id, actor_counter, + execution_dependencies, resources, placement_resources) self.raylet_client.submit_task(task) return task.returns() @@ -2098,7 +2099,7 @@ def connect(info, worker.current_task_id, worker.task_index, ray.ObjectID(NIL_ACTOR_ID), - ray.ObjectID(NIL_ACTOR_ID), + ray.ObjectID(NIL_ACTOR_ID), 0, ray.ObjectID(NIL_ACTOR_ID), ray.ObjectID(NIL_ACTOR_ID), nil_actor_counter, [], {"CPU": 0}, {}) @@ -2512,6 +2513,7 @@ def make_decorator(num_return_vals=None, resources=None, max_calls=None, checkpoint_interval=None, + max_reconstructions=None, worker=None): def decorator(function_or_class): if (inspect.isfunction(function_or_class) @@ -2520,6 +2522,9 @@ def make_decorator(num_return_vals=None, if checkpoint_interval is not None: raise Exception("The keyword 'checkpoint_interval' is not " "allowed for remote functions.") + if max_reconstructions is not None: + raise Exception("The keyword 'max_reconstructions' is not " + "allowed for remote functions.") return ray.remote_function.RemoteFunction( function_or_class, num_cpus, num_gpus, resources, @@ -2549,7 +2554,7 @@ def make_decorator(num_return_vals=None, return worker.make_actor(function_or_class, cpus_to_use, num_gpus, resources, actor_method_cpus, - checkpoint_interval) + checkpoint_interval, max_reconstructions) raise Exception("The @ray.remote decorator must be applied to " "either a function or to a class.") @@ -2591,6 +2596,11 @@ def remote(*args, **kwargs): third-party libraries or to reclaim resources that cannot easily be released, e.g., GPU memory that was acquired by TensorFlow). By default this is infinite. + * **max_reconstructions**: Only for *actors*. This specifies the maximum + number of times that the actor should be reconstructed when it dies + unexpectedly. The minimum valid value is 0 (default), which indicates + that the actor doesn't need to be reconstructed. And the maximum valid + value is ray.ray_constants.INFINITE_RECONSTRUCTIONS. This can be done as follows: @@ -2616,14 +2626,15 @@ def remote(*args, **kwargs): "with no arguments and no parentheses, for example " "'@ray.remote', or it must be applied using some of " "the arguments 'num_return_vals', 'num_cpus', 'num_gpus', " - "'resources', 'max_calls', or 'checkpoint_interval', like " + "'resources', 'max_calls', 'checkpoint_interval'," + "or 'max_reconstructions', like " "'@ray.remote(num_return_vals=2, " "resources={\"CustomResource\": 1})'.") assert len(args) == 0 and len(kwargs) > 0, error_string for key in kwargs: assert key in [ "num_return_vals", "num_cpus", "num_gpus", "resources", - "max_calls", "checkpoint_interval" + "max_calls", "checkpoint_interval", "max_reconstructions" ], error_string num_cpus = kwargs["num_cpus"] if "num_cpus" in kwargs else None @@ -2641,6 +2652,7 @@ def remote(*args, **kwargs): num_return_vals = kwargs.get("num_return_vals") max_calls = kwargs.get("max_calls") checkpoint_interval = kwargs.get("checkpoint_interval") + max_reconstructions = kwargs.get("max_reconstructions") return make_decorator( num_return_vals=num_return_vals, @@ -2649,4 +2661,5 @@ def remote(*args, **kwargs): resources=resources, max_calls=max_calls, checkpoint_interval=checkpoint_interval, + max_reconstructions=max_reconstructions, worker=worker) diff --git a/src/ray/gcs/format/gcs.fbs b/src/ray/gcs/format/gcs.fbs index 6414613d4..99207d9b4 100644 --- a/src/ray/gcs/format/gcs.fbs +++ b/src/ray/gcs/format/gcs.fbs @@ -60,6 +60,9 @@ table TaskInfo { actor_creation_id: string; // The dummy object ID of the actor creation task if this is an actor method. actor_creation_dummy_object_id: string; + // The max number of times this actor should be recontructed. + // If this number of 0 or negative, the actor won't be reconstructed on failure. + max_actor_reconstructions: int; // Actor ID of the task. This is the actor that this task is executed on // or NIL_ACTOR_ID if the task is just a normal task. actor_id: string; @@ -167,8 +170,11 @@ table ClassTableData { enum ActorState:int { // Actor is alive. ALIVE = 0, + // Actor is dead, now being reconstructed. + // After reconstruction finishes, the state will become alive again. + RECONSTRUCTING = 1, // Actor is already dead and won't be reconstructed. - DEAD + DEAD = 2 } table ActorTableData { @@ -184,6 +190,10 @@ table ActorTableData { node_manager_id: string; // Current state of this actor. state: ActorState; + // Max number of times this actor should be reconstructed. + max_reconstructions: int; + // Remaining number of reconstructions. + remaining_reconstructions: int; } table ErrorTableData { diff --git a/src/ray/gcs/tables.h b/src/ray/gcs/tables.h index 5cca066fb..20a5b3db2 100644 --- a/src/ray/gcs/tables.h +++ b/src/ray/gcs/tables.h @@ -393,7 +393,11 @@ class FunctionTable : public Table { using ClassTable = Table; -// TODO(swang): Set the pubsub channel for the actor table. +/// Actor table starts with an ALIVE entry, which represents the first time the actor +/// is created. This may be followed by 0 or more pairs of RECONSTRUCTING, ALIVE entries, +/// which represent each time the actor fails (RECONSTRUCTING) and gets recreated (ALIVE). +/// These may be followed by a DEAD entry, which means that the actor has failed and will +/// not be reconstructed. class ActorTable : public Log { public: ActorTable(const std::vector> &contexts, diff --git a/src/ray/raylet/actor_registration.cc b/src/ray/raylet/actor_registration.cc index 7ea95e656..aecf55eff 100644 --- a/src/ray/raylet/actor_registration.cc +++ b/src/ray/raylet/actor_registration.cc @@ -25,6 +25,18 @@ const ObjectID ActorRegistration::GetExecutionDependency() const { return execution_dependency_; } +const DriverID ActorRegistration::GetDriverId() const { + return DriverID::from_binary(actor_table_data_.driver_id); +} + +const int64_t ActorRegistration::GetMaxReconstructions() const { + return actor_table_data_.max_reconstructions; +} + +const int64_t ActorRegistration::GetRemainingReconstructions() const { + return actor_table_data_.remaining_reconstructions; +} + const std::unordered_map &ActorRegistration::GetFrontier() const { return frontier_; @@ -39,10 +51,6 @@ void ActorRegistration::ExtendFrontier(const ActorHandleID &handle_id, dummy_objects_.push_back(execution_dependency); } -bool ActorRegistration::IsAlive() const { - return actor_table_data_.state == ActorState::ALIVE; -} - int ActorRegistration::NumHandles() const { return frontier_.size(); } } // namespace raylet diff --git a/src/ray/raylet/actor_registration.h b/src/ray/raylet/actor_registration.h index 4cf9b110a..9c4664455 100644 --- a/src/ray/raylet/actor_registration.h +++ b/src/ray/raylet/actor_registration.h @@ -46,6 +46,9 @@ class ActorRegistration { /// \return The actor's current state. const ActorState &GetState() const { return actor_table_data_.state; } + /// Update actor's state. + void SetState(const ActorState &state) { actor_table_data_.state = state; } + /// Get the actor's node manager location. /// /// \return The actor's node manager location. All tasks for the actor should @@ -59,6 +62,15 @@ class ActorRegistration { /// \return The execution dependency returned by the actor's creation task. const ObjectID GetActorCreationDependency() const; + /// Get actor's driver ID. + const DriverID GetDriverId() const; + + /// Get the max number of times this actor should be reconstructed. + const int64_t GetMaxReconstructions() const; + + /// Get the remaining number of times this actor should be reconstructed. + const int64_t GetRemainingReconstructions() const; + /// Get the object that represents the actor's current state. This is the /// execution dependency returned by the task most recently executed on the /// actor. The next task to execute on the actor should be marked as @@ -88,12 +100,6 @@ class ActorRegistration { void ExtendFrontier(const ActorHandleID &handle_id, const ObjectID &execution_dependency); - /// Return whether the actor is alive or not. This should only be called on - /// local actors. - /// - /// \return True if the local actor is alive and false if it is dead. - bool IsAlive() const; - /// Returns num handles to this actor entry. /// /// \return int. @@ -111,6 +117,7 @@ class ActorRegistration { /// executed so far and which tasks may execute next, based on execution /// dependencies. This is indexed by handle. std::unordered_map frontier_; + /// All of the dummy object IDs from this actor's tasks. std::vector dummy_objects_; }; diff --git a/src/ray/raylet/lib/python/common_extension.cc b/src/ray/raylet/lib/python/common_extension.cc index 2ac9d2caf..88a905012 100644 --- a/src/ray/raylet/lib/python/common_extension.cc +++ b/src/ray/raylet/lib/python/common_extension.cc @@ -377,6 +377,9 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) { ActorID actor_creation_id = ActorID::nil(); // The dummy object for the actor creation task (if this is an actor method). ObjectID actor_creation_dummy_object_id = ObjectID::nil(); + // Max number of times to reconstruct this actor (only used for actor creation + // task). + int32_t max_actor_reconstructions; // Arguments of the task that are execution-dependent. These must be // PyObjectIDs). PyObject *execution_arguments = nullptr; @@ -384,13 +387,14 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) { PyObject *resource_map = nullptr; // Dictionary of required placement resources for this task. PyObject *placement_resource_map = nullptr; - if (!PyArg_ParseTuple(args, "O&O&OiO&i|O&O&O&O&iOOO", &PyObjectToUniqueID, &driver_id, + if (!PyArg_ParseTuple(args, "O&O&OiO&i|O&O&iO&O&iOOO", &PyObjectToUniqueID, &driver_id, &PyObjectToUniqueID, &function_id, &arguments, &num_returns, &PyObjectToUniqueID, &parent_task_id, &parent_counter, &PyObjectToUniqueID, &actor_creation_id, &PyObjectToUniqueID, - &actor_creation_dummy_object_id, &PyObjectToUniqueID, &actor_id, - &PyObjectToUniqueID, &actor_handle_id, &actor_counter, - &execution_arguments, &resource_map, &placement_resource_map)) { + &actor_creation_dummy_object_id, &max_actor_reconstructions, + &PyObjectToUniqueID, &actor_id, &PyObjectToUniqueID, + &actor_handle_id, &actor_counter, &execution_arguments, + &resource_map, &placement_resource_map)) { return -1; } @@ -439,9 +443,9 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) { self->task_spec = new ray::raylet::TaskSpecification( driver_id, parent_task_id, parent_counter, actor_creation_id, - actor_creation_dummy_object_id, actor_id, actor_handle_id, actor_counter, - function_id, task_args, num_returns, required_resources, - required_placement_resources, Language::PYTHON); + actor_creation_dummy_object_id, max_actor_reconstructions, actor_id, + actor_handle_id, actor_counter, function_id, task_args, num_returns, + required_resources, required_placement_resources, Language::PYTHON); /* Set the task's execution dependencies. */ self->execution_dependencies = new std::vector(); diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index ea8baa1ad..e5cd2a7df 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -134,15 +134,19 @@ ray::Status NodeManager::RegisterGcs() { JobID::nil(), gcs_client_->client_table().GetLocalClientId(), task_lease_notification_callback, task_lease_empty_callback, nullptr)); - // Register a callback for actor creation notifications. - auto actor_creation_callback = [this](gcs::AsyncGcsClient *client, - const ActorID &actor_id, - const std::vector &data) { - HandleActorStateTransition(actor_id, data.back()); + // Register a callback to handle actor notifications. + auto actor_notification_callback = [this](gcs::AsyncGcsClient *client, + const ActorID &actor_id, + const std::vector &data) { + if (!data.empty()) { + // We only need the last entry, because it represents the latest state of + // this actor. + HandleActorStateTransition(actor_id, data.back()); + } }; RAY_RETURN_NOT_OK(gcs_client_->actor_table().Subscribe( - UniqueID::nil(), UniqueID::nil(), actor_creation_callback, nullptr)); + UniqueID::nil(), UniqueID::nil(), actor_notification_callback, nullptr)); // Register a callback on the client table for new clients. auto node_manager_client_added = [this](gcs::AsyncGcsClient *client, const UniqueID &id, @@ -405,8 +409,12 @@ void NodeManager::ClientRemoved(const ClientTableDataT &client_data) { // TODO(swang): This could be very slow if there are many actors. for (const auto &actor_entry : actor_registry_) { if (actor_entry.second.GetNodeManagerId() == client_id && - actor_entry.second.IsAlive()) { - HandleDisconnectedActor(actor_entry.first, /*was_local=*/false); + actor_entry.second.GetState() == ActorState::ALIVE) { + RAY_LOG(INFO) << "Actor " << actor_entry.first + << " is disconnected, because its node " << client_id + << " is removed from cluster. It may be reconstructed."; + HandleDisconnectedActor(actor_entry.first, /*was_local=*/false, + /*intentional_disconnect=*/false); } } } @@ -467,53 +475,49 @@ void NodeManager::HeartbeatBatchAdded(const HeartbeatBatchTableDataT &heartbeat_ } } -void NodeManager::HandleDisconnectedActor(const ActorID &actor_id, bool was_local) { - RAY_LOG(DEBUG) << "Actor disconnected " << actor_id; - auto actor_entry = actor_registry_.find(actor_id); - RAY_CHECK(actor_entry != actor_registry_.end()); +void NodeManager::PublishActorStateTransition( + const ActorID &actor_id, const ActorTableDataT &data, + const ray::gcs::ActorTable::WriteCallback &failure_callback) { + // Copy the actor notification data. + auto actor_notification = std::make_shared(data); - // Release all the dummy objects for the dead actor. - if (was_local) { - for (auto &dummy_object : actor_entry->second.GetDummyObjects()) { - HandleObjectMissing(dummy_object); - } + // The actor log starts with an ALIVE entry. This is followed by 0 to N pairs + // of (RECONSTRUCTING, ALIVE) entries, where N is the maximum number of + // reconstructions. This is followed optionally by a DEAD entry. + int log_length = 2 * (actor_notification->max_reconstructions - + actor_notification->remaining_reconstructions); + if (actor_notification->state != ActorState::ALIVE) { + // RECONSTRUCTING or DEAD entries have an odd index. + log_length += 1; } - - auto new_actor_data = - std::make_shared(actor_entry->second.GetTableData()); - new_actor_data->state = ActorState::DEAD; - HandleActorStateTransition(actor_id, *new_actor_data); - ray::gcs::ActorTable::WriteCallback failure_callback = nullptr; - if (was_local) { - // The actor was local to this node, so we are the only one who should try - // to update the log. - failure_callback = [](gcs::AsyncGcsClient *client, const ActorID &id, - const ActorTableDataT &data) { - RAY_LOG(FATAL) << "Failed to update state to DEAD for actor " << id; - }; - } - // Actor reconstruction is disabled, so the actor can only go from ALIVE to - // DEAD. The DEAD entry must therefore be at the second index in the log. - RAY_CHECK_OK(gcs_client_->actor_table().AppendAt(JobID::nil(), actor_id, new_actor_data, - nullptr, failure_callback, - /*log_index=*/1)); + RAY_CHECK_OK(gcs_client_->actor_table().AppendAt( + JobID::nil(), actor_id, actor_notification, nullptr, failure_callback, log_length)); } void NodeManager::HandleActorStateTransition(const ActorID &actor_id, const ActorTableDataT &data) { - RAY_LOG(DEBUG) << "Actor creation notification received: " << actor_id << " " - << static_cast(data.state); - - // Register the new actor. ActorRegistration actor_registration(data); + RAY_LOG(DEBUG) << "Actor notification received: actor_id = " << actor_id + << ", node_manager_id = " << actor_registration.GetNodeManagerId() + << ", state = " << static_cast(actor_registration.GetState()) + << ", remaining_reconstructions = " + << actor_registration.GetRemainingReconstructions(); // Update local registry. auto it = actor_registry_.find(actor_id); if (it == actor_registry_.end()) { it = actor_registry_.emplace(actor_id, actor_registration).first; } else { - RAY_CHECK(it->second.GetNodeManagerId() == actor_registration.GetNodeManagerId()); - if (actor_registration.GetState() > it->second.GetState()) { - // The new state is later than our current state. + // Only process the state transition if it is to a later state than ours. + if (actor_registration.GetState() > it->second.GetState() && + actor_registration.GetRemainingReconstructions() == + it->second.GetRemainingReconstructions()) { + // The new state is later than ours if it is about the same lifetime, but + // a greater state. + it->second = actor_registration; + } else if (actor_registration.GetRemainingReconstructions() < + it->second.GetRemainingReconstructions()) { + // The new state is also later than ours it is about a later lifetime of + // the actor. it->second = actor_registration; } else { // Our state is already at or past the update, so skip the update. @@ -521,7 +525,7 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id, } } - if (it->second.IsAlive()) { + if (actor_registration.GetState() == ActorState::ALIVE) { // The actor's location is now known. Dequeue any methods that were // submitted before the actor's location was known. // (See design_docs/task_states.rst for the state transition diagram.) @@ -543,9 +547,9 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id, } // Maintain the invariant that if a task is in the // MethodsWaitingForActorCreation queue, then it is subscribed to its - // respective actor creation task and that task only. Since the actor - // location is now known, we can remove the task from the queue and - // forget its dependency on the actor creation task. + // respective actor creation task. Since the actor location is now known, + // we can remove the task from the queue and forget its dependency on the + // actor creation task. RAY_CHECK(task_dependency_manager_.UnsubscribeDependencies( method.GetTaskSpecification().TaskId())); // The task's uncommitted lineage was already added to the local lineage @@ -553,7 +557,7 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id, // empty lineage this time. SubmitTask(method, Lineage()); } - } else { + } else if (actor_registration.GetState() == ActorState::DEAD) { // When an actor dies, loop over all of the queued tasks for that actor // and treat them as failed. auto tasks_to_remove = local_queues_.GetTaskIdsForActor(actor_id); @@ -561,6 +565,17 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id, for (auto const &task : removed_tasks) { TreatTaskAsFailed(task); } + } else { + RAY_CHECK(actor_registration.GetState() == ActorState::RECONSTRUCTING); + RAY_LOG(DEBUG) << "Actor is being reconstructed: " << actor_id; + // When an actor fails but can be reconstructed, resubmit all of the queued + // tasks for that actor. This will mark the tasks as waiting for actor + // creation. + auto tasks_to_remove = local_queues_.GetTaskIdsForActor(actor_id); + auto removed_tasks = local_queues_.RemoveTasks(tasks_to_remove); + for (auto const &task : removed_tasks) { + SubmitTask(task, Lineage()); + } } } @@ -640,7 +655,7 @@ void NodeManager::ProcessClientMessage( return; } break; case protocol::MessageType::IntentionalDisconnectClient: { - ProcessDisconnectClientMessage(client, /* push_warning = */ false); + ProcessDisconnectClientMessage(client, /* intentional_disconnect = */ true); // We don't need to receive future messages from this client, // because it's already disconnected. return; @@ -702,6 +717,50 @@ void NodeManager::ProcessRegisterClientRequestMessage( } } +void NodeManager::HandleDisconnectedActor(const ActorID &actor_id, bool was_local, + bool intentional_disconnect) { + auto actor_entry = actor_registry_.find(actor_id); + RAY_CHECK(actor_entry != actor_registry_.end()); + auto &actor_registration = actor_entry->second; + RAY_LOG(DEBUG) << "The actor with ID " << actor_id << " died " + << (intentional_disconnect ? "intentionally" : "unintentionally") + << ", remaining reconstructions = " + << actor_registration.GetRemainingReconstructions(); + + // Check if this actor needs to be reconstructed. + ActorState new_state = + actor_registration.GetRemainingReconstructions() > 0 && !intentional_disconnect + ? ActorState::RECONSTRUCTING + : ActorState::DEAD; + if (was_local) { + // Clean up the dummy objects from this actor. + RAY_LOG(DEBUG) << "Removing dummy objects for actor: " << actor_id; + for (auto &id : actor_entry->second.GetDummyObjects()) { + HandleObjectMissing(id); + } + } + // Update the actor's state. + ActorTableDataT new_actor_data = actor_entry->second.GetTableData(); + new_actor_data.state = new_state; + if (was_local) { + // If the actor was local, immediately update the state in actor registry. + // So if we receive any actor tasks before we receive GCS notification, + // these tasks can be correctly routed to the `MethodsWaitingForActorCreation` queue, + // instead of being assigned to the dead actor. + HandleActorStateTransition(actor_id, new_actor_data); + } + ray::gcs::ActorTable::WriteCallback failure_callback = nullptr; + if (was_local) { + failure_callback = [](gcs::AsyncGcsClient *client, const ActorID &id, + const ActorTableDataT &data) { + // If the disconnected actor was local, only this node will try to update actor + // state. So the update shouldn't fail. + RAY_LOG(FATAL) << "Failed to update state for actor " << id; + }; + } + PublishActorStateTransition(actor_id, new_actor_data, failure_callback); +} + void NodeManager::ProcessGetTaskMessage( const std::shared_ptr &client) { std::shared_ptr worker = worker_pool_.GetRegisteredWorker(client); @@ -721,7 +780,7 @@ void NodeManager::ProcessGetTaskMessage( } void NodeManager::ProcessDisconnectClientMessage( - const std::shared_ptr &client, bool push_warning) { + const std::shared_ptr &client, bool intentional_disconnect) { std::shared_ptr worker = worker_pool_.GetRegisteredWorker(client); bool is_worker = false, is_driver = false; if (worker) { @@ -768,7 +827,7 @@ void NodeManager::ProcessDisconnectClientMessage( const JobID &job_id = worker->GetAssignedDriverId(); - if (push_warning) { + if (!intentional_disconnect) { // TODO(rkn): Define this constant somewhere else. std::string type = "worker_died"; std::ostringstream error_message; @@ -786,7 +845,7 @@ void NodeManager::ProcessDisconnectClientMessage( if (!actor_id.is_nil()) { RAY_LOG(DEBUG) << "The actor with ID " << actor_id << " died on " << gcs_client_->client_table().GetLocalClientId(); - HandleDisconnectedActor(actor_id, /*was_local=*/true); + HandleDisconnectedActor(actor_id, /*was_local=*/true, intentional_disconnect); } const ClientID &client_id = gcs_client_->client_table().GetLocalClientId(); @@ -1064,7 +1123,7 @@ void NodeManager::TreatTaskAsFailed(const Task &task) { // Loop over the return IDs (except the dummy ID) and store a fake object in // the object store. int64_t num_returns = spec.NumReturns(); - if (spec.IsActorTask()) { + if (spec.IsActorCreationTask() || spec.IsActorTask()) { // TODO(rkn): We subtract 1 to avoid the dummy ID. However, this leaks // information about the TaskSpecification implementation. num_returns -= 1; @@ -1100,7 +1159,12 @@ void NodeManager::TreatTaskAsFailed(const Task &task) { void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineage, bool forwarded) { - const TaskID &task_id = task.GetTaskSpecification().TaskId(); + const TaskSpecification &spec = task.GetTaskSpecification(); + const TaskID &task_id = spec.TaskId(); + RAY_LOG(DEBUG) << "Submitting task: task_id = " << task_id + << ", actor_id = " << spec.ActorId() + << ", actor_creation_id = " << spec.ActorCreationId(); + if (local_queues_.HasTask(task_id)) { RAY_LOG(WARNING) << "Submitted task " << task_id << " is already queued and will not be reconstructed. This is most " @@ -1115,49 +1179,58 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag << " already in lineage cache. This is most likely due to reconstruction."; } - const TaskSpecification &spec = task.GetTaskSpecification(); if (spec.IsActorTask()) { // Check whether we know the location of the actor. const auto actor_entry = actor_registry_.find(spec.ActorId()); - if (actor_entry != actor_registry_.end()) { - if (!actor_entry->second.IsAlive()) { + bool seen = actor_entry != actor_registry_.end(); + // If we have already seen this actor and this actor is not being reconstructed, + // its location is known. + bool location_known = + seen && actor_entry->second.GetState() != ActorState::RECONSTRUCTING; + if (location_known) { + if (actor_entry->second.GetState() == ActorState::DEAD) { + // If this actor is dead, either because the actor process is dead + // or because its residing node is dead, treat this task as failed. TreatTaskAsFailed(task); } else { - // We have a known location for the actor. + // If this actor is alive, check whether this actor is local. auto node_manager_id = actor_entry->second.GetNodeManagerId(); if (node_manager_id == gcs_client_->client_table().GetLocalClientId()) { - // Queue the task for local execution, bypassing placement. + // If this actor is local, queue the task for local execution, bypassing + // placement. EnqueuePlaceableTask(task); } else { - // If the node manager has been removed, then it must have already been - // marked as DEAD in the handler for a removed GCS client. - RAY_CHECK(!gcs_client_->client_table().IsRemoved(node_manager_id)); - // The actor is remote. Attempt to forward the task to the node manager - // that owns the actor. If this fails to forward the task, the task - // will be resubmitted locally. + // The actor is remote. Forward the task to the node manager that owns + // the actor. + // Attempt to forward the task. If this fails to forward the task, + // the task will be resubmit locally. ForwardTaskOrResubmit(task, node_manager_id); } } } else { - // We do not have a registered location for the object, so either the - // actor has not yet been created or we missed the notification for the - // actor creation because this node joined the cluster after the actor - // was already created. Look up the actor's registered location in case - // we missed the creation notification. - // NOTE(swang): This codepath needs to be tested in a cluster setting. - auto lookup_callback = [this](gcs::AsyncGcsClient *client, const ActorID &actor_id, - const std::vector &data) { - if (!data.empty()) { - // The actor has been created. - HandleActorStateTransition(actor_id, data.back()); - } else { - // The actor has not yet been created. - // TODO(swang): Set a timer for reconstructing the actor creation - // task. - } - }; - RAY_CHECK_OK(gcs_client_->actor_table().Lookup(JobID::nil(), spec.ActorId(), - lookup_callback)); + ObjectID actor_creation_dummy_object; + if (!seen) { + // We do not have a registered location for the object, so either the + // actor has not yet been created or we missed the notification for the + // actor creation because this node joined the cluster after the actor + // was already created. Look up the actor's registered location in case + // we missed the creation notification. + auto lookup_callback = [this](gcs::AsyncGcsClient *client, + const ActorID &actor_id, + const std::vector &data) { + if (!data.empty()) { + // The actor has been created. We only need the last entry, because + // it represents the latest state of this actor. + HandleActorStateTransition(actor_id, data.back()); + } + }; + RAY_CHECK_OK(gcs_client_->actor_table().Lookup(JobID::nil(), spec.ActorId(), + lookup_callback)); + actor_creation_dummy_object = spec.ActorCreationDummyObjectId(); + } else { + actor_creation_dummy_object = actor_entry->second.GetActorCreationDependency(); + } + // Keep the task queued until we discover the actor's location. // (See design_docs/task_states.rst for the state transition diagram.) local_queues_.QueueMethodsWaitingForActorCreation({task}); @@ -1169,7 +1242,7 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag // waiting queue, the caller must make the corresponding call to // UnsubscribeDependencies. task_dependency_manager_.SubscribeDependencies(spec.TaskId(), - {spec.ActorCreationDummyObjectId()}); + {actor_creation_dummy_object}); // Mark the task as pending. It will be canceled once we discover the // actor's location and either execute the task ourselves or forward it // to another node. @@ -1443,38 +1516,47 @@ void NodeManager::FinishAssignedTask(Worker &worker) { // If this was an actor creation task, then convert the worker to an actor. auto actor_id = task.GetTaskSpecification().ActorCreationId(); worker.AssignActorId(actor_id); - const auto driver_id = task.GetTaskSpecification().DriverId(); - // Publish the actor creation event to all other nodes so that methods for // the actor will be forwarded directly to this node. - RAY_CHECK(actor_registry_.find(actor_id) == actor_registry_.end()) - << "Created an actor that already exists"; - auto actor_data = std::make_shared(); - actor_data->actor_id = actor_id.binary(); - actor_data->actor_creation_dummy_object_id = - task.GetTaskSpecification().ActorDummyObject().binary(); - actor_data->driver_id = driver_id.binary(); - actor_data->node_manager_id = gcs_client_->client_table().GetLocalClientId().binary(); - actor_data->state = ActorState::ALIVE; + auto actor_entry = actor_registry_.find(actor_id); + ActorTableDataT new_actor_data; + if (actor_entry == actor_registry_.end()) { + // Set all of the static fields for the actor. These fields will not + // change even if the actor fails or is reconstructed. + new_actor_data.actor_id = actor_id.binary(); + new_actor_data.actor_creation_dummy_object_id = + task.GetTaskSpecification().ActorDummyObject().binary(); + new_actor_data.driver_id = task.GetTaskSpecification().DriverId().binary(); + new_actor_data.max_reconstructions = + task.GetTaskSpecification().MaxActorReconstructions(); + // This is the first time that the actor has been created, so the number + // of remaining reconstructions is the max. + new_actor_data.remaining_reconstructions = + task.GetTaskSpecification().MaxActorReconstructions(); + } else { + // If we've already seen this actor, it means that this actor was reconstructed. + // Thus, its previous state must be RECONSTRUCTING. + RAY_CHECK(actor_entry->second.GetState() == ActorState::RECONSTRUCTING); + // Copy the static fields from the current actor entry. + new_actor_data = actor_entry->second.GetTableData(); + // We are reconstructing the actor, so subtract its + // remaining_reconstructions by 1. + new_actor_data.remaining_reconstructions--; + } - RAY_LOG(DEBUG) << "Publishing actor creation: " << actor_id - << " driver_id: " << driver_id; - HandleActorStateTransition(actor_id, *actor_data); - // The actor should not have been created before, so writing to the first - // index in the log should succeed. - auto failure_callback = [](gcs::AsyncGcsClient *client, const ActorID &id, - const ActorTableDataT &data) { - // TODO(swang): Instead of making this a fatal check, we could just kill - // the duplicate actor process. If we do this, we must make sure to - // either resubmit the tasks that went to the duplicate actor, or wait - // for success before handling the actor state transition to ALIVE. - RAY_LOG(FATAL) << "Failed to update state to ALIVE for actor " << id; - }; - RAY_CHECK_OK(gcs_client_->actor_table().AppendAt( - JobID::nil(), actor_id, actor_data, nullptr, failure_callback, /*log_index=*/0)); - - // Resources required by an actor creation task are acquired for the - // lifetime of the actor, so we do not release any resources here. + // Set the new fields for the actor's state to indicate that the actor is + // now alive on this node manager. + new_actor_data.node_manager_id = + gcs_client_->client_table().GetLocalClientId().binary(); + new_actor_data.state = ActorState::ALIVE; + HandleActorStateTransition(actor_id, new_actor_data); + PublishActorStateTransition( + actor_id, new_actor_data, + /*failure_callback=*/ + [](gcs::AsyncGcsClient *client, const ActorID &id, const ActorTableDataT &data) { + // Only one node at a time should succeed at creating the actor. + RAY_LOG(FATAL) << "Failed to update state to ALIVE for actor " << id; + }); } else { // Release task's resources. local_available_resources_.Release(worker.GetTaskResourceIds()); @@ -1488,8 +1570,6 @@ void NodeManager::FinishAssignedTask(Worker &worker) { // If the finished task was an actor task, mark the returned dummy object as // locally available. This is not added to the object table, so the update // will be invisible to both the local object manager and the other nodes. - // NOTE(swang): These objects are never cleaned up. We should consider - // removing the objects, e.g., when an actor is terminated. if (task.GetTaskSpecification().IsActorCreationTask() || task.GetTaskSpecification().IsActorTask()) { ActorID actor_id; @@ -1557,23 +1637,22 @@ void NodeManager::HandleTaskReconstruction(const TaskID &task_id) { } void NodeManager::ResubmitTask(const Task &task) { - if (task.GetTaskSpecification().IsActorTask()) { - // Actor reconstruction is turned off by default right now. - const ActorID actor_id = task.GetTaskSpecification().ActorId(); - auto it = actor_registry_.find(actor_id); - RAY_CHECK(it != actor_registry_.end()); - if (it->second.IsAlive()) { - // If the actor is still alive, then do not resubmit. - RAY_LOG(ERROR) << "The output of an actor task is required, but the actor may " - "still be alive. If the output has been evicted, the job may " - "hang."; + RAY_LOG(DEBUG) << "Attempting to resubmit task " + << task.GetTaskSpecification().TaskId(); + + // Actors should only be recreated if the first initialization failed or if + // the most recent instance of the actor failed. + if (task.GetTaskSpecification().IsActorCreationTask()) { + const auto &actor_id = task.GetTaskSpecification().ActorCreationId(); + const auto it = actor_registry_.find(actor_id); + if (it != actor_registry_.end() && it->second.GetState() == ActorState::ALIVE) { + // If the actor is still alive, then do not resubmit the task. If the + // actor actually is dead and a result is needed, then reconstruction + // for this task will be triggered again. + RAY_LOG(WARNING) + << "Actor creation task resubmitted, but the actor is still alive."; return; } - // The actor is dead. The actor task will get resubmitted, at which point - // it will be treated as failed. - } else { - RAY_LOG(INFO) << "Reconstructing task " << task.GetTaskSpecification().TaskId() - << " on client " << gcs_client_->client_table().GetLocalClientId(); } // Driver tasks cannot be reconstructed. If this is a driver task, push an @@ -1591,6 +1670,8 @@ void NodeManager::ResubmitTask(const Task &task) { return; } + RAY_LOG(INFO) << "Resubmitting task " << task.GetTaskSpecification().TaskId() + << " on client " << gcs_client_->client_table().GetLocalClientId(); // The task may be reconstructed. Submit it with an empty lineage, since any // uncommitted lineage must already be in the lineage cache. At this point, // the task should not yet exist in the local scheduling queue. If it does, @@ -1609,6 +1690,7 @@ void NodeManager::HandleObjectLocal(const ObjectID &object_id) { // First filter out the tasks that should not be moved to READY. local_queues_.FilterState(ready_task_id_set, TaskState::BLOCKED); local_queues_.FilterState(ready_task_id_set, TaskState::DRIVER); + local_queues_.FilterState(ready_task_id_set, TaskState::WAITING_FOR_ACTOR_CREATION); // Make sure that the remaining tasks are all WAITING. auto ready_task_id_set_copy = ready_task_id_set; @@ -1793,10 +1875,13 @@ std::string NodeManager::DebugString() const { result << "\nActorRegistry:"; int live_actors = 0; int dead_actors = 0; + int reconstructing_actors = 0; int max_num_handles = 0; for (auto &pair : actor_registry_) { - if (pair.second.IsAlive()) { + if (pair.second.GetState() == ActorState::ALIVE) { live_actors += 1; + } else if (pair.second.GetState() == ActorState::RECONSTRUCTING) { + reconstructing_actors += 1; } else { dead_actors += 1; } @@ -1805,6 +1890,7 @@ std::string NodeManager::DebugString() const { } } result << "\n- num live actors: " << live_actors; + result << "\n- num reconstructing actors: " << live_actors; result << "\n- num dead actors: " << dead_actors; result << "\n- max num handles: " << max_num_handles; result << "\nRemoteConnections:"; diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index fd73ddec4..951fc2fa7 100644 --- a/src/ray/raylet/node_manager.h +++ b/src/ray/raylet/node_manager.h @@ -260,22 +260,25 @@ class NodeManager { /// \return Void. void KillWorker(std::shared_ptr worker); - /// Methods for actor scheduling. - /// Handler for an actor state transition, for a newly created actor or an - /// actor that died. This method is idempotent and will ignore old state - /// transitions. + /// The callback for handling an actor state transition (e.g., from ALIVE to + /// DEAD), whether as a notification from the actor table or as a handler for + /// a local actor's state transition. This method is idempotent and will ignore + /// old state transition. /// - /// \param actor_id The actor ID of the actor that was created. - /// \param data Data associated with the actor state transition. + /// \param actor_id The actor ID of the actor whose state was updated. + /// \param data Data associated with this notification. /// \return Void. void HandleActorStateTransition(const ActorID &actor_id, const ActorTableDataT &data); - /// Handler for an actor dying. The actor may be remote. + /// Publish an actor's state transition to all other nodes. /// - /// \param actor_id The actor ID of the actor that died. - /// \param was_local Whether the actor was local. - /// \return Void. - void HandleDisconnectedActor(const ActorID &actor_id, bool was_local); + /// \param actor_id The actor ID of the actor whose state was updated. + /// \param data Data to publish. + /// \param failure_callback An optional callback to call if the publish is + /// unsuccessful. + void PublishActorStateTransition( + const ActorID &actor_id, const ActorTableDataT &data, + const ray::gcs::ActorTable::WriteCallback &failure_callback); /// When a driver dies, loop over all of the queued tasks for that driver and /// treat them as failed. @@ -332,10 +335,11 @@ class NodeManager { /// client. /// /// \param client The client that sent the message. - /// \param push_warning Propogate error message if true. + /// \param intentional_disconnect Wether the client was intentionally disconnected. /// \return Void. void ProcessDisconnectClientMessage( - const std::shared_ptr &client, bool push_warning = true); + const std::shared_ptr &client, + bool intentional_disconnect = false); /// Process client message of SubmitTask /// @@ -365,6 +369,18 @@ class NodeManager { /// \return Void. void ProcessPushErrorRequestMessage(const uint8_t *message_data); + /// Handle the case where an actor is disconnected, determine whether this + /// actor needs to be reconstructed and then update actor table. + /// This function needs to be called either when actor process dies or when + /// a node dies. + /// + /// \param actor_id Id of this actor. + /// \param was_local Whether the disconnected was on this local node. + /// \param intentional_disconnect Wether the client was intentionally disconnected. + /// \return Void. + void HandleDisconnectedActor(const ActorID &actor_id, bool was_local, + bool intentional_disconnect); + boost::asio::io_service &io_service_; ObjectManager &object_manager_; /// A Plasma object store client. This is used exclusively for creating new diff --git a/src/ray/raylet/scheduling_queue.cc b/src/ray/raylet/scheduling_queue.cc index b9045e579..47c353a9a 100644 --- a/src/ray/raylet/scheduling_queue.cc +++ b/src/ray/raylet/scheduling_queue.cc @@ -36,8 +36,7 @@ inline void QueueTasks(TaskQueue &queue, const std::vector &t // Helper function to filter out tasks of a given state. template inline void FilterStateFromQueue(const TaskQueue &queue, - std::unordered_set &task_ids, - ray::raylet::TaskState filter_state) { + std::unordered_set &task_ids) { for (auto it = task_ids.begin(); it != task_ids.end();) { if (queue.HasTask(*it)) { it = task_ids.erase(it); @@ -173,16 +172,19 @@ void SchedulingQueue::FilterState(std::unordered_set &task_ids, TaskState filter_state) const { switch (filter_state) { case TaskState::PLACEABLE: - FilterStateFromQueue(placeable_tasks_, task_ids, filter_state); + FilterStateFromQueue(placeable_tasks_, task_ids); + break; + case TaskState::WAITING_FOR_ACTOR_CREATION: + FilterStateFromQueue(methods_waiting_for_actor_creation_, task_ids); break; case TaskState::WAITING: - FilterStateFromQueue(waiting_tasks_, task_ids, filter_state); + FilterStateFromQueue(waiting_tasks_, task_ids); break; case TaskState::READY: - FilterStateFromQueue(ready_tasks_, task_ids, filter_state); + FilterStateFromQueue(ready_tasks_, task_ids); break; case TaskState::RUNNING: - FilterStateFromQueue(running_tasks_, task_ids, filter_state); + FilterStateFromQueue(running_tasks_, task_ids); break; case TaskState::BLOCKED: { const auto blocked_ids = GetBlockedTaskIds(); @@ -195,7 +197,7 @@ void SchedulingQueue::FilterState(std::unordered_set &task_ids, } } break; case TaskState::INFEASIBLE: - FilterStateFromQueue(infeasible_tasks_, task_ids, filter_state); + FilterStateFromQueue(infeasible_tasks_, task_ids); break; case TaskState::DRIVER: { const auto driver_ids = GetDriverTaskIds(); diff --git a/src/ray/raylet/scheduling_queue.h b/src/ray/raylet/scheduling_queue.h index bdd065fa1..dad94a6d3 100644 --- a/src/ray/raylet/scheduling_queue.h +++ b/src/ray/raylet/scheduling_queue.h @@ -18,6 +18,8 @@ enum class TaskState { INIT, // The task may be placed on a node. PLACEABLE, + // The task is for an actor whose location we do not know yet. + WAITING_FOR_ACTOR_CREATION, // The task has been placed on a node and is waiting for some object // dependencies to become local. WAITING, diff --git a/src/ray/raylet/task_spec.cc b/src/ray/raylet/task_spec.cc index 0a914adcb..540df9b63 100644 --- a/src/ray/raylet/task_spec.cc +++ b/src/ray/raylet/task_spec.cc @@ -61,14 +61,15 @@ TaskSpecification::TaskSpecification( const std::unordered_map &required_resources, const Language &language) : TaskSpecification(driver_id, parent_task_id, parent_counter, ActorID::nil(), - ObjectID::nil(), ActorID::nil(), ActorHandleID::nil(), -1, + ObjectID::nil(), 0, ActorID::nil(), ActorHandleID::nil(), -1, function_id, task_arguments, num_returns, required_resources, std::unordered_map(), language) {} TaskSpecification::TaskSpecification( const UniqueID &driver_id, const TaskID &parent_task_id, int64_t parent_counter, const ActorID &actor_creation_id, const ObjectID &actor_creation_dummy_object_id, - const ActorID &actor_id, const ActorHandleID &actor_handle_id, int64_t actor_counter, + const int64_t max_actor_reconstructions, const ActorID &actor_id, + const ActorHandleID &actor_handle_id, int64_t actor_counter, const FunctionID &function_id, const std::vector> &task_arguments, int64_t num_returns, const std::unordered_map &required_resources, @@ -96,8 +97,8 @@ TaskSpecification::TaskSpecification( auto spec = CreateTaskInfo( fbb, to_flatbuf(fbb, driver_id), to_flatbuf(fbb, task_id), to_flatbuf(fbb, parent_task_id), parent_counter, to_flatbuf(fbb, actor_creation_id), - to_flatbuf(fbb, actor_creation_dummy_object_id), to_flatbuf(fbb, actor_id), - to_flatbuf(fbb, actor_handle_id), actor_counter, false, + to_flatbuf(fbb, actor_creation_dummy_object_id), max_actor_reconstructions, + to_flatbuf(fbb, actor_id), to_flatbuf(fbb, actor_handle_id), actor_counter, false, to_flatbuf(fbb, function_id), fbb.CreateVector(arguments), fbb.CreateVector(returns), map_to_flatbuf(fbb, required_resources), map_to_flatbuf(fbb, required_placement_resources), language); @@ -220,6 +221,11 @@ ObjectID TaskSpecification::ActorCreationDummyObjectId() const { return from_flatbuf(*message->actor_creation_dummy_object_id()); } +int64_t TaskSpecification::MaxActorReconstructions() const { + auto message = flatbuffers::GetRoot(spec_.data()); + return message->max_actor_reconstructions(); +} + ActorID TaskSpecification::ActorId() const { auto message = flatbuffers::GetRoot(spec_.data()); return from_flatbuf(*message->actor_id()); diff --git a/src/ray/raylet/task_spec.h b/src/ray/raylet/task_spec.h index 5a86a443c..2799f0568 100644 --- a/src/ray/raylet/task_spec.h +++ b/src/ray/raylet/task_spec.h @@ -130,8 +130,9 @@ class TaskSpecification { TaskSpecification( const UniqueID &driver_id, const TaskID &parent_task_id, int64_t parent_counter, const ActorID &actor_creation_id, const ObjectID &actor_creation_dummy_object_id, - const ActorID &actor_id, const ActorHandleID &actor_handle_id, - int64_t actor_counter, const FunctionID &function_id, + int64_t max_actor_reconstructions, const ActorID &actor_id, + const ActorHandleID &actor_handle_id, int64_t actor_counter, + const FunctionID &function_id, const std::vector> &task_arguments, int64_t num_returns, const std::unordered_map &required_resources, @@ -192,6 +193,7 @@ class TaskSpecification { bool IsActorTask() const; ActorID ActorCreationId() const; ObjectID ActorCreationDummyObjectId() const; + int64_t MaxActorReconstructions() const; ActorID ActorId() const; ActorHandleID ActorHandleId() const; int64_t ActorCounter() const; diff --git a/src/ray/raylet/worker_pool_test.cc b/src/ray/raylet/worker_pool_test.cc index abaf675ff..3933ec76c 100644 --- a/src/ray/raylet/worker_pool_test.cc +++ b/src/ray/raylet/worker_pool_test.cc @@ -64,7 +64,7 @@ static inline TaskSpecification ExampleTaskSpec( const ActorID actor_id = ActorID::nil(), const Language &language = Language::PYTHON) { return TaskSpecification(UniqueID::nil(), UniqueID::nil(), 0, ActorID::nil(), - ObjectID::nil(), actor_id, ActorHandleID::nil(), 0, + ObjectID::nil(), 0, actor_id, ActorHandleID::nil(), 0, FunctionID::nil(), {}, 0, {{}}, {{}}, language); } diff --git a/test/actor_test.py b/test/actor_test.py index bf1b98a13..b65ba6426 100644 --- a/test/actor_test.py +++ b/test/actor_test.py @@ -8,12 +8,14 @@ import random import numpy as np import os import pytest +import signal import sys import time import ray import ray.ray_constants as ray_constants import ray.test.test_utils +import ray.test.cluster_utils @pytest.fixture @@ -32,6 +34,23 @@ def shutdown_only(): ray.shutdown() +@pytest.fixture +def head_node_cluster(): + cluster = ray.test.cluster_utils.Cluster( + initialize_head=True, + connect=True, + head_node_args={ + "_internal_config": json.dumps({ + "initial_reconstruction_timeout_milliseconds": 200, + "num_heartbeats_timeout": 10, + }) + }) + yield cluster + # The code after the yield will run as teardown code. + ray.shutdown() + cluster.shutdown() + + def test_actor_init_error_propagated(ray_start_regular): @ray.remote class Actor(object): @@ -1259,15 +1278,8 @@ def test_blocking_actor_task(shutdown_only): assert remaining_ids == [x_id] -def test_exception_raised_when_actor_node_dies(shutdown_only): - ray.worker._init( - start_ray_local=True, - num_local_schedulers=2, - num_cpus=1, - _internal_config=json.dumps({ - "initial_reconstruction_timeout_milliseconds": 200, - "num_heartbeats_timeout": 10, - })) +def test_exception_raised_when_actor_node_dies(head_node_cluster): + remote_node = head_node_cluster.add_node() @ray.remote class Counter(object): @@ -1281,18 +1293,14 @@ def test_exception_raised_when_actor_node_dies(shutdown_only): self.x += 1 return self.x - local_plasma = ray.worker.global_worker.plasma_client.store_socket_name - # Create an actor that is not on the local scheduler. actor = Counter.remote() - while ray.get(actor.local_plasma.remote()) == local_plasma: + while (ray.get(actor.local_plasma.remote()) != + remote_node.get_plasma_store_name()): actor = Counter.remote() - # Kill the second plasma store to get rid of the cached objects and - # trigger the corresponding local scheduler to exit. - process = ray.services.all_processes[ - ray.services.PROCESS_TYPE_PLASMA_STORE][1] - process.kill() + # Kill the second node. + head_node_cluster.remove_node(remote_node) # Submit some new actor tasks both before and after the node failure is # detected. Make sure that getting the result raises an exception. @@ -1306,126 +1314,68 @@ def test_exception_raised_when_actor_node_dies(shutdown_only): # dies). ray.get(x_id) - # Make sure the process has exited. - process.wait() - -@pytest.mark.skip("This test does not work yet.") @pytest.mark.skipif( os.environ.get("RAY_USE_NEW_GCS") == "on", reason="Hanging with new GCS API.") -def test_local_scheduler_dying(shutdown_only): - ray.worker._init( - start_ray_local=True, - num_local_schedulers=2, - num_cpus=1, - redirect_output=True) +def test_actor_init_fails(head_node_cluster): + remote_node = head_node_cluster.add_node() - @ray.remote + @ray.remote(max_reconstructions=1) class Counter(object): def __init__(self): self.x = 0 - def local_plasma(self): - return ray.worker.global_worker.plasma_client.store_socket_name - def inc(self): self.x += 1 return self.x - local_plasma = ray.worker.global_worker.plasma_client.store_socket_name - - # Create an actor that is not on the local scheduler. - actor = Counter.remote() - while ray.get(actor.local_plasma.remote()) == local_plasma: - actor = Counter.remote() - - ids = [actor.inc.remote() for _ in range(100)] - - # Wait for the last task to finish running. - ray.get(ids[-1]) - - # Kill the second plasma store to get rid of the cached objects and - # trigger the corresponding local scheduler to exit. - process = ray.services.all_processes[ - ray.services.PROCESS_TYPE_PLASMA_STORE][1] - process.kill() - process.wait() + # Create many actors. It should take a while to finish initializing them. + actors = [Counter.remote() for _ in range(100)] + # Allow some time to forward the actor creation tasks to the other node. + time.sleep(0.1) + # Kill the second node. + head_node_cluster.remove_node(remote_node) # Get all of the results - results = ray.get(ids) - - assert results == list(range(1, 1 + len(results))) + results = ray.get([actor.inc.remote() for actor in actors]) + assert results == [1 for actor in actors] -@pytest.mark.skip("This test does not work yet.") -@pytest.mark.skipif( - os.environ.get("RAY_USE_NEW_GCS") == "on", - reason="Hanging with new GCS API.") -def test_many_local_schedulers_dying(shutdown_only): - # This test can be made more stressful by increasing the numbers below. - # The total number of actors created will be - # num_actors_at_a_time * num_local_schedulers. - num_local_schedulers = 5 - num_actors_at_a_time = 3 - num_function_calls_at_a_time = 10 +def test_reconstruction_suppression(head_node_cluster): + num_local_schedulers = 10 + worker_nodes = [ + head_node_cluster.add_node() for _ in range(num_local_schedulers) + ] - ray.worker._init( - start_ray_local=True, - num_local_schedulers=num_local_schedulers, - num_cpus=3, - redirect_output=True) - - @ray.remote - class SlowCounter(object): + @ray.remote(max_reconstructions=1) + class Counter(object): def __init__(self): self.x = 0 - def inc(self, duration): - time.sleep(duration) + def inc(self): self.x += 1 return self.x - # Create some initial actors. - actors = [SlowCounter.remote() for _ in range(num_actors_at_a_time)] + @ray.remote + def inc(actor_handle): + return ray.get(actor_handle.inc.remote()) - # Wait for the actors to start up. - time.sleep(1) + # Make sure all of the actors have started. + actors = [Counter.remote() for _ in range(20)] + ray.get([actor.inc.remote() for actor in actors]) - # This is a mapping from actor handles to object IDs returned by - # methods on that actor. - result_ids = collections.defaultdict(lambda: []) + # Kill a node. + head_node_cluster.remove_node(worker_nodes[0]) - # In a loop we are going to create some actors, run some methods, kill - # a local scheduler, and run some more methods. - for i in range(num_local_schedulers - 1): - # Create some actors. - actors.extend( - [SlowCounter.remote() for _ in range(num_actors_at_a_time)]) - # Run some methods. - for j in range(len(actors)): - actor = actors[j] - for _ in range(num_function_calls_at_a_time): - result_ids[actor].append(actor.inc.remote(j**2 * 0.000001)) - # Kill a plasma store to get rid of the cached objects and trigger - # exit of the corresponding local scheduler. Don't kill the first - # local scheduler since that is the one that the driver is - # connected to. - process = ray.services.all_processes[ - ray.services.PROCESS_TYPE_PLASMA_STORE][i + 1] - process.kill() - process.wait() - - # Run some more methods. - for j in range(len(actors)): - actor = actors[j] - for _ in range(num_function_calls_at_a_time): - result_ids[actor].append(actor.inc.remote(j**2 * 0.000001)) - - # Get the results and check that they have the correct values. - for _, result_id_list in result_ids.items(): - results = list(range(1, len(result_id_list) + 1)) - assert ray.get(result_id_list) == results + # Submit several tasks per actor. These should be randomly scheduled to the + # nodes, so that multiple nodes will detect and try to reconstruct the + # actor that died, but only one should succeed. + results = [] + for _ in range(10): + results += [inc.remote(actor) for actor in actors] + # Make sure that we can get the results from the reconstructed actor. + results = ray.get(results) def setup_counter_actor(test_checkpoint=False, @@ -2142,3 +2092,173 @@ def test_creating_more_actors_than_resources(shutdown_only): ray.wait([object_id]) ray.get(results) + + +def test_actor_reconstruction(ray_start_regular): + """Test actor reconstruction when actor process is killed.""" + + @ray.remote(max_reconstructions=1) + class ReconstructableActor(object): + """An actor that will be reconstructed at most once.""" + + def __init__(self): + self.value = 0 + + def increase(self): + self.value += 1 + return self.value + + def get_pid(self): + return os.getpid() + + def kill_actor(actor): + """Kill actor process.""" + pid = ray.get(actor.get_pid.remote()) + os.kill(pid, signal.SIGKILL) + time.sleep(1) + + actor = ReconstructableActor.remote() + # Call increase 3 times + for _ in range(3): + ray.get(actor.increase.remote()) + # kill actor process + kill_actor(actor) + # Call increase again. + # Check that actor is reconstructed and value is 4. + assert ray.get(actor.increase.remote()) == 4 + # kill actor process one more time. + kill_actor(actor) + # The actor has exceeded max reconstructions, and this task should fail. + with pytest.raises(ray.worker.RayTaskError): + ray.get(actor.increase.remote()) + + # Create another actor. + actor = ReconstructableActor.remote() + # Intentionlly exit the actor + actor.__ray_terminate__.remote() + # Check that the actor won't be reconstructed. + with pytest.raises(ray.worker.RayTaskError): + ray.get(actor.increase.remote()) + + +def test_actor_reconstruction_on_node_failure(head_node_cluster): + """Test actor reconstruction when node dies unexpectedly.""" + cluster = head_node_cluster + max_reconstructions = 3 + # Add a few nodes to the cluster. + # Use custom resource to make sure the actor is only created on worker + # nodes, not on the head node. + for _ in range(max_reconstructions + 2): + cluster.add_node( + resources={"a": 1}, + _internal_config=json.dumps({ + "initial_reconstruction_timeout_milliseconds": 200, + "num_heartbeats_timeout": 10, + }), + ) + + def kill_node(object_store_socket): + node_to_remove = None + for node in cluster.worker_nodes: + if object_store_socket == node.get_plasma_store_name(): + node_to_remove = node + cluster.remove_node(node_to_remove) + + @ray.remote(max_reconstructions=max_reconstructions, resources={"a": 1}) + class MyActor(object): + def __init__(self): + self.value = 0 + + def increase(self): + self.value += 1 + return self.value + + def get_object_store_socket(self): + return ray.worker.global_worker.plasma_client.store_socket_name + + actor = MyActor.remote() + # Call increase 3 times. + for _ in range(3): + ray.get(actor.increase.remote()) + + for i in range(max_reconstructions): + object_store_socket = ray.get(actor.get_object_store_socket.remote()) + # Kill actor's node and the actor should be reconstructed + # on a different node. + kill_node(object_store_socket) + # Call increase again. + # Check that the actor is reconstructed and value is correct. + assert ray.get(actor.increase.remote()) == 4 + i + # Check that the actor is now on a different node. + assert object_store_socket != ray.get( + actor.get_object_store_socket.remote()) + + # kill the node again. + object_store_socket = ray.get(actor.get_object_store_socket.remote()) + kill_node(object_store_socket) + # The actor has exceeded max reconstructions, and this task should fail. + with pytest.raises(ray.worker.RayTaskError): + ray.get(actor.increase.remote()) + + +def test_multiple_actor_reconstruction(head_node_cluster): + # This test can be made more stressful by increasing the numbers below. + # The total number of actors created will be + # num_actors_at_a_time * num_local_schedulers. + num_local_schedulers = 5 + num_actors_at_a_time = 3 + num_function_calls_at_a_time = 10 + + worker_nodes = [ + head_node_cluster.add_node( + resources={"CPU": 3}, + _internal_config=json.dumps({ + "initial_reconstruction_timeout_milliseconds": 200, + "num_heartbeats_timeout": 10, + })) for _ in range(num_local_schedulers) + ] + + @ray.remote(max_reconstructions=ray.ray_constants.INFINITE_RECONSTRUCTION) + class SlowCounter(object): + def __init__(self): + self.x = 0 + + def inc(self, duration): + time.sleep(duration) + self.x += 1 + return self.x + + # Create some initial actors. + actors = [SlowCounter.remote() for _ in range(num_actors_at_a_time)] + + # Wait for the actors to start up. + time.sleep(1) + + # This is a mapping from actor handles to object IDs returned by + # methods on that actor. + result_ids = collections.defaultdict(lambda: []) + + # In a loop we are going to create some actors, run some methods, kill + # a local scheduler, and run some more methods. + for node in worker_nodes: + # Create some actors. + actors.extend( + [SlowCounter.remote() for _ in range(num_actors_at_a_time)]) + # Run some methods. + for j in range(len(actors)): + actor = actors[j] + for _ in range(num_function_calls_at_a_time): + result_ids[actor].append(actor.inc.remote(j**2 * 0.000001)) + # Kill a node. + head_node_cluster.remove_node(node) + + # Run some more methods. + for j in range(len(actors)): + actor = actors[j] + for _ in range(num_function_calls_at_a_time): + result_ids[actor].append(actor.inc.remote(j**2 * 0.000001)) + + # Get the results and check that they have the correct values. + for _, result_id_list in result_ids.items(): + results = list(range(1, len(result_id_list) + 1)) + assert ray.get(result_id_list) == results