Job arrays

docs/executor.rst

@@ -14,6 +14,64 @@ In other words, you can write your pipeline script once and have it running on y
 or the cloud — simply change the executor definition in the Nextflow configuration file.
 
 
+.. _array-executor:
+
+Array
+=====
+
+.. note:: This feature requires Nextflow version ``23.05.0-edge`` or later.
+
+.. warning:: This feature is experimental and may change in a future release.
+
+Many execution platforms support "array jobs", that is, a collection of jobs with the same resource requirements and script
+parameterized by an index. An array job incurs significantly less scheduling overhead compared to submitting each task separately,
+and they are a best practice in HPC environments. Nextflow supports array jobs through the Array executor.
+
+The Array executor takes two config options: an array size (default: ``100``) and a "target" executor (default: ``'local'``).
+These options can be specified in the Nextflow configuration as follows::
+
+    process {
+        executor = 'array'
+    }
+
+    executor {
+        $array {
+            arraySize = 100
+            target = 'slurm'
+        }
+    }
+
+The target executor can be any Nextflow executor that supports array jobs, which currently includes the following:
+
+* :ref:`local-executor`
+* :ref:`slurm-executor`
+
+The Array executor submits tasks in batches to the target executor, as soon as a full batch is ready. Batches are separated
+by process. Any "leftover" tasks are submitted as a partial batch at the end. If any tasks in an array job fail and can be retried,
+they will be retried in another array job without interfering with the tasks that succeeded. Because the order in which tasks are
+executed varies across workflow runs, so too will the grouping of tasks into array jobs.
+
+Aside from the batching, tasks in an array job are executed in the same way, i.e. each task is executed in its own work directory with
+its own script. However, certain configuration properties must be the same for all tasks created by a process.
+
+The following process directives msut be the same for all tasks when using array jobs, because they are specified once for the array job:
+
+* :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, the following additional directives must be uniform:
+
+* :ref:`process-container`
+* :ref:`process-containerOptions`
+
+
 .. _awsbatch-executor:
 
 AWS Batch

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -424,6 +424,9 @@ class Session implements ISession {
         return result
     }
 
+    void registerObserver(TraceObserver observer) {
+        observers << observer
+    }
 
     /*
      * intercepts interruption signal i.e. CTRL+C
@@ -935,6 +938,17 @@ class Session implements ISession {
         }
     }
 
+    void notifyProcessClose(String process) {
+        observers.each { observer ->
+            try {
+                observer.onProcessClose(process)
+            }
+            catch( Exception e ) {
+                log.debug(e.getMessage(), e)
+            }
+        }
+    }
+
     void notifyProcessTerminate(TaskProcessor process) {
         for( int i=0; i<observers.size(); i++ ) {
             final observer = observers.get(i)

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

@@ -21,6 +21,7 @@ import java.nio.file.Path
 import groovy.transform.CompileStatic
 import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
+import nextflow.processor.TaskHandler
 import nextflow.processor.TaskMonitor
 import nextflow.processor.TaskPollingMonitor
 import nextflow.processor.TaskProcessor
@@ -36,7 +37,7 @@ import org.apache.commons.lang.StringUtils
  */
 @Slf4j
 @CompileStatic
-abstract class AbstractGridExecutor extends Executor {
+abstract class AbstractGridExecutor extends Executor implements ArrayTaskAware {
 
     protected Duration queueInterval
 
@@ -65,7 +66,7 @@ abstract class AbstractGridExecutor extends Executor {
     /*
      * Prepare and launch the task in the underlying execution platform
      */
-    GridTaskHandler createTaskHandler(TaskRun task) {
+    TaskHandler createTaskHandler(TaskRun task) {
         assert task
         assert task.workDir
 
@@ -407,5 +408,43 @@ abstract class AbstractGridExecutor extends Executor {
         // Instead, it is the command wrapper script that is launched run within a container process.
         return isFusionEnabled()
     }
+
+    @Override
+    ArrayTaskHandler createArrayTaskHandler(List<TaskRun> array) {
+        final handlers = array.collect { task -> createTaskHandler(task) }
+        new ArrayGridTaskHandler(handlers, this)
+    }
+
+    String createArrayTaskWrapper(ArrayGridTaskHandler handler) {
+        final array = handler.array
+        final task = array.first().getTask()
+
+        final arrayHeader = getArrayDirective(array.size())
+        final taskHeaders = getHeaders(task)
+        final files = array
+            .collect { h -> ((GridTaskHandler)h).wrapperFile }
+            .join(' ')
+
+        final builder = new StringBuilder()
+            << '#!/bin/bash\n'
+            << "${headerToken} ${arrayHeader}\n"
+            << taskHeaders
+            << "declare -a array=( ${files} )\n"
+            << "bash \${array[\$${arrayIndexName}]}\n"
+
+        return builder.toString()
+    }
+
+    protected String getArrayDirective(int arraySize) {
+        throw new UnsupportedOperationException("Executor '${name}' does not support array jobs")
+    }
+
+    protected String getArrayIndexName() {
+        throw new UnsupportedOperationException("Executor '${name}' does not support array jobs")
+    }
+
+    protected List<String> getArraySubmitCommandLine() {
+        throw new UnsupportedOperationException("Executor '${name}' does not support array jobs")
+    }
 }
 

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

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2013-2023, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package nextflow.executor
+
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import groovy.transform.CompileStatic
+import groovy.util.logging.Slf4j
+import nextflow.processor.ArrayTaskPollingMonitor
+import nextflow.processor.TaskHandler
+import nextflow.processor.TaskMonitor
+import nextflow.processor.TaskProcessor
+import nextflow.processor.TaskRun
+import nextflow.trace.TraceObserver
+import nextflow.util.Duration
+import org.codehaus.groovy.runtime.typehandling.GroovyCastException
+/**
+ * Executor that submits tasks in batches to a target executor
+ * that supports array jobs.
+ *
+ * @author Ben Sherman <bentshermann@gmail.com>
+ */
+@Slf4j
+@CompileStatic
+class ArrayExecutor extends Executor implements TraceObserver {
+
+    private ArrayTaskAware target
+
+    private Integer arraySize
+
+    private Map<String,Queue<TaskRun>> queues = new ConcurrentHashMap<>()
+
+    private Map<String,Boolean> closed = new ConcurrentHashMap<>()
+
+    /**
+     * Initialize the executor class
+     */
+    @Override
+    protected void register() {
+        super.register()
+
+        session.registerObserver(this)
+
+        final targetName = session.getExecConfigProp('array', 'target', 'local') as String
+        try {
+            target = (ArrayTaskAware)session.executorFactory.getExecutor(targetName, session)
+        }
+        catch( GroovyCastException e ) {
+            throw new IllegalArgumentException("Executor '${targetName}' does not support array jobs")
+        }
+
+        arraySize = session.getExecConfigProp('array', 'arraySize', 100) as Integer
+
+        log.debug "Creating 'array' executor > target executor: '${targetName}', array size: ${arraySize}"
+    }
+
+    @Override
+    TaskMonitor createTaskMonitor() {
+        return ArrayTaskPollingMonitor.create(session, name, 100, Duration.of('5 sec'))
+    }
+
+    /**
+     * Add submitted tasks to the queue, and schedule an array job when
+     * the queue reaches the desired size.
+     *
+     * @param task
+     */
+    @Override
+    synchronized void submit( TaskRun task ) {
+        log.trace "Scheduling process: ${task}"
+
+        if( session.isTerminated() )
+            new IllegalStateException("Session terminated - Cannot add process to execution array: ${task}")
+
+        final process = task.processor.name
+
+        // submit task directly if process has already closed
+        if( closed[process] ) {
+            ((Executor)target).submit(task)
+            return
+        }
+
+        // initialize process queue
+        if( process !in queues )
+            queues[process] = new ConcurrentLinkedQueue<>()
+
+        // add task to the process queue
+        final queue = queues[process]
+        queue.add(task)
+
+        // schedule array job when a batch is ready
+        if( queue.size() >= arraySize ) {
+            log.debug "[ARRAY] Submitting array job for process '${process}'"
+            submit0(queue, arraySize)
+        }
+    }
+
+    synchronized private void submit0( Queue<TaskRun> queue, int size ) {
+        def array = new ArrayList<TaskRun>()
+        def iter = queue.iterator()
+
+        for( int i : 1..size ) {
+            array << iter.next()
+            iter.remove()
+        }
+
+        monitor.schedule(target.createArrayTaskHandler(array))
+    }
+
+    @Override
+    TaskHandler createTaskHandler(TaskRun task) {
+        throw new UnsupportedOperationException()
+    }
+
+    /**
+     * Submit any remaining tasks as a partial batch when a process is closed.
+     *
+     * @param process
+     */
+    @Override
+    void onProcessClose(String process) {
+        final queue = queues[process]
+
+        if( queue != null && queue.size() > 0 ) {
+            log.debug "[ARRAY] Submitting remainder array job for process '${process}'"
+            submit0(queue, queue.size())
+        }
+
+        closed[process] = true
+    }
+
+}