[java] support creating an actor with parameters (#2817)

Previously `Ray.createActor` only support creating an actor without any parameter. This PR adds the support for creating an actor with parameters. Moreover, besides using a constructor, it's now also allowed to create an actor with a factory method. For more usage, prefer refer to `ActorTest.java`.

java/runtime-common/src/main/java/org/ray/core/AbstractRayRuntime.java

@@ -14,9 +14,7 @@ import org.ray.api.Ray;
 import org.ray.api.RayActor;
 import org.ray.api.RayObject;
 import org.ray.api.WaitResult;
-import org.ray.api.annotation.RayRemote;
 import org.ray.api.function.RayFunc;
-import org.ray.api.function.RayFunc2;
 import org.ray.api.id.UniqueId;
 import org.ray.api.runtime.RayRuntime;
 import org.ray.core.model.RayParameters;
@@ -50,9 +48,9 @@ public abstract class AbstractRayRuntime implements RayRuntime {
   protected PathConfig pathConfig;
 
   /**
-   * Actor ID -> actor instance.
+   * Actor ID -> local actor instance.
    */
-  private Map<UniqueId, Object> actors = new HashMap<>();
+  Map<UniqueId, Object> localActors = new HashMap<>();
 
   // app level Ray.init()
   // make it private so there is no direct usage but only from Ray.init
@@ -137,7 +135,7 @@ public abstract class AbstractRayRuntime implements RayRuntime {
       objectStoreProxy = new ObjectStoreProxy(plink, slink);
     }
     
-    worker = new Worker(localSchedulerClient, functions);
+    worker = new Worker(this);
   }
 
   private static AbstractRayRuntime instantiate(RayParameters params) {
@@ -195,13 +193,15 @@ public abstract class AbstractRayRuntime implements RayRuntime {
 
   @Override
   public <T> RayObject<T> put(T obj) {
-    UniqueId objectId = getCurrentTaskNextPutId();
+    UniqueId objectId = UniqueIdHelper.computePutId(
+        WorkerContext.currentTask().taskId, WorkerContext.nextPutIndex());
+
     put(objectId, obj);
     return new RayObjectImpl<>(objectId);
   }
 
   public <T> void put(UniqueId objectId, T obj) {
-    UniqueId taskId = getCurrentTaskId();
+    UniqueId taskId = WorkerContext.currentTask().taskId;
     RayLog.core.info("Putting object {}, for task {} ", objectId, taskId);
     if (!params.use_raylet) {
       localSchedulerClient.markTaskPutDependency(taskId, objectId);
@@ -209,21 +209,6 @@ public abstract class AbstractRayRuntime implements RayRuntime {
     objectStoreProxy.put(objectId, obj, null);
   }
 
-  /**
-   * get the task identity of the currently running task, UniqueId.NIL if not inside any
-   */
-  public UniqueId getCurrentTaskId() {
-    return worker.getCurrentTaskId();
-  }
-
-  /**
-   * get the to-be-returned objects identities of the currently running task, empty array if not
-   * inside any.
-   */
-  public UniqueId getCurrentTaskNextPutId() {
-    return worker.getCurrentTaskNextPutId();
-  }
-
   @Override
   public <T> T get(UniqueId objectId) throws TaskExecutionException {
     List<T> ret = get(ImmutableList.of(objectId));
@@ -233,7 +218,7 @@ public abstract class AbstractRayRuntime implements RayRuntime {
   @Override
   public <T> List<T> get(List<UniqueId> objectIds) {
     boolean wasBlocked = false;
-    UniqueId taskId = getCurrentTaskId();
+    UniqueId taskId = WorkerContext.currentTask().taskId;
 
     try {
       int numObjectIds = objectIds.size();
@@ -343,7 +328,7 @@ public abstract class AbstractRayRuntime implements RayRuntime {
 
   @Override
   public RayObject call(RayFunc func, Object[] args) {
-    TaskSpec spec = createTaskSpec(func, RayActorImpl.NIL, args, null);
+    TaskSpec spec = createTaskSpec(func, RayActorImpl.NIL, args, false);
     localSchedulerClient.submitTask(spec);
     return new RayObjectImpl(spec.returnIds[0]);
   }
@@ -354,7 +339,7 @@ public abstract class AbstractRayRuntime implements RayRuntime {
       throw new IllegalArgumentException("Unsupported actor type: " + actor.getClass().getName());
     }
     RayActorImpl actorImpl = (RayActorImpl)actor;
-    TaskSpec spec = createTaskSpec(func, actorImpl, args, null);
+    TaskSpec spec = createTaskSpec(func, actorImpl, args, false);
     actorImpl.setTaskCursor(spec.returnIds[1]);
     localSchedulerClient.submitTask(spec);
     return new RayObjectImpl(spec.returnIds[0]);
@@ -362,30 +347,13 @@ public abstract class AbstractRayRuntime implements RayRuntime {
 
   @Override
   @SuppressWarnings("unchecked")
-  public <T> RayActor<T> createActor(Class<T> actorClass) {
-    RayFunc2<UniqueId, String, Object> func = AbstractRayRuntime::createLocalActor;
-    TaskSpec spec = createTaskSpec(func, RayActorImpl.NIL, null, actorClass);
-    RayActorImpl actor = new RayActorImpl(spec.returnIds[0]);
+  public <T> RayActor<T> createActor(RayFunc actorFactoryFunc, Object[] args) {
+    TaskSpec spec = createTaskSpec(actorFactoryFunc, RayActorImpl.NIL, args, true);
+    RayActorImpl<?> actor = new RayActorImpl(spec.returnIds[0]);
     actor.increaseTaskCounter();
     actor.setTaskCursor(spec.returnIds[0]);
     localSchedulerClient.submitTask(spec);
-    return actor;
-  }
-
-  @RayRemote
-  private static Object createLocalActor(UniqueId actorId, String className) {
-    try {
-      Class<?> cls = Class.forName(className, true, Thread.currentThread().getContextClassLoader());
-      Object actor = cls.getConstructor().newInstance();
-      getInstance().actors.put(actorId, actor);
-      RayLog.core.info("Created actor: {}, actor id: {}", className, actorId);
-      return null;
-    } catch (ClassNotFoundException | InstantiationException | IllegalAccessException
-        | IllegalArgumentException | InvocationTargetException | NoSuchMethodException
-        | SecurityException e) {
-      RayLog.core.error("Failed to create actor {}", className, e);
-      throw new TaskExecutionException(e);
-    }
+    return (RayActor<T>) actor;
   }
 
   /**
@@ -404,12 +372,11 @@ public abstract class AbstractRayRuntime implements RayRuntime {
    * @param func The target remote function.
    * @param actor The actor handle. If the task is not an actor task, actor id must be NIL.
    * @param args The arguments for the remote function.
-   * @param actorClassForCreation If the task is a actor creation task, the argument should be
-   *     the actor class. Otherwise, it should be null.
+   * @param isActorCreationTask Whether this task is an actor creation task.
    * @return A TaskSpec object.
    */
   private TaskSpec createTaskSpec(RayFunc func, RayActorImpl actor, Object[] args,
-      Class actorClassForCreation) {
+      boolean isActorCreationTask) {
     final TaskSpec current = WorkerContext.currentTask();
     UniqueId taskId = localSchedulerClient.generateTaskId(current.driverId,
         current.taskId,
@@ -418,8 +385,7 @@ public abstract class AbstractRayRuntime implements RayRuntime {
     UniqueId[] returnIds = genReturnIds(taskId, numReturns);
 
     UniqueId actorCreationId = UniqueId.NIL;
-    if (actorClassForCreation != null) {
-      args = new Object[] {returnIds[0], actorClassForCreation.getName()};
+    if (isActorCreationTask) {
       actorCreationId = returnIds[0];
     }
 
@@ -448,24 +414,25 @@ public abstract class AbstractRayRuntime implements RayRuntime {
         returnIds,
         actor.getHandleId(),
         actorCreationId,
-        ResourceUtil.getResourcesMapFromArray(rayMethod.remoteAnnotation.resources()),
+        ResourceUtil.getResourcesMapFromArray(rayMethod.remoteAnnotation),
         actor.getTaskCursor()
     );
   }
 
-  /***********
-   * Internal Methods.
-   ***********/
-
   public void loop() {
     worker.loop();
   }
 
-  /**
-   * get actor with given id.
-   */
-  public Object getLocalActor(UniqueId id) {
-    return actors.get(id);
+  public Worker getWorker() {
+    return worker;
   }
 
-}
+  public LocalSchedulerLink getLocalSchedulerClient() {
+    return localSchedulerClient;
+  }
+
+  public LocalFunctionManager getLocalFunctionManager() {
+    return functions;
+  }
+}
+

java/runtime-common/src/main/java/org/ray/core/ArgumentsBuilder.java

@@ -9,12 +9,16 @@ import org.ray.api.id.UniqueId;
 import org.ray.spi.model.FunctionArg;
 import org.ray.spi.model.TaskSpec;
 
-/**
- * arguments wrap and unwrap.
- */
 public class ArgumentsBuilder {
 
-  @SuppressWarnings({"rawtypes", "unchecked"})
+  private static boolean checkSimpleValue(Object o) {
+    // TODO(raulchen): implement this.
+    return true;
+  }
+
+  /**
+   * Convert real function arguments to task spec arguments.
+   */
   public static FunctionArg[] wrap(Object[] args) {
     FunctionArg[] ret = new FunctionArg[args.length];
     for (int i = 0; i < ret.length; i++) {
@@ -38,13 +42,11 @@ public class ArgumentsBuilder {
     return ret;
   }
 
-  private static boolean checkSimpleValue(Object o) {
-    return true;//TODO I think Ray don't want to pass big parameter
-  }
-
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  public static Pair<Object, Object[]> unwrap(TaskSpec task, Method m, ClassLoader classLoader) {
-    // the last arg is className
+  /**
+   * Convert task spec arguments to real function arguments.
+   */
+  public static Object[] unwrap(TaskSpec task, ClassLoader classLoader) {
+    // Ignore the last arg, which is the class name
     Object[] realArgs = new Object[task.args.length - 1];
     for (int i = 0; i < task.args.length - 1; i++) {
       FunctionArg arg = task.args[i];
@@ -57,8 +59,6 @@ public class ArgumentsBuilder {
         realArgs[i] = Ray.get(arg.id);
       }
     }
-    Object actor = task.actorId.isNil()
-        ? null : AbstractRayRuntime.getInstance().getLocalActor(task.actorId);
-    return Pair.of(actor, realArgs);
+    return realArgs;
   }
 }

java/runtime-common/src/main/java/org/ray/core/InvocationExecutor.java

@@ -1,94 +0,0 @@
-package org.ray.core;
-
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import org.apache.commons.lang3.tuple.Pair;
-import org.ray.api.id.UniqueId;
-import org.ray.spi.model.RayMethod;
-import org.ray.spi.model.TaskSpec;
-import org.ray.util.exception.TaskExecutionException;
-import org.ray.util.logger.RayLog;
-
-/**
- * how to execute a invocation.
- */
-public class InvocationExecutor {
-
-
-  public static void execute(TaskSpec task, Pair<ClassLoader, RayMethod> pr)
-      throws TaskExecutionException {
-    String taskdesc =
-        "[" + pr.getRight().fullName + "_" + task.taskId.toString() + " actorId = " + task.actorId
-            + "]";
-    TaskExecutionException ex = null;
-
-    // switch to current driver's loader
-    ClassLoader oldLoader = Thread.currentThread().getContextClassLoader();
-    if (pr.getLeft() != null) {
-      Thread.currentThread().setContextClassLoader(pr.getLeft());
-    }
-
-    // execute
-    try {
-      executeInternal(task, pr);
-    } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
-      if (!task.actorId.isNil()
-          && AbstractRayRuntime.getInstance().getLocalActor(task.actorId) == null) {
-        ex = new TaskExecutionException("Task " + taskdesc + " execution on actor " + task.actorId
-            + " failed as the actor is not present ", e);
-        RayLog.core.error("Task " + taskdesc + " execution on actor " + task.actorId
-            + " failed as the actor is not present ", e);
-      } else {
-        ex = new TaskExecutionException(
-            formatTaskExecutionExceptionMsg(task, pr.getRight().fullName), e);
-        RayLog.core.error("Task " + taskdesc + " execution failed ", e);
-      }
-      RayLog.core.error(e.getMessage());
-      RayLog.core.error("task info: \n" + task.toString());
-    } catch (Throwable e) {
-      ex = new TaskExecutionException(formatTaskExecutionExceptionMsg(task, pr.getRight().fullName),
-          e);
-      RayLog.core.error("Task " + taskdesc + " execution with unknown error ", e);
-      RayLog.core.error(e.getMessage());
-    }
-
-    // recover loader
-    if (pr.getLeft() != null) {
-      Thread.currentThread().setContextClassLoader(oldLoader);
-    }
-
-    // set exception as the output results
-    if (ex != null) {
-      throw ex;
-    }
-  }
-
-  private static void executeInternal(TaskSpec task, Pair<ClassLoader, RayMethod> pr)
-      throws IllegalAccessException, IllegalArgumentException, InvocationTargetException {
-    Method m = pr.getRight().invokable;
-    Pair<Object, Object[]> realArgs = ArgumentsBuilder.unwrap(task, m, pr.getLeft());
-
-    // execute
-    Object result = null;
-    try {
-      result = m.invoke(realArgs.getLeft(), realArgs.getRight());
-    } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
-      RayLog.core.error("invoke failed:" + m);
-      throw e;
-    }
-
-    if (task.returnIds == null || task.returnIds.length == 0) {
-      return;
-    }
-    AbstractRayRuntime.getInstance().put(task.returnIds[0], result);
-  }
-
-  private static String formatTaskExecutionExceptionMsg(TaskSpec task, String funcName) {
-    return "Execute task " + task.taskId
-        + " failed with function name = " + funcName;
-  }
-
-  private static void safePut(UniqueId objectId, Object obj) {
-    AbstractRayRuntime.getInstance().put(objectId, obj);
-  }
-}

java/runtime-common/src/main/java/org/ray/core/LocalFunctionManager.java

@@ -62,8 +62,7 @@ public class LocalFunctionManager {
    * it may block for a while if the related resources (e.g., jars) are not ready on local machine
    */
   public Pair<ClassLoader, RayMethod> getMethod(UniqueId driverId, UniqueId actorId,
-      UniqueId methodId,
-      FunctionArg[] args) throws NoSuchMethodException, SecurityException, ClassNotFoundException {
+      UniqueId methodId, FunctionArg[] args) {
     Preconditions.checkArgument(args.length >= 1, "method's args len %s<=1", args.length);
     String className = (String) Serializer.decode(args[args.length - 1].data);
     return getMethod(driverId, actorId, methodId, className);
@@ -84,20 +83,20 @@ public class LocalFunctionManager {
 
     final ClassLoader classLoader;
     final ConcurrentHashMap<String, RayTaskMethods> taskMethods = new ConcurrentHashMap<>();
-    final ConcurrentHashMap<String, RayActorMethods> actors = new ConcurrentHashMap<>();
+    final ConcurrentHashMap<String, RayActorMethods> actorMethods = new ConcurrentHashMap<>();
 
     FunctionTable(ClassLoader classLoader) {
       this.classLoader = classLoader;
     }
 
     RayMethod getTaskMethod(UniqueId methodId, String className) {
-      RayTaskMethods tasks = taskMethods.get(className);
-      if (tasks == null) {
-        tasks = RayTaskMethods.fromClass(className, classLoader);
-        RayLog.core.info("create RayTaskMethods:" + tasks);
-        taskMethods.put(className, tasks);
+      RayTaskMethods taskMethods = this.taskMethods.get(className);
+      if (taskMethods == null) {
+        taskMethods = RayTaskMethods.fromClass(className, classLoader);
+        RayLog.core.info("create RayTaskMethods: {}", taskMethods);
+        this.taskMethods.put(className, taskMethods);
       }
-      RayMethod m = tasks.functions.get(methodId);
+      RayMethod m = taskMethods.functions.get(methodId);
       if (m != null) {
         return m;
       }
@@ -110,13 +109,14 @@ public class LocalFunctionManager {
     }
 
     private RayMethod getActorMethod(UniqueId methodId, String className, boolean isStatic) {
-      RayActorMethods actor = actors.get(className);
-      if (actor == null) {
-        actor = RayActorMethods.fromClass(className, classLoader);
-        RayLog.core.info("create RayActorMethods:" + actor);
-        actors.put(className, actor);
+      RayActorMethods actorMethods = this.actorMethods.get(className);
+      if (actorMethods == null) {
+        actorMethods = RayActorMethods.fromClass(className, classLoader);
+        RayLog.core.info("create RayActorMethods: {}", actorMethods);
+        this.actorMethods.put(className, actorMethods);
       }
-      return isStatic ? actor.staticFunctions.get(methodId) : actor.functions.get(methodId);
+      return isStatic ? actorMethods.staticFunctions.get(methodId)
+          : actorMethods.functions.get(methodId);
     }
   }
-}
+}

java/runtime-common/src/main/java/org/ray/core/UniqueIdHelper.java

@@ -68,5 +68,4 @@ public class UniqueIdHelper {
 
     return new UniqueId(taskId);
   }
-
 }

java/runtime-common/src/main/java/org/ray/core/Worker.java

@@ -1,9 +1,8 @@
 package org.ray.core;
 
-import org.apache.commons.lang3.ArrayUtils;
 import org.apache.commons.lang3.tuple.Pair;
+import org.ray.api.exception.RayException;
 import org.ray.api.id.UniqueId;
-import org.ray.spi.LocalSchedulerLink;
 import org.ray.spi.model.RayMethod;
 import org.ray.spi.model.TaskSpec;
 import org.ray.util.logger.RayLog;
@@ -14,51 +13,58 @@ import org.ray.util.logger.RayLog;
  */
 public class Worker {
 
-  private final LocalSchedulerLink scheduler;
-  private final LocalFunctionManager functions;
+  private final AbstractRayRuntime runtime;
 
-  public Worker(LocalSchedulerLink scheduler, LocalFunctionManager functions) {
-    this.scheduler = scheduler;
-    this.functions = functions;
+  public Worker(AbstractRayRuntime runtime) {
+    this.runtime = runtime;
   }
 
   public void loop() {
     while (true) {
       RayLog.core.info(Thread.currentThread().getName() + ":fetching new task...");
-      TaskSpec task = scheduler.getTask();
-      execute(task, functions);
+      TaskSpec task = runtime.getLocalSchedulerClient().getTask();
+      execute(task);
     }
   }
 
-  public static void execute(TaskSpec task, LocalFunctionManager funcs) {
-    RayLog.core.info("Executing task {}", task.taskId);
-
-    if (!task.actorId.isNil() || (task.createActorId != null && !task.createActorId.isNil())) {
-      task.returnIds = ArrayUtils.subarray(task.returnIds, 0, task.returnIds.length - 1);
-    }
-
+  /**
+   * Execute a task.
+   */
+  public void execute(TaskSpec spec) {
+    RayLog.core.info("Executing task {}", spec.taskId);
+    UniqueId returnId = spec.returnIds[0];
+    ClassLoader oldLoader = Thread.currentThread().getContextClassLoader();
     try {
-      Pair<ClassLoader, RayMethod> pr = funcs
-          .getMethod(task.driverId, task.actorId, task.functionId, task.args);
-      WorkerContext.prepare(task, pr.getLeft());
-      InvocationExecutor.execute(task, pr);
-      RayLog.core.info("Finished executing task {}", task.taskId);
+      // Get method
+      Pair<ClassLoader, RayMethod> pair = runtime.getLocalFunctionManager().getMethod(
+          spec.driverId, spec.actorId, spec.functionId, spec.args);
+      ClassLoader classLoader = pair.getLeft();
+      RayMethod method = pair.getRight();
+      // Set context
+      WorkerContext.prepare(spec, classLoader);
+      Thread.currentThread().setContextClassLoader(classLoader);
+      // Get local actor object and arguments.
+      Object actor = spec.isActorTask() ? runtime.localActors.get(spec.actorId) : null;
+      Object[] args = ArgumentsBuilder.unwrap(spec, classLoader);
+      // Execute the task.
+      Object result;
+      if (!method.isConstructor()) {
+        result = method.getMethod().invoke(actor, args);
+      } else {
+        result = method.getConstructor().newInstance(args);
+      }
+      // Set result
+      if (!spec.isActorCreationTask()) {
+        runtime.put(returnId, result);
+      } else {
+        runtime.localActors.put(returnId, result);
+      }
+      RayLog.core.info("Finished executing task {}", spec.taskId);
     } catch (Exception e) {
-      RayLog.core.error("Failed to execute task " + task.taskId, e);
-      AbstractRayRuntime.getInstance().put(task.returnIds[0], e);
+      RayLog.core.error("Error executing task " + spec, e);
+      runtime.put(returnId, new RayException("Error executing task " + spec, e));
+    } finally {
+      Thread.currentThread().setContextClassLoader(oldLoader);
     }
   }
-
-  public UniqueId getCurrentTaskId() {
-    return WorkerContext.currentTask().taskId;
-  }
-
-  public UniqueId getCurrentTaskNextPutId() {
-    return UniqueIdHelper.computePutId(
-        WorkerContext.currentTask().taskId, WorkerContext.nextPutIndex());
-  }
-
-  public UniqueId[] getCurrentTaskReturnIDs() {
-    return WorkerContext.currentTask().returnIds;
-  }
 }

java/runtime-common/src/main/java/org/ray/spi/model/RayActorMethods.java

@@ -1,10 +1,15 @@
 package org.ray.spi.model;
 
 import com.google.common.base.Preconditions;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Executable;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import org.ray.api.annotation.RayRemote;
 import org.ray.api.id.UniqueId;
@@ -28,22 +33,21 @@ public final class RayActorMethods {
     this.staticFunctions = Collections.unmodifiableMap(new HashMap<>(staticFunctions));
   }
 
-  public static RayActorMethods fromClass(String clazzName, ClassLoader classLoader) {
+  public static RayActorMethods fromClass(String className, ClassLoader classLoader) {
     try {
-      Class clazz = Class.forName(clazzName, true, classLoader);
+      Class clazz = Class.forName(className, true, classLoader);
       RayRemote remoteAnnotation = (RayRemote) clazz.getAnnotation(RayRemote.class);
-      Preconditions
-          .checkNotNull(remoteAnnotation, "%s must declare @RayRemote", clazzName);
-      Method[] methods = clazz.getDeclaredMethods();
-      Map<UniqueId, RayMethod> functions = new HashMap<>(methods.length * 2);
-      Map<UniqueId, RayMethod> staticFunctions = new HashMap<>(methods.length * 2);
+      Preconditions.checkNotNull(remoteAnnotation,
+          "%s must be annotated with @RayRemote", className);
 
-      for (Method m : methods) {
-        if (!Modifier.isPublic(m.getModifiers())) {
-          continue;
-        }
-        RayMethod rayMethod = RayMethod.from(m, remoteAnnotation);
-        if (Modifier.isStatic(m.getModifiers())) {
+      List<Executable> executables = new ArrayList<>(Arrays.asList(clazz.getDeclaredMethods()));
+
+      Map<UniqueId, RayMethod> functions = new HashMap<>();
+      Map<UniqueId, RayMethod> staticFunctions = new HashMap<>();
+
+      for (Executable e : executables) {
+        RayMethod rayMethod = RayMethod.from(e, remoteAnnotation);
+        if (Modifier.isStatic(e.getModifiers())) {
           staticFunctions.put(rayMethod.getFuncId(), rayMethod);
         } else {
           functions.put(rayMethod.getFuncId(), rayMethod);
@@ -51,7 +55,7 @@ public final class RayActorMethods {
       }
       return new RayActorMethods(clazz, remoteAnnotation, functions, staticFunctions);
     } catch (Exception e) {
-      throw new RuntimeException("failed to get RayActorMethods from " + clazzName, e);
+      throw new RuntimeException("failed to get RayActorMethods from " + className, e);
     }
   }
 

java/runtime-common/src/main/java/org/ray/spi/model/RayMethod.java

@@ -1,5 +1,7 @@
 package org.ray.spi.model;
 
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Executable;
 import java.lang.reflect.Method;
 import org.ray.api.annotation.RayRemote;
 import org.ray.api.id.UniqueId;
@@ -10,29 +12,40 @@ import org.ray.util.MethodId;
  */
 public class RayMethod {
 
-  public final Method invokable;
+  public final Executable invokable;
   public final String fullName;
   public final RayRemote remoteAnnotation;
   private final UniqueId funcId;
 
-  private RayMethod(Method m, RayRemote remoteAnnotation, UniqueId funcId) {
-    this.invokable = m;
+  private RayMethod(Executable e, RayRemote remoteAnnotation, UniqueId funcId) {
+    this.invokable = e;
     this.remoteAnnotation = remoteAnnotation;
     this.funcId = funcId;
-    fullName = m.getDeclaringClass().getName() + "." + m.getName();
+    fullName = e.getDeclaringClass().getName() + "." + e.getName();
   }
 
-  public static RayMethod from(Method m, RayRemote parentRemoteAnnotation) {
-    Class<?> clazz = m.getDeclaringClass();
-    RayRemote remoteAnnotation = m.getAnnotation(RayRemote.class);
-    MethodId mid = MethodId.fromMethod(m);
+  public static RayMethod from(Executable e, RayRemote parentRemoteAnnotation) {
+    RayRemote remoteAnnotation = e.getAnnotation(RayRemote.class);
+    MethodId mid = MethodId.fromExecutable(e);
     UniqueId funcId = new UniqueId(mid.getSha1Hash());
-    RayMethod method = new RayMethod(m,
+    RayMethod method = new RayMethod(e,
         remoteAnnotation != null ? remoteAnnotation : parentRemoteAnnotation,
         funcId);
     return method;
   }
 
+  public boolean isConstructor() {
+    return invokable instanceof Constructor;
+  }
+
+  public Constructor<?> getConstructor() {
+    return (Constructor<?>) invokable;
+  }
+
+  public Method getMethod() {
+    return (Method) invokable;
+  }
+
   @Override
   public String toString() {
     return fullName;

java/runtime-common/src/main/java/org/ray/spi/model/RayTaskMethods.java

@@ -1,9 +1,14 @@
 package org.ray.spi.model;
 
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Executable;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import org.ray.api.annotation.RayRemote;
 import org.ray.api.id.UniqueId;
@@ -23,20 +28,19 @@ public final class RayTaskMethods {
   public static RayTaskMethods fromClass(String clazzName, ClassLoader classLoader) {
     try {
       Class clazz = Class.forName(clazzName, true, classLoader);
-      Method[] methods = clazz.getDeclaredMethods();
-      Map<UniqueId, RayMethod> functions = new HashMap<>(methods.length * 2);
+      List<Executable> executables = new ArrayList<>();
+      executables.addAll(Arrays.asList(clazz.getDeclaredMethods()));
+      executables.addAll(Arrays.asList(clazz.getConstructors()));
+      Map<UniqueId, RayMethod> functions = new HashMap<>(executables.size());
 
-      for (Method m : methods) {
-        if (!Modifier.isStatic(m.getModifiers())) {
+      for (Executable e : executables) {
+        // This executable must be either a constructor or a static method.
+        if (!(e instanceof Constructor)
+            && !Modifier.isStatic(e.getModifiers())) {
           continue;
         }
-        //task method only for static.
-        RayRemote remoteAnnotation = m.getAnnotation(RayRemote.class);
-        if (remoteAnnotation == null) {
-          continue;
-        }
-        m.setAccessible(true);
-        RayMethod rayMethod = RayMethod.from(m, null);
+        e.setAccessible(true);
+        RayMethod rayMethod = RayMethod.from(e, null);
         functions.put(rayMethod.getFuncId(), rayMethod);
       }
       return new RayTaskMethods(clazz, functions);

java/runtime-common/src/main/java/org/ray/spi/model/TaskSpec.java

@@ -94,4 +94,11 @@ public class TaskSpec {
     return builder.toString();
   }
 
+  public boolean isActorTask() {
+    return !actorId.isNil();
+  }
+
+  public boolean isActorCreationTask() {
+    return !createActorId.isNil();
+  }
 }