From c3080b3f166ba058b497fa64931f127a3f0414c1 Mon Sep 17 00:00:00 2001 From: Glenn Hickey Date: Thu, 23 Jul 2026 11:47:14 -0400 Subject: [PATCH 1/2] cactus-panpatch: call the shared setup functions in main() cactus-panpatch's main() was missing the standard per-tool boilerplate that every other cactus entry point runs after parsing options: - cactus_override_toil_options(options): sets toil defaults (retryCount, realtime logging, default memory) and exports CACTUS_MAX_MEMORY / CACTUS_DEFAULT_MEMORY. panpatch calls cactus_clamp_memory(), which reads those env vars, so without this it would KeyError. - setupBinaries(options): configures binariesMode/docker so cactus_call works as requested (panpatch runs vg, bgzip, etc. via cactus_call). - set_logging_from_options(options) and enableDumpStack(): consistent logging setup and debug stack dumps. Co-Authored-By: Claude Opus 4.8 --- src/cactus/refmap/cactus_panpatch.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/cactus/refmap/cactus_panpatch.py b/src/cactus/refmap/cactus_panpatch.py index 90fe96b04..e158edc47 100644 --- a/src/cactus/refmap/cactus_panpatch.py +++ b/src/cactus/refmap/cactus_panpatch.py @@ -34,6 +34,9 @@ from cactus.shared.common import cactus_clamp_memory from cactus.shared.common import delete_directory from cactus.shared.common import getOptionalAttrib, findRequiredNode +from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import setupBinaries +from cactus.shared.common import enableDumpStack from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.version import cactus_commit from cactus.refmap.cactus_minigraph import check_sample_names @@ -42,7 +45,7 @@ from cactus.refmap.cactus_pangenome import pangenome_config_overrides, pangenome_end_to_end_workflow from toil.job import Job from toil.common import Toil -from toil.statsAndLogging import logger +from toil.statsAndLogging import logger, set_logging_from_options from toil.realtimeLogger import RealtimeLogger def main(): @@ -107,6 +110,10 @@ def main(): options = parser.parse_args() + setupBinaries(options) + set_logging_from_options(options) + enableDumpStack() + panpatch_validate_options(options) # we need the minigraph event name to filter it out of the seqfiles below @@ -137,6 +144,9 @@ def main(): # out of the join workflow before any clipping or filtering disable_pangenome_outputs(options) + # Mess with some toil options to create useful defaults. + cactus_override_toil_options(options) + logger.info('Cactus Command: {}'.format(' '.join(sys.argv))) logger.info('Cactus Commit: {}'.format(cactus_commit)) for run in runs: From f9e94fafe374e132e57185bd918786d0850d1606 Mon Sep 17 00:00:00 2001 From: Glenn Hickey Date: Thu, 23 Jul 2026 11:59:25 -0400 Subject: [PATCH 2/2] Add per-job "fast" walltime to route small coordination jobs to a fast Slurm partition Newer Toil exposes a per-job `walltime` requirement that its Slurm backend uses to pick a partition (and set --time). This gives Cactus's many small "coordination" jobs a short walltime so Slurm routes them to a fast/short partition, while everything else falls back to Toil's --defaultWalltime. Machinery (src/cactus/shared/common.py): - cactus_fast_walltime(): reads CACTUS_FAST_WALLTIME (mirrors cactus_clamp_memory). - add_cactus_toil_options(): adds --fastWalltime (default 1800 = 30 min, very conservative for these jobs; 0 disables). - cactus_override_toil_options() exports CACTUS_FAST_WALLTIME so it reaches workers. - RoundedJob / ChildTreeJob now forward walltime to Toil. Application: - Wired add_cactus_toil_options(parser) into every tool's main(). - Tagged the small coordination jobs' scheduling call sites with walltime=cactus_fast_walltime() (the FAST set plus a hand-picked subset of the ambiguous ones; compute-heavy jobs left on --defaultWalltime). Docs (doc/progressive.md, doc/pangenome.md): recommend --defaultWalltime (+ the new --fastWalltime) instead of --slurmTime, which is now framed as a global override. IMPORTANT: toil-requirement.txt is temporarily pinned to Toil master because per-job walltime is not in a Toil release yet. Revert to a released toil[aws]== before merging. Co-Authored-By: Claude Opus 4.8 --- doc/pangenome.md | 2 +- doc/progressive.md | 14 ++--- src/cactus/blast/cactus_blast.py | 7 +-- src/cactus/maf/cactus_hal2chains.py | 11 ++-- src/cactus/maf/cactus_hal2maf.py | 19 +++---- src/cactus/maf/cactus_maf2bigmaf.py | 7 +-- src/cactus/maf/cactus_phast.py | 29 ++++++----- src/cactus/paf/local_alignment.py | 20 ++++---- .../preprocessor/cactus_preprocessor.py | 15 +++--- src/cactus/preprocessor/unmasking.py | 4 +- src/cactus/progressive/cactus_prepare.py | 5 +- src/cactus/progressive/cactus_progressive.py | 27 +++++----- src/cactus/refmap/cactus_graphmap.py | 15 +++--- src/cactus/refmap/cactus_graphmap_join.py | 17 ++++--- src/cactus/refmap/cactus_graphmap_split.py | 17 ++++--- src/cactus/refmap/cactus_minigraph.py | 19 +++---- src/cactus/refmap/cactus_pangenome.py | 51 ++++++++++--------- src/cactus/refmap/cactus_panpatch.py | 15 +++--- src/cactus/refmap/cactus_refmap.py | 13 ++--- src/cactus/setup/cactus_align.py | 18 ++++--- src/cactus/shared/common.py | 39 ++++++++++++-- src/cactus/update/cactus_hal2seqfile.py | 7 +-- toil-requirement.txt | 4 +- 23 files changed, 215 insertions(+), 160 deletions(-) diff --git a/doc/pangenome.md b/doc/pangenome.md index a6f29b944..f9240ee4f 100644 --- a/doc/pangenome.md +++ b/doc/pangenome.md @@ -341,7 +341,7 @@ cactus-panpatch ./js chromfile.txt --outDir patched --batch ``` cactus-panpatch ./js chromfile.txt --outDir patched --batch --requireTelomeres \ - --batchSystem slurm --slurmTime 10:00:00 --doubleMem true --maxMemory 1.5T --retryCount 10 \ + --batchSystem slurm --defaultWalltime 36000 --doubleMem true --maxMemory 1.5T --retryCount 10 \ --consCores 16 --mgCores 32 --mapCores 16 ``` diff --git a/doc/progressive.md b/doc/progressive.md index e28220263..97dd5e8df 100644 --- a/doc/progressive.md +++ b/doc/progressive.md @@ -231,7 +231,7 @@ cactus-hal2maf ./js evolverMammals.hal evolverMammals.maf.gz --refGenome simHuma Exporting a MAF for each reference in an 8-way [ape alignment](https://cglgenomics.ucsc.edu/february-2024-t2t-apes/) on UCSC Slurm cluster: ``` -for i in hs1 hg38 GCA_028858775.2 GCA_028885655.2 GCA_028885625.2 GCA_028878055.2 GCA_029281585.2 GCA_029289425.2; do cactus-hal2maf ./js_hal2maf8 ./8-t2t-apes-2023v2.hal ./8-t2t-apes-2023v2.${i}.maf.gz --filterGapCausingDupes --outType norm single --refGenome $i --chunkSize 500000 --batchCores 64 --noAncestors --batchCount 16 --batchSystem slurm --logFile ./8-t2t-apes-2023v2.${i}.gz.log --batchLogsDir batch-logs-8apes --slurmTime 200:00:00 --slurmPartition long;done +for i in hs1 hg38 GCA_028858775.2 GCA_028885655.2 GCA_028885625.2 GCA_028878055.2 GCA_029281585.2 GCA_029289425.2; do cactus-hal2maf ./js_hal2maf8 ./8-t2t-apes-2023v2.hal ./8-t2t-apes-2023v2.${i}.maf.gz --filterGapCausingDupes --outType norm single --refGenome $i --chunkSize 500000 --batchCores 64 --noAncestors --batchCount 16 --batchSystem slurm --logFile ./8-t2t-apes-2023v2.${i}.gz.log --batchLogsDir batch-logs-8apes --defaultWalltime 720000 --slurmPartition long;done ``` Note that this invocation creates two MAFs per reference (as dictated by the `--outType` option). Notably, the `.single.maf.gz` files will be filtered so that each genome appears at most once per block, which is often required by browsers and other MAF-reading tools. @@ -342,7 +342,7 @@ cactus-phast ./js-vgp \ --geneAnnotation https://hgdownload.soe.ucsc.edu/goldenPath/hg38/database/ncbiRefSeq.txt.gz \ --bigwig --batchSystem slurm --chunkCores 32 --phyloFitCores 32 \ --substMod REV --modFreqs --precision HIGH \ - --slurmPartition medium --slurmTime 10:00:00 \ + --slurmPartition medium --defaultWalltime 36000 \ --doubleMem true ``` @@ -416,8 +416,10 @@ These are the most relevant options for running on a cluster On a cluster with partitions and/or time limits, make sure to use -* `--slurmTime` to specify the time for each job. Unfortunately cactus does not yet try to set this itself, so you need to give one value that will be applied to all jobs, ex `--slurmTime 200:00:00` -* `--slurmPartition / --slurmGPUPartition` to specify the slurm partition where CPU / GPU jobs end up on. Cactus will try to figure this out on its own using the `--slurmTime` value along with whether or not the job needs GPU. But this option will allow you to override that. +* `--defaultWalltime` to specify the time, **in seconds**, for each job, ex `--defaultWalltime 720000` (200 hours). Slurm uses each job's walltime to choose a partition, so give the big jobs plenty of time here. +* `--fastWalltime` sets the walltime, **in seconds**, that Cactus gives its many small "coordination" jobs; it defaults to `1800` (30 minutes), which is very conservative for these jobs. On a cluster with a fast/short partition, this lets those little jobs get routed there instead of tying up a long partition; everything else falls back to `--defaultWalltime`. Pass `--fastWalltime 0` to disable it. +* `--slurmPartition / --slurmGPUPartition` to specify the slurm partition where CPU / GPU jobs end up on. Cactus will try to figure this out on its own from each job's walltime along with whether or not the job needs GPU. But this option will allow you to override that. +* `--slurmTime` is a global override that forces a single time onto *every* job, ignoring the per-job `--defaultWalltime` / `--fastWalltime` values above (so it also disables the fast-partition routing). Reach for it only as an escape hatch, e.g. if a job's time estimate turns out too low and it keeps getting killed: `--slurmTime 200:00:00`. You can also use @@ -436,13 +438,13 @@ source /private/groups/cgl/cactus/venv-cactus-latest/bin/activate Some recommended options: ``` -cactus ./js ./examples/evolverMammals.txt evolverMammals.hal --batchSystem slurm --batchLogsDir batch-logs --consCores 64 --maxMemory 1.4Ti --doubleMem true --slurmTime 200:00:00 +cactus ./js ./examples/evolverMammals.txt evolverMammals.hal --batchSystem slurm --batchLogsDir batch-logs --consCores 64 --maxMemory 1.4Ti --doubleMem true --defaultWalltime 720000 ``` To run the same command step by step, ``` -cactus-prepare ./examples/evolverMammals.txt --outDir mammals-prepare --outHal mammals-prepare/evolverMammals.hal --cactusOptions "--maxMemory 1.4Ti --doubleMem true --slurmTime 200:00:00 --batchSystem slurm" --alignCores 64 --script > mammals.sh +cactus-prepare ./examples/evolverMammals.txt --outDir mammals-prepare --outHal mammals-prepare/evolverMammals.hal --cactusOptions "--maxMemory 1.4Ti --doubleMem true --defaultWalltime 720000 --batchSystem slurm" --alignCores 64 --script > mammals.sh chmod+x mammals.sh ./mammals.sh ``` diff --git a/src/cactus/blast/cactus_blast.py b/src/cactus/blast/cactus_blast.py index 275ed4f23..cc3d16593 100644 --- a/src/cactus/blast/cactus_blast.py +++ b/src/cactus/blast/cactus_blast.py @@ -12,12 +12,12 @@ import timeit from cactus.progressive.progressive_decomposition import compute_outgroups, parse_seqfile, get_subtree, get_spanning_subtree, get_event_set, get_ancestor_scaled_tree -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import getOptionalAttrib from cactus.shared.version import cactus_commit from cactus.progressive.cactus_prepare import human2bytesN @@ -34,6 +34,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("seqFile", help="Seq file") parser.add_argument("outputFile", type=str, help="Output pairwise alignment file") @@ -156,7 +157,7 @@ def runCactusBlastOnly(options): input_seq_id_map[genome] = toil.importFile(seq) paf_id = toil.start(Job.wrapJobFn(sanitize_then_make_paf_alignments, NXNewick().writeString(spanning_tree), - input_seq_id_map, options.root, config_node, options.outputFile)) + input_seq_id_map, options.root, config_node, options.outputFile, walltime=cactus_fast_walltime())) # export the alignments toil.exportFile(paf_id, makeURL(options.outputFile)) diff --git a/src/cactus/maf/cactus_hal2chains.py b/src/cactus/maf/cactus_hal2chains.py index 4603f9f86..b0cbdce3d 100644 --- a/src/cactus/maf/cactus_hal2chains.py +++ b/src/cactus/maf/cactus_hal2chains.py @@ -14,12 +14,12 @@ from operator import itemgetter from cactus.progressive.seqFile import SeqFile -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.version import cactus_commit @@ -41,6 +41,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("halFile", help = "HAL file to convert to MAF") parser.add_argument("outDir", help = "Output directory") @@ -154,7 +155,7 @@ def main(): config.substituteAllPredefinedConstantsWithLiterals(options) hal_id = toil.importFile(options.halFile) - chains_id_dict = toil.start(Job.wrapJobFn(hal2chains_workflow, config, options, hal_id)) + chains_id_dict = toil.start(Job.wrapJobFn(hal2chains_workflow, config, options, hal_id, walltime=cactus_fast_walltime())) #export the chains for query_genome in chains_id_dict.keys(): @@ -183,8 +184,8 @@ def hal2chains_workflow(job, config, options, hal_id): disk=int(hal_id.size * 1.2)) leaf_genomes = get_genomes_job.rv(0) distance_matrix = get_genomes_job.rv(1) - chrom_info_job = get_genomes_job.addFollowOnJobFn(hal2chains_chrom_info_all, config, options, hal_id, leaf_genomes) - hal2chains_all_job = chrom_info_job.addFollowOnJobFn(hal2chains_all, config, options, hal_id, chrom_info_job.rv(), distance_matrix) + chrom_info_job = get_genomes_job.addFollowOnJobFn(hal2chains_chrom_info_all, config, options, hal_id, leaf_genomes, walltime=cactus_fast_walltime()) + hal2chains_all_job = chrom_info_job.addFollowOnJobFn(hal2chains_all, config, options, hal_id, chrom_info_job.rv(), distance_matrix, walltime=cactus_fast_walltime()) return hal2chains_all_job.rv() def hal2chains_check_tools(job, options): diff --git a/src/cactus/maf/cactus_hal2maf.py b/src/cactus/maf/cactus_hal2maf.py index ffcba36c3..da2cda59b 100644 --- a/src/cactus/maf/cactus_hal2maf.py +++ b/src/cactus/maf/cactus_hal2maf.py @@ -15,12 +15,12 @@ from operator import itemgetter from cactus.progressive.seqFile import SeqFile -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.common import clean_jobstore_files @@ -44,6 +44,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("halFile", help = "HAL file to convert to MAF") parser.add_argument("outputMAF", help = "Output MAF (will be gzipped if ends in .gz). Suffix with .taf or .taf.gz if you want TAF output") @@ -220,7 +221,7 @@ def main(): bed_id = toil.importFile(options.bedRanges) if options.bedRanges else None hal_id = toil.importFile(options.halFile) - toil.start(Job.wrapJobFn(hal2maf_workflow, hal_id, bed_id, options, config)) + toil.start(Job.wrapJobFn(hal2maf_workflow, hal_id, bed_id, options, config, walltime=cactus_fast_walltime())) end_time = timeit.default_timer() run_time = end_time - start_time @@ -235,9 +236,9 @@ def hal2maf_workflow(job, hal_id, bed_id, options, config): hal2maf_ranges_job = job.addChildJobFn(hal2maf_ranges, hal_id, bed_id, options, cores=1, disk=hal_id.size) chunks, genome_list = hal2maf_ranges_job.rv(0), hal2maf_ranges_job.rv(1) - hal2maf_all_job = hal2maf_ranges_job.addFollowOnJobFn(hal2maf_all, hal_id, chunks, genome_list, options, config) + hal2maf_all_job = hal2maf_ranges_job.addFollowOnJobFn(hal2maf_all, hal_id, chunks, genome_list, options, config, walltime=cactus_fast_walltime()) hal2maf_merge_job = hal2maf_all_job.addFollowOnJobFn(hal2maf_merge_all, hal2maf_all_job.rv(), options, genome_list, - disk=hal_id.size) + disk=hal_id.size, walltime=cactus_fast_walltime()) hal2maf_ranges_job.addFollowOn(hal2maf_merge_job) # note: merge job also handles exporting (and some cleanup), indexing and coverage @@ -666,13 +667,13 @@ def hal2maf_merge_all(job, output_dicts, options, genome_list): output_ext += '.gz' if out_type != 'norm' and len(options.outType) > 1: output_name += '.{}'.format(out_type) - export_job = merge_job.addFollowOnJobFn(export_file, merge_job.rv(), output_name + output_ext) - merge_job.addFollowOnJobFn(clean_jobstore_files, file_ids=maf_ids) + export_job = merge_job.addFollowOnJobFn(export_file, merge_job.rv(), output_name + output_ext, walltime=cactus_fast_walltime()) + merge_job.addFollowOnJobFn(clean_jobstore_files, file_ids=maf_ids, walltime=cactus_fast_walltime()) if options.index: index_job = merge_job.addFollowOnJobFn(taffy_index, merge_job.rv(), output_name + output_ext, disk=int(1.1 * maf_size), memory=cactus_clamp_memory(maf_size / 10)) - index_job.addFollowOnJobFn(export_file, index_job.rv(), output_name + output_ext + '.tai') + index_job.addFollowOnJobFn(export_file, index_job.rv(), output_name + output_ext + '.tai', walltime=cactus_fast_walltime()) if options.coverage: @@ -680,7 +681,7 @@ def hal2maf_merge_all(job, output_dicts, options, genome_list): genome_list, options, disk=int(1.1 * maf_size), memory=cactus_clamp_memory(maf_size / 10)) - coverage_job.addFollowOnJobFn(export_file, coverage_job.rv(), output_name + output_ext + '.cov.tsv') + coverage_job.addFollowOnJobFn(export_file, coverage_job.rv(), output_name + output_ext + '.cov.tsv', walltime=cactus_fast_walltime()) return export_job.rv() diff --git a/src/cactus/maf/cactus_maf2bigmaf.py b/src/cactus/maf/cactus_maf2bigmaf.py index 4ea7a16e4..ec5aeb710 100644 --- a/src/cactus/maf/cactus_maf2bigmaf.py +++ b/src/cactus/maf/cactus_maf2bigmaf.py @@ -14,12 +14,12 @@ from operator import itemgetter from cactus.progressive.seqFile import SeqFile -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.common import cactus_clamp_memory @@ -36,6 +36,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("mafFile", help = "MAF file to convert to BigMaf (can be gzipped)") parser.add_argument("outFile", help = "Output bigMaf file (.bb)") @@ -93,7 +94,7 @@ def main(): if options.halFile: hal_id = toil.importFile(options.halFile) - bigmaf_id_dict = toil.start(Job.wrapJobFn(maf2bigmaf_workflow, config, options, maf_id, hal_id)) + bigmaf_id_dict = toil.start(Job.wrapJobFn(maf2bigmaf_workflow, config, options, maf_id, hal_id, walltime=cactus_fast_walltime())) #export the big maf out_bm_path = makeURL(options.outFile) diff --git a/src/cactus/maf/cactus_phast.py b/src/cactus/maf/cactus_phast.py index 51219ba5f..99a891dad 100644 --- a/src/cactus/maf/cactus_phast.py +++ b/src/cactus/maf/cactus_phast.py @@ -23,12 +23,12 @@ import timeit import xml.etree.ElementTree as ET -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import cactus_clamp_memory from cactus.shared.common import cactus_cpu_count @@ -51,6 +51,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("inMaf", help="Input alignment as produced by cactus-hal2maf. " "Accepts .maf, .maf.gz, .taf, or .taf.gz (auto-detected " @@ -415,7 +416,7 @@ def main(): toil.start(Job.wrapJobFn(phast_workflow, config, options, maf_id, tai_id, tai_built, hal_id, ann_id, model_id, - bed_id)) + bed_id, walltime=cactus_fast_walltime())) end_time = timeit.default_timer() logger.info("cactus-phast finished in {} seconds".format(end_time - start_time)) @@ -814,20 +815,20 @@ def phast_workflow(job, config, options, maf_id, tai_id, tai_built, hal_id, ann_ # Convenience sed script: maps phast canonical names back to HAL genome # names. Only relevant when at least one HAL genome name contains '.'; # the job no-ops otherwise. - setup_job.addFollowOnJobFn(export_name_map_job, options, species_list) + setup_job.addFollowOnJobFn(export_name_map_job, options, species_list, walltime=cactus_fast_walltime()) # build .tai if missing, else just chain through the existing one if tai_built: idx_job = setup_job.addFollowOnJobFn(taffy_index_job, maf_id, os.path.basename(options.inMaf), disk=int(maf_id.size * 1.1)) tai_id = idx_job.rv() - idx_job.addFollowOnJobFn(export_file, tai_id, tai_export_path(options)) + idx_job.addFollowOnJobFn(export_file, tai_id, tai_export_path(options), walltime=cactus_fast_walltime()) plan_parent = idx_job else: plan_parent = setup_job plan_job = plan_parent.addFollowOnJobFn(plan_chunks_job, options, ref_seq_lengths, tai_id, - bed_id, aligned_contigs) + bed_id, aligned_contigs, walltime=cactus_fast_walltime()) chunk_specs = plan_job.rv() # The single multi-core chunker job: localizes the source MAF once, @@ -867,7 +868,7 @@ def phast_workflow(job, config, options, maf_id, tai_id, tai_built, hal_id, ann_ # ----- phyloFit branch (also runs in phyloP mode if no model was given) ----- if need_train: train_job = chunk_job.addFollowOnJobFn(train_workflow, options, chunks, - species_list, tree_str, ann_id) + species_list, tree_str, ann_id, walltime=cactus_fast_walltime()) # train_workflow returns the model file id trained_model_id = train_job.rv() if options.mode == 'phyloFit': @@ -890,7 +891,7 @@ def phast_workflow(job, config, options, maf_id, tai_id, tai_built, hal_id, ann_ for sub in track_inputs: score_job = score_parent.addFollowOnJobFn(phyloP_workflow, options, chunks, model_id, ref_seq_lengths, species_list, - sub, effective_root_name) + sub, effective_root_name, walltime=cactus_fast_walltime()) track_rvs.append(score_job.rv()) return track_rvs @@ -1301,7 +1302,7 @@ def train_workflow(job, options, chunks, species_list, tree_str, ann_id): # projection); with --root, it's restricted to clade leaves. leaves_csv = ','.join(phast_name(g) for g in newick_leaves(tree_str)) extract_job = job.addChildJobFn(extract_4d_all, options, chunks, ann_id, - species_list, leaves_csv) + species_list, leaves_csv, walltime=cactus_fast_walltime()) ss_results = extract_job.rv() # flat list of ss file ids (one per chunk-group) # 4d-site SS files are tiny even at 447-way (~300 MB aggregated). 4 GiB @@ -1310,7 +1311,7 @@ def train_workflow(job, options, chunks, species_list, tree_str, ann_id): species_list, tree_str, disk=4 * 1024**3) aggregate_id = aggregate_job.rv() - aggregate_job.addFollowOnJobFn(export_file, aggregate_id, ss_export_path(options)) + aggregate_job.addFollowOnJobFn(export_file, aggregate_id, ss_export_path(options), walltime=cactus_fast_walltime()) # phyloFit on 577-way × 1.4 GB SS peaked at ~300 MiB; default memory is # fine unless the user overrides via --phyloFitMemory. @@ -1329,7 +1330,7 @@ def train_workflow(job, options, chunks, species_list, tree_str, ann_id): else: export_parent = fit_job - export_parent.addFollowOnJobFn(export_file, model_id, model_export_path(options)) + export_parent.addFollowOnJobFn(export_file, model_id, model_export_path(options), walltime=cactus_fast_walltime()) return model_id @@ -1571,7 +1572,7 @@ def phyloP_workflow(job, options, chunks, model_id, ref_seq_lengths, species_lis else: RealtimeLogger.info('phyloP track: global conservation (no --subtree)') score_job = job.addChildJobFn(phyloP_all, options, chunks, model_id, species_list, - track_subtree) + track_subtree, walltime=cactus_fast_walltime()) per_chunk_wigs = score_job.rv() # flat list of (contig, start, wig_id_or_None) per chunk # disk estimates: per-base wig text is ~10 bytes/ref_bp uncompressed; bgzip @@ -1586,7 +1587,7 @@ def phyloP_workflow(job, options, chunks, model_id, ref_seq_lengths, species_lis disk=merge_disk, cores=merge_cores) wig_id = merge_job.rv() - merge_job.addFollowOnJobFn(export_file, wig_id, wig_export_path(options, track_subtree)) + merge_job.addFollowOnJobFn(export_file, wig_id, wig_export_path(options, track_subtree), walltime=cactus_fast_walltime()) if options.bigwig: # decompressed wig + .bw output + bbiFile scratch @@ -1602,7 +1603,7 @@ def phyloP_workflow(job, options, chunks, model_id, ref_seq_lengths, species_lis memory=cactus_clamp_memory(bw_mem), cores=merge_cores) bw_job.addFollowOnJobFn(export_file, bw_job.rv(), - bigwig_export_path(options, track_subtree)) + bigwig_export_path(options, track_subtree), walltime=cactus_fast_walltime()) return wig_id diff --git a/src/cactus/paf/local_alignment.py b/src/cactus/paf/local_alignment.py index 193df4bd3..44b4a7736 100755 --- a/src/cactus/paf/local_alignment.py +++ b/src/cactus/paf/local_alignment.py @@ -20,7 +20,7 @@ import copy from Bio import SeqIO from cactus.paf.paf import get_event_pairs, get_leaves, get_node, get_distances -from cactus.shared.common import cactus_call, getOptionalAttrib, zip_gz +from cactus.shared.common import cactus_call, getOptionalAttrib, zip_gz, cactus_fast_walltime from cactus.preprocessor.checkUniqueHeaders import sanitize_fasta_headers from cactus.preprocessor.unmasking import unmask_contigs_all from cactus.preprocessor.cactus_preprocessor import clean_if_different @@ -421,7 +421,7 @@ def invert_alignments(job, alignment_file): def make_ingroup_to_outgroup_alignments_0(job, ingroup_event, outgroup_events, event_names_to_sequences, distances, params): # Generate the alignments fle alignment_file = job.addChildJobFn(make_ingroup_to_outgroup_alignments_1, ingroup_event, outgroup_events, - event_names_to_sequences, distances, params).rv() + event_names_to_sequences, distances, params, walltime=cactus_fast_walltime()).rv() # Invert the final alignment so that the query is the outgroup and the target is the ingroup return job.addFollowOnJobFn(invert_alignments, alignment_file).rv() @@ -493,7 +493,7 @@ def make_ingroup_to_outgroup_alignments_2(job, alignments, ingroup_event, outgro # recursively make alignments with the remaining outgroups alignments2 = root_job.addChildJobFn(make_ingroup_to_outgroup_alignments_1, ingroup_event, outgroup_events, - event_names_to_sequences, distances, params).rv() + event_names_to_sequences, distances, params, walltime=cactus_fast_walltime()).rv() return root_job.addFollowOnJobFn(make_ingroup_to_outgroup_alignments_3, ingroup_event, event_names_to_sequences[ingroup_event.iD], alignments, alignments2).rv() @@ -654,7 +654,7 @@ def chain_alignments(job, alignment_files, alignment_names, reference_event_name memory=cactus_clamp_memory(4 * split_size)).rv() ) - return job.addFollowOnJobFn(merge_processed_alignments, processed_rvs).rv() + return job.addFollowOnJobFn(merge_processed_alignments, processed_rvs, walltime=cactus_fast_walltime()).rv() def chain_tile_trim_filter_one_contig(job, split_file_id, reference_event_name, params): @@ -738,9 +738,9 @@ def merge_processed_alignments(job, processed_file_ids): def sanitize_then_make_paf_alignments(job, event_tree_string, event_names_to_sequences, ancestor_event_string, params, output_path=None): - sanitize_job = job.addChildJobFn(sanitize_fasta_headers, event_names_to_sequences) + sanitize_job = job.addChildJobFn(sanitize_fasta_headers, event_names_to_sequences, walltime=cactus_fast_walltime()) paf_job = sanitize_job.addFollowOnJobFn(make_paf_alignments, event_tree_string, sanitize_job.rv(), - ancestor_event_string, params) + ancestor_event_string, params, walltime=cactus_fast_walltime()) # gzip the output if requested if output_path and output_path.endswith('.gz'): gzip_job = paf_job.addFollowOnJobFn(zip_gz, output_path, paf_job.rv()) @@ -792,7 +792,7 @@ def make_paf_alignments(job, event_tree_string, event_names_to_sequences, ancest if getOptionalAttrib(lastz_params_node.find("unmask"), 'action', typeFn=str, default='none') != 'none': ingroups = [ingroup.iD for ingroup in ingroup_events] # Pass a copy of event_names_to_sequences to unmask_job to avoid circular reference - unmask_job = root_job.addChildJobFn(unmask_contigs_all, input_sequence_map, ingroups, params) + unmask_job = root_job.addChildJobFn(unmask_contigs_all, input_sequence_map, ingroups, params, walltime=cactus_fast_walltime()) for i,ingroup in enumerate(ingroups): event_names_to_sequences[ingroup] = unmask_job.rv(i) new_root_job = Job() @@ -822,7 +822,7 @@ def make_paf_alignments(job, event_tree_string, event_names_to_sequences, ancest # for each ingroup make alignments to the outgroups if int(params.find("blast").attrib["trimIngroups"]): # Trim the ingroup sequences outgroup_alignments = [root_job.addChildJobFn(make_ingroup_to_outgroup_alignments_0, ingroup, outgroup_events, - dict(event_names_to_sequences), distances, params).rv() + dict(event_names_to_sequences), distances, params, walltime=cactus_fast_walltime()).rv() for ingroup in ingroup_events] if len(outgroup_events) > 0 else [] else: outgroup_alignments = [root_job.addChildJobFn(make_chunked_alignments, @@ -844,11 +844,11 @@ def make_paf_alignments(job, event_tree_string, event_names_to_sequences, ancest ingroup_alignments, ingroup_alignment_names, outgroup_alignments, outgroup_alignment_names, ancestor_event_string, params, - total_sequence_size=total_sequence_size).rv() + total_sequence_size=total_sequence_size, walltime=cactus_fast_walltime()).rv() # Delete the unmasked fastas (todo: should we do the unmasking somewhere further upstream?) for ingroup in ingroup_events: - root_job.addFollowOnJobFn(clean_if_different, event_names_to_sequences[ingroup.iD], input_sequence_map[ingroup.iD]) + root_job.addFollowOnJobFn(clean_if_different, event_names_to_sequences[ingroup.iD], input_sequence_map[ingroup.iD], walltime=cactus_fast_walltime()) return root_job.addFollowOnJobFn(chain_alignments, ingroup_alignments + outgroup_alignments, ingroup_alignment_names + outgroup_alignment_names, diff --git a/src/cactus/preprocessor/cactus_preprocessor.py b/src/cactus/preprocessor/cactus_preprocessor.py index b4cfa18a3..275ef5b32 100644 --- a/src/cactus/preprocessor/cactus_preprocessor.py +++ b/src/cactus/preprocessor/cactus_preprocessor.py @@ -19,7 +19,7 @@ from sonLib.bioio import getTempDirectory from toil.common import Toil from toil.job import Job -from cactus.shared.common import cactus_call +from cactus.shared.common import cactus_call, cactus_fast_walltime from cactus.shared.common import RoundedJob from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.common import runGetChunks @@ -36,7 +36,7 @@ from toil.statsAndLogging import set_logging_from_options from toil.realtimeLogger import RealtimeLogger -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.preprocessor.checkUniqueHeaders import checkUniqueHeaders from cactus.preprocessor.lastzRepeatMasking.cactus_lastzRepeatMask import LastzRepeatMaskJob from cactus.preprocessor.lastzRepeatMasking.cactus_lastzRepeatMask import RepeatMaskOptions @@ -305,7 +305,7 @@ def run(self, fileStore): ppJob = self.addChild(PreprocessSequence(prepOptions, self.inSequenceID)) outSeqID = ppJob.rv() - self.addFollowOnJobFn(clean_if_different, self.inSequenceID, outSeqID) + self.addFollowOnJobFn(clean_if_different, self.inSequenceID, outSeqID, walltime=cactus_fast_walltime()) else: logger.info("Skipping inactive preprocessor {}".format(prepNode.attrib["preprocessJob"])) outSeqID = self.inSequenceID @@ -419,7 +419,7 @@ def stageWorkflow(outputSequenceDir, configNode, inputSequences, toil, restart=F inputSequenceIDs.append(toil.importFile(makeURL(seq))) maskFileID = toil.importFile(makeURL(maskFile)) if maskFile else None unzip_job = Job.wrapJobFn(unzip_then_pp, configNode, inputSequences, inputSequenceIDs, inputEventNames, - maskFile, maskFileID, maskAction, minLength) + maskFile, maskFileID, maskAction, minLength, walltime=cactus_fast_walltime()) outputSequenceIDs = toil.start(unzip_job) else: outputSequenceIDs = toil.restart() @@ -435,13 +435,13 @@ def stageWorkflow(outputSequenceDir, configNode, inputSequences, toil, restart=F def unzip_then_pp(job, config_node, input_fa_paths, input_fa_ids, input_event_names, mask_file_path, mask_file_id, mask_file_action, min_length): """ unzip then preprocess """ - unzip_job = job.addChildJobFn(unzip_gzs, input_fa_paths, input_fa_ids) + unzip_job = job.addChildJobFn(unzip_gzs, input_fa_paths, input_fa_ids, walltime=cactus_fast_walltime()) if mask_file_id is not None: - mask_unzip_job = unzip_job.addChildJobFn(unzip_gzs, [mask_file_path], [mask_file_id]) + mask_unzip_job = unzip_job.addChildJobFn(unzip_gzs, [mask_file_path], [mask_file_id], walltime=cactus_fast_walltime()) config_node = mask_unzip_job.addFollowOnJobFn(maskJobOverride, config_node, mask_file_path, mask_unzip_job.rv(0), mask_file_action, min_length, disk=mask_file_id.size*20).rv() pp_job = unzip_job.addFollowOn(CactusPreprocessor([unzip_job.rv(i) for i in range(len(input_fa_ids))], config_node, eventNames=input_event_names)) - zip_job = pp_job.addFollowOnJobFn(zip_gzs, input_fa_paths, pp_job.rv(), list_elems = [0]) + zip_job = pp_job.addFollowOnJobFn(zip_gzs, input_fa_paths, pp_job.rv(), list_elems = [0], walltime=cactus_fast_walltime()) return zip_job.rv() def runCactusPreprocessor(outputSequenceDir, configFile, inputSequences, toilDir): @@ -453,6 +453,7 @@ def runCactusPreprocessor(outputSequenceDir, configFile, inputSequences, toilDir def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("inSeqFile", type=str, nargs='?', default=None, help = "Input Seq file") parser.add_argument("outSeqFile", type=str, nargs='?', default=None, help = "Output Seq file (ex generated with cactus-prepare)") parser.add_argument("--configFile", default=os.path.join(cactusRootPath(), "cactus_progressive_config.xml")) diff --git a/src/cactus/preprocessor/unmasking.py b/src/cactus/preprocessor/unmasking.py index e2e42d208..efdab86b7 100644 --- a/src/cactus/preprocessor/unmasking.py +++ b/src/cactus/preprocessor/unmasking.py @@ -5,7 +5,7 @@ from Bio.SeqRecord import SeqRecord import os from toil.job import Job -from cactus.shared.common import cactus_clamp_memory, catFiles, clean_jobstore_files +from cactus.shared.common import cactus_clamp_memory, catFiles, clean_jobstore_files, cactus_fast_walltime from cactus.shared.common import cactus_call, getOptionalAttrib from toil.realtimeLogger import RealtimeLogger from cactus.preprocessor.cactus_preprocessor import CactusPreprocessor @@ -98,7 +98,7 @@ def unmask_contigs_one(job, event, fasta_id, params): disk=fasta_id.size * 4) fasta_id = fa_merge_job.rv() # delete the pp_id - fa_merge_job.addFollowOnJobFn(clean_jobstore_files, file_ids=[pp_id]) + fa_merge_job.addFollowOnJobFn(clean_jobstore_files, file_ids=[pp_id], walltime=cactus_fast_walltime()) else: RealtimeLogger.warning('Remasking activated but no maskers active in preprocessor: doing nothing') diff --git a/src/cactus/progressive/cactus_prepare.py b/src/cactus/progressive/cactus_prepare.py index e3a1dbf59..313269541 100644 --- a/src/cactus/progressive/cactus_prepare.py +++ b/src/cactus/progressive/cactus_prepare.py @@ -32,7 +32,7 @@ from cactus.shared.common import findRequiredNode from cactus.shared.common import makeURL, cactus_call, RoundedJob from cactus.shared.common import write_s3, has_s3, get_aws_region -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_clamp_memory from toil.job import Job @@ -51,6 +51,7 @@ def main(toil_mode=False): parser = ArgumentParser() if toil_mode: parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("--latest", dest="latest", action="store_true", help="Use the latest version of the docker container " "rather than pulling one matching this version of cactus") @@ -1265,6 +1266,7 @@ def main_hal2fasta(): """ cli wrapper for hal2fasta so that cactus-perpare can send it to slurm consistently with other cactus commands """ parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("halFile", help="input HAL file") parser.add_argument("genome", help="Genome to convert to FASTA") @@ -1332,6 +1334,7 @@ def hal2fasta(job, hal_id, hal_path, genome, fa_path): def main_hal_append_subtrees(): """ toil wrapper for halAppendSubtree command(s) """ parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("tgtFile", help="input HAL file to append to") parser.add_argument("subFiles", nargs='+', help="subtree HAL files to append") diff --git a/src/cactus/progressive/cactus_progressive.py b/src/cactus/progressive/cactus_progressive.py index 61ff4d963..17373a7af 100755 --- a/src/cactus/progressive/cactus_progressive.py +++ b/src/cactus/progressive/cactus_progressive.py @@ -18,7 +18,7 @@ from base64 import b64encode from toil.lib.bioio import getTempFile -from cactus.shared.common import cactus_cpu_count +from cactus.shared.common import cactus_cpu_count, cactus_fast_walltime from toil.statsAndLogging import logger from toil.statsAndLogging import set_logging_from_options from toil.realtimeLogger import RealtimeLogger @@ -33,7 +33,7 @@ from cactus.shared.version import cactus_commit from cactus.shared.common import cactusRootPath from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import write_s3 from cactus.shared.common import cactus_clamp_memory from cactus.shared.common import clean_jobstore_files @@ -68,7 +68,7 @@ def preprocess_all(job, options, config_node, input_seq_id_map): pp_seq_ids[event] = preprocessor_job.rv(i) # do the logging and checkpointing - root_job.addFollowOnJobFn(save_preprocessed_files, options, config_node, pp_seq_ids) + root_job.addFollowOnJobFn(save_preprocessed_files, options, config_node, pp_seq_ids, walltime=cactus_fast_walltime()) return pp_seq_ids @@ -137,7 +137,7 @@ def progressive_schedule(job, options, config_node, seq_id_map, tree, og_map, ro # to be consistent with pre-refactor (and work with updating tests), we include the root when its id is input if event in seq_id_map and seq_id_map[event]: event_id_map[event] = seq_id_map[event] - event_job = Job.wrapJobFn(progressive_step, options, config_node, event_id_map, tree, og_map, event) + event_job = Job.wrapJobFn(progressive_step, options, config_node, event_id_map, tree, og_map, event, walltime=cactus_fast_walltime()) job_table[event] = event_job for dep in dep_table[event]: if dep in job_table: @@ -174,7 +174,7 @@ def progressive_step(job, options, config_node, seq_id_map, tree, og_map, event) # do the blast paf_job = job.addChildJobFn(make_paf_alignments, NXNewick().writeString(spanning_tree), - subtree_eventmap, event, config_node).encapsulate() + subtree_eventmap, event, config_node, walltime=cactus_fast_walltime()).encapsulate() outgroups = og_map[event] if event in og_map else [] # trim the outgroups @@ -182,14 +182,14 @@ def progressive_step(job, options, config_node, seq_id_map, tree, og_map, event) trim_sequences = paf_job.addChildJobFn(trim_unaligned_sequences, [subtree_eventmap[i] for i in outgroups], paf_job.rv(), config_node) cons_job = paf_job.addFollowOnJobFn(progressive_step_2, trim_sequences.rv(), options, config_node, subtree_eventmap, - spanning_tree, og_map, event) + spanning_tree, og_map, event, walltime=cactus_fast_walltime()) else: # Without outgroup trimming (or if there are no outgroups to trim) cons_job = paf_job.addChildJobFn(cactus_cons_with_resources, spanning_tree, event, config_node, subtree_eventmap, og_map, paf_job.rv(), cons_cores=options.consCores, cons_memory=options.consMemory, - intermediate_results_url=options.intermediateResultsUrl) + intermediate_results_url=options.intermediateResultsUrl, walltime=cactus_fast_walltime()) # erase the paf since its now longer needed - cons_job.addFollowOnJobFn(clean_jobstore_files, file_ids=[paf_job.rv()]) + cons_job.addFollowOnJobFn(clean_jobstore_files, file_ids=[paf_job.rv()], walltime=cactus_fast_walltime()) return cons_job.rv() @@ -203,7 +203,7 @@ def progressive_step_2(job, trimmed_outgroups_and_alignments, options, config_no # now do consolidated return job.addChildJobFn(cactus_cons_with_resources, spanning_tree, event, config_node, subtree_eventmap, og_map, pafs, cons_cores=options.consCores, cons_memory=options.consMemory, - intermediate_results_url=options.intermediateResultsUrl).rv() + intermediate_results_url=options.intermediateResultsUrl, walltime=cactus_fast_walltime()).rv() def export_hal(job, mc_tree, config_node, seq_id_map, og_map, results, event=None, cacheBytes=None, @@ -300,11 +300,11 @@ def progressive_workflow(job, options, config_node, mc_tree, og_map, input_seq_i ''' run the entire progressive workflow ''' # run the usual unzip / rename, even before preprocessing - sanitize_job = job.addChildJobFn(sanitize_fasta_headers, input_seq_id_map) + sanitize_job = job.addChildJobFn(sanitize_fasta_headers, input_seq_id_map, walltime=cactus_fast_walltime()) # start with the preprocessor if not options.skipPreprocessor: - pp_job = sanitize_job.addFollowOnJobFn(preprocess_all, options, config_node, sanitize_job.rv()) + pp_job = sanitize_job.addFollowOnJobFn(preprocess_all, options, config_node, sanitize_job.rv(), walltime=cactus_fast_walltime()) seq_id_map = pp_job.rv() sanitize_job = pp_job else: @@ -322,7 +322,7 @@ def progressive_workflow(job, options, config_node, mc_tree, og_map, input_seq_i scaled_tree = get_ancestor_scaled_tree(mc_tree, root_event, max_div, branch_scale=options.branchScale, upweight_ancestors=upweight_ancestors) - progressive_job = sanitize_job.addFollowOnJobFn(progressive_schedule, options, config_node, seq_id_map, scaled_tree, og_map, root_event) + progressive_job = sanitize_job.addFollowOnJobFn(progressive_schedule, options, config_node, seq_id_map, scaled_tree, og_map, root_event, walltime=cactus_fast_walltime()) # then do the hal export hal_export_job = progressive_job.addFollowOnJobFn(export_hal, mc_tree, config_node, seq_id_map, og_map, @@ -332,6 +332,7 @@ def progressive_workflow(job, options, config_node, mc_tree, og_map, input_seq_i def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("seqFile", help = "Seq file") parser.add_argument("outputHal", type=str, help = "Output HAL file") @@ -493,7 +494,7 @@ def main(): loadDnaBrnnModel(toil, config_node) # run the whole workflow - hal_id = toil.start(Job.wrapJobFn(progressive_workflow, options, config_node, mc_tree, og_map, input_seq_id_map)) + hal_id = toil.start(Job.wrapJobFn(progressive_workflow, options, config_node, mc_tree, og_map, input_seq_id_map, walltime=cactus_fast_walltime())) toil.exportFile(hal_id, makeURL(options.outputHal)) diff --git a/src/cactus/refmap/cactus_graphmap.py b/src/cactus/refmap/cactus_graphmap.py index 2e534bf7c..ad15b545a 100644 --- a/src/cactus/refmap/cactus_graphmap.py +++ b/src/cactus/refmap/cactus_graphmap.py @@ -17,12 +17,12 @@ from operator import itemgetter from cactus.progressive.seqFile import SeqFile -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.common import unzip_gz, zip_gz @@ -42,6 +42,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("seqFile", help = "Seq file (will be modified if necessary to include graph Fasta sequence) (or chromfile with --batch)") parser.add_argument("minigraphGFA", nargs='?', default='', type=str, @@ -233,7 +234,7 @@ def graph_map(options): # run the workflow # output_dict is chrom -> paf_id, gfa_fa_id, gaf_id, unfiltered_paf_id, paf_filter_log, paf_was_filtered - output_dict = toil.start(Job.wrapJobFn(minigraph_batch_workflow, options, config_wrapper, input_dict, graph_event, True)) + output_dict = toil.start(Job.wrapJobFn(minigraph_batch_workflow, options, config_wrapper, input_dict, graph_event, True, walltime=cactus_fast_walltime())) export_graphmap_output(options, config_node, input_map, output_dict, toil) @@ -304,7 +305,7 @@ def minigraph_batch_workflow(job, options, config, input_dict, graph_event, sani else: chrom_options = options mgwf_job = job.addChildJobFn(minigraph_workflow, chrom_options, config, seq_id_map, gfa_id, graph_event, - sanitize, ref_collapse_paf_id, pansn_gfa_input) + sanitize, ref_collapse_paf_id, pansn_gfa_input, walltime=cactus_fast_walltime()) output_dict[chrom] = mgwf_job.rv() return output_dict @@ -325,7 +326,7 @@ def minigraph_workflow(job, options, config, seq_id_map, gfa_id, graph_event, sa # enforce unique prefixes and unzip fastas if sanitize: - sanitize_job = root_job.addChildJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True) + sanitize_job = root_job.addChildJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True, walltime=cactus_fast_walltime()) seq_id_map = sanitize_job.rv() # add unique prefixes to the input PAF @@ -356,7 +357,7 @@ def minigraph_workflow(job, options, config, seq_id_map, gfa_id, graph_event, sa gfa_id = gfa_unzip_job.rv() gfa_id_size *= 10 options.minigraphGFA = options.minigraphGFA[:-3] - paf_job = Job.wrapJobFn(minigraph_map_all, options, config, gfa_id, seq_id_map, graph_event) + paf_job = Job.wrapJobFn(minigraph_map_all, options, config, gfa_id, seq_id_map, graph_event, walltime=cactus_fast_walltime()) root_job.addFollowOn(paf_job) collapse_paf_id = ref_collapse_paf_id @@ -372,7 +373,7 @@ def minigraph_workflow(job, options, config, seq_id_map, gfa_id, graph_event, sa paf_job.addFollowOn(gfa2paf_job) collapse_mode = getOptionalAttrib(findRequiredNode(config.xmlRoot, "graphmap"), "collapse", typeFn=str, default="none") if collapse_mode in ['reference', 'all', 'nonref']: - collapse_job = paf_job.addChildJobFn(self_align_all, config, seq_id_map, options.reference, collapse_mode) + collapse_job = paf_job.addChildJobFn(self_align_all, config, seq_id_map, options.reference, collapse_mode, walltime=cactus_fast_walltime()) if ref_collapse_paf_id: collapse_paf_id = collapse_job.addFollowOnJobFn(merge_pafs, {"1":collapse_job.rv(), "2":ref_collapse_paf_id}, diff --git a/src/cactus/refmap/cactus_graphmap_join.py b/src/cactus/refmap/cactus_graphmap_join.py index 2d22e081e..37324784b 100644 --- a/src/cactus/refmap/cactus_graphmap_join.py +++ b/src/cactus/refmap/cactus_graphmap_join.py @@ -35,12 +35,12 @@ from cactus.progressive.seqFile import SeqFile from cactus.progressive.multiCactusTree import MultiCactusTree -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.common import unzip_gz, write_s3 @@ -64,6 +64,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("--vg", required=False, nargs='+', default=None, help = "Input vg files (PackedGraph or HashGraph format)") parser.add_argument("--vgFull", nargs='+', default=None, @@ -583,7 +584,7 @@ def graphmap_join(options): vg_ids = bypass_full_ids or bypass_clip_ids or bypass_filter_ids wf_output = toil.start(Job.wrapJobFn(graphmap_join_workflow, options, config, vg_ids, hal_ids, sv_gfa_ids, - bypass_full_ids, bypass_clip_ids, bypass_filter_ids)) + bypass_full_ids, bypass_clip_ids, bypass_filter_ids, walltime=cactus_fast_walltime())) else: # load up the vgs vg_ids = [] @@ -591,7 +592,7 @@ def graphmap_join(options): vg_ids.append(toil.importFile(makeURL(vg_path))) # run the workflow - wf_output = toil.start(Job.wrapJobFn(graphmap_join_workflow, options, config, vg_ids, hal_ids, sv_gfa_ids)) + wf_output = toil.start(Job.wrapJobFn(graphmap_join_workflow, options, config, vg_ids, hal_ids, sv_gfa_ids, walltime=cactus_fast_walltime())) #export the split data export_join_data(toil, options, wf_output[0], wf_output[1], wf_output[2], wf_output[3], wf_output[4], wf_output[5]) @@ -750,7 +751,7 @@ def graphmap_join_workflow(job, options, config, vg_ids, hal_ids, sv_gfa_ids, hal_id_dict = hal_merge_job.rv() out_dicts.append(hal_id_dict) # delete the chromosome hals - hal_merge_job.addFollowOnJobFn(clean_jobstore_files, file_ids=hal_ids) + hal_merge_job.addFollowOnJobFn(clean_jobstore_files, file_ids=hal_ids, walltime=cactus_fast_walltime()) # optional minigraph gfa merge if sv_gfa_ids: @@ -759,7 +760,7 @@ def graphmap_join_workflow(job, options, config, vg_ids, hal_ids, sv_gfa_ids, sv_gfa_id_dict = sv_gfa_merge_job.rv() out_dicts.append(sv_gfa_id_dict) # delete the chromosome gfas - sv_gfa_merge_job.addFollowOnJobFn(clean_jobstore_files, file_ids=sv_gfa_ids) + sv_gfa_merge_job.addFollowOnJobFn(clean_jobstore_files, file_ids=sv_gfa_ids, walltime=cactus_fast_walltime()) if options.indexMemory: index_mem = options.indexMemory @@ -858,7 +859,7 @@ def graphmap_join_workflow(job, options, config, vg_ids, hal_ids, sv_gfa_ids, for vcf_ref in options.vcfReference: vcf_job = gfa_root_job.addFollowOnJobFn(make_vcf, config, options, workflow_phase, index_mem, vcf_ref, phase_vg_ids, - ref_fasta_job.rv() if ref_fasta_job else None) + ref_fasta_job.rv() if ref_fasta_job else None, walltime=cactus_fast_walltime()) if ref_fasta_job: ref_fasta_job.addFollowOn(vcf_job) out_dicts.append(vcf_job.rv()) @@ -2102,7 +2103,7 @@ def build_vg_indexes_and_vcf(parent_job, options, config, phase_vg_ids, vg_ids, vcf_job = gfa_root_job.addFollowOnJobFn(make_vcf, config, options, tag.rstrip('.'), index_mem, vcf_ref, phase_vg_ids, ref_fasta_dict, vcftag=vcftag, - is_gref=is_gref, decon_L=decon_L) + is_gref=is_gref, decon_L=decon_L, walltime=cactus_fast_walltime()) out_dicts.append(vcf_job.rv()) # optional haplo index diff --git a/src/cactus/refmap/cactus_graphmap_split.py b/src/cactus/refmap/cactus_graphmap_split.py index dd413612f..3bd355c19 100644 --- a/src/cactus/refmap/cactus_graphmap_split.py +++ b/src/cactus/refmap/cactus_graphmap_split.py @@ -14,12 +14,12 @@ from cactus.progressive.seqFile import SeqFile from cactus.progressive.multiCactusTree import MultiCactusTree -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.common import unzip_gz, write_s3 @@ -42,6 +42,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("seqFile", help = "Seq file (gzipped fastas supported)") parser.add_argument("minigraphGFA", help = "Minigraph-compatible reference graph in GFA format (can be gzipped)") @@ -175,7 +176,7 @@ def cactus_graphmap_split(options): # run the workflow wf_output = toil.start(Job.wrapJobFn(graphmap_split_workflow, options, config, input_seq_id_map, input_name_map, gfa_id, options.minigraphGFA, - paf_id, options.graphmapPAF)) + paf_id, options.graphmapPAF, walltime=cactus_fast_walltime())) #export the split data export_split_data(toil, wf_output[0], wf_output[1], wf_output[2], wf_output[3], options.outDir, config) @@ -208,7 +209,7 @@ def graphmap_split_workflow(job, options, config, seq_id_map, seq_name_map, gfa_ # fix up the headers if sanitize: - sanitize_job = root_job.addChildJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True) + sanitize_job = root_job.addChildJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True, walltime=cactus_fast_walltime()) seq_id_map = sanitize_job.rv() else: sanitize_job = Job() @@ -250,7 +251,7 @@ def graphmap_split_workflow(job, options, config, seq_id_map, seq_name_map, gfa_ mask_bed_id = None if options.maskFilter: - mask_bed_id = sanitize_job.addFollowOnJobFn(get_mask_bed, seq_id_map, options.maskFilter).rv() + mask_bed_id = sanitize_job.addFollowOnJobFn(get_mask_bed, seq_id_map, options.maskFilter, walltime=cactus_fast_walltime()).rv() # use rgfa-split to split the gfa and paf up by contig split_gfa_job = root_job.addFollowOnJobFn(split_gfa, config, gfa_id, [paf_id], ref_contigs, @@ -259,14 +260,14 @@ def graphmap_split_workflow(job, options, config, seq_id_map, seq_name_map, gfa_ memory=cactus_clamp_memory((gfa_size + paf_size) * 3)) # use the output of the above splitting to do the fasta splitting - split_fas_job = split_gfa_job.addFollowOnJobFn(split_fas, seq_id_map, seq_name_map, split_gfa_job.rv(0)) + split_fas_job = split_gfa_job.addFollowOnJobFn(split_fas, seq_id_map, seq_name_map, split_gfa_job.rv(0), walltime=cactus_fast_walltime()) # gather everythign up into a table - gather_fas_job = split_fas_job.addFollowOnJobFn(gather_fas, split_gfa_job.rv(0), split_fas_job.rv(0), split_fas_job.rv(1)) + gather_fas_job = split_fas_job.addFollowOnJobFn(gather_fas, split_gfa_job.rv(0), split_fas_job.rv(0), split_fas_job.rv(1), walltime=cactus_fast_walltime()) # lump "other" contigs together into one file (to make fewer align jobs downstream) bin_other_job = gather_fas_job.addFollowOnJobFn(bin_other_contigs, config, ref_contigs, options.otherContig, gather_fas_job.rv(0), - disk=(gfa_size + paf_size) * 2) + disk=(gfa_size + paf_size) * 2, walltime=cactus_fast_walltime()) # return all the files, as well as the 2 split logs return (seq_name_map, bin_other_job.rv(), split_gfa_job.rv(1), gather_fas_job.rv(1)) diff --git a/src/cactus/refmap/cactus_minigraph.py b/src/cactus/refmap/cactus_minigraph.py index 5a238a0c2..fa8e5dee7 100644 --- a/src/cactus/refmap/cactus_minigraph.py +++ b/src/cactus/refmap/cactus_minigraph.py @@ -17,12 +17,12 @@ import gzip from cactus.progressive.seqFile import SeqFile -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles, write_s3 from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.common import clean_jobstore_files @@ -43,6 +43,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("seqFile", help = "Seq file (or chromfile with --batch)") parser.add_argument("outputGFA", help = "Output Minigraph GFA (or directory in --batch mode)") @@ -124,7 +125,7 @@ def main(): input_dict = minigraph_construct_import_sequences(options, config_wrapper, input_seqfiles, toil) # output_dict: chrom-> (gfa_id, pansn_gfa_id, train_id) - output_dict = toil.start(Job.wrapJobFn(minigraph_construct_batch_workflow, options, config_node, input_dict, options.outputGFA)) + output_dict = toil.start(Job.wrapJobFn(minigraph_construct_batch_workflow, options, config_node, input_dict, options.outputGFA, walltime=cactus_fast_walltime())) export_minigraph_construct_output(options, input_seqfiles, output_dict, toil) @@ -263,7 +264,7 @@ def minigraph_construct_batch_workflow(job, options, config_node, input_dict, gf gfa_path = os.path.join(options.outputGFA, '{}.gfa.gz'.format(chrom)) else: gfa_path = options.outputGFA - mgwf_job = job.addChildJobFn(minigraph_construct_workflow, options, config_node, seq_id_map, seq_order, gfa_path, sanitize) + mgwf_job = job.addChildJobFn(minigraph_construct_workflow, options, config_node, seq_id_map, seq_order, gfa_path, sanitize, walltime=cactus_fast_walltime()) output_dict[chrom] = mgwf_job.rv() return output_dict @@ -281,7 +282,7 @@ def minigraph_construct_workflow(job, options, config_node, seq_id_map, seq_orde seq_id_map, seq_order = refonly_seq_id_map, refonly_seq_order ref_size = seq_id_map[options.reference[0]].size if sanitize: - sanitize_job = job.addChildJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True) + sanitize_job = job.addChildJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True, walltime=cactus_fast_walltime()) sanitized_seq_id_map = sanitize_job.rv() else: sanitized_seq_id_map = seq_id_map @@ -290,12 +291,12 @@ def minigraph_construct_workflow(job, options, config_node, seq_id_map, seq_orde xml_node = findRequiredNode(config_node, "graphmap") sort_type = getOptionalAttrib(xml_node, "minigraphSortInput", str, default=None) if sort_type == "mash" and len(seq_id_map) > 2: - sort_job = sanitize_job.addFollowOnJobFn(sort_minigraph_input_with_mash, options, config_node, sanitized_seq_id_map, seq_order) + sort_job = sanitize_job.addFollowOnJobFn(sort_minigraph_input_with_mash, options, config_node, sanitized_seq_id_map, seq_order, walltime=cactus_fast_walltime()) seq_order = sort_job.rv() prev_job = sort_job else: prev_job = sanitize_job - minigraph_job = prev_job.addFollowOnJobFn(minigraph_construct_in_batches, options, config_node, sanitized_seq_id_map, seq_order, gfa_path) + minigraph_job = prev_job.addFollowOnJobFn(minigraph_construct_in_batches, options, config_node, sanitized_seq_id_map, seq_order, gfa_path, walltime=cactus_fast_walltime()) train_id = None if options.lastTrain and len(seq_id_map) > 1: # note: somehow last training memory overruns don't seem to be detected by slurm so we @@ -344,7 +345,7 @@ def sort_minigraph_input_with_mash(job, options, config_node, seq_id_map, seq_or disk = 2 * sum(seq_id_map[x].size for x in names) + seq_id_map[seq_order[0]].size).rv() dist_maps.append(dist_map) - return dist_root_job.addFollowOnJobFn(mash_distance_order, options, config_node, seq_order, dist_maps).rv() + return dist_root_job.addFollowOnJobFn(mash_distance_order, options, config_node, seq_order, dist_maps, walltime=cactus_fast_walltime()).rv() def mash_sketch(job, ref_seq, seq_id_map): """ get the sketch """ @@ -495,7 +496,7 @@ def minigraph_construct_in_batches(job, options, config_node, seq_id_map, seq_or if prev_job: prev_job.addFollowOn(minigraph_job) # delete the output of the previous batch from the job store - minigraph_job.addFollowOnJobFn(clean_jobstore_files, file_ids=[prev_job.rv()]) + minigraph_job.addFollowOnJobFn(clean_jobstore_files, file_ids=[prev_job.rv()], walltime=cactus_fast_walltime()) else: job.addChild(minigraph_job) prev_job = minigraph_job diff --git a/src/cactus/refmap/cactus_pangenome.py b/src/cactus/refmap/cactus_pangenome.py index 826c990f0..5cc96a354 100644 --- a/src/cactus/refmap/cactus_pangenome.py +++ b/src/cactus/refmap/cactus_pangenome.py @@ -21,12 +21,12 @@ from operator import itemgetter from cactus.progressive.seqFile import SeqFile -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import cactusRootPath from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import cactus_call from cactus.shared.common import getOptionalAttrib, findRequiredNode from cactus.shared.common import clean_jobstore_files @@ -217,6 +217,7 @@ def pangenome_config_overrides(options, config_node): def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("seqFile", help = "Seq file (will be modified if necessary to include graph Fasta sequence)") parser.add_argument("--outDir", help = "Output directory", required=True) @@ -293,7 +294,7 @@ def main(): elif genome in input_seq_order: input_seq_order.remove(genome) - toil.start(Job.wrapJobFn(pangenome_end_to_end_workflow, options, config_wrapper, input_seq_id_map, input_path_map, input_seq_order, ref_collapse_paf_id, last_scores_id)) + toil.start(Job.wrapJobFn(pangenome_end_to_end_workflow, options, config_wrapper, input_seq_id_map, input_path_map, input_seq_order, ref_collapse_paf_id, last_scores_id, walltime=cactus_fast_walltime())) end_time = timeit.default_timer() run_time = end_time - start_time @@ -370,7 +371,7 @@ def sanitize_fasta_headers_batch(job, chromfile_id_map): seq_id_map = value[0] else: seq_id_map = value - sanitize_job = job.addChildJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True) + sanitize_job = job.addChildJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True, walltime=cactus_fast_walltime()) if type(value) is list: out_id_map[chrom] = [sanitize_job.rv()] + value[1:] else: @@ -489,7 +490,7 @@ def pangenome_end_to_end_workflow(job, options, config_wrapper, seq_id_map, seq_ root_job = vcflib_checks(root_job, options, config_node) # sanitize headers (once here, skip in all workflows below) - sanitize_job = root_job.addFollowOnJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True) + sanitize_job = root_job.addFollowOnJobFn(sanitize_fasta_headers, seq_id_map, pangenome=True, walltime=cactus_fast_walltime()) seq_id_map = sanitize_job.rv() assert type(options.reference) == list @@ -514,13 +515,13 @@ def pangenome_end_to_end_workflow(job, options, config_wrapper, seq_id_map, seq_ else: split_config_node = config_node split_config_wrapper = config_wrapper - minigraph_job = sanitize_job.addFollowOnJobFn(minigraph_construct_workflow, mg_options, split_config_node, seq_id_map, seq_order, sv_gfa_path, sanitize=False) + minigraph_job = sanitize_job.addFollowOnJobFn(minigraph_construct_workflow, mg_options, split_config_node, seq_id_map, seq_order, sv_gfa_path, sanitize=False, walltime=cactus_fast_walltime()) sv_gfa_id = minigraph_job.rv(0) pansn_sv_gfa_id = minigraph_job.rv(1) if not last_scores_id: last_scores_id = minigraph_job.rv(2) # only build reference graph on first pass when doing minigraph-by-chrom pipeline - minigraph_wrapper_job = minigraph_job.addFollowOnJobFn(export_minigraph_wrapper, options, pansn_sv_gfa_id, sv_gfa_path, last_scores_id) + minigraph_wrapper_job = minigraph_job.addFollowOnJobFn(export_minigraph_wrapper, options, pansn_sv_gfa_id, sv_gfa_path, last_scores_id, walltime=cactus_fast_walltime()) # cactus_graphmap paf_path = os.path.join(options.outDir, options.outName + '.paf') @@ -531,31 +532,31 @@ def pangenome_end_to_end_workflow(job, options, config_wrapper, seq_id_map, seq_ gm_options = copy.deepcopy(options) if options.mgSplit: gm_options.collapse = False - graphmap_job = minigraph_wrapper_job.addFollowOnJobFn(minigraph_workflow, gm_options, split_config_wrapper, seq_id_map, sv_gfa_id, graph_event, False, ref_collapse_paf_id, pansn_gfa_input=False) + graphmap_job = minigraph_wrapper_job.addFollowOnJobFn(minigraph_workflow, gm_options, split_config_wrapper, seq_id_map, sv_gfa_id, graph_event, False, ref_collapse_paf_id, pansn_gfa_input=False, walltime=cactus_fast_walltime()) paf_id, gfa_fa_id, gaf_id, unfiltered_paf_id, paf_filter_log = graphmap_job.rv(0), graphmap_job.rv(1), graphmap_job.rv(2), graphmap_job.rv(3), graphmap_job.rv(4) - graphmap_export_job = graphmap_job.addFollowOnJobFn(export_graphmap_wrapper, options, paf_id, paf_path, gaf_id, unfiltered_paf_id, paf_filter_log) + graphmap_export_job = graphmap_job.addFollowOnJobFn(export_graphmap_wrapper, options, paf_id, paf_path, gaf_id, unfiltered_paf_id, paf_filter_log, walltime=cactus_fast_walltime()) # we need to update the seqfile with the phonied in minigraph event - update_seqfile_job = graphmap_export_job.addFollowOnJobFn(update_seqfile, options, seq_id_map, seq_path_map, seq_order, gfa_fa_id, gfa_fa_path, graph_event) + update_seqfile_job = graphmap_export_job.addFollowOnJobFn(update_seqfile, options, seq_id_map, seq_path_map, seq_order, gfa_fa_id, gfa_fa_path, graph_event, walltime=cactus_fast_walltime()) seq_id_map, seq_path_map, seq_name_map = update_seqfile_job.rv(0), update_seqfile_job.rv(1), update_seqfile_job.rv(2) if options.noSplit: # we phony in the entire alignment as one chromsome called 'all' - phony_chromfile_job = update_seqfile_job.addFollowOnJobFn(phony_chromfile, options, paf_path) + phony_chromfile_job = update_seqfile_job.addFollowOnJobFn(phony_chromfile, options, paf_path, walltime=cactus_fast_walltime()) chromfile_path = phony_chromfile_job.rv() split_export_job = phony_chromfile_job else: # cactus_graphmap_split split_job = update_seqfile_job.addFollowOnJobFn(graphmap_split_workflow, options, split_config_wrapper, seq_id_map, seq_name_map, sv_gfa_id, - sv_gfa_path, paf_id, paf_path, sanitize=False, pansn_gfa_input=False) + sv_gfa_path, paf_id, paf_path, sanitize=False, pansn_gfa_input=False, walltime=cactus_fast_walltime()) wf_output = split_job.rv() split_out_path = os.path.join(options.outDir, 'chrom-subproblems') - split_export_job = split_job.addFollowOnJobFn(export_split_wrapper, wf_output, split_out_path, split_config_wrapper) + split_export_job = split_job.addFollowOnJobFn(export_split_wrapper, wf_output, split_out_path, split_config_wrapper, walltime=cactus_fast_walltime()) chromfile_path = os.path.join(split_out_path, 'chromfile.txt') # clean out some jobstore files we no longer need clean_jobstore_job = split_export_job.addFollowOnJobFn(clean_jobstore_files, file_id_maps=[seq_id_map] if not options.noSplit else None, - file_ids=[sv_gfa_id, paf_id]) + file_ids=[sv_gfa_id, paf_id], walltime=cactus_fast_walltime()) options.batch = True minigraph_pansn_sv_gfa_ids = [] @@ -565,14 +566,14 @@ def pangenome_end_to_end_workflow(job, options, config_wrapper, seq_id_map, seq_ chromfile_path) input_seqfiles = minigraph_batch_import_job.rv(0) raw_input_map = minigraph_batch_import_job.rv(1) - sanitize_job = minigraph_batch_import_job.addFollowOnJobFn(sanitize_fasta_headers_batch, raw_input_map) + sanitize_job = minigraph_batch_import_job.addFollowOnJobFn(sanitize_fasta_headers_batch, raw_input_map, walltime=cactus_fast_walltime()) input_map = sanitize_job.rv() options.outputGFA='' minigraph_batch_job = sanitize_job.addFollowOnJobFn(minigraph_construct_batch_workflow, options, config_node, - input_map, None, sanitize=False) + input_map, None, sanitize=False, walltime=cactus_fast_walltime()) minigraph_batch_results = minigraph_batch_job.rv() minigraph_batch_export_job = minigraph_batch_job.addFollowOnJobFn(export_minigraph_batch_wrapper, options, config_node, - input_seqfiles, input_map, minigraph_batch_results) + input_seqfiles, input_map, minigraph_batch_results, walltime=cactus_fast_walltime()) # now rerun cactus_graphmap but on a per-chromosome bassis graphmap_input_dict = minigraph_batch_export_job.rv(0) @@ -581,16 +582,16 @@ def pangenome_end_to_end_workflow(job, options, config_wrapper, seq_id_map, seq_ minigraph_pansn_sv_gfa_ids = minigraph_batch_export_job.rv(3) graphmap_batch_job = minigraph_batch_export_job.addFollowOnJobFn(minigraph_batch_workflow, options, config_wrapper, graphmap_input_dict, graph_event, sanitize=False, - pansn_gfa_input=False) + pansn_gfa_input=False, walltime=cactus_fast_walltime()) graphmap_batch_results = graphmap_batch_job.rv() graphmap_batch_export_job = graphmap_batch_job.addFollowOnJobFn(export_graphmap_batch_wrapper, options, config_node, - graphmap_batch_results, input_seqfiles) + graphmap_batch_results, input_seqfiles, walltime=cactus_fast_walltime()) graphmap_file_ids = graphmap_batch_export_job.rv(0) chromfile_path = graphmap_batch_export_job.rv(1) # clean out the jobstore, as cactus_align reads everything from disk - clean_jobstore_job = graphmap_batch_export_job.addFollowOnJobFn(clean_jobstore_files, file_ids=graphmap_file_ids) + clean_jobstore_job = graphmap_batch_export_job.addFollowOnJobFn(clean_jobstore_files, file_ids=graphmap_file_ids, walltime=cactus_fast_walltime()) clean_jobstore_job = clean_jobstore_job.addFollowOnJobFn(clean_jobstore_files, file_id_maps=minigraph_output_maps, - file_ids=minigraph_output_ids, allow_none=True) + file_ids=minigraph_output_ids, allow_none=True, walltime=cactus_fast_walltime()) # cactus_align options.scoresFromChromfile = options.lastTrain and options.mgSplit @@ -598,7 +599,7 @@ def pangenome_end_to_end_workflow(job, options, config_wrapper, seq_id_map, seq_ last_scores_id) align_jobs = align_jobs_make_job.rv() - align_job = align_jobs_make_job.addFollowOnJobFn(batch_align_jobs, align_jobs) + align_job = align_jobs_make_job.addFollowOnJobFn(batch_align_jobs, align_jobs, walltime=cactus_fast_walltime()) results_dict = align_job.rv() align_export_job = align_job.addFollowOnJobFn(export_align_wrapper, options, results_dict) join_options, vg_ids, hal_ids = align_export_job.rv(0), align_export_job.rv(1), align_export_job.rv(2) @@ -608,16 +609,16 @@ def pangenome_end_to_end_workflow(job, options, config_wrapper, seq_id_map, seq_ # hal_ids, and it's also the only thing that frees the chromosome hals from the jobstore, so we # need to do that ourselves here or they'd leak for the rest of the run join_job = align_export_job.addFollowOnJobFn(graphmap_join_workflow, join_options, config_wrapper, vg_ids, - [] if options.noHal else hal_ids, minigraph_pansn_sv_gfa_ids) + [] if options.noHal else hal_ids, minigraph_pansn_sv_gfa_ids, walltime=cactus_fast_walltime()) join_wf_output = join_job.rv() if options.noHal: - join_job.addFollowOnJobFn(clean_jobstore_files, file_ids=hal_ids) + join_job.addFollowOnJobFn(clean_jobstore_files, file_ids=hal_ids, walltime=cactus_fast_walltime()) # cactus-panpatch exports the chromosome vgs itself (they're the only thing export_join_data # would write, given it turns every other output off), so let it skip this to avoid writing # the biggest output twice if not options.noJoinExport: - join_job.addFollowOnJobFn(export_join_wrapper, join_options, join_wf_output) + join_job.addFollowOnJobFn(export_join_wrapper, join_options, join_wf_output, walltime=cactus_fast_walltime()) return join_options, join_wf_output, seq_id_map diff --git a/src/cactus/refmap/cactus_panpatch.py b/src/cactus/refmap/cactus_panpatch.py index e158edc47..d0ca51ce4 100644 --- a/src/cactus/refmap/cactus_panpatch.py +++ b/src/cactus/refmap/cactus_panpatch.py @@ -28,7 +28,7 @@ from cactus.progressive.seqFile import SeqFile from cactus.progressive.cactus_prepare import human2bytesN -from cactus.shared.common import importSingularityImage +from cactus.shared.common import importSingularityImage, cactus_fast_walltime, add_cactus_toil_options from cactus.shared.common import makeURL from cactus.shared.common import cactus_call from cactus.shared.common import cactus_clamp_memory @@ -50,6 +50,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("seqFile", nargs='?', default=None, help = "Seq file (as with cactus-pangenome), or, with --batch, a chromfile with one " @@ -222,7 +223,7 @@ def main(): ref_collapse_paf_id, last_scores_id, target_fasta_ids)) toil.start(Job.wrapJobFn(panpatch_batch_workflow, options, config_wrapper, run_inputs, - exclude_bed_id)) + exclude_bed_id, walltime=cactus_fast_walltime())) end_time = timeit.default_timer() run_time = end_time - start_time @@ -477,18 +478,18 @@ def panpatch_batch_workflow(job, options, config_wrapper, run_inputs, exclude_be for run, pg_options, seq_id_map, seq_path_map, seq_order, ref_collapse_paf_id, last_scores_id, target_fasta_ids in run_inputs: job.addChildJobFn(panpatch_run_workflow, options, pg_options, config_wrapper, run, seq_id_map, seq_path_map, seq_order, ref_collapse_paf_id, last_scores_id, exclude_bed_id, - target_fasta_ids) + target_fasta_ids, walltime=cactus_fast_walltime()) def panpatch_run_workflow(job, options, pg_options, config_wrapper, run, seq_id_map, seq_path_map, seq_order, ref_collapse_paf_id, last_scores_id, exclude_bed_id, target_fasta_ids): """ cactus-pangenome, then panpatch on the chromosome graphs it made """ pangenome_job = job.addChildJobFn(pangenome_end_to_end_workflow, pg_options, config_wrapper, seq_id_map, - seq_path_map, seq_order, ref_collapse_paf_id, last_scores_id) + seq_path_map, seq_order, ref_collapse_paf_id, last_scores_id, walltime=cactus_fast_walltime()) # a follow-on of the job hosting the pangenome workflow only runs once that job's entire child # subtree is done, which is what makes this safe pangenome_job.addFollowOnJobFn(panpatch_workflow, options, run, pangenome_job.rv(0), pangenome_job.rv(1), - pangenome_job.rv(2), exclude_bed_id, target_fasta_ids) + pangenome_job.rv(2), exclude_bed_id, target_fasta_ids, walltime=cactus_fast_walltime()) def panpatch_workflow(job, options, run, join_options, join_wf_output, seq_id_map, exclude_bed_id, target_fasta_ids): """ the pangenome's promises have resolved by now. run one single-threaded panpatch job per @@ -526,7 +527,7 @@ def panpatch_workflow(job, options, run, join_options, join_wf_output, seq_id_ma target_fasta_ids, memory=cactus_clamp_memory(max(2**32, ref_size * 2)), disk=gather_disk) gather_job.addFollowOnJobFn(export_panpatch_wrapper, options, run, gather_job.rv(), full_vg_ids, vg_names, - disk=sum(vg_id.size for vg_id in full_vg_ids) * 2 + 2**30) + disk=sum(vg_id.size for vg_id in full_vg_ids) * 2 + 2**30, walltime=cactus_fast_walltime()) def run_panpatch_chrom(job, options, run, vg_id, vg_name, exclude_bed_id): """ run single-threaded panpatch on one chromosome graph. returns the per-haplotype fastas, the @@ -720,7 +721,7 @@ def export_panpatch_wrapper(job, options, run, output_id_map, full_vg_ids, vg_na job.fileStore.exportFile(vg_id, makeURL(os.path.join(chrom_dir, vg_name))) if not options.keepPangenome: - job.addFollowOnJobFn(cleanup_pangenome_wrapper, options, run) + job.addFollowOnJobFn(cleanup_pangenome_wrapper, options, run, walltime=cactus_fast_walltime()) def cleanup_pangenome_wrapper(job, options, run): """ the cactus-pangenome output is only ever scratch for us. everything the user asked for has diff --git a/src/cactus/refmap/cactus_refmap.py b/src/cactus/refmap/cactus_refmap.py index 4c2f2c867..9f9ccf904 100644 --- a/src/cactus/refmap/cactus_refmap.py +++ b/src/cactus/refmap/cactus_refmap.py @@ -36,12 +36,12 @@ from cactus.refmap import fasta_preprocessing from cactus.refmap import apply_dipcall_bed_filter -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import makeURL from cactus.shared.common import cactus_call from cactus.shared.configWrapper import ConfigWrapper from cactus.shared.common import cactusRootPath -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.progressive.progressive_decomposition import compute_outgroups, parse_seqfile, get_subtree, get_spanning_subtree, get_event_set from cactus.preprocessor.checkUniqueHeaders import sanitize_fasta_headers @@ -139,8 +139,8 @@ def run_cactus_reference_align(job, assembly_files, reference, debug_export=Fals """ Preprocesses assemblies, then runs mappings. """ - sanitize_job = job.addChildJobFn(sanitize_fasta_headers, assembly_files) - mappings = sanitize_job.addFollowOnJobFn(map_all_to_ref, sanitize_job.rv(), reference, debug_export, dipcall_bed_filter, dipcall_vcf_filter).rv() + sanitize_job = job.addChildJobFn(sanitize_fasta_headers, assembly_files, walltime=cactus_fast_walltime()) + mappings = sanitize_job.addFollowOnJobFn(map_all_to_ref, sanitize_job.rv(), reference, debug_export, dipcall_bed_filter, dipcall_vcf_filter, walltime=cactus_fast_walltime()).rv() return mappings def map_all_to_ref(job, assembly_files, reference, debug_export=False, dipcall_bed_filter=False, dipcall_vcf_filter=False): @@ -162,7 +162,7 @@ def map_all_to_ref(job, assembly_files, reference, debug_export=False, dipcall_b * Filters out all mappings below min_var_len=50k and min_mapq=5 from a lastz file Defaults to False. """ - lead_job = job.addChildJobFn(empty) + lead_job = job.addChildJobFn(empty, walltime=cactus_fast_walltime()) # map all assemblies to the reference. Don't map reference to reference, though. ref_mappings = dict() @@ -228,6 +228,7 @@ def map_a_to_b(job, a, b, dipcall_filter): def get_options(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) # addCactusWorkflowOptions(parser) # ### For quick debugging of apply_dipcall_bed_filter: @@ -327,7 +328,7 @@ def main(): ## Perform alignments: if not toil.options.restart: - alignments = toil.start(Job.wrapJobFn(run_cactus_reference_align, input_seq_id_map, options.reference, options.debug_export, options.dipcall_bed_filter, options.dipcall_vcf_filter)) + alignments = toil.start(Job.wrapJobFn(run_cactus_reference_align, input_seq_id_map, options.reference, options.debug_export, options.dipcall_bed_filter, options.dipcall_vcf_filter, walltime=cactus_fast_walltime())) else: alignments = toil.restart() diff --git a/src/cactus/setup/cactus_align.py b/src/cactus/setup/cactus_align.py index 3bd92d14c..a1b1094c6 100644 --- a/src/cactus/setup/cactus_align.py +++ b/src/cactus/setup/cactus_align.py @@ -15,13 +15,13 @@ import multiprocessing from operator import itemgetter -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.pipeline.cactus_workflow import cactus_cons_with_resources from cactus.progressive.progressive_decomposition import compute_outgroups, parse_seqfile, get_subtree, get_spanning_subtree, get_event_set, get_ancestor_scaled_tree from cactus.progressive.cactus_progressive import export_hal from cactus.shared.common import makeURL, catFiles from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import findRequiredNode from cactus.shared.common import getOptionalAttrib from cactus.shared.common import cactus_call @@ -48,6 +48,7 @@ def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("seqFile", help = "Seq file (or chromfile with --batch)") parser.add_argument("pafFile", nargs='?', default='', type=str, help = "Pairiwse aliginments (from cactus-blast, cactus-refmap or cactus-graphmap)") @@ -190,7 +191,7 @@ def main(): results_dict = toil.restart() else: align_jobs = make_batch_align_jobs(options, toil) - results_dict = toil.start(Job.wrapJobFn(batch_align_jobs, align_jobs)) + results_dict = toil.start(Job.wrapJobFn(batch_align_jobs, align_jobs, walltime=cactus_fast_walltime())) # when using s3 output urls, things get checkpointed as they're made so no reason to export # todo: make a more unified interface throughout cactus for this @@ -398,7 +399,7 @@ def make_align_job(options, toil, config_wrapper=None, chrom_name=None): do_filter_paf=options.pangenome, chrom_name=chrom_name, scores_id=scores_id, - branch_scale=options.branchScale) + branch_scale=options.branchScale, walltime=cactus_fast_walltime()) return align_job def cactus_align(job, config_wrapper, mc_tree, input_seq_map, input_seq_id_map, paf_id, paf_path, root_name, og_map, checkpointInfo, doVG, doGFA, delay=0, @@ -422,7 +423,7 @@ def cactus_align(job, config_wrapper, mc_tree, input_seq_map, input_seq_id_map, apply_scores_to_config(score_dict, config_wrapper.xmlRoot) # unzip the input sequences and enforce unique header prefixes - sanitize_job = head_job.addChildJobFn(sanitize_fasta_headers, input_seq_id_map, pangenome=doVG or doGFA or do_filter_paf) + sanitize_job = head_job.addChildJobFn(sanitize_fasta_headers, input_seq_id_map, pangenome=doVG or doGFA or do_filter_paf, walltime=cactus_fast_walltime()) new_seq_id_map = sanitize_job.rv() # run pangenome-specific paf filter @@ -446,7 +447,7 @@ def cactus_align(job, config_wrapper, mc_tree, input_seq_map, input_seq_id_map, # run consolidated cons_job = head_job.addFollowOnJobFn(cactus_cons_with_resources, spanning_tree, root_name, config_wrapper.xmlRoot, new_seq_id_map, og_map, paf_id, - cons_cores = cons_cores, cons_memory=cons_memory, chrom_name=chrom_name) + cons_cores = cons_cores, cons_memory=cons_memory, chrom_name=chrom_name, walltime=cactus_fast_walltime()) results = {root_name : (cons_job.rv(1), cons_job.rv(2))} # get the immediate subtree (which is all export_hal can use) @@ -459,7 +460,7 @@ def cactus_align(job, config_wrapper, mc_tree, input_seq_map, input_seq_id_map, memory_override=cons_memory) # clean out some of the intermediate jobstore files - hal_job.addFollowOnJobFn(clean_jobstore_files, file_id_maps=[new_seq_id_map], file_ids=[paf_id]) + hal_job.addFollowOnJobFn(clean_jobstore_files, file_id_maps=[new_seq_id_map], file_ids=[paf_id], walltime=cactus_fast_walltime()) # optionally create the VG if doVG or doGFA: @@ -546,6 +547,7 @@ def main_batch(): cons: less efficient use of resources """ parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("chromFile", help = "chroms file") parser.add_argument("outHal", type=str, help = "Output directory (can be s3://)") @@ -630,7 +632,7 @@ def main_batch(): chrom_dict[chrom] = toil.importFile(makeURL(seqfile)), toil.importFile(makeURL(alnFile)) if chrom in options.configOverrides: options.configOverrides[chrom][1] = toil.importFile(makeURL(options.configOverrides[chrom][0])) - results_dict = toil.start(Job.wrapJobFn(align_toil_batch, chrom_dict, config_id, options)) + results_dict = toil.start(Job.wrapJobFn(align_toil_batch, chrom_dict, config_id, options, walltime=cactus_fast_walltime())) # when using s3 output urls, things get checkpointed as they're made so no reason to export # todo: make a more unified interface throughout cactus for this diff --git a/src/cactus/shared/common.py b/src/cactus/shared/common.py index bf3d83f8d..bd6acabe3 100644 --- a/src/cactus/shared/common.py +++ b/src/cactus/shared/common.py @@ -107,6 +107,13 @@ def cactus_override_toil_options(options): os.environ['CACTUS_MAX_MEMORY'] = str(max_mem) os.environ['CACTUS_DEFAULT_MEMORY'] = str(human2bytes(str(options.defaultMemory)) if options.defaultMemory else 2**31) + # store the "fast" walltime (in seconds) for small coordination jobs here so we can get + # at it without carrying options around, and so it propagates to workers (see + # cactus_fast_walltime). An explicit --fastWalltime wins over any inherited env var. + fast_walltime = getattr(options, 'fastWalltime', None) + if fast_walltime: + os.environ['CACTUS_FAST_WALLTIME'] = str(int(fast_walltime)) + # auto-set cactus_log_memory try: subprocess.check_call(['/usr/bin/time', '-v', 'ls'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) @@ -118,6 +125,29 @@ def cactus_clamp_memory(memory_bytes): """ use the environment variables from --maxMemory and --defaultMemory to clamp a given memory value """ return max(min(int(os.environ['CACTUS_MAX_MEMORY']), int(memory_bytes)), int(os.environ['CACTUS_DEFAULT_MEMORY'])) +def cactus_fast_walltime(): + """ Walltime, in seconds, to assign to small "coordination" jobs (ones that just schedule + other jobs or do trivial work) so that Slurm can route them to a fast partition. Set via + --fastWalltime, which stores CACTUS_FAST_WALLTIME so the value reaches workers just like + cactus_clamp_memory (or export CACTUS_FAST_WALLTIME directly). Returns None (ie no walltime + override, so the job falls back to Toil's --defaultWalltime) when it isn't set. """ + val = os.environ.get('CACTUS_FAST_WALLTIME') + if not val: + return None + seconds = int(val) + # treat 0 (or negative) as "disabled" so it stays a no-op rather than requesting a + # 0-second job (which Slurm would map to its shortest partition) + return seconds if seconds > 0 else None + +def add_cactus_toil_options(parser): + """ Add cactus-specific options on top of Toil's default argument parser + (Job.Runner.getDefaultArgumentParser). Call this right after creating the parser. """ + parser.add_argument("--fastWalltime", type=int, default=1800, + help="Walltime, in seconds, to assign to small \"coordination\" jobs so that " + "(on Slurm) Toil can route them to a fast partition [default: 1800 (30 min), " + "which is very conservative for these jobs]. Set to 0 to disable. Pair with " + "Toil's --defaultWalltime, which sets the walltime for every other job.") + def makeURL(path_or_url): if urlparse(path_or_url).scheme == '': return "file://" + os.path.abspath(path_or_url) @@ -945,7 +975,7 @@ class RoundedJob(Job): # Default rounding amount: 100 MiB roundingAmount = 100*1024*1024 def __init__(self, memory=None, cores=None, disk=None, preemptable=None, - unitName=None, checkpoint=False, accelerators=None): + unitName=None, checkpoint=False, accelerators=None, walltime=None): if memory is not None: memory = self.roundUp(memory) if disk is not None: @@ -955,7 +985,8 @@ def __init__(self, memory=None, cores=None, disk=None, preemptable=None, disk = 1500*1024*1024 + self.roundUp(disk) super(RoundedJob, self).__init__(memory=memory, cores=cores, disk=disk, preemptable=preemptable, unitName=unitName, - checkpoint=checkpoint, accelerators=accelerators) + checkpoint=checkpoint, accelerators=accelerators, + walltime=walltime) def roundUp(self, bytesRequirement): """ @@ -1005,12 +1036,12 @@ class ChildTreeJob(RoundedJob): slightly, but reducing the wall-clock time taken dramatically. """ def __init__(self, memory=None, cores=None, disk=None, preemptable=None, - unitName=None, checkpoint=False, maxChildrenPerJob=20): + unitName=None, checkpoint=False, maxChildrenPerJob=20, walltime=None): self.queuedChildJobs = [] self.maxChildrenPerJob = maxChildrenPerJob super(ChildTreeJob, self).__init__(memory=memory, cores=cores, disk=disk, preemptable=preemptable, unitName=unitName, - checkpoint=checkpoint) + checkpoint=checkpoint, walltime=walltime) def addChild(self, job): self.queuedChildJobs.append(job) diff --git a/src/cactus/update/cactus_hal2seqfile.py b/src/cactus/update/cactus_hal2seqfile.py index f443a4175..0eab5633f 100644 --- a/src/cactus/update/cactus_hal2seqfile.py +++ b/src/cactus/update/cactus_hal2seqfile.py @@ -30,9 +30,9 @@ from toil.statsAndLogging import logger from toil.statsAndLogging import set_logging_from_options -from cactus.shared.common import setupBinaries, importSingularityImage +from cactus.shared.common import setupBinaries, importSingularityImage, cactus_fast_walltime from cactus.shared.common import enableDumpStack -from cactus.shared.common import cactus_override_toil_options +from cactus.shared.common import cactus_override_toil_options, add_cactus_toil_options from cactus.shared.common import makeURL, cactus_call, cactus_clamp_memory from cactus.shared.version import cactus_commit @@ -117,6 +117,7 @@ def hal2fasta_gz(job, hal_id, hal_name, genome): def main(): parser = Job.Runner.getDefaultArgumentParser() + add_cactus_toil_options(parser) parser.add_argument("halFile", help="input HAL alignment") parser.add_argument("outDir", help="output directory for the gzipped per-genome FASTAs (created if needed)") @@ -170,7 +171,7 @@ def main(): # defaults to symlink=True in Toil, but we set it explicitly so this holds within the tool hal_id = toil.importFile(makeURL(options.halFile), symlink=True) fa_ids = toil.start(Job.wrapJobFn(export_subtree_fastas, hal_id, - os.path.basename(options.halFile), genomes, lengths)) + os.path.basename(options.halFile), genomes, lengths, walltime=cactus_fast_walltime())) # export each gzipped fasta to /.fa.gz for genome, fa_id in fa_ids.items(): diff --git a/toil-requirement.txt b/toil-requirement.txt index 30248feae..f6fd9eecb 100644 --- a/toil-requirement.txt +++ b/toil-requirement.txt @@ -1,2 +1,4 @@ backports.zoneinfo[tzdata];python_version<"3.9" -toil[aws]==9.5.0 +# Pinned to Toil master for per-job walltime support (not yet in a release: 9.5.0 lacks it). +# TODO: revert to a released version (toil[aws]==) once walltime lands in a Toil release. +toil[aws] @ git+https://github.com/DataBiosphere/toil.git@091ed8861390c06230495819ba0fbef4cc4db5a0