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

[Streaming] Supports multiple downstream collector (#9240)

Browse Source
This commit is contained in:
Lixin Wei
2020-07-03 11:05:07 +08:00
committed by GitHub
parent 4b62a888cc
commit aea3d53545
15 changed files with 134 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/collector/OutputCollector.java
+6 -6
View File
@@ -9,7 +9,7 @@ import io.ray.streaming.message.Record;
import io.ray.streaming.runtime.serialization.CrossLangSerializer;
import io.ray.streaming.runtime.serialization.JavaSerializer;
import io.ray.streaming.runtime.serialization.Serializer;
import io.ray.streaming.runtime.transfer.ChannelID;
import io.ray.streaming.runtime.transfer.ChannelId;
import io.ray.streaming.runtime.transfer.DataWriter;
import java.nio.ByteBuffer;
import java.util.Collection;
@@ -20,7 +20,7 @@ public class OutputCollector implements Collector<Record> {
private static final Logger LOGGER = LoggerFactory.getLogger(OutputCollector.class);
private final DataWriter writer;
private final ChannelID[] outputQueues;
private final ChannelId[] outputQueues;
private final Collection<BaseActorHandle> targetActors;
private final Language[] targetLanguages;
private final Partition partition;
@@ -28,18 +28,18 @@ public class OutputCollector implements Collector<Record> {
private final Serializer crossLangSerializer = new CrossLangSerializer();
public OutputCollector(DataWriter writer,
Collection<String> outputQueueIds,
Collection<String> outputChannelIds,
Collection<BaseActorHandle> targetActors,
Partition partition) {
this.writer = writer;
this.outputQueues = outputQueueIds.stream().map(ChannelID::from).toArray(ChannelID[]::new);
this.outputQueues = outputChannelIds.stream().map(ChannelId::from).toArray(ChannelId[]::new);
this.targetActors = targetActors;
this.targetLanguages = targetActors.stream()
.map(actor -> actor instanceof PyActorHandle ? Language.PYTHON : Language.JAVA)
.toArray(Language[]::new);
this.partition = partition;
LOGGER.debug("OutputCollector constructed, outputQueueIds:{}, partition:{}.",
outputQueueIds, this.partition);
LOGGER.debug("OutputCollector constructed, outputChannelIds:{}, partition:{}.",
outputChannelIds, this.partition);
}
@Override
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionEdge.java
+4
View File
@@ -52,6 +52,10 @@ public class ExecutionEdge implements Serializable {
return targetExecutionVertex;
}
public String getTargetExecutionJobVertexName() {
return getTargetExecutionVertex().getExecutionJobVertexName();
}
public int getSourceVertexId() {
return sourceExecutionVertex.getExecutionVertexId();
}
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelCreationParametersBuilder.java
+7 -6
View File
@@ -12,7 +12,6 @@ import io.ray.runtime.functionmanager.PyFunctionDescriptor;
import io.ray.streaming.runtime.worker.JobWorker;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
/**
* Save channel initial parameters needed by DataWriter/DataReader.
@@ -108,26 +107,28 @@ public class ChannelCreationParametersBuilder {
public ChannelCreationParametersBuilder buildInputQueueParameters(
List<String> queues,
Map<String, BaseActorHandle> actors) {
List<BaseActorHandle> actors) {
return buildParameters(queues, actors, javaWriterAsyncFuncDesc, javaWriterSyncFuncDesc,
pyWriterAsyncFunctionDesc, pyWriterSyncFunctionDesc);
}
public ChannelCreationParametersBuilder buildOutputQueueParameters(List<String> queues,
Map<String, BaseActorHandle> actors) {
List<BaseActorHandle> actors) {
return buildParameters(queues, actors, javaReaderAsyncFuncDesc, javaReaderSyncFuncDesc,
pyReaderAsyncFunctionDesc, pyReaderSyncFunctionDesc);
}
private ChannelCreationParametersBuilder buildParameters(List<String> queues,
Map<String, BaseActorHandle> actors,
List<BaseActorHandle> actors,
JavaFunctionDescriptor javaAsyncFunctionDesc, JavaFunctionDescriptor javaSyncFunctionDesc,
PyFunctionDescriptor pyAsyncFunctionDesc, PyFunctionDescriptor pySyncFunctionDesc
) {
parameters = new ArrayList<>(queues.size());
for (String queue : queues) {
for (int i = 0; i < queues.size(); ++i) {
String queue = queues.get(i);
BaseActorHandle actor = actors.get(i);
Parameter parameter = new Parameter();
BaseActorHandle actor = actors.get(queue);
Preconditions.checkArgument(actor != null);
parameter.setActorId(actor.getId());
/// LocalModeRayActor used in single-process mode.
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelID.java → streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelId.java
+18 -18
View File
@@ -15,7 +15,7 @@ import sun.nio.ch.DirectBuffer;
* ChannelID is used to identify a transfer channel between a upstream worker
* and downstream worker.
*/
public class ChannelID {
public class ChannelId {
public static final int ID_LENGTH = 20;
private static final FinalizableReferenceQueue REFERENCE_QUEUE = new FinalizableReferenceQueue();
// This ensures that the FinalizablePhantomReference itself is not garbage-collected.
@@ -27,7 +27,7 @@ public class ChannelID {
private final long address;
private final long nativeIdPtr;
private ChannelID(String strId, byte[] idBytes) {
private ChannelId(String strId, byte[] idBytes) {
this.strId = strId;
this.bytes = idBytes;
ByteBuffer directBuffer = ByteBuffer.allocateDirect(ID_LENGTH);
@@ -36,7 +36,7 @@ public class ChannelID {
this.buffer = directBuffer;
this.address = ((DirectBuffer) (buffer)).address();
long nativeIdPtr = 0;
nativeIdPtr = createNativeID(address);
nativeIdPtr = createNativeId(address);
this.nativeIdPtr = nativeIdPtr;
}
@@ -72,7 +72,7 @@ public class ChannelID {
if (o == null || getClass() != o.getClass()) {
return false;
}
ChannelID that = (ChannelID) o;
ChannelId that = (ChannelId) o;
return strId.equals(that.strId);
}
@@ -81,33 +81,33 @@ public class ChannelID {
return strId.hashCode();
}
private static native long createNativeID(long idAddress);
private static native long createNativeId(long idAddress);
private static native void destroyNativeID(long nativeIdPtr);
private static native void destroyNativeId(long nativeIdPtr);
/**
* @param id hex string representation of channel id
*/
public static ChannelID from(String id) {
return from(id, ChannelID.idStrToBytes(id));
public static ChannelId from(String id) {
return from(id, ChannelId.idStrToBytes(id));
}
/**
* @param idBytes bytes representation of channel id
*/
public static ChannelID from(byte[] idBytes) {
public static ChannelId from(byte[] idBytes) {
return from(idBytesToStr(idBytes), idBytes);
}
private static ChannelID from(String strID, byte[] idBytes) {
ChannelID id = new ChannelID(strID, idBytes);
private static ChannelId from(String strID, byte[] idBytes) {
ChannelId id = new ChannelId(strID, idBytes);
long nativeIdPtr = id.nativeIdPtr;
if (nativeIdPtr != 0) {
Reference<ChannelID> reference =
new FinalizablePhantomReference<ChannelID>(id, REFERENCE_QUEUE) {
Reference<ChannelId> reference =
new FinalizablePhantomReference<ChannelId>(id, REFERENCE_QUEUE) {
@Override
public void finalizeReferent() {
destroyNativeID(nativeIdPtr);
destroyNativeId(nativeIdPtr);
references.remove(this);
}
};
@@ -122,7 +122,7 @@ public class ChannelID {
public static String genRandomIdStr() {
StringBuilder sb = new StringBuilder();
Random random = new Random();
for (int i = 0; i < ChannelID.ID_LENGTH * 2; ++i) {
for (int i = 0; i < ChannelId.ID_LENGTH * 2; ++i) {
sb.append((char) (random.nextInt(6) + 'A'));
}
return sb.toString();
@@ -156,7 +156,7 @@ public class ChannelID {
channelName[18] = (byte) ((toTaskId & 0xffff) >> 8);
channelName[19] = (byte) (toTaskId & 0xff);
return ChannelID.idBytesToStr(channelName);
return ChannelId.idBytesToStr(channelName);
}
/**
@@ -165,7 +165,7 @@ public class ChannelID {
*/
static byte[] idStrToBytes(String id) {
byte[] idBytes = BaseEncoding.base16().decode(id.toUpperCase());
assert idBytes.length == ChannelID.ID_LENGTH;
assert idBytes.length == ChannelId.ID_LENGTH;
return idBytes;
}
@@ -174,7 +174,7 @@ public class ChannelID {
* @return hex string representation of channel id
*/
static String idBytesToStr(byte[] id) {
assert id.length == ChannelID.ID_LENGTH;
assert id.length == ChannelId.ID_LENGTH;
return BaseEncoding.base16().encode(id).toLowerCase();
}
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInitException.java
-24
View File
@@ -1,24 +0,0 @@
package io.ray.streaming.runtime.transfer;
import java.util.ArrayList;
import java.util.List;
public class ChannelInitException extends Exception {
private final List<byte[]> abnormalQueues;
public ChannelInitException(String message, List<byte[]> abnormalQueues) {
super(message);
this.abnormalQueues = abnormalQueues;
}
public List<byte[]> getAbnormalChannels() {
return abnormalQueues;
}
public List<String> getAbnormalChannelsString() {
List<String> res = new ArrayList<>();
abnormalQueues.forEach(ele -> res.add(ChannelID.idBytesToStr(ele)));
return res;
}
}
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInterruptException.java
-11
View File
@@ -1,11 +0,0 @@
package io.ray.streaming.runtime.transfer;
public class ChannelInterruptException extends RuntimeException {
public ChannelInterruptException() {
super();
}
public ChannelInterruptException(String message) {
super(message);
}
}
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java
+4 -5
View File
@@ -9,7 +9,6 @@ import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,14 +29,14 @@ public class DataReader {
* @param workerConfig configuration
*/
public DataReader(List<String> inputChannels,
Map<String, BaseActorHandle> fromActors,
List<BaseActorHandle> fromActors,
StreamingWorkerConfig workerConfig) {
Preconditions.checkArgument(inputChannels.size() > 0);
Preconditions.checkArgument(inputChannels.size() == fromActors.size());
ChannelCreationParametersBuilder initialParameters =
new ChannelCreationParametersBuilder().buildInputQueueParameters(inputChannels, fromActors);
byte[][] inputChannelsBytes = inputChannels.stream()
.map(ChannelID::idStrToBytes).toArray(byte[][]::new);
.map(ChannelId::idStrToBytes).toArray(byte[][]::new);
long[] seqIds = new long[inputChannels.size()];
long[] msgIds = new long[inputChannels.size()];
for (int i = 0; i < inputChannels.size(); i++) {
@@ -227,9 +226,9 @@ public class DataReader {
}
private String getQidString(ByteBuffer buffer) {
byte[] bytes = new byte[ChannelID.ID_LENGTH];
byte[] bytes = new byte[ChannelId.ID_LENGTH];
buffer.get(bytes);
return ChannelID.idBytesToStr(bytes);
return ChannelId.idBytesToStr(bytes);
}
public int getMagicNum() {
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataWriter.java
+5 -6
View File
@@ -8,7 +8,6 @@ import io.ray.streaming.runtime.util.Platform;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,14 +33,14 @@ public class DataWriter {
* @param workerConfig configuration
*/
public DataWriter(List<String> outputChannels,
Map<String, BaseActorHandle> toActors,
List<BaseActorHandle> toActors,
StreamingWorkerConfig workerConfig) {
Preconditions.checkArgument(!outputChannels.isEmpty());
Preconditions.checkArgument(outputChannels.size() == toActors.size());
ChannelCreationParametersBuilder initialParameters =
new ChannelCreationParametersBuilder().buildOutputQueueParameters(outputChannels, toActors);
byte[][] outputChannelsBytes = outputChannels.stream()
.map(ChannelID::idStrToBytes).toArray(byte[][]::new);
.map(ChannelId::idStrToBytes).toArray(byte[][]::new);
long channelSize = workerConfig.transferConfig.channelSize();
long[] msgIds = new long[outputChannels.size()];
for (int i = 0; i < outputChannels.size(); i++) {
@@ -70,7 +69,7 @@ public class DataWriter {
* @param id channel id
* @param item message item data section is specified by [position, limit).
*/
public void write(ChannelID id, ByteBuffer item) {
public void write(ChannelId id, ByteBuffer item) {
int size = item.remaining();
ensureBuffer(size);
buffer.clear();
@@ -85,10 +84,10 @@ public class DataWriter {
* @param item message item data section is specified by [position, limit).
* item doesn't have to be a direct buffer.
*/
public void write(Set<ChannelID> ids, ByteBuffer item) {
public void write(Set<ChannelId> ids, ByteBuffer item) {
int size = item.remaining();
ensureBuffer(size);
for (ChannelID id : ids) {
for (ChannelId id : ids) {
buffer.clear();
buffer.put(item.duplicate());
writeMessageNative(nativeWriterPtr, id.getNativeIdPtr(), bufferAddress, size);
streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java
+43 -21
View File
@@ -4,12 +4,13 @@ import io.ray.api.BaseActorHandle;
import io.ray.api.Ray;
import io.ray.streaming.api.collector.Collector;
import io.ray.streaming.api.context.RuntimeContext;
import io.ray.streaming.api.partition.Partition;
import io.ray.streaming.runtime.config.worker.WorkerInternalConfig;
import io.ray.streaming.runtime.core.collector.OutputCollector;
import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionEdge;
import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
import io.ray.streaming.runtime.core.processor.Processor;
import io.ray.streaming.runtime.transfer.ChannelID;
import io.ray.streaming.runtime.transfer.ChannelId;
import io.ray.streaming.runtime.transfer.DataReader;
import io.ray.streaming.runtime.transfer.DataWriter;
import io.ray.streaming.runtime.worker.JobWorker;
@@ -18,6 +19,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,45 +62,65 @@ public abstract class StreamTask implements Runnable {
WorkerInternalConfig.OP_NAME_INTERNAL, executionVertex.getExecutionJobVertexName());
// producer
List<ExecutionEdge> outputEdges = executionVertex.getOutputEdges();
Map<String, BaseActorHandle> outputActors = new HashMap<>();
// merge all output edges to create writer
List<String> outputChannelIds = new ArrayList<>();
List<BaseActorHandle> targetActors = new ArrayList<>();
for (ExecutionEdge edge : outputEdges) {
String queueName = ChannelID.genIdStr(
String channelId = ChannelId.genIdStr(
taskId,
edge.getTargetExecutionVertex().getExecutionVertexId(),
executionVertex.getBuildTime());
outputActors.put(queueName, edge.getTargetExecutionVertex().getWorkerActor());
outputChannelIds.add(channelId);
targetActors.add(edge.getTargetExecutionVertex().getWorkerActor());
}
if (!outputActors.isEmpty()) {
List<String> channelIDs = new ArrayList<>();
outputActors.forEach((vertexId, actorId) -> {
channelIDs.add(vertexId);
});
if (!targetActors.isEmpty()) {
DataWriter writer = new DataWriter(
outputChannelIds, targetActors, jobWorker.getWorkerConfig()
);
DataWriter writer = new DataWriter(channelIDs, outputActors, jobWorker.getWorkerConfig());
collectors.add(new OutputCollector(writer, channelIDs, outputActors.values(),
executionVertex.getOutputEdges().get(0).getPartition()));
// create a collector for each output operator
Map<String, List<String>> opGroupedChannelId = new HashMap<>();
Map<String, List<BaseActorHandle>> opGroupedActor = new HashMap<>();
Map<String, Partition> opPartitionMap = new HashMap<>();
for (int i = 0; i < outputEdges.size(); ++i) {
ExecutionEdge edge = outputEdges.get(i);
String opName = edge.getTargetExecutionJobVertexName();
if (!opPartitionMap.containsKey(opName)) {
opGroupedChannelId.put(opName, new ArrayList<>());
opGroupedActor.put(opName, new ArrayList<>());
}
opGroupedChannelId.get(opName).add(outputChannelIds.get(i));
opGroupedActor.get(opName).add(targetActors.get(i));
opPartitionMap.put(opName, edge.getPartition());
}
opPartitionMap.keySet().forEach(opName -> {
collectors.add(new OutputCollector(
writer, opGroupedChannelId.get(opName),
opGroupedActor.get(opName), opPartitionMap.get(opName)
));
});
}
// consumer
List<ExecutionEdge> inputEdges = executionVertex.getInputEdges();
Map<String, BaseActorHandle> inputActors = new HashMap<>();
List<String> inputChannelIds = new ArrayList<>();
List<BaseActorHandle> inputActors = new ArrayList<>();
for (ExecutionEdge edge : inputEdges) {
String queueName = ChannelID.genIdStr(
String queueName = ChannelId.genIdStr(
edge.getSourceExecutionVertex().getExecutionVertexId(),
taskId,
executionVertex.getBuildTime());
inputActors.put(queueName, edge.getSourceExecutionVertex().getWorkerActor());
inputChannelIds.add(queueName);
inputActors.add(edge.getSourceExecutionVertex().getWorkerActor());
}
if (!inputActors.isEmpty()) {
List<String> channelIDs = new ArrayList<>();
inputActors.forEach((k, v) -> {
channelIDs.add(k);
});
LOG.info("Register queue consumer, queues {}.", channelIDs);
reader = new DataReader(channelIDs, inputActors, jobWorker.getWorkerConfig());
LOG.info("Register queue consumer, channels {}.", inputChannelIds);
reader = new DataReader(inputChannelIds, inputActors, jobWorker.getWorkerConfig());
}
RuntimeContext runtimeContext = new StreamingRuntimeContext(executionVertex,
streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/StreamingQueueTest.java
+2 -2
View File
@@ -9,7 +9,7 @@ import io.ray.streaming.api.function.impl.FlatMapFunction;
import io.ray.streaming.api.function.impl.ReduceFunction;
import io.ray.streaming.api.stream.DataStreamSource;
import io.ray.streaming.runtime.BaseUnitTest;
import io.ray.streaming.runtime.transfer.ChannelID;
import io.ray.streaming.runtime.transfer.ChannelId;
import io.ray.streaming.runtime.util.EnvUtil;
import io.ray.streaming.util.Config;
import java.io.File;
@@ -99,7 +99,7 @@ public class StreamingQueueTest extends BaseUnitTest implements Serializable {
List<String> inputQueueList = new ArrayList<>();
int queueNum = 2;
for (int i = 0; i < queueNum; ++i) {
String qid = ChannelID.genRandomIdStr();
String qid = ChannelId.genRandomIdStr();
LOGGER.info("getRandomQueueId: {}", qid);
inputQueueList.add(qid);
outputQueueList.add(qid);
streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/Worker.java
+7 -6
View File
@@ -5,7 +5,7 @@ import io.ray.api.Ray;
import io.ray.api.ActorHandle;
import io.ray.runtime.functionmanager.JavaFunctionDescriptor;
import io.ray.streaming.runtime.config.StreamingWorkerConfig;
import io.ray.streaming.runtime.transfer.ChannelID;
import io.ray.streaming.runtime.transfer.ChannelId;
import io.ray.streaming.runtime.transfer.ChannelCreationParametersBuilder;
import io.ray.streaming.runtime.transfer.DataMessage;
import io.ray.streaming.runtime.transfer.DataReader;
@@ -14,6 +14,7 @@ import io.ray.streaming.runtime.transfer.TransferHandler;
import io.ray.streaming.util.Config;
import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -53,7 +54,7 @@ class ReaderWorker extends Worker {
private String name = null;
private List<String> inputQueueList = null;
Map<String, BaseActorHandle> inputActors = new HashMap<>();
List<BaseActorHandle> inputActors = new ArrayList<>();
private DataReader dataReader = null;
private long handler = 0;
private ActorHandle<WriterWorker> peerActor = null;
@@ -88,7 +89,7 @@ class ReaderWorker extends Worker {
LOGGER.info("java.library.path = {}", System.getProperty("java.library.path"));
for (String queue : this.inputQueueList) {
inputActors.put(queue, this.peerActor);
inputActors.add(this.peerActor);
LOGGER.info("ReaderWorker actorId: {}", this.peerActor.getId());
}
@@ -173,7 +174,7 @@ class WriterWorker extends Worker {
private String name = null;
private List<String> outputQueueList = null;
Map<String, BaseActorHandle> outputActors = new HashMap<>();
List<BaseActorHandle> outputActors = new ArrayList<>();
DataWriter dataWriter = null;
ActorHandle<ReaderWorker> peerActor = null;
int msgCount = 0;
@@ -205,7 +206,7 @@ class WriterWorker extends Worker {
LOGGER.info("WriterWorker init:");
for (String queue : this.outputQueueList) {
outputActors.put(queue, this.peerActor);
outputActors.add(this.peerActor);
LOGGER.info("WriterWorker actorId: {}", this.peerActor.getId());
}
@@ -260,7 +261,7 @@ class WriterWorker extends Worker {
}
bb.clear();
ChannelID qid = ChannelID.from(outputQueueList.get(j));
ChannelId qid = ChannelId.from(outputQueueList.get(j));
dataWriter.write(qid, bb);
}
}
streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/ChannelIDTest.java → streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/ChannelIdTest.java
+4 -4
View File
@@ -6,7 +6,7 @@ import io.ray.streaming.runtime.BaseUnitTest;
import io.ray.streaming.runtime.util.EnvUtil;
import org.testng.annotations.Test;
public class ChannelIDTest extends BaseUnitTest {
public class ChannelIdTest extends BaseUnitTest {
static {
EnvUtil.loadNativeLibraries();
@@ -14,9 +14,9 @@ public class ChannelIDTest extends BaseUnitTest {
@Test
public void testIdStrToBytes() {
String idStr = ChannelID.genRandomIdStr();
assertEquals(idStr.length(), ChannelID.ID_LENGTH * 2);
assertEquals(ChannelID.idStrToBytes(idStr).length, ChannelID.ID_LENGTH);
String idStr = ChannelId.genRandomIdStr();
assertEquals(idStr.length(), ChannelId.ID_LENGTH * 2);
assertEquals(ChannelId.idStrToBytes(idStr).length, ChannelId.ID_LENGTH);
}
}