flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [7/7] flink git commit: [FLINK-2826] [runtime] Fix race condition and locking in BroadcastVariableMaterialization#decrementReferenceInternal
Date Wed, 11 Nov 2015 15:36:57 GMT
[FLINK-2826] [runtime] Fix race condition and locking in BroadcastVariableMaterialization#decrementReferenceInternal

This closes #1339


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6019f085
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6019f085
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6019f085

Branch: refs/heads/master
Commit: 6019f085f0d3dadc20a11f900784ce2796a89ce0
Parents: 976bacc
Author: tedyu <yuzhihong@gmail.com>
Authored: Sun Nov 8 07:43:03 2015 -0800
Committer: Stephan Ewen <sewen@apache.org>
Committed: Wed Nov 11 14:47:45 2015 +0100

----------------------------------------------------------------------
 .../runtime/broadcast/BroadcastVariableMaterialization.java    | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/6019f085/flink-runtime/src/main/java/org/apache/flink/runtime/broadcast/BroadcastVariableMaterialization.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/broadcast/BroadcastVariableMaterialization.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/broadcast/BroadcastVariableMaterialization.java
index 86e0111..eb7e311 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/broadcast/BroadcastVariableMaterialization.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/broadcast/BroadcastVariableMaterialization.java
@@ -137,7 +137,7 @@ public class BroadcastVariableMaterialization<T, C> {
 				while ((element = readerIterator.next(element)) != null);
 				
 				synchronized (materializationMonitor) {
-					while (!this.materialized) {
+					while (!this.materialized && !disposed) {
 						materializationMonitor.wait();
 					}
 				}
@@ -209,7 +209,7 @@ public class BroadcastVariableMaterialization<T, C> {
 			throw new IllegalStateException("The Broadcast Variable has been disposed");
 		}
 		
-		synchronized (this) {
+		synchronized (references) {
 			if (transformed != null) {
 				if (transformed instanceof List) {
 					@SuppressWarnings("unchecked")
@@ -233,7 +233,7 @@ public class BroadcastVariableMaterialization<T, C> {
 			throw new IllegalStateException("The Broadcast Variable has been disposed");
 		}
 		
-		synchronized (this) {
+		synchronized (references) {
 			if (transformed == null) {
 				transformed = initializer.initializeBroadcastVariable(data);
 				data = null;


Mime
View raw message