dvc: move checkout out of cache (#5415)

Checkout is an operation that builds object from cache(aka object
database) in a particular tree(aka filesystem). Thus it shouldn't be
part of the object database itself nor of the filesystem.

Pre-requisite for #5337
This commit is contained in:
Ruslan Kuprieiev
2021-02-05 03:27:00 +02:00
committed by GitHub
parent 8676617ae0
commit 320715a5f8
7 changed files with 319 additions and 308 deletions
+5 -296
View File
@@ -11,14 +11,8 @@ from typing import Optional
from funcy import decorator
from shortuuid import uuid
import dvc.prompt as prompt
from dvc.dir_info import DirInfo
from dvc.exceptions import (
CacheLinkError,
CheckoutError,
ConfirmRemoveError,
DvcException,
)
from dvc.exceptions import CacheLinkError, DvcException
from dvc.progress import Tqdm
from dvc.remote.slow_link_detection import ( # type: ignore[attr-defined]
slow_link_guard,
@@ -93,77 +87,6 @@ class CloudCache:
return DirInfo.from_list(d)
def _filter_hash_info(self, hash_info, path_info, filter_info):
if not filter_info or path_info == filter_info:
return hash_info
dir_info = self.get_dir_cache(hash_info)
hash_key = filter_info.relative_to(path_info).parts
# Check whether it is a file that exists on the trie
hi = dir_info.trie.get(hash_key)
if hi:
return hi
depth = len(hash_key)
filtered_dir_info = DirInfo()
try:
for key, value in dir_info.trie.items(hash_key):
filtered_dir_info.trie[key[depth:]] = value
except KeyError:
return None
return self._get_dir_info_hash(filtered_dir_info)[0]
def changed(self, path_info, tree, hash_info, filter_info=None):
"""Checks if data has changed.
A file is considered changed if:
- It doesn't exist on the working directory (was unlinked)
- Hash value is not computed (saving a new file)
- The hash value stored is different from the given one
- There's no file in the cache
Args:
path_info: dict with path information.
hash: expected hash value for this data.
filter_info: an optional argument to target a specific path.
Returns:
bool: True if data has changed, False otherwise.
"""
path = filter_info or path_info
logger.trace(
"checking if '%s'('%s') has changed.", path_info, hash_info
)
if not tree.exists(path):
logger.debug("'%s' doesn't exist.", path)
return True
hi = self._filter_hash_info(hash_info, path_info, filter_info)
if not hi:
logger.debug("hash value for '%s' is missing.", path)
return True
if self.changed_cache(hi):
logger.debug("cache for '%s'('%s') has changed.", path, hi)
return True
actual = tree.get_hash(path)
if hi != actual:
logger.debug(
"hash value '%s' for '%s' has changed (actual '%s').",
hi,
actual,
path,
)
return True
logger.trace("'%s' hasn't changed.", path)
return False
def link(self, from_info, to_info):
self._link(from_info, to_info, self.cache_types)
@@ -301,7 +224,7 @@ class CloudCache:
(
hash_info,
to_info,
) = local_cache._get_dir_info_hash( # pylint: disable=protected-access
) = local_cache.get_dir_info_hash( # pylint: disable=protected-access
dir_info
)
@@ -320,7 +243,7 @@ class CloudCache:
)
return self._transfer_file(from_tree, from_info)
def _cache_is_copy(self, path_info):
def cache_is_copy(self, path_info):
"""Checks whether cache uses copies."""
if self.cache_type_confirmed:
return self.cache_types[0] == "copy"
@@ -342,7 +265,7 @@ class CloudCache:
self.cache_type_confirmed = True
return self.cache_types[0] == "copy"
def _get_dir_info_hash(self, dir_info):
def get_dir_info_hash(self, dir_info):
import tempfile
from dvc.path_info import PathInfo
@@ -372,7 +295,7 @@ class CloudCache:
):
return hash_info
hi, tmp_info = self._get_dir_info_hash(dir_info)
hi, tmp_info = self.get_dir_info_hash(dir_info)
new_info = self.tree.hash_to_path_info(hi.value)
if self.changed_cache_file(hi):
self.makedirs(new_info.parent)
@@ -515,220 +438,6 @@ class CloudCache:
)
return self.changed_cache_file(hash_info)
def already_cached(self, path_info, tree):
current = tree.get_hash(path_info)
if not current:
return False
return not self.changed_cache(current)
def safe_remove(self, path_info, tree, force=False):
if not tree.exists(path_info):
return
if not force and not self.already_cached(path_info, tree):
msg = (
"file '{}' is going to be removed."
" Are you sure you want to proceed?".format(str(path_info))
)
if not prompt.confirm(msg):
raise ConfirmRemoveError(str(path_info))
tree.remove(path_info)
def _checkout_file(
self,
path_info,
tree,
hash_info,
force,
progress_callback=None,
relink=False,
):
"""The file is changed we need to checkout a new copy"""
cache_info = self.tree.hash_to_path_info(hash_info.value)
if tree.exists(path_info):
added = False
if not relink and self.changed(path_info, tree, hash_info):
modified = True
self.safe_remove(path_info, tree, force=force)
self.link(cache_info, path_info)
else:
modified = False
if tree.iscopy(path_info) and self._cache_is_copy(path_info):
self.unprotect(path_info)
else:
self.safe_remove(path_info, tree, force=force)
self.link(cache_info, path_info)
else:
self.link(cache_info, path_info)
added, modified = True, False
tree.state.save(path_info, hash_info)
if progress_callback:
progress_callback(str(path_info))
return added, modified
def _checkout_dir(
self,
path_info,
tree,
hash_info,
force,
progress_callback=None,
relink=False,
filter_info=None,
):
added, modified = False, False
# Create dir separately so that dir is created
# even if there are no files in it
if not tree.exists(path_info):
added = True
self.makedirs(path_info)
dir_info = self.get_dir_cache(hash_info)
logger.debug("Linking directory '%s'.", path_info)
for entry_info, entry_hash_info in dir_info.items(path_info):
if filter_info and not entry_info.isin_or_eq(filter_info):
continue
entry_added, entry_modified = self._checkout_file(
entry_info,
tree,
entry_hash_info,
force,
progress_callback,
relink,
)
if entry_added or entry_modified:
modified = True
modified = (
self._remove_redundant_files(path_info, tree, dir_info, force)
or modified
)
tree.state.save(path_info, hash_info)
# relink is not modified, assume it as nochange
return added, not added and modified and not relink
def _remove_redundant_files(self, path_info, tree, dir_info, force):
existing_files = set(tree.walk_files(path_info))
needed_files = {info for info, _ in dir_info.items(path_info)}
redundant_files = existing_files - needed_files
for path in redundant_files:
self.safe_remove(path, tree, force)
return bool(redundant_files)
@use_state
def checkout(
self,
path_info,
tree,
hash_info,
force=False,
progress_callback=None,
relink=False,
filter_info=None,
quiet=False,
):
if path_info.scheme not in ["local", self.tree.scheme]:
raise NotImplementedError
failed = None
skip = False
if not hash_info:
if not quiet:
logger.warning(
"No file hash info found for '%s'. It won't be created.",
path_info,
)
self.safe_remove(path_info, tree, force=force)
failed = path_info
elif not relink and not self.changed(
path_info, tree, hash_info, filter_info=filter_info
):
logger.trace("Data '%s' didn't change.", path_info)
skip = True
elif self.changed_cache(
hash_info, path_info=path_info, filter_info=filter_info
):
if not quiet:
logger.warning(
"Cache '%s' not found. File '%s' won't be created.",
hash_info,
path_info,
)
self.safe_remove(path_info, tree, force=force)
failed = path_info
if failed or skip:
if progress_callback:
progress_callback(
str(path_info),
self.get_files_number(
self.tree.path_info, hash_info, filter_info
),
)
if failed:
raise CheckoutError([failed])
return
logger.debug(
"Checking out '%s' with cache '%s'.", path_info, hash_info
)
return self._checkout(
path_info,
tree,
hash_info,
force,
progress_callback,
relink,
filter_info,
)
def _checkout(
self,
path_info,
tree,
hash_info,
force=False,
progress_callback=None,
relink=False,
filter_info=None,
):
if not hash_info.isdir:
ret = self._checkout_file(
path_info, tree, hash_info, force, progress_callback, relink
)
else:
ret = self._checkout_dir(
path_info,
tree,
hash_info,
force,
progress_callback,
relink,
filter_info,
)
tree.state.save_link(path_info)
return ret
def get_files_number(self, path_info, hash_info, filter_info):
from funcy.py3 import ilen
-5
View File
@@ -81,11 +81,6 @@ class LocalCache(CloudCache):
)
]
def already_cached(self, path_info, tree):
assert path_info.scheme in ["", "local"]
return super().already_cached(path_info, tree)
def _verify_link(self, path_info, link_type):
if link_type == "hardlink" and self.tree.getsize(path_info) == 0:
return
+297
View File
@@ -0,0 +1,297 @@
import logging
import dvc.prompt as prompt
from dvc.dir_info import DirInfo
from dvc.exceptions import CheckoutError, ConfirmRemoveError
logger = logging.getLogger(__name__)
def _filter_hash_info(cache, hash_info, path_info, filter_info):
if not filter_info or path_info == filter_info:
return hash_info
dir_info = cache.get_dir_cache(hash_info)
hash_key = filter_info.relative_to(path_info).parts
# Check whether it is a file that exists on the trie
hi = dir_info.trie.get(hash_key)
if hi:
return hi
depth = len(hash_key)
filtered_dir_info = DirInfo()
try:
for key, value in dir_info.trie.items(hash_key):
filtered_dir_info.trie[key[depth:]] = value
except KeyError:
return None
return cache.get_dir_info_hash(filtered_dir_info)[0]
def _changed(path_info, tree, hash_info, cache, filter_info=None):
"""Checks if data has changed.
A file is considered changed if:
- It doesn't exist on the working directory (was unlinked)
- Hash value is not computed (saving a new file)
- The hash value stored is different from the given one
- There's no file in the cache
Args:
path_info: dict with path information.
hash: expected hash value for this data.
filter_info: an optional argument to target a specific path.
Returns:
bool: True if data has changed, False otherwise.
"""
path = filter_info or path_info
logger.trace("checking if '%s'('%s') has changed.", path_info, hash_info)
if not tree.exists(path):
logger.debug("'%s' doesn't exist.", path)
return True
hi = _filter_hash_info(cache, hash_info, path_info, filter_info)
if not hi:
logger.debug("hash value for '%s' is missing.", path)
return True
if cache.changed_cache(hi):
logger.debug("cache for '%s'('%s') has changed.", path, hi)
return True
actual = tree.get_hash(path)
if hi != actual:
logger.debug(
"hash value '%s' for '%s' has changed (actual '%s').",
hi,
actual,
path,
)
return True
logger.trace("'%s' hasn't changed.", path)
return False
def _is_cached(cache, path_info, tree):
current = tree.get_hash(path_info)
if not current:
return False
return not cache.changed_cache(current)
def _remove(path_info, tree, cache, force=False):
if not tree.exists(path_info):
return
if not force and not _is_cached(cache, path_info, tree):
msg = (
"file '{}' is going to be removed."
" Are you sure you want to proceed?".format(str(path_info))
)
if not prompt.confirm(msg):
raise ConfirmRemoveError(str(path_info))
tree.remove(path_info)
def _checkout_file(
path_info,
tree,
hash_info,
cache,
force,
progress_callback=None,
relink=False,
):
"""The file is changed we need to checkout a new copy"""
cache_info = cache.tree.hash_to_path_info(hash_info.value)
if tree.exists(path_info):
added = False
if not relink and _changed(path_info, tree, hash_info, cache):
modified = True
_remove(path_info, tree, cache, force=force)
cache.link(cache_info, path_info)
else:
modified = False
if tree.iscopy(path_info) and cache.cache_is_copy(path_info):
cache.unprotect(path_info)
else:
_remove(path_info, tree, cache, force=force)
cache.link(cache_info, path_info)
else:
cache.link(cache_info, path_info)
added, modified = True, False
tree.state.save(path_info, hash_info)
if progress_callback:
progress_callback(str(path_info))
return added, modified
def _remove_redundant_files(path_info, tree, dir_info, cache, force):
existing_files = set(tree.walk_files(path_info))
needed_files = {info for info, _ in dir_info.items(path_info)}
redundant_files = existing_files - needed_files
for path in redundant_files:
_remove(path, tree, cache, force)
return bool(redundant_files)
def _checkout_dir(
path_info,
tree,
hash_info,
cache,
force,
progress_callback=None,
relink=False,
filter_info=None,
):
added, modified = False, False
# Create dir separately so that dir is created
# even if there are no files in it
if not tree.exists(path_info):
added = True
tree.makedirs(path_info)
dir_info = cache.get_dir_cache(hash_info)
logger.debug("Linking directory '%s'.", path_info)
for entry_info, entry_hash_info in dir_info.items(path_info):
if filter_info and not entry_info.isin_or_eq(filter_info):
continue
entry_added, entry_modified = _checkout_file(
entry_info,
tree,
entry_hash_info,
cache,
force,
progress_callback,
relink,
)
if entry_added or entry_modified:
modified = True
modified = (
_remove_redundant_files(path_info, tree, dir_info, cache, force)
or modified
)
tree.state.save(path_info, hash_info)
# relink is not modified, assume it as nochange
return added, not added and modified and not relink
def _checkout(
path_info,
tree,
hash_info,
cache,
force=False,
progress_callback=None,
relink=False,
filter_info=None,
):
if not hash_info.isdir:
ret = _checkout_file(
path_info, tree, hash_info, cache, force, progress_callback, relink
)
else:
ret = _checkout_dir(
path_info,
tree,
hash_info,
cache,
force,
progress_callback,
relink,
filter_info,
)
tree.state.save_link(path_info)
return ret
def checkout(
path_info,
tree,
hash_info,
cache,
force=False,
progress_callback=None,
relink=False,
filter_info=None,
quiet=False,
):
if path_info.scheme not in ["local", cache.tree.scheme]:
raise NotImplementedError
failed = None
skip = False
if not hash_info:
if not quiet:
logger.warning(
"No file hash info found for '%s'. It won't be created.",
path_info,
)
_remove(path_info, tree, cache, force=force)
failed = path_info
elif not relink and not _changed(
path_info, tree, hash_info, cache, filter_info=filter_info
):
logger.trace("Data '%s' didn't change.", path_info)
skip = True
elif cache.changed_cache(
hash_info, path_info=path_info, filter_info=filter_info
):
if not quiet:
logger.warning(
"Cache '%s' not found. File '%s' won't be created.",
hash_info,
path_info,
)
_remove(path_info, tree, cache, force=force)
failed = path_info
if failed or skip:
if progress_callback:
progress_callback(
str(path_info),
cache.get_files_number(path_info, hash_info, filter_info),
)
if failed:
raise CheckoutError([failed])
return
logger.debug("Checking out '%s' with cache '%s'.", path_info, hash_info)
return _checkout(
path_info,
tree,
hash_info,
cache,
force,
progress_callback,
relink,
filter_info,
)
+3 -1
View File
@@ -70,6 +70,8 @@ class RepoDependency(LocalDependency):
return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo}
def download(self, to, jobs=None):
from dvc.checkout import checkout
cache = self.repo.cache.local
with self._make_repo(cache_dir=cache.cache_dir) as repo:
@@ -84,7 +86,7 @@ class RepoDependency(LocalDependency):
follow_subrepos=False,
)
cache.checkout(to.path_info, to.tree, hash_info)
checkout(to.path_info, to.tree, hash_info, cache)
def update(self, rev=None):
if rev:
+6 -2
View File
@@ -8,6 +8,7 @@ from voluptuous import Any
import dvc.prompt as prompt
from dvc.cache import NamedCache
from dvc.checkout import checkout
from dvc.exceptions import (
CheckoutError,
CollectCacheError,
@@ -315,10 +316,12 @@ class BaseOutput:
self.hash_info,
filter_info=filter_info,
)
self.cache.checkout(
checkout(
self.path_info,
self.tree,
self.hash_info,
self.cache,
relink=True,
filter_info=filter_info,
)
@@ -387,10 +390,11 @@ class BaseOutput:
return None
try:
res = self.cache.checkout(
res = checkout(
self.path_info,
self.tree,
self.hash_info,
self.cache,
force=force,
progress_callback=progress_callback,
relink=relink,
+3 -1
View File
@@ -150,6 +150,8 @@ def test_memory_runs_of_multiple_stages(tmp_dir, dvc, run_copy, mocker):
def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote):
from dvc.output import base
tmp_dir.gen("foo", "foo")
stage = run_copy("foo", "bar", name="copy-foo-bar")
@@ -157,7 +159,7 @@ def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote):
mock_restore = mocker.spy(dvc.stage_cache, "restore")
mock_run = mocker.patch("dvc.stage.run.cmd_run")
mock_checkout = mocker.spy(dvc.cache.local, "checkout")
mock_checkout = mocker.spy(base, "checkout")
# removing any information that `dvc` could use to re-generate from
(tmp_dir / "bar").unlink()
+5 -3
View File
@@ -2,6 +2,8 @@ import os
import pytest
from dvc.output import base
def test_stage_cache(tmp_dir, dvc, mocker):
tmp_dir.gen("dep", "dep")
@@ -42,7 +44,7 @@ def test_stage_cache(tmp_dir, dvc, mocker):
assert os.path.isfile(cache_file)
run_spy = mocker.patch("dvc.stage.run.cmd_run")
checkout_spy = mocker.spy(dvc.cache.local, "checkout")
checkout_spy = mocker.spy(base, "checkout")
with dvc.lock, dvc.state:
stage.run()
@@ -95,7 +97,7 @@ def test_stage_cache_params(tmp_dir, dvc, mocker):
assert os.path.isfile(cache_file)
run_spy = mocker.patch("dvc.stage.run.cmd_run")
checkout_spy = mocker.spy(dvc.cache.local, "checkout")
checkout_spy = mocker.spy(base, "checkout")
with dvc.lock, dvc.state:
stage.run()
@@ -149,7 +151,7 @@ def test_stage_cache_wdir(tmp_dir, dvc, mocker):
assert os.path.isfile(cache_file)
run_spy = mocker.patch("dvc.stage.run.cmd_run")
checkout_spy = mocker.spy(dvc.cache.local, "checkout")
checkout_spy = mocker.spy(base, "checkout")
with dvc.lock, dvc.state:
stage.run()