From 6b9d842b26992e94e55843a3a788267608ff60ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20Rowlands=20=28=EB=B3=80=EA=B8=B0=ED=98=B8=29?= Date: Wed, 16 Dec 2020 02:54:04 +0900 Subject: [PATCH] experiments: handle `exp run` from within subdirs/subrepos (#5093) * experiments: add test cases for stage in subdir & subrepo * git: fix relpath handling in dulwich add() * repro: move exp file git staging from stage.run into repro * experiments: support running experiments from outside DVC root * experiments: setup logger inside multiprocessing (executor) context * git: fix dulwich add in submodules --- dvc/repo/experiments/__init__.py | 6 +- dvc/repo/experiments/executor.py | 24 +++++-- dvc/repo/reproduce.py | 12 ++++ dvc/scm/git/backend/dulwich.py | 36 ++++++++-- dvc/stage/run.py | 4 -- tests/func/experiments/test_experiments.py | 77 ++++++++++++++++++++++ 6 files changed, 145 insertions(+), 14 deletions(-) diff --git a/dvc/repo/experiments/__init__.py b/dvc/repo/experiments/__init__.py index 0c95ac373..24720f19a 100644 --- a/dvc/repo/experiments/__init__.py +++ b/dvc/repo/experiments/__init__.py @@ -269,7 +269,7 @@ class Experiments: logger.debug("Using experiment params '%s'", params) for params_fname in params: - path = PathInfo(self.repo.root_dir) / params_fname + path = PathInfo(params_fname) suffix = path.suffix.lower() modify_data = MODIFIERS[suffix] with modify_data(path, tree=self.repo.tree) as data: @@ -496,6 +496,8 @@ class Experiments: manager = Manager() pid_q = manager.Queue() + + rel_cwd = relpath(os.getcwd(), self.repo.root_dir) with ProcessPoolExecutor(max_workers=jobs) as workers: futures = {} for rev, executor in executors.items(): @@ -505,6 +507,8 @@ class Experiments: pid_q, rev, name=executor.name, + rel_cwd=rel_cwd, + log_level=logger.getEffectiveLevel(), ) futures[future] = (rev, executor) diff --git a/dvc/repo/experiments/executor.py b/dvc/repo/experiments/executor.py index dae89f445..cb0fe86cc 100644 --- a/dvc/repo/experiments/executor.py +++ b/dvc/repo/experiments/executor.py @@ -196,8 +196,9 @@ class BaseExecutor: dvc_dir: str, queue: "Queue", rev: str, - cwd: Optional[str] = None, + rel_cwd: Optional[str] = None, name: Optional[str] = None, + log_level: Optional[int] = None, ) -> Tuple[Optional[str], bool]: """Run dvc repro and return the result. @@ -211,6 +212,7 @@ class BaseExecutor: unchanged = [] queue.put((rev, os.getpid())) + cls._set_log_level(log_level) def filter_pipeline(stages): unchanged.extend( @@ -223,9 +225,11 @@ class BaseExecutor: try: dvc = Repo(dvc_dir) old_cwd = os.getcwd() - new_cwd = cwd if cwd else dvc.root_dir - os.chdir(new_cwd) - logger.debug("Running repro in '%s'", cwd) + if rel_cwd: + os.chdir(os.path.join(dvc.root_dir, rel_cwd)) + else: + os.chdir(dvc.root_dir) + logger.debug("Running repro in '%s'", os.getcwd()) args_path = os.path.join( dvc.tmp_dir, BaseExecutor.PACKED_ARGS_FILE @@ -321,6 +325,18 @@ class BaseExecutor: scm.set_ref(EXEC_BRANCH, branch, symbolic=True) return new_rev + @staticmethod + def _set_log_level(level): + from dvc.logger import disable_other_loggers + + # When executor.reproduce is run in a multiprocessing child process, + # dvc.main will not be called for that child process so we need to + # setup logging ourselves + dvc_logger = logging.getLogger("dvc") + disable_other_loggers() + if level is not None: + dvc_logger.setLevel(level) + class LocalExecutor(BaseExecutor): """Local machine experiment executor.""" diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index 99cdd4058..b5385e1c7 100644 --- a/dvc/repo/reproduce.py +++ b/dvc/repo/reproduce.py @@ -1,4 +1,5 @@ import logging +import os import typing from functools import partial @@ -18,6 +19,7 @@ logger = logging.getLogger(__name__) def _reproduce_stage(stage, **kwargs): def _run_callback(repro_callback): _dump_stage(stage) + _track_stage(stage) repro_callback([stage]) checkpoint_func = kwargs.pop("checkpoint_func", None) @@ -42,7 +44,10 @@ def _reproduce_stage(stage, **kwargs): return [] if not kwargs.get("dry", False): + track = checkpoint_func is not None _dump_stage(stage) + if track: + _track_stage(stage) return [stage] @@ -54,6 +59,13 @@ def _dump_stage(stage): dvcfile.dump(stage, update_pipeline=False) +def _track_stage(stage): + for out in stage.outs: + if not out.use_scm_ignore and out.is_in_repo: + stage.repo.scm.track_file(os.fspath(out.path_info)) + stage.repo.scm.track_changed_files() + + def _get_active_graph(G): import networkx as nx diff --git a/dvc/scm/git/backend/dulwich.py b/dvc/scm/git/backend/dulwich.py index f3c4b0543..f721825ef 100644 --- a/dvc/scm/git/backend/dulwich.py +++ b/dvc/scm/git/backend/dulwich.py @@ -4,8 +4,9 @@ import logging import os import stat from io import BytesIO, StringIO -from typing import Callable, Iterable, Optional, Tuple +from typing import Callable, Dict, Iterable, Optional, Tuple +from dvc.path_info import PathInfo from dvc.scm.base import SCMError from dvc.utils import relpath @@ -67,8 +68,24 @@ class DulwichBackend(BaseGitBackend): # pylint:disable=abstract-method except NotGitRepository as exc: raise SCMError(f"{root_dir} is not a git repository") from exc + self._submodules: Dict[str, "PathInfo"] = self._find_submodules() self._stashes: dict = {} + def _find_submodules(self) -> Dict[str, "PathInfo"]: + """Return dict mapping submodule names to submodule paths. + + Submodule paths will be relative to Git repo root. + """ + from dulwich.config import ConfigFile, parse_submodules + + submodules: Dict[str, "PathInfo"] = {} + config_path = os.path.join(self.root_dir, ".gitmodules") + if os.path.isfile(config_path): + config = ConfigFile.from_path(config_path) + for path, _url, section in parse_submodules(config): + submodules[os.fsdecode(section)] = PathInfo(os.fsdecode(path)) + return submodules + def close(self): self.repo.close() @@ -101,8 +118,19 @@ class DulwichBackend(BaseGitBackend): # pylint:disable=abstract-method files = [] for path in paths: - if not os.path.isabs(path): - path = os.path.join(self.root_dir, path) + if not os.path.isabs(path) and self._submodules: + # NOTE: If path is inside a submodule, Dulwich expects the + # staged paths to be relative to the submodule root (not the + # parent git repo root). We append path to root_dir here so + # that the result of relpath(path, root_dir) is actually the + # path relative to the submodule root. + path_info = PathInfo(path).relative_to(self.root_dir) + for sm_path in self._submodules.values(): + if path_info.isin(sm_path): + path = os.path.join( + self.root_dir, path_info.relative_to(sm_path) + ) + break if os.path.isdir(path): files.extend(walk_files(path)) else: @@ -138,8 +166,6 @@ class DulwichBackend(BaseGitBackend): # pylint:disable=abstract-method raise NotImplementedError def is_tracked(self, path: str) -> bool: - from dvc.path_info import PathInfo - rel = PathInfo(path).relative_to(self.root_dir).as_posix().encode() rel_dir = rel + b"/" for path in self.repo.open_index(): diff --git a/dvc/stage/run.py b/dvc/stage/run.py index b7660bfd0..6653770bb 100644 --- a/dvc/stage/run.py +++ b/dvc/stage/run.py @@ -211,9 +211,5 @@ def _kill_nt(proc): def _run_callback(stage, callback_func): stage.save(allow_missing=True) stage.commit(allow_missing=True) - for out in stage.outs: - if not out.use_scm_ignore and out.is_in_repo: - stage.repo.scm.track_file(os.fspath(out.path_info)) - stage.repo.scm.track_changed_files() logger.debug("Running checkpoint callback for stage '%s'", stage) callback_func() diff --git a/tests/func/experiments/test_experiments.py b/tests/func/experiments/test_experiments.py index 0955f0b4a..e0525e094 100644 --- a/tests/func/experiments/test_experiments.py +++ b/tests/func/experiments/test_experiments.py @@ -5,6 +5,7 @@ import stat import pytest from funcy import first +from dvc.dvcfile import PIPELINE_FILE from dvc.repo.experiments.utils import exp_refs_by_rev from dvc.utils.serialize import PythonFileCorruptedError from tests.func.test_repro_multistage import COPY_SCRIPT @@ -436,3 +437,79 @@ def test_list(tmp_dir, scm, dvc, exp_stage): baseline_a: {ref_info_a.name, ref_info_b.name}, baseline_c: {ref_info_c.name}, } + + +def test_subdir(tmp_dir, scm, dvc): + subdir = tmp_dir / "dir" + subdir.gen("copy.py", COPY_SCRIPT) + subdir.gen("params.yaml", "foo: 1") + + with subdir.chdir(): + dvc.run( + cmd="python copy.py params.yaml metrics.yaml", + metrics_no_cache=["metrics.yaml"], + params=["foo"], + name="copy-file", + no_exec=True, + ) + scm.add( + [subdir / "dvc.yaml", subdir / "copy.py", subdir / "params.yaml"] + ) + scm.commit("init") + + results = dvc.experiments.run(PIPELINE_FILE, params=["foo=2"]) + assert results + + exp = first(results) + ref_info = first(exp_refs_by_rev(scm, exp)) + + tree = scm.get_tree(exp) + for fname in ["metrics.yaml", "dvc.lock"]: + assert tree.exists(subdir / fname) + with tree.open(subdir / "metrics.yaml") as fobj: + assert fobj.read().strip() == "foo: 2" + + assert dvc.experiments.get_exact_name(exp) == ref_info.name + assert scm.resolve_rev(ref_info.name) == exp + + +def test_subrepo(tmp_dir, scm): + from tests.unit.tree.test_repo import make_subrepo + + subrepo = tmp_dir / "dir" / "repo" + make_subrepo(subrepo, scm) + + subrepo.gen("copy.py", COPY_SCRIPT) + subrepo.gen("params.yaml", "foo: 1") + + with subrepo.chdir(): + subrepo.dvc.run( + cmd="python copy.py params.yaml metrics.yaml", + metrics_no_cache=["metrics.yaml"], + params=["foo"], + name="copy-file", + no_exec=True, + ) + scm.add( + [ + subrepo / "dvc.yaml", + subrepo / "copy.py", + subrepo / "params.yaml", + ] + ) + scm.commit("init") + + results = subrepo.dvc.experiments.run(PIPELINE_FILE, params=["foo=2"]) + assert results + + exp = first(results) + ref_info = first(exp_refs_by_rev(scm, exp)) + + tree = scm.get_tree(exp) + for fname in ["metrics.yaml", "dvc.lock"]: + assert tree.exists(subrepo / fname) + with tree.open(subrepo / "metrics.yaml") as fobj: + assert fobj.read().strip() == "foo: 2" + + assert subrepo.dvc.experiments.get_exact_name(exp) == ref_info.name + assert scm.resolve_rev(ref_info.name) == exp