WorkQueue: Fix parked-task core-pool size races

When a task is parked, WorkQueue temporarily increments corePoolSize so
a replacement worker can run. The parked slot must be decremented
exactly once when that parked task is either unparked or cancelled.
This change ensures that when a completing thread unparks or cleans up a
ready or canceled task in updateParked(), it will not proceed to
potentially start working on a queued task until the thread pool has
been decremented, thus avoiding potentially temporarily running more
tasks than allowed concurrently.

Add TaskParkerIT coverage for interrupting a task while it is being
unparked, plus a small helper to interrupt task threads by name.

Change-Id: I3f9bdb42b34f1ec79763f93b23b54812fe77f7fe
Release-Notes: Avoid running more WorkQueue tasks than allowed

java/com/google/gerrit/server/git/WorkQueue.java

diff --git a/java/com/google/gerrit/server/git/WorkQueue.java b/java/com/google/gerrit/server/git/WorkQueue.java
index 95d6d5f..49d0be0 100644
--- a/java/com/google/gerrit/server/git/WorkQueue.java
+++ b/java/com/google/gerrit/server/git/WorkQueue.java
@@ -60,6 +60,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -347,6 +348,7 @@
       private final CancellableCountDownLatch latch = new CancellableCountDownLatch(1);
       private final Task<?> task;
       private final Long priority = priorityGenerator.getAndIncrement();
+      private final AtomicBoolean isParked = new AtomicBoolean(true);
 
       public ParkedTask(Task<?> task) {
         this.task = task;
@@ -366,6 +368,7 @@
        * method.
        */
       public void cancel() {
+        close();
         latch.cancel();
       }
 
@@ -383,12 +386,15 @@
       }
 
       public void unpark() {
+        close();
         latch.countDown();
       }
 
       @Override
       public void close() {
-        incrementCorePoolSizeBy(-1);
+        if (isParked.compareAndSet(true, false)) {
+          incrementCorePoolSizeBy(-1);
+        }
       }
     }
 

javatests/com/google/gerrit/acceptance/server/util/TaskParkerIT.java

diff --git a/javatests/com/google/gerrit/acceptance/server/util/TaskParkerIT.java b/javatests/com/google/gerrit/acceptance/server/util/TaskParkerIT.java
index 3b82ebe..df63b69 100644
--- a/javatests/com/google/gerrit/acceptance/server/util/TaskParkerIT.java
+++ b/javatests/com/google/gerrit/acceptance/server/util/TaskParkerIT.java
@@ -463,12 +463,34 @@
     assertStateIsEventually(forwarder.task, State.PARKED);
 
     // interrupt the thread with parked task
-    for (Thread t : Thread.getAllStackTraces().keySet()) {
-      if (t.getName().contains(taskName)) {
-        t.interrupt();
-        break;
-      }
-    }
+    interruptThreadContaining(taskName);
+
+    assertCorePoolSizeIsEventually(1);
+  }
+
+  @Test
+  public void interruptingTaskWhileUnparkingDoesNotDoubleDecrementCorePoolSize()
+      throws InterruptedException {
+    String taskName = "to-be-unparked";
+    LatchedRunnable parkedRunnable = new LatchedRunnable(taskName);
+    LatchedRunnable blockerRunnable = new LatchedRunnable("blocker");
+    assertCorePoolSizeIs(1);
+
+    // park parkedRunnable
+    executor.execute(parkedRunnable);
+    parker.isReadyToStart.assertCalledEventuallyThenComplete(false);
+    assertCorePoolSizeIsEventually(2);
+    assertStateIsEventually(forwarder.task, State.PARKED);
+
+    // start blockerRunnable and unblock it to trigger updateParked() from onStop()
+    executor.execute(blockerRunnable);
+    blockerRunnable.run.assertCalledEventuallyThenComplete(null);
+
+    // Wait for updateParked() to poll parkedRunnable and call isReadyToStart(), then interrupt
+    // the parked thread before releasing isReadyToStart() as ready.
+    parker.isReadyToStart.assertCalledEventually();
+    interruptThreadContaining(taskName);
+    parker.isReadyToStart.complete(true);
 
     assertCorePoolSizeIsEventually(1);
   }
@@ -502,6 +524,15 @@
     TaskListenerIT.assertTaskCountIsEventually(workQueue, count);
   }
 
+  private void interruptThreadContaining(String taskName) {
+    for (Thread t : Thread.getAllStackTraces().keySet()) {
+      if (t.getName().contains(taskName)) {
+        t.interrupt();
+        break;
+      }
+    }
+  }
+
   private void assertCorePoolSizeIs(int count) {
     assertThat(count).isEqualTo(((ScheduledThreadPoolExecutor) executor).getCorePoolSize());
   }