From 666f0b47816f55a3310cee7d5da0d42b8e354e5a Mon Sep 17 00:00:00 2001
From: Kohsuke Kawaguchi <kk@kohsuke.org>
Date: Tue, 22 Apr 2014 11:26:09 -0700
Subject: [PATCH] Fixing a bug.

pending was null as soon as task started running, so it was possible that another thread submits the task and have that executed while the first one was still running
---
 .../jenkins/util/AtmostOneTaskExecutor.java   | 34 ++++++++++++--
 .../util/AtmostOneTaskExecutorTest.groovy     | 47 ++++++++++---------
 2 files changed, 55 insertions(+), 26 deletions(-)

diff --git a/core/src/main/java/jenkins/util/AtmostOneTaskExecutor.java b/core/src/main/java/jenkins/util/AtmostOneTaskExecutor.java
index c3e9504f4c..9e6e82b33b 100644
--- a/core/src/main/java/jenkins/util/AtmostOneTaskExecutor.java
+++ b/core/src/main/java/jenkins/util/AtmostOneTaskExecutor.java
@@ -1,5 +1,6 @@
 package jenkins.util;
 
+import com.google.common.util.concurrent.SettableFuture;
 import hudson.remoting.AtmostOneThreadExecutor;
 
 import java.util.concurrent.Callable;
@@ -41,7 +42,9 @@ public class AtmostOneTaskExecutor<V> {
      * If a task is already submitted and pending execution, non-null.
      * Guarded by "synchronized(this)"
      */
-    private Future<V> pending;
+    private SettableFuture<V> pending;
+
+    private SettableFuture<V> inprogress;
 
     public AtmostOneTaskExecutor(ExecutorService base, Callable<V> task) {
         this.base = base;
@@ -57,18 +60,39 @@ public class AtmostOneTaskExecutor<V> {
             // if a task is already pending, just join that
             return pending;
 
-        pending = base.submit(new Callable<V>() {
+        pending = SettableFuture.create();
+        if (inprogress==null) {
+            // if the task isn't currently running, schedule one
+            run();
+        }
+        return pending;
+    }
+
+    private synchronized void run() {
+        base.submit(new Callable<Void>() {
             @Override
-            public V call() throws Exception {
+            public Void call() throws Exception {
                 // before we get going, everyone who submits after this
                 // should form a next batch
                 synchronized (AtmostOneTaskExecutor.this) {
+                    inprogress = pending;
                     pending = null;
                 }
 
-                return task.call();
+                try {
+                    inprogress.set(task.call());
+                } catch (Throwable t) {
+                    inprogress.setException(t);
+                } finally {
+                    synchronized (AtmostOneTaskExecutor.this) {
+                        // if next one is pending, get that scheduled
+                        inprogress = null;
+                        if (pending!=null)
+                            run();
+                    }
+                }
+                return null;
             }
         });
-        return pending;
     }
 }
diff --git a/core/src/test/groovy/jenkins/util/AtmostOneTaskExecutorTest.groovy b/core/src/test/groovy/jenkins/util/AtmostOneTaskExecutorTest.groovy
index 894eef7c4f..e39addae77 100644
--- a/core/src/test/groovy/jenkins/util/AtmostOneTaskExecutorTest.groovy
+++ b/core/src/test/groovy/jenkins/util/AtmostOneTaskExecutorTest.groovy
@@ -1,5 +1,6 @@
 package jenkins.util
 
+import hudson.util.OneShotEvent
 import org.junit.Test
 
 import java.util.concurrent.Callable
@@ -14,29 +15,33 @@ import java.util.concurrent.atomic.AtomicInteger
 class AtmostOneTaskExecutorTest {
     def counter = new AtomicInteger()
 
-    def lock = new Object()
+    def lock = new OneShotEvent()
 
     @Test
     public void doubleBooking() {
-        synchronized (lock) {
-            def base = Executors.newCachedThreadPool()
-            def es = new AtmostOneTaskExecutor(base,
-                    { ->
-                        counter.incrementAndGet()
-                        synchronized (lock) {
-                            lock.wait()
-                        }
-                    } as Callable);
-            es.submit()
-            while (counter.get()==0)
-                ;   // spin lock until executor gets to the choking point
-
-            def f = es.submit() // this should hang
-            Thread.sleep(500)   // make sure the 2nd task is hanging
-            assert counter.get()==1
-            assert !f.isDone()
-
-            notifyAll() // let the first one go
-        }
+        def f1,f2;
+
+        def base = Executors.newCachedThreadPool()
+        def es = new AtmostOneTaskExecutor(base,
+                { ->
+                    counter.incrementAndGet()
+                    lock.block()
+                } as Callable);
+        f1 = es.submit()
+        while (counter.get() == 0)
+        ;   // spin lock until executor gets to the choking point
+
+        f2 = es.submit() // this should hang
+        Thread.sleep(500)   // make sure the 2nd task is hanging
+        assert counter.get() == 1
+        assert !f2.isDone()
+
+        lock.signal() // let the first one go
+
+        f1.get();   // first one should complete
+
+        // now 2nd one gets going and hits the choke point
+        f2.get()
+        assert counter.get()==2
     }
 }
-- 
GitLab