text: user136: 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:
  
  To learn Nextflow better
  Because the traces provided by Nextflow are super convenient (though I could get this information from SLURM, too).
  
  
  user10775: 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
  }
  
  
original_text: 136: <p>I'm trying to test variations of CPU and memory for a specific process (<code>samtools sort</code>) to measure computational efficiency (measured only by time). To do this, I'd like to set the CPU and memory for a <code>Nextflow</code> <code>process</code> dynamically.</p>
  <p><strong>Ultimate question:</strong> How can I dynamically set CPU and memory for many instances of a single <code>Nextflow</code> process definition?</p>
  <h1>Attempt 1</h1>
  <p>The most obvious solution was to submit distinct <code>Nextflow</code> workflows for each combination of CPU and memory by specifying the CPU and memory on the command line (i.e., when calling <code>nextflow run ...</code>), where my <code>Nextflow</code> workflow has global parameters <code>params.samtools_total_threads</code> and <code>params.samtools_mem_per_thread</code>.</p>
  <p>This didn't work perfectly because there appeared to be 'collisions' between parent <code>Nextflow</code> instances trying to lock the history file. I got the following error:</p>
  <pre><code>Can't lock file: /path/to/project/.nextflow/history -- Nextflow needs to run in a file system that supports file locks
  </code></pre>
  <p>I could try delaying submissions, but I suspect that will still result in collisions.</p>
  <p>Here's the <code>SLURM</code> script that I'm using:</p>
  <pre><code>TOTAL_THREADS=<span class="math-container">$1
  MEM_PER_THREAD=$</span>2
  
  nextflow run 02-SAMTOOLS_SORT.nf \
      -with-report &quot;reports/report.<span class="math-container">${TOTAL_THREADS}_threads.$</span>{MEM_PER_THREAD}_GB_per_thread.html&quot; \
      -with-trace &quot;traces/trace.<span class="math-container">${TOTAL_THREADS}_threads.$</span>{MEM_PER_THREAD}_GB_per_thread.txt&quot; \
      --samtools_total_threads <span class="math-container">$TOTAL_THREADS \
      --samtools_mem_per_thread $</span>MEM_PER_THREAD
  </code></pre>
  <p>My workflow had the following global parameters:</p>
  <pre><code>/*
   * Make this pipeline a nextflow 2 implementation
   */
  nextflow.enable.dsl=2
  
  ...
  
  params.samtools_total_threads = '4'
  params.samtools_mem_per_thread = '8'
  
  ...
  </code></pre>
  <h1>Attempt 2</h1>
  <p>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 <code>process</code>, but only with global variables.</p>
  <p>I get the following <code>No such variable</code> error with the below code (note: most of this code is courtesy of @Steve):</p>
  <pre><code>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
  </code></pre>
  <p><strong>Process code</strong></p>
  <pre><code>process samtools_coordinate_sort_proc {
  
      tag { &quot;${bam.baseName}&quot; }
  
      executor='slurm'
      queue = 'normal'
      cpus = total_threads
      memory = total_mem.GB
      clusterOptions = &quot;--time 5:00:00 --account &lt;account&gt;&quot;
  
      /* delete files upon completion (I think) */
      cleanup = true
  
      input:
      path(bam)
      val(total_threads)
      val(total_mem)
  
      output:
      path(&quot;${bam.baseName}.csorted.bam&quot;)
  
      script:
  
      def additional_threads = task.cpus - 1
  
      def avail_mem = task.memory ? task.memory.toGiga().intdiv(task.cpus) : 0
      println &quot;avail_mem: <span class="math-container">$avail_mem"
      def mem_per_thread = avail_mem ? "$</span>{avail_mem}G&quot; : ''
      println &quot;mem_per_thread: $mem_per_thread&quot;
  
      &quot;&quot;&quot;
      samtools sort \\
          -@ &quot;<span class="math-container">${additional_threads}" \\
      -m $</span>{mem_per_thread} \\
          -o &quot;<span class="math-container">${bam.baseName}.csorted.bam" \\
      -T "$</span>{bam.baseName}.csorted&quot; \\
          --write-index \\
          &quot;${bam}&quot;
      &quot;&quot;&quot;
  }
  </code></pre>
  <h1>Possible attempt 3</h1>
  <p>Of course, it would be <em>super</em> easy to do this with a couple of <code>bash</code> scripts, but I'm trying to do it in <code>Nextflow</code> for two reasons:</p>
  <ol>
  <li>To learn <code>Nextflow</code> better</li>
  <li>Because the traces provided by <code>Nextflow</code> are super convenient (though I could get this information from <code>SLURM</code>, too).</li>
  </ol>
   10775: <p>It's possible to specify <a href="https://www.nextflow.io/docs/latest/process.html#dynamic-computing-resources" rel="noreferrer">dynamic computing resources</a> using a closure. For example, this sets the <a href="https://www.nextflow.io/docs/latest/process.html#cpus" rel="noreferrer">cpus</a> and <a href="https://www.nextflow.io/docs/latest/process.html#memory" rel="noreferrer">memory</a> directives using the values supplied via the input declaration:</p>
  <pre><code>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 ? &quot;-m ${avail_mem}G&quot; : ''
  
      &quot;&quot;&quot;
      echo samtools sort \\
          -@ &quot;<span class="math-container">${additional_threads}" \\
          $</span>{mem_per_thread} \\
          -o &quot;<span class="math-container">${bam.baseName}.csorted.bam" \\
          -T "$</span>{bam.baseName}.csorted&quot; \\
          --write-index \\
          &quot;${bam}&quot;
      &quot;&quot;&quot;
  }
  
  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
  }
  </code></pre>
  