[flink] 02/02: [FLINK-12843][network] Refactor the pin logic in ReleaseOnConsumptionResultPartition

[ch...@apache.org]

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

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

commit 0afd7bc3e789dae3fcbb20afecb8edc2ebccbc57
Author: Zhijiang <wa...@aliyun.com>
AuthorDate: Mon Jun 17 17:55:25 2019 +0800

    [FLINK-12843][network] Refactor the pin logic in ReleaseOnConsumptionResultPartition
    
    The pin logic is for adding the reference counter based on number of subpartitions for ReleaseOnConsumptionResultPartition. It seems
    not necessary to do it in while loop as now, because the atomic counter would not be accessed by other threads during pin. If the
    ReleaseOnConsumptionResultPartition was not created yet, the createSubpartitionView would not be called actually resulting in
    PartitionNotFoundException. So we could simple increase the reference counter in ReleaseOnConsumptionResultPartition constructor directly.
---
 .../ReleaseOnConsumptionResultPartition.java          | 19 +------------------
 .../runtime/io/network/partition/ResultPartition.java |  6 ------
 .../io/network/partition/ResultPartitionFactory.java  |  3 ---
 3 files changed, 1 insertion(+), 27 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
index 5f2d390..d056271 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ReleaseOnConsumptionResultPartition.java
@@ -47,25 +47,8 @@ public class ReleaseOnConsumptionResultPartition extends ResultPartition {
 			ResultPartitionManager partitionManager,
 			FunctionWithException<BufferPoolOwner, BufferPool, IOException> bufferPoolFactory) {
 		super(owningTaskName, partitionId, partitionType, subpartitions, numTargetKeyGroups, partitionManager, bufferPoolFactory);
-	}
 
-	/**
-	 * The partition can only be released after each subpartition has been consumed once per pin operation.
-	 */
-	@Override
-	void pin() {
-		while (true) {
-			int refCnt = pendingReferences.get();
-
-			if (refCnt >= 0) {
-				if (pendingReferences.compareAndSet(refCnt, refCnt + subpartitions.length)) {
-					break;
-				}
-			}
-			else {
-				throw new IllegalStateException("Released.");
-			}
-		}
+		pendingReferences.set(subpartitions.length);
 	}
 
 	@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
index a303421..d0cf25d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
@@ -330,12 +330,6 @@ public class ResultPartition implements ResultPartitionWriter, BufferPoolOwner {
 	// ------------------------------------------------------------------------
 
 	/**
-	 * Pins the result partition.
-	 */
-	void pin() {
-	}
-
-	/**
 	 * Notification when a subpartition is released.
 	 */
 	void onConsumedSubpartition(int subpartitionIndex) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
index 67ad2c5..2eb620e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
@@ -122,9 +122,6 @@ public class ResultPartitionFactory {
 
 		createSubpartitions(partition, type, subpartitions);
 
-		// Initially, partitions should be consumed once before release.
-		partition.pin();
-
 		LOG.debug("{}: Initialized {}", taskNameWithSubtaskAndId, this);
 
 		return partition;