Automatic task cleanup

docs/config.md

@@ -1579,11 +1579,32 @@ Read the {ref}`trace-report` page to learn more about the execution report that
 There are additional variables that can be defined within a configuration file that do not have a dedicated scope.
 
 `cleanup`
-: If `true`, on a successful completion of a run all files in *work* directory are automatically deleted.
-
-  :::{warning}
-  The use of the `cleanup` option will prevent the use of the *resume* feature on subsequent executions of that pipeline run. Also, be aware that deleting all scratch files can take a lot of time, especially when using a shared file system or remote cloud storage.
+: :::{versionchanged} 23.10.0
+  Added `'lazy'`, `'eager'`, and `'aggressive'` strategies.
   :::
+: Automatically delete task directories using one of the following strategies:
+
+  `false` (default)
+  : Disable automatic cleanup.
+
+  `true`
+  : Equivalent to `'lazy'`.
+
+  `'lazy'`
+  : If a workflow completes successfully, delete all task directories.
+  : This strategy supports resumability for both successful and failed runs.
+  : Note that deleting all work directories at once can take a lot of time, especially when using a shared file system or remote cloud storage.
+
+  `'eager'`
+  : Delete each task directory as soon as it is no longer needed by downstream tasks.
+  : A task can be deleted once all of the tasks that use any of its output files have completed.
+  : This strategy supports resumability for both successful and failed runs.
+  : Output files that are published via symlink will be invalidated when the original task directory is deleted. Avoid using the following publish modes: `copyNoFollow`, `rellink`, `symlink`.
+
+  `'aggressive'`
+  : Equivalent to `'eager'`, but also deletes individual output files as soon as they are no longer needed.
+  : An output file can be deleted once the tasks that use it have completed. In some cases, an output file can be deleted sooner than its originating task.
+  : This strategy supports resumability for successful runs, but not necessarily for failed runs. Therefore, it is recommended when you want to minimize disk storage during the pipeline and you don't need resumability.
 
 `dumpHashes`
 : If `true`, dump task hash keys in the log file, for debugging purposes.

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -372,6 +372,9 @@ class Session implements ISession {
         // -- file porter config
         this.filePorter = new FilePorter(this)
 
+        // -- normalize cleanup config
+        if( config.cleanup == true )
+            config.cleanup = 'lazy'
     }
 
     protected Path cloudCachePath(Map cloudcache, Path workDir) {
@@ -964,6 +967,17 @@ class Session implements ISession {
         }
     }
 
+    void notifyProcessClose(TaskProcessor process) {
+        observers.each { observer ->
+            try {
+                observer.onProcessClose(process)
+            }
+            catch( Exception e ) {
+                log.debug(e.getMessage(), e)
+            }
+        }
+    }
+
     void notifyProcessTerminate(TaskProcessor process) {
         for( int i=0; i<observers.size(); i++ ) {
             final observer = observers.get(i)
@@ -1149,37 +1163,6 @@ class Session implements ISession {
         errorAction = action
     }
 
-    /**
-     * Delete the workflow work directory from tasks temporary files
-     */
-    void cleanup() {
-        if( !workDir || !config.cleanup )
-            return
-
-        if( aborted || cancelled || error )
-            return
-
-        CacheDB db = null
-        try {
-            log.trace "Cleaning-up workdir"
-            db = CacheFactory.create(uniqueId, runName).openForRead()
-            db.eachRecord { HashCode hash, TraceRecord record ->
-                def deleted = db.removeTaskEntry(hash)
-                if( deleted ) {
-                    // delete folder
-                    FileHelper.deletePath(FileHelper.asPath(record.workDir))
-                }
-            }
-            log.trace "Clean workdir complete"
-        }
-        catch( Exception e ) {
-            log.warn("Failed to cleanup work dir: ${workDir.toUriString()}")
-        }
-        finally {
-            db.close()
-        }
-    }
-
     @Memoized
     CondaConfig getCondaConfig() {
         final cfg = config.conda as Map ?: Collections.emptyMap()

modules/nextflow/src/main/groovy/nextflow/cache/CacheDB.groovy

@@ -77,17 +77,19 @@ class CacheDB implements Closeable {
      * @param processor The {@link TaskProcessor} instance to be assigned to the retrieved task
      * @return A {link TaskEntry} instance or {@code null} if a task for the given hash does not exist
      */
-    TaskEntry getTaskEntry(HashCode taskHash, TaskProcessor processor) {
+    TaskEntry getTaskEntry(HashCode taskHash, Map<String,TaskProcessor> processorLookup=[:]) {
 
         final payload = store.getEntry(taskHash)
         if( !payload )
             return null
 
         final record = (List)KryoHelper.deserialize(payload)
         TraceRecord trace = TraceRecord.deserialize( (byte[])record[0] )
+        final processor = processorLookup[trace.get('process')]
         TaskContext ctx = record[1]!=null && processor!=null ? TaskContext.deserialize(processor, (byte[])record[1]) : null
+        final consumers = record[3]!=null ? ((List<String>)record[3]).collect( s -> HashCode.fromString(s) ) : null
 
-        return new TaskEntry(trace,ctx)
+        return new TaskEntry(processor, trace, ctx, consumers)
     }
 
     void incTaskEntry( HashCode hash ) {
@@ -99,7 +101,7 @@ class CacheDB implements Closeable {
 
         final record = (List)KryoHelper.deserialize(payload)
         // third record contains the reference count for this record
-        record[2] = ((Integer)record[2]) +1
+        record[2] = ((Integer)record[2]) + 1
         // save it again
         store.putEntry(hash, KryoHelper.serialize(record))
 
@@ -114,7 +116,7 @@ class CacheDB implements Closeable {
 
         final record = (List)KryoHelper.deserialize(payload)
         // third record contains the reference count for this record
-        def count = record[2] = ((Integer)record[2]) -1
+        def count = record[2] = ((Integer)record[2]) - 1
         // save or delete
         if( count > 0 ) {
             store.putEntry(hash, KryoHelper.serialize(record))
@@ -128,9 +130,10 @@ class CacheDB implements Closeable {
 
 
     /**
-     * Save task runtime information to th cache DB
+     * Save task runtime information to the cache DB
      *
-     * @param handler A {@link TaskHandler} instance
+     * @param handler
+     * @param trace
      */
     @PackageScope
     void writeTaskEntry0( TaskHandler handler, TraceRecord trace ) {
@@ -143,10 +146,11 @@ class CacheDB implements Closeable {
         // only the 'cache' is active and
         TaskContext ctx = proc.isCacheable() && task.hasCacheableValues() ? task.context : null
 
-        def record = new ArrayList(3)
+        final record = new ArrayList(4)
         record[0] = trace.serialize()
         record[1] = ctx != null ? ctx.serialize() : null
         record[2] = 1
+        record[3] = null
 
         // -- save in the db
         store.putEntry( key, KryoHelper.serialize(record) )
@@ -157,6 +161,31 @@ class CacheDB implements Closeable {
         writer.send { writeTaskEntry0(handler, trace) }
     }
 
+    /**
+     * Finalize task entry in the cache DB with the list of
+     * consumer tasks.
+     *
+     * @param hash
+     * @param consumers
+     */
+    @PackageScope
+    void finalizeTaskEntry0( HashCode hash, List<HashCode> consumers ) {
+
+        final payload = store.getEntry(hash)
+        if( !payload ) {
+            log.debug "Unable to finalize task with key: $hash"
+            return
+        }
+
+        final record = (List)KryoHelper.deserialize(payload)
+        record[3] = consumers.collect( h -> h.toString() )
+        store.putEntry(hash, KryoHelper.serialize(record))
+    }
+
+    void finalizeTaskAsync( HashCode hash, List<HashCode> consumers ) {
+        writer.send { finalizeTaskEntry0(hash, consumers) }
+    }
+
     void cacheTaskAsync( TaskHandler handler ) {
         writer.send {
             writeTaskIndex0(handler,true)
@@ -224,7 +253,7 @@ class CacheDB implements Closeable {
     }
 
     TraceRecord getTraceRecord( HashCode hashCode ) {
-        final result = getTaskEntry(hashCode, null)
+        final result = getTaskEntry(hashCode)
         return result ? result.trace : null
     }
 

modules/nextflow/src/main/groovy/nextflow/processor/PublishDir.groovy

@@ -183,7 +183,7 @@ class PublishDir {
 
         def result = new PublishDir()
         if( params.path )
-            result.path = params.path
+        result.path = params.path
 
         if( params.mode )
             result.mode = params.mode
@@ -217,6 +217,18 @@ class PublishDir {
         return result
     }
 
+    boolean canPublish(Path source, TaskRun task) {
+        if( !sourceDir ) {
+            this.sourceDir = task.targetDir
+            this.sourceFileSystem = sourceDir.fileSystem
+            this.stageInMode = task.config.stageInMode
+            this.task = task
+            validatePublishMode()
+        }
+
+        return getPublishTarget(source) != null
+    }
+
     protected void apply0(Set<Path> files) {
         assert path
 
@@ -301,13 +313,8 @@ class PublishDir {
 
     protected void apply1(Path source, boolean inProcess ) {
 
-        def target = sourceDir ? sourceDir.relativize(source) : source.getFileName()
-        if( matcher && !matcher.matches(target) ) {
-            // skip not matching file
-            return
-        }
-
-        if( saveAs && !(target=saveAs.call(target.toString()))) {
+        final target = getPublishTarget(source)
+        if( !target ) {
             // skip this file
             return
         }
@@ -339,6 +346,17 @@ class PublishDir {
 
     }
 
+    protected def getPublishTarget(Path source) {
+        def target = sourceDir ? sourceDir.relativize(source) : source.getFileName()
+        if( matcher && !matcher.matches(target) )
+            return null
+
+        if( saveAs && !(target=saveAs.call(target.toString())))
+            return null
+
+        return target
+    }
+
     protected Path resolveDestination(target) {
 
         if( target instanceof Path ) {

modules/nextflow/src/main/groovy/nextflow/processor/TaskEntry.groovy

@@ -16,9 +16,11 @@
 
 package nextflow.processor
 
+import com.google.common.hash.HashCode
 import groovy.transform.EqualsAndHashCode
 import groovy.transform.ToString
 import groovy.transform.TupleConstructor
+import nextflow.processor.TaskProcessor
 import nextflow.trace.TraceRecord
 /**
  * Model a task entry persisted in the {@link nextflow.cache.CacheDB}
@@ -30,8 +32,12 @@ import nextflow.trace.TraceRecord
 @TupleConstructor
 class TaskEntry {
 
+    TaskProcessor processor
+
     TraceRecord trace
 
     TaskContext context
 
+    List<HashCode> consumers
+
 }