[GitHub] [flink] StephanEwen commented on a change in pull request #13523: [FLINK-15981][network] Implement FileRegion way to shuffle file-based blocking partition in network stack

[GitBox <gi...@apache.org>]

StephanEwen commented on a change in pull request #13523:
URL: https://github.com/apache/flink/pull/13523#discussion_r512269411



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PartitionData.java
##########
@@ -45,6 +51,14 @@
 
 	public abstract boolean isBuffer();
 
+	/**
+	 * Returns the buffer-format partition data with the provided memory segment or not.
+	 *
+	 * @param segment it might be needed to read the partition data into.
+	 * @return the buffer represents the partition data.
+	 */
+	public abstract Buffer getBuffer(@Nullable MemorySegment segment) throws IOException;

Review comment:
       I would change this to `Supplier<MemorySegment>` so that the code lazily fetches (and allocates) the memory segment only when needed. See also next comment about lazy creation.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferReaderWriterUtil.java
##########
@@ -181,7 +181,7 @@ private static boolean tryReadByteBuffer(FileChannel channel, ByteBuffer b) thro
 		}
 	}
 
-	private static void readByteBufferFully(FileChannel channel, ByteBuffer b) throws IOException {
+	public static void readByteBufferFully(FileChannel channel, ByteBuffer b) throws IOException {

Review comment:
       The remaining class uses package-private visibility, would suggest to use that here as well for consistency.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -238,6 +248,12 @@ public void setup() throws IOException {
 		checkState(this.bufferPool == null, "Bug in input gate setup logic: Already registered buffer pool.");
 		setupChannels();
 
+		//TODO we can further judge the condition whether the type is file-based or mmap-based
+		if (consumedPartitionType.isBlocking()) {

Review comment:
       I would simply make this a simple lazy-created MemorySegement, without any logic trying to determine what the shuffle type is. If this is fetched only if needed (see using `Supplier<MemorySegment>` above) then that will be good enough.
   
   I think it would also be fine if this was an unpooled segment, not obtained from the `memorySegmentProvider` but just created on first access. Would make things simpler, you don't have to worry about recycling, races between creation and early disposal, etc.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org