svn commit: r1499971 - in /hama/trunk: CHANGES.txt core/src/main/java/org/apache/hama/bsp/BSPPeerImpl.java core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java

[kennethxian]

Author: kennethxian
Date: Fri Jul  5 11:35:34 2013
New Revision: 1499971

URL: http://svn.apache.org/r1499971
Log:
HAMA-769: Intermediate queue's close method is not called, clean work may be 
omitted.

Modified:
    hama/trunk/CHANGES.txt
    hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPPeerImpl.java
    
hama/trunk/core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java

Modified: hama/trunk/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/hama/trunk/CHANGES.txt?rev=1499971&r1=1499970&r2=1499971&view=diff
==============================================================================
--- hama/trunk/CHANGES.txt (original)
+++ hama/trunk/CHANGES.txt Fri Jul  5 11:35:34 2013
@@ -6,6 +6,7 @@ Release 0.6.3 (unreleased changes)
 
   BUG FIXES
 
+  HAMA-769: Intermediate queue's close method is not called, clean work may be 
omitted (kennethxian)
   HAMA-771: Determining the count of active vertices (edwardyoon)
 
   IMPROVEMENTS

Modified: hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPPeerImpl.java
URL: 
http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPPeerImpl.java?rev=1499971&r1=1499970&r2=1499971&view=diff
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPPeerImpl.java 
(original)
+++ hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPPeerImpl.java Fri Jul  
5 11:35:34 2013
@@ -385,6 +385,10 @@ public final class BSPPeerImpl<K1, V1, K
       } catch (Exception e) {
         LOG.error("Error while sending messages", e);
       }
+      MessageQueue<M> msgQueue = (MessageQueue<M>) messages;
+      if (msgQueue != null) {
+        msgQueue.close();
+      }
     }
 
     if (this.faultToleranceService != null) {

Modified: 
hama/trunk/core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java
URL: 
http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java?rev=1499971&r1=1499970&r2=1499971&view=diff
==============================================================================
--- 
hama/trunk/core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java
 (original)
+++ 
hama/trunk/core/src/main/java/org/apache/hama/bsp/message/AbstractMessageManager.java
 Fri Jul  5 11:35:34 2013
@@ -157,6 +157,9 @@ public abstract class AbstractMessageMan
    */
   @Override
   public final void clearOutgoingQueues() {
+    if (localQueue != null) {
+      localQueue.close();
+    }
     localQueue = localQueueForNextIteration.getMessageQueue();
     localQueue.prepareRead();
     localQueueForNextIteration = getSynchronizedReceiverQueue();