限制 Nextflow 工作流中单个流程的数量

Question

我有以下简单的工作流程：

workflow {

  Channel.fromPath(params.file_list)
        .splitText(){it.trim()}
        .set { file_list }

  data = GetFromHPSS(file_list)
  data_pairs = CoupleDETXToFile(data, file(params.detx_path))
  SingleDUTimeResFit(data_pairs)

}

其中 file_list 是磁带机系统上的路径列表。 GetFromHPSS 是从磁带系统检索文件的进程，我需要将并行进程限制在相当低的数量。

目前，我正在使用

executor {
  queueSize = 100
}

在配置文件中但是有两个问题：

它限制了并行作业的总体最大数量，而我可以运行数千个 SingleDUTimeResFit 并行进程
它总是先等到它处理完 GetFromHPSS 中的所有内容，而不是继续进行后续处理

这是一个例子：

N E X T F L O W  ~  version 21.04.3
Launching `workflows/singledu_timeresfit.nf` [wise_galileo] - revision: 8084ac1482
executor >  sge (502)
[13/ca3e8a] process > GetFromHPSS (426)  [ 18%] 402 of 22840
[-        ] process > CoupleDETXToFile   [  0%] 0 of 402
[-        ] process > SingleDUTimeResFit -

有没有办法限制 GetFromHPSS 特定数量的并行执行并让剩余的进程运行设置另一个队列限制？

编辑：我猜这是我最好的尝试之一，但它不接受配置：


process {
  executor {
    queueSize = 100
    submitRateLimit = "10sec"
  }

  withName: GetFromHPSS {
    executor.queueSize = 10
  }
}

有了这个 process 顶级配置，我得到：

N E X T F L O W  ~  version 21.04.3
Launching `workflows/singledu_timeresfit.nf` [confident_pasteur] - revision: 8084ac1482
Unknown config attribute `process.withName:GetFromHPSS` -- check config file: /sps/km3net/users/tgal/dev/PhD/workflows/nextflow.config

Answer 1

我认为您在这里寻找的是 maxForks 指令，它可以仅应用于 'GetFromHPSS' 进程，而无需更改执行程序的 queueSize:

process 'GetFromHPSS' {

    maxForks 1

    """
    <your script here>
    """
}

如果您认为有意义，您甚至可以对其进行参数化：

params.hpss_forks = 5

process 'GetFromHPSS' {

    maxForks params.hpss_forks

    """
    <your script here>
    """
}

限制 Nextflow 工作流中单个流程的数量

Limit the number of single processes in Nextflow workflows

workflow

nextflow