Coverage for python / lsst / ctrl / bps / parsl / sites / ccin2p3.py: 29%
58 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-31 01:13 -0800
1import copy
2import platform
3from typing import TYPE_CHECKING, Any
5import parsl.config
6from parsl.executors import HighThroughputExecutor
7from parsl.executors.base import ParslExecutor
8from parsl.providers import SlurmProvider
10from ..configuration import get_bps_config_value
11from ..site import SiteConfig
13if TYPE_CHECKING:
14 from .job import ParslJob
16__all__ = ("Ccin2p3",)
18Kwargs = dict[str, Any]
21class Ccin2p3(SiteConfig):
22 """Configuration for executing Parsl jobs in CC-IN2P3 Slurm batch farm.
24 This class provides four job slot sizes each with its specific
25 requirements, in particular in terms of memory. Those slot sizes are named
26 "small", "medium", "large" and "xlarge".
28 Sensible default values for those requirements are provided for each
29 job slot but you can overwrite those defaults either in the
30 the BPS submission file or in a site configuration file that you
31 include in your BPS submission file.
33 If you don't need to modify the default requirements for the job slot
34 sizes, use the site specification below in your BPS configuration
35 file:
37 .. code-block:: yaml
39 wmsServiceClass: lsst.ctrl.bps.parsl.ParslService
40 computeSite: ccin2p3
42 site:
43 ccin2p3:
44 class: lsst.ctrl.bps.parsl.sites.ccin2p3.Ccin2p3
46 If you do need to modify those defaults, you can overwrite them for
47 all job slots or for specific each job slots. Requirements specified
48 for a job slot take priority over those specified for all job slots
49 at the level of entry '.site.ccin2p3:'.
51 This is an example of how to overwrite selected requirements in your BPS
52 submission file:
54 .. code-block:: yaml
56 wmsServiceClass: lsst.ctrl.bps.parsl.ParslService
57 computeSite: ccin2p3
59 site:
60 ccin2p3:
61 class: lsst.ctrl.bps.parsl.sites.ccin2p3.Ccin2p3
62 walltime: "72:00:00"
63 scheduler_options:
64 - "--licenses=sps"
65 - "--qos=normal"
66 small:
67 memory: 6
68 partition: "flash"
69 medium:
70 memory: 10
71 partition: "lsst,htc"
72 large:
73 memory: 80
74 xlarge:
75 memory: 180
76 partition: "lsst"
77 scheduler_options:
78 - "--constraint=el7"
79 - "--licenses=my_product"
80 - "--reservation=my_reservation"
82 At the level of entry 'site.ccin2p3:' in the BPS submission file, the
83 following configuration parameters are accepted, which apply to all slot
84 sizes:
86 - `partition` (`str`): name of the one or more configured partitions. If
87 more than one, separate them with comma (',').
88 (Default: "lsst,htc")
89 - `walltime` (`str`): walltime to require for the job (Default: "72:00:00")
90 - `scheduler_options` (`list` [`str`] ): scheduler options to send to Slurm
91 for scheduling purposes.
92 (Default: "--licenses=sps")
94 In addition, as shown in the previous example, for each job slot (i.e.
95 "small", "medium", etc.) you can specify the requirements above as well as
96 the following:
98 - `max_blocks` (`int`): maximum number of Slurm jobs that your workflow can
99 simultaneously use.
100 - `memory` (`int`): required amount of memory for each job, in Gigabytes.
101 (Defaults: 4 for "small", 10 for "medium", 50 fo "large" and
102 150 for "xlarge").
104 Parameters
105 ----------
106 *args : optional
107 Arguments to initialize the super-class.
108 **kwargs : optional
109 Keyword arguments to initialize the super-class.
111 Returns
112 -------
113 Ccin2p3 : `SiteConfig`
114 Concrete instance of a `SiteConfig` specific for the CC-IN2P3 Slurm
115 farm.
116 """
118 DEFAULT_ACCOUNT: str = "lsst"
119 DEFAULT_WALLTIME: str = "72:00:00"
120 DEFAULT_SCHEDULER_OPTIONS: list[str] = [
121 "--licenses=sps",
122 ]
124 def __init__(self, *args, **kwargs):
125 # Have BPS-defined resource requests for each job passed to executor.
126 kwargs["resource_list"] = ["priority"]
127 super().__init__(*args, **kwargs)
128 self._account = get_bps_config_value(self.site, ".account", str, self.DEFAULT_ACCOUNT)
129 self._scheduler_options = get_bps_config_value(
130 self.site, ".scheduler_options", list, self.DEFAULT_SCHEDULER_OPTIONS
131 )
132 self._slot_size = {
133 "small": {
134 "memory": get_bps_config_value(self.site, ".small.memory", int, 4),
135 "walltime": self._get_walltime_for_slot("small"),
136 "partition": self._get_partition_for_slot("small"),
137 "max_blocks": get_bps_config_value(self.site, ".small.max_blocks", int, 3_000),
138 "scheduler_options": get_bps_config_value(self.site, ".small.scheduler_options", list, []),
139 },
140 "medium": {
141 "memory": get_bps_config_value(self.site, ".medium.memory", int, 10),
142 "walltime": self._get_walltime_for_slot("medium"),
143 "partition": self._get_partition_for_slot("medium"),
144 "max_blocks": get_bps_config_value(self.site, ".medium.max_blocks", int, 1_000),
145 "scheduler_options": get_bps_config_value(self.site, ".medium.scheduler_options", list, []),
146 },
147 "large": {
148 "memory": get_bps_config_value(self.site, ".large.memory", int, 50),
149 "walltime": self._get_walltime_for_slot("large"),
150 "partition": self._get_partition_for_slot("large"),
151 "max_blocks": get_bps_config_value(self.site, ".large.max_blocks", int, 100),
152 "scheduler_options": get_bps_config_value(self.site, ".large.scheduler_options", list, []),
153 },
154 "xlarge": {
155 "memory": get_bps_config_value(self.site, ".xlarge.memory", int, 150),
156 "walltime": self._get_walltime_for_slot("xlarge"),
157 "partition": self._get_partition_for_slot("xlarge"),
158 "max_blocks": get_bps_config_value(self.site, ".xlarge.max_blocks", int, 10),
159 "scheduler_options": get_bps_config_value(self.site, ".xlarge.scheduler_options", list, []),
160 },
161 }
163 def _get_partition_for_slot(self, slot: str) -> str:
164 """Return the Slurm partition Parsl must use to submit jobs for the
165 job slot `slot`. Values of `slot` can be "small", "medium", "large"
166 or "xlarge".
167 """
168 # The target Slurm partition must be selected according to the type of
169 # the job slot but also according to the CPU architecture of the
170 # compute node.
171 #
172 # Parsl requires that the CPU architecture of its orchestrator to
173 # be identical to the architecture of its executors. Therefore,
174 # we need to ensure that Slurm schedules our Parsl executors on
175 # compute nodes with the same architecture as the host where this
176 # orchestrator runs.
178 # Default target Slurm partitions per CPU architecture
179 default_partition = {
180 "aarch64": {
181 "small": "htc_arm",
182 "medium": "htc_arm",
183 "large": "htc_arm",
184 "xlarge": "htc_arm",
185 },
186 "x86_64": {
187 "small": "lsst,htc",
188 "medium": "lsst",
189 "large": "lsst",
190 "xlarge": "lsst",
191 },
192 }
193 architecture = platform.machine()
194 if architecture not in default_partition:
195 raise ValueError(f"architecture {architecture} is not supported")
197 # If a partition was specified in the workflow description file
198 # specifically for this job slot, use that partition. For instance:
199 #
200 # site:
201 # ccin2p3:
202 # class: lsst.ctrl.bps.parsl.sites.ccin2p3.Ccin2p3
203 # small:
204 # partition: htc
205 slot_partition = get_bps_config_value(self.site, f".{slot}.partition", str, "")
206 if slot_partition != "":
207 return slot_partition
209 # If a partition was specified in the workflow description file at
210 # the site level, use that partition. For instance:
211 #
212 # site:
213 # ccin2p3:
214 # class: lsst.ctrl.bps.parsl.sites.ccin2p3.Ccin2p3
215 # partition: htc
216 #
217 # Otherwise, use the default for this slot on this architecture.
218 return get_bps_config_value(self.site, ".partition", str, default_partition[architecture][slot])
220 def _get_walltime_for_slot(self, slot: str) -> str:
221 """Return the value for walltime Parsl must use to submit jobs for the
222 job slot `slot`. Values of `slot` can be "small", "medium", "large"
223 or "xlarge".
224 """
225 # If a specific walltime value was specified for this job slot in the
226 # configuration use that value. For instance:
227 #
228 # site:
229 # ccin2p3:
230 # class: lsst.ctrl.bps.parsl.sites.ccin2p3.Ccin2p3
231 # small:
232 # walltime: "3:00:00"
233 slot_walltime = get_bps_config_value(self.site, f".{slot}.walltime", str, "")
234 if slot_walltime != "":
235 return slot_walltime
237 # If a walltime value was specified for the site use that value.
238 # Otherwise, use the default walltime. For instance:
239 #
240 # site:
241 # ccin2p3:
242 # class: lsst.ctrl.bps.parsl.sites.ccin2p3.Ccin2p3
243 # walltime: "3:00:00"
244 return get_bps_config_value(self.site, ".walltime", str, self.DEFAULT_WALLTIME)
246 def get_executors(self) -> list[ParslExecutor]:
247 """Get a list of Parsl executors that can be used for processing a
248 workflow.
250 Each executor must have a unique ``label``.
251 """
252 executors: list[ParslExecutor] = []
253 for label, slot in self._slot_size.items():
254 # Compute the scheduler options for this job slot. Options
255 # specified at the slot level in the configuration file
256 # overwrite those specified at the site level.
257 scheduler_options = copy.deepcopy(self._scheduler_options)
258 if slot_scheduler_options := slot.get("scheduler_options", []):
259 scheduler_options = copy.deepcopy(slot_scheduler_options)
261 options = f"#SBATCH {' '.join(opt for opt in scheduler_options)}" if scheduler_options else ""
263 executor = HighThroughputExecutor(
264 label,
265 provider=SlurmProvider(
266 # Slurm partition to request blocks from.
267 partition=slot["partition"],
268 # Slurm account to which to charge resources used by the
269 # job.
270 account=self._account,
271 # Nodes to provision per block (1 block = 1 CPU core).
272 nodes_per_block=1,
273 # Number of CPU cores to provision per node.
274 cores_per_node=1,
275 # Memory per node (GB) for each Slurm job.
276 mem_per_node=slot["memory"],
277 # Initial number of blocks.
278 init_blocks=0,
279 # Minimum number of blocks to maintain.
280 min_blocks=0,
281 # Maximum number of blocks to maintain.
282 max_blocks=slot["max_blocks"],
283 # Time limit for each Slurm job.
284 walltime=slot["walltime"],
285 # '#SBATCH' directives to prepend to the Slurm submission
286 # script.
287 scheduler_options=options,
288 # Set the number of file descriptors and processes to
289 # the maximum allowed.
290 worker_init="ulimit -n hard && ulimit -u hard",
291 # Requests nodes which are not shared with other running
292 # jobs.
293 exclusive=False,
294 ),
295 # Address to connect to the main Parsl process.
296 address=self.get_address(),
297 # GB of memory required per worker. If specified the node
298 # manager will check the available memory at startup and limit
299 # the number of workers such that the there’s sufficient memory
300 # for each worker.
301 mem_per_worker=None,
302 # Caps the number of workers launched per node.
303 max_workers_per_node=1,
304 # Timeout period (in milliseconds) to be used by the
305 # executor components.
306 poll_period=1_000,
307 # Retry submitting to Slurm in case of submission error.
308 block_error_handler=False,
309 )
310 executors.append(executor)
312 return executors
314 def select_executor(self, job: "ParslJob") -> str:
315 """Get the ``label`` of the executor to use to execute ``job``.
317 Parameters
318 ----------
319 job : `ParslJob`
320 Job to be executed.
322 Returns
323 -------
324 label : `str`
325 Label of executor to use to execute ``job``.
326 """
327 # We choose the executor to use based only on the memory required
328 # by the job.
329 memory = job.generic.request_memory / 1024 # Convert to GB
330 for label in ("small", "medium", "large"):
331 if memory <= self._slot_size[label]["memory"]:
332 return label
334 return "xlarge"
336 def get_parsl_config(self) -> parsl.config.Config:
337 """Get Parsl configuration for using CC-IN2P3 Slurm farm as a
338 Parsl execution site.
340 Returns
341 -------
342 config : `parsl.config.Config`
343 The configuration to be used to initialize Parsl for this site.
344 """
345 executors = self.get_executors()
346 monitor = self.get_monitor()
348 # Number of retries in case of job failure.
349 retries = get_bps_config_value(self.site, ".retries", int, 0)
351 # Path to run directory.
352 run_dir = get_bps_config_value(self.site, ".run_dir", str, "parsl_runinfo")
354 # Strategy for scaling blocks according to workflow needs.
355 # Use a strategy that allows for scaling up and down Parsl workers.
356 strategy = get_bps_config_value(self.site, ".strategy", str, "htex_auto_scale")
358 return parsl.config.Config(
359 executors=executors,
360 monitoring=monitor,
361 retries=retries,
362 checkpoint_mode="task_exit",
363 run_dir=run_dir,
364 strategy=strategy,
365 )