From d8a4ecd68fc29d6b0840b5dd4979e4f166a413b3 Mon Sep 17 00:00:00 2001 From: karajan1001 Date: Wed, 31 Aug 2022 15:02:38 +0800 Subject: [PATCH] Remove the stale manager code --- .../experiments/executor/manager/__init__.py | 0 dvc/repo/experiments/executor/manager/base.py | 258 ------------------ .../experiments/executor/manager/local.py | 101 ------- dvc/repo/experiments/executor/manager/ssh.py | 109 -------- dvc/repo/experiments/queue/celery.py | 1 - 5 files changed, 469 deletions(-) delete mode 100644 dvc/repo/experiments/executor/manager/__init__.py delete mode 100644 dvc/repo/experiments/executor/manager/base.py delete mode 100644 dvc/repo/experiments/executor/manager/local.py delete mode 100644 dvc/repo/experiments/executor/manager/ssh.py diff --git a/dvc/repo/experiments/executor/manager/__init__.py b/dvc/repo/experiments/executor/manager/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/dvc/repo/experiments/executor/manager/base.py b/dvc/repo/experiments/executor/manager/base.py deleted file mode 100644 index b487a2da6..000000000 --- a/dvc/repo/experiments/executor/manager/base.py +++ /dev/null @@ -1,258 +0,0 @@ -import logging -import os -from abc import ABC -from collections import defaultdict, deque -from collections.abc import Mapping -from typing import TYPE_CHECKING, Deque, Dict, Generator, Optional, Tuple, Type - -from dvc.proc.manager import ProcessManager - -from ...exceptions import CheckpointExistsError, ExperimentExistsError -from ...refs import ExpRefInfo -from ...stash import ExpStashEntry -from ..base import EXEC_PID_DIR, BaseExecutor -from ..local import TempDirExecutor, WorkspaceExecutor - -if TYPE_CHECKING: - from scmrepo.git import Git - - from dvc.repo import Repo - -logger = logging.getLogger(__name__) - - -class BaseExecutorManager(ABC, Mapping): # noqa: B024 - """Manages executors for a collection of experiments to be run.""" - - EXECUTOR_CLS: Type = BaseExecutor - - def __init__( - self, - scm: "Git", - wdir: str, - **kwargs, - ): - from dvc.utils.fs import makedirs - - self.scm = scm - makedirs(wdir, exist_ok=True) - self.wdir = wdir - self.proc = ProcessManager(self.pid_dir) - self._attached: Dict[str, "BaseExecutor"] = {} - self._detached: Dict[str, "BaseExecutor"] = dict(self._load_infos()) - self._queue: Deque[Tuple[str, "BaseExecutor"]] = deque() - - def __getitem__(self, key: str) -> "BaseExecutor": - try: - return self._attached[key] - except KeyError: - pass - return self._detached[key] - - def __iter__(self): - yield from self._attached - yield from self._detached - - def __len__(self): - return len(self._attached) + len(self._detached) - - @property - def pid_dir(self) -> str: - return os.path.join(self.wdir, EXEC_PID_DIR) - - def enqueue(self, rev: str, executor: "BaseExecutor"): - assert rev not in self - self._queue.append((rev, executor)) - - def _load_infos(self) -> Generator[Tuple[str, "BaseExecutor"], None, None]: - import json - from urllib.parse import urlparse - - from ..base import ExecutorInfo - from ..ssh import SSHExecutor - - def make_executor(info: "ExecutorInfo"): - if info.git_url: - scheme = urlparse(info.git_url).scheme - if scheme == "file": - cls: Type = TempDirExecutor - elif scheme == "ssh": - cls = SSHExecutor - else: - raise NotImplementedError - else: - cls = WorkspaceExecutor - return cls.from_info(info) - - for name in self.proc: - infofile = self.get_infofile_path(name) - try: - with open(infofile, encoding="utf-8") as fobj: - info = ExecutorInfo.from_dict(json.load(fobj)) - yield name, make_executor(info) - except OSError: - continue - - def get_infofile_path(self, name: str) -> str: - return os.path.join( - self.pid_dir, - name, - f"{name}{BaseExecutor.INFOFILE_EXT}", - ) - - @classmethod - def from_stash_entries( - cls, - scm: "Git", - wdir: str, - repo: "Repo", - to_run: Dict[str, ExpStashEntry], - **kwargs, - ): - manager = cls(scm, wdir) - manager._enqueue_stash_entries(scm, repo, to_run, **kwargs) - return manager - - def _enqueue_stash_entries( - self, - scm: "Git", - repo: "Repo", - to_run: Dict[str, ExpStashEntry], - **kwargs, - ): - for stash_rev, entry in to_run.items(): - executor = self.EXECUTOR_CLS.from_stash_entry( - repo, - entry, - **kwargs, - ) - - executor.init_git(scm, stash_rev, entry, branch=entry.branch) - executor.init_cache(repo, stash_rev) - - self.enqueue(stash_rev, executor) - - def exec_queue( - self, repo: "Repo", jobs: Optional[int] = 1, detach: bool = False - ): - """Run dvc repro for queued executors in parallel.""" - if detach: - raise NotImplementedError - # TODO use ProcessManager.spawn() to support detached runs - return self._exec_attached(repo, jobs=jobs) - - def _exec_attached(self, repo: "Repo", jobs: Optional[int] = 1): - import signal - from concurrent.futures import ( - CancelledError, - ProcessPoolExecutor, - wait, - ) - from multiprocessing import Manager - - from dvc.stage.monitor import CheckpointKilledError - - result: Dict[str, Dict[str, str]] = defaultdict(dict) - - manager = Manager() - pid_q = manager.Queue() - - with ProcessPoolExecutor(max_workers=jobs) as workers: - futures = {} - while self._queue: - rev, executor = self._queue.popleft() - infofile = self.get_infofile_path(rev) - future = workers.submit( - executor.reproduce, - info=executor.info, - rev=rev, - queue=pid_q, - infofile=infofile, - log_level=logger.getEffectiveLevel(), - ) - futures[future] = (rev, executor) - self._attached[rev] = executor - - try: - wait(futures) - except KeyboardInterrupt: - # forward SIGINT to any running executor processes and - # cancel any remaining futures - workers.shutdown(wait=False) - pids = {} - for future, (rev, _) in futures.items(): - if future.running(): - # if future has already been started by the scheduler - # we still have to wait until it tells us its PID - while rev not in pids: - rev, pid = pid_q.get() - pids[rev] = pid - os.kill(pids[rev], signal.SIGINT) - elif not future.done(): - future.cancel() - - for future, (rev, executor) in futures.items(): - rev, executor = futures[future] - - try: - exc = future.exception() - if exc is None: - exec_result = future.result() - result[rev].update( - self._collect_executor(repo, executor, exec_result) - ) - elif not isinstance(exc, CheckpointKilledError): - logger.error( - "Failed to reproduce experiment '%s'", rev[:7] - ) - except CancelledError: - logger.error( - "Cancelled before attempting to reproduce experiment " - "'%s'", - rev[:7], - ) - finally: - self.cleanup_executor(rev, executor) - - return result - - def _collect_executor(self, repo, executor, exec_result) -> Dict[str, str]: - # NOTE: GitPython Repo instances cannot be re-used - # after process has received SIGINT or SIGTERM, so we - # need this hack to re-instantiate git instances after - # checkpoint runs. See: - # https://github.com/gitpython-developers/GitPython/issues/427 - # del self.repo.scm - - results = {} - - def on_diverged(ref: str, checkpoint: bool): - ref_info = ExpRefInfo.from_ref(ref) - if checkpoint: - raise CheckpointExistsError(ref_info.name) - raise ExperimentExistsError(ref_info.name) - - for ref in executor.fetch_exps( - self.scm, - force=exec_result.force, - on_diverged=on_diverged, - ): - exp_rev = self.scm.get_ref(ref) - if exp_rev: - logger.debug("Collected experiment '%s'.", exp_rev[:7]) - results[exp_rev] = exec_result.exp_hash - - if exec_result.ref_info is not None: - executor.collect_cache(repo, exec_result.ref_info) - - return results - - def cleanup_executor(self, rev: str, executor: "BaseExecutor"): - from dvc.utils.fs import remove - - executor.cleanup() - try: - self.proc.remove(rev) - except KeyError: - pass - remove(os.path.join(self.pid_dir, rev)) diff --git a/dvc/repo/experiments/executor/manager/local.py b/dvc/repo/experiments/executor/manager/local.py deleted file mode 100644 index b4659a3b7..000000000 --- a/dvc/repo/experiments/executor/manager/local.py +++ /dev/null @@ -1,101 +0,0 @@ -import logging -from collections import defaultdict -from typing import TYPE_CHECKING, Dict, Optional - -from ...refs import EXEC_BRANCH -from ...stash import ExpStashEntry -from ..local import TempDirExecutor, WorkspaceExecutor -from .base import BaseExecutorManager - -if TYPE_CHECKING: - from scmrepo.git import Git - - from dvc.repo import Repo - -logger = logging.getLogger(__name__) - - -class TempDirExecutorManager(BaseExecutorManager): - EXECUTOR_CLS = TempDirExecutor - - -class WorkspaceExecutorManager(BaseExecutorManager): - EXECUTOR_CLS = WorkspaceExecutor - - @classmethod - def from_stash_entries( - cls, - scm: "Git", - wdir: str, - repo: "Repo", - to_run: Dict[str, ExpStashEntry], - **kwargs, - ): - manager = cls(scm, wdir) - assert len(to_run) == 1 - for stash_rev, entry in to_run.items(): - executor = cls.EXECUTOR_CLS.from_stash_entry( - repo, - entry, - **kwargs, - ) - executor.init_git(repo.scm, stash_rev, entry, branch=entry.branch) - executor.init_cache(repo, stash_rev) - - manager.enqueue(stash_rev, executor) - return manager - - def _collect_executor(self, repo, executor, exec_result) -> Dict[str, str]: - results = {} - exp_rev = self.scm.get_ref(EXEC_BRANCH) - if exp_rev: - logger.debug("Collected experiment '%s'.", exp_rev[:7]) - results[exp_rev] = exec_result.exp_hash - return results - - def exec_queue( - self, repo: "Repo", jobs: Optional[int] = 1, detach: bool = False - ): - """Run a single WorkspaceExecutor. - - Workspace execution is done within the main DVC process - (rather than in multiprocessing context) - """ - from dvc.exceptions import DvcException - from dvc.stage.monitor import CheckpointKilledError - - assert len(self._queue) == 1 - assert not detach - result: Dict[str, Dict[str, str]] = defaultdict(dict) - rev, executor = self._queue.popleft() - - exec_name = "workspace" - infofile = self.get_infofile_path(exec_name) - try: - exec_result = executor.reproduce( - info=executor.info, - rev=rev, - infofile=infofile, - log_level=logger.getEffectiveLevel(), - log_errors=not isinstance(executor, WorkspaceExecutor), - ) - if not exec_result.exp_hash: - raise DvcException( - f"Failed to reproduce experiment '{rev[:7]}'" - ) - if exec_result.ref_info: - result[rev].update( - self._collect_executor(repo, executor, exec_result) - ) - except CheckpointKilledError: - # Checkpoint errors have already been logged - return {} - except DvcException: - raise - except Exception as exc: - raise DvcException( - f"Failed to reproduce experiment '{rev[:7]}'" - ) from exc - finally: - self.cleanup_executor(exec_name, executor) - return result diff --git a/dvc/repo/experiments/executor/manager/ssh.py b/dvc/repo/experiments/executor/manager/ssh.py deleted file mode 100644 index c7960bf5a..000000000 --- a/dvc/repo/experiments/executor/manager/ssh.py +++ /dev/null @@ -1,109 +0,0 @@ -import logging -import posixpath -from collections import defaultdict -from typing import TYPE_CHECKING, Callable, Dict, Generator, Optional, Tuple - -from ...stash import ExpStashEntry -from ..base import BaseExecutor -from ..ssh import SSHExecutor, _sshfs -from .base import BaseExecutorManager - -if TYPE_CHECKING: - from scmrepo.git import Git - - from dvc.repo import Repo - -logger = logging.getLogger(__name__) - - -class SSHExecutorManager(BaseExecutorManager): - EXECUTOR_CLS = SSHExecutor - - def __init__( - self, - scm: "Git", - wdir: str, - host: Optional[str] = None, - port: Optional[int] = None, - username: Optional[str] = None, - fs_factory: Optional[Callable] = None, - **kwargs, - ): - assert host - super().__init__(scm, wdir, **kwargs) - self.host = host - self.port = port - self.username = username - self._fs_factory = fs_factory - - def _load_infos(self) -> Generator[Tuple[str, "BaseExecutor"], None, None]: - # TODO: load existing infos using sshfs - yield from [] - - @classmethod - def from_stash_entries( - cls, - scm: "Git", - wdir: str, - repo: "Repo", - to_run: Dict[str, ExpStashEntry], - **kwargs, - ): - machine_name: Optional[str] = kwargs.get("machine_name", None) - manager = cls( - scm, wdir, **repo.machine.get_executor_kwargs(machine_name) - ) - manager._enqueue_stash_entries(scm, repo, to_run, **kwargs) - return manager - - def sshfs(self): - return _sshfs(self._fs_factory, host=self.host, port=self.port) - - def get_infofile_path(self, name: str) -> str: - return f"{name}{BaseExecutor.INFOFILE_EXT}" - - def _exec_attached(self, repo: "Repo", jobs: Optional[int] = 1): - from dvc.exceptions import DvcException - from dvc.stage.monitor import CheckpointKilledError - - assert len(self._queue) == 1 - result: Dict[str, Dict[str, str]] = defaultdict(dict) - rev, executor = self._queue.popleft() - info = executor.info - infofile = posixpath.join( - info.root_dir, - info.dvc_dir, - "tmp", - self.get_infofile_path(rev), - ) - try: - exec_result = executor.reproduce( - info=executor.info, - rev=rev, - infofile=infofile, - log_level=logger.getEffectiveLevel(), - fs_factory=self._fs_factory, - ) - if not exec_result.exp_hash: - raise DvcException( - f"Failed to reproduce experiment '{rev[:7]}'" - ) - if exec_result.ref_info: - result[rev].update( - self._collect_executor(repo, executor, exec_result) - ) - except CheckpointKilledError: - # Checkpoint errors have already been logged - return {} - except DvcException: - raise - except Exception as exc: - raise DvcException( - f"Failed to reproduce experiment '{rev[:7]}'" - ) from exc - finally: - self.cleanup_executor(rev, executor) - return result - - def cleanup_executor(self, rev: str, executor: "BaseExecutor"): - executor.cleanup() diff --git a/dvc/repo/experiments/queue/celery.py b/dvc/repo/experiments/queue/celery.py index 734d7dcef..0e85e9393 100644 --- a/dvc/repo/experiments/queue/celery.py +++ b/dvc/repo/experiments/queue/celery.py @@ -253,7 +253,6 @@ class LocalCeleryQueue(BaseStashQueue): def _load_collected(rev: str) -> Optional[ExecutorResult]: executor_info = _load_info(rev) - print("executor_info is", executor_info.status) if executor_info.status > TaskStatus.SUCCESS: return executor_info.result raise FileNotFoundError