diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy index d55801cf2b..b122e65f46 100644 --- a/modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy +++ b/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) } diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy index c2bcb7b73f..0f25c19719 100644 --- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy +++ b/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 { diff --git a/modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy b/modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy index 775f565355..ecd72e6d3a 100644 --- a/modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy +++ b/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 + } + } diff --git a/modules/nextflow/src/test/groovy/nextflow/processor/TaskPollingMonitorTest.groovy b/modules/nextflow/src/test/groovy/nextflow/processor/TaskPollingMonitorTest.groovy index 18e3a4a203..b22b71494b 100644 --- a/modules/nextflow/src/test/groovy/nextflow/processor/TaskPollingMonitorTest.groovy +++ b/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 + } + }