wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-06-30 20:00:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

[Java] Make both RayActor and RayPyActor inheriting from BaseActor (#7462)

Browse Source
This commit is contained in:
Kai Yang
2020-03-17 21:45:56 +08:00
committed by GitHub
parent dfa5d9b8e9
commit 6b888b0247
21 changed files with 83 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/core/graph/ExecutionGraph.java
+9 -9
View File
@@ -6,7 +6,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.ray.api.RayActor;
import org.ray.api.BaseActor;
/**
* Physical execution graph.
@@ -18,19 +18,19 @@ import org.ray.api.RayActor;
public class ExecutionGraph implements Serializable {
private long buildTime;
private List<ExecutionNode> executionNodeList;
private List<RayActor> sourceWorkers = new ArrayList<>();
private List<RayActor> sinkWorkers = new ArrayList<>();
private List<BaseActor> sourceWorkers = new ArrayList<>();
private List<BaseActor> sinkWorkers = new ArrayList<>();
public ExecutionGraph(List<ExecutionNode> executionNodes) {
this.executionNodeList = executionNodes;
for (ExecutionNode executionNode : executionNodeList) {
if (executionNode.getNodeType() == ExecutionNode.NodeType.SOURCE) {
List<RayActor> actors = executionNode.getExecutionTasks().stream()
List<BaseActor> actors = executionNode.getExecutionTasks().stream()
.map(ExecutionTask::getWorker).collect(Collectors.toList());
sourceWorkers.addAll(actors);
}
if (executionNode.getNodeType() == ExecutionNode.NodeType.SINK) {
List<RayActor> actors = executionNode.getExecutionTasks().stream()
List<BaseActor> actors = executionNode.getExecutionTasks().stream()
.map(ExecutionTask::getWorker).collect(Collectors.toList());
sinkWorkers.addAll(actors);
}
@@ -38,11 +38,11 @@ public class ExecutionGraph implements Serializable {
buildTime = System.currentTimeMillis();
}
public List<RayActor> getSourceWorkers() {
public List<BaseActor> getSourceWorkers() {
return sourceWorkers;
}
public List<RayActor> getSinkWorkers() {
public List<BaseActor> getSinkWorkers() {
return sinkWorkers;
}
@@ -81,10 +81,10 @@ public class ExecutionGraph implements Serializable {
throw new RuntimeException("Task " + taskId + " does not exist!");
}
public Map<Integer, RayActor> getTaskId2WorkerByNodeId(int nodeId) {
public Map<Integer, BaseActor> getTaskId2WorkerByNodeId(int nodeId) {
for (ExecutionNode executionNode : executionNodeList) {
if (executionNode.getNodeId() == nodeId) {
Map<Integer, RayActor> taskId2Worker = new HashMap<>();
Map<Integer, BaseActor> taskId2Worker = new HashMap<>();
for (ExecutionTask executionTask : executionNode.getExecutionTasks()) {
taskId2Worker.put(executionTask.getTaskId(), executionTask.getWorker());
}
streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/core/graph/ExecutionTask.java
+5 -5
View File
@@ -1,7 +1,7 @@
package org.ray.streaming.runtime.core.graph;
import java.io.Serializable;
import org.ray.api.RayActor;
import org.ray.api.BaseActor;
/**
* ExecutionTask is minimal execution unit.
@@ -11,9 +11,9 @@ import org.ray.api.RayActor;
public class ExecutionTask implements Serializable {
private int taskId;
private int taskIndex;
private RayActor worker;
private BaseActor worker;
public ExecutionTask(int taskId, int taskIndex, RayActor worker) {
public ExecutionTask(int taskId, int taskIndex, BaseActor worker) {
this.taskId = taskId;
this.taskIndex = taskIndex;
this.worker = worker;
@@ -35,11 +35,11 @@ public class ExecutionTask implements Serializable {
this.taskIndex = taskIndex;
}
public RayActor getWorker() {
public BaseActor getWorker() {
return worker;
}
public void setWorker(RayActor worker) {
public void setWorker(BaseActor worker) {
this.worker = worker;
}
}
streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/schedule/JobSchedulerImpl.java
+2 -1
View File
@@ -3,6 +3,7 @@ package org.ray.streaming.runtime.schedule;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.ray.api.BaseActor;
import org.ray.api.Ray;
import org.ray.api.RayActor;
import org.ray.api.RayObject;
@@ -57,7 +58,7 @@ public class JobSchedulerImpl implements JobScheduler {
List<ExecutionTask> executionTasks = executionNode.getExecutionTasks();
for (ExecutionTask executionTask : executionTasks) {
int taskId = executionTask.getTaskId();
RayActor worker = executionTask.getWorker();
BaseActor worker = executionTask.getWorker();
switch (executionNode.getLanguage()) {
case JAVA:
RayActor<JobWorker> jobWorker = (RayActor<JobWorker>) worker;
streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/schedule/TaskAssignerImpl.java
+2 -2
View File
@@ -4,8 +4,8 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.ray.api.BaseActor;
import org.ray.api.Ray;
import org.ray.api.RayActor;
import org.ray.streaming.jobgraph.JobEdge;
import org.ray.streaming.jobgraph.JobGraph;
import org.ray.streaming.jobgraph.JobVertex;
@@ -58,7 +58,7 @@ public class TaskAssignerImpl implements TaskAssigner {
return new ExecutionGraph(executionNodes);
}
private RayActor createWorker(JobVertex jobVertex) {
private BaseActor createWorker(JobVertex jobVertex) {
switch (jobVertex.getLanguage()) {
case PYTHON:
return Ray.createPyActor(
streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/worker/tasks/StreamTask.java
+3 -3
View File
@@ -4,8 +4,8 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.ray.api.BaseActor;
import org.ray.api.Ray;
import org.ray.api.RayActor;
import org.ray.api.id.ActorId;
import org.ray.streaming.api.collector.Collector;
import org.ray.streaming.api.context.RuntimeContext;
@@ -65,7 +65,7 @@ public abstract class StreamTask implements Runnable {
List<Collector> collectors = new ArrayList<>();
for (ExecutionEdge edge : outputEdges) {
Map<String, ActorId> outputActorIds = new HashMap<>();
Map<Integer, RayActor> taskId2Worker = executionGraph
Map<Integer, BaseActor> taskId2Worker = executionGraph
.getTaskId2WorkerByNodeId(edge.getTargetNodeId());
taskId2Worker.forEach((targetTaskId, targetActor) -> {
String queueName = ChannelID.genIdStr(taskId, targetTaskId, executionGraph.getBuildTime());
@@ -91,7 +91,7 @@ public abstract class StreamTask implements Runnable {
List<ExecutionEdge> inputEdges = executionNode.getInputsEdges();
Map<String, ActorId> inputActorIds = new HashMap<>();
for (ExecutionEdge edge : inputEdges) {
Map<Integer, RayActor> taskId2Worker = executionGraph
Map<Integer, BaseActor> taskId2Worker = executionGraph
.getTaskId2WorkerByNodeId(edge.getSrcNodeId());
taskId2Worker.forEach((srcTaskId, srcActor) -> {
String queueName = ChannelID.genIdStr(srcTaskId, taskId, executionGraph.getBuildTime());