This is an automated email from the ASF dual-hosted git repository.
ethanfeng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new f0b9236ff [CELEBORN-340][FLINK] Reuse file channels in map partition
read. (#1276)
f0b9236ff is described below
commit f0b9236ff21b52610148444b833e97300ab422db
Author: Ethan Feng <[email protected]>
AuthorDate: Fri Feb 24 19:26:51 2023 +0800
[CELEBORN-340][FLINK] Reuse file channels in map partition read. (#1276)
---
.../common/network/server/BufferStreamManager.java | 50 ++++++++++++++--------
1 file changed, 31 insertions(+), 19 deletions(-)
diff --git
a/common/src/main/java/org/apache/celeborn/common/network/server/BufferStreamManager.java
b/common/src/main/java/org/apache/celeborn/common/network/server/BufferStreamManager.java
index ea87dffd8..e48f28765 100644
---
a/common/src/main/java/org/apache/celeborn/common/network/server/BufferStreamManager.java
+++
b/common/src/main/java/org/apache/celeborn/common/network/server/BufferStreamManager.java
@@ -18,6 +18,7 @@
package org.apache.celeborn.common.network.server;
import java.io.FileInputStream;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
@@ -117,9 +118,11 @@ public class BufferStreamManager {
streams.put(streamId, new StreamState(channel, fileInfo.getBufferSize()));
synchronized (activeMapPartitions) {
- MapDataPartition mapDataPartition =
- activeMapPartitions.computeIfAbsent(fileInfo, (f) -> new
MapDataPartition(fileInfo));
-
+ MapDataPartition mapDataPartition = activeMapPartitions.get(fileInfo);
+ if (mapDataPartition == null) {
+ mapDataPartition = new MapDataPartition(fileInfo);
+ activeMapPartitions.put(fileInfo, mapDataPartition);
+ }
mapDataPartition.addStream(streamId);
addCredit(initialCredit, streamId);
servingStreams.put(streamId, mapDataPartition);
@@ -214,6 +217,7 @@ public class BufferStreamManager {
memoryManager.recycleReadBuffer(buffer);
}
activeMapPartitions.remove(fileInfo);
+ mapDataPartition.close();
}
}
}
@@ -231,16 +235,21 @@ public class BufferStreamManager {
/** All available buffers can be used by the partition readers for
reading. */
protected Queue<ByteBuf> buffers;
- public MapDataPartition(FileInfo fileInfo) {
+ protected FileChannel dataFileChanel;
+ protected FileChannel indexChannel;
+
+ public MapDataPartition(FileInfo fileInfo) throws FileNotFoundException {
this.fileInfo = fileInfo;
readExecutor =
storageFetcherPool.getExecutorPool(fileInfo.getMountPoint());
+ this.dataFileChanel = new
FileInputStream(fileInfo.getFile()).getChannel();
+ this.indexChannel = new
FileInputStream(fileInfo.getIndexPath()).getChannel();
}
public synchronized void setupDataPartitionReader(
int startSubIndex, int endSubIndex, long streamId) throws IOException {
DataPartitionReader dataPartitionReader =
new DataPartitionReader(startSubIndex, endSubIndex, fileInfo,
streamId);
- dataPartitionReader.open();
+ dataPartitionReader.open(dataFileChanel, indexChannel);
// allocate resources when the first reader is registered
boolean allocateResources = readers.isEmpty();
readers.add(dataPartitionReader);
@@ -305,6 +314,11 @@ public class BufferStreamManager {
public synchronized void removeStream(long streamId) {
activeStreamIds.remove(streamId);
}
+
+ public void close() {
+ IOUtils.closeQuietly(dataFileChanel);
+ IOUtils.closeQuietly(indexChannel);
+ }
}
private class Buffer {
@@ -324,8 +338,6 @@ public class BufferStreamManager {
private final int startPartitionIndex;
private final int endPartitionIndex;
private int numRegions;
- private FileChannel dataFileChannel;
- private FileChannel indexFileChannel;
private int numRemainingPartitions;
private int currentDataRegion = -1;
private long dataConsumingOffset;
@@ -355,6 +367,9 @@ public class BufferStreamManager {
@GuardedBy("lock")
protected boolean isError;
+ private FileChannel dataFileChannel;
+ private FileChannel indexFileChannel;
+
public DataPartitionReader(
int startPartitionIndex, int endPartitionIndex, FileInfo fileInfo,
long streamId) {
this.startPartitionIndex = startPartitionIndex;
@@ -370,10 +385,9 @@ public class BufferStreamManager {
this.isClosed = false;
}
- public void open() throws IOException {
- this.dataFileChannel = new
FileInputStream(fileInfo.getFile()).getChannel();
- this.indexFileChannel = new
FileInputStream(fileInfo.getIndexPath()).getChannel();
-
+ public void open(FileChannel dataFileChannel, FileChannel
indexFileChannel) throws IOException {
+ this.dataFileChannel = dataFileChannel;
+ this.indexFileChannel = indexFileChannel;
long indexFileSize = indexFileChannel.size();
// index is (offset,length)
long indexRegionSize = fileInfo.getNumReducerPartitions() * (long)
INDEX_ENTRY_SIZE;
@@ -518,8 +532,8 @@ public class BufferStreamManager {
public synchronized boolean readAndSend(Queue<ByteBuf> bufferQueue,
Consumer<ByteBuf> consumer)
throws IOException {
- boolean hasReaming = hasRemaining();
- boolean continueReading = hasReaming;
+ boolean hasRemaing = hasRemaining();
+ boolean continueReading = hasRemaing;
int numDataBuffers = 0;
while (continueReading) {
@@ -538,19 +552,19 @@ public class BufferStreamManager {
throw throwable;
}
- hasReaming = hasRemaining();
- addBuffer(buffer, hasReaming, consumer);
+ hasRemaing = hasRemaining();
+ addBuffer(buffer, hasRemaing, consumer);
++numDataBuffers;
}
if (numDataBuffers > 0) {
notifyBacklog(numDataBuffers);
}
- if (!hasReaming) {
+ if (!hasRemaing) {
closeReader();
}
- return hasReaming;
+ return hasRemaing;
}
protected void addBuffer(ByteBuf buffer, boolean hasRemaining,
Consumer<ByteBuf> consumer) {
@@ -638,8 +652,6 @@ public class BufferStreamManager {
public void closeReader() {
isClosed = true;
- IOUtils.closeQuietly(indexFileChannel);
- IOUtils.closeQuietly(dataFileChannel);
logger.debug("Closed read for stream {}", this.streamId);
}
