add resource limits to workflow

FastOMA_light.nf

@@ -133,31 +133,34 @@ Parameters:
 
 
 process check_input{
-    publishDir params.output_folder, mode: 'copy'
-    input:
-        path proteome_folder
-        path hogmap_folder
-        path species_tree
-        path omamerdb
-        path splice_folder
-    output:
-        path "species_tree_checked.nwk"
-        val "check_completed"
-    script:
-        """
-        fastoma-check-input --proteomes ${proteome_folder} \
-                            --species-tree ${species_tree} \
-                            --out-tree species_tree_checked.nwk \
-                            --splice ${splice_folder} \
-                            --hogmap ${hogmap_folder} \
-                            --omamer_db ${omamerdb} \
-                            -vv
-        """
+  label "process_single"
+  publishDir params.output_folder, mode: 'copy'
+
+  input:
+    path proteome_folder
+    path hogmap_folder
+    path species_tree
+    path omamerdb
+    path splice_folder
+  output:
+    path "species_tree_checked.nwk"
+    val "check_completed"
+  script:
+    """
+    fastoma-check-input --proteomes ${proteome_folder} \
+                        --species-tree ${species_tree} \
+                        --out-tree species_tree_checked.nwk \
+                        --splice ${splice_folder} \
+                        --hogmap ${hogmap_folder} \
+                        --omamer_db ${omamerdb} \
+                        -vv
+    """
 }
 
 
 process omamer_run{
-  time {4.h}
+  label "process_single"
+  tag "$proteome"
   publishDir params.hogmap_folder, mode: 'copy'
 
   input:
@@ -178,11 +181,11 @@ process omamer_run{
 
 
 process infer_roothogs{
+  label "process_single"
   publishDir = [
     path: params.temp_output,
     enabled: params.debug_enabled,
-    //mode: 'copy', saveAs: { filename -> filename.equals('temp_omamer_rhogs') ? null : filename }
-    ]
+  ]
 
   input:
     path hogmaps, stageAs: "hogmaps/*"
@@ -219,8 +222,8 @@ process batch_roothogs{
 }
 
 process hog_big{
-  cpus  2
-  time {20.h}     // for very big rhog it might need more, or you could re-run and add `-resume`
+  label "process_high"
+
   input:
     each rhogsbig
     path species_tree
@@ -239,6 +242,8 @@ process hog_big{
 }
 
 process hog_rest{
+  label "process_single"
+
   input:
     each rhogsrest
     path species_tree

conf/base.config

@@ -0,0 +1,55 @@
+/*
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+    dessimozlab/FastOMA Nextflow base config file
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+    A 'blank slate' config file, appropriate for general use on most high performance
+    compute environments. Assumes that all software is installed and available on
+    the PATH. Runs in `local` mode - all jobs will be run on the logged in environment.
+----------------------------------------------------------------------------------------
+*/
+
+process {
+
+    cpus   = { check_max( 1    * task.attempt, 'cpus'   ) }
+    memory = { check_max( 6.GB * task.attempt, 'memory' ) }
+    time   = { check_max( 4.h  * task.attempt, 'time'   ) }
+    shell  = ['/bin/bash', '-euo', 'pipefail']
+
+    errorStrategy = { task.exitStatus in ((130..145) + 104) ? 'retry' : 'finish' }
+    maxRetries    = 1
+    maxErrors     = '-1'
+
+    withLabel:process_single {
+        cpus   = { check_max( 1                   , 'cpus'    ) }
+        memory = { check_max( 12.GB * task.attempt, 'memory'  ) }
+        time   = { check_max( 4.h   * task.attempt, 'time'    ) }
+    }
+    withLabel:process_low {
+        cpus   = { check_max( 2     * task.attempt, 'cpus'    ) }
+        memory = { check_max( 12.GB * task.attempt, 'memory'  ) }
+        time   = { check_max( 4.h   * task.attempt, 'time'    ) }
+    }
+    withLabel:process_medium {
+        cpus   = { check_max( 6     * task.attempt, 'cpus'    ) }
+        memory = { check_max( 36.GB * task.attempt, 'memory'  ) }
+        time   = { check_max( 8.h   * task.attempt, 'time'    ) }
+    }
+    withLabel:process_high {
+        cpus   = { check_max( 12    * task.attempt, 'cpus'    ) }
+        memory = { check_max( 72.GB * task.attempt, 'memory'  ) }
+        time   = { check_max( 16.h  * task.attempt, 'time'    ) }
+    }
+    withLabel:process_long {
+        time   = { check_max( 20.h  * task.attempt, 'time'    ) }
+    }
+    withLabel:process_high_memory {
+        memory = { check_max( 200.GB * task.attempt, 'memory' ) }
+    }
+    withLabel:error_ignore {
+        errorStrategy = 'ignore'
+    }
+    withLabel:error_retry {
+        errorStrategy = 'retry'
+        maxRetries    = 2
+    }
+}

nextflow.config

@@ -18,6 +18,12 @@ params {
 
   output_folder = "Output"
   statsdir = "${params.output_folder}/stats"
+
+  // Max resource options
+  // Defaults only, expecting to be overwritten
+  max_memory                 = '128.GB'
+  max_cpus                   = 24
+  max_time                   = '120.h'
 }
 
 // Profiles configure nextflow depending on the environment (local, docker, singularity)
@@ -69,11 +75,54 @@ profiles {
   }
 }
 
+def trace_timestamp = new java.util.Date().format( 'yyyy-MM-dd_HH-mm-ss')
 timeline {
   enabled = params.report
-  file = "${params.statsdir}/timeline.html"
+  file = "${params.statsdir}/timeline_${trace_timestamp}.html"
 }
 report {
   enabled = params.report
-  file = "${params.statsdir}/report.html"
+  file = "${params.statsdir}/report_{trace_timestamp}.html"
+}
+trace {
+    enabled = params.report
+    file    = "${params.statsdir}/trace_${trace_timestamp}.txt"
+}
+dag {
+    enabled = params.report
+    file    = "${params.statsdir}/pipeline_dag_${trace_timestamp}.html"
+}
+
+includeConfig "conf/base.config"
+
+// function to check maximum resources
+def check_max(obj, type) {
+    if (type == 'memory') {
+        try {
+            if (obj.compareTo(params.max_memory as nextflow.util.MemoryUnit) == 1)
+                return params.max_memory as nextflow.util.MemoryUnit
+            else
+                return obj
+        } catch (all) {
+            println "   ### ERROR ###   Max memory '${params.max_memory}' is not valid! Using default value: $obj"
+            return obj
+        }
+    } else if (type == 'time') {
+        try {
+            if (obj.compareTo(params.max_time as nextflow.util.Duration) == 1)
+                return params.max_time as nextflow.util.Duration
+            else
+                return obj
+        } catch (all) {
+            println "   ### ERROR ###   Max time '${params.max_time}' is not valid! Using default value: $obj"
+            return obj
+        }
+    } else if (type == 'cpus') {
+        try {
+            return Math.min( obj, params.max_cpus as int )
+        } catch (all) {
+            println "   ### ERROR ###   Max cpus '${params.max_cpus}' is not valid! Using default value: $obj"
+            return obj
+        }
+    }
 }