diff --git a/modules/nextflow/src/main/groovy/nextflow/Session.groovy b/modules/nextflow/src/main/groovy/nextflow/Session.groovy index bc444875a7..9704a1de16 100644 --- a/modules/nextflow/src/main/groovy/nextflow/Session.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/Session.groovy @@ -22,6 +22,7 @@ import java.nio.file.Paths import java.util.concurrent.ConcurrentLinkedQueue import java.util.concurrent.ExecutorService import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicBoolean import java.util.function.Consumer import com.google.common.hash.HashCode @@ -271,7 +272,7 @@ class Session implements ISession { private volatile Throwable error - private volatile boolean shutdownInitiated + private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false) private Queue shutdownCallbacks = new ConcurrentLinkedQueue<>() @@ -759,8 +760,10 @@ class Session implements ISession { } final protected void shutdown0() { + // guard against adding shutdown hooks after shutdown, or calling shutdown more than once + if( !shutdownInitiated.compareAndSet(false, true) ) + return log.trace "Invoking ${shutdownCallbacks.size()} shutdown callbacks" - shutdownInitiated = true while( shutdownCallbacks.size() ) { final hook = shutdownCallbacks.poll() try { @@ -987,8 +990,8 @@ class Session implements ISession { log.warn "Shutdown hook cannot be null\n${ExceptionUtils.getStackTrace(new Exception())}" return } - if( shutdownInitiated ) - throw new IllegalStateException("Session shutdown already initiated — Hook cannot be added: $hook") + if( shutdownInitiated.get() ) + throw new IllegalStateException("Session shutdown already initiated -- Hook cannot be added: ${hook.class.name}") shutdownCallbacks.add(hook) } diff --git a/modules/nextflow/src/test/groovy/nextflow/SessionTest.groovy b/modules/nextflow/src/test/groovy/nextflow/SessionTest.groovy index b418081fb7..370b210d92 100644 --- a/modules/nextflow/src/test/groovy/nextflow/SessionTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/SessionTest.groovy @@ -31,6 +31,7 @@ import nextflow.script.ScriptFile import nextflow.script.WorkflowMetadata import nextflow.trace.TraceFileObserver import nextflow.trace.TraceHelper +import nextflow.trace.TraceObserverV2 import nextflow.trace.WorkflowStatsObserver import nextflow.util.Duration import nextflow.util.VersionNumber @@ -415,4 +416,23 @@ class SessionTest extends Specification { true | true } + + def 'should notify flow complete only once when abort and destroy race' () { + given: + def observer = Mock(TraceObserverV2) + def session = new Session() + session.@observersV2 = [observer] + + when: + // certain error conditions can abort the session on a separate thread + // while the main thread winds it down via destroy() + // both call shutdown0() -> notifyFlowComplete() concurrently + def t1 = Thread.start { session.abort() } + def t2 = Thread.start { session.destroy() } + t1.join() + t2.join() + + then: + 1 * observer.onFlowComplete() + } }