[FEATURE] Modules for Batch tasks

Is your feature request related to a problem? Please describe

Many short running tasks are a problem for distributed computing. While grouping of processes in future may help, there is still likely going to be overhead.

Describe the solution you'd like

Allow batch processing modules that take a group files and parallelise short running tasks over them.

E.g. Batch processing for indexing Bam files with Samtools:
There would be a BATCH_INDEX process under the samtools folder.

Generalised toy demo of batch processing:
Make input with touch Sample{A,B,C}_{1,2}.fastq.gz.

#! /usr/bin/env nextflow

workflow {
    ch_input = Channel.fromFilePairs( 'Sample*_{1,2}.fastq.gz', checkIfExists: true )
        .toSortedList { a, b -> a[0] <=> b[0] } // Sorted just because. toList() also works, and collect(flatten:false)
        .map { lst -> def tlst = lst.transpose(); [ tlst[0].collect{ [ id: it ] }, tlst[1].flatten() ] }
        .view()
    /* ch_input:
        [ 
            [ 
                [ id: 'SampleA' ], 
                [ id: 'SampleB' ], 
                [ id: 'SampleC' ] 
            ], 
            [ 
                'SampleA_1.fastq.gz', 
                'SampleA_2.fastq.gz', 
                'SampleB_1.fastq.gz', 
                'SampleB_2.fastq.gz',
                'SampleC_1.fastq.gz',
                'SampleC_2.fastq.gz' 
            ] 
        ]
    */
    BATCH_TASK( ch_input, 2 ).batch_fq
        .flatMap{ meta_list, file_list -> meta_list.collect{ meta -> [ meta.findAll{ it.key != 'group'}, file_list.findAll { it.name.startsWith(meta.group) } ] } }
        .view()
}

process BATCH_TASK {

    input:
    tuple val(meta), path(files) // [ [ meta1, meta2, ..., metaN ], [ flattened files ] ]
    val collate_size

    // directives:
    label 'batch'
    cpus { Math.min(6, meta.size()) } // Defined by 'batch' label in config
    debug true // Show echo command output

    script:
    def args = task.ext.args?: ''
    def prefix = task.ext.prefix?: meta*.id  // This is a list of prefixes
    // Update meta with grouping information
    updated_meta = prefix.withIndex().collect{ pfix, idx -> meta[idx] + [ group: pfix ] }
    // Makes a string to pass to printf
    // formats the line as "prefixA fileA1 fileA2 prefixB fileB1 fileB2 ..."
    def cmd_input_str = files.collate( collate_size ).withIndex().collect{ fgroup, idx -> [ prefix[idx], *fgroup ].join(' ') }.join(' ') 
    """
    task(){
        # \$1: prefix
        # \$2: file1
        # \$3: file2
        # TODO: Generally only need to update these commands. 
        echo "command -p \$1 -1 \$2 -2 \$3 $args"
        touch \${1}_{1,2}.cmd.fastq.gz
    }
    export -f task

    # Parallelise
    printf "%s\\n" $cmd_input_str | \\
        xargs -n ${collate_size + 1} \\
            -P$task.cpus \\
            bash -c 'task "\$@"' bash

    cat <<-END_VERSIONS > versions.yml
    "${task.process}":
        tool: \$( tool --version |& sed '1!d; s/.* //' )
    END_VERSIONS
    """
    
    output:
    tuple val(updated_meta), path("*.cmd.fastq.gz"), emit: batch_fq
    path "versions.yml"                            , emit: versions
}

Describe alternatives you've considered

There is perhaps a batch processing / grouping of tasks, but one doesn't know if this will be implemented, and this likely still incurs overhead by all the separate work directories, file staging, and so on of multiple processes even if they're grouped in a single job submission.

Additional context

Related to https://github.com/nf-core/modules/issues/1179

No repository file or test is named; start by reviewing related issue #1179 and the BATCH_TASK toy workflow, including the proposed samtools BATCH_INDEX example. Done would require an agreed module interface and an implemented, tested batch module that groups files and parallelises tasks while addressing the described per-task overhead.