wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-13 17:45:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

Java call Python use structured function descriptors (#7634)

Browse Source
This commit is contained in:
fyrestone
2020-03-20 17:29:45 +08:00
committed by GitHub
parent 7d08b418fc
commit a1ae935839
14 changed files with 475 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files
streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/schedule/JobSchedulerImpl.java
+3 -1
View File
@@ -8,6 +8,7 @@ import org.ray.api.Ray;
import org.ray.api.RayActor;
import org.ray.api.RayObject;
import org.ray.api.RayPyActor;
import org.ray.api.function.PyActorMethod;
import org.ray.streaming.api.Language;
import org.ray.streaming.jobgraph.JobGraph;
import org.ray.streaming.runtime.core.graph.ExecutionGraph;
@@ -68,7 +69,8 @@ public class JobSchedulerImpl implements JobScheduler {
case PYTHON:
byte[] workerContextBytes = buildPythonWorkerContext(
taskId, executionGraphPb, jobConfig);
waits.add(((RayPyActor)worker).call("init", workerContextBytes));
waits.add(((RayPyActor)worker).call(new PyActorMethod("init", Object.class),
workerContextBytes));
break;
default:
throw new UnsupportedOperationException(
streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/schedule/TaskAssignerImpl.java
+3 -2
View File
@@ -6,6 +6,7 @@ import java.util.List;
import java.util.Map;
import org.ray.api.BaseActor;
import org.ray.api.Ray;
import org.ray.api.function.PyActorClass;
import org.ray.streaming.jobgraph.JobEdge;
import org.ray.streaming.jobgraph.JobGraph;
import org.ray.streaming.jobgraph.JobVertex;
@@ -61,8 +62,8 @@ public class TaskAssignerImpl implements TaskAssigner {
private BaseActor createWorker(JobVertex jobVertex) {
switch (jobVertex.getLanguage()) {
case PYTHON:
return Ray.createPyActor(
"ray.streaming.runtime.worker", "JobWorker");
return Ray.createActor(
new PyActorClass("ray.streaming.runtime.worker", "JobWorker"));
case JAVA:
return Ray.createActor(JobWorker::new);
default: