Release the maxForks slot when task submission fails

modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy

@@ -86,6 +86,7 @@ class ParallelPollingMonitor extends TaskPollingMonitor {
             protected void onFailure(Throwable e) {
                 if( !session.success )
                     return // ignore error when the session has been interrupted
+                releaseSubmitSlot(handler)
                 handleException(handler, e)
                 notifyTaskComplete(handler)
             }

modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy

@@ -659,6 +659,7 @@ class TaskPollingMonitor implements TaskMonitor {
                 submit(handler)
             }
             catch ( Throwable e ) {
+                releaseSubmitSlot(handler)
                 handleException(handler, e)
                 notifyTaskComplete(handler)
             }
@@ -671,6 +672,21 @@ class TaskPollingMonitor implements TaskMonitor {
     }
 
 
+    /**
+     * Give back the {@code maxForks} slot acquired just before a submit attempt that
+     * then failed. The handler is not in the running queue in that case -- {@link #submit}
+     * adds it only after {@code handler.submit()} returns -- so {@link #evict} returns
+     * false and {@link #handleException} would not release the slot.
+     *
+     * @param handler
+     *      The {@link TaskHandler} whose submission threw
+     */
+    protected void releaseSubmitSlot(TaskHandler handler) {
+        if( !handler || runningQueue.contains(handler) )
+            return
+        handler.decProcessForks()
+    }
+
     final protected void handleException( TaskHandler handler, Throwable error ) {
         def fault = null
         try {

modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy

@@ -19,8 +19,10 @@ package nextflow.processor
 
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.LongAdder
 
 import nextflow.Session
+import nextflow.exception.ProcessException
 import nextflow.util.Duration
 import nextflow.util.ThrottlingExecutor
 import spock.lang.Specification
@@ -140,4 +142,32 @@ class ParallelPollingMonitorTest extends Specification {
         e.message.contains("Process 'oversized_array' declares array size (10) which exceeds the executor queue size (5)")
     }
 
+    def 'should release the forks slot when the parallel submit fails' () {
+        given:
+        def adder = new LongAdder()
+        def processor = Mock(TaskProcessor) { getForksCount() >> adder }
+        def session = Mock(Session) { isSuccess() >> true }
+        and:
+        def opts = new ThrottlingExecutor.Options().withRateLimit('100/sec')
+        def exec = ThrottlingExecutor.create(opts)
+        def monitor = Spy(new ParallelPollingMonitor(exec, [session: session, name: 'foo', pollInterval: '1sec']))
+        and:
+        def handler = Spy(TaskHandler) { getTraceRecord() >> null }
+        handler.task = Mock(TaskRun) { getProcessor() >> processor }
+        handler.submit() >> { throw new ProcessException('Cannot submit task') }
+
+        when:
+        // the slot is taken by the monitor before handing the submit to the executor
+        handler.incProcessForks()
+        monitor.submit(handler)
+        exec.shutdown()
+        exec.awaitTermination(1, TimeUnit.MINUTES)
+
+        then:
+        // onFailure must give the slot back -- the handler never reached the running queue
+        adder.intValue() == 0
+        and:
+        monitor.runningQueue.size() == 0
+    }
+
 }

modules/nextflow/src/test/groovy/nextflow/processor/TaskPollingMonitorTest.groovy

@@ -16,8 +16,11 @@
 
 package nextflow.processor
 
+import java.util.concurrent.atomic.LongAdder
+import java.util.concurrent.locks.ReentrantLock
 
 import nextflow.Session
+import nextflow.exception.ProcessException
 import nextflow.executor.ExecutorConfig
 import nextflow.util.Duration
 import nextflow.util.RateUnit
@@ -223,4 +226,75 @@ class TaskPollingMonitorTest extends Specification {
         10       | 5             | 3          | true     | false    | false            | false     // Not ready
     }
 
+    /**
+     * Initialise the locks/conditions normally created by {@link TaskPollingMonitor#start()},
+     * without spawning the polling and submitter threads.
+     */
+    private static TaskPollingMonitor initLocks(TaskPollingMonitor monitor) {
+        final lock = new ReentrantLock()
+        monitor.@pendingLock = lock
+        monitor.@taskAvail = lock.newCondition()
+        monitor.@slotAvail = lock.newCondition()
+        return monitor
+    }
+
+    def 'should release the forks slot when the task submit throws' () {
+        given:
+        def adder = new LongAdder()
+        def processor = Mock(TaskProcessor) { getForksCount() >> adder }
+        def session = Mock(Session) { canSubmitTasks() >> true }
+        and:
+        def monitor = Spy(new TaskPollingMonitor(name: 'foo', session: session, pollInterval: Duration.of('1min')))
+        initLocks(monitor)
+        and:
+        def handler = Spy(TaskHandler) {
+            isReady() >> true
+            canForkProcess() >> true
+            getTraceRecord() >> null
+        }
+        handler.task = Mock(TaskRun) { getProcessor() >> processor }
+        // the submit fails, so the handler is never added to the running queue
+        handler.submit() >> { throw new ProcessException('Cannot submit task') }
+
+        when:
+        monitor.schedule(handler)
+        monitor.submitPendingTasks()
+
+        then:
+        // the slot acquired before the failed submit must be given back
+        adder.intValue() == 0
+        and:
+        monitor.runningQueue.size() == 0
+    }
+
+    def 'should not release the forks slot twice when a running task completes' () {
+        given:
+        def adder = new LongAdder(); adder.increment()
+        def processor = Mock(TaskProcessor) { getForksCount() >> adder }
+        def session = Mock(Session)
+        and:
+        def monitor = Spy(new TaskPollingMonitor(name: 'foo', session: session, pollInterval: Duration.of('1min')))
+        initLocks(monitor)
+        and:
+        def handler = Spy(TaskHandler) {
+            checkIfRunning() >> false
+            checkIfCompleted() >> true
+        }
+        handler.task = Mock(TaskRun) { getProcessor() >> processor }
+        and:
+        monitor.runningQueue.add(handler)
+        // run the finalizer inline instead of on the (unstarted) finalizer pool
+        monitor.@enableAsyncFinalizer = false
+        monitor.finalizeTask(_) >> null
+
+        when:
+        monitor.checkTaskStatus(handler)
+
+        then:
+        // checkTaskStatus already decrements and evicts -- exactly one release
+        adder.intValue() == 0
+        and:
+        monitor.runningQueue.size() == 0
+    }
+
 }