diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy index d55801cf2b..fad02fd9dc 100644 --- a/modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy @@ -75,6 +75,14 @@ class ParallelPollingMonitor extends TaskPollingMonitor { // using an thread-pool via the executor service final wrapper = (Callable)new ThrottlingExecutor.Recoverable() { @Override protected Object invoke() { + // the session may have been aborted or cancelled while this + // submission was sitting in the submitter queue -- submitting now + // would create a job that the shutdown kill sweep has already + // missed and that the polling loop will never observe + if( !session.canSubmitTasks() ) { + log.debug "Skipping task submission -- session no longer accepting submissions > $handler" + return null + } return submit0(handler) } diff --git a/modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy b/modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy index 775f565355..2d5322dff6 100644 --- a/modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy @@ -65,6 +65,9 @@ class ParallelPollingMonitorTest extends Specification { then: handler.submit() >> { println "c=$count"; if(count.getAndIncrement()<2) throw new IllegalArgumentException("Ooops!") } + // the session is healthy in this scenario -- the retries come from the + // submit operation failing, not from the session being aborted + _ * session.canSubmitTasks() >> true success.get() == 1 retry.get() == 2 @@ -118,6 +121,51 @@ class ParallelPollingMonitorTest extends Specification { 10 | 1 | false | false } + def 'should not submit a queued task once the session can no longer submit' () { + given: + def session = Mock(Session) + def handler = Mock(TaskHandler) + and: + def opts = new ThrottlingExecutor.Options().withRateLimit('100/sec') + def exec = ThrottlingExecutor.create(opts) + def mon = new ParallelPollingMonitor(exec, [session:session, name:'foo', pollInterval:'1sec']) + + when: + mon.submit(handler) + exec.shutdown() + exec.awaitTermination(1, TimeUnit.MINUTES) + + then: + // the session was aborted while the wrapper was sitting in the submitter queue + _ * session.canSubmitTasks() >> false + and: + 0 * handler.prepareLauncher() + 0 * handler.submit() + 0 * session.notifyTaskSubmit(_) + } + + def 'should submit a queued task while the session can still submit' () { + given: + def session = Mock(Session) + def handler = Mock(TaskHandler) + and: + def opts = new ThrottlingExecutor.Options().withRateLimit('100/sec') + def exec = ThrottlingExecutor.create(opts) + def mon = new ParallelPollingMonitor(exec, [session:session, name:'foo', pollInterval:'1sec']) + + when: + mon.submit(handler) + exec.shutdown() + exec.awaitTermination(1, TimeUnit.MINUTES) + + then: + _ * session.canSubmitTasks() >> true + and: + 1 * handler.prepareLauncher() + 1 * handler.submit() + 1 * session.notifyTaskSubmit(handler) + } + def 'should inherit array size validation from parent TaskPollingMonitor' () { given: def session = Mock(Session)