fixed: multiplexing deadlock

VonRosenchild · Jul 19, 2013 · 4542555 · 4542555
1 parent 0540148
commit 4542555
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 8 deletions.
diff --git a/...e/nephele-server/src/main/java/eu/stratosphere/nephele/executiongraph/ExecutionGraph.java b/...e/nephele-server/src/main/java/eu/stratosphere/nephele/executiongraph/ExecutionGraph.java
@@ -1402,8 +1402,11 @@ private void calculateConnectionIDs() {
  for (int i = 0; i < lastStage.getNumberOfStageMembers(); ++i) {
 
  final ExecutionGroupVertex groupVertex = lastStage.getStageMember(i);
+
+ int currentConnectionID = 0;
+
  if (groupVertex.isOutputVertex()) {
-  groupVertex.calculateConnectionID(0, alreadyVisited);
+  currentConnectionID = groupVertex.calculateConnectionID(currentConnectionID, alreadyVisited);
  }
  }
  }

diff --git a/...ele-server/src/main/java/eu/stratosphere/nephele/executiongraph/ExecutionGroupVertex.java b/...ele-server/src/main/java/eu/stratosphere/nephele/executiongraph/ExecutionGroupVertex.java
@@ -927,19 +927,25 @@ public Iterator<ExecutionVertex> iterator() {
  * the current connection ID
  * @param alreadyVisited
  * the set of already visited group vertices
+ * @return maximum assigned connectionID
  */
- void calculateConnectionID(final int currentConnectionID, final Set<ExecutionGroupVertex> alreadyVisited) {
+ int calculateConnectionID(int currentConnectionID, final Set<ExecutionGroupVertex> alreadyVisited) {
 
  if (!alreadyVisited.add(this)) {
- return;
+ return currentConnectionID;
  }
-
- int nextConnectionID = currentConnectionID;
+
  for (final ExecutionGroupEdge backwardLink : this.backwardLinks) {
- backwardLink.setConnectionID(nextConnectionID);
- backwardLink.getSourceVertex().calculateConnectionID(nextConnectionID, alreadyVisited);
- ++nextConnectionID;
+
+ backwardLink.setConnectionID(currentConnectionID);
+
+ ++currentConnectionID;
+
+ currentConnectionID = backwardLink.getSourceVertex()
+ .calculateConnectionID(currentConnectionID, alreadyVisited);
  }
+
+ return currentConnectionID;
  }
 
  /**