Source code for reframe.core.launchers

# Copyright 2016-2024 Swiss National Supercomputing Centre (CSCS/ETH Zurich)
# ReFrame Project Developers. See the top-level LICENSE file for details.
#
# SPDX-License-Identifier: BSD-3-Clause

import abc
import reframe.utility.typecheck as typ
from reframe.core.meta import RegressionTestMeta
from reframe.core.warnings import user_deprecation_warning


class _JobLauncherMeta(RegressionTestMeta, abc.ABCMeta):
    '''Job launcher metaclass.'''


[docs] class JobLauncher(metaclass=_JobLauncherMeta): '''Abstract base class for job launchers. A job launcher is the executable that actually launches a distributed program to multiple nodes, e.g., ``mpirun``, ``srun`` etc. .. note:: .. versionchanged:: 4.0.0 Users may create job launchers directly. .. versionchanged:: 2.8 Job launchers do not get a reference to a job during their initialization. ''' #: List of options to be passed to the job launcher invocation. #: #: :type: :class:`List[str]` #: :default: ``[]`` options = variable(typ.List[str], value=[]) #: Optional modifier of the launcher command. #: #: This will be combined with the :attr:`modifier_options` and prepended to #: the parallel launch command. #: #: :type: :class:`str` #: :default: ``''`` #: #: .. versionadded:: 4.6.0 modifier = variable(str, value='') #: Options to be passed to the launcher :attr:`modifier`. #: #: If the modifier is empty, these options will be ignored. #: #: :type: :clas:`List[str]` #: :default: ``[]`` #: #: :versionadded:: 4.6.0 modifier_options = variable(typ.List[str], value=[]) def __init__(self): self.options = []
[docs] @abc.abstractmethod def command(self, job): '''The launcher command to be emitted for a specific job. Launcher backends provide concrete implementations of this method. :param job: A job descriptor. :returns: the basic launcher command as a list of tokens. '''
[docs] def run_command(self, job): '''The full launcher command to be emitted for a specific job. This includes any user options. :param job: a job descriptor. :returns: the launcher command as a string. ''' cmd_tokens = [] if self.modifier: cmd_tokens.append(self.modifier) cmd_tokens += self.modifier_options cmd_tokens += self.command(job) + self.options return ' '.join(cmd_tokens)
[docs] class LauncherWrapper(JobLauncher): '''Wrap a launcher object so as to modify its invocation. This is useful for parallel debuggers. For example, to launch a regression test using the `ARM DDT <https://www.arm.com/products/development-tools/server-and-hpc/forge>`__ debugger, you can do the following: .. code:: python @run_after('setup') def set_launcher(self): self.job.launcher = LauncherWrapper(self.job.launcher, 'ddt', ['--offline']) If the current system partition uses native Slurm for job submission, this setup will generate the following command in the submission script: :: ddt --offline srun <test_executable> If the current partition uses ``mpirun`` instead, it will generate :: ddt --offline mpirun -np <num_tasks> ... <test_executable> :arg target_launcher: The launcher to wrap. :arg wrapper_command: The wrapper command. :arg wrapper_options: List of options to pass to the wrapper command. ''' def __init__(self, target_launcher, wrapper_command, wrapper_options=None): super().__init__() user_deprecation_warning("'LauncherWrapper is deprecated; " "please use the launcher's 'modifier' and " "'modifier_options' instead") wrapper_options = wrapper_options or [] self.options = target_launcher.options self._target_launcher = target_launcher self._wrapper_command = [wrapper_command] + wrapper_options
[docs] def command(self, job): return self._wrapper_command + self._target_launcher.command(job)