Fix a signaling bug in newDirectExecutorService() Also switch to impl…

…icit

monitors to take advantage of biased locking.

If thread A calls awaitTermination while there are no running tasks, then
Thread B calls shutdown(). Thread A will never be woken up.
-------------
Created by MOE: http://code.google.com/p/moe-java
MOE_MIGRATED_REVID=87280837

guava-tests/test/com/google/common/util/concurrent/MoreExecutorsTest.java

@@ -53,6 +53,7 @@
 import org.mockito.InOrder;
 import org.mockito.Mockito;
 
+import java.lang.Thread.State;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -243,6 +244,51 @@ public Void call() throws Exception {
         throwableFromOtherThread.get());
   }
 
+  /**
+   * Test for a bug where threads weren't getting signaled when shutdown was called, only when
+   * tasks completed.
+   */
+
+  public void testDirectExecutorService_awaitTermination_missedSignal() {
+    final ExecutorService service = MoreExecutors.newDirectExecutorService();
+    Thread waiter = new Thread() {
+      @Override public void run() {
+        try {
+          service.awaitTermination(1, TimeUnit.DAYS);
+        } catch (InterruptedException e) {
+          return;
+        }
+      }
+    };
+    waiter.start();
+    awaitTimedWaiting(waiter);
+    service.shutdown();
+    Uninterruptibles.joinUninterruptibly(waiter, 10, TimeUnit.SECONDS);
+    if (waiter.isAlive()) {
+      waiter.interrupt();
+      fail("awaitTermination failed to trigger after shutdown()");
+    }
+  }
+
+  /** Wait for the given thread to reach the {@link State#TIMED_WAITING} thread state. */
+  void awaitTimedWaiting(Thread thread) {
+    while (true) {
+      switch (thread.getState()) {
+        case BLOCKED:
+        case NEW:
+        case RUNNABLE:
+        case WAITING:
+          Thread.yield();
+          break;
+        case TIMED_WAITING:
+          return;
+        case TERMINATED:
+        default:
+          throw new AssertionError();
+      }
+    }
+  }
+
   public void testDirectExecutorService_shutdownNow() {
     ExecutorService executor = newDirectExecutorService();
     assertEquals(ImmutableList.of(), executor.shutdownNow());

guava/src/com/google/common/util/concurrent/MoreExecutors.java

@@ -51,9 +51,8 @@
 import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
+
+import javax.annotation.concurrent.GuardedBy;
 
 /**
  * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link
@@ -290,12 +289,9 @@ private static class DirectExecutorService
       extends AbstractListeningExecutorService {
     /**
      * Lock used whenever accessing the state variables
-     * (runningTasks, shutdown, terminationCondition) of the executor
+     * (runningTasks, shutdown) of the executor
      */
-    private final Lock lock = new ReentrantLock();
-
-    /** Signaled after the executor is shutdown and running tasks are done */
-    private final Condition termination = lock.newCondition();
+    private final Object lock = new Object();
 
     /*
      * Conceptually, these two variables describe the executor being in
@@ -304,8 +300,8 @@ private static class DirectExecutorService
      *   - Shutdown: runningTasks > 0 and shutdown == true
      *   - Terminated: runningTasks == 0 and shutdown == true
      */
-    private int runningTasks = 0;
-    private boolean shutdown = false;
+    @GuardedBy("lock") private int runningTasks = 0;
+    @GuardedBy("lock") private boolean shutdown = false;
 
     @Override
     public void execute(Runnable command) {
@@ -319,21 +315,18 @@ public void execute(Runnable command) {
 
     @Override
     public boolean isShutdown() {
-      lock.lock();
-      try {
+      synchronized (lock) {
         return shutdown;
-      } finally {
-        lock.unlock();
       }
     }
 
     @Override
     public void shutdown() {
-      lock.lock();
-      try {
+      synchronized (lock) {
         shutdown = true;
-      } finally {
-        lock.unlock();
+        if (runningTasks == 0) {
+          lock.notifyAll();
+        }
       }
     }
 
@@ -346,31 +339,27 @@ public List<Runnable> shutdownNow() {
 
     @Override
     public boolean isTerminated() {
-      lock.lock();
-      try {
+      synchronized (lock) {
         return shutdown && runningTasks == 0;
-      } finally {
-        lock.unlock();
       }
     }
 
     @Override
     public boolean awaitTermination(long timeout, TimeUnit unit)
         throws InterruptedException {
       long nanos = unit.toNanos(timeout);
-      lock.lock();
-      try {
+      synchronized (lock) {
         for (;;) {
-          if (isTerminated()) {
+          if (shutdown && runningTasks == 0) {
             return true;
           } else if (nanos <= 0) {
             return false;
           } else {
-            nanos = termination.awaitNanos(nanos);
+            long now = System.nanoTime();
+            TimeUnit.NANOSECONDS.timedWait(lock, nanos);
+            nanos -= System.nanoTime() - now;  // subtract the actual time we waited
           }
         }
-      } finally {
-        lock.unlock();
       }
     }
 
@@ -382,29 +371,23 @@ public boolean awaitTermination(long timeout, TimeUnit unit)
      *         shutdown
      */
     private void startTask() {
-      lock.lock();
-      try {
-        if (isShutdown()) {
+      synchronized (lock) {
+        if (shutdown) {
           throw new RejectedExecutionException("Executor already shutdown");
         }
         runningTasks++;
-      } finally {
-        lock.unlock();
       }
     }
 
     /**
      * Decrements the running task count.
      */
     private void endTask() {
-      lock.lock();
-      try {
-        runningTasks--;
-        if (isTerminated()) {
-          termination.signalAll();
+      synchronized (lock) {
+        int numRunning = --runningTasks;
+        if (numRunning == 0) {
+          lock.notifyAll();
         }
-      } finally {
-        lock.unlock();
       }
     }
   }