Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-09 23:10:24 +01:00
Code Issues Packages Projects Releases Wiki Activity

dvc: load state when taking repo lock in @locked

Browse Source 
We are always loading state when taking a repo lock, so doing it right
in @locked decorator simplifies the code for Repo methods.
This commit is contained in:
Ruslan Kuprieiev
2019-10-10 01:08:28 +03:00
parent f254004fae
commit cba8ac7a8f
15 changed files with 172 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/repo/__init__.py
+1 -1
View File
@@ -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()
dvc/repo/add.py
+7 -8
View File
@@ -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
dvc/repo/checkout.py
+19 -24
View File
@@ -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)
dvc/repo/commit.py
+4 -5
View File
@@ -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()
dvc/repo/fetch.py
+33 -37
View File
@@ -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
dvc/repo/gc.py
+40 -36
View File
@@ -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)
dvc/repo/imp_url.py
+4 -6
View File
@@ -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
dvc/repo/move.py
+2 -3
View File
@@ -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()
dvc/repo/push.py
+11 -12
View File
@@ -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)
dvc/repo/reproduce.py
+3 -4
View File
@@ -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
dvc/repo/run.py
+3 -5
View File
@@ -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()
dvc/repo/status.py
+11 -12
View File
@@ -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)
dvc/repo/update.py
+1 -2
View File
@@ -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()
dvc/utils/compat.py
+2
View File
@@ -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
tests/func/test_tag.py
+31 -28
View File
@@ -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):