svn commit: r1554344 - /hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/AsyncDispatcher.java

[vinodkv]

Author: vinodkv
Date: Tue Dec 31 01:10:01 2013
New Revision: 1554344

URL: http://svn.apache.org/r1554344
Log:
YARN-1121. Addendum patch. Fixed AsyncDispatcher hang issue during stop due to 
a race condition caused by the previous patch. Contributed by Jian He. 

Modified:
    
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/AsyncDispatcher.java

Modified: 
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/AsyncDispatcher.java
URL: 
http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/AsyncDispatcher.java?rev=1554344&r1=1554343&r2=1554344&view=diff
==============================================================================
--- 
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/AsyncDispatcher.java
 (original)
+++ 
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/event/AsyncDispatcher.java
 Tue Dec 31 01:10:01 2013
@@ -56,6 +56,7 @@ public class AsyncDispatcher extends Abs
   // Indicates all the remaining dispatcher's events on stop have been drained
   // and processed.
   private volatile boolean drained = true;
+  private Object waitForDrained = new Object();
 
   // For drainEventsOnStop enabled only, block newly coming events into the
   // queue while stopping.
@@ -82,6 +83,16 @@ public class AsyncDispatcher extends Abs
       public void run() {
         while (!stopped && !Thread.currentThread().isInterrupted()) {
           drained = eventQueue.isEmpty();
+          // blockNewEvents is only set when dispatcher is draining to stop,
+          // adding this check is to avoid the overhead of acquiring the lock
+          // and calling notify every time in the normal run of the loop.
+          if (blockNewEvents) {
+            synchronized (waitForDrained) {
+              if (drained) {
+                waitForDrained.notify();
+              }
+            }
+          }
           Event event;
           try {
             event = eventQueue.take();
@@ -125,8 +136,11 @@ public class AsyncDispatcher extends Abs
     if (drainEventsOnStop) {
       blockNewEvents = true;
       LOG.info("AsyncDispatcher is draining to stop, igonring any new 
events.");
-      while(!drained) {
-        Thread.yield();
+      synchronized (waitForDrained) {
+        while (!drained && eventHandlingThread.isAlive()) {
+          waitForDrained.wait(1000);
+          LOG.info("Waiting for AsyncDispatcher to drain.");
+        }
       }
     }
     stopped = true;