Fix run hanging forever when a shutdown callback blocks during abort

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<Runnable> 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

modules/nextflow/src/main/groovy/nextflow/util/SimpleAgent.groovy

@@ -36,6 +36,11 @@ import groovy.util.logging.Slf4j
 @CompileStatic
 class SimpleAgent<T> {
 
+    /**
+     * 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<T> {
         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<T> {
      *  the cloned state otherwise the state object itself.
      */
     T getQuickValue() {
+        if( Thread.currentThread()==runner )
+            return currentValue0()
         final retrieve = new RetrieveValueClosure<T>(state)
         events.offerFirst(retrieve)
         return retrieve.getResult()
     }
 
     T getValue() {
+        if( Thread.currentThread()==runner )
+            return currentValue0()
         final retrieve = new RetrieveValueClosure<T>(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> {
 
         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) {

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()
+    }
 }

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]
+    }
+
 }