Coverage for python / lsst / ctrl / bps / parsl / sites / torque.py: 34%
50 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 01:46 -0800
1# This file is part of ctrl_bps_parsl.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org) and the LSST DESC (https://www.lsstdesc.org/).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
28from typing import TYPE_CHECKING, Any
30from parsl.executors import HighThroughputExecutor
31from parsl.executors.base import ParslExecutor
32from parsl.launchers import MpiRunLauncher
33from parsl.providers import TorqueProvider
35from ..configuration import get_bps_config_value, get_workflow_name
36from ..site import SiteConfig
38if TYPE_CHECKING:
39 from ..job import ParslJob
41__all__ = ("Torque",)
44Kwargs = dict[str, Any]
47class Torque(SiteConfig):
48 """Configuration for generic Torque cluster.
50 This can be used directly as the site configuration for a Torque cluster by
51 setting the BPS config, e.g.:
53 .. code-block:: yaml
55 computeSite: torque
56 site:
57 torque:
58 class: lsst.ctrl.bps.parsl.sites.Torque
59 nodes: 4
60 tasks_per_node: 20
61 walltime: "00:59:00" # Note: always quote walltime in YAML
63 Alternatively, it can be used as a base class for Torque cluster
64 configurations.
66 The following BPS configuration parameters are recognised (and required
67 unless there is a default mentioned here, or provided by a subclass):
69 - ``queue`` (`int`): Queue for the Torque job.
70 - ``nodes`` (`int`): number of nodes for each Torque job.
71 - ``tasks_per_node`` (`int`): number of cores per node for each Torque job;
72 by default we use all cores on the node.
73 - ``walltime`` (`str`): time limit for each Torque job.
74 - ``scheduler_options`` (`str`): text to prepend to the Torque submission
75 script (each line usually starting with ``#PBS``).
76 """
78 def __init__(self, *args, **kwargs):
79 # Have BPS-defined resource requests for each job passed to executor.
80 kwargs["resource_list"] = ["priority"]
81 super().__init__(*args, **kwargs)
83 def make_executor(
84 self,
85 label: str,
86 *,
87 queue: str | None = None,
88 nodes: int | None = None,
89 tasks_per_node: int | None = None,
90 walltime: str | None = None,
91 mem_per_worker: float | None = None,
92 scheduler_options: str | None = None,
93 worker_init: str | None = None,
94 provider_options: Kwargs | None = None,
95 executor_options: Kwargs | None = None,
96 ) -> ParslExecutor:
97 """Return an executor for running on a Torque cluster.
99 Parameters
100 ----------
101 label : `str`
102 Label for executor.
103 queue : `str`, optional
104 Queue for the Torque job.
105 nodes : `int`, optional
106 Default number of nodes for each Torque job.
107 tasks_per_node : `int`, optional
108 Default number of cores per node for each Torque job.
109 walltime : `str`, optional
110 Default time limit for each Torque job.
111 mem_per_worker : `float`, optional
112 Minimum memory per worker (GB), limited by the executor.
113 worker_init : `str`, optional
114 Environment initiation command
115 scheduler_options : `str`, optional
116 ``#SBATCH`` directives to prepend to the Torque submission script.
117 provider_options : `dict`, optional
118 Additional arguments for `TorqueProvider` constructor.
119 executor_options : `dict`, optional
120 Additional arguments for `HighThroughputExecutor` constructor.
122 Returns
123 -------
124 executor : `HighThroughputExecutor`
125 Executor for Torque jobs.
126 """
127 nodes = get_bps_config_value(self.site, "nodes", int, nodes, required=True)
128 walltime = get_bps_config_value(self.site, "walltime", str, walltime, required=True)
129 queue = get_bps_config_value(self.site, "queue", str, queue)
130 tasks_per_node = get_bps_config_value(self.site, "tasks_per_node", int, tasks_per_node)
131 worker_init = get_bps_config_value(self.site, "worker_init", str, "")
132 scheduler_options = get_bps_config_value(self.site, "scheduler_options", str, scheduler_options)
134 if tasks_per_node is None:
135 tasks_per_node = 1
137 job_name = get_workflow_name(self.config)
139 if scheduler_options is None:
140 scheduler_options = ""
141 else:
142 scheduler_options += "\n"
143 scheduler_options += f"#PBS -N {job_name}\n"
144 if queue:
145 scheduler_options += f"#PBS -q {queue}\n"
147 if worker_init is None:
148 worker_init = ""
150 launcher = PbsMpiRunLauncher(overrides=f"--map-by core:{tasks_per_node}")
152 return HighThroughputExecutor(
153 label,
154 provider=PbsTorqueProvider(
155 nodes_per_block=nodes,
156 tasks_per_node=tasks_per_node,
157 queue=queue,
158 walltime=walltime,
159 scheduler_options=scheduler_options,
160 worker_init=worker_init,
161 launcher=launcher,
162 **(provider_options or {}),
163 ),
164 max_workers_per_node=1,
165 mem_per_worker=mem_per_worker,
166 address=self.get_address(),
167 **(executor_options or {}),
168 )
170 def get_executors(self) -> list[ParslExecutor]:
171 """Get a list of executors to be used in processing.
173 Each executor should have a unique ``label``.
174 """
175 return [self.make_executor("torque")]
177 def select_executor(self, job: "ParslJob") -> str:
178 """Get the ``label`` of the executor to use to execute a job.
180 Parameters
181 ----------
182 job : `ParslJob`
183 Job to be executed.
185 Returns
186 -------
187 label : `str`
188 Label of executor to use to execute ``job``.
189 """
190 return "torque"
193class PbsTorqueProvider(TorqueProvider):
194 """Torque Execution Provider
196 This provider uses qsub to submit, qstat for status, and qdel to cancel
197 jobs. The qsub script to be used is created from a template file in this
198 same module.
200 This subclass allows the ``tasks_per_node`` to be set at construction time
201 instead of at submission time.
202 """
204 def __init__(self, *args, tasks_per_node: int = 1, **kwargs):
205 super().__init__(*args, **kwargs)
206 self.tasks_per_node = tasks_per_node
208 def submit(self, command, tasks_per_node, job_name="parsl.torque"):
209 """Submit the command onto an Local Resource Manager job.
211 This function returns an ID that corresponds to the task that was just
212 submitted.
214 The ``tasks_per_node`` parameter is ignored in this provider, as it is
215 set at construction time.
217 Parameters
218 ----------
219 command : `str`
220 Command-line invocation to be made on the remote side.
221 tasks_per_node : `int`
222 Number of tasks to be launched per node. This is ignored in this
223 provider.
224 job_name : `str`:
225 Name for job, must be unique.
227 Returns
228 -------
229 None: At capacity, cannot provision more
230 job_id (string): Identifier for the job
232 """
233 return super().submit(
234 command=command,
235 tasks_per_node=self.tasks_per_node,
236 job_name=job_name,
237 )
240class PbsMpiRunLauncher(MpiRunLauncher):
241 """Worker launcher that wraps the user's command with the framework to
242 launch multiple command invocations via ``mpirun``.
244 This wrapper sets the bash env variable ``CORES`` to the number of cores on
245 the machine.
247 This launcher makes the following assumptions:
248 - mpirun is installed and can be located in ``$PATH``
249 - The provider makes available the ``$PBS_NODEFILE`` environment variable
250 """
252 def __init__(
253 self,
254 debug: bool = True,
255 bash_location: str = "/bin/bash",
256 overrides: str = "",
257 ):
258 super().__init__(debug=debug, bash_location=bash_location, overrides=overrides)
260 def __call__(self, command: str, tasks_per_node: int, nodes_per_block: int) -> str:
261 """Wrap the user's command with mpirun invocation"""
262 worker_count = nodes_per_block * tasks_per_node
263 debug_num = int(self.debug)
265 return f"""set -e
266export CORES=$(getconf _NPROCESSORS_ONLN)
267[[ "{debug_num}" == "1" ]] && echo "Found cores : $CORES"
268WORKERCOUNT={worker_count}
270cat << MPIRUN_EOF > cmd_$JOBNAME.sh
271{command}
272MPIRUN_EOF
273chmod u+x cmd_$JOBNAME.sh
275mpirun -np $WORKERCOUNT {self.overrides} {self.bash_location} cmd_$JOBNAME.sh
277[[ "{debug_num}" == "1" ]] && echo "All workers done"
278"""