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