Coverage for python / lsst / ctrl / bps / parsl / sites / work_queue.py: 48%
31 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-18 01:09 -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 WorkQueueExecutor
31from parsl.executors.base import ParslExecutor
32from parsl.launchers import SrunLauncher
33from parsl.providers import LocalProvider
35try:
36 from parsl.providers.base import ExecutionProvider
37except ImportError:
38 from parsl.providers.provider_base import ExecutionProvider # type: ignore
40from ..configuration import get_bps_config_value
41from ..site import SiteConfig
43if TYPE_CHECKING:
44 from ..job import ParslJob
46__all__ = ("LocalSrunWorkQueue", "WorkQueue")
49class WorkQueue(SiteConfig):
50 """Base class configuraton for `WorkQueueExecutor`.
52 Subclasses must provide implementations for ``.get_executors``
53 and ``.select_executor``. In ``.get_executors``, the site-specific
54 `ExecutionProvider` must be defined.
56 Parameters
57 ----------
58 *args : `~typing.Any`
59 Parameters forwarded to base class constructor.
60 **kwargs : `~typing.Any`
61 Keyword arguments passed to base class constructor, augmented by
62 the ``resource_list`` argument.
64 Notes
65 -----
66 The following BPS configuration parameters are recognized, overriding the
67 defaults:
69 - ``port`` (`int`): The port used by work_queue. Default: ``9000``.
70 - ``worker_options (`str`): Extra options to pass to work_queue workers.
71 A typical option specifies the memory available per worker, e.g.,
72 ``"--memory=90000"``, which sets the available memory to 90 GB.
73 Default: ``""``
74 - ``wq_max_retries`` (`int`): The number of retries that work_queue
75 will make in case of task failures. Set to ``None`` to have work_queue
76 retry forever; set to ``1`` to have retries managed by Parsl.
77 Default: ``1``
78 """
80 def __init__(self, *args, **kwargs):
81 # Have BPS-defined resource requests for each job passed to work_queue.
82 kwargs["resource_list"] = ["memory", "cores", "disk", "running_time_min", "priority"]
83 super().__init__(*args, **kwargs)
85 def make_executor(
86 self,
87 label: str,
88 provider: ExecutionProvider,
89 *,
90 port: int = 9000,
91 worker_options: str = "",
92 wq_max_retries: int = 1,
93 ) -> ParslExecutor:
94 """Return a `WorkQueueExecutor`. The ``provider`` contains the
95 site-specific configuration.
97 Parameters
98 ----------
99 label : `str`
100 Label for executor.
101 provider : `ExecutionProvider`
102 Parsl execution provider, e.g., `SlurmProvider`.
103 port : `int`, optional
104 Port used by work_queue. Default: ``9000``.
105 worker_options : `str`, optional
106 Extra options to pass to work_queue workers, e.g.,
107 ``"--memory=90000"``. Default: `""`.
108 wq_max_retries : `int`, optional
109 Number of retries for work_queue to attempt per job. Set to
110 ``None`` to have it try indefinitely; set to ``1`` to have Parsl
111 control the number of retries. Default: ``1``.
112 """
113 port = get_bps_config_value(self.site, "port", int, port)
114 worker_options = get_bps_config_value(self.site, "worker_options", str, worker_options)
115 max_retries = get_bps_config_value(self.site, "wq_max_retries", int, wq_max_retries)
116 return WorkQueueExecutor(
117 label=label,
118 provider=provider,
119 port=port,
120 worker_options=worker_options,
121 max_retries=max_retries,
122 shared_fs=True,
123 autolabel=False,
124 )
127class LocalSrunWorkQueue(WorkQueue):
128 """Configuration for a `WorkQueueExecutor` that uses a `LocalProvider`
129 to manage resources.
131 This can be used directly as the site configuration within a
132 multi-node allocation when Slurm is available. For running on a
133 single node, e.g., a laptop, a `SingleNodeLauncher` is used, and
134 Slurm need not be available.
136 The following BPS configuration parameters are recognized, overriding the
137 defaults:
139 - ``port`` (`int`): The port used by work_queue. Default: ``9000``.
140 - ``worker_options (`str`): Extra options to pass to work_queue workers.
141 A typical option specifies the memory available per worker, e.g.,
142 ``"--memory=90000"``, which sets the available memory to 90 GB.
143 Default: ``""``
144 - ``wq_max_retries`` (`int`): The number of retries that work_queue
145 will make in case of task failures. Set to ``None`` to have work_queue
146 retry forever; set to ``1`` to have retries managed by Parsl. Default:
147 ``1``.
148 - ``nodes_per_block`` (`int`): The number of allocated nodes.
149 Default: ``1``.
150 """
152 def get_executors(self) -> list[ParslExecutor]:
153 """Get a list of executors to be used in processing."""
154 nodes_per_block = get_bps_config_value(self.site, "nodes_per_block", int, 1)
155 provider_options: dict[str, Any] = {
156 "nodes_per_block": nodes_per_block,
157 "init_blocks": 0,
158 "min_blocks": 0,
159 "max_blocks": 1,
160 "parallelism": 0,
161 "cmd_timeout": 300,
162 }
163 if nodes_per_block > 1:
164 provider_options["launcher"] = SrunLauncher(overrides="-K0 -k --slurmd-debug=verbose")
165 provider = LocalProvider(**provider_options)
166 return [self.make_executor("work_queue", provider)]
168 def select_executor(self, job: "ParslJob") -> str:
169 """Get the ``label`` of the executor to use to execute a job.
171 Parameters
172 ----------
173 job : `ParslJob`
174 Job to be executed.
176 Returns
177 -------
178 label : `str`
179 Label of executor to use to execute ``job``.
180 """
181 return "work_queue"