diff --git a/modules/nextflow/src/main/groovy/nextflow/Session.groovy b/modules/nextflow/src/main/groovy/nextflow/Session.groovy index 34df841e17..cf527f4222 100644 --- a/modules/nextflow/src/main/groovy/nextflow/Session.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/Session.groovy @@ -20,8 +20,10 @@ import java.nio.file.Files import java.nio.file.Path import java.nio.file.Paths import java.util.concurrent.ConcurrentLinkedQueue +import java.util.concurrent.CountDownLatch import java.util.concurrent.ExecutorService import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean import java.util.function.Consumer @@ -86,6 +88,7 @@ import nextflow.trace.event.TaskEvent import nextflow.trace.event.WorkflowOutputEvent import nextflow.util.Barrier import nextflow.util.ClassLoaderFactory +import nextflow.util.Duration import nextflow.util.HistoryFile import nextflow.util.LoggerHelper import nextflow.util.NameGenerator @@ -104,6 +107,12 @@ import sun.misc.SignalHandler @CompileStatic class Session implements ISession { + /** + * Max time to wait for the shutdown callbacks completion, when they are + * executed by another thread e.g. the thread that aborted the session + */ + static private final Duration SHUTDOWN_TIMEOUT = Duration.of('5min') + /** * Keep a list of all processor created */ @@ -275,6 +284,8 @@ class Session implements ISession { private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false) + private final CountDownLatch shutdownComplete = new CountDownLatch(1) + private Queue shutdownCallbacks = new ConcurrentLinkedQueue<>() private int poolSize @@ -762,21 +773,32 @@ 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) ) + if( !shutdownInitiated.compareAndSet(false, true) ) { + // the callbacks are being executed by another thread e.g. the thread that + // aborted the session -- await their completion, but not indefinitely, so that + // a stuck callback cannot prevent the pipeline execution from terminating + if( !shutdownComplete.await(SHUTDOWN_TIMEOUT.millis, TimeUnit.MILLISECONDS) ) + log.warn "Timed out awaiting the completion of the shutdown callbacks (>$SHUTDOWN_TIMEOUT) -- Forcing pipeline termination" return - log.trace "Invoking ${shutdownCallbacks.size()} shutdown callbacks" - while( shutdownCallbacks.size() ) { - final hook = shutdownCallbacks.poll() - try { - hook.run() - } - catch( Exception e ) { - log.debug "Failed to execute shutdown hook: ${hook.class.name}", e - } } + try { + log.trace "Invoking ${shutdownCallbacks.size()} shutdown callbacks" + while( shutdownCallbacks.size() ) { + final hook = shutdownCallbacks.poll() + try { + hook.run() + } + catch( Exception e ) { + log.debug "Failed to execute shutdown hook: ${hook.class.name}", e + } + } - // -- invoke observers completion handlers - notifyFlowComplete() + // -- invoke observers completion handlers + notifyFlowComplete() + } + finally { + shutdownComplete.countDown() + } } /** @@ -830,15 +852,17 @@ class Session implements ISession { // dump threads status if( log.isTraceEnabled() ) log.trace(SysHelper.dumpThreads()) - // invoke shutdown callbacks - shutdown0() - notifyError(null) - // force termination - logObserver?.forceTermination() + // force termination *before* running the shutdown callbacks, otherwise a callback + // taking too long (or hanging) would prevent the release of the threads awaiting + // the pipeline termination, and therefore hang the execution -- see issue #7444 executorFactory?.signalExecutors() processesBarrier.forceTermination() monitorsBarrier.forceTermination() operatorsForceTermination() + // invoke shutdown callbacks + shutdown0() + notifyError(null) + logObserver?.forceTermination() } catch( Throwable e ) { log.debug "Unexpected error while aborting execution", e diff --git a/modules/nextflow/src/main/groovy/nextflow/util/SimpleAgent.groovy b/modules/nextflow/src/main/groovy/nextflow/util/SimpleAgent.groovy index aa026dda44..53bbce8acb 100644 --- a/modules/nextflow/src/main/groovy/nextflow/util/SimpleAgent.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/util/SimpleAgent.groovy @@ -36,6 +36,11 @@ import groovy.util.logging.Slf4j @CompileStatic class SimpleAgent { + /** + * Max time to wait for the agent runner thread to provide the current state + */ + static private final Duration GET_VALUE_TIMEOUT = Duration.of('1min') + private T state private BlockingDeque events = new LinkedBlockingDeque<>() private Thread runner @@ -45,7 +50,7 @@ class SimpleAgent { if(state == null) throw new IllegalArgumentException("Missing state argument") this.state = state - this.runner = Threads.start(this.&run) + this.runner = Threads.start("agent-${state.getClass().getSimpleName()}".toString(), this.&run) } SimpleAgent onError(@ClosureParams(value = SimpleType, options = ['java.lang.Throwable']) Closure handler) { @@ -72,17 +77,31 @@ class SimpleAgent { * the cloned state otherwise the state object itself. */ T getQuickValue() { + if( Thread.currentThread()==runner ) + return currentValue0() final retrieve = new RetrieveValueClosure(state) events.offerFirst(retrieve) return retrieve.getResult() } T getValue() { + if( Thread.currentThread()==runner ) + return currentValue0() final retrieve = new RetrieveValueClosure(state) events.offer(retrieve) return retrieve.getResult() } + /** + * Retrieve the state directly, without going through the events queue. It's meant to be + * used only when the invoking thread is the agent runner itself, that otherwise would + * deadlock awaiting for an event that only it can serve. + */ + @CompileDynamic + private T currentValue0() { + return (T)(state instanceof Cloneable ? state.clone() : state) + } + protected void run() { while(true) { try { @@ -129,7 +148,12 @@ class SimpleAgent { T getResult() { try { - sync.await() + // note: do not await indefinitely, otherwise a stalled runner thread + // would hang the invoking thread forever -- see issue #7444 + if( !sync.await(GET_VALUE_TIMEOUT.millis, TimeUnit.MILLISECONDS) ) { + log.warn "Timed out awaiting the agent result (>$GET_VALUE_TIMEOUT) -- Returning the current state" + return (T)s0 + } return (T)result } catch (InterruptedException e) { diff --git a/modules/nextflow/src/test/groovy/nextflow/SessionTest.groovy b/modules/nextflow/src/test/groovy/nextflow/SessionTest.groovy index 370b210d92..f795b2fef5 100644 --- a/modules/nextflow/src/test/groovy/nextflow/SessionTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/SessionTest.groovy @@ -19,6 +19,7 @@ package nextflow import java.nio.file.Files import java.nio.file.Paths import java.nio.file.attribute.PosixFilePermission +import java.util.concurrent.CountDownLatch import nextflow.config.Manifest import nextflow.container.ContainerConfig @@ -27,6 +28,7 @@ import nextflow.container.PodmanConfig import nextflow.container.SarusConfig import nextflow.exception.AbortOperationException import nextflow.file.FileHelper +import nextflow.processor.TaskProcessor import nextflow.script.ScriptFile import nextflow.script.WorkflowMetadata import nextflow.trace.TraceFileObserver @@ -435,4 +437,29 @@ class SessionTest extends Specification { then: 1 * observer.onFlowComplete() } + + def 'should release the await barrier when a shutdown callback is blocking' () { + given: + def blocked = new CountDownLatch(1) + def release = new CountDownLatch(1) + def session = new Session() + def processor = Mock(TaskProcessor) + // register a process, so that `await` blocks on the processes barrier + session.processRegister(processor) + // register a shutdown callback that never returns until it's released + session.onShutdown { blocked.countDown(); release.await() } + + when: + Thread.start { session.abort() } + blocked.await() + // the main thread must be able to complete the await, even though + // the shutdown callback is still hanging + def main = Thread.start { session.await() } + main.join(30_000) + then: + !main.isAlive() + + cleanup: + release.countDown() + } } diff --git a/modules/nextflow/src/test/groovy/nextflow/util/SimpleAgentTest.groovy b/modules/nextflow/src/test/groovy/nextflow/util/SimpleAgentTest.groovy index e59c1432db..7e59f6e884 100644 --- a/modules/nextflow/src/test/groovy/nextflow/util/SimpleAgentTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/util/SimpleAgentTest.groovy @@ -16,6 +16,9 @@ package nextflow.util +import java.util.concurrent.CompletableFuture +import java.util.concurrent.TimeUnit + import spock.lang.Specification /** @@ -50,4 +53,21 @@ class SimpleAgentTest extends Specification { } + def 'should get the value when invoked by the runner thread' () { + given: + def state = [] + def result = new CompletableFuture() + SimpleAgent agent + // the error handler is invoked by the agent runner thread itself, therefore + // it cannot await for an event that only that thread is able to serve + agent = new SimpleAgent(state).onError { result.complete(agent.getValue()) } + + when: + agent.send { state<<1 } + agent.send { throw new RuntimeException('Oops') } + + then: + result.get(30, TimeUnit.SECONDS) == [1] + } + }