From cba8ac7a8fb1e0637629ae2e4153c5a3eeefa0b4 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 7 Oct 2019 11:55:44 +0300 Subject: [PATCH] dvc: load state when taking repo lock in @locked We are always loading state when taking a repo lock, so doing it right in @locked decorator simplifies the code for Repo methods. --- dvc/repo/__init__.py | 2 +- dvc/repo/add.py | 15 ++++----- dvc/repo/checkout.py | 43 +++++++++++------------- dvc/repo/commit.py | 9 +++-- dvc/repo/fetch.py | 70 ++++++++++++++++++-------------------- dvc/repo/gc.py | 76 ++++++++++++++++++++++-------------------- dvc/repo/imp_url.py | 10 +++--- dvc/repo/move.py | 5 ++- dvc/repo/push.py | 23 ++++++------- dvc/repo/reproduce.py | 7 ++-- dvc/repo/run.py | 8 ++--- dvc/repo/status.py | 23 ++++++------- dvc/repo/update.py | 3 +- dvc/utils/compat.py | 2 ++ tests/func/test_tag.py | 59 ++++++++++++++++---------------- 15 files changed, 172 insertions(+), 183 deletions(-) diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 060f32cff..1b9bdd500 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -29,7 +29,7 @@ logger = logging.getLogger(__name__) def locked(f): @wraps(f) def wrapper(repo, *args, **kwargs): - with repo.lock: + with repo.lock, repo.state: ret = f(repo, *args, **kwargs) # Our graph cache is no longer valid after we release the repo.lock repo._reset() diff --git a/dvc/repo/add.py b/dvc/repo/add.py index a6485b905..ce2b4d1d0 100644 --- a/dvc/repo/add.py +++ b/dvc/repo/add.py @@ -36,18 +36,17 @@ def add(repo, target, recursive=False, no_commit=False, fname=None): ) ) - with repo.state: - stages = _create_stages(repo, targets, fname) + stages = _create_stages(repo, targets, fname) - repo.check_modified_graph(stages) + repo.check_modified_graph(stages) - for stage in stages: - stage.save() + for stage in stages: + stage.save() - if not no_commit: - stage.commit() + if not no_commit: + stage.commit() - stage.dump() + stage.dump() return stages diff --git a/dvc/repo/checkout.py b/dvc/repo/checkout.py index 4647a21ed..36d8c977d 100644 --- a/dvc/repo/checkout.py +++ b/dvc/repo/checkout.py @@ -35,28 +35,23 @@ def _checkout( raise raise CheckoutErrorSuggestGit(target, exc) - with self.state: - _cleanup_unused_links(self, self.stages) - total = get_all_files_numbers(stages) - if total == 0: - logger.info("Nothing to do") - failed = [] - with Tqdm( - total=total, unit="file", desc="Checkout", disable=total == 0 - ) as pbar: - for stage in stages: - if stage.locked: - logger.warning( - "DVC-file '{path}' is locked. Its dependencies are" - " not going to be checked out.".format( - path=stage.relpath - ) - ) - - failed.extend( - stage.checkout( - force=force, progress_callback=pbar.update_desc - ) + _cleanup_unused_links(self, self.stages) + total = get_all_files_numbers(stages) + if total == 0: + logger.info("Nothing to do") + failed = [] + with Tqdm( + total=total, unit="file", desc="Checkout", disable=total == 0 + ) as pbar: + for stage in stages: + if stage.locked: + logger.warning( + "DVC-file '{path}' is locked. Its dependencies are" + " not going to be checked out.".format(path=stage.relpath) ) - if failed: - raise CheckoutError(failed) + + failed.extend( + stage.checkout(force=force, progress_callback=pbar.update_desc) + ) + if failed: + raise CheckoutError(failed) diff --git a/dvc/repo/commit.py b/dvc/repo/commit.py index 4f48e1dab..ceba50e59 100644 --- a/dvc/repo/commit.py +++ b/dvc/repo/commit.py @@ -4,8 +4,7 @@ from . import locked @locked def commit(self, target, with_deps=False, recursive=False, force=False): stages = self.collect(target, with_deps=with_deps, recursive=recursive) - with self.state: - for stage in stages: - stage.check_can_commit(force=force) - stage.commit() - stage.dump() + for stage in stages: + stage.check_can_commit(force=force) + stage.commit() + stage.dump() diff --git a/dvc/repo/fetch.py b/dvc/repo/fetch.py index eea05db1b..b12ac23c4 100644 --- a/dvc/repo/fetch.py +++ b/dvc/repo/fetch.py @@ -33,48 +33,44 @@ def _fetch( config.NoRemoteError: thrown when downloading only local files and no remote is configured """ - with self.state: - used = self.used_cache( - targets, - all_branches=all_branches, - all_tags=all_tags, - with_deps=with_deps, - force=True, - remote=remote, - jobs=jobs, - recursive=recursive, + used = self.used_cache( + targets, + all_branches=all_branches, + all_tags=all_tags, + with_deps=with_deps, + force=True, + remote=remote, + jobs=jobs, + recursive=recursive, + ) + + downloaded = 0 + failed = 0 + + try: + downloaded += self.cloud.pull( + used["local"], jobs, remote=remote, show_checksums=show_checksums ) + except NoRemoteError: + if not used["repo"] and used["local"]: + raise - downloaded = 0 - failed = 0 + except DownloadError as exc: + failed += exc.amount + for dep in used["repo"]: try: - downloaded += self.cloud.pull( - used["local"], - jobs, - remote=remote, - show_checksums=show_checksums, - ) - except NoRemoteError: - if not used["repo"] and used["local"]: - raise - + out = dep.fetch() + downloaded += out.get_files_number() except DownloadError as exc: failed += exc.amount + except (CloneError, OutputNotFoundError): + failed += 1 + logger.exception( + "failed to fetch data for '{}'".format(dep.stage.outs[0]) + ) - for dep in used["repo"]: - try: - out = dep.fetch() - downloaded += out.get_files_number() - except DownloadError as exc: - failed += exc.amount - except (CloneError, OutputNotFoundError): - failed += 1 - logger.exception( - "failed to fetch data for '{}'".format(dep.stage.outs[0]) - ) + if failed: + raise DownloadError(failed) - if failed: - raise DownloadError(failed) - - return downloaded + return downloaded diff --git a/dvc/repo/gc.py b/dvc/repo/gc.py index b6719a6b0..ef15d65f9 100644 --- a/dvc/repo/gc.py +++ b/dvc/repo/gc.py @@ -33,18 +33,17 @@ def _load_all_used_cache( clists = [] for repo in repos: - with repo.state: - repo_clist = repo.used_cache( - targets=None, - all_branches=all_branches, - with_deps=with_deps, - all_tags=all_tags, - remote=remote, - force=force, - jobs=jobs, - ) + repo_clist = repo.used_cache( + targets=None, + all_branches=all_branches, + with_deps=with_deps, + all_tags=all_tags, + remote=remote, + force=force, + jobs=jobs, + ) - clists.append(repo_clist) + clists.append(repo_clist) return clists @@ -67,45 +66,50 @@ def gc( jobs=None, repos=None, ): + from dvc.utils.compat import ExitStack from dvc.repo import Repo - all_repos = [self] + all_repos = [] if repos: - all_repos.extend(Repo(path) for path in repos) + all_repos = [Repo(path) for path in repos] - all_clists = _load_all_used_cache( - all_repos, - all_branches=all_branches, - with_deps=with_deps, - all_tags=all_tags, - remote=remote, - force=force, - jobs=jobs, - ) + with ExitStack() as stack: + for repo in all_repos: + stack.enter_context(repo.lock) + stack.enter_context(repo.state) + + all_clists = _load_all_used_cache( + all_repos + [self], + all_branches=all_branches, + with_deps=with_deps, + all_tags=all_tags, + remote=remote, + force=force, + jobs=jobs, + ) if len(all_clists) > 1: clist = _merge_cache_lists(all_clists) else: clist = all_clists[0] - with self.state: - _do_gc("local", self.cache.local.gc, clist) + _do_gc("local", self.cache.local.gc, clist) - if self.cache.s3: - _do_gc("s3", self.cache.s3.gc, clist) + if self.cache.s3: + _do_gc("s3", self.cache.s3.gc, clist) - if self.cache.gs: - _do_gc("gs", self.cache.gs.gc, clist) + if self.cache.gs: + _do_gc("gs", self.cache.gs.gc, clist) - if self.cache.ssh: - _do_gc("ssh", self.cache.ssh.gc, clist) + if self.cache.ssh: + _do_gc("ssh", self.cache.ssh.gc, clist) - if self.cache.hdfs: - _do_gc("hdfs", self.cache.hdfs.gc, clist) + if self.cache.hdfs: + _do_gc("hdfs", self.cache.hdfs.gc, clist) - if self.cache.azure: - _do_gc("azure", self.cache.azure.gc, clist) + if self.cache.azure: + _do_gc("azure", self.cache.azure.gc, clist) - if cloud: - _do_gc("remote", self.cloud.get_remote(remote, "gc -c").gc, clist) + if cloud: + _do_gc("remote", self.cloud.get_remote(remote, "gc -c").gc, clist) diff --git a/dvc/repo/imp_url.py b/dvc/repo/imp_url.py index 55cc0cef6..4ce67104c 100644 --- a/dvc/repo/imp_url.py +++ b/dvc/repo/imp_url.py @@ -11,18 +11,16 @@ def imp_url(self, url, out=None, fname=None, erepo=None, locked=True): out = out or pathlib.PurePath(url).name - with self.state: - stage = Stage.create( - self, cmd=None, deps=[url], outs=[out], fname=fname, erepo=erepo - ) + stage = Stage.create( + self, cmd=None, deps=[url], outs=[out], fname=fname, erepo=erepo + ) if stage is None: return None self.check_modified_graph([stage]) - with self.state: - stage.run() + stage.run() stage.locked = locked diff --git a/dvc/repo/move.py b/dvc/repo/move.py index 58c9777f4..6fa86eaa6 100644 --- a/dvc/repo/move.py +++ b/dvc/repo/move.py @@ -66,8 +66,7 @@ def move(self, from_path, to_path): to_out = Output.loads_from(stage, [to_path], out.use_cache, out.metric)[0] - with self.state: - out.move(to_out) - stage.save() + out.move(to_out) + stage.save() stage.dump() diff --git a/dvc/repo/push.py b/dvc/repo/push.py index a30739964..ec5eb0afb 100644 --- a/dvc/repo/push.py +++ b/dvc/repo/push.py @@ -14,15 +14,14 @@ def push( all_tags=False, recursive=False, ): - with self.state: - used = self.used_cache( - targets, - all_branches=all_branches, - all_tags=all_tags, - with_deps=with_deps, - force=True, - remote=remote, - jobs=jobs, - recursive=recursive, - )["local"] - return self.cloud.push(used, jobs, remote=remote) + used = self.used_cache( + targets, + all_branches=all_branches, + all_tags=all_tags, + with_deps=with_deps, + force=True, + remote=remote, + jobs=jobs, + recursive=recursive, + )["local"] + return self.cloud.push(used, jobs, remote=remote) diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index f65369a5f..dd26d6df9 100644 --- a/dvc/repo/reproduce.py +++ b/dvc/repo/reproduce.py @@ -94,10 +94,9 @@ def reproduce( targets = self.collect(target, recursive=recursive, graph=active_graph) ret = [] - with self.state: - for target in targets: - stages = _reproduce(self, active_graph, target, **kwargs) - ret.extend(stages) + for target in targets: + stages = _reproduce(self, active_graph, target, **kwargs) + ret.extend(stages) return ret diff --git a/dvc/repo/run.py b/dvc/repo/run.py index 65c4933dd..2c590716f 100644 --- a/dvc/repo/run.py +++ b/dvc/repo/run.py @@ -9,17 +9,15 @@ from .scm_context import scm_context def run(self, no_exec=False, **kwargs): from dvc.stage import Stage - with self.state: - stage = Stage.create(self, **kwargs) + stage = Stage.create(self, **kwargs) if stage is None: return None self.check_modified_graph([stage]) - with self.state: - if not no_exec: - stage.run(no_commit=kwargs.get("no_commit", False)) + if not no_exec: + stage.run(no_commit=kwargs.get("no_commit", False)) stage.dump() diff --git a/dvc/repo/status.py b/dvc/repo/status.py index c5da186d0..422504404 100644 --- a/dvc/repo/status.py +++ b/dvc/repo/status.py @@ -104,15 +104,14 @@ def status( with_deps=False, all_tags=False, ): - with self.state: - if cloud or remote: - return _cloud_status( - self, - targets, - jobs, - all_branches=all_branches, - with_deps=with_deps, - remote=remote, - all_tags=all_tags, - ) - return _local_status(self, targets, with_deps=with_deps) + if cloud or remote: + return _cloud_status( + self, + targets, + jobs, + all_branches=all_branches, + with_deps=with_deps, + remote=remote, + all_tags=all_tags, + ) + return _local_status(self, targets, with_deps=with_deps) diff --git a/dvc/repo/update.py b/dvc/repo/update.py index 05c2f27cf..9533bb6d3 100644 --- a/dvc/repo/update.py +++ b/dvc/repo/update.py @@ -6,7 +6,6 @@ def update(self, target): from dvc.stage import Stage stage = Stage.load(self, target) - with self.state: - stage.update() + stage.update() stage.dump() diff --git a/dvc/utils/compat.py b/dvc/utils/compat.py index 911664f07..6ab98ea2a 100644 --- a/dvc/utils/compat.py +++ b/dvc/utils/compat.py @@ -112,6 +112,7 @@ if is_py2: from io import open # noqa: F401 import pathlib2 as pathlib # noqa: F401 from collections import Mapping # noqa: F401 + from contextlib2 import ExitStack # noqa: F401 builtin_str = str # noqa: F821 bytes = str # noqa: F821 @@ -155,6 +156,7 @@ elif is_py3: from io import StringIO, BytesIO # noqa: F401 import configparser as ConfigParser # noqa: F401 from collections.abc import Mapping # noqa: F401 + from contextlib import ExitStack # noqa: F401 builtin_str = str # noqa: F821 str = str # noqa: F821 diff --git a/tests/func/test_tag.py b/tests/func/test_tag.py index 8230d82d6..cd675a8d4 100644 --- a/tests/func/test_tag.py +++ b/tests/func/test_tag.py @@ -47,47 +47,50 @@ class TestTag(TestDvc): class TestTagAll(TestDvc): def test(self): - ret = main(["add", self.FOO, self.BAR]) - self.assertEqual(ret, 0) + with self._caplog.at_level(logging.INFO, logger="dvc"): + ret = main(["add", self.FOO, self.BAR]) + self.assertEqual(ret, 0) - self._caplog.clear() + self._caplog.clear() - ret = main(["tag", "list"]) - self.assertEqual(ret, 0) + ret = main(["tag", "list"]) + self.assertEqual(ret, 0) - self.assertEqual("", self._caplog.text) + self.assertEqual("", self._caplog.text) - ret = main(["tag", "add", "v1"]) - self.assertEqual(ret, 0) + ret = main(["tag", "add", "v1"]) + self.assertEqual(ret, 0) - self._caplog.clear() + self._caplog.clear() - ret = main(["tag", "list"]) - self.assertEqual(ret, 0) + ret = main(["tag", "list"]) + self.assertEqual(ret, 0) - expected = { - "foo.dvc": { - "foo": {"v1": {"md5": "acbd18db4cc2f85cedef654fccc4a4d8"}} - }, - "bar.dvc": { - "bar": {"v1": {"md5": "8978c98bb5a48c2fb5f2c4c905768afa"}} - }, - } + expected = { + "foo.dvc": { + "foo": {"v1": {"md5": "acbd18db4cc2f85cedef654fccc4a4d8"}} + }, + "bar.dvc": { + "bar": {"v1": {"md5": "8978c98bb5a48c2fb5f2c4c905768afa"}} + }, + } - stdout = "\n".join(record.message for record in self._caplog.records) - received = from_yaml_string(stdout) + stdout = "\n".join( + record.message for record in self._caplog.records + ) + received = from_yaml_string(stdout) - assert expected == received + assert expected == received - ret = main(["tag", "remove", "v1"]) - self.assertEqual(ret, 0) + ret = main(["tag", "remove", "v1"]) + self.assertEqual(ret, 0) - self._caplog.clear() + self._caplog.clear() - ret = main(["tag", "list"]) - self.assertEqual(ret, 0) + ret = main(["tag", "list"]) + self.assertEqual(ret, 0) - self.assertEqual("", self._caplog.text) + self.assertEqual("", self._caplog.text) class TestTagAddNoChecksumInfo(TestDvc):