Source code for toil.batchSystems.options

# Copyright (C) 2015-2021 Regents of the University of California
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and

import logging
import sys
from argparse import ArgumentParser, _ArgumentGroup
from typing import Any, Callable, List, Optional, TypeVar, Union

if sys.version_info >= (3, 8):
    from typing import Protocol
    from typing_extensions import Protocol

from toil.batchSystems.registry import (DEFAULT_BATCH_SYSTEM,
from toil.lib.threading import cpu_count

logger = logging.getLogger(__name__)

[docs] class OptionSetter(Protocol): """ Protocol for the setOption function we get to let us set up CLI options for each batch system. Actual functionality is defined in the Config class. """ OptionType = TypeVar('OptionType')
[docs] def __call__( self, option_name: str, parsing_function: Optional[Callable[[Any], OptionType]] = None, check_function: Optional[Callable[[OptionType], Union[None, bool]]] = None, default: Optional[OptionType] = None, env: Optional[List[str]] = None, old_names: Optional[List[str]] = None ) -> bool: ...
[docs] def set_batchsystem_options(batch_system: Optional[str], set_option: OptionSetter) -> None: """ Call set_option for all the options for the given named batch system, or all batch systems if no name is provided. """ if batch_system is not None: # Use this batch system batch_system_type = get_batch_system(batch_system) batch_system_type.setOptions(set_option) else: for name in get_batch_systems(): # All the batch systems are responsible for setting their own options # with their setOptions() class methods. try: batch_system_type = get_batch_system(name) except ImportError: # Skip anything we can't import continue # Ask the batch system to tell us all its options. batch_system_type.setOptions(set_option) # Options shared between multiple batch systems set_option("disableAutoDeployment") # Make limits maximum if set to 0 set_option("max_jobs") set_option("max_local_jobs") set_option("manualMemArgs") set_option("run_local_jobs_on_workers") set_option("statePollingWait") set_option("batch_logs_dir")
[docs] def add_all_batchsystem_options(parser: Union[ArgumentParser, _ArgumentGroup]) -> None: from toil.common import SYS_MAX_SIZE # Do the global cross-batch-system arguments parser.add_argument( "--batchSystem", dest="batchSystem", default=DEFAULT_BATCH_SYSTEM, choices=get_batch_systems(), help=f"The type of batch system to run the job(s) with, currently can be one " f"of {', '.join(get_batch_systems())}. default={DEFAULT_BATCH_SYSTEM}", ) parser.add_argument( "--disableHotDeployment", dest="disableAutoDeployment", default=False, action="store_true", help="Hot-deployment was renamed to auto-deployment. Option now redirects to " "--disableAutoDeployment. Left in for backwards compatibility.", ) parser.add_argument( "--disableAutoDeployment", dest="disableAutoDeployment", default=False, help="Should auto-deployment of the user script be deactivated? If True, the user " "script/package should be present at the same location on all workers. Default = False.", ) parser.add_argument( "--maxJobs", dest="max_jobs", default=SYS_MAX_SIZE, # This is *basically* unlimited and saves a lot of Optional[] type=lambda x: int(x) or SYS_MAX_SIZE, help="Specifies the maximum number of jobs to submit to the " "backing scheduler at once. Not supported on Mesos or " "AWS Batch. Use 0 for unlimited. Defaults to unlimited.", ) parser.add_argument( "--maxLocalJobs", dest="max_local_jobs", default=None, type=lambda x: int(x) or 0, help=f"Specifies the maximum number of housekeeping jobs to " f"run sumultaneously on the local system. Use 0 for " f"unlimited. Defaults to the number of local cores ({cpu_count()}).", ) parser.add_argument( "--manualMemArgs", default=False, dest="manualMemArgs", action="store_true", help="Do not add the default arguments: 'hv=MEMORY' & 'h_vmem=MEMORY' to the qsub " "call, and instead rely on TOIL_GRIDGENGINE_ARGS to supply alternative arguments. " "Requires that TOIL_GRIDGENGINE_ARGS be set.", ) parser.add_argument( "--runLocalJobsOnWorkers", "--runCwlInternalJobsOnWorkers", dest="run_local_jobs_on_workers", default=False, action="store_true", help="Whether to run jobs marked as local (e.g. CWLScatter) on the worker nodes " "instead of the leader node. If false (default), then all such jobs are run on " "the leader node. Setting this to true can speed up CWL pipelines for very large " "workflows with many sub-workflows and/or scatters, provided that the worker " "pool is large enough.", ) parser.add_argument( "--coalesceStatusCalls", dest="coalesceStatusCalls", action="store_true", default=True, help=( "Ask for job statuses from the batch system in a batch. Deprecated; now always " "enabled where supported." ), ) parser.add_argument( "--statePollingWait", dest="statePollingWait", type=int, default=None, help="Time, in seconds, to wait before doing a scheduler query for job state. " "Return cached results if within the waiting period. Only works for grid " "engine batch systems such as gridengine, htcondor, torque, slurm, and lsf." ) parser.add_argument( "--batchLogsDir", dest="batch_logs_dir", default=None, env_var="TOIL_BATCH_LOGS_DIR", help="Directory to tell the backing batch system to log into. Should be available " "on both the leader and the workers, if the backing batch system writes logs " "to the worker machines' filesystems, as many HPC schedulers do. If unset, " "the Toil work directory will be used. Only works for grid engine batch " "systems such as gridengine, htcondor, torque, slurm, and lsf." ) for name in get_batch_systems(): # All the batch systems are responsible for adding their own options # with the add_options class method. try: batch_system_type = get_batch_system(name) except ImportError: # Skip anything we can't import continue # Ask the batch system to create its options in the parser logger.debug('Add options for %s batch system', name) batch_system_type.add_options(parser)