11554988762

Committed 28 Oct 2024 01:28PM UTC coverage: 92.418% (-2.0%) from 94.369%

Build # 11554988762

Build Type

Pull #459

github

Committed by

web-flow

Commit Message

Merge e33fa6df5 into de646f910

Pull Request Pull Request #459: Cache: create method

Run Details

6 of 25 new or added lines in 2 files covered. (24.0%)

6 existing lines in 1 file now uncovered.

841 of 910 relevant lines covered (92.42%)

0.92 hits per line

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

53.66

/executorlib/cache/executor.py

import os
from typing import Optional

from executorlib.base.executor import ExecutorBase
from executorlib.cache.shared import execute_tasks_h5
from executorlib.standalone.cache.spawner import (
    execute_in_subprocess,
    terminate_subprocess,
)
from executorlib.standalone.inputcheck import (
    check_executor,
    check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
    from executorlib.standalone.cache.queue import execute_with_pysqa
except ImportError:
    # If pysqa is not available fall back to executing tasks in a subprocess
    execute_with_pysqa = execute_in_subprocess


class FileExecutor(ExecutorBase):
    def __init__(
        self,
        cache_directory: str = "cache",
        resource_dict: Optional[dict] = None,
        execute_function: callable = execute_with_pysqa,
        terminate_function: Optional[callable] = None,
        pysqa_config_directory: Optional[str] = None,
        backend: Optional[str] = None,
    ):
        """
        Initialize the FileExecutor.

        Args:
            cache_directory (str, optional): The directory to store cache files. Defaults to "cache".
            resource_dict (dict): A dictionary of resources required by the task. With the following keys:
                              - cores (int): number of MPI cores to be used for each function call
                              - cwd (str/None): current working directory where the parallel python task is executed
            execute_function (callable, optional): The function to execute tasks. Defaults to execute_in_subprocess.
            terminate_function (callable, optional): The function to terminate the tasks.
            pysqa_config_directory (str, optional): path to the pysqa config directory (only for pysqa based backend).
            backend (str, optional): name of the backend used to spawn tasks.
        """
        super().__init__()
        default_resource_dict = {
            "cores": 1,
            "cwd": None,
        }
        if resource_dict is None:
            resource_dict = {}
        resource_dict.update(
            {k: v for k, v in default_resource_dict.items() if k not in resource_dict}
        )
        if execute_function == execute_in_subprocess and terminate_function is None:
            terminate_function = terminate_subprocess
        cache_directory_path = os.path.abspath(cache_directory)
        os.makedirs(cache_directory_path, exist_ok=True)
        self._set_process(
            RaisingThread(
                target=execute_tasks_h5,
                kwargs={
                    "future_queue": self._future_queue,
                    "execute_function": execute_function,
                    "cache_directory": cache_directory_path,
                    "resource_dict": resource_dict,
                    "terminate_function": terminate_function,
                    "pysqa_config_directory": pysqa_config_directory,
                    "backend": backend,
                },
            )
        )


def create_file_executor(
    max_workers: int = 1,
    backend: str = "pysqa_flux",
    max_cores: int = 1,
    cache_directory: Optional[str] = None,
    resource_dict: Optional[dict] = None,
    flux_executor=None,
    flux_executor_pmi_mode: Optional[str] = None,
    flux_executor_nesting: bool = False,
    pysqa_config_directory: Optional[str] = None,
    hostname_localhost: Optional[bool] = None,
    block_allocation: bool = False,
    init_function: Optional[callable] = None,
):
    if cache_directory is None:
        cache_directory = "executorlib_cache"
    if max_workers != 1:
        raise ValueError(
            "The number of workers cannot be controlled with the pysqa based backend."
        )
    if max_cores != 1:
        raise ValueError(
            "The number of cores cannot be controlled with the pysqa based backend."
        )
    if hostname_localhost is not None:
        raise ValueError(
            "The option to connect to hosts based on their hostname is not available with the pysqa based backend."
        )
    if block_allocation:
        raise ValueError(
            "The option block_allocation is not available with the pysqa based backend."
        )
    if init_function is not None:
        raise ValueError(
            "The option to specify an init_function is not available with the pysqa based backend."
        )
    if flux_executor_pmi_mode is not None:
        raise ValueError(
            "The option to specify the flux pmi mode is not available with the pysqa based backend."
        )
    check_executor(executor=flux_executor)
    check_nested_flux_executor(nested_flux_executor=flux_executor_nesting)
    return FileExecutor(
        cache_directory=cache_directory,
        resource_dict=resource_dict,
        pysqa_config_directory=pysqa_config_directory,
        backend=backend.split("pysqa_")[-1],
    )

1	import os	1✔
2	from typing import Optional	1✔
3
4	from executorlib.base.executor import ExecutorBase	1✔
5	from executorlib.cache.shared import execute_tasks_h5	1✔
6	from executorlib.standalone.cache.spawner import (	1✔
7	execute_in_subprocess,
8	terminate_subprocess,
9	)
10	from executorlib.standalone.inputcheck import (	1✔
11	check_executor,
12	check_nested_flux_executor,
13	)
14	from executorlib.standalone.thread import RaisingThread	1✔
15
16	try:	1✔
17	from executorlib.standalone.cache.queue import execute_with_pysqa	1✔
18	except ImportError:	×
19	# If pysqa is not available fall back to executing tasks in a subprocess
20	execute_with_pysqa = execute_in_subprocess	×
21
22
23	class FileExecutor(ExecutorBase):	1✔
24	def __init__(	1✔
25	self,
26	cache_directory: str = "cache",
27	resource_dict: Optional[dict] = None,
28	execute_function: callable = execute_with_pysqa,
29	terminate_function: Optional[callable] = None,
30	pysqa_config_directory: Optional[str] = None,
31	backend: Optional[str] = None,
32	):
33	"""
34	Initialize the FileExecutor.
35
36	Args:
37	cache_directory (str, optional): The directory to store cache files. Defaults to "cache".
38	resource_dict (dict): A dictionary of resources required by the task. With the following keys:
39	- cores (int): number of MPI cores to be used for each function call
40	- cwd (str/None): current working directory where the parallel python task is executed
41	execute_function (callable, optional): The function to execute tasks. Defaults to execute_in_subprocess.
42	terminate_function (callable, optional): The function to terminate the tasks.
43	pysqa_config_directory (str, optional): path to the pysqa config directory (only for pysqa based backend).
44	backend (str, optional): name of the backend used to spawn tasks.
45	"""
46	super().__init__()	1✔
47	default_resource_dict = {	1✔
48	"cores": 1,
49	"cwd": None,
50	}
51	if resource_dict is None:	1✔
52	resource_dict = {}	1✔
53	resource_dict.update(	1✔
54	{k: v for k, v in default_resource_dict.items() if k not in resource_dict}
55	)
56	if execute_function == execute_in_subprocess and terminate_function is None:	1✔
57	terminate_function = terminate_subprocess	1✔
58	cache_directory_path = os.path.abspath(cache_directory)	1✔
59	os.makedirs(cache_directory_path, exist_ok=True)	1✔
60	self._set_process(	1✔
61	RaisingThread(
62	target=execute_tasks_h5,
63	kwargs={
64	"future_queue": self._future_queue,
65	"execute_function": execute_function,
66	"cache_directory": cache_directory_path,
67	"resource_dict": resource_dict,
68	"terminate_function": terminate_function,
69	"pysqa_config_directory": pysqa_config_directory,
70	"backend": backend,
71	},
72	)
73	)
74
75
76	def create_file_executor(	1✔
77	max_workers: int = 1,
78	backend: str = "pysqa_flux",
79	max_cores: int = 1,
80	cache_directory: Optional[str] = None,
81	resource_dict: Optional[dict] = None,
82	flux_executor=None,
83	flux_executor_pmi_mode: Optional[str] = None,
84	flux_executor_nesting: bool = False,
85	pysqa_config_directory: Optional[str] = None,
86	hostname_localhost: Optional[bool] = None,
87	block_allocation: bool = False,
88	init_function: Optional[callable] = None,
89	):
NEW 90	if cache_directory is None:	×
NEW 91	cache_directory = "executorlib_cache"	×
NEW 92	if max_workers != 1:	×
NEW 93	raise ValueError(	×
94	"The number of workers cannot be controlled with the pysqa based backend."
95	)
NEW 96	if max_cores != 1:	×
NEW 97	raise ValueError(	×
98	"The number of cores cannot be controlled with the pysqa based backend."
99	)
NEW 100	if hostname_localhost is not None:	×
NEW 101	raise ValueError(	×
102	"The option to connect to hosts based on their hostname is not available with the pysqa based backend."
103	)
NEW 104	if block_allocation:	×
NEW 105	raise ValueError(	×
106	"The option block_allocation is not available with the pysqa based backend."
107	)
NEW 108	if init_function is not None:	×
NEW 109	raise ValueError(	×
110	"The option to specify an init_function is not available with the pysqa based backend."
111	)
NEW 112	if flux_executor_pmi_mode is not None:	×
NEW 113	raise ValueError(	×
114	"The option to specify the flux pmi mode is not available with the pysqa based backend."
115	)
NEW 116	check_executor(executor=flux_executor)	×
NEW 117	check_nested_flux_executor(nested_flux_executor=flux_executor_nesting)	×
NEW 118	return FileExecutor(	×
119	cache_directory=cache_directory,
120	resource_dict=resource_dict,
121	pysqa_config_directory=pysqa_config_directory,
122	backend=backend.split("pysqa_")[-1],
123	)

pyiron / executorlib / 11554988762

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