Implement actor dummy object gc in java (#3822)

* Add dummy object gc in java

* Fix

* Address comments.

* Refine

* Address comments.

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

@@ -221,16 +221,17 @@ public abstract class AbstractRayRuntime implements RayRuntime {
   }
 
   @Override
-  public RayObject call(RayFunc func, RayActor actor, Object[] args) {
+  public RayObject call(RayFunc func, RayActor<?> actor, Object[] args) {
     if (!(actor instanceof RayActorImpl)) {
       throw new IllegalArgumentException("Unsupported actor type: " + actor.getClass().getName());
     }
-    RayActorImpl actorImpl = (RayActorImpl)actor;
+    RayActorImpl<?> actorImpl = (RayActorImpl) actor;
     TaskSpec spec;
     synchronized (actor) {
       spec = createTaskSpec(func, actorImpl, args, false, null);
       spec.getExecutionDependencies().add(((RayActorImpl) actor).getTaskCursor());
       actorImpl.setTaskCursor(spec.returnIds[1]);
+      actorImpl.clearNewActorHandles();
     }
     rayletClient.submitTask(spec);
     return new RayObjectImpl(spec.returnIds[0]);
@@ -257,7 +258,7 @@ public abstract class AbstractRayRuntime implements RayRuntime {
    * @param isActorCreationTask Whether this task is an actor creation task.
    * @return A TaskSpec object.
    */
-  private TaskSpec createTaskSpec(RayFunc func, RayActorImpl actor, Object[] args,
+  private TaskSpec createTaskSpec(RayFunc func, RayActorImpl<?> actor, Object[] args,
       boolean isActorCreationTask, BaseTaskOptions taskOptions) {
     UniqueId taskId = rayletClient.generateTaskId(workerContext.getCurrentDriverId(),
         workerContext.getCurrentTaskId(), workerContext.nextTaskIndex());
@@ -285,7 +286,9 @@ public abstract class AbstractRayRuntime implements RayRuntime {
     if (taskOptions instanceof ActorCreationOptions) {
       maxActorReconstruction = ((ActorCreationOptions) taskOptions).maxReconstructions;
     }
+
     RayFunction rayFunction = functionManager.getFunction(workerContext.getCurrentDriverId(), func);
+
     return new TaskSpec(
         workerContext.getCurrentDriverId(),
         taskId,
@@ -296,6 +299,7 @@ public abstract class AbstractRayRuntime implements RayRuntime {
         actor.getId(),
         actor.getHandleId(),
         actor.increaseTaskCounter(),
+        actor.getNewActorHandles().toArray(new UniqueId[0]),
         ArgumentsBuilder.wrap(args),
         returnIds,
         resources,

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

@@ -4,6 +4,8 @@ import java.io.Externalizable;
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.List;
 import org.ray.api.RayActor;
 import org.ray.api.id.UniqueId;
 import org.ray.runtime.util.Sha1Digestor;
@@ -29,6 +31,14 @@ public final class RayActorImpl<T> implements RayActor<T>, Externalizable {
    */
   private int numForks;
 
+  /**
+   * The new actor handles that were created from this handle
+   * since the last task on this handle was submitted. This is
+   * used to garbage-collect dummy objects that are no longer
+   * necessary in the backend.
+   */
+  private List<UniqueId> newActorHandles;
+
   public RayActorImpl() {
     this(UniqueId.NIL, UniqueId.NIL);
   }
@@ -42,6 +52,7 @@ public final class RayActorImpl<T> implements RayActor<T>, Externalizable {
     this.handleId = handleId;
     this.taskCounter = 0;
     this.taskCursor = null;
+    this.newActorHandles = new ArrayList<>();
     numForks = 0;
   }
 
@@ -59,6 +70,14 @@ public final class RayActorImpl<T> implements RayActor<T>, Externalizable {
     this.taskCursor = taskCursor;
   }
 
+  public List<UniqueId> getNewActorHandles() {
+    return this.newActorHandles;
+  }
+
+  public void clearNewActorHandles() {
+    this.newActorHandles.clear();
+  }
+
   public UniqueId getTaskCursor() {
     return taskCursor;
   }
@@ -74,6 +93,7 @@ public final class RayActorImpl<T> implements RayActor<T>, Externalizable {
     ret.numForks = 0;
     ret.taskCursor = this.taskCursor;
     ret.handleId = this.computeNextActorHandleId();
+    newActorHandles.add(ret.handleId);
     return ret;
   }
 

java/runtime/src/main/java/org/ray/runtime/raylet/RayletClientImpl.java

@@ -142,6 +142,11 @@ public class RayletClientImpl implements RayletClient {
     UniqueId actorId = UniqueId.fromByteBuffer(info.actorIdAsByteBuffer());
     UniqueId actorHandleId = UniqueId.fromByteBuffer(info.actorHandleIdAsByteBuffer());
     int actorCounter = info.actorCounter();
+
+    // Deserialize new actor handles
+    UniqueId[] newActorHandles = UniqueIdUtil.getUniqueIdsFromByteBuffer(
+      info.newActorHandlesAsByteBuffer());
+
     // Deserialize args
     FunctionArg[] args = new FunctionArg[info.argsLength()];
     for (int i = 0; i < info.argsLength(); i++) {
@@ -175,7 +180,7 @@ public class RayletClientImpl implements RayletClient {
         info.functionDescriptor(0), info.functionDescriptor(1), info.functionDescriptor(2)
     );
     return new TaskSpec(driverId, taskId, parentTaskId, parentCounter, actorCreationId,
-        maxActorReconstructions, actorId, actorHandleId, actorCounter,
+        maxActorReconstructions, actorId, actorHandleId, actorCounter, newActorHandles,
         args, returnIds, resources, functionDescriptor);
   }
 

java/runtime/src/main/java/org/ray/runtime/task/TaskSpec.java

@@ -68,8 +68,8 @@ public class TaskSpec {
 
   public TaskSpec(UniqueId driverId, UniqueId taskId, UniqueId parentTaskId, int parentCounter,
       UniqueId actorCreationId, int maxActorReconstructions, UniqueId actorId,
-      UniqueId actorHandleId, int actorCounter, FunctionArg[] args, UniqueId[] returnIds,
-      Map<String, Double> resources, FunctionDescriptor functionDescriptor) {
+      UniqueId actorHandleId, int actorCounter, UniqueId[] newActorHandles, FunctionArg[] args,
+      UniqueId[] returnIds, Map<String, Double> resources, FunctionDescriptor functionDescriptor) {
     this.driverId = driverId;
     this.taskId = taskId;
     this.parentTaskId = parentTaskId;
@@ -79,8 +79,7 @@ public class TaskSpec {
     this.actorId = actorId;
     this.actorHandleId = actorHandleId;
     this.actorCounter = actorCounter;
-    // TODO: Initialize the new actor handles.
-    this.newActorHandles = new UniqueId[] {};
+    this.newActorHandles = newActorHandles;
     this.args = args;
     this.returnIds = returnIds;
     this.resources = resources;