How to wait for process to finish before starting the next

[Apompetti-Cori]

Hello I have a problem which doesn't seem to be answered well by the scatter gather solution I've found in prior discussions.

So I'm running trim_galore on raw fastq file which are then fed into bismark for alignment. I'd like to wait for the entirety of the trim_galore step to finish so that I don't hog all the resources by running alignment and trimming at the same time.

This is the setup of my pipeline.

#!/usr/bin/env nextflow
/*
Enable Nextflow DSL2
*/
nextflow.enable.dsl=2

params.fastq_folder = "${workflow.projectDir}/fastq"
params.reads = "${params.fastq_folder}/*{_R,_}{1,2}*.{fastq,fq}.gz"
params.singleEnd = false

Channel
.fromFilePairs(params.reads, size: params.singleEnd ? 1 : 2)
.ifEmpty {exit 1, "Cannot find any reads matching ${params.reads}\nNB: Path needs to be enclosed in quotes!\nIf this is single-end data, please specify --singleEnd on the command line."}
.set{ reads_ch }

process trim_galore {
    maxForks 4
    memory '8 GB'
    cpus 2

    publishDir "./results/trim_galore", mode: 'copy'

    input:
    tuple val(file_id), path(reads)

    output:
    tuple val(file_id), path("*.fq.gz")
    path("*trimming_report.txt"), emit: trimming_report

    script:
    def singleEnd = params.singleEnd ? '' : '--paired'
    def rrbs = params.rrbs ? '--rrbs' : ''

    """
    trim_galore \
    ${singleEnd} \
    ${rrbs} \
    --length 35 \
    --quality 28 \
    --phred33 \
    --cores ${task.cpus} \
    $reads
    """
}

process bismark_align {
    maxForks 4
    memory '40 GB'
    cpus 8
    
    publishDir "./results/bismark_align", mode: 'copy'

    input:
    tuple val(file_id), path(reads)

    output:
    tuple val(file_id), path("*.bam")
    path("*report.txt"), emit: align_report

    script:
    """
    bismark --bowtie2 --parallel ${task.cpus} ${params.db} -1 ${reads[0]} -2 ${reads[1]}
    """
}

workflow {
trim_galore(reads_ch)
bismark_align(trim_galore.out[0])
}

What I'd like ideally is for trim_galore to finish before allowing bismark to start on all the processed files.

From what I see is that if I do a collect on the trim_galore.out, it does end up waiting for the process to finish but the output is just one big list that looks roughly like this:

[file_id1, [file_id1_val_1.fq.gz, file_id1_val_2.fq.gz], file_id2, [file_id2_val_1.fq.gz, file_id2_val_2.fq.gz]]

I'm not sure exactly how I can get this list to be read into the bismark_align process so that it submits 1 process for each file_id/val_(1,2) pair.

I believe I want some sort of state dependency implemented simply but I'm not entirely sure how.

Please let me know if you need further clarification and thank you for any help you can provide.

[mribeirodantas]

If you collect with flat: false and use the flatMap channel operator afterward you should get the waiting desired effect and at the same time have the channel in its original output structure. Check the example below:

process FOO {
  input:
    val x
  output:
    tuple val(y), val(z)
  exec:
    y = x+1
    z = y*2
}

workflow {
  Channel
    .of(1..10)
    | FOO
    | collect(flat: false)
    | flatMap
    | view
}

Output:

N E X T F L O W  ~  version 22.10.6
Launching `hehe.nf` [mighty_ardinghelli] DSL2 - revision: 72de2b765e
executor >  local (10)
[23/ec3933] process > FOO (10) [100%] 10 of 10 ✔
[8, 16]
[3, 6]
[7, 14]
[4, 8]
[9, 18]
[5, 10]
[6, 12]
[2, 4]
[10, 20]
[11, 22]

[mribeirodantas]

An even better example is with 2 processes and using sleep so that you can see when BAR awaits till FOO is finished or not. Play with the code below by uncommenting the two commented lines at the end or not.

process FOO {
  input:
    val x
  output:
    tuple val(y), val(z)
  exec:
    y = x+1
    z = y*2
    sleep 1000
}

process BAR {
  debug true
  input:
    tuple val(y), val(z)
  output:
    stdout
  script:
    """
    echo [$y, $z]
    """
}

workflow {
  Channel
    .of(1..10)
    | FOO
 //   | collect(flat: false)
 //   | flatMap
    | BAR
}

[Apompetti-Cori]

This worked perfectly in changing collect from a single emission to multiple emissions. Thank you!

[mribeirodantas]

You're welcome 😉

[yinshiyi]

@mribeirodantas you are a genius, thank you so much