12425443902

Committed 20 Dec 2024 04:17AM UTC coverage: 95.252% (-0.5%) from 95.793%

Build # 12425443902

Build Type

Pull #529

github

Committed by

web-flow

Commit Message

Merge ca71e531d into f1649fab2

Pull Request Pull Request #529: Add validate_max_workers_slurm method

Run Details

4 of 12 new or added lines in 2 files covered. (33.33%)

7 existing lines in 1 file now uncovered.

983 of 1032 relevant lines covered (95.25%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.33

/executorlib/interactive/slurm.py

import os
from typing import Optional

from executorlib.standalone.interactive.spawner import SubprocessSpawner

SLURM_COMMAND = "srun"


def validate_max_workers(max_workers, cores, threads_per_core):
    cores_total = os.environ["SLURM_NTASKS"] * os.environ["SLURM_CPUS_PER_TASK"]
    cores_requested = max_workers * cores * threads_per_core
    if cores_total < cores_requested:
        raise ValueError(
            "The number of requested cores is larger than the available cores "
            + str(cores_total)
            + " < "
            + str(cores_requested)
        )


class SrunSpawner(SubprocessSpawner):
    def __init__(
        self,
        cwd: Optional[str] = None,
        cores: int = 1,
        threads_per_core: int = 1,
        gpus_per_core: int = 0,
        openmpi_oversubscribe: bool = False,
        slurm_cmd_args: list[str] = [],
    ):
        """
        Srun interface implementation.

        Args:
            cwd (str, optional): The current working directory. Defaults to None.
            cores (int, optional): The number of cores to use. Defaults to 1.
            threads_per_core (int, optional): The number of threads per core. Defaults to 1.
            gpus_per_core (int, optional): The number of GPUs per core. Defaults to 0.
            openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
            slurm_cmd_args (list[str], optional): Additional command line arguments. Defaults to [].
        """
        super().__init__(
            cwd=cwd,
            cores=cores,
            openmpi_oversubscribe=openmpi_oversubscribe,
            threads_per_core=threads_per_core,
        )
        self._gpus_per_core = gpus_per_core
        self._slurm_cmd_args = slurm_cmd_args

    def generate_command(self, command_lst: list[str]) -> list[str]:
        """
        Generate the command list for the Srun interface.

        Args:
            command_lst (list[str]): The command list.

        Returns:
            list[str]: The generated command list.
        """
        command_prepend_lst = generate_slurm_command(
            cores=self._cores,
            cwd=self._cwd,
            threads_per_core=self._threads_per_core,
            gpus_per_core=self._gpus_per_core,
            openmpi_oversubscribe=self._openmpi_oversubscribe,
            slurm_cmd_args=self._slurm_cmd_args,
        )
        return super().generate_command(
            command_lst=command_prepend_lst + command_lst,
        )


def generate_slurm_command(
    cores: int,
    cwd: str,
    threads_per_core: int = 1,
    gpus_per_core: int = 0,
    openmpi_oversubscribe: bool = False,
    slurm_cmd_args: list[str] = [],
) -> list[str]:
    """
    Generate the command list for the SLURM interface.

    Args:
        cores (int): The number of cores.
        cwd (str): The current working directory.
        threads_per_core (int, optional): The number of threads per core. Defaults to 1.
        gpus_per_core (int, optional): The number of GPUs per core. Defaults to 0.
        openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
        slurm_cmd_args (list[str], optional): Additional command line arguments. Defaults to [].

    Returns:
        list[str]: The generated command list.
    """
    command_prepend_lst = [SLURM_COMMAND, "-n", str(cores)]
    if cwd is not None:
        command_prepend_lst += ["-D", cwd]
    if threads_per_core > 1:
        command_prepend_lst += ["--cpus-per-task" + str(threads_per_core)]
    if gpus_per_core > 0:
        command_prepend_lst += ["--gpus-per-task=" + str(gpus_per_core)]
    if openmpi_oversubscribe:
        command_prepend_lst += ["--oversubscribe"]
    if len(slurm_cmd_args) > 0:
        command_prepend_lst += slurm_cmd_args
    return command_prepend_lst

1	import os	1✔
2	from typing import Optional	1✔
3
4	from executorlib.standalone.interactive.spawner import SubprocessSpawner	1✔
5
6	SLURM_COMMAND = "srun"	1✔
7
8
9	def validate_max_workers(max_workers, cores, threads_per_core):	1✔
NEW 10	cores_total = os.environ["SLURM_NTASKS"] * os.environ["SLURM_CPUS_PER_TASK"]	×
NEW 11	cores_requested = max_workers * cores * threads_per_core	×
NEW 12	if cores_total < cores_requested:	×
NEW 13	raise ValueError(	×
14	"The number of requested cores is larger than the available cores "
15	+ str(cores_total)
16	+ " < "
17	+ str(cores_requested)
18	)
19
20
21	class SrunSpawner(SubprocessSpawner):	1✔
22	def __init__(	1✔
23	self,
24	cwd: Optional[str] = None,
25	cores: int = 1,
26	threads_per_core: int = 1,
27	gpus_per_core: int = 0,
28	openmpi_oversubscribe: bool = False,
29	slurm_cmd_args: list[str] = [],
30	):
31	"""
32	Srun interface implementation.
33
34	Args:
35	cwd (str, optional): The current working directory. Defaults to None.
36	cores (int, optional): The number of cores to use. Defaults to 1.
37	threads_per_core (int, optional): The number of threads per core. Defaults to 1.
38	gpus_per_core (int, optional): The number of GPUs per core. Defaults to 0.
39	openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
40	slurm_cmd_args (list[str], optional): Additional command line arguments. Defaults to [].
41	"""
42	super().__init__(	1✔
43	cwd=cwd,
44	cores=cores,
45	openmpi_oversubscribe=openmpi_oversubscribe,
46	threads_per_core=threads_per_core,
47	)
48	self._gpus_per_core = gpus_per_core	1✔
49	self._slurm_cmd_args = slurm_cmd_args	1✔
50
51	def generate_command(self, command_lst: list[str]) -> list[str]:	1✔
52	"""
53	Generate the command list for the Srun interface.
54
55	Args:
56	command_lst (list[str]): The command list.
57
58	Returns:
59	list[str]: The generated command list.
60	"""
61	command_prepend_lst = generate_slurm_command(	1✔
62	cores=self._cores,
63	cwd=self._cwd,
64	threads_per_core=self._threads_per_core,
65	gpus_per_core=self._gpus_per_core,
66	openmpi_oversubscribe=self._openmpi_oversubscribe,
67	slurm_cmd_args=self._slurm_cmd_args,
68	)
69	return super().generate_command(	1✔
70	command_lst=command_prepend_lst + command_lst,
71	)
72
73
74	def generate_slurm_command(	1✔
75	cores: int,
76	cwd: str,
77	threads_per_core: int = 1,
78	gpus_per_core: int = 0,
79	openmpi_oversubscribe: bool = False,
80	slurm_cmd_args: list[str] = [],
81	) -> list[str]:
82	"""
83	Generate the command list for the SLURM interface.
84
85	Args:
86	cores (int): The number of cores.
87	cwd (str): The current working directory.
88	threads_per_core (int, optional): The number of threads per core. Defaults to 1.
89	gpus_per_core (int, optional): The number of GPUs per core. Defaults to 0.
90	openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
91	slurm_cmd_args (list[str], optional): Additional command line arguments. Defaults to [].
92
93	Returns:
94	list[str]: The generated command list.
95	"""
96	command_prepend_lst = [SLURM_COMMAND, "-n", str(cores)]	1✔
97	if cwd is not None:	1✔
98	command_prepend_lst += ["-D", cwd]	1✔
99	if threads_per_core > 1:	1✔
100	command_prepend_lst += ["--cpus-per-task" + str(threads_per_core)]	×
101	if gpus_per_core > 0:	1✔
102	command_prepend_lst += ["--gpus-per-task=" + str(gpus_per_core)]	1✔
103	if openmpi_oversubscribe:	1✔
104	command_prepend_lst += ["--oversubscribe"]	1✔
105	if len(slurm_cmd_args) > 0:	1✔
106	command_prepend_lst += slurm_cmd_args	1✔
107	return command_prepend_lst	1✔

pyiron / executorlib / 12425443902

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous