# Nextflow: dynamically setting resource directives for a process

I'm trying to test variations of CPU and memory for a specific process (samtools sort) to measure computational efficiency (measured only by time). To do this, I'd like to set the CPU and memory for a Nextflow process dynamically.

Ultimate question: How can I dynamically set CPU and memory for many instances of a single Nextflow process definition?

# Attempt 1

The most obvious solution was to submit distinct Nextflow workflows for each combination of CPU and memory by specifying the CPU and memory on the command line (i.e., when calling nextflow run ...), where my Nextflow workflow has global parameters params.samtools_total_threads and params.samtools_mem_per_thread.

This didn't work perfectly because there appeared to be 'collisions' between parent Nextflow instances trying to lock the history file. I got the following error:

Can't lock file: /path/to/project/.nextflow/history -- Nextflow needs to run in a file system that supports file locks


I could try delaying submissions, but I suspect that will still result in collisions.

Here's the SLURM script that I'm using:

TOTAL_THREADS=$$1 MEM_PER_THREAD=$$2

nextflow run 02-SAMTOOLS_SORT.nf \
-with-report "reports/report.$${TOTAL_THREADS}_threads.$${MEM_PER_THREAD}_GB_per_thread.html" \
-with-trace "traces/trace.$${TOTAL_THREADS}_threads.$${MEM_PER_THREAD}_GB_per_thread.txt" \
--samtools_total_threads $$TOTAL_THREADS \ --samtools_mem_per_thread$$MEM_PER_THREAD


My workflow had the following global parameters:

/*
* Make this pipeline a nextflow 2 implementation
*/
nextflow.enable.dsl=2

...

...


# Attempt 2

After Attempt 1 failed, I figured the most 'Nextflow-like way' would be to generate tuples for each combination of CPU and memory that I want to test. Unfortunately, it doesn't appear that it's possible to set CPU and memory directives using parameters passed into the process, but only with global variables.

I get the following No such variable error with the below code (note: most of this code is courtesy of @Steve):

N E X T F L O W  ~  version 21.04.3
Launching 02-SAMTOOLS_SORT.nf [desperate_aryabhata] - revision: fb9e79055a

-- Check script '02-SAMTOOLS_SORT.nf' at line: 33 or see '.nextflow.log' file for more details


Process code

process samtools_coordinate_sort_proc {

tag { "${bam.baseName}" } executor='slurm' queue = 'normal' cpus = total_threads memory = total_mem.GB clusterOptions = "--time 5:00:00 --account <account>" /* delete files upon completion (I think) */ cleanup = true input: path(bam) val(total_threads) val(total_mem) output: path("${bam.baseName}.csorted.bam")

script:

println "avail_mem: $$avail_mem" def mem_per_thread = avail_mem ? "$${avail_mem}G" : ''
println "mem_per_thread: $mem_per_thread" """ samtools sort \\ -@ "$${additional_threads}" \\ -m$${mem_per_thread} \\ -o "$${bam.baseName}.csorted.bam" \\ -T "$${bam.baseName}.csorted" \\ --write-index \\ "${bam}"
"""
}


# Possible attempt 3

Of course, it would be super easy to do this with a couple of bash scripts, but I'm trying to do it in Nextflow for two reasons:

1. To learn Nextflow better
2. Because the traces provided by Nextflow are super convenient (though I could get this information from SLURM, too).

It's possible to specify dynamic computing resources using a closure. For example, this sets the cpus and memory directives using the values supplied via the input declaration:

nextflow.enable.dsl=2

process test_samtools {

echo true

cpus { cpus }
memory { memory }

input:
tuple path(bam), val(cpus), val(memory)

script:

def mem_per_thread = avail_mem ? "-m ${avail_mem}G" : '' """ echo samtools sort \\ -@ "$${additional_threads}" \\$${mem_per_thread} \\ -o "$${bam.baseName}.csorted.bam" \\ -T "$${bam.baseName}.csorted" \\ --write-index \\ "${bam}"
"""
}

workflow {

cpu_ch = Channel.of( 1, 2, 3, 4, 5)
memory_ch = Channel.of( 1.GB, 2.GB, 4.GB )

Channel.fromPath( './data/*.bam' )
| combine( cpu_ch )
| combine( memory_ch )
| test_samtools
}