Allow to queue async periodic work (#10716)

Author: krisstern

core/src/main/java/hudson/model/AsyncPeriodicWork.java

@@ -9,6 +9,7 @@
 import java.io.PrintStream;
 import java.util.Date;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.LogRecord;
@@ -73,6 +74,8 @@ public abstract class AsyncPeriodicWork extends PeriodicWork {
 
     private Thread thread;
 
+    private final AtomicBoolean pending = new AtomicBoolean(false);
+
     protected AsyncPeriodicWork(String name) {
         this.name = name;
         this.logRotateMillis = TimeUnit.MINUTES.toMillis(
@@ -88,7 +91,12 @@ protected AsyncPeriodicWork(String name) {
     public final void doRun() {
         try {
             if (thread != null && thread.isAlive()) {
-                logger.log(this.getSlowLoggingLevel(), "{0} thread is still running. Execution aborted.", name);
+                if (queueIfAlreadyRunning()) {
+                    logger.log(this.getSlowLoggingLevel(), "Scheduling another run of {0} since it is already running", name);
+                    pending.set(true);
+                } else {
+                    logger.log(this.getSlowLoggingLevel(), "{0} thread is still running. Execution aborted.", name);
+                }
                 return;
             }
             thread = new Thread(() -> {
@@ -112,6 +120,11 @@ public final void doRun() {
 
                 logger.log(Level.FINE, "Finished {0}. {1,number} ms",
                         new Object[]{name, stopTime - startTime});
+                thread = null;
+                if (pending.getAndSet(false)) {
+                    logger.log(this.getSlowLoggingLevel(), "An execution of {0} was requested while it was running, scheduling another run now", name);
+                    doRun();
+                }
             }, name + " thread");
             thread.start();
         } catch (Throwable t) {
@@ -262,4 +275,12 @@ protected Level getErrorLoggingLevel() {
      *      The caller will record the exception and moves on.
      */
     protected abstract void execute(TaskListener listener) throws IOException, InterruptedException;
+
+    /**
+     * @return true if a new run should be queued if it is already running while called.
+     * @since TODO
+     */
+    protected boolean queueIfAlreadyRunning() {
+        return false;
+    }
 }

test/src/test/java/hudson/model/AsyncPeriodicWorkTest.java

@@ -0,0 +1,80 @@
+package hudson.model;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.lessThan;
+
+import hudson.ExtensionList;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import org.junit.Rule;
+import org.junit.Test;
+import org.jvnet.hudson.test.JenkinsRule;
+import org.jvnet.hudson.test.TestExtension;
+
+public class AsyncPeriodicWorkTest {
+    @Rule
+    public JenkinsRule r = new JenkinsRule();
+
+    @Test
+    public void extraCallGetsIgnored() {
+        var instance = ExtensionList.lookupSingleton(AsyncPeriodicWorkTestImpl.class);
+        assertThat(instance.getCount(), is(0));
+        instance.run();
+        // Gets ignored
+        instance.run();
+        await().until(instance::getCount, is(1));
+        await("Should never reach 2 as the second call should be ignored").during(Duration.ofSeconds(2)).until(instance::getCount, lessThan(2));
+    }
+
+    @Test
+    public void extraCallGetsQueued() {
+        var instance = ExtensionList.lookupSingleton(AsyncPeriodicWorkTestImpl.class);
+        instance.setQueueIfAlreadyRunning(true);
+        assertThat(instance.getCount(), is(0));
+        instance.run();
+        // Gets queued
+        instance.run();
+        await("The second call has been queued and executed later").until(instance::getCount, is(2));
+    }
+
+    @TestExtension
+    public static class AsyncPeriodicWorkTestImpl extends AsyncPeriodicWork {
+        private boolean queueIfAlreadyRunning;
+        private int count = 0;
+
+        public AsyncPeriodicWorkTestImpl() {
+            super(AsyncPeriodicWorkTestImpl.class.getSimpleName());
+        }
+
+        @Override
+        protected void execute(TaskListener listener) throws InterruptedException {
+            count++;
+            Thread.sleep(100);
+        }
+
+        public int getCount() {
+            return count;
+        }
+
+        @Override
+        public long getRecurrencePeriod() {
+            return TimeUnit.DAYS.toMillis(1);
+        }
+
+        @Override
+        public long getInitialDelay() {
+            return TimeUnit.DAYS.toMillis(1);
+        }
+
+        public void setQueueIfAlreadyRunning(boolean queueIfAlreadyRunning) {
+            this.queueIfAlreadyRunning = queueIfAlreadyRunning;
+        }
+
+        @Override
+        protected boolean queueIfAlreadyRunning() {
+            return queueIfAlreadyRunning;
+        }
+    }
+}