Source code for torchx.runner.config

#!/usr/bin/env python3
# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.

Status: Beta

You can store the scheduler run cfg (run configs) for your project
by storing them in the ``.torchxconfig`` file. Currently this file is only read
and honored when running the component from the CLI.

CLI Usage

**Scheduler Config**

#. ``cd`` into the directory where you want the ``.torchxconfig`` file to be dropped.
   The CLI only picks up ``.torchxconfig`` files from the current-working-directory (CWD)
   so chose a directory where you typically run ``torchx`` from. Typically this
   is the root of your project directory.

#. Generate the config file by running

   .. code-block:: shell-session

    $ torchx configure -s <comma,delimited,scheduler,names>

    # -- or for all registered schedulers --
    $ torchx configure

#. If you specified ``-s local_cwd,kubernetes``, you should see a ``.torchxconfig``
   file as shown below:

   .. code-block:: shell-session

    $ cat .torchxconfig

    queue = #FIXME:(str) Volcano queue to schedule job in

#. ``.torchxconfig`` in in INI format and the section names map to the scheduler names.
   Each section contains the run configs for the scheduler as ``$key = $value`` pairs.
   You may find that certain schedulers have empty sections, this means that
   the scheduler defines sensible defaults for all its run configs hence no run configs
   are required at runtime. If you'd like to override the default you can add them.
   **TIP:** To see all the run options for a scheduler use ``torchx runopts <scheduler_name>``.

#. The sections with ``FIXME`` placeholders are run configs that are required
   by the scheduler. Replace these with the values that apply to you.

#. **IMPORTANT:** If you are happy with the scheduler provided defaults for a particular
   run config, you **should not** redundantly specify them in ``.torchxconfig`` with the
   same default value. This is because the scheduler may decide to change the default
   value at a later date which would leave you with a stale default.

#. Now you can run your component without having to specify the scheduler run configs
   each time. Just make sure the directory you are running ``torchx`` cli from actually
   has ``.torchxconfig``!

   .. code-block:: shell-session

    $ ls .torchxconfig

    $ torchx run -s local_cwd ./

#. In addition, it is possible to specify a different config other than .torchxconfig to
   load at runtime. Requirements are that the config path is specified by enviornment
   variable TORCHX_CONFIG. It also disables hierarchy loading configs from multiple
   directories as the cases otherwise.

#. User level .torchxconfig
   In addition to the project-level .torchxconfig at the root of the project directory,
   you can create one in ``$HOME/.torchxconfig`` to override or specify additional default configs.
   This config file will get overlaid on top of the one defined at the project root.

#. Config options take the following precedence (high to low):
    1. Options specified directly from the CLI
    2. If TORCHXCONFIG env variable is set, the options specified in that file
    3. If TORCHXCONFIG env variable is not set,
        a. Options specified in user level .torchxconfig (``$HOME/.torchxconfig``)
        b. Options specified in .torchxconfig
    4. Any default values in the code

   Note that malformed or unrecognized options are simply skipped and not applied

**Component Config**

You can specify component defaults by adding a section prefixed with

.. code-block:: ini


Now when you run the ``dist.ddp`` component those configs are automatically
picked up.

.. code-block:: shell-session

    $ torchx run -s local_cwd dist.ddp
    ... runs with -j 2x8 --cpu 4

**CLI Subcommand Config**

The default arguments for the ``torchx`` subcommands can be overwritten. Any
``--foo FOO`` argument can be set via the correspond ``[cli:<cmd>]`` settings

For the ``run`` command you can additionally set ``component`` to set the
default component to run.

.. code-block:: ini


Programmatic Usage

Unlike the cli, ``.torchxconfig`` file **is not** picked up automatically
from ``CWD`` if you are programmatically running your component with :py:class:`torchx.runner.Runner`.
You'll have to manually specify the directory containing ``.torchxconfig``.

Below is an example

.. doctest:: [runner_config_example]

 from torchx.runner import get_runner
 from torchx.runner.config import apply
 import torchx.specs as specs

 def my_component(a: int) -> specs.AppDef:
    # <... component body omitted for brevity ...>

 scheduler = "local_cwd"
 cfg = {"log_dir": "/these/take/outmost/precedence"}

 apply(scheduler, cfg, dirs=["/home/bob"])  # looks for /home/bob/.torchxconfig
 get_runner().run(my_component(1), scheduler, cfg)

You may also specify multiple directories (in preceding order) which is useful when
you want to keep personal config overrides on top of a project defined default.

import configparser as configparser
import logging
import os
from pathlib import Path
from typing import Dict, Iterable, List, Optional, TextIO

from torchx.schedulers import get_scheduler_factories, Scheduler
from torchx.specs import CfgVal, get_type_name
from torchx.specs.api import runopt

CONFIG_FILE = ".torchxconfig"
DEFAULT_CONFIG_DIRS = [str(Path.home()), str(Path.cwd())]

_NONE = "None"

log: logging.Logger = logging.getLogger(__name__)

def _configparser() -> configparser.ConfigParser:
    Sets up the configparser and returns it. The same config parser
    should be used between dumps() and loads() methods for ser/de compatibility

    config = configparser.ConfigParser()
    # if optionxform is not overridden, configparser will by default lowercase
    # the option keys because it is compatible with Windows INI files
    # which are expected to be parsed case insensitive.
    # override since torchx's runopts are case-sensitive
    # see:
    # pyre-ignore[8]
    config.optionxform = lambda option: option

    return config

def _get_scheduler(name: str) -> Scheduler:
    schedulers = get_scheduler_factories()
    if name not in schedulers:
        raise ValueError(
            f"`{name}` is not a registered scheduler. Valid scheduler names: {schedulers.keys()}"
    sched = schedulers[name](session_name="_")
    return sched

def _fixme_placeholder(runopt: runopt, max_len: int = 60) -> str:
    ph = f"#FIXME:({get_type_name(runopt.opt_type)}) {}"
    return ph if len(ph) <= max_len else f"{ph[:max_len]}..."

[docs]def dump( f: TextIO, schedulers: Optional[List[str]] = None, required_only: bool = False ) -> None: """ Dumps a default INI-style config template containing the :py:class:torchx.specs.runopts for the given scheduler names into the file-like object specified by ``f``. If no ``schedulers`` are specified dumps all known registered schedulers. Optional runopts are pre-filled with their default values. Required runopts are set with a ``FIXME: ...`` placeholder. To only dump required runopts pass ``required_only=True``. Each scheduler's runopts are written in the section called ``[{scheduler_name}]``. For example: :: [kubernetes] namespace = default queue = #FIXME (str)Volcano queue to schedule job in Raises: ValueError: if given a scheduler name that is not known """ if schedulers: scheds = schedulers else: scheds = get_scheduler_factories().keys() config = _configparser() for sched_name in scheds: try: sched = _get_scheduler(sched_name) except ModuleNotFoundError: continue section = f"{sched_name}" config.add_section(section) for opt_name, opt in sched.run_opts(): if opt.is_required: val = _fixme_placeholder(opt) else: # not required runopts MUST have a default if required_only: continue # serialize list elements with `;` delimiter (consistent with torchx cli) if opt.opt_type == List[str]: # deal with empty or None default lists if opt.default: # pyre-ignore[6] opt.default type checked already as List[str] val = ";".join(opt.default) else: val = _NONE else: val = f"{opt.default}" config.set(section, opt_name, val) config.write(f, space_around_delimiters=True)
[docs]def apply( scheduler: str, cfg: Dict[str, CfgVal], dirs: Optional[List[str]] = None, ) -> None: """ Loads a ``.torchxconfig`` INI file from the specified directories in preceding order and applies the run configs for the scheduler onto the given ``cfg``. If no ``dirs`` is specified, then it looks for ``.torchxconfig`` in the current working directory. If a specified directory does not have ``.torchxconfig`` then it is ignored. Note that the configs already present in the given ``cfg`` take precedence over the ones in the config file and only new configs are added. The same holds true for the configs loaded in list order. For instance if ``cfg={"foo":"bar"}`` and the config file is: :: # dir_1/.torchxconfig [local_cwd] foo = baz hello = world # dir_2/.torchxconfig [local_cwd] hello = bob Then after the method call, ``cfg={"foo":"bar","hello":"world"}``. """ for configfile in find_configs(dirs): with open(configfile, "r") as f: load(scheduler, f, cfg)"loaded configs from {configfile}")
[docs]def load_sections( prefix: str, dirs: Optional[List[str]] = None, ) -> Dict[str, Dict[str, str]]: """ Loads the content of the sections in the given ``.torchxconfig`` file that start with the specified prefix. Returns a map of maps of the section name WITHOUT the prefix with the contents of the section loaded into a map. ``":"`` is used as the prefix delimiter. Example config format for specifying defaults for the builtin component ``dist.ddp`` is shown below: :: [component:dist.ddp] j = 1x2 image = # calling `load_sections(prefix="component")` returns # { # "dist.ddp": { # "j":"1x2", # "image":"", # }, # } The keys in the section must match the parameter name of the component function. The example below shows how to represent the various types that are allowable as component parameter types. :: [] int = 1 float = 1.2 bool = True # or False str = foobar list = a,b,c map = A=B,C=D vararg = -a b --c=d e # to call the component as: "-a", "b", "--c=d", "e", int=1, float=1.2, bool=True, str="foobar", list=["a", "b", "c"], map={"A":"B", "C": "D"}) """ def strip_prefix(section_name: str) -> Optional[str]: # returns the section_name with the prefix removed # or None if the section name does not match the prefix idx = section_name.find(CONFIG_PREFIX_DELIM) # index guard (e.g ":foo" and "foo:" are bad, just return None) if 0 < idx < len(section_name) - 1: if section_name[0:idx] == prefix: return section_name[idx + 1 :] return None sections: Dict[str, Dict[str, str]] = {} for configfile in find_configs(dirs): with open(configfile, "r") as f: config = _configparser() config.read_file(f) for section_name in config.sections(): name = strip_prefix(section_name) if name: section = sections.setdefault(name, {}) for key, value in config.items(section_name): if key not in section: log.debug( f"Loaded config: {prefix}.{key}={value} from {configfile}" ) section[key] = value return sections
[docs]def get_configs( prefix: str, name: str, dirs: Optional[List[str]] = None, ) -> Dict[str, str]: """ Gets all the config values in the section ``["{prefix}:{name}"]``. Or an empty map if the section does not exist. Example: :: # for config file: # [foo:bar] # baz = 1 get_configs(prefix="foo", name="bar") # returns {"baz": "1"} get_config(prefix="foo", name="barr") # returns {} """ sections = load_sections(prefix, dirs) return sections.get(name, {})
[docs]def get_config( prefix: str, name: str, key: str, dirs: Optional[List[str]] = None, ) -> Optional[str]: """ Gets the config value for the ``key`` in the section ``["{prefix}:{name}"]``. Or ``None`` if no section or key exists Example: :: # for config file: # [foo:bar] # baz = 1 get_config(prefix="foo", name="bar", key="baz") == 1 get_config(prefix="foo", name="bar", key="bazz") == None get_config(prefix="foo", name="barr", key="baz") == None get_config(prefix="fooo", name="bar", key="baz") == None """ return get_configs(prefix, name, dirs).get(key, None)
[docs]def find_configs(dirs: Optional[Iterable[str]] = None) -> List[str]: """ Finds and returns the filepath to ``.torchxconfig`` files based on the following logic: 1. If the environment variable ``TORCHXCONFIG`` exists, then its value is returned in a single-element list and the directories specified through the ``dirs`` parameter is NOT searched. 2. Otherwise, a ``.torchxconfig`` file is looked for in ``dirs`` and the filepaths to existing config files are returned. If ``dirs`` is not specified or is empty then ``dirs`` defaults to ``[$HOME, $CWD]`` where CWD is current working dir. """ config = os.getenv(ENV_TORCHXCONFIG) if config is not None: configfile = Path(config) if not configfile.is_file(): raise FileNotFoundError( f"`{ENV_TORCHXCONFIG}={config}` does not exist or is not a file." ) return [str(configfile)] else: config_files = [] if not dirs: dirs = DEFAULT_CONFIG_DIRS for d in dirs: configfile = Path(d) / CONFIG_FILE if configfile.exists(): config_files.append(str(configfile)) return config_files
[docs]def load(scheduler: str, f: TextIO, cfg: Dict[str, CfgVal]) -> None: """ loads the section ``[{scheduler}]`` from the given configfile ``f`` (in .INI format) into the provided ``runcfg``, only adding configs that are NOT currently in the given ``runcfg`` (e.g. does not override existing values in ``runcfg``). If no section is found, does nothing. """ config = _configparser() config.read_file(f) runopts = _get_scheduler(scheduler).run_opts() section = f"{scheduler}" if config.has_section(section): for name, value in config.items(section): if name in cfg.keys(): # DO NOT OVERRIDE existing configs continue if value == _NONE: # should map to None (not str 'None') # this also handles empty or None lists cfg[name] = None else: runopt = runopts.get(name) if runopt is None: log.warning( f"`{name} = {value}` was declared in the [{section}] section " f" of the config file but is not a runopt of `{scheduler}` scheduler." f" Remove the entry from the config file to no longer see this warning" ) else: if runopt.opt_type is bool: # need to handle bool specially since str -> bool is based on # str emptiness not value (e.g. bool("False") == True) cfg[name] = config.getboolean(section, name) elif runopt.opt_type is List[str]: cfg[name] = value.split(";") else: # pyre-ignore[29] cfg[name] = runopt.opt_type(value)


Access comprehensive developer documentation for PyTorch

View Docs


Get in-depth tutorials for beginners and advanced developers

View Tutorials


Find development resources and get your questions answered

View Resources