You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by re...@apache.org on 2023/03/29 02:23:53 UTC
[incubator-celeborn] 38/42: [CELEBORN-440][FLINK] Implement numConcurrentReading in RemoteShuffleInputGate.
This is an automated email from the ASF dual-hosted git repository.
rexxiong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit aad554c4925e752eac99b369f8842fb6d56117ce
Author: fengmingxiao.fmx <fe...@alibaba-inc.com>
AuthorDate: Wed Mar 22 12:11:40 2023 +0800
[CELEBORN-440][FLINK] Implement numConcurrentReading in RemoteShuffleInputGate.
---
.../plugin/flink/RemoteBufferStreamReader.java | 8 +-
.../plugin/flink/RemoteShuffleInputGate.java | 13 ++-
.../flink/RemoteShuffleInputGateFactory.java | 8 +-
.../plugin/flink/RemoteShuffleInputGate.java | 111 +++++++++++----------
.../flink/RemoteShuffleInputGateFactory.java | 8 +-
.../plugin/flink/RemoteShuffleResultPartition.java | 3 +-
dev/reformat | 1 +
7 files changed, 86 insertions(+), 66 deletions(-)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
index 08c0d1fdf..f972a2788 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
@@ -42,7 +42,8 @@ public class RemoteBufferStreamReader extends CreditListener {
private int partitionId;
private int subPartitionIndexStart;
private int subPartitionIndexEnd;
- private boolean isOpen;
+ // To indicate that this reader has opened.
+ private boolean isOpened;
private Consumer<ByteBuf> dataListener;
private Consumer<Throwable> failureListener;
private RssBufferStream bufferStream;
@@ -89,11 +90,10 @@ public class RemoteBufferStreamReader extends CreditListener {
} catch (InterruptedException e) {
throw new RuntimeException("Failed to openStream.", e);
}
- isOpen = true;
+ isOpened = true;
}
public void close() {
- isOpen = false;
// need set closed first before remove Handler
closed = true;
if (this.bufferStream != null) {
@@ -108,7 +108,7 @@ public class RemoteBufferStreamReader extends CreditListener {
}
public boolean isOpened() {
- return isOpen;
+ return isOpened;
}
public void notifyAvailableCredits(int numCredits) {
diff --git a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
index 1192125cf..dd6eb9882 100644
--- a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
+++ b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
@@ -96,6 +96,7 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
new TransferBufferPool(Collections.emptySet());
private final List<RemoteBufferStreamReader> bufferReaders = new ArrayList<>();
+
private final List<InputChannelInfo> channelsInfo;
/** Map from channel index to shuffle client index. */
private final int[] clientIndexMap;
@@ -123,10 +124,9 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
/** Number of pending {@link EndOfData} events to be received. */
private long pendingEndOfDataEvents;
+
/** Max concurrent reader count */
- private int numConcurrentReading = Integer.MAX_VALUE;
- /** Keep compatibility with streaming mode. */
- private boolean shouldDrainOnEndOfData = true;
+ private final int numConcurrentReading;
/** Data decompressor. */
private final BufferDecompressor bufferDecompressor;
@@ -139,7 +139,8 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
int gateIndex,
InputGateDeploymentDescriptor gateDescriptor,
SupplierWithException<BufferPool, IOException> bufferPoolFactory,
- BufferDecompressor bufferDecompressor) {
+ BufferDecompressor bufferDecompressor,
+ int numConcurrentReading) {
this.taskName = taskName;
this.gateIndex = gateIndex;
@@ -164,6 +165,8 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
this.numUnconsumedSubpartitions = initShuffleReadClients();
this.pendingEndOfDataEvents = numUnconsumedSubpartitions;
this.channelsInfo = createChannelInfos();
+ this.numConcurrentReading = numConcurrentReading;
+ LOG.debug("Initial input gate with numConcurrentReading {}", this.numConcurrentReading);
}
private long initShuffleReadClients() {
@@ -541,7 +544,7 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
} else {
// the real end.
bufferReaders.get(clientIndexMap[channelInfo.getInputChannelIdx()]).close();
- // tryOpenSomeChannels();
+ tryOpenSomeChannels();
if (allReadersEOF()) {
availabilityHelper.getUnavailableToResetAvailable().complete(null);
}
diff --git a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
index ad3620518..42f8a2215 100644
--- a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
+++ b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
@@ -121,7 +121,13 @@ public class RemoteShuffleInputGateFactory {
SupplierWithException<BufferPool, IOException> bufferPoolFactory,
BufferDecompressor bufferDecompressor) {
return new RemoteShuffleInputGate(
- celebornConf, owningTaskName, gateIndex, igdd, bufferPoolFactory, bufferDecompressor);
+ celebornConf,
+ owningTaskName,
+ gateIndex,
+ igdd,
+ bufferPoolFactory,
+ bufferDecompressor,
+ numConcurrentReading);
}
private SupplierWithException<BufferPool, IOException> createBufferPoolFactory(
diff --git a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
index 33581d791..e98b0ef80 100644
--- a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
+++ b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
@@ -17,12 +17,22 @@
package org.apache.celeborn.plugin.flink;
-import org.apache.celeborn.common.CelebornConf;
-import org.apache.celeborn.common.identity.UserIdentifier;
-import org.apache.celeborn.plugin.flink.buffer.BufferPacker;
-import org.apache.celeborn.plugin.flink.buffer.TransferBufferPool;
-import org.apache.celeborn.plugin.flink.readclient.FlinkShuffleClientImpl;
-import org.apache.celeborn.plugin.flink.utils.BufferUtils;
+import static org.apache.celeborn.plugin.flink.utils.Utils.checkState;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentProvider;
@@ -66,21 +76,12 @@ import org.apache.flink.util.function.SupplierWithException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Optional;
-import java.util.Queue;
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-import static org.apache.celeborn.plugin.flink.utils.Utils.checkState;
+import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.plugin.flink.buffer.BufferPacker;
+import org.apache.celeborn.plugin.flink.buffer.TransferBufferPool;
+import org.apache.celeborn.plugin.flink.readclient.FlinkShuffleClientImpl;
+import org.apache.celeborn.plugin.flink.utils.BufferUtils;
/** A {@link IndexedInputGate} which ingest data from remote shuffle workers. */
public class RemoteShuffleInputGate extends IndexedInputGate {
@@ -112,13 +113,13 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
private final List<RemoteBufferStreamReader> bufferReaders = new ArrayList<>();
private final List<InputChannelInfo> channelsInfo;
/** Map from channel index to shuffle client index. */
- private final int[] clientIndexMap;
+ private final int[] channelIndexToReaderIndex;
/** Map from shuffle client index to channel index. */
- private final int[] channelIndexMap;
+ private final int[] readerIndexToChannelIndex;
/** The number of subpartitions that has not consumed per channel. */
- private final int[] numSubPartitionsHasNotConsumed;
+ private final int[] numSubPartitionsNotConsumed;
/** The overall number of subpartitions that has not been consumed. */
private long numUnconsumedSubpartitions;
@@ -137,6 +138,7 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
/** Number of pending {@link EndOfData} events to be received. */
private long pendingEndOfDataEvents;
+
/** Max concurrent reader count */
private int numConcurrentReading = Integer.MAX_VALUE;
/** Keep compatibility with streaming mode. */
@@ -153,7 +155,8 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
int gateIndex,
InputGateDeploymentDescriptor gateDescriptor,
SupplierWithException<BufferPool, IOException> bufferPoolFactory,
- BufferDecompressor bufferDecompressor) {
+ BufferDecompressor bufferDecompressor,
+ int numConcurrentReading) {
this.taskName = taskName;
this.gateIndex = gateIndex;
@@ -161,9 +164,9 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
this.bufferPoolFactory = bufferPoolFactory;
int numChannels = gateDescriptor.getShuffleDescriptors().length;
- this.clientIndexMap = new int[numChannels];
- this.channelIndexMap = new int[numChannels];
- this.numSubPartitionsHasNotConsumed = new int[numChannels];
+ this.channelIndexToReaderIndex = new int[numChannels];
+ this.readerIndexToChannelIndex = new int[numChannels];
+ this.numSubPartitionsNotConsumed = new int[numChannels];
this.bufferDecompressor = bufferDecompressor;
RemoteShuffleDescriptor remoteShuffleDescriptor =
@@ -175,12 +178,13 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
celebornConf,
new UserIdentifier("default", "default"));
- this.numUnconsumedSubpartitions = initShuffleReadClients();
- this.pendingEndOfDataEvents = numUnconsumedSubpartitions;
- this.channelsInfo = createChannelInfos();
+ initShuffleReadClients();
+ channelsInfo = createChannelInfos();
+ this.numConcurrentReading = numConcurrentReading;
+ LOG.debug("Initial input gate with numConcurrentReading {}", this.numConcurrentReading);
}
- private long initShuffleReadClients() {
+ private void initShuffleReadClients() {
int startSubIdx = gateDescriptor.getConsumedSubpartitionIndex();
int endSubIdx = gateDescriptor.getConsumedSubpartitionIndex();
int numSubpartitionsPerChannel = endSubIdx - startSubIdx + 1;
@@ -192,7 +196,7 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
.mapToObj(i -> Pair.of(i, gateDescriptor.getShuffleDescriptors()[i]))
.collect(Collectors.toList());
- int clientIndex = 0;
+ int readerIndex = 0;
for (Pair<Integer, ShuffleDescriptor> descriptor : descriptors) {
RemoteShuffleDescriptor remoteDescriptor = (RemoteShuffleDescriptor) descriptor.getRight();
ShuffleResourceDescriptor shuffleDescriptor =
@@ -213,13 +217,15 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
getFailureListener(remoteDescriptor.getResultPartitionID()));
bufferReaders.add(reader);
- numSubPartitionsHasNotConsumed[descriptor.getLeft()] = numSubpartitionsPerChannel;
+ numSubPartitionsNotConsumed[descriptor.getLeft()] = numSubpartitionsPerChannel;
numUnconsumedSubpartitions += numSubpartitionsPerChannel;
- clientIndexMap[descriptor.getLeft()] = clientIndex;
- channelIndexMap[clientIndex] = descriptor.getLeft();
- ++clientIndex;
+ channelIndexToReaderIndex[descriptor.getLeft()] = readerIndex;
+ readerIndexToChannelIndex[readerIndex] = descriptor.getLeft();
+ ++readerIndex;
}
- return numUnconsumedSubpartitions;
+
+ this.numUnconsumedSubpartitions = numUnconsumedSubpartitions;
+ this.pendingEndOfDataEvents = numUnconsumedSubpartitions;
}
/** Setup gate and build network connections. */
@@ -317,9 +323,9 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
Throwable closeException = null;
// Do not check closed flag, thus to allow calling this method from both task thread and
// cancel thread.
- for (RemoteBufferStreamReader shuffleReadClient : bufferReaders) {
+ for (RemoteBufferStreamReader reader : bufferReaders) {
try {
- shuffleReadClient.close();
+ reader.close();
} catch (Throwable throwable) {
closeException = closeException == null ? throwable : closeException;
LOG.error("Failed to close shuffle read client.", throwable);
@@ -378,7 +384,7 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
/** Try to open more readers to {@link #numConcurrentReading}. */
private void tryOpenSomeChannels() throws IOException {
- List<RemoteBufferStreamReader> clientsToOpen = new ArrayList<>();
+ List<RemoteBufferStreamReader> readersToOpen = new ArrayList<>();
synchronized (lock) {
if (closed) {
@@ -393,25 +399,25 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
"Trying reader: {}, isOpened={}, numSubPartitionsHasNotConsumed={}.",
bufferStreamReader,
bufferStreamReader.isOpened(),
- numSubPartitionsHasNotConsumed[channelIndexMap[i]]);
+ numSubPartitionsNotConsumed[readerIndexToChannelIndex[i]]);
if (numOnGoing >= numConcurrentReading) {
break;
}
if (bufferStreamReader.isOpened()
- && numSubPartitionsHasNotConsumed[channelIndexMap[i]] > 0) {
+ && numSubPartitionsNotConsumed[readerIndexToChannelIndex[i]] > 0) {
numOnGoing++;
continue;
}
if (!bufferStreamReader.isOpened()) {
- clientsToOpen.add(bufferStreamReader);
+ readersToOpen.add(bufferStreamReader);
numOnGoing++;
}
}
}
- for (RemoteBufferStreamReader reader : clientsToOpen) {
+ for (RemoteBufferStreamReader reader : readersToOpen) {
reader.open(0);
}
}
@@ -545,17 +551,17 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
}
if (event.getClass() == EndOfPartitionEvent.class) {
checkState(
- numSubPartitionsHasNotConsumed[channelInfo.getInputChannelIdx()] > 0,
+ numSubPartitionsNotConsumed[channelInfo.getInputChannelIdx()] > 0,
"BUG -- EndOfPartitionEvent received repeatedly.");
- numSubPartitionsHasNotConsumed[channelInfo.getInputChannelIdx()]--;
+ numSubPartitionsNotConsumed[channelInfo.getInputChannelIdx()]--;
numUnconsumedSubpartitions--;
// not the real end.
- if (numSubPartitionsHasNotConsumed[channelInfo.getInputChannelIdx()] != 0) {
+ if (numSubPartitionsNotConsumed[channelInfo.getInputChannelIdx()] != 0) {
return Optional.empty();
} else {
// the real end.
- bufferReaders.get(clientIndexMap[channelInfo.getInputChannelIdx()]).close();
- // tryOpenSomeChannels();
+ bufferReaders.get(channelIndexToReaderIndex[channelInfo.getInputChannelIdx()]).close();
+ tryOpenSomeChannels();
if (allReadersEOF()) {
availabilityHelper.getUnavailableToResetAvailable().complete(null);
}
@@ -663,13 +669,12 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
new FakedMemorySegmentProvider(),
0,
new ThroughputCalculator(SystemClock.getInstance()),
- null ),
+ null),
channelIndex,
new ResultPartitionID(),
0,
new ConnectionID(
- new TaskManagerLocation(
- ResourceID.generate(), InetAddress.getLoopbackAddress(), 1),
+ new TaskManagerLocation(ResourceID.generate(), InetAddress.getLoopbackAddress(), 1),
0),
new LocalConnectionManager(),
0,
diff --git a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
index a3485721f..a8e05613e 100644
--- a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
+++ b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
@@ -118,7 +118,13 @@ public class RemoteShuffleInputGateFactory {
SupplierWithException<BufferPool, IOException> bufferPoolFactory,
BufferDecompressor bufferDecompressor) {
return new RemoteShuffleInputGate(
- this.celebornConf, owningTaskName, gateIndex, igdd, bufferPoolFactory, bufferDecompressor);
+ this.celebornConf,
+ owningTaskName,
+ gateIndex,
+ igdd,
+ bufferPoolFactory,
+ bufferDecompressor,
+ numConcurrentReading);
}
private SupplierWithException<BufferPool, IOException> createBufferPoolFactory(
diff --git a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
index d481c0f7a..aacf61647 100644
--- a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
+++ b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
@@ -56,8 +56,7 @@ import org.apache.celeborn.plugin.flink.utils.Utils;
* A {@link ResultPartition} which appends records and events to {@link SortBuffer} and after the
* {@link SortBuffer} is full, all data in the {@link SortBuffer} will be copied and spilled to the
* remote shuffle service in subpartition index order sequentially. Large records that can not be
- * appended to an empty {@link SortBuffer} will be
- * spilled directly.
+ * appended to an empty {@link SortBuffer} will be spilled directly.
*/
public class RemoteShuffleResultPartition extends ResultPartition {
diff --git a/dev/reformat b/dev/reformat
index 420d13618..26d74fed1 100755
--- a/dev/reformat
+++ b/dev/reformat
@@ -20,5 +20,6 @@ set -x
PROJECT_DIR="$(cd "`dirname "$0"`/.."; pwd)"
${PROJECT_DIR}/build/mvn spotless:apply -Pflink-1.14
+${PROJECT_DIR}/build/mvn spotless:apply -Pflink-1.15
${PROJECT_DIR}/build/mvn spotless:apply -Pspark-2.4
${PROJECT_DIR}/build/mvn spotless:apply -Pspark-3.3