From 9101fce76143b6b1cf10be96615886f3976eb4b3 Mon Sep 17 00:00:00 2001 From: Peter Rowlands Date: Tue, 7 Mar 2023 17:29:59 +0900 Subject: [PATCH] proc: remove dvc.proc --- dvc/proc/__init__.py | 0 dvc/proc/exceptions.py | 23 ---- dvc/proc/manager.py | 149 -------------------------- dvc/proc/process.py | 179 -------------------------------- tests/unit/proc/__init__.py | 0 tests/unit/proc/test_manager.py | 114 -------------------- tests/unit/proc/test_process.py | 61 ----------- 7 files changed, 526 deletions(-) delete mode 100644 dvc/proc/__init__.py delete mode 100644 dvc/proc/exceptions.py delete mode 100644 dvc/proc/manager.py delete mode 100644 dvc/proc/process.py delete mode 100644 tests/unit/proc/__init__.py delete mode 100644 tests/unit/proc/test_manager.py delete mode 100644 tests/unit/proc/test_process.py diff --git a/dvc/proc/__init__.py b/dvc/proc/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/dvc/proc/exceptions.py b/dvc/proc/exceptions.py deleted file mode 100644 index bdf593760..000000000 --- a/dvc/proc/exceptions.py +++ /dev/null @@ -1,23 +0,0 @@ -from dvc.exceptions import DvcException - - -class ProcessNotTerminatedError(DvcException): - def __init__(self, name): - super().__init__(f"Managed process '{name}' has not been terminated.") - - -class ProcessNotFoundError(DvcException): - def __init__(self, name): - super().__init__(f"Managed process '{name}' does not exist.") - - -class TimeoutExpired(DvcException): - def __init__(self, cmd, timeout): - super().__init__(f"'{cmd}' did not complete before timeout '{timeout}'") - self.cmd = cmd - self.timeout = timeout - - -class UnsupportedSignalError(DvcException): - def __init__(self, sig): - super().__init__(f"Unsupported signal: {sig}") diff --git a/dvc/proc/manager.py b/dvc/proc/manager.py deleted file mode 100644 index 4fc9d1a89..000000000 --- a/dvc/proc/manager.py +++ /dev/null @@ -1,149 +0,0 @@ -"""Serverless process manager.""" - -import json -import logging -import os -import signal -import sys -from typing import Generator, List, Optional, Tuple, Union - -from funcy.flow import reraise -from shortuuid import uuid - -from .exceptions import ProcessNotTerminatedError, UnsupportedSignalError -from .process import ManagedProcess, ProcessInfo - -logger = logging.getLogger(__name__) - - -class ProcessManager: - """Manager for controlling background ManagedProcess(es). - - Spawned process entries are kept in the manager directory until they - are explicitly removed (with remove() or cleanup()) so that return - value and log information can be accessed after a process has completed. - """ - - def __init__(self, wdir: Optional[str] = None): - self.wdir = wdir or "." - - def __iter__(self) -> Generator[str, None, None]: - if not os.path.exists(self.wdir): - return - yield from os.listdir(self.wdir) - - def __getitem__(self, key: str) -> "ProcessInfo": - info_path = os.path.join(self.wdir, key, f"{key}.json") - try: - with open(info_path, encoding="utf-8") as fobj: - return ProcessInfo.from_dict(json.load(fobj)) - except FileNotFoundError: - raise KeyError # noqa: B904 - - @reraise(FileNotFoundError, KeyError) - def __setitem__(self, key: str, value: "ProcessInfo"): - info_path = os.path.join(self.wdir, key, f"{key}.json") - with open(info_path, "w", encoding="utf-8") as fobj: - return json.dump(value.asdict(), fobj) - - def __delitem__(self, key: str) -> None: - from dvc.utils.fs import remove - - path = os.path.join(self.wdir, key) - if os.path.exists(path): - remove(path) - - def get(self, key: str, default=None): - try: - return self[key] - except KeyError: - return default - - def processes(self) -> Generator[Tuple[str, "ProcessInfo"], None, None]: - for name in self: - try: - yield name, self[name] - except KeyError: - continue - - def spawn(self, args: Union[str, List[str]], name: Optional[str] = None): - """Run the given command in the background.""" - name = name or uuid() - pid = ManagedProcess.spawn( - args, - wdir=os.path.join(self.wdir, name), - name=name, - ) - logger.debug( - "Spawned managed process '%s' (PID: '%d')", - name, - pid, - ) - - def send_signal(self, name: str, sig: int): - """Send `signal` to the specified named process.""" - process_info = self[name] - if sys.platform == "win32" and sig not in ( - signal.SIGTERM, - signal.CTRL_C_EVENT, # pylint: disable=no-member - signal.CTRL_BREAK_EVENT, # pylint: disable=no-member - ): - raise UnsupportedSignalError(sig) - - def handle_closed_process(): - logging.warning("Process %s had already aborted unexpectedly.", name) - process_info.returncode = -1 - self[name] = process_info - - if process_info.returncode is None: - try: - os.kill(process_info.pid, sig) - except ProcessLookupError: - handle_closed_process() - raise - except OSError as exc: - if getattr(exc, "winerror", None) == 87: - handle_closed_process() - raise ProcessLookupError from exc - raise - - def terminate(self, name: str): - """Terminate the specified named process.""" - self.send_signal(name, signal.SIGTERM) - - def kill(self, name: str): - """Kill the specified named process.""" - if sys.platform == "win32": - self.send_signal(name, signal.SIGTERM) - else: - self.send_signal(name, signal.SIGKILL) # pylint: disable=no-member - - def remove(self, name: str, force: bool = False): - """Remove the specified named process from this manager. - - If the specified process is still running, it will be forcefully killed - if `force` is True`, otherwise an exception will be raised. - - Raises: - ProcessNotTerminatedError if the specified process is still - running and was not forcefully killed. - """ - try: - process_info = self[name] - except KeyError: - return - if process_info.returncode is None and not force: - raise ProcessNotTerminatedError(name) - try: - self.kill(name) - except ProcessLookupError: - pass - del self[name] - - def cleanup(self, force: bool = False): - """Remove stale (terminated) processes from this manager.""" - for name in self: - try: - self.remove(name, force) - except ProcessNotTerminatedError: - continue diff --git a/dvc/proc/process.py b/dvc/proc/process.py deleted file mode 100644 index 3d999ef5c..000000000 --- a/dvc/proc/process.py +++ /dev/null @@ -1,179 +0,0 @@ -import json -import logging -import os -import shlex -import subprocess -from contextlib import AbstractContextManager -from dataclasses import asdict, dataclass -from typing import BinaryIO, List, Optional, Union - -from shortuuid import uuid - -from dvc.utils.objects import cached_property - -from .exceptions import TimeoutExpired - -logger = logging.getLogger(__name__) - - -@dataclass -class ProcessInfo: - pid: int - stdin: Optional[str] - stdout: Optional[str] - stderr: Optional[str] - returncode: Optional[int] - - @classmethod - def from_dict(cls, d): - return cls(**d) - - def asdict(self): - return asdict(self) - - -class ManagedProcess(AbstractContextManager): - """Run the specified command with redirected output. - - stdout and stderr will both be redirected to .out. - Interactive processes (requiring stdin input) are currently unsupported. - - Parameters: - args: Command to be run. - wdir: If specified, redirected output files will be placed in `wdir`. - name: Name to use for this process, if not specified a UUID will be - generated instead. - """ - - def __init__( - self, - args: Union[str, List[str]], - wdir: Optional[str] = None, - name: Optional[str] = None, - ): - self.args: List[str] = ( - shlex.split(args, posix=os.name == "posix") - if isinstance(args, str) - else list(args) - ) - self.wdir = wdir - self.name = name or uuid() - self.returncode: Optional[int] = None - self._stdout: Optional[BinaryIO] = None - self._stderr: Optional[BinaryIO] = None - - self._run() - - def __exit__(self, exc_type, exc_value, traceback): - self.wait() - - def _close_fds(self): - if self._stdout: - self._stdout.close() - self._stdout = None - if self._stderr: - self._stderr.close() - self._stderr = None - - def _make_path(self, path: str) -> str: - return os.path.join(self.wdir, path) if self.wdir else path - - @cached_property - def stdout_path(self) -> str: - return self._make_path(f"{self.name}.out") - - @cached_property - def info_path(self) -> str: - return self._make_path(f"{self.name}.json") - - @cached_property - def pidfile_path(self) -> str: - return self._make_path(f"{self.name}.pid") - - @property - def info(self) -> "ProcessInfo": - return ProcessInfo( - pid=self.pid, - stdin=None, - stdout=self.stdout_path, - stderr=None, - returncode=self.returncode, - ) - - def _make_wdir(self): - if self.wdir: - os.makedirs(self.wdir, exist_ok=True) - - def _dump(self): - self._make_wdir() - with open(self.info_path, "w", encoding="utf-8") as fobj: - json.dump(self.info.asdict(), fobj) - with open(self.pidfile_path, "w", encoding="utf-8") as fobj: - fobj.write(str(self.pid)) - - def _run(self) -> None: - self._make_wdir() - logger.debug( - "Appending output to '%s'", - self.stdout_path, - ) - self._stdout = open(self.stdout_path, "ab") # noqa: SIM115 - try: - self._proc = subprocess.Popen( - self.args, - stdin=subprocess.DEVNULL, - stdout=self._stdout, - stderr=subprocess.STDOUT, - close_fds=True, - shell=False, - ) - self.pid: int = self._proc.pid - self._dump() - except Exception: - if self._proc is not None: - self._proc.kill() - self._close_fds() - raise - - def wait(self, timeout: Optional[int] = None) -> Optional[int]: - """Block until a process started with `run` has completed. - - Raises: - TimeoutExpired if `timeout` was set and the process - did not terminate after `timeout` seconds. - """ - if self.returncode is not None: - return self.returncode - try: - self._proc.wait(timeout=timeout) - except subprocess.TimeoutExpired as exc: - raise TimeoutExpired(exc.cmd, exc.timeout) from exc - self.returncode = self._proc.returncode - self._close_fds() - self._dump() - return self.returncode - - @classmethod - def spawn(cls, *args, **kwargs) -> Optional[int]: - """Spawn a ManagedProcess command in the background. - - Returns: The spawned process PID. - """ - import multiprocessing as mp - - proc = mp.Process( - target=cls._spawn, - args=args, - kwargs=kwargs, - daemon=True, - ) - proc.start() - # Do not terminate the child daemon when the main process exits - # pylint: disable=protected-access - mp.process._children.discard(proc) # type: ignore[attr-defined] - return proc.pid - - @classmethod - def _spawn(cls, *args, **kwargs): - with cls(*args, **kwargs): - pass diff --git a/tests/unit/proc/__init__.py b/tests/unit/proc/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/unit/proc/test_manager.py b/tests/unit/proc/test_manager.py deleted file mode 100644 index f01a7f8ff..000000000 --- a/tests/unit/proc/test_manager.py +++ /dev/null @@ -1,114 +0,0 @@ -import json -import os -import signal -import sys - -import pytest - -from dvc.proc.exceptions import ProcessNotTerminatedError, UnsupportedSignalError -from dvc.proc.manager import ProcessManager -from dvc.proc.process import ProcessInfo - -PID_FINISHED = 1234 -PID_RUNNING = 5678 - - -def create_process(root: str, name: str, pid: int, returncode=None): - info_path = os.path.join(root, name, f"{name}.json") - os.makedirs(os.path.join(root, name)) - process_info = ProcessInfo( - pid=pid, stdin=None, stdout=None, stderr=None, returncode=returncode - ) - with open(info_path, "w", encoding="utf-8") as fobj: - json.dump(process_info.asdict(), fobj) - - -@pytest.fixture -def finished_process(tmp_dir): - key = "finished" - create_process(tmp_dir, key, PID_FINISHED, 0) - return key - - -@pytest.fixture -def running_process(tmp_dir): - key = "running" - create_process(tmp_dir, key, PID_RUNNING) - return key - - -def test_send_signal(tmp_dir, mocker, finished_process, running_process): - m = mocker.patch("os.kill") - process_manager = ProcessManager(tmp_dir) - process_manager.send_signal(running_process, signal.SIGTERM) - m.assert_called_once_with(PID_RUNNING, signal.SIGTERM) - - m.reset_mock() - process_manager.send_signal(finished_process, signal.SIGTERM) - m.assert_not_called() - - if sys.platform == "win32": - with pytest.raises(UnsupportedSignalError): - process_manager.send_signal(finished_process, signal.SIGABRT) - - -def test_dead_process(tmp_dir, mocker, running_process): - process_manager = ProcessManager(tmp_dir) - - def side_effect(*args): - if sys.platform == "win32": - err = OSError() - err.winerror = 87 - raise err - raise ProcessLookupError() - - mocker.patch("os.kill", side_effect=side_effect) - with pytest.raises(ProcessLookupError): - process_manager.send_signal(running_process, signal.SIGTERM) - assert process_manager[running_process].returncode == -1 - - -def test_kill(tmp_dir, mocker, finished_process, running_process): - m = mocker.patch("os.kill") - process_manager = ProcessManager(tmp_dir) - process_manager.kill(running_process) - if sys.platform == "win32": - m.assert_called_once_with(PID_RUNNING, signal.SIGTERM) - else: - m.assert_called_once_with(PID_RUNNING, signal.SIGKILL) - - m.reset_mock() - process_manager.kill(finished_process) - m.assert_not_called() - - -def test_terminate(tmp_dir, mocker, running_process, finished_process): - m = mocker.patch("os.kill") - process_manager = ProcessManager(tmp_dir) - process_manager.terminate(running_process) - m.assert_called_once_with(PID_RUNNING, signal.SIGTERM) - - m.reset_mock() - process_manager.terminate(finished_process) - m.assert_not_called() - - -def test_remove(mocker, tmp_dir, running_process, finished_process): - mocker.patch("os.kill", return_value=None) - process_manager = ProcessManager(tmp_dir) - process_manager.remove(finished_process) - assert not (tmp_dir / finished_process).exists() - with pytest.raises(ProcessNotTerminatedError): - process_manager.remove(running_process) - assert (tmp_dir / running_process).exists() - process_manager.remove(running_process, True) - assert not (tmp_dir / running_process).exists() - - -@pytest.mark.parametrize("force", [True, False]) -def test_cleanup(mocker, tmp_dir, running_process, finished_process, force): - mocker.patch("os.kill", return_value=None) - process_manager = ProcessManager(tmp_dir) - process_manager.cleanup(force) - assert (tmp_dir / running_process).exists() != force - assert not (tmp_dir / finished_process).exists() diff --git a/tests/unit/proc/test_process.py b/tests/unit/proc/test_process.py deleted file mode 100644 index 7802814aa..000000000 --- a/tests/unit/proc/test_process.py +++ /dev/null @@ -1,61 +0,0 @@ -import json -import subprocess - -import pytest - -from dvc.proc.process import ManagedProcess, ProcessInfo - -TEST_PID = 1234 - - -@pytest.fixture(autouse=True) -def mock_popen(mocker): - mocker.patch( - "subprocess.Popen", - return_value=mocker.MagicMock(pid=TEST_PID, returncode=None), - ) - - -@pytest.mark.parametrize( - "args", - [ - "/bin/foo -o option", - ["/bin/foo", "-o", "option"], - ], -) -def test_init_args(tmp_dir, request, args, mocker): - expected = ["/bin/foo", "-o", "option"] - proc = ManagedProcess(args) - request.addfinalizer(proc._close_fds) - assert expected == proc.args - - -def test_run(tmp_dir, request, mocker): - proc = ManagedProcess("/bin/foo") - - request.addfinalizer(proc._close_fds) - assert proc.pid == TEST_PID - - with open(proc.info_path, encoding="utf-8") as fobj: - info = ProcessInfo.from_dict(json.load(fobj)) - assert info.pid == TEST_PID - - -def test_wait(tmp_dir, mocker): - from dvc.proc.exceptions import TimeoutExpired - - proc = ManagedProcess("/bin/foo") - proc._proc.wait = mocker.Mock(side_effect=subprocess.TimeoutExpired("/bin/foo", 5)) - with pytest.raises(TimeoutExpired): - proc.wait(5) - - proc._proc.wait = mocker.Mock(return_value=None) - proc._proc.returncode = 0 - assert proc.wait() == 0 - proc._proc.wait.assert_called_once() - - # once subprocess return code is set, future ManagedProcess.wait() calls - # should not block - proc._proc.wait.reset_mock() - assert proc.wait() == 0 - proc._proc.wait.assert_not_called()