How can I ensure that processes respect the order of a queue channel?

[Subject-13]

Hello there,

As the title suggests, I'm struggling with how to ensure that processes respect the order of a queue channel (correct me here if it is not) and process its elements in sequence?

I am using version 23.10.1 and have the following process that should take into account the order of the channel orderedConvertAndFilterSequences:

process ampliconSequences {
    fair true
    publishDir "${params.outdir}/${file_name}/3_amplicon"
    cpus ("${params.cpus}".toInteger() >= 7 ? 7 : "${params.cpus}".toInteger() - 1)
    tag "${name}"
    //maxForks 1

    input:
    tuple val(order), file(filtered_sequences), val(file_name), file(converted_table)

    output:
    tuple path("*_trimmed.bed"), val(file_name), emit: trimmed_sequences_bed_tuple

    script:
    """
    seqkit amplicon ${filtered_sequences} -p ${converted_table} -m ${params.mismat} -j ${task.cpus} -M --bed > ${filtered_sequences.simpleName}_trimmed.bed
    """
}

workflow {
   ...
   orderedConvertAndFilterSequences = orderedConvertAndFilterSequences
                                                    | toSortedList { a, b -> a[0] <=> b[0] }
                                                    | flatMap

   orderedConvertAndFilterSequences.view()
   ampliconSequences(orderedConvertAndFilterSequences)
   ...
}

The orderedConvertAndFilterSequences channel looks like this after I have sorted it based on the first element:

[1, .../work/2d/0af90d74c93d81c7af1a3fdda9d63b/file_a.fasta, file_a, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[2, .../work/fe/0741a0ed175de49c75b5e7647af299/file_b.fasta, file_b, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[4, .../work/52/8a081642a144f4b25f236a4ebed4aa/file_c.fasta, file_c, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[7, .../work/61/6d3fe624cb44c67ced91bee70711aa/file_d.fasta, file_d, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[8, .../work/f3/8f559e96dc8eedbc6a69168b7847df/file_e.fasta, file_e, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[9, .../work/f6/e919b57ce3049f82e675cc5be39c0e/file_f.fasta, file_f, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[10, .../work/7e/e62ffc552a80d0536416f1c21618cb/file_g.fasta, file_g, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[10, .../work/97/5ab083e293121e6ee7629d3dc7c4d6/file_h.fasta, file_h, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[10, .../work/02/5794d676aee1125b94508590c8d47d/file_i.fasta, file_i, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]
[10, .../work/d0/0dc2225bad727bd8639bed0ca0830f/file_j.fasta, file_j, .../work/6e/f411aff4ab4e17bbcdf520b3ba4ed7/table.tsv]

I have shortened the paths here for display purposes, in the original they are complete and exist.

Sorting my channel based on the first element works. The script does not throw an error and the process is created and executed 10 times. However, the order of my previously sorted channel is not taken into account. The choice of elements that are processed first appears random. I have run the script several times and the choice of the first element of the process appears random and not sorted. I tried to force the process with maxForks 1 to process the elements sequentially, even if more resources were available. Also setting the boolean fair true has no effect on the order.

Is there anything else I can try or am I missing something?

I would appreciate support and answers!

EDIT: Typos

[bentsherman]

The fair directive ensures that the process outputs are emitted in the same order in which they were received, but it does not affect the order in which they are processed. Anything that can be executed in parallel will be, based on the available resources.

Why do you need to ensure that the inputs are processed in order?

[Subject-13]

Thank you for your answer. I am working with large data sets from a sequencer that contains several different samples. I have already built a very large pipeline with Nextflow for my analysis. Since parts of the analysis take a very long time, it would be advantageous to send the samples with a higher weighting through this process first, so that they can already be forwarded to the next processes. Background: I have programmed a GUI for this analysis that is to be made available to several users. I would like some samples to be considered first by the analysis. To do this, I allow the user to assign a priority order to the samples in advance so that I can sort the samples within Nextflow in the channel and start the processes accordingly.

[bentsherman]

I see. I'm not sure if that can be done at the moment. It might become possible once we make the process syntax more flexible, because then you would be able to invoke the tasks in a specific order. But it is not possible right now

[Subject-13]

Of course it would be nice if this would be a possibility in the future. Until then, I am open to any suggestions and advice. However, thanks for the clarification!

[Subject-13]

I've been trying to figure out where I should make a change for myself, but I can't really identify where exactly I should make the adjustment. Would you have a tip for me where I could start with this idea? Would it still work within the source code of Nextflow or would that be a modification of GPars?

[bentsherman]

A quick and dirty heuristic... create a "large" and "small" version of the process (e.g. with an include alias) and process the large and small samples separately. Add a fake dependency between the large and small so that the large samples are processed first