Nextflow: dynamically setting resource directives for a process

Question

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

...

params.samtools_total_threads = '4'
params.samtools_mem_per_thread = '8'

...

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
No such variable: total_threads

 -- 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:

    def additional_threads = task.cpus - 1

    def avail_mem = task.memory ? task.memory.toGiga().intdiv(task.cpus) : 0
    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).

Answer 1

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 additional_threads = task.cpus - 1

    def avail_mem = task.memory ? task.memory.toGiga().intdiv(task.cpus) : 0
    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
}