wait for previous process to finish

[davidaray]

This may have been asked previously but I can't find the topic if it has been.

How does one hold a process until the previous processes have completed?

In the workflow below, I need the final process, TEAID_END, to wait for all output files from all previous processes are complete before executing. As currently written, TEAID_END starts while the previous processes are running and, as a result, is archiving output before it's ready.

The 'when' option (https://www.nextflow.io/docs/latest/process.html#when) looks like it could be helpful but I'm unable to figure out how to implement it.

#!/usr/bin/env nextflow
 
/*
 * Defines the pipeline inputs parameters (giving a default value for each for them)
 * Each of the following parameters can be specified as command line options
 */

params.NAME = "iRic1"
params.MINORF = 500
params.MINCOPY = 10
params.WORKDIR = "/lustre/scratch/daray/ixodes/iRic1_nextflow"
params.TARGET = "iRic1_extended_rep.fa.classified"
params.EXTENSIONSDIR = "/lustre/scratch/daray/ixodes/iRic1_nextflow/extensions"
params.PFAM = "/lustre/work/daray/software/pfam_db"
params.AIDPATH = "/lustre/work/daray/software/TE-Aid"
params.ASSEMBLIESDIR = "/lustre/scratch/daray/ixodes/assemblies"
params.AIDOUT = "/lustre/scratch/daray/ixodes/iRic1_nextflow/te-aid"

/*
 * Call TEcurate_setup to generate the initial table and output directories.
 */

process TECURATE_SETUP { 

	output: 
	stdout into SETUP

	shell:
	"""
	#!/bin/bash
		
	sh $params.WORKDIR/TEcurate_setup.sh \
		$params.NAME \
		$params.MINORF \
		$params.MINCOPY \
		$params.WORKDIR \
		$params.TARGET \
		$params.EXTENSIONSDIR \
		$params.PFAM \
		$params.AIDPATH \
		$params.ASSEMBLIESDIR \
	"""
	}

/*
 * Call detectCOUNTS.sh to generate the processing files to generate array jobs.
 */


process DETECTCOUNTS {
	input:
	stdin from SETUP
	
	output:
	stdout into DETECTCOUNTS
	
	shell:
	"""
	#!/bin/bash
	
	sh $params.WORKDIR/detectCOUNTS.sh \
		$params.NAME \
		$params.WORKDIR 
		
	"""
	}

/*
 *Call RC_array.sh to check orientation of ORF-containing hits and 
 *reverse complement if necessary using array jobs.
 */

process RC_ARRAY {
	input:
	stdin from DETECTCOUNTS
		
	output:
	stdout into RC

	shell:
	"""
	#!/bin/bash 
	
	sbatch $params.WORKDIR/prioritize/RC_array_LINE.sh \
		$params.NAME \
		$params.WORKDIR \
		$params.AIDOUT \
		$params.AIDPATH 
	sbatch $params.WORKDIR/prioritize/RC_array_DNA.sh \
		$params.NAME \
		$params.WORKDIR \
		$params.AIDOUT \
		$params.AIDPATH 
	sbatch $params.WORKDIR/prioritize/RC_array_LTR.sh \
		$params.NAME \
		$params.WORKDIR \
		$params.AIDOUT \
		$params.AIDPATH 

	"""
	}

/*
 *Call TE-aid1_array.sh to run TE-Aid and other housekeeping.
 */

process TEAID1_ARRAY {
	input:
	stdin from RC
		
	output:
	stdout into AID_ONE

	shell:
	"""
	#!/bin/bash 
	
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_LINE.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid1_array_LINE.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_DNA.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid1_array_DNA.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_LTR.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid1_array_LTR.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_NOHIT.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid1_array_NOHIT.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_SINE.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid1_array_SINE.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_RC.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid1_array_RC.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	"""
	}

/*
 *Call TE-aid2_array.sh to run TE-Aid and other housekeeping on files with no orientation.
 */

process TEAID2_ARRAY {
	input:
	stdin from AID_ONE
		
	output:
	stdout into AID_TWO

	shell:
	"""
	#!/bin/bash 
	
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_LINE.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid2_array_LINE.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_DNA.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid2_array_DNA.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_LTR.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid2_array_LTR.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_SINE.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid2_array_SINE.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	if [ -f $params.WORKDIR/prioritize/TE-aid1_array_RC.sh ]; then
		sbatch $params.WORKDIR/prioritize/TE-aid2_array_RC.sh \
			$params.NAME \
			$params.WORKDIR \
			$params.AIDOUT \
			$params.AIDPATH \
			$params.ASSEMBLIESDIR
	fi
	"""
	}

/*
 *Call TE-aid_end.sh to filter and generate output archives.
 */

process TEAID_END {
	input:
	stdin from AID_ONE
		
	output:
	stdout into END

	shell:
	"""
	#!/bin/bash 
	
	sbatch $params.WORKDIR/TEcurate_end.sh \
		$params.NAME \
		$params.WORKDIR \
		$params.AIDOUT
	"""
	}

[lbrandhoff-w]

Although its probably a bit hacky using the each qualifier worked for me in a similar scenario. Using it in the declaration of the input channel for TEAID_END will wait for all upstream processes of that channel to finish before running the first TEAID_END instance.

Although you might have to transform the AID_ONE channel to use each but Im not sure.

[bentsherman]

You need to use channels to create dependencies between the upstream process and your final process. For example, feed the output channels of each upstream process into a collect operator and feed that to the final process. That way, it won't start until all the upstream processes are finished.

If the final process doesn't actually need the upstream outputs, you can also create a state dependency as shown in the patterns docs.

[Rohit-Satyam]

I Use collect to serialize the process execution. An exemplary case is when we have to run fastqc first on all samples and then run multiqc on all the produced html and zip files. The following code waits for FASTQC process to finish first and then execute MULTIQC process.

nextflow.enable.dsl=2
params.raw = "data/*.fastq.gz"
params.outdir="results/01_rawfastqc"
process FASTQC {
	publishDir "$params.outdir", mode: 'copy'
	input:
	path reads
	output:
	file "*"
	script:
	"""
	fastqc -t 10 ${reads}
	"""
}

process MULTIQC {
	publishDir "$params.outdir", mode: 'copy'
	input:
	file x
	output:
	file "multiqc_report.html"
	file "multiqc_data"

    script:
    """
    multiqc -f $x
    """
}



workflow {
 Channel.fromPath(params.raw, checkIfExists: true )| FASTQC | collect | MULTIQC
  
}