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

dvc: introduce File/Tree cache objects (#5424)

Browse Source
This commit is contained in:
Ruslan Kuprieiev
2021-02-09 04:00:06 +02:00
committed by GitHub
parent 02a0d1af52
commit 226c68f8ff
20 changed files with 587 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cache/base.py
Vendored
+46 -262
View File
@@ -1,8 +1,5 @@
import errno
import itertools
import json
import logging
import os
from concurrent import futures
from concurrent.futures import ThreadPoolExecutor
from copy import copy
@@ -18,18 +15,11 @@ from dvc.remote.slow_link_detection import ( # type: ignore[attr-defined]
slow_link_guard,
)
from ..tree.base import RemoteActionNotImplemented
from ..objects import File, HashFile, ObjectFormatError, Tree
logger = logging.getLogger(__name__)
class DirCacheError(DvcException):
def __init__(self, hash_):
super().__init__(
f"Failed to load dir cache for hash value: '{hash_}'."
)
@decorator
def use_state(call):
tree = call._args[0].tree # pylint: disable=protected-access
@@ -52,40 +42,6 @@ class CloudCache:
self.DEFAULT_CACHE_TYPES
)
self.cache_type_confirmed = False
self._dir_info = {}
def get_dir_cache(self, hash_info):
assert hash_info
dir_info = self._dir_info.get(hash_info.value)
if dir_info:
return dir_info
try:
dir_info = self.load_dir_cache(hash_info)
except DirCacheError:
dir_info = DirInfo()
self._dir_info[hash_info.value] = dir_info
return dir_info
def load_dir_cache(self, hash_info):
path_info = self.tree.hash_to_path_info(hash_info.value)
try:
with self.tree.open(path_info, "r") as fobj:
d = json.load(fobj)
except (ValueError, FileNotFoundError) as exc:
raise DirCacheError(hash_info) from exc
if not isinstance(d, list):
logger.error(
"dir cache file format error '%s' [skipping the file]",
path_info,
)
d = []
return DirInfo.from_list(d)
def link(self, from_info, to_info):
self._link(from_info, to_info, self.cache_types)
@@ -141,31 +97,39 @@ class CloudCache:
"Created '%s': %s -> %s", self.cache_types[0], from_info, to_info,
)
def add(self, path_info, tree, hash_info, **kwargs):
if not self.changed_cache_file(hash_info):
return
def get(self, hash_info):
""" get raw object """
return HashFile(
# Prefer string path over PathInfo when possible due to performance
self.hash_to_path(hash_info.value),
self.tree,
hash_info,
)
cache_info = self.tree.hash_to_path_info(hash_info.value)
@use_state
def add(self, obj, **kwargs):
try:
self.check(obj.hash_info)
return
except (ObjectFormatError, FileNotFoundError):
pass
cache_info = self.tree.hash_to_path_info(obj.hash_info.value)
# using our makedirs to create dirs with proper permissions
self.makedirs(cache_info.parent)
if isinstance(tree, type(self.tree)):
self.tree.move(path_info, cache_info)
if isinstance(obj.tree, type(self.tree)):
self.tree.move(obj.path_info, cache_info)
else:
with tree.open(path_info, mode="rb") as fobj:
with obj.tree.open(obj.path_info, mode="rb") as fobj:
self.tree.upload_fobj(fobj, cache_info)
tree.state.save(path_info, hash_info)
obj.tree.state.save(obj.path_info, obj.hash_info)
self.protect(cache_info)
self.tree.state.save(cache_info, hash_info)
self.tree.state.save(cache_info, obj.hash_info)
callback = kwargs.get("download_callback")
if callback:
callback(1)
def _save_file(self, path_info, tree, hash_info, **kwargs):
assert hash_info
self.add(path_info, tree, hash_info, **kwargs)
def _transfer_file(self, from_tree, from_info):
from dvc.utils import tmp_fname
from dvc.utils.stream import HashedStreamReader
@@ -225,19 +189,11 @@ class CloudCache:
) in self._transfer_directory_contents(
from_tree, from_info, jobs, pbar
):
self.add(entry_tmp_info, self.tree, entry_hash)
obj = File(entry_tmp_info, self.tree, entry_hash)
self.add(obj)
dir_info.trie[entry_info.parts] = entry_hash
local_cache = self.repo.cache.local
(
hash_info,
to_info,
) = local_cache.get_dir_info_hash( # pylint: disable=protected-access
dir_info
)
self.add(to_info, self.repo.cache.local.tree, hash_info)
return hash_info
return Tree.save_dir_info(self, dir_info)
def transfer(self, from_tree, from_info, jobs=None, no_progress_bar=False):
jobs = jobs or min((from_tree.jobs, self.tree.jobs))
@@ -250,7 +206,8 @@ class CloudCache:
no_progress_bar=no_progress_bar,
)
tmp_info, hash_info = self._transfer_file(from_tree, from_info)
self.add(tmp_info, self.tree, hash_info)
obj = File(tmp_info, self.tree, hash_info)
self.add(obj)
return hash_info
def cache_is_copy(self, path_info):
@@ -264,6 +221,7 @@ class CloudCache:
workspace_file = path_info.with_name("." + uuid())
test_cache_file = self.tree.path_info / ".cache_type_test_file"
if not self.tree.exists(test_cache_file):
self.makedirs(test_cache_file.parent)
with self.tree.open(test_cache_file, "wb") as fobj:
fobj.write(bytes(1))
try:
@@ -275,89 +233,9 @@ class CloudCache:
self.cache_type_confirmed = True
return self.cache_types[0] == "copy"
def get_dir_info_hash(self, dir_info):
import tempfile
from dvc.path_info import PathInfo
from dvc.utils import tmp_fname
tmp = tempfile.NamedTemporaryFile(delete=False).name
with open(tmp, "w+") as fobj:
json.dump(dir_info.to_list(), fobj, sort_keys=True)
from_info = PathInfo(tmp)
to_info = self.tree.path_info / tmp_fname("")
self.tree.upload(from_info, to_info, no_progress_bar=True)
hash_info = self.tree.get_file_hash(to_info)
hash_info.value += self.tree.CHECKSUM_DIR_SUFFIX
hash_info.dir_info = dir_info
hash_info.nfiles = dir_info.nfiles
return hash_info, to_info
@use_state
def save_dir_info(self, dir_info, hash_info=None):
if (
hash_info
and hash_info.name == self.tree.PARAM_CHECKSUM
and not self.changed_cache_file(hash_info)
):
return hash_info
hi, tmp_info = self.get_dir_info_hash(dir_info)
self.add(tmp_info, self.tree, hi)
return hi
def _save_dir(
self, path_info, tree, hash_info, filter_info=None, **kwargs,
):
if not hash_info.dir_info:
hash_info.dir_info = tree.cache.get_dir_cache(hash_info)
hi = self.save_dir_info(hash_info.dir_info, hash_info)
for entry_info, entry_hash in Tqdm(
hi.dir_info.items(path_info),
desc="Saving " + path_info.name,
unit="file",
):
if filter_info and not entry_info.isin_or_eq(filter_info):
continue
self._save_file(entry_info, tree, entry_hash, **kwargs)
cache_info = self.tree.hash_to_path_info(hi.value)
self.tree.state.save(cache_info, hi)
tree.state.save(path_info, hi)
@use_state
def save(self, path_info, tree, hash_info, **kwargs):
if path_info.scheme != self.tree.scheme:
raise RemoteActionNotImplemented(
f"save {path_info.scheme} -> {self.tree.scheme}",
self.tree.scheme,
)
if not hash_info:
kw = kwargs.copy()
kw.pop("download_callback", None)
hash_info = tree.get_hash(path_info, **kw)
if not hash_info:
raise FileNotFoundError(
errno.ENOENT, os.strerror(errno.ENOENT), path_info
)
self._save(path_info, tree, hash_info, **kwargs)
return hash_info
def _save(self, path_info, tree, hash_info, **kwargs):
to_info = self.tree.hash_to_path_info(hash_info.value)
logger.debug("Saving '%s' to '%s'.", path_info, to_info)
if tree.isdir(path_info):
self._save_dir(path_info, tree, hash_info, **kwargs)
else:
self._save_file(path_info, tree, hash_info, **kwargs)
def save(self, obj, **kwargs):
obj.save(self, **kwargs)
# Override to return path as a string instead of PathInfo for clouds
# which support string paths (see local)
@@ -376,7 +254,7 @@ class CloudCache:
def set_exec(self, path_info): # pylint: disable=unused-argument
pass
def changed_cache_file(self, hash_info):
def check(self, hash_info):
"""Compare the given hash with the (corresponding) actual one.
- Use `State` as a cache for computed hashes
@@ -388,119 +266,25 @@ class CloudCache:
- Remove the file from cache if it doesn't match the actual hash
"""
# Prefer string path over PathInfo when possible due to performance
cache_info = self.hash_to_path(hash_info.value)
if self.is_protected(cache_info):
obj = self.get(hash_info)
if self.is_protected(obj.path_info):
logger.trace(
"Assuming '%s' is unchanged since it is read-only", cache_info
"Assuming '%s' is unchanged since it is read-only",
obj.path_info,
)
return False
return
actual = self.tree.get_hash(cache_info)
logger.trace(
"cache '%s' expected '%s' actual '%s'",
cache_info,
hash_info,
actual,
)
if not hash_info or not actual:
return True
if actual.value.split(".")[0] == hash_info.value.split(".")[0]:
# making cache file read-only so we don't need to check it
# next time
self.protect(cache_info)
return False
if self.tree.exists(cache_info):
logger.warning("corrupted cache file '%s'.", cache_info)
self.tree.remove(cache_info)
return True
def _changed_dir_cache(self, hash_info, path_info=None, filter_info=None):
if self.changed_cache_file(hash_info):
return True
dir_info = self.get_dir_cache(hash_info)
for entry_info, entry_hash in dir_info.items(path_info):
if path_info and filter_info:
if not entry_info.isin_or_eq(filter_info):
continue
if self.changed_cache_file(entry_hash):
return True
return False
def changed_cache(self, hash_info, path_info=None, filter_info=None):
if hash_info.isdir:
return self._changed_dir_cache(
hash_info, path_info=path_info, filter_info=filter_info
)
return self.changed_cache_file(hash_info)
def get_files_number(self, path_info, hash_info, filter_info):
from funcy.py3 import ilen
if not hash_info:
return 0
if not hash_info.isdir:
return 1
if not filter_info:
return self.get_dir_cache(hash_info).nfiles
return ilen(
filter_info.isin_or_eq(path_info / relpath)
for relpath, _ in self.get_dir_cache(hash_info).items()
)
def _get_dir_size(self, dir_info):
try:
return sum(
self.tree.getsize(self.tree.hash_to_path_info(hi.value))
for _, hi in dir_info.items()
)
except FileNotFoundError:
return None
obj.check(self)
except ObjectFormatError:
logger.warning("corrupted cache file '%s'.", obj.path_info)
self.tree.remove(obj.path_info)
raise
def merge(self, ancestor_info, our_info, their_info):
assert our_info
assert their_info
if ancestor_info:
ancestor = self.get_dir_cache(ancestor_info)
else:
ancestor = DirInfo()
our = self.get_dir_cache(our_info)
their = self.get_dir_cache(their_info)
merged = our.merge(ancestor, their)
hash_info = self.save_dir_info(merged)
hash_info.size = self._get_dir_size(merged)
return hash_info
@use_state
def get_hash(self, tree, path_info):
hash_info = tree.get_hash(path_info)
if not hash_info.isdir:
assert hash_info.name == self.tree.PARAM_CHECKSUM
return hash_info
hi = self.save_dir_info(hash_info.dir_info, hash_info)
hi.size = hash_info.size
return hi
def set_dir_info(self, hash_info):
assert hash_info.isdir
hash_info.dir_info = self.get_dir_cache(hash_info)
hash_info.nfiles = hash_info.dir_info.nfiles
# making cache file read-only so we don't need to check it
# next time
self.protect(obj.path_info)
def _list_paths(self, prefix=None, progress_callback=None):
if prefix:
dvc/cache/local.py
Vendored
+16 -12
View File
@@ -6,6 +6,7 @@ from funcy import cached_property
from shortuuid import uuid
from dvc.hash_info import HashInfo
from dvc.objects import ObjectFormatError
from dvc.path_info import PathInfo
from dvc.progress import Tqdm
@@ -68,18 +69,21 @@ class LocalCache(CloudCache):
def hashes_exist(
self, hashes, jobs=None, name=None
): # pylint: disable=unused-argument
return [
hash_
for hash_ in Tqdm(
hashes,
unit="file",
desc="Querying "
+ ("cache in " + name if name else "local cache"),
)
if not self.changed_cache_file(
HashInfo(self.tree.PARAM_CHECKSUM, hash_)
)
]
ret = []
for hash_ in Tqdm(
hashes,
unit="file",
desc="Querying " + ("cache in " + name if name else "local cache"),
):
hash_info = HashInfo(self.tree.PARAM_CHECKSUM, hash_)
try:
self.check(hash_info)
ret.append(hash_)
except (FileNotFoundError, ObjectFormatError):
pass
return ret
def _verify_link(self, path_info, link_type):
if link_type == "hardlink" and self.tree.getsize(path_info) == 0:
dvc/checkout.py
+75 -156
View File
@@ -1,101 +1,58 @@
import logging
import dvc.prompt as prompt
from dvc.dir_info import DirInfo
from dvc.exceptions import CheckoutError, ConfirmRemoveError
from dvc.objects import ObjectFormatError, check, load
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)
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
def _changed(path_info, tree, obj, cache):
logger.trace("checking if '%s'('%s') has changed.", path_info, obj)
try:
actual = tree.get_hash(path)
except FileNotFoundError:
logger.debug("'%s' doesn't exist.", path)
return True
if hi != actual:
check(cache, obj)
except (FileNotFoundError, ObjectFormatError):
logger.debug(
"hash value '%s' for '%s' has changed (actual '%s').",
hi,
actual,
path,
"cache for '%s'('%s') has changed.", path_info, obj.hash_info
)
return True
logger.trace("'%s' hasn't changed.", path)
try:
actual = tree.get_hash(path_info)
except FileNotFoundError:
logger.debug("'%s' doesn't exist.", path_info)
return True
if obj.hash_info != actual:
logger.debug(
"hash value '%s' for '%s' has changed (actual '%s').",
obj.hash_info,
actual,
path_info,
)
return True
logger.trace("'%s' hasn't changed.", path_info)
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):
if force:
tree.remove(path_info)
return
current = tree.get_hash(path_info)
try:
obj = load(cache, current)
check(cache, obj)
except (FileNotFoundError, ObjectFormatError):
msg = (
"file '{}' is going to be removed."
" Are you sure you want to proceed?".format(str(path_info))
f"file/directory '{path_info}' is going to be removed. "
"Are you sure you want to proceed?"
)
if not prompt.confirm(msg):
@@ -105,26 +62,17 @@ def _remove(path_info, tree, cache, force=False):
def _checkout_file(
path_info,
tree,
hash_info,
cache,
force,
progress_callback=None,
relink=False,
path_info, tree, obj, 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)
modified = False
cache_info = cache.tree.hash_to_path_info(obj.hash_info.value)
if tree.exists(path_info):
added = False
if not relink and _changed(path_info, tree, hash_info, cache):
if not relink and _changed(path_info, tree, obj, 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:
@@ -132,13 +80,13 @@ def _checkout_file(
cache.link(cache_info, path_info)
else:
cache.link(cache_info, path_info)
added, modified = True, False
modified = True
tree.state.save(path_info, hash_info)
tree.state.save(path_info, obj.hash_info)
if progress_callback:
progress_callback(str(path_info))
return added, modified
return modified
def _remove_redundant_files(path_info, tree, dir_info, cache, force):
@@ -153,77 +101,59 @@ def _remove_redundant_files(path_info, tree, dir_info, cache, force):
def _checkout_dir(
path_info,
tree,
hash_info,
cache,
force,
progress_callback=None,
relink=False,
filter_info=None,
path_info, tree, obj, cache, force, progress_callback=None, relink=False,
):
added, modified = False, False
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
modified = 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(
for entry_info, entry_hash_info in obj.hash_info.dir_info.items(path_info):
entry_modified = _checkout_file(
entry_info,
tree,
entry_hash_info,
cache.get(entry_hash_info),
cache,
force,
progress_callback,
relink,
)
if entry_added or entry_modified:
if entry_modified:
modified = True
modified = (
_remove_redundant_files(path_info, tree, dir_info, cache, force)
_remove_redundant_files(
path_info, tree, obj.hash_info.dir_info, cache, force
)
or modified
)
tree.state.save(path_info, hash_info)
tree.state.save(path_info, obj.hash_info)
# relink is not modified, assume it as nochange
return added, not added and modified and not relink
return modified and not relink
def _checkout(
path_info,
tree,
hash_info,
obj,
cache,
force=False,
progress_callback=None,
relink=False,
filter_info=None,
):
if not hash_info.isdir:
if not obj.hash_info.isdir:
ret = _checkout_file(
path_info, tree, hash_info, cache, force, progress_callback, relink
path_info, tree, obj, cache, force, progress_callback, relink
)
else:
ret = _checkout_dir(
path_info,
tree,
hash_info,
cache,
force,
progress_callback,
relink,
filter_info,
path_info, tree, obj, cache, force, progress_callback, relink,
)
tree.state.save_link(path_info)
@@ -234,12 +164,11 @@ def _checkout(
def checkout(
path_info,
tree,
hash_info,
obj,
cache,
force=False,
progress_callback=None,
relink=False,
filter_info=None,
quiet=False,
):
if path_info.scheme not in ["local", cache.tree.scheme]:
@@ -247,7 +176,7 @@ def checkout(
failed = None
skip = False
if not hash_info:
if not obj:
if not quiet:
logger.warning(
"No file hash info found for '%s'. It won't be created.",
@@ -256,43 +185,33 @@ def checkout(
_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
):
elif not relink and not _changed(path_info, tree, obj, cache):
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
else:
try:
check(cache, obj)
except (FileNotFoundError, ObjectFormatError):
if not quiet:
logger.warning(
"Cache '%s' not found. File '%s' won't be created.",
obj.hash_info,
path_info,
)
_remove(path_info, tree, cache, force=force)
failed = path_info
if failed or skip:
if progress_callback:
if progress_callback and obj:
progress_callback(
str(path_info),
cache.get_files_number(path_info, hash_info, filter_info),
str(path_info), len(obj),
)
if failed:
raise CheckoutError([failed])
return
logger.debug("Checking out '%s' with cache '%s'.", path_info, hash_info)
logger.debug("Checking out '%s' with cache '%s'.", path_info, obj)
return _checkout(
path_info,
tree,
hash_info,
cache,
force,
progress_callback,
relink,
filter_info,
path_info, tree, obj, cache, force, progress_callback, relink,
)
dvc/dependency/repo.py
+5 -3
View File
@@ -71,6 +71,7 @@ class RepoDependency(LocalDependency):
def download(self, to, jobs=None):
from dvc.checkout import checkout
from dvc.objects import stage
cache = self.repo.cache.local
@@ -78,15 +79,16 @@ class RepoDependency(LocalDependency):
if self.def_repo.get(self.PARAM_REV_LOCK) is None:
self.def_repo[self.PARAM_REV_LOCK] = repo.get_rev()
path_info = PathInfo(repo.root_dir) / self.def_path
hash_info = cache.save(
obj = stage(
cache,
path_info,
repo.repo_tree,
None,
jobs=jobs,
follow_subrepos=False,
)
cache.save(obj, jobs=jobs)
checkout(to.path_info, to.tree, hash_info, cache)
checkout(to.path_info, to.tree, obj, cache)
def update(self, rev=None):
if rev:
dvc/objects.py
+254
View File
@@ -0,0 +1,254 @@
import json
import logging
from .dir_info import DirInfo
from .exceptions import DvcException
logger = logging.getLogger(__name__)
class ObjectError(DvcException):
pass
class ObjectFormatError(ObjectError):
pass
def _get_hash(path_info, tree, name, **kwargs):
if tree.PARAM_CHECKSUM == name:
return tree.get_hash(path_info, **kwargs)
if name == "md5":
from dvc.hash_info import HashInfo
from dvc.utils import file_md5
return HashInfo("md5", file_md5(path_info, tree, **kwargs))
raise AssertionError(f"unsupported hash '{name}'")
class HashFile:
def __init__(self, path_info, tree, hash_info):
self.path_info = path_info
self.tree = tree
self.hash_info = hash_info
def __len__(self):
return 1
def __str__(self):
return f"object {self.hash_info}"
def __bool__(self):
return bool(self.hash_info)
def check(self, odb):
actual = _get_hash(self.path_info, self.tree, odb.tree.PARAM_CHECKSUM)
logger.trace(
"cache '%s' expected '%s' actual '%s'",
self.path_info,
self.hash_info,
actual,
)
assert actual.name == self.hash_info.name
if actual.value.split(".")[0] != self.hash_info.value.split(".")[0]:
raise ObjectFormatError(f"{self} is corrupted")
@classmethod
def load(cls, odb, hash_info):
return odb.get(hash_info)
def save(self, odb, **kwargs):
odb.add(self, **kwargs)
class File(HashFile):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.src = None
@classmethod
def stage(cls, odb, path_info, tree, **kwargs):
hash_info = _get_hash(
path_info, tree, odb.tree.PARAM_CHECKSUM, **kwargs
)
raw = odb.get(hash_info)
obj = cls(raw.path_info, raw.tree, hash_info)
obj.src = HashFile(path_info, tree, hash_info)
return obj
def save(self, odb, **kwargs):
self.src.save(odb, **kwargs)
class Tree(HashFile):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.src_hash_info = None
self.src_path_info = None
self.src_tree = None
def __len__(self):
return self.hash_info.dir_info.nfiles
def __iter__(self):
for (
key,
hash_info,
) in self.hash_info.dir_info.trie.iteritems(): # noqa: B301
yield (key, hash_info)
@classmethod
def save_dir_info(cls, odb, dir_info, hash_info=None):
if hash_info and hash_info.name == odb.tree.PARAM_CHECKSUM:
try:
odb.check(hash_info)
assert hash_info.dir_info == dir_info
return hash_info
except (FileNotFoundError, ObjectFormatError):
pass
from dvc.path_info import PathInfo
from dvc.tree.memory import MemoryTree
from dvc.utils import tmp_fname
tree = MemoryTree(None, {})
path_info = PathInfo(tmp_fname(""))
with tree.open(path_info, "w") as fobj:
json.dump(dir_info.to_list(), fobj, sort_keys=True)
tmp_info = odb.tree.path_info / tmp_fname("")
with tree.open(path_info, "rb") as fobj:
odb.tree.upload_fobj(fobj, tmp_info)
hash_info = _get_hash(tmp_info, odb.tree, odb.tree.PARAM_CHECKSUM)
hash_info.value += odb.tree.CHECKSUM_DIR_SUFFIX
hash_info.dir_info = dir_info
hash_info.nfiles = dir_info.nfiles
obj = HashFile(tmp_info, odb.tree, hash_info)
odb.add(obj)
return hash_info
@classmethod
def stage(cls, odb, path_info, tree, **kwargs):
hash_info = _get_hash(
path_info, tree, odb.tree.PARAM_CHECKSUM, **kwargs
)
hi = cls.save_dir_info(odb, hash_info.dir_info, hash_info)
hi.size = hash_info.size
raw = odb.get(hi)
obj = cls(raw.path_info, raw.tree, hi)
obj.src_hash_info = hash_info
obj.src_path_info = path_info
obj.src_tree = tree
return obj
@classmethod
def load(cls, odb, hash_info):
obj = odb.get(hash_info)
try:
with obj.tree.open(obj.path_info, "r") as fobj:
raw = json.load(fobj)
except ValueError as exc:
raise ObjectFormatError(f"{obj} is corrupted") from exc
if not isinstance(raw, list):
logger.error(
"dir cache file format error '%s' [skipping the file]",
obj.path_info,
)
raise ObjectFormatError(f"{obj} is corrupted")
dir_info = DirInfo.from_list(raw)
hash_info.dir_info = dir_info
hash_info.nfiles = dir_info.nfiles
return cls(obj.path_info, obj.tree, hash_info)
def save(self, odb, **kwargs):
from dvc.progress import Tqdm
assert self.src_hash_info.dir_info
hi = self.save_dir_info(
odb, self.src_hash_info.dir_info, self.hash_info
)
for entry_info, entry_hash in Tqdm(
hi.dir_info.items(self.src_path_info),
desc="Saving " + self.src_path_info.name,
unit="file",
):
entry_obj = HashFile(entry_info, self.src_tree, entry_hash)
entry_obj.save(odb, **kwargs)
cache_info = odb.tree.hash_to_path_info(hi.value)
odb.tree.state.save(cache_info, hi)
self.src_tree.state.save(self.src_path_info, hi)
def filter(self, odb, prefix):
hash_info = self.hash_info.dir_info.trie.get(prefix)
if hash_info:
return load(odb, hash_info)
depth = len(prefix)
dir_info = DirInfo()
try:
for key, value in self.hash_info.dir_info.trie.items(prefix):
dir_info.trie[key[depth:]] = value
except KeyError:
return None
return load(odb, self.save_dir_info(odb, dir_info))
def stage(odb, path_info, tree, **kwargs):
if tree.isdir(path_info):
return Tree.stage(odb, path_info, tree, **kwargs)
return File.stage(odb, path_info, tree, **kwargs)
def check(odb, obj):
odb.check(obj.hash_info)
if isinstance(obj, Tree):
for _, hash_info in obj:
odb.check(hash_info)
def load(odb, hash_info):
if hash_info.isdir:
return Tree.load(odb, hash_info)
return File.load(odb, hash_info)
def _get_dir_size(odb, dir_info):
try:
return sum(
odb.tree.getsize(odb.tree.hash_to_path_info(hi.value))
for _, hi in dir_info.items()
)
except FileNotFoundError:
return None
def merge(odb, ancestor_info, our_info, their_info):
assert our_info
assert their_info
if ancestor_info:
ancestor = load(odb, ancestor_info).hash_info.dir_info
else:
ancestor = DirInfo()
our = load(odb, our_info).hash_info.dir_info
their = load(odb, their_info).hash_info.dir_info
merged = our.merge(ancestor, their)
hash_info = Tree.save_dir_info(odb, merged)
hash_info.size = _get_dir_size(odb, merged)
return hash_info
dvc/output/base.py
+80 -31
View File
@@ -6,6 +6,7 @@ from urllib.parse import urlparse
from voluptuous import Any
import dvc.objects as objects
import dvc.prompt as prompt
from dvc.cache import NamedCache
from dvc.checkout import checkout
@@ -144,6 +145,7 @@ class BaseOutput:
if self.use_cache and self.cache is None:
raise RemoteCacheRequiredError(self.path_info)
self.obj = None
self.isexec = False if self.IS_DEPENDENCY else isexec
def _parse_path(self, tree, path):
@@ -181,7 +183,7 @@ class BaseOutput:
@property
def dir_cache(self):
return self.cache.get_dir_cache(self.hash_info)
return self.hash_info.dir_info
@classmethod
def supported(cls, url):
@@ -194,7 +196,7 @@ class BaseOutput:
def get_hash(self):
if not self.use_cache:
return self.tree.get_hash(self.path_info)
return self.cache.get_hash(self.tree, self.path_info)
return objects.stage(self.cache, self.path_info, self.tree).hash_info
@property
def is_dir_checksum(self):
@@ -211,9 +213,15 @@ class BaseOutput:
if not self.use_cache or not self.hash_info:
return True
return self.cache.changed_cache(
self.hash_info, filter_info=filter_info
)
obj = self.get_obj(filter_info=filter_info)
if not obj:
return True
try:
objects.check(self.cache, obj)
return False
except (FileNotFoundError, objects.ObjectFormatError):
return True
def workspace_status(self):
if not self.exists:
@@ -296,7 +304,8 @@ class BaseOutput:
logger.debug("Output '%s' didn't change. Skipping saving.", self)
return
self.hash_info = self.get_hash()
self.obj = objects.stage(self.cache, self.path_info, self.tree)
self.hash_info = self.obj.hash_info
self.isexec = self.isfile() and self.tree.isexec(self.path_info)
def set_exec(self):
@@ -310,20 +319,16 @@ class BaseOutput:
assert self.hash_info
if self.use_cache:
self.cache.save(
self.path_info,
self.tree,
self.hash_info,
filter_info=filter_info,
obj = objects.stage(
self.cache, filter_info or self.path_info, self.tree
)
self.cache.save(obj)
checkout(
self.path_info,
filter_info or self.path_info,
self.tree,
self.hash_info,
obj,
self.cache,
relink=True,
filter_info=filter_info,
)
self.set_exec()
@@ -373,6 +378,23 @@ class BaseOutput:
def download(self, to, jobs=None):
self.tree.download(self.path_info, to.path_info, jobs=jobs)
def get_obj(self, filter_info=None):
if self.obj:
obj = self.obj
elif self.hash_info:
try:
obj = objects.load(self.cache, self.hash_info)
except FileNotFoundError:
return None
else:
return None
if filter_info and filter_info != self.path_info:
prefix = filter_info.relative_to(self.path_info).parts
obj = obj.filter(self.cache, prefix)
return obj
def checkout(
self,
force=False,
@@ -389,16 +411,22 @@ class BaseOutput:
)
return None
obj = self.get_obj(filter_info=filter_info)
if not obj and (filter_info and filter_info != self.path_info):
# backward compatibility
return None
added = not self.exists
try:
res = checkout(
self.path_info,
modified = checkout(
filter_info or self.path_info,
self.tree,
self.hash_info,
obj,
self.cache,
force=force,
progress_callback=progress_callback,
relink=relink,
filter_info=filter_info,
**kwargs,
)
except CheckoutError:
@@ -406,7 +434,7 @@ class BaseOutput:
return None
raise
self.set_exec()
return res
return added, False if added else modified
def remove(self, ignore_remove=False):
self.tree.remove(self.path_info)
@@ -431,11 +459,23 @@ class BaseOutput:
self.repo.scm.ignore(self.fspath)
def get_files_number(self, filter_info=None):
if not self.use_cache:
from funcy import ilen
if not self.use_cache or not self.hash_info:
return 0
return self.cache.get_files_number(
self.path_info, self.hash_info, filter_info
if not self.hash_info.isdir:
return 1
if not filter_info or filter_info == self.path_info:
return self.hash_info.nfiles
if not self.hash_info.dir_info:
return 0
return ilen(
filter_info.isin_or_eq(self.path_info / relpath)
for relpath, _ in self.hash_info.dir_info.items()
)
def unprotect(self):
@@ -443,17 +483,25 @@ class BaseOutput:
self.cache.unprotect(self.path_info)
def get_dir_cache(self, **kwargs):
if not self.is_dir_checksum:
raise DvcException("cannot get dir cache for file checksum")
if self.cache.changed_cache_file(self.hash_info):
try:
objects.check(self.cache, self.cache.get(self.hash_info))
except (FileNotFoundError, objects.ObjectFormatError):
self.repo.cloud.pull(
NamedCache.make("local", self.hash_info.value, str(self)),
show_checksums=False,
**kwargs,
)
try:
objects.load(self.cache, self.hash_info)
assert self.hash_info.dir_info
except (objects.ObjectFormatError, FileNotFoundError):
self.hash_info.dir_info = None
if not self.hash_info.dir_info:
self.cache.set_dir_info(self.hash_info)
return self.dir_cache
def collect_used_dir_cache(
@@ -485,7 +533,9 @@ class BaseOutput:
except DvcException:
logger.debug(f"failed to pull cache for '{self}'")
if self.cache.changed_cache_file(self.hash_info):
try:
objects.check(self.cache, self.cache.get(self.hash_info))
except (FileNotFoundError, objects.ObjectFormatError):
msg = (
"Missing cache for directory '{}'. "
"Cache for files inside will be lost. "
@@ -496,8 +546,7 @@ class BaseOutput:
"unable to fully collect used cache"
" without cache for directory '{}'".format(self)
)
else:
return cache
return cache
path = str(self.path_info)
filter_path = str(filter_info) if filter_info else None
@@ -615,6 +664,6 @@ class BaseOutput:
self._check_can_merge(self)
self._check_can_merge(other)
self.hash_info = self.cache.merge(
ancestor_info, self.hash_info, other.hash_info
self.hash_info = objects.merge(
self.cache, ancestor_info, self.hash_info, other.hash_info
)
dvc/repo/fetch.py
+6 -3
View File
@@ -81,6 +81,7 @@ def fetch(
def _fetch_external(self, repo_url, repo_rev, files, jobs):
from dvc.external_repo import external_repo
from dvc.objects import stage
from dvc.path_info import PathInfo
from dvc.scm.base import CloneError
@@ -109,14 +110,16 @@ def _fetch_external(self, repo_url, repo_rev, files, jobs):
cb(repo.cloud.pull(used, jobs))
except (NoOutputOrStageError, NoRemoteError):
pass
self.cache.local.save(
obj = stage(
cache,
path_info,
repo.repo_tree,
None,
jobs=jobs,
download_callback=cb,
follow_subrepos=False,
)
cache.save(
obj, jobs=jobs, download_callback=cb,
)
except CloneError:
failed += 1
logger.exception(
dvc/tree/base.py
+13 -3
View File
@@ -1,4 +1,6 @@
import errno
import logging
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from functools import partial
from multiprocessing import cpu_count
@@ -252,7 +254,9 @@ class BaseTree:
)
if not self.exists(path_info):
return None
raise FileNotFoundError(
errno.ENOENT, os.strerror(errno.ENOENT), path_info
)
# pylint: disable=assignment-from-none
hash_info = self.state.get(path_info)
@@ -272,7 +276,10 @@ class BaseTree:
if hash_info:
assert hash_info.name == self.PARAM_CHECKSUM
if hash_info.isdir:
self.cache.set_dir_info(hash_info)
from dvc.objects import Tree
# NOTE: loading the tree will restore hash_info.dir_info
Tree.load(self.cache, hash_info)
return hash_info
if self.isdir(path_info):
@@ -347,9 +354,12 @@ class BaseTree:
@use_state
def get_dir_hash(self, path_info, **kwargs):
from dvc.objects import Tree
dir_info = self._collect_dir(path_info, **kwargs)
hash_info = self.repo.cache.local.save_dir_info(dir_info)
hash_info = Tree.save_dir_info(self.repo.cache.local, dir_info)
hash_info.size = dir_info.size
hash_info.dir_info = dir_info
return hash_info
def upload(
dvc/tree/dvc.py
+6 -1
View File
@@ -132,7 +132,12 @@ class DvcTree(BaseTree): # pylint:disable=abstract-method
out.get_dir_cache(**kwargs)
dir_cache = out.dir_cache
hash_info = out.cache.save_dir_info(dir_cache)
if not dir_cache:
raise FileNotFoundError
from dvc.objects import Tree
hash_info = Tree.save_dir_info(out.cache, dir_cache)
if hash_info != out.hash_info:
raise FileNotFoundError
dvc/tree/memory.py
+37
View File
@@ -0,0 +1,37 @@
from dvc.hash_info import HashInfo
from dvc.utils import file_md5
from .base import BaseTree
class MemoryTree(BaseTree):
scheme = "local"
PARAM_CHECKSUM = "md5"
def __init__(self, repo, config):
from fsspec.implementations.memory import MemoryFileSystem
super().__init__(repo, config)
self.fs = MemoryFileSystem()
def exists(self, path_info, use_dvcignore=True):
return self.fs.exists(path_info.path)
def open(self, path_info, mode="r", encoding=None, **kwargs):
return self.fs.open(
path_info.fspath, mode=mode, encoding=encoding, **kwargs
)
def stat(self, path_info):
import os
info = self.fs.info(path_info.fspath)
return os.stat_result((0, 0, 0, 0, 0, 0, info["size"], 0, 0, 0))
def get_file_hash(self, path_info):
return HashInfo(self.PARAM_CHECKSUM, file_md5(path_info, self))
def walk_files(self, path_info, **kwargs):
raise NotImplementedError
setup.py
+1
View File
@@ -87,6 +87,7 @@ install_requires = [
"python-benedict>=0.21.1",
"pyparsing==2.4.7",
"typing_extensions>=3.7.4",
"fsspec>=0.8.5",
]
tests/func/test_add.py
+3 -1
View File
@@ -96,6 +96,8 @@ def test_add_unsupported_file(dvc):
def test_add_directory(tmp_dir, dvc):
from dvc.objects import load
(stage,) = tmp_dir.dvc_gen({"dir": {"file": "file"}})
assert stage is not None
@@ -104,7 +106,7 @@ def test_add_directory(tmp_dir, dvc):
hash_info = stage.outs[0].hash_info
dir_info = dvc.cache.local.load_dir_cache(hash_info)
dir_info = load(dvc.cache.local, hash_info).hash_info.dir_info
for path, _ in dir_info.trie.items():
assert "\\" not in path
tests/func/test_cache.py
+7 -9
View File
@@ -6,10 +6,9 @@ import configobj
import pytest
from dvc.cache import Cache
from dvc.cache.base import DirCacheError
from dvc.dir_info import DirInfo
from dvc.hash_info import HashInfo
from dvc.main import main
from dvc.objects import ObjectFormatError
from dvc.utils import relpath
from tests.basic_env import TestDir, TestDvc
@@ -49,24 +48,23 @@ class TestCacheLoadBadDirCache(TestDvc):
self.assertEqual(len(ret), 0)
def test(self):
from dvc.objects import load
dir_hash = "123.dir"
fname = os.fspath(
self.dvc.cache.local.tree.hash_to_path_info(dir_hash)
)
self.create(fname, "<clearly>not,json")
with pytest.raises(DirCacheError):
self.dvc.cache.local.load_dir_cache(HashInfo("md5", dir_hash))
with pytest.raises(ObjectFormatError):
load(self.dvc.cache.local, HashInfo("md5", dir_hash))
dir_hash = "234.dir"
fname = os.fspath(
self.dvc.cache.local.tree.hash_to_path_info(dir_hash)
)
self.create(fname, '{"a": "b"}')
dir_info = self.dvc.cache.local.load_dir_cache(
HashInfo("md5", dir_hash)
)
self.assertTrue(isinstance(dir_info, DirInfo))
self.assertEqual(dir_info.nfiles, 0)
with pytest.raises(ObjectFormatError):
load(self.dvc.cache.local, HashInfo("md5", dir_hash))
class TestExternalCacheDir(TestDvc):
tests/func/test_checkout.py
+5 -1
View File
@@ -85,6 +85,8 @@ class TestCheckoutCorruptedCacheFile(TestRepro):
class TestCheckoutCorruptedCacheDir(TestDvc):
def test(self):
from dvc.objects import load
# NOTE: using 'copy' so that cache and link don't have same inode
ret = main(["config", "cache.type", "copy"])
self.assertEqual(ret, 0)
@@ -98,7 +100,9 @@ class TestCheckoutCorruptedCacheDir(TestDvc):
# NOTE: modifying cache file for one of the files inside the directory
# to check if dvc will detect that the cache is corrupted.
_, entry_hash = next(
self.dvc.cache.local.load_dir_cache(out.hash_info).items()
load(
self.dvc.cache.local, out.hash_info
).hash_info.dir_info.items()
)
cache = os.fspath(
self.dvc.cache.local.tree.hash_to_path_info(entry_hash.value)
tests/func/test_commit.py
+1
View File
@@ -148,6 +148,7 @@ def test_commit_granular_dir(tmp_dir, dvc):
dvc.commit(os.path.join("data", "subdir"))
assert set(cache.glob("*/*")) == {
cache / "1a" / "ca2c799df82929bbdd976557975546.dir",
cache / "26" / "d6b64d96a660707412f523e8184b5f.dir",
cache / "ac" / "bd18db4cc2f85cedef654fccc4a4d8",
cache / "4c" / "e8d2a2cf314a52fa7f315ca37ca445",
cache / "68" / "dde2c3c4e7953c2290f176bbdc9a54",
tests/func/test_external_repo.py
+5 -12
View File
@@ -3,7 +3,7 @@ import os
from mock import ANY, patch
from dvc.external_repo import CLONES, external_repo
from dvc.hash_info import HashInfo
from dvc.objects import stage
from dvc.path_info import PathInfo
from dvc.scm.git import Git
from dvc.tree.local import LocalTree
@@ -200,20 +200,13 @@ def test_subrepos_are_ignored(tmp_dir, erepo_dir):
remove(cache_dir)
makedirs(cache_dir)
expected_hash = HashInfo("md5", "e1d9e8eae5374860ae025ec84cfd85c7.dir")
assert (
repo.repo_tree.get_hash(
os.path.join(repo.root_dir, "dir"), follow_subrepos=False
)
== expected_hash
)
repo.cache.local.save(
obj = stage(
repo.cache.local,
PathInfo(repo.root_dir) / "dir",
repo.repo_tree,
expected_hash,
link=False,
follow_subrepos=False,
)
repo.cache.local.save(obj)
assert set(cache_dir.glob("*/*")) == {
cache_dir / "e1" / "d9e8eae5374860ae025ec84cfd85c7.dir",
cache_dir / "37" / "b51d194a7513e45b56f6524f2d51f2",
tests/unit/output/test_local.py
+12 -13
View File
@@ -2,7 +2,6 @@ import os
from mock import patch
from dvc.cache.local import LocalCache
from dvc.dir_info import DirInfo
from dvc.hash_info import HashInfo
from dvc.output import LocalOutput
@@ -79,19 +78,19 @@ class TestGetFilesNumber(TestDvc):
o.use_cache = False
self.assertEqual(0, o.get_files_number())
@patch.object(
LocalCache,
"get_dir_cache",
return_value=DirInfo.from_list(
[
{"relpath": "foo", "md5": "asdf"},
{"relpath": "bar", "md5": "qwe"},
]
),
)
def test_return_multiple_for_dir(self, _mock_get_dir_cache):
def test_return_multiple_for_dir(self):
o = self._get_output()
o.hash_info = HashInfo("md5", "12345678.dir")
o.hash_info = HashInfo(
"md5",
"12345678.dir",
nfiles=2,
dir_info=DirInfo.from_list(
[
{"relpath": "foo", "md5": "asdf"},
{"relpath": "bar", "md5": "qwe"},
]
),
)
self.assertEqual(2, o.get_files_number())
@patch.object(LocalOutput, "is_dir_checksum", False)
tests/unit/test_external_repo.py
+7 -2
View File
@@ -4,6 +4,7 @@ from unittest.mock import call
import pytest
from dvc.external_repo import external_repo
from dvc.objects import stage
from dvc.path_info import PathInfo
from tests.unit.tree.test_repo import make_subrepo
@@ -109,12 +110,16 @@ def test_fetch_external_repo_jobs(tmp_dir, scm, mocker, dvc, local_remote):
with external_repo(str(tmp_dir)) as repo:
spy = mocker.spy(repo.cloud, "pull")
dvc.cache.local.save(
obj = stage(
dvc.cache.local,
PathInfo(repo.root_dir) / "dir1",
repo.repo_tree,
None,
follow_subrepos=False,
jobs=3,
)
dvc.cache.local.save(
obj, jobs=3,
)
run_jobs = tuple(spy.call_args_list[0])[1].get("jobs")
assert run_jobs == 3
tests/unit/tree/test_dvc.py
+4 -4
View File
@@ -23,7 +23,7 @@ def test_open(tmp_dir, dvc):
(tmp_dir / "foo").unlink()
tree = DvcTree(dvc)
with tree.open("foo", "r") as fobj:
with tree.open(PathInfo(tmp_dir) / "foo", "r") as fobj:
assert fobj.read() == "foo"
@@ -32,7 +32,7 @@ def test_open_dirty_hash(tmp_dir, dvc):
(tmp_dir / "file").write_text("something")
tree = DvcTree(dvc)
with tree.open("file", "r") as fobj:
with tree.open(PathInfo(tmp_dir) / "file", "r") as fobj:
# NOTE: Unlike RepoTree, DvcTree should not
# be affected by a dirty workspace.
assert fobj.read() == "file"
@@ -46,7 +46,7 @@ def test_open_dirty_no_hash(tmp_dir, dvc):
# NOTE: Unlike RepoTree, DvcTree should not
# be affected by a dirty workspace.
with pytest.raises(FileNotFoundError):
with tree.open("file", "r"):
with tree.open(PathInfo(tmp_dir) / "file", "r"):
pass
@@ -66,7 +66,7 @@ def test_open_in_history(tmp_dir, scm, dvc):
continue
tree = DvcTree(dvc)
with tree.open("foo", "r") as fobj:
with tree.open(PathInfo(tmp_dir) / "foo", "r") as fobj:
assert fobj.read() == "foo"
tests/unit/tree/test_repo.py
+4 -4
View File
@@ -24,7 +24,7 @@ def test_open(tmp_dir, dvc):
(tmp_dir / "foo").unlink()
tree = RepoTree(dvc)
with tree.open("foo", "r") as fobj:
with tree.open(PathInfo(tmp_dir) / "foo", "r") as fobj:
assert fobj.read() == "foo"
@@ -33,7 +33,7 @@ def test_open_dirty_hash(tmp_dir, dvc):
(tmp_dir / "file").write_text("something")
tree = RepoTree(dvc)
with tree.open("file", "r") as fobj:
with tree.open(PathInfo(tmp_dir) / "file", "r") as fobj:
assert fobj.read() == "something"
@@ -42,7 +42,7 @@ def test_open_dirty_no_hash(tmp_dir, dvc):
(tmp_dir / "file.dvc").write_text("outs:\n- path: file\n")
tree = RepoTree(dvc)
with tree.open("file", "r") as fobj:
with tree.open(PathInfo(tmp_dir) / "file", "r") as fobj:
assert fobj.read() == "file"
@@ -62,7 +62,7 @@ def test_open_in_history(tmp_dir, scm, dvc):
continue
tree = RepoTree(dvc)
with tree.open("foo", "r") as fobj:
with tree.open(PathInfo(tmp_dir) / "foo", "r") as fobj:
assert fobj.read() == "foo"