Merge pull request #226 from uclahs-cds/jarbet-mark-dup-spark-mem-retry

add retry method for run_MarkDuplicatesSpark_GATK and run_sort_SAMtools

Author: jarbet

.gitmodules

@@ -2,3 +2,6 @@
 	path = external/nextflow-modules
 	url = [email protected]:uclahs-cds/pipeline-Nextflow-module.git
 	branch = main
+[submodule "external/nextflow-config"]
+	path = external/nextflow-config
+	url = [email protected]:uclahs-cds/pipeline-Nextflow-config.git

CHANGELOG.md

@@ -25,6 +25,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Use SAMtools `sort` instead of Picard `SortSam`
 
 ### Added
+- Add `retry` method to `run_sort_SAMtools` and `run_MarkDuplicatesSpark_GATK` (if run out of RAM then retry with more memory)
 - Add process `run_merge_SAMtools`: use when `params.mark_duplicates=false` to ensure multiple BAM outputs are merged
 - `.github/CODEOWNERS`
 - Add config file for F16 node

config/F32.config

@@ -16,6 +16,12 @@ process {
     withName: run_sort_SAMtools {
         cpus = 12
         memory = 15.GB
+        retry_strategy {
+            memory {
+                strategy = 'exponential'
+                operand = 2
+                }
+            }
         }
     withName: run_merge_SAMtools {
         cpus = 12
@@ -28,5 +34,11 @@ process {
     withName: run_MarkDuplicatesSpark_GATK {
         cpus = 16
         memory = 35.GB
+        retry_strategy {
+            memory {
+                strategy = 'add'
+                operand = 29.GB
+                }
+            }
         }
     }

config/F72.config

@@ -16,6 +16,12 @@ process {
     withName: run_sort_SAMtools {
         cpus = 12
         memory = 15.GB
+        retry_strategy {
+            memory {
+                strategy = 'exponential'
+                operand = 2
+                }
+            }
         }
     withName: run_merge_SAMtools {
         cpus = 12
@@ -28,5 +34,11 @@ process {
     withName: run_MarkDuplicatesSpark_GATK {
         cpus = 16
         memory = 35.GB
+        retry_strategy {
+            memory {
+                strategy = 'exponential'
+                operand = 2
+                }
+            }
         }
     }

config/M64.config

@@ -16,6 +16,12 @@ process {
     withName: run_sort_SAMtools {
         cpus = 12
         memory = 15.GB
+        retry_strategy {
+            memory {
+                strategy = 'exponential'
+                operand = 2
+                }
+            }
         }
     withName: run_merge_SAMtools {
         cpus = 12
@@ -28,6 +34,12 @@ process {
     withName: run_MarkDuplicatesSpark_GATK {
         cpus = 16
         memory = 35.GB
+        retry_strategy {
+            memory {
+                strategy = 'exponential'
+                operand = 2
+                }
+            }
         }
     }
 

config/base.config

@@ -5,7 +5,12 @@
 process {
     cpus = { methods.check_max( 1 * task.attempt, 'cpus' ) }
 
-    errorStrategy = { task.exitStatus in [143, 137, 104, 134, 139] ? 'retry' : 'finish' }
+    commonRetryCodes = [104, 134, 137, 139, 143, 247] // Common out-of-memory error codes
+    withName: 'run_MarkDuplicatesSpark_GATK' {
+        ext.retry_codes = [52] // Spark OutOfMemory error codes
+    }
+        
+    errorStrategy = { task.exitStatus in (((task?.ext?.retry_codes) ? task.ext.retry_codes : []) + process.commonRetryCodes) ? 'retry' : 'terminate' }
     maxRetries = 1
 
     withLabel:process_low {

config/methods.config

@@ -1,3 +1,4 @@
+includeConfig "../external/nextflow-config/config/retry/retry.config"
 
 methods {
 
@@ -186,6 +187,44 @@ methods {
         }
     }
 
+    // Function to ensure that resource requirements don't go beyond
+    // a maximum limit or below a minimum limit
+    // needed for ../external/nextflow-config/config/retry/retry.config
+    check_limits = { 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 if (obj.compareTo(params.min_memory as nextflow.util.MemoryUnit) == -1)
+                    return params.min_memory as nextflow.util.MemoryUnit
+                else
+                    return obj
+            } catch (all) {
+                println "   ### WARNING ###   Max memory '${params.max_memory}' or min memory '${params.min_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 if (obj.compareTo(params.min_time as nextflow.util.Duration) == -1)
+                    return params.min_time as nextflow.util.Duration
+                else
+                    return obj
+            } catch (all) {
+                println "   ### WARNING ###   Max time '${params.max_time}' or min time '${params.min_time}' is not valid! Using default value: $obj"
+                return obj
+            }
+        } else if (type == 'cpus') {
+            try {
+                return Math.max( Math.min( obj, params.max_cpus as int ), params.min_cpus as int )
+            } catch (all) {
+                println "   ### WARNING ###   Max cpus '${params.max_cpus}' or min cpus '${params.min_cpus}' is not valid! Using default value: $obj"
+                return obj
+            }
+        }
+    }
+
     // Resource allocation here. Static node-based allocations included here
     set_resources_allocation = {
         // Function to ensure that resource requirements don't go beyond
@@ -263,5 +302,6 @@ methods {
         methods.set_pipeline_logs()
         methods.check_aligner()
         methods.set_resources_allocation()
+        retry.setup_retry()
         }
     }