[08/10] drill git commit: DRILL-2547: Don't allow Drill to shut down while queries are still executing This will cause Drillbit.close() to block until all currently executing fragments have completed.

[ja...@apache.org]

DRILL-2547: Don't allow Drill to shut down while queries are still executing
This will cause Drillbit.close() to block until all currently executing
fragments have completed.

WorkManager
- added waitForExit() and indicateIfSafeToExit(), which use a latch to
  wait to shut down if there are active fragments
  - waitForExit() times out after 5 seconds

Drillbit
- call WorkManager.waitForExit() in close()


Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/26463d38
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/26463d38
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/26463d38

Branch: refs/heads/0.8.0
Commit: 26463d38f356c138ba42b78144842950c33e1cef
Parents: f89c59b
Author: Chris Westin <cw...@yahoo.com>
Authored: Tue Mar 24 16:21:33 2015 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Wed Mar 25 21:11:23 2015 -0700

----------------------------------------------------------------------
 .../org/apache/drill/exec/server/Drillbit.java  |  3 ++
 .../org/apache/drill/exec/work/WorkManager.java | 47 +++++++++++++++++++-
 2 files changed, 49 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/drill/blob/26463d38/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
index 0f531b8..958f2dc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
@@ -252,6 +252,9 @@ public class Drillbit implements AutoCloseable {
       return;
     }
 
+    // wait for anything that is running to complete
+    manager.waitToExit();
+
     if (coord != null && registrationHandle != null) {
       coord.unregister(registrationHandle);
     }

http://git-wip-us.apache.org/repos/asf/drill/blob/26463d38/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java
index 231e49a..e2bcec3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/WorkManager.java
@@ -21,6 +21,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -168,6 +169,46 @@ public class WorkManager implements AutoCloseable {
     return dContext;
   }
 
+  private CountDownLatch exitLatch = null; // used to wait to exit when things are still running
+
+  /**
+   * Waits until it is safe to exit. Blocks until all currently running fragments have completed.
+   *
+   * <p>This is intended to be used by {@link org.apache.drill.exec.server.Drillbit#close()}.</p>
+   */
+  public void waitToExit() {
+    synchronized(this) {
+      if (queries.isEmpty() && runningFragments.isEmpty()) {
+        return;
+      }
+
+      exitLatch = new CountDownLatch(1);
+    }
+
+    while(true) {
+      try {
+        exitLatch.await(5, TimeUnit.SECONDS);
+      } catch(InterruptedException e) {
+        // keep waiting
+      }
+      break;
+    }
+  }
+
+  /**
+   * If it is safe to exit, and the exitLatch is in use, signals it so that waitToExit() will
+   * unblock.
+   */
+  private void indicateIfSafeToExit() {
+    synchronized(this) {
+      if (exitLatch != null) {
+        if (queries.isEmpty() && runningFragments.isEmpty()) {
+          exitLatch.countDown();
+        }
+      }
+    }
+  }
+
   /**
    * Narrowed interface to WorkManager that is made available to tasks it is managing.
    */
@@ -196,8 +237,11 @@ public class WorkManager implements AutoCloseable {
       final QueryId queryId = foreman.getQueryId();
       final boolean wasRemoved = queries.remove(queryId, foreman);
       if (!wasRemoved) {
-        throw new IllegalStateException("Couldn't find retiring Foreman for query " + queryId);
+        logger.warn("Couldn't find retiring Foreman for query " + queryId);
+//        throw new IllegalStateException("Couldn't find retiring Foreman for query " + queryId);
       }
+
+      indicateIfSafeToExit();
     }
 
     public Foreman getForemanForQueryId(final QueryId queryId) {
@@ -219,6 +263,7 @@ public class WorkManager implements AutoCloseable {
         @Override
         protected void cleanup() {
           runningFragments.remove(fragmentHandle);
+          indicateIfSafeToExit();
         }
       });
     }