[GitHub] [flink] zhuzhurk commented on a change in pull request #15259: [FLINK-20757][network] Optimize data broadcast for sort-merge blocking shuffle

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

zhuzhurk commented on a change in pull request #15259:
URL: https://github.com/apache/flink/pull/15259#discussion_r603185899



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java
##########
@@ -227,84 +233,102 @@ public void broadcastEvent(AbstractEvent event, boolean isPriorityEvent) throws
     }
 
     private void broadcast(ByteBuffer record, DataType dataType) throws IOException {
-        for (int channelIndex = 0; channelIndex < numSubpartitions; ++channelIndex) {
-            record.rewind();
-            emit(record, channelIndex, dataType);
-        }
+        emit(record, 0, dataType, true);
     }
 
-    private void emit(ByteBuffer record, int targetSubpartition, DataType dataType)
+    private void emit(
+            ByteBuffer record, int targetSubpartition, DataType dataType, boolean isBroadcast)
             throws IOException {
         checkInProduceState();
 
-        SortBuffer sortBuffer = getSortBuffer();
+        SortBuffer sortBuffer = isBroadcast ? getBroadcastSortBuffer() : getUnicastSortBuffer();
         if (sortBuffer.append(record, targetSubpartition, dataType)) {
             return;
         }
 
         if (!sortBuffer.hasRemaining()) {
             // the record can not be appended to the free sort buffer because it is too large
-            currentSortBuffer.finish();
-            currentSortBuffer.release();
-            writeLargeRecord(record, targetSubpartition, dataType);
+            sortBuffer.finish();
+            sortBuffer.release();
+            writeLargeRecord(record, targetSubpartition, dataType, isBroadcast);
             return;
         }
 
-        flushCurrentSortBuffer();
-        emit(record, targetSubpartition, dataType);
+        flushSortBuffer(sortBuffer, isBroadcast);
+        emit(record, targetSubpartition, dataType, isBroadcast);
     }
 
-    private void releaseCurrentSortBuffer() {
-        if (currentSortBuffer != null) {
-            currentSortBuffer.release();
+    private void releaseSortBuffer(SortBuffer sortBuffer) {
+        if (sortBuffer != null) {
+            sortBuffer.release();
         }
     }
 
-    private SortBuffer getSortBuffer() {
-        if (currentSortBuffer != null && !currentSortBuffer.isFinished()) {
-            return currentSortBuffer;
+    private SortBuffer getUnicastSortBuffer() throws IOException {
+        flushSortBuffer(broadcastSortBuffer, true);

Review comment:
       How about introducing a `flushBroadcastSortBuffer()` and a `flushUnicastSortBuffer()`?
   Think can help to avoid unexpected inconsistent invocations.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionTest.java
##########
@@ -246,6 +257,36 @@ public void testWriteLargeRecord() throws Exception {
         assertEquals(recordWritten, recordRead);
     }
 
+    @Test
+    public void testDataBroadcast() throws Exception {
+        int numSubpartitions = 10;
+        int numBuffers = 100;
+        int numRecords = 10000;
+
+        BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
+        SortMergeResultPartition partition =
+                createSortMergedPartition(numSubpartitions, bufferPool);
+
+        for (int i = 0; i < numRecords; ++i) {
+            ByteBuffer record = generateRandomData(bufferSize, new Random());
+            partition.broadcastRecord(record);
+        }
+        partition.finish();
+        partition.close();
+
+        int eventSize = EventSerializer.toSerializedEvent(EndOfPartitionEvent.INSTANCE).remaining();
+        long dataSize = numSubpartitions * numRecords * bufferSize + numSubpartitions * eventSize;
+        assertNotNull(partition.getResultFile());
+        assertEquals(2, checkNotNull(fileChannelManager.getPaths()[0].list()).length);
+        for (File file : checkNotNull(fileChannelManager.getPaths()[0].listFiles())) {
+            assertTrue(file.length() < numSubpartitions * numRecords * bufferSize);

Review comment:
       Is this check for compression? Should we just check the data file?




-- 
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