`maxForks 1` + `fair true` do not guarantee FIFO task submission order when channel emits all items simultaneously

[thealanjason]

Bug report

When a channel emits all items simultaneously (via toSortedList + flatMap), processes with maxForks 1 and fair true do not submit tasks in channel order. Instead, tasks appear to be submitted in an arbitrary order determined by thread scheduling at task-creation time.

This causes a critical correctness problem when two separate processes (with separate environments) need to coordinate via IPC on the same experiment key simultaneously. If they pick different keys, they deadlock waiting on each other's IPC signals.

Expected behavior and actual behavior

When two processes (SERVER and CLIENT) each have maxForks 1 and fair true, and each receives a sorted channel of N keys, both processes should submit their tasks in channel order — so that SERVER(exp_01) and CLIENT(exp_01) run concurrently.

Instead, the first submitted tasks are chosen arbitrarily from the middle of the sorted list (e.g. SERVER starts on exp_05, CLIENT on exp_04). The two processes deadlock immediately: SERVER is waiting for exp_05/DONE while CLIENT is waiting for exp_04/READY.

Steps to reproduce the problem

Save the following as main.nf and run nextflow run main.nf:

#!/usr/bin/env nextflow
                                                                                                                                        
def N_EXPERIMENTS = 6
                                                                                                                                        
workflow {      

    keys = Channel.of( (1..N_EXPERIMENTS).collect { sprintf('exp_%02d', it) } ).flatten()
                                                                                                                                        
    comm_dirs = keys.map { key ->
        def comm = "/tmp/mwe_ipc_${workflow.sessionId}/${key}"                                                                          
        [key, comm]                                                                                                                     
    }                                                                                                                                   
                                                                                                                                        
    comm_dirs.multiMap { key, comm ->                                                                                                   
        server: [key, comm]
        client: [key, comm]                                                                                                             
    }.set { _split }
                                                                                                                                        
    server_inputs = _split.server
        .toSortedList { a, b -> a[0] <=> b[0] }                                                                                         
        .flatMap { it }
                                                                                                                                        
    client_inputs = _split.client
        .toSortedList { a, b -> a[0] <=> b[0] }                                                                                         
        .flatMap { it }
                                                                                                                                        
    SERVER(server_inputs)
    CLIENT(client_inputs)                                                                                                               
}               

process SERVER {
    maxForks 1
    fair true

    input:
    tuple val(key), val(comm_dir)
                                                                                                                                        
    script:
    """                                                                                                                                 
    mkdir -p "${comm_dir}"
    echo "[SERVER] key=${key}  writing READY to ${comm_dir}"
    touch "${comm_dir}/READY"                                                                                                           
    echo "[SERVER] key=${key}  waiting for DONE..."
    until [ -e "${comm_dir}/DONE" ]; do sleep 0.2; done                                                                                 
    echo "[SERVER] key=${key}  got DONE, exiting"                                                                                       
    rm -f "${comm_dir}/READY" "${comm_dir}/DONE"
    """                                                                                                                                 
}               
                                                                                                                                        
process CLIENT {
    maxForks 1
    fair true

    input:
    tuple val(key), val(comm_dir)

    output:
    val key

    script:
    """
    echo "[CLIENT] key=${key}  waiting for READY in ${comm_dir}..."                                                                     
    until [ -e "${comm_dir}/READY" ]; do sleep 0.2; done
    echo "[CLIENT] key=${key}  found READY — doing work"                                                                                
    sleep 1                                                                                                                             
    echo "[CLIENT] key=${key}  done, writing DONE"
    touch "${comm_dir}/DONE"                                                                                                            
    """         
}

The workflow hangs immediately. Checking the work directories for the two submitted tasks confirms SERVER and CLIENT are processing different keys.

Program output

executor >  local (2)
[51/0c3aab] SERVER (5) [  0%] 0 of 6
[ef/e721c1] CLIENT (4) [  0%] 0 of 6                                                                                                    

Checking the work directories: SERVER (5) ran exp_05, CLIENT (4) ran exp_04 — a mismatch causing a deadlock.

The .nextflow.log 5-minute task dump shows all 12 tasks were pre-created simultaneously in scrambled order, with exp_04 and exp_05 tasks (ids 7 and 9) submitted before exp_01 tasks (ids 1 and 2):

[Task submitter] Submitted process > CLIENT (4) ← exp_04, task id=7
[Task submitter] Submitted process > SERVER (5) ← exp_05, task id=9

%% executor local > tasks in the submission queue:
~> id: 5; CLIENT (3)
~> id: 4; CLIENT (2)
~> id: 3; SERVER (2)
~> id: 6; SERVER (3)
~> id: 1; SERVER (1) ← exp_01, never reached
~> id: 10; CLIENT (5)
~> id: 12; CLIENT (6)
~> id: 8; SERVER (4)
~> id: 2; CLIENT (1) ← exp_01, never reached
~> id: 11; SERVER (6)

.nextflow.log attached.

Environment

• Nextflow version: 25.04.6 (build 5954, 01-07-2025)
• Java version: OpenJDK 64-Bit Server VM 17.0.7+4-jvmci-23.0-b10
• Operating system: Linux 6.12.85+deb13-amd64
• Bash version: GNU bash 5.2.37(1)-release (x86_64-pc-linux-gnu)

Additional context

toSortedList + flatMap makes all N items available in the channel simultaneously. Nextflow's task processor then creates all N tasks concurrently via its thread pool, assigning task IDs non-deterministically by thread scheduling — so lower-indexed channel items do not get lower task IDs. fair true does not prevent this because it governs round-robin input selection across multiple input channels, not FIFO submission order when all tasks are created simultaneously in a race.

The use case driving this report is two processes that must coordinate via IPC (one acts as an HTTP model server, the other as a client/sampler) and therefore must run concurrently on the same key. They cannot be merged into a single process because they require different execution environments (different conda/container environments). Is there a supported Nextflow idiom to guarantee that two processes consuming paired channels always submit their Nth tasks concurrently?

[thealanjason]

I am able to avoid the race condition of the Task Scheduler by adding a delay to the queue channel.

    server_inputs = _split.server
        .toSortedList { a, b -> a[0] <=> b[0] }                                                                                         
        .flatMap { it }
        .map { item -> Thread.sleep(500); item }                                                                                                         
    client_inputs = _split.client
        .toSortedList { a, b -> a[0] <=> b[0] }                                                                                         
        .flatMap { it }
        .map { item -> Thread.sleep(500); item }

Does this make sense?

[adamrtalbot]

The answer is no. Nextflow is designed around isolated, concurrent tasks that only communicate via inputs and outputs. Synchronising processes with maxForks and fair forces Nextflow into a role it wasn't built for. Note that fair only orders outputs to match inputs; it has nothing to do with submission.

That said, your sleep example works because the blocking operation forces Nextflow to wait for inputs to arrive. Below, I use a process with a random sleep duration. The outputs of SLEEP are passed to both downstream processes (SERVER and CLIENT) at the same time. This is incidental: the inputs pop out of SLEEP with enough gaps that Nextflow looks pseudo-synchronised. If two SLEEP tasks happened to wait the same length of time, results would get mixed up again. Without SLEEP, you're back to the original case: everything is sent at once and it's first-come-first-served.

process SLEEP {
    tag "${key}"

    input:
    val key

    output:
    val key, emit: sleep_out

    script:
    def random = new Random()
    def sleepTime = random.nextInt(10) + 1
    """
    sleep ${sleepTime}
    echo "${key} slept for ${sleepTime} seconds"
    """
}

process SERVER {

    tag "${key}"

    maxForks 1

    input:
    tuple val(key), path(comm_dir)

    script:
    """
    sleep 5
    echo ${key}
    """
}

process CLIENT {

    tag "${key}"

    maxForks 1

    input:
    tuple val(key), path(comm_dir)

    output:
    val key

    script:
    """
    sleep 5
    echo ${key}
    """
}

workflow {
    def N_EXPERIMENTS = 6

    keys = channel.of(1..N_EXPERIMENTS)

    ch_keys_with_delay = SLEEP(keys)
    ch_paths = ch_keys_with_delay.sleep_out
                        .map { f -> ["experiment_${f}", file("/tmp/mwe_ipc_${workflow.sessionId}/experiment_${f}")] }

     SERVER(ch_paths)
     CLIENT(ch_paths)
}

To coordinate multiple channels, you could use a blocking operator like .join() to hold task submission until inputs are ready.

Here, I sleep the same channel twice with random gaps, so SLEEP_A and SLEEP_B emit out of sync. .join() pairs them by key and emits each tuple on a match. The synchronisation is again incidental, relying on the join to force a block.

process SLEEP_A {
    tag "${key}"

    input:
    val key

    output:
    val key, emit: sleep_out

    script:
    def random = new Random()
    def sleepTime = random.nextInt(10) + 1
    """
    sleep ${sleepTime}
    echo "${key} slept for ${sleepTime} seconds"
    """
}

process SLEEP_B {
    tag "${key}"

    input:
    val key

    output:
    val key, emit: sleep_out

    script:
    def random = new Random()
    def sleepTime = random.nextInt(10) + 1
    """
    sleep ${sleepTime}
    echo "${key} slept for ${sleepTime} seconds"
    """
}

process SERVER {

    tag "${key}"

    input:
    tuple val(key), path(comm_dir)

    script:
    """
    sleep 5
    echo ${key}
    """
}

process CLIENT {

    tag "${key}"

    input:
    tuple val(key), path(comm_dir)

    output:
    val key

    script:
    """
    sleep 5
    echo ${key}
    """
}

workflow {
    def N_EXPERIMENTS = 6

    keys = channel.of(1..N_EXPERIMENTS)

    ch_keys_a_with_delay = SLEEP_A(keys)
    ch_a_paths = ch_keys_a_with_delay.sleep_out
                        .map { f -> ["experiment_${f}", file("/tmp/mwe_ipc_${workflow.sessionId}/experiment_${f}")] }

    ch_keys_b_with_delay = SLEEP_B(keys)
    ch_b_paths = ch_keys_b_with_delay.sleep_out
                        .map { f -> ["experiment_${f}", file("/tmp/mwe_ipc_${workflow.sessionId}/experiment_${f}")] }

    ch_synced_paths = ch_a_paths
                        .join(ch_b_paths, by: 0)

     SERVER(ch_synced_paths)
     CLIENT(ch_synced_paths)
}

And we can break this again with groupTuple/groupBy which will wait for all inputs to be emitted:

ch_synced_paths = ch_a_paths
                        .join(ch_b_paths, by: 0)
                        .groupTuple() // wait for everything to be emitted and group by the item in index 0
                        .map { key, file1, file2 -> [key, file1] } // drop duplicated file

To summarise:

maxForks 1 + fair true do not guarantee FIFO task submission order when channel emits all items simultaneously

No, by design they do not.

Is there a supported Nextflow idiom to guarantee that two processes consuming paired channels always submit their Nth tasks concurrently?

Not really. You can force it with a blocking operation, but I'd avoid the pattern.

Your best options:

• Combine them into one Nextflow process and launch both environments inside the task. This is the only way Nextflow can schedule the pair atomically.
• Make it sequential. Have SERVER produce an output that CLIENT consumes. This removes the concurrent IPC, so it only works if the protocol allows.
• Make the processes tolerant of asynchrony, then you can submit them all at once.

[thealanjason]

Hi @adamrtalbot thanks for your response :)

I made this issue after considering the suggested best options (which I already got using Claude 😉)

I understand that Nextflow uses a data flow paradigm and is not built for such a task. But I think it is possible, if the tasks avaialble for execution are stored in a Queue type buffer instead of a HashMap type buffer.

• Combine them into one Nextflow process and launch both environments inside the task. This is the only way Nextflow can schedule the pair atomically.

This is not possible for me, since I would like the SERVER and CLIENT to have different computational environments (conda, docker, apptainer, etc.) and I cannot combine them into one process.

• Make it sequential. Have SERVER produce an output that CLIENT consumes. This removes the concurrent IPC, so it only works if the protocol allows.

In this case the SERVER process (task) would have to complete. I can achieve this by creating an orphan process with the & operator in the bash script when running the server, but it is then not taken care of when a SIGTERM or SIGKILL, etc is passed to nextflow.

• Make the processes tolerant of asynchrony, then you can submit them all at once.

I did, by actually using the key. The issue is that I would like to limit the number of (SERVER and CLIENT) task pairs based on the number of resources available. This maximum number would then be set with the help of maxForks.
Here is when things start to go south. The nextflow scheduler schedules tasks in the order that they were made ready by the FIFO queue but, the executor just randomly picks a task irrespective of the spawning order (which I don't understand why?).

Hence I would like to argue that it is incorrect to say that Nextflow queue channels are FIFO. Because they are only FIFO until they reach the barrier of a Process. Here they get accumulated and even though the are spawned with correct incremental task IDs, the executors just randomly pick tasks from a basket of spawned tasks (which is a sad). It would be nice if (even though the execution of tasks is asynchronous), at least the order of picking from the available tasks is maintained by using a Queue type buffer with Queue.pop(), and not a HashMap.pop().

@bentsherman @adamrtalbot what are your thoughts?

[adamrtalbot]

The issue is that I would like to limit the number of (SERVER and CLIENT) task pairs based on the number of resources available. This maximum number would then be set with the help of maxForks.

If CLIENT is lightweight, you could submit all of them at once and only let them exit when the matching SERVER is scheduled and run, as fast as the resources will allow?

[thealanjason]

This might work, let me try :) Nevertheless, it's still a workaround. The executor picking any of the tasks at random (and not in the order that they were scheduled by the scheduler) is what bothers me, and I'd still call it a bug.

…

On Thu, May 14, 2026 at 5:14 PM Adam Talbot ***@***.***> wrote: *adamrtalbot* left a comment (nextflow-io/nextflow#7117) <#7117 (comment)> The issue is that I would like to limit the number of (SERVER and CLIENT) task pairs based on the number of resources available. This maximum number would then be set with the help of maxForks. If CLIENT is lightweight, you could submit all of them at once and only let them exit when the matching SERVER is scheduled and run, as fast as the resources will allow? — Reply to this email directly, view it on GitHub <#7117 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AJX55NSKIDOML2MRUGMRNEL42XPGNAVCNFSM6AAAAACYUYMTOKVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHM2DINJRHE4TENZYGE> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>. You are receiving this because you authored the thread.Message ID: ***@***.***>