Hi there. New to Nextflow and trying to understand how to best approach it, cause had already a couple of first failed attempts. First experiment started with a workflow that would handle things string by string, one string at a time (kind of). And the workflow alone looked like this:

workflow {
    main:
    printPythonVersions()

    def repo_root = projectDir.parent

    def cells_channel = channel.fromPath(params.lastpass_export, checkIfExists: true)
        .splitCsv()
        .flatten()
        // .take(params.take_only) //TODO REMOVE DEMO
        .view { cell -> "Processing cell: ${cell}" }

    def base_tokens = tokenizeLine(params.tokenize_bases_script, repo_root, cells_channel)
        .splitText()
        .map { line -> line.trim() } // makes the stdout into a channel of lines
        .filter { token -> !token.isEmpty() }
        .unique()

    def case_variants_tokens = case_variants(params.tokenize_cases_script, repo_root, base_tokens)
        .splitText().map { line -> line.trim() }

    def one_leet_variants_tokens = one_leet_variants(params.one_leet_variants_script, repo_root, base_tokens)
        .splitText().map { line -> line.trim() }

    def commonAffixesChannel = channel.fromList(get_common_affixes())

    def prefixed_tokens = base_tokens.combine(commonAffixesChannel).map { token, affix -> "${affix}${token}" }
    def suffixed_tokens = base_tokens.combine(commonAffixesChannel).map { token, affix -> "${token}${affix}" }

    def all_candidates_file = base_tokens
        .mix(case_variants_tokens)
        .mix(one_leet_variants_tokens)
        .mix(prefixed_tokens)
        .mix(suffixed_tokens)
        .filter { candidate -> max_length_filter(candidate) } // Example: filter out candidates longer than 128 characters
        .unique()
        .collectFile(name: "all_candidates_from_nf.txt", newLine: true)

    publish:
    all_candidates = all_candidates_file
 }

The result of this first experiment would be to burn all of the laptop's resources for 4 hours and then, by spinning hundrends of thousands of tasks, etc. leading to an out of memory. In a second experiment I tried to redesign the whole thing to work in batches:

workflow {
    main:
    printPythonVersions()

    def repo_root = projectDir.parent

    def cells_channel = channel.fromPath(params.lastpass_export, checkIfExists: true)
        .splitCsv()
        .flatten()
        // .take(params.take_only) //TODO REMOVE DEMO
        // .view { cell -> "Processing cell: ${cell}" } //too slow in full run

    def base_tokens = tokenizeLine(params.tokenize_bases_script, repo_root, cells_channel)
        .splitText()
        .map { line -> line.trim() } // makes the stdout into a channel of lines
        .filter { token -> !token.isEmpty() }
        .unique()

    def base_batches = base_tokens
        .collate(params.batch_size)
        .map { batch -> batch.join('\n') + '\n' }

    def base_parts = write_batch_parts(base_batches)
    // def base_file = base_parts.collectFile(name: 'base_parts_from_nf.txt', newLine: false)

    def case_variants_parts = case_variants_batches(params.tokenize_cases_script, repo_root, base_batches)
    // def case_variants_file = case_variants_parts.collectFile(name: 'case_variants.txt', newLine: false)

    def one_leet_variants_parts = one_leet_variants_batches(params.one_leet_variants_script, repo_root, base_batches)
    // one_leet_variants_parts
    //     .count()
    //     .view { cnt -> "one_leet_variants_parts count: ${cnt}" }
    // def one_leet_variants_file = one_leet_variants_parts.collectFile(name: 'one_leet_variants.txt', newLine: false)

    def common_affixes = get_common_affixes()

    def prefixed_parts = prefixed_batches(base_batches, common_affixes, true)
    // def prefixed_file = prefixed_parts.collectFile(name: 'prefixed_variants.txt', newLine: false)

    def suffixed_parts = suffixed_batches(base_batches, common_affixes, false)
    // def suffixed_file = suffixed_parts.collectFile(name: 'suffixed_variants.txt', newLine: false)

    def all_parts = base_parts
        .mix(case_variants_parts)
        .mix(one_leet_variants_parts)
        .mix(prefixed_parts)
        .mix(suffixed_parts)

    def all_file = all_parts.collectFile(name: 'all_candidates_from_nf.txt', newLine: false)

    all_final = filter_by_length_and_make_unique(all_file)

    // Unlike mix(), concat() preserves source order: all items from file1, then all items from file2.
    def files =
        // base_file
        // .concat(case_variants_file)
        // .concat(one_leet_variants_file)
        // .concat(prefixed_file)
        // .concat(suffixed_file)
        // .concat(all_file)
        all_file
        .concat(all_final)
        .map { file -> tuple(file.baseName, file) }

    count_lines(files)
        .view { tpl -> "file name ${tpl[0]} with count: ${tpl[1].trim()}" }

    publish:
    all_candidates = all_file
    pure_candidates = all_final
    // basic = base_file
    // case_variants = case_variants_file
    // one_leet_variants = one_leet_variants_file
    // prefixed = prefixed_file
    // suffixed = suffixed_file
 }

This time it executed for several hours but completed successfully without any issues with batches of count 1000. This was more of a toy-experiment as the whole thing could have been coded in a single Python file, which made it as fast as complete execution in under 30 minutes and the python was single threaded .... So the genuine questions from this first experience: 1. Why Nextflow does not manage itself the load that it gets and requires me to engineer it in batches ? 2. In what kind of scenarios to think of nextflow to actually take advantage its capacity to run in parallel ? I understand that splitting the python in mini python scripts might not have been the best idea of all times 🙂

For this kind of job (not having really read your code in much detail but going on your description) - doing sting manipulation, yup, I would probably also use Python 😅

yes there were process declarations, but only shared bit of the code to not overflow. Yeah probably need to go back in training and see real use cases to understand nextflow's advantages.

Most people use either: docker, singularity / apptainer or conda / pixi / mamba. Those are the big 3.