Automatic task cleanup

docs/config.md

@@ -1431,6 +1431,21 @@ There are additional variables that can be defined within a configuration file t
   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.
   :::
 
+  :::{versionadded} 23.10.0
+  :::
+
+  Setting `cleanup = 'eager'` enables the "eager" cleanup strategy. This strategy will delete each task directory as soon as it is no longer needed by downstream tasks, rather than at the end of the workflow run. This feature is useful for minimizing disk usage during the workflow run.
+
+  The lifetime of a task is determined by the downstream tasks that use the task's output files. When all of these tasks finish, the upstream task can be deleted.
+
+  The following caveats apply when using eager cleanup:
+
+  - Eager cleanup will break the resumability of your pipeline. If a workflow run fails, you will have to restart from the beginning, whereas with `cleanup = true` the cleanup would not have happened. As a result, eager cleanup is designed to be used only when you are confident that the workflow run will not fail.
+
+  - Output files should not be published via symlink when using eager cleanup, because the symlinks will be invalidated when the original task directory is deleted. Avoid using the following publish modes: `copyNoFollow`, `rellink`, `symlink`.
+
+  - Eager cleanup currently does not work properly with processes that forward input files with the `includeInputs` option. In this case, the forwarded input files will be deleted prematurely, and any process that consumes the forwarded output channel may fail or produce incorrect output.
+
 `dumpHashes`
 : If `true`, dump task hash keys in the log file, for debugging purposes.
 

modules/nextflow/build.gradle

@@ -34,7 +34,7 @@ dependencies {
     api "commons-codec:commons-codec:1.15"
     api "commons-io:commons-io:2.11.0"
     api "com.beust:jcommander:1.35"
-    api("com.esotericsoftware.kryo:kryo:2.24.0") { exclude group: 'com.esotericsoftware.minlog', module: 'minlog' }
+    api("com.esotericsoftware:kryo:5.5.0")
     api('org.iq80.leveldb:leveldb:0.12')
     api('org.eclipse.jgit:org.eclipse.jgit:6.5.0.202303070854-r')
     api ('javax.activation:activation:1.1.1')

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

@@ -938,6 +938,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)
@@ -1123,7 +1134,7 @@ class Session implements ISession {
      * Delete the workflow work directory from tasks temporary files
      */
     void cleanup() {
-        if( !workDir || !config.cleanup )
+        if( !workDir || config.cleanup != true )
             return
 
         if( aborted || cancelled || error )

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<HashCode>)record[3]) : 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
+        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

@@ -168,7 +168,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
@@ -202,6 +202,19 @@ class PublishDir {
         return result
     }
 
+    @CompileStatic
+    boolean canPublish(Path source, TaskRun task) {
+        if( !sourceDir ) {
+            this.sourceDir = task.targetDir
+            this.sourceFileSystem = sourceDir.fileSystem
+            this.stageInMode = task.config.stageInMode
+            this.taskName = task.name
+            validatePublishMode()
+        }
+
+        return getPublishTarget(source) != null
+    }
+
     @CompileStatic
     protected void apply0(Set<Path> files) {
         assert path
@@ -290,13 +303,8 @@ class PublishDir {
     @CompileStatic
     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
         }
@@ -328,6 +336,18 @@ class PublishDir {
 
     }
 
+    @CompileStatic
+    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
+    }
+
     @CompileStatic
     protected Path resolveDestination(target) {
 

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