ThreadPools: schedule a timeout check after adding command to queue

Our thread pools have support for timeout on a task. To support this, a special background task is schedule to run at timeout. That background task fires and check if the main task is still in the executor queue and then cancels it if needed. Currently we schedule this background task before adding the main task to the queue. If the timeout is very small (in tests we often use numbers like 2 ms) the background task can fire before the main one is added to the queue causing the timeout to be missed.

See http://build-us-00.elastic.co/job/es_g1gc_master_metal/11780/testReport/junit/org.elasticsearch.cluster/ClusterServiceTests/testTimeoutUpdateTask/

Closes elastic#12319

src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java

@@ -84,6 +84,7 @@ public void execute(Runnable command, final ScheduledExecutorService timer, fina
         } else if (!(command instanceof PrioritizedFutureTask)) { // it might be a callable wrapper...
             command = new TieBreakingPrioritizedRunnable(command, Priority.NORMAL, insertionOrder.incrementAndGet());
         }
+        super.execute(command);
         if (timeout.nanos() >= 0) {
             if (command instanceof TieBreakingPrioritizedRunnable) {
                 ((TieBreakingPrioritizedRunnable) command).scheduleTimeout(timer, timeoutCallback, timeout);
@@ -93,7 +94,6 @@ public void execute(Runnable command, final ScheduledExecutorService timer, fina
                 throw new UnsupportedOperationException("Execute with timeout is not supported for future tasks");
             }
         }
-        super.execute(command);
     }
 
     @Override
@@ -140,7 +140,8 @@ private final class TieBreakingPrioritizedRunnable extends PrioritizedRunnable {
 
         private Runnable runnable;
         private final long insertionOrder;
-        private ScheduledFuture<?> timeoutFuture;
+        private volatile ScheduledFuture<?> timeoutFuture;
+        private volatile boolean started = false;
 
         TieBreakingPrioritizedRunnable(PrioritizedRunnable runnable, long insertionOrder) {
             this(runnable, runnable.priority(), insertionOrder);
@@ -154,6 +155,9 @@ private final class TieBreakingPrioritizedRunnable extends PrioritizedRunnable {
 
         @Override
         public void run() {
+            // make the task as stared. This is needed for synchronization with the timeout handling
+            // see  #scheduleTimeout()
+            started = true;
             FutureUtils.cancel(timeoutFuture);
             runAndClean(runnable);
         }
@@ -176,6 +180,10 @@ public void run() {
                     }
                 }
             }, timeValue.nanos(), TimeUnit.NANOSECONDS);
+            if (started) {
+                // if the actual action already it might have missed the setting of the future. Clean it ourselves.
+                FutureUtils.cancel(timeoutFuture);
+            }
         }
 
         /**

src/test/java/org/elasticsearch/common/util/concurrent/PrioritizedExecutorsTests.java

@@ -253,7 +253,7 @@ public void testTimeoutCleanup() throws Exception {
                              public void run() {
                                  invoked.countDown();
                              }
-                         }, timer, TimeValue.timeValueMillis(1000), new Runnable() {
+                         }, timer, TimeValue.timeValueHours(1), new Runnable() {
                     @Override
                     public void run() {
                         // We should never get here