From aea3d53545ddf1312fe16af5a25c54ed1b731222 Mon Sep 17 00:00:00 2001 From: Lixin Wei Date: Fri, 3 Jul 2020 11:05:07 +0800 Subject: [PATCH] [Streaming] Supports multiple downstream collector (#9240) --- .../core/collector/OutputCollector.java | 12 ++-- .../graph/executiongraph/ExecutionEdge.java | 4 ++ .../ChannelCreationParametersBuilder.java | 13 ++-- .../{ChannelID.java => ChannelId.java} | 36 +++++------ .../transfer/ChannelInitException.java | 24 ------- .../transfer/ChannelInterruptException.java | 11 ---- .../runtime/transfer/DataReader.java | 9 ++- .../runtime/transfer/DataWriter.java | 11 ++-- .../runtime/worker/tasks/StreamTask.java | 64 +++++++++++++------ .../streamingqueue/StreamingQueueTest.java | 4 +- .../runtime/streamingqueue/Worker.java | 13 ++-- ...{ChannelIDTest.java => ChannelIdTest.java} | 8 +-- ...ray_streaming_runtime_transfer_ChannelID.h | 31 --------- ...y_streaming_runtime_transfer_ChannelId.cc} | 6 +- ...ray_streaming_runtime_transfer_ChannelId.h | 31 +++++++++ 15 files changed, 134 insertions(+), 143 deletions(-) rename streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/{ChannelID.java => ChannelId.java} (82%) delete mode 100644 streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInitException.java delete mode 100644 streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInterruptException.java rename streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/{ChannelIDTest.java => ChannelIdTest.java} (60%) delete mode 100644 streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelID.h rename streaming/src/lib/java/{io_ray_streaming_runtime_transfer_ChannelID.cc => io_ray_streaming_runtime_transfer_ChannelId.cc} (84%) create mode 100644 streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.h diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/collector/OutputCollector.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/collector/OutputCollector.java index 090d91123..0ee7e250a 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/collector/OutputCollector.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/collector/OutputCollector.java @@ -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 { private static final Logger LOGGER = LoggerFactory.getLogger(OutputCollector.class); private final DataWriter writer; - private final ChannelID[] outputQueues; + private final ChannelId[] outputQueues; private final Collection targetActors; private final Language[] targetLanguages; private final Partition partition; @@ -28,18 +28,18 @@ public class OutputCollector implements Collector { private final Serializer crossLangSerializer = new CrossLangSerializer(); public OutputCollector(DataWriter writer, - Collection outputQueueIds, + Collection outputChannelIds, Collection 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 diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionEdge.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionEdge.java index a98c77a16..ff6b549cb 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionEdge.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/graph/executiongraph/ExecutionEdge.java @@ -52,6 +52,10 @@ public class ExecutionEdge implements Serializable { return targetExecutionVertex; } + public String getTargetExecutionJobVertexName() { + return getTargetExecutionVertex().getExecutionJobVertexName(); + } + public int getSourceVertexId() { return sourceExecutionVertex.getExecutionVertexId(); } diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelCreationParametersBuilder.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelCreationParametersBuilder.java index 486d3aaaf..d5a9db374 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelCreationParametersBuilder.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelCreationParametersBuilder.java @@ -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 queues, - Map actors) { + List actors) { return buildParameters(queues, actors, javaWriterAsyncFuncDesc, javaWriterSyncFuncDesc, pyWriterAsyncFunctionDesc, pyWriterSyncFunctionDesc); } public ChannelCreationParametersBuilder buildOutputQueueParameters(List queues, - Map actors) { + List actors) { return buildParameters(queues, actors, javaReaderAsyncFuncDesc, javaReaderSyncFuncDesc, pyReaderAsyncFunctionDesc, pyReaderSyncFunctionDesc); } private ChannelCreationParametersBuilder buildParameters(List queues, - Map actors, + List 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. diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelID.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelId.java similarity index 82% rename from streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelID.java rename to streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelId.java index 08c7fbacb..4d0c38578 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelID.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelId.java @@ -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 reference = - new FinalizablePhantomReference(id, REFERENCE_QUEUE) { + Reference reference = + new FinalizablePhantomReference(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(); } diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInitException.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInitException.java deleted file mode 100644 index 3ac6d129f..000000000 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInitException.java +++ /dev/null @@ -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 abnormalQueues; - - public ChannelInitException(String message, List abnormalQueues) { - super(message); - this.abnormalQueues = abnormalQueues; - } - - public List getAbnormalChannels() { - return abnormalQueues; - } - - public List getAbnormalChannelsString() { - List res = new ArrayList<>(); - abnormalQueues.forEach(ele -> res.add(ChannelID.idBytesToStr(ele))); - return res; - } -} diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInterruptException.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInterruptException.java deleted file mode 100644 index d78a62829..000000000 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/ChannelInterruptException.java +++ /dev/null @@ -1,11 +0,0 @@ -package io.ray.streaming.runtime.transfer; - -public class ChannelInterruptException extends RuntimeException { - public ChannelInterruptException() { - super(); - } - - public ChannelInterruptException(String message) { - super(message); - } -} diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java index d12e21fd6..e23dca519 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataReader.java @@ -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 inputChannels, - Map fromActors, + List 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() { diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataWriter.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataWriter.java index f9d752fd7..7fc617cbe 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataWriter.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/transfer/DataWriter.java @@ -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 outputChannels, - Map toActors, + List 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 ids, ByteBuffer item) { + public void write(Set 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); diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java index 64cb0f85d..00f6bb094 100644 --- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java +++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java @@ -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 outputEdges = executionVertex.getOutputEdges(); - Map outputActors = new HashMap<>(); + + // merge all output edges to create writer + List outputChannelIds = new ArrayList<>(); + List 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 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> opGroupedChannelId = new HashMap<>(); + Map> opGroupedActor = new HashMap<>(); + Map 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 inputEdges = executionVertex.getInputEdges(); - Map inputActors = new HashMap<>(); + List inputChannelIds = new ArrayList<>(); + List 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 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, diff --git a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/StreamingQueueTest.java b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/StreamingQueueTest.java index 7dcccc735..b1760ceb6 100644 --- a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/StreamingQueueTest.java +++ b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/StreamingQueueTest.java @@ -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 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); diff --git a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/Worker.java b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/Worker.java index f7427e999..314dcf181 100644 --- a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/Worker.java +++ b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/streamingqueue/Worker.java @@ -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 inputQueueList = null; - Map inputActors = new HashMap<>(); + List inputActors = new ArrayList<>(); private DataReader dataReader = null; private long handler = 0; private ActorHandle 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 outputQueueList = null; - Map outputActors = new HashMap<>(); + List outputActors = new ArrayList<>(); DataWriter dataWriter = null; ActorHandle 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); } } diff --git a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/ChannelIDTest.java b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/ChannelIdTest.java similarity index 60% rename from streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/ChannelIDTest.java rename to streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/ChannelIdTest.java index 6a656222a..fe841844c 100644 --- a/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/ChannelIDTest.java +++ b/streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/transfer/ChannelIdTest.java @@ -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); } } \ No newline at end of file diff --git a/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelID.h b/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelID.h deleted file mode 100644 index c97613afa..000000000 --- a/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelID.h +++ /dev/null @@ -1,31 +0,0 @@ -/* DO NOT EDIT THIS FILE - it is machine generated */ -#include -/* Header for class io_ray_streaming_runtime_transfer_ChannelID */ - -#ifndef _Included_io_ray_streaming_runtime_transfer_ChannelID -#define _Included_io_ray_streaming_runtime_transfer_ChannelID -#ifdef __cplusplus -extern "C" { -#endif -#undef io_ray_streaming_runtime_transfer_ChannelID_ID_LENGTH -#define io_ray_streaming_runtime_transfer_ChannelID_ID_LENGTH 20L -/* - * Class: io_ray_streaming_runtime_transfer_ChannelID - * Method: createNativeID - * Signature: (J)J - */ -JNIEXPORT jlong JNICALL Java_io_ray_streaming_runtime_transfer_ChannelID_createNativeID - (JNIEnv *, jclass, jlong); - -/* - * Class: io_ray_streaming_runtime_transfer_ChannelID - * Method: destroyNativeID - * Signature: (J)V - */ -JNIEXPORT void JNICALL Java_io_ray_streaming_runtime_transfer_ChannelID_destroyNativeID - (JNIEnv *, jclass, jlong); - -#ifdef __cplusplus -} -#endif -#endif diff --git a/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelID.cc b/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.cc similarity index 84% rename from streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelID.cc rename to streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.cc index b8591fcc6..1666daf54 100644 --- a/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelID.cc +++ b/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.cc @@ -1,15 +1,15 @@ -#include "io_ray_streaming_runtime_transfer_ChannelID.h" +#include "io_ray_streaming_runtime_transfer_ChannelId.h" #include "streaming_jni_common.h" using namespace ray::streaming; -JNIEXPORT jlong JNICALL Java_io_ray_streaming_runtime_transfer_ChannelID_createNativeID( +JNIEXPORT jlong JNICALL Java_io_ray_streaming_runtime_transfer_ChannelId_createNativeId( JNIEnv *env, jclass cls, jlong qid_address) { auto id = ray::ObjectID::FromBinary( std::string(reinterpret_cast(qid_address), ray::ObjectID::Size())); return reinterpret_cast(new ray::ObjectID(id)); } -JNIEXPORT void JNICALL Java_io_ray_streaming_runtime_transfer_ChannelID_destroyNativeID( +JNIEXPORT void JNICALL Java_io_ray_streaming_runtime_transfer_ChannelId_destroyNativeId( JNIEnv *env, jclass cls, jlong native_id_ptr) { auto id = reinterpret_cast(native_id_ptr); STREAMING_CHECK(id != nullptr); diff --git a/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.h b/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.h new file mode 100644 index 000000000..8bd0c0aab --- /dev/null +++ b/streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.h @@ -0,0 +1,31 @@ +/* DO NOT EDIT THIS FILE - it is machine generated */ +#include +/* Header for class io_ray_streaming_runtime_transfer_ChannelId */ + +#ifndef _Included_io_ray_streaming_runtime_transfer_ChannelId +#define _Included_io_ray_streaming_runtime_transfer_ChannelId +#ifdef __cplusplus +extern "C" { +#endif +#undef io_ray_streaming_runtime_transfer_ChannelId_ID_LENGTH +#define io_ray_streaming_runtime_transfer_ChannelId_ID_LENGTH 20L +/* + * Class: io_ray_streaming_runtime_transfer_ChannelId + * Method: createNativeId + * Signature: (J)J + */ +JNIEXPORT jlong JNICALL Java_io_ray_streaming_runtime_transfer_ChannelId_createNativeId + (JNIEnv *, jclass, jlong); + +/* + * Class: io_ray_streaming_runtime_transfer_ChannelId + * Method: destroyNativeId + * Signature: (J)V + */ +JNIEXPORT void JNICALL Java_io_ray_streaming_runtime_transfer_ChannelId_destroyNativeId + (JNIEnv *, jclass, jlong); + +#ifdef __cplusplus +} +#endif +#endif