[flink] branch master updated: [FLINK-17799][network] Fix performance regression in the benchmarks

[pn...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new a875235  [FLINK-17799][network] Fix performance regression in the benchmarks
a875235 is described below

commit a87523538992c5d3d9588ed161f14a04345c7662
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon May 18 15:59:08 2020 +0200

    [FLINK-17799][network] Fix performance regression in the benchmarks
    
    FLINK-16536 (re) introduced requestPartitions on critical path of AbstractRecordReader, which
    amounts to couple of viritual calls, one lock acquisition and one volatile read. This overhead
    is what's cuasing performance drop.
    
    We can avoid subsequentional redundant calls, by checking against a local variable if we have
    already requested partitions or not.
---
 .../flink/runtime/io/network/api/reader/AbstractRecordReader.java  | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
index 6975ac6..1c98d0c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
@@ -41,6 +41,8 @@ abstract class AbstractRecordReader<T extends IOReadableWritable> extends Abstra
 
 	private RecordDeserializer<T> currentRecordDeserializer;
 
+	private boolean requestedPartitions;
+
 	private boolean isFinished;
 
 	/**
@@ -66,7 +68,10 @@ abstract class AbstractRecordReader<T extends IOReadableWritable> extends Abstra
 		// The action of partition request was removed from InputGate#setup since FLINK-16536, and this is the only
 		// unified way for launching partition request for batch jobs. In order to avoid potential performance concern,
 		// we might consider migrating this action back to the setup based on some condition judgement future.
-		inputGate.requestPartitions();
+		if (!requestedPartitions) {
+			inputGate.requestPartitions();
+			requestedPartitions = true;
+		}
 
 		if (isFinished) {
 			return false;