Job arrays

docs/process.md

@@ -1349,6 +1349,64 @@ Allowed values for the `arch` directive are as follows, grouped by equivalent fa
 
 Examples of values for the architecture `target` option are `cascadelake`, `icelake`, `zen2` and `zen3`. See the Spack documentation for the full and up-to-date [list of meaningful targets](https://spack.readthedocs.io/en/latest/basic_usage.html#support-for-specific-microarchitectures).
 
+(process-array)=
+
+## array
+
+:::{versionadded} 24.02.0-edge
+:::
+
+:::{warning} *Experimental: may change in a future release.*
+:::
+
+The `array` directive allows you to submit tasks as *job arrays* for executors that support it.
+
+A job array is a collection of jobs with the same resource requirements and the same script (parameterized by an index). Job arrays incur significantly less scheduling overhead compared to individual jobs, and as a result they are preferred by HPC schedulers where possible.
+
+The directive should be specified with a given array size, along with an executor that supports job arrays. For example:
+
+```groovy
+process cpu_task {
+    executor 'slurm'
+    array 100
+
+    '''
+    your_command --here
+    '''
+}
+```
+
+Nextflow currently supports job arrays for the following executors:
+
+- {ref}`awsbatch-executor`
+- {ref}`google-batch-executor`
+- {ref}`lsf-executor`
+- {ref}`pbs-executor`
+- {ref}`pbspro-executor`
+- {ref}`sge-executor`
+- {ref}`slurm-executor`
+
+A process using job arrays will collect tasks and submit each batch as a job array when it is ready. Any "leftover" tasks will be submitted as a partial job array.
+
+Once a job array is submitted, each "child" task is executed as an independent job. Any tasks that fail (and can be retried) will be retried without interfering with the tasks that succeeded. Retried tasks are submitted individually rather than through a job array, in order to allow for the use of [dynamic resources](#dynamic-computing-resources).
+
+The following directives must be uniform across all tasks in a process that uses job arrays, because these directives are specified once for the entire job array:
+
+- {ref}`process-accelerator`
+- {ref}`process-clusterOptions`
+- {ref}`process-cpus`
+- {ref}`process-disk`
+- {ref}`process-machineType`
+- {ref}`process-memory`
+- {ref}`process-queue`
+- {ref}`process-resourcelabels`
+- {ref}`process-time`
+
+For cloud-based executors like AWS Batch, or when using Fusion with any executor, the following additional directives must be uniform:
+
+- {ref}`process-container`
+- {ref}`process-containerOptions`
+
 (process-beforescript)=
 
 ### beforeScript

modules/nextflow/src/main/groovy/nextflow/executor/AbstractGridExecutor.groovy

@@ -36,7 +36,7 @@ import org.apache.commons.lang.StringUtils
  */
 @Slf4j
 @CompileStatic
-abstract class AbstractGridExecutor extends Executor {
+abstract class AbstractGridExecutor extends Executor implements TaskArrayAware {
 
     protected Duration queueInterval
 
@@ -407,5 +407,10 @@ abstract class AbstractGridExecutor extends Executor {
         // Instead, it is the command wrapper script that is launched run within a container process.
         return isFusionEnabled()
     }
+
+    String getArrayTaskId(String jobId, int index) {
+        throw new UnsupportedOperationException("Executor '${name}' does not support job arrays")
+    }
+
 }
 

modules/nextflow/src/main/groovy/nextflow/executor/CrgExecutor.groovy

@@ -18,6 +18,7 @@ package nextflow.executor
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import nextflow.processor.TaskArray
 import nextflow.processor.TaskRun
 /**
  * An executor specialised for CRG cluster
@@ -41,6 +42,11 @@ class CrgExecutor extends SgeExecutor {
             task.config.penv = 'smp'
         }
 
+        if( task instanceof TaskArray ) {
+            final arraySize = task.getArraySize()
+            result << '-t' << "1-${arraySize}".toString()
+        }
+
         result << '-N' << getJobNameFor(task)
         result << '-o' << quote(task.workDir.resolve(TaskRun.CMD_LOG))
         result << '-j' << 'y'

modules/nextflow/src/main/groovy/nextflow/executor/Executor.groovy

@@ -81,7 +81,7 @@ abstract class Executor {
      *
      * @param task A {@code TaskRun} instance
      */
-    final void submit( TaskRun task ) {
+    void submit( TaskRun task ) {
         log.trace "Scheduling process: ${task}"
 
         if( session.isTerminated() ) {

modules/nextflow/src/main/groovy/nextflow/executor/GridTaskHandler.groovy

@@ -38,11 +38,13 @@ import nextflow.exception.ProcessNonZeroExitStatusException
 import nextflow.file.FileHelper
 import nextflow.fusion.FusionAwareTask
 import nextflow.fusion.FusionHelper
+import nextflow.processor.TaskArray
 import nextflow.processor.TaskHandler
 import nextflow.processor.TaskRun
 import nextflow.trace.TraceRecord
 import nextflow.util.CmdLineHelper
 import nextflow.util.Duration
+import nextflow.util.Escape
 import nextflow.util.Throttle
 /**
  * Handles a job execution in the underlying grid platform
@@ -100,6 +102,27 @@ class GridTaskHandler extends TaskHandler implements FusionAwareTask {
         this.sanityCheckInterval = duration
     }
 
+    @Override
+    void prepareLauncher() {
+        // -- create the wrapper script
+        createTaskWrapper(task).build()
+    }
+
+    @Override
+    String getWorkDir() {
+        fusionEnabled()
+            ? FusionHelper.toContainerMount(task.workDir).toString()
+            : task.workDir.toString()
+    }
+
+    @Override
+    List<String> getLaunchCommand() {
+        final workDir = Escape.path(getWorkDir())
+        final cmd = "bash ${workDir}/${TaskRun.CMD_RUN} 2>&1 | tee ${workDir}/${TaskRun.CMD_LOG}"
+
+        List.of('bash', '-o', 'pipefail', '-c', cmd.toString())
+    }
+
     protected ProcessBuilder createProcessBuilder() {
 
         // -- log the qsub command
@@ -189,7 +212,7 @@ class GridTaskHandler extends TaskHandler implements FusionAwareTask {
             // -- wait the the process completes
             final result = process.text
             final exitStatus = process.waitFor()
-            final cmd = launchCmd0(builder,pipeScript)
+            final cmd = submitCmd0(builder,pipeScript)
 
             if( exitStatus ) {
                 throw new ProcessNonZeroExitStatusException("Failed to submit process to grid scheduler for execution", result, exitStatus, cmd)
@@ -237,12 +260,12 @@ class GridTaskHandler extends TaskHandler implements FusionAwareTask {
         return result
     }
 
-    protected String launchCmd0(ProcessBuilder builder, String pipeScript) {
+    protected String submitCmd0(ProcessBuilder builder, String pipeScript) {
         def result = CmdLineHelper.toLine(builder.command())
         if( pipeScript ) {
-            result = "cat << 'LAUNCH_COMMAND_EOF' | ${result}\n"
+            result = "cat << 'SUBMIT_COMMAND_EOF' | ${result}\n"
             result += pipeScript.trim() + '\n'
-            result += 'LAUNCH_COMMAND_EOF\n'
+            result += 'SUBMIT_COMMAND_EOF\n'
         }
         return result
     }
@@ -254,17 +277,15 @@ class GridTaskHandler extends TaskHandler implements FusionAwareTask {
     void submit() {
         ProcessBuilder builder = null
         try {
-            // -- create the wrapper script
-            createTaskWrapper(task).build()
             // -- start the execution and notify the event to the monitor
             builder = createProcessBuilder()
             // -- forward the job launcher script to the command stdin if required
             final stdinScript = executor.pipeLauncherScript() ? stdinLauncherScript() : null
             // -- execute with a re-triable strategy
             final result = safeExecute( () -> processStart(builder, stdinScript) )
-            // -- save the JobId in the
-            this.jobId = executor.parseJobId(result)
-            this.status = SUBMITTED
+            // -- save the job id
+            final jobId = (String)executor.parseJobId(result)
+            this.onSubmit(jobId)
             log.debug "[${executor.name.toUpperCase()}] submitted process ${task.name} > jobId: $jobId; workDir: ${task.workDir}"
 
         }
@@ -284,6 +305,18 @@ class GridTaskHandler extends TaskHandler implements FusionAwareTask {
 
     }
 
+    void onSubmit(String jobId) {
+        if( task instanceof TaskArray ) {
+            task.children.eachWithIndex { handler, i ->
+                final arrayTaskId = executor.getArrayTaskId(jobId, i)
+                ((GridTaskHandler)handler).onSubmit(arrayTaskId)
+            }
+        }
+        else {
+            this.jobId = jobId
+            this.status = SUBMITTED
+        }
+    }
 
     private long startedMillis
 

modules/nextflow/src/main/groovy/nextflow/executor/LsfExecutor.groovy

@@ -23,6 +23,7 @@ import java.util.regex.Pattern
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import nextflow.fusion.FusionHelper
+import nextflow.processor.TaskArray
 import nextflow.processor.TaskRun
 /**
  * Processor for LSF resource manager
@@ -104,7 +105,13 @@ class LsfExecutor extends AbstractGridExecutor {
         }
 
         // -- the job name
-        result << '-J' << getJobNameFor(task)
+        if( task instanceof TaskArray ) {
+            final arraySize = task.getArraySize()
+            result << '-J' << "${getJobNameFor(task)}[1-${arraySize}]".toString()
+        }
+        else {
+            result << '-J' << getJobNameFor(task)
+        }
 
         // -- at the end append the command script wrapped file name
         result.addAll( task.config.getClusterOptionsAsList() )
@@ -304,4 +311,14 @@ class LsfExecutor extends AbstractGridExecutor {
     boolean isFusionEnabled() {
         return FusionHelper.isFusionEnabled(session)
     }
+
+    @Override
+    String getArrayIndexName() { 'LSB_JOBINDEX' }
+
+    @Override
+    int getArrayIndexStart() { 1 }
+
+    @Override
+    String getArrayTaskId(String jobId, int index) { "${jobId}[${index + 1}]" }
+
 }

modules/nextflow/src/main/groovy/nextflow/executor/PbsExecutor.groovy

@@ -21,6 +21,7 @@ import java.util.regex.Pattern
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import nextflow.processor.TaskArray
 import nextflow.processor.TaskRun
 /**
  * Implements a executor for PBS/Torque cluster
@@ -43,6 +44,11 @@ class PbsExecutor extends AbstractGridExecutor {
     protected List<String> getDirectives( TaskRun task, List<String> result ) {
         assert result !=null
 
+        if( task instanceof TaskArray ) {
+            final arraySize = task.getArraySize()
+            result << '-J' << "0-${arraySize - 1}".toString()
+        }
+
         result << '-N' << getJobNameFor(task)
         result << '-o' << quote(task.workDir.resolve(TaskRun.CMD_LOG))
         result << '-j' << 'oe'
@@ -174,4 +180,13 @@ class PbsExecutor extends AbstractGridExecutor {
     static protected boolean matchOptions(String value) {
         value ? OPTS_REGEX.matcher(value).find() : null
     }
+
+    @Override
+    String getArrayIndexName() { 'PBS_ARRAY_INDEX' }
+
+    @Override
+    String getArrayTaskId(String jobId, int index) {
+        jobId.replace('[]', "[$index]")
+    }
+
 }

modules/nextflow/src/main/groovy/nextflow/executor/PbsProExecutor.groovy

@@ -18,6 +18,7 @@ package nextflow.executor
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import nextflow.processor.TaskArray
 import nextflow.processor.TaskRun
 /**
  * Implements a executor for PBSPro cluster executor
@@ -44,7 +45,12 @@ class PbsProExecutor extends PbsExecutor {
     @Override
     protected List<String> getDirectives(TaskRun task, List<String> result ) {
         assert result !=null
-
+
+        if( task instanceof TaskArray ) {
+            final arraySize = task.getArraySize()
+            result << '-J' << "0-${arraySize - 1}".toString()
+        }
+
         // when multiple competing directives are provided, only the first one will take effect
         // therefore clusterOptions is added as first to give priority over other options as expected
         // by the clusterOptions semantics -- see https://github.com/nextflow-io/nextflow/pull/2036

modules/nextflow/src/main/groovy/nextflow/executor/SgeExecutor.groovy

@@ -19,6 +19,7 @@ import java.nio.file.Path
 
 import groovy.transform.CompileStatic
 import nextflow.fusion.FusionHelper
+import nextflow.processor.TaskArray
 import nextflow.processor.TaskRun
 /**
  * Execute a task script by running it on the SGE/OGE cluster
@@ -37,6 +38,11 @@ class SgeExecutor extends AbstractGridExecutor {
      */
     protected List<String> getDirectives(TaskRun task, List<String> result) {
 
+        if( task instanceof TaskArray ) {
+            final arraySize = task.getArraySize()
+            result << '-t' << "1-${arraySize}".toString()
+        }
+
         result << '-N' << getJobNameFor(task)
         result << '-o' << quote(task.workDir.resolve(TaskRun.CMD_LOG))
         result << '-j' << 'y'
@@ -114,8 +120,14 @@ class SgeExecutor extends AbstractGridExecutor {
             if( entry.toString().isLong() )
                 return entry
 
+            if( (id=entry.tokenize('.').get(0)).isLong() )
+                return id
+
             if( entry.startsWith('Your job') && entry.endsWith('has been submitted') && (id=entry.tokenize().get(2)) )
                 return id
+
+            if( entry.startsWith('Your job array') && entry.endsWith('has been submitted') && (id=entry.tokenize().get(3)) )
+                return id.tokenize('.').get(0)
         }
 
         throw new IllegalStateException("Invalid SGE submit response:\n$text\n\n")
@@ -185,4 +197,14 @@ class SgeExecutor extends AbstractGridExecutor {
     boolean isFusionEnabled() {
         return FusionHelper.isFusionEnabled(session)
     }
+
+    @Override
+    String getArrayIndexName() { 'SGE_TASK_ID' }
+
+    @Override
+    int getArrayIndexStart() { 1 }
+
+    @Override
+    String getArrayTaskId(String jobId, int index) { "${jobId}.${index}" }
+
 }

modules/nextflow/src/main/groovy/nextflow/executor/SlurmExecutor.groovy

@@ -22,6 +22,7 @@ import java.util.regex.Pattern
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import nextflow.fusion.FusionHelper
+import nextflow.processor.TaskArray
 import nextflow.processor.TaskRun
 /**
  * Processor for SLURM resource manager
@@ -54,6 +55,11 @@ class SlurmExecutor extends AbstractGridExecutor {
      */
     protected List<String> getDirectives(TaskRun task, List<String> result) {
 
+        if( task instanceof TaskArray ) {
+            final arraySize = task.getArraySize()
+            result << '--array' << "0-${arraySize - 1}".toString()
+        }
+
         result << '-J' << getJobNameFor(task)
         result << '-o' << quote(task.workDir.resolve(TaskRun.CMD_LOG))     // -o OUTFILE and no -e option => stdout and stderr merged to stdout/OUTFILE
         result << '--no-requeue' << '' // note: directive need to be returned as pairs
@@ -211,4 +217,11 @@ class SlurmExecutor extends AbstractGridExecutor {
     boolean isFusionEnabled() {
         return FusionHelper.isFusionEnabled(session)
     }
+
+    @Override
+    String getArrayIndexName() { 'SLURM_ARRAY_TASK_ID' }
+
+    @Override
+    String getArrayTaskId(String jobId, int index) { "${jobId}_${index}" }
+
 }