From 90ca5bd35adba55678c8a307d2a4cab9ac2a5062 Mon Sep 17 00:00:00 2001 From: pared Date: Mon, 3 Jun 2019 14:03:15 +0200 Subject: [PATCH 1/6] remote: local: add cache status check hardlink optimization --- dvc/remote/base.py | 12 +++-- dvc/remote/local/__init__.py | 96 +++++++++++++++++++++++++++++++++ tests/func/remote/__init__.py | 0 tests/func/remote/test_local.py | 59 ++++++++++++++++++++ tests/utils/__init__.py | 11 ++++ 5 files changed, 175 insertions(+), 3 deletions(-) create mode 100644 tests/func/remote/__init__.py create mode 100644 tests/func/remote/test_local.py diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 30f13a8f4..fa9a5c6b9 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -372,6 +372,7 @@ class RemoteBASE(object): self.state.save_link(path_info) self.state.save(cache_info, checksum) self.state.save(path_info, checksum) + return dir_info def is_empty(self, path_info): return False @@ -483,9 +484,7 @@ class RemoteBASE(object): ) def gc(self, cinfos): - from dvc.remote.local import RemoteLOCAL - - used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]} + used = self.extract_used_local_checksums(cinfos) if self.scheme != "": used |= { @@ -636,6 +635,7 @@ class RemoteBASE(object): self.state.save_link(path_info) self.state.save(path_info, checksum) + return dir_info def _remove_redundant_files(self, path_info, dir_info, force): existing_files = set( @@ -701,3 +701,9 @@ class RemoteBASE(object): @staticmethod def unprotect(path_info): pass + + def extract_used_local_checksums(self, cinfos): + from dvc.remote import RemoteLOCAL + + used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]} + return used diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index 3e314da4f..7a4ab20aa 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -38,6 +38,7 @@ from dvc.exceptions import DvcException from dvc.progress import progress from concurrent.futures import ThreadPoolExecutor +from dvc.utils.fs import get_mtime_and_size, get_inode logger = logging.getLogger(__name__) @@ -48,6 +49,8 @@ class RemoteLOCAL(RemoteBASE): PARAM_CHECKSUM = "md5" PARAM_PATH = "path" + STATUS_VERIFICATION_DIR_SUFFIX = "unpacked" + DEFAULT_CACHE_TYPES = ["reflink", "copy"] CACHE_TYPE_MAP = { "copy": shutil.copyfile, @@ -566,3 +569,96 @@ class RemoteLOCAL(RemoteBASE): @staticmethod def protect(path_info): os.chmod(path_info.path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH) + + def append_verififcation_suffix(self, string): + return string + "." + self.STATUS_VERIFICATION_DIR_SUFFIX + + def get_verification_dir_info(self, checksum): + verification_dir_info = self.checksum_to_path_info(checksum) + verification_dir_info.path = self.append_verififcation_suffix( + verification_dir_info.path + ) + + return verification_dir_info + + def create_status_verification_dir(self, checksum, dir_info): + verification_dir_info = self.get_verification_dir_info(checksum) + + self.makedirs(verification_dir_info) + + for entry in dir_info: + entry_cache_info = self.checksum_to_path_info( + entry[self.PARAM_CHECKSUM] + ) + relpath = entry[self.PARAM_RELPATH] + + verification_entry_info = copy(entry_cache_info) + verification_entry_info.path = self.ospath.join( + verification_dir_info.path, relpath + ) + self.hardlink(entry_cache_info, verification_entry_info) + + self.state.save(verification_dir_info, checksum) + + def _path_metadata_up_to_date(self, path): + mtime, size = get_mtime_and_size(path) + inode = get_inode(path) + + entry = self.state.get_state_record_for_inode(inode) + + if entry: + state_mtime, state_size, _, _ = entry + if state_mtime == mtime and state_size == size: + return True + return False + + def _is_verification_dir_up_to_date(self, checksum): + status_verification_dir_info = self.get_verification_dir_info(checksum) + + if os.path.exists(status_verification_dir_info.path): + return self._path_metadata_up_to_date( + status_verification_dir_info.path + ) + return False + + def _changed_dir_cache(self, checksum): + if self._is_verification_dir_up_to_date(checksum): + return False + return super(RemoteLOCAL, self)._changed_dir_cache(checksum) + + def _save_dir(self, path_info, checksum): + dir_info = super(RemoteLOCAL, self)._save_dir(path_info, checksum) + self.create_status_verification_dir(checksum, dir_info) + return dir_info + + def _checkout_dir( + self, path_info, checksum, force, progress_callback=None + ): + dir_info = super(RemoteLOCAL, self)._checkout_dir( + path_info, checksum, force, progress_callback + ) + self.create_status_verification_dir(checksum, dir_info) + return dir_info + + def extract_used_local_checksums(self, cinfos): + used = super(RemoteLOCAL, self).extract_used_local_checksums(cinfos) + verification_dirs = set() + for checksum in used: + if self.is_dir_checksum(checksum): + verification_dirs.add( + self.append_verififcation_suffix(checksum) + ) + return used | verification_dirs + + def hardlink(self, from_info, to_info): + cache = from_info.path + path = to_info.path + + assert os.path.isfile(cache) + + dname = os.path.dirname(path) + if not os.path.exists(dname): + os.makedirs(dname) + + if not os.path.lexists(path): + System.hardlink(cache, path) diff --git a/tests/func/remote/__init__.py b/tests/func/remote/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/func/remote/test_local.py b/tests/func/remote/test_local.py new file mode 100644 index 000000000..4c088a6fb --- /dev/null +++ b/tests/func/remote/test_local.py @@ -0,0 +1,59 @@ +import os +import shutil + +from dvc.main import main +from dvc.remote.local import RemoteLOCAL +from dvc.repo import Repo +from dvc.utils import file_md5 +from dvc.utils.fs import get_inode +from tests.utils import trees_equal + + +def test_should_create_cache_status_check_dir_on_add(dvc_repo, repo_dir): + + stages = dvc_repo.add(repo_dir.DATA_DIR) + verification_dir = ( + stages[0].outs[0].cache_path + + "." + + RemoteLOCAL.STATUS_VERIFICATION_DIR_SUFFIX + ) + + assert os.path.exists(verification_dir) + + trees_equal(repo_dir.DATA_DIR, verification_dir) + data_cache_path = dvc_repo.cache.local.get(file_md5(repo_dir.DATA)[0]) + data_sub_cache_path = dvc_repo.cache.local.get( + file_md5(repo_dir.DATA_SUB)[0] + ) + assert get_inode(os.path.join(verification_dir, "data")) == get_inode( + data_cache_path + ) + assert get_inode( + os.path.join(verification_dir, "data_sub_dir", "data_sub") + ) == get_inode(data_sub_cache_path) + + +def test_should_create_cache_status_check_dir_on_checkout(dvc_repo, repo_dir): + local_storage = repo_dir.mkdtemp() + ret = main(["remote", "add", "-d", "storage", local_storage]) + assert ret == 0 + + dvc_repo = Repo(dvc_repo.dvc_dir) + + stages = dvc_repo.add(repo_dir.DATA_DIR) + verification_dir = ( + stages[0].outs[0].cache_path + + "." + + RemoteLOCAL.STATUS_VERIFICATION_DIR_SUFFIX + ) + + assert dvc_repo.push() == 1 + + shutil.rmtree(dvc_repo.cache.local.cache_dir) + shutil.rmtree(repo_dir.DATA_DIR) + assert not os.path.exists(verification_dir) + + dvc_repo.pull("{}.dvc".format(repo_dir.DATA_DIR)) + + assert os.path.exists(verification_dir) + trees_equal(repo_dir.DATA_DIR, verification_dir) diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py index 3bbe0c3c3..b43af8d06 100644 --- a/tests/utils/__init__.py +++ b/tests/utils/__init__.py @@ -1,4 +1,5 @@ import os +from filecmp import dircmp from dvc.scm import Git from mock import MagicMock @@ -29,3 +30,13 @@ def cd(newdir): yield finally: os.chdir(prevdir) + + +def trees_equal(dir_path_1, dir_path_2): + + comparison = dircmp(dir_path_1, dir_path_2) + + assert set(comparison.left_only) == set(comparison.right_only) == set() + + for d in comparison.common_dirs: + trees_equal(os.path.join(dir_path_1, d), os.path.join(dir_path_2, d)) From 5d25d5ce965720b6339845c84802801eca14b995 Mon Sep 17 00:00:00 2001 From: pawel Date: Wed, 5 Jun 2019 13:22:10 +0200 Subject: [PATCH 2/6] remote: local: refactor link method --- dvc/remote/base.py | 2 +- dvc/remote/local/__init__.py | 86 ++++++++++++++++++++---------------- tests/func/test_repro.py | 1 + 3 files changed, 50 insertions(+), 39 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index fa9a5c6b9..dac99e1b9 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -333,7 +333,7 @@ class RemoteBASE(object): logger.debug("'{}' hasn't changed.".format(path_info)) return False - def link(self, from_info, to_info): + def link(self, from_info, to_info, link_type=None): self.copy(from_info, to_info) def _save_file(self, path_info, checksum, save_link=True): diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index 7a4ab20aa..fa5de7453 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -126,41 +126,64 @@ class RemoteLOCAL(RemoteBASE): if not self.exists(path_info): os.makedirs(path_info.path) - @slow_link_guard - def link(self, cache_info, path_info): - cache = cache_info.path - path = path_info.path + def link(self, from_info, to_info, link_type=None): + from_path = from_info.path + to_path = to_info.path - assert os.path.isfile(cache) + assert os.path.isfile(from_path) - dname = os.path.dirname(path) + dname = os.path.dirname(to_path) if not os.path.exists(dname): os.makedirs(dname) # NOTE: just create an empty file for an empty cache - if os.path.getsize(cache) == 0: - open(path, "w+").close() + if os.path.getsize(from_path) == 0: + open(to_path, "w+").close() - msg = "Created empty file: {} -> {}".format(cache, path) + msg = "Created empty file: {} -> {}".format(from_path, to_path) logger.debug(msg) return + if not link_type: + self._default_link(from_path, to_path) + else: + link_method = self._get_link_method(link_type) + self._link(from_path, to_path, link_method) + + @classmethod + def _get_link_method(cls, link_type): + try: + return cls.CACHE_TYPE_MAP[link_type] + except KeyError: + raise DvcException( + "Cache type: '{}' not supported!".format(link_type) + ) + + def _link(self, from_path, to_path, link_method): + if os.path.lexists(to_path): + # TODO handle existing link + pass + else: + link_method(from_path, to_path) + if self.protected: + self.protect(to_path) + + msg = "Created {}'{}': {} -> {}".format( + "protected " if self.protected else "", + self.cache_types[0], + from_path, + to_path, + ) + + logger.debug(msg) + + @slow_link_guard + def _default_link(self, from_path, to_path): i = len(self.cache_types) while i > 0: + link_method = self._get_link_method(self.cache_types[0]) try: - self.CACHE_TYPE_MAP[self.cache_types[0]](cache, path) - - if self.protected: - self.protect(path_info) - - msg = "Created {}'{}': {} -> {}".format( - "protected " if self.protected else "", - self.cache_types[0], - cache, - path, - ) - - logger.debug(msg) + self._link(from_path, to_path, link_method) return except DvcException as exc: @@ -567,8 +590,8 @@ class RemoteLOCAL(RemoteBASE): RemoteLOCAL._unprotect_file(path) @staticmethod - def protect(path_info): - os.chmod(path_info.path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH) + def protect(path): + os.chmod(path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH) def append_verififcation_suffix(self, string): return string + "." + self.STATUS_VERIFICATION_DIR_SUFFIX @@ -596,7 +619,7 @@ class RemoteLOCAL(RemoteBASE): verification_entry_info.path = self.ospath.join( verification_dir_info.path, relpath ) - self.hardlink(entry_cache_info, verification_entry_info) + self.link(entry_cache_info, verification_entry_info, "hardlink") self.state.save(verification_dir_info, checksum) @@ -649,16 +672,3 @@ class RemoteLOCAL(RemoteBASE): self.append_verififcation_suffix(checksum) ) return used | verification_dirs - - def hardlink(self, from_info, to_info): - cache = from_info.path - path = to_info.path - - assert os.path.isfile(cache) - - dname = os.path.dirname(path) - if not os.path.exists(dname): - os.makedirs(dname) - - if not os.path.lexists(path): - System.hardlink(cache, path) diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py index a58fcbd6f..93375e801 100644 --- a/tests/func/test_repro.py +++ b/tests/func/test_repro.py @@ -1044,6 +1044,7 @@ class TestReproExternalHDFS(TestReproExternalBase): self.assertEqual(p.returncode, 0) +@pytest.mark.skip @flaky(max_runs=3, min_passes=1) class TestReproExternalSSH(TestReproExternalBase): _dir = None From b5464cbc59740d546f9089fe1b492542191d68a3 Mon Sep 17 00:00:00 2001 From: pawel Date: Wed, 5 Jun 2019 16:19:23 +0200 Subject: [PATCH 3/6] remote: local: check if status verification dir already exists --- dvc/remote/base.py | 9 ++- dvc/remote/local/__init__.py | 119 +++++++++++++++----------------- tests/func/remote/test_local.py | 42 +++++++---- tests/func/test_repro.py | 1 - 4 files changed, 91 insertions(+), 80 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index dac99e1b9..ea620eeed 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -386,7 +386,8 @@ class RemoteBASE(object): def walk(self, path_info): raise NotImplementedError - def protect(self, path_info): + @staticmethod + def protect(path_info): pass def save(self, path_info, checksum_info): @@ -538,6 +539,9 @@ class RemoteBASE(object): if self.changed_cache_file(checksum): return True + if not self._changed_unpacked_dir(checksum): + return False + for entry in self.get_dir_cache(checksum): checksum = entry[self.PARAM_CHECKSUM] if self.changed_cache_file(checksum): @@ -707,3 +711,6 @@ class RemoteBASE(object): used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]} return used + + def _changed_unpacked_dir(self, checksum): + return True diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index fa5de7453..038863adc 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -38,7 +38,6 @@ from dvc.exceptions import DvcException from dvc.progress import progress from concurrent.futures import ThreadPoolExecutor -from dvc.utils.fs import get_mtime_and_size, get_inode logger = logging.getLogger(__name__) @@ -49,7 +48,7 @@ class RemoteLOCAL(RemoteBASE): PARAM_CHECKSUM = "md5" PARAM_PATH = "path" - STATUS_VERIFICATION_DIR_SUFFIX = "unpacked" + UNPACKED_DIR_SUFFIX = ".unpacked" DEFAULT_CACHE_TYPES = ["reflink", "copy"] CACHE_TYPE_MAP = { @@ -145,10 +144,10 @@ class RemoteLOCAL(RemoteBASE): return if not link_type: - self._default_link(from_path, to_path) + self._default_link(from_info, to_info) else: link_method = self._get_link_method(link_type) - self._link(from_path, to_path, link_method) + self._link(from_info, to_info, link_method) @classmethod def _get_link_method(cls, link_type): @@ -159,31 +158,32 @@ class RemoteLOCAL(RemoteBASE): "Cache type: '{}' not supported!".format(link_type) ) - def _link(self, from_path, to_path, link_method): - if os.path.lexists(to_path): - # TODO handle existing link - pass + def _link(self, from_info, to_info, link_method): + if os.path.lexists(to_info.path): + raise DvcException( + "Link '{}' already exists!".format(to_info.path) + ) else: - link_method(from_path, to_path) + link_method(from_info.path, to_info.path) if self.protected: - self.protect(to_path) + self.protect(to_info) msg = "Created {}'{}': {} -> {}".format( "protected " if self.protected else "", self.cache_types[0], - from_path, - to_path, + from_info.path, + to_info.path, ) logger.debug(msg) @slow_link_guard - def _default_link(self, from_path, to_path): + def _default_link(self, from_info, to_info): i = len(self.cache_types) while i > 0: link_method = self._get_link_method(self.cache_types[0]) try: - self._link(from_path, to_path, link_method) + self._link(from_info, to_info, link_method) return except DvcException as exc: @@ -590,68 +590,63 @@ class RemoteLOCAL(RemoteBASE): RemoteLOCAL._unprotect_file(path) @staticmethod - def protect(path): - os.chmod(path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH) + def protect(path_info): + os.chmod(path_info.path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH) - def append_verififcation_suffix(self, string): - return string + "." + self.STATUS_VERIFICATION_DIR_SUFFIX + @classmethod + def _append_unpacked_suffix(cls, string): + return string + cls.UNPACKED_DIR_SUFFIX - def get_verification_dir_info(self, checksum): - verification_dir_info = self.checksum_to_path_info(checksum) - verification_dir_info.path = self.append_verififcation_suffix( - verification_dir_info.path + def _get_unpacked_dir_path_info(self, checksum): + unpacked_dir_info = self.checksum_to_path_info(checksum) + unpacked_dir_info.path = self._append_unpacked_suffix( + unpacked_dir_info.path ) - return verification_dir_info + return unpacked_dir_info - def create_status_verification_dir(self, checksum, dir_info): - verification_dir_info = self.get_verification_dir_info(checksum) + def _update_unpacked_dir(self, checksum, dir_info): + unpacked_dir_info = self._get_unpacked_dir_path_info(checksum) - self.makedirs(verification_dir_info) + if self.exists(unpacked_dir_info): + if self.state.get(unpacked_dir_info): + return + shutil.rmtree(unpacked_dir_info.path) + + try: + self._create_unpacked_dir(checksum, dir_info, unpacked_dir_info) + except Exception: + logger.warning( + "Could not create '{}'".format(unpacked_dir_info.path) + ) + + if self.exists(unpacked_dir_info): + self.remove(unpacked_dir_info) + + def _create_unpacked_dir(self, checksum, dir_info, unpacked_dir_info): + self.makedirs(unpacked_dir_info) for entry in dir_info: entry_cache_info = self.checksum_to_path_info( entry[self.PARAM_CHECKSUM] ) relpath = entry[self.PARAM_RELPATH] - verification_entry_info = copy(entry_cache_info) - verification_entry_info.path = self.ospath.join( - verification_dir_info.path, relpath + unpacked_entry_info = copy(entry_cache_info) + unpacked_entry_info.path = self.ospath.join( + unpacked_dir_info.path, relpath ) - self.link(entry_cache_info, verification_entry_info, "hardlink") + self.link(entry_cache_info, unpacked_entry_info, "hardlink") + self.state.save(unpacked_dir_info, checksum) - self.state.save(verification_dir_info, checksum) + def _changed_unpacked_dir(self, checksum): + status_unpacked_dir_info = self._get_unpacked_dir_path_info(checksum) - def _path_metadata_up_to_date(self, path): - mtime, size = get_mtime_and_size(path) - inode = get_inode(path) - - entry = self.state.get_state_record_for_inode(inode) - - if entry: - state_mtime, state_size, _, _ = entry - if state_mtime == mtime and state_size == size: - return True - return False - - def _is_verification_dir_up_to_date(self, checksum): - status_verification_dir_info = self.get_verification_dir_info(checksum) - - if os.path.exists(status_verification_dir_info.path): - return self._path_metadata_up_to_date( - status_verification_dir_info.path - ) - return False - - def _changed_dir_cache(self, checksum): - if self._is_verification_dir_up_to_date(checksum): - return False - return super(RemoteLOCAL, self)._changed_dir_cache(checksum) + return not self.state.get(status_unpacked_dir_info) def _save_dir(self, path_info, checksum): dir_info = super(RemoteLOCAL, self)._save_dir(path_info, checksum) - self.create_status_verification_dir(checksum, dir_info) + self._update_unpacked_dir(checksum, dir_info) return dir_info def _checkout_dir( @@ -660,15 +655,13 @@ class RemoteLOCAL(RemoteBASE): dir_info = super(RemoteLOCAL, self)._checkout_dir( path_info, checksum, force, progress_callback ) - self.create_status_verification_dir(checksum, dir_info) + self._update_unpacked_dir(checksum, dir_info) return dir_info def extract_used_local_checksums(self, cinfos): used = super(RemoteLOCAL, self).extract_used_local_checksums(cinfos) - verification_dirs = set() + unpacked_dir = set() for checksum in used: if self.is_dir_checksum(checksum): - verification_dirs.add( - self.append_verififcation_suffix(checksum) - ) - return used | verification_dirs + unpacked_dir.add(self._append_unpacked_suffix(checksum)) + return used | unpacked_dir diff --git a/tests/func/remote/test_local.py b/tests/func/remote/test_local.py index 4c088a6fb..2cbbbadd0 100644 --- a/tests/func/remote/test_local.py +++ b/tests/func/remote/test_local.py @@ -1,6 +1,9 @@ import os import shutil +import mock + +import dvc from dvc.main import main from dvc.remote.local import RemoteLOCAL from dvc.repo import Repo @@ -12,24 +15,22 @@ from tests.utils import trees_equal def test_should_create_cache_status_check_dir_on_add(dvc_repo, repo_dir): stages = dvc_repo.add(repo_dir.DATA_DIR) - verification_dir = ( - stages[0].outs[0].cache_path - + "." - + RemoteLOCAL.STATUS_VERIFICATION_DIR_SUFFIX + unpacked_dir = ( + stages[0].outs[0].cache_path + RemoteLOCAL.UNPACKED_DIR_SUFFIX ) - assert os.path.exists(verification_dir) + assert os.path.exists(unpacked_dir) - trees_equal(repo_dir.DATA_DIR, verification_dir) + trees_equal(repo_dir.DATA_DIR, unpacked_dir) data_cache_path = dvc_repo.cache.local.get(file_md5(repo_dir.DATA)[0]) data_sub_cache_path = dvc_repo.cache.local.get( file_md5(repo_dir.DATA_SUB)[0] ) - assert get_inode(os.path.join(verification_dir, "data")) == get_inode( + assert get_inode(os.path.join(unpacked_dir, "data")) == get_inode( data_cache_path ) assert get_inode( - os.path.join(verification_dir, "data_sub_dir", "data_sub") + os.path.join(unpacked_dir, "data_sub_dir", "data_sub") ) == get_inode(data_sub_cache_path) @@ -41,19 +42,30 @@ def test_should_create_cache_status_check_dir_on_checkout(dvc_repo, repo_dir): dvc_repo = Repo(dvc_repo.dvc_dir) stages = dvc_repo.add(repo_dir.DATA_DIR) - verification_dir = ( - stages[0].outs[0].cache_path - + "." - + RemoteLOCAL.STATUS_VERIFICATION_DIR_SUFFIX + unpacked_dir = ( + stages[0].outs[0].cache_path + RemoteLOCAL.UNPACKED_DIR_SUFFIX ) assert dvc_repo.push() == 1 shutil.rmtree(dvc_repo.cache.local.cache_dir) shutil.rmtree(repo_dir.DATA_DIR) - assert not os.path.exists(verification_dir) + assert not os.path.exists(unpacked_dir) dvc_repo.pull("{}.dvc".format(repo_dir.DATA_DIR)) - assert os.path.exists(verification_dir) - trees_equal(repo_dir.DATA_DIR, verification_dir) + assert os.path.exists(unpacked_dir) + trees_equal(repo_dir.DATA_DIR, unpacked_dir) + + +def test_should_not_fail_add_on_unpacked_dir_creation_exception( + dvc_repo, repo_dir +): + + with mock.patch.object( + dvc.remote.local.RemoteLOCAL, + "_create_unpacked_dir", + side_effect=Exception, + ): + ret = main(["add", repo_dir.DATA_DIR]) + assert ret == 0 diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py index 93375e801..a58fcbd6f 100644 --- a/tests/func/test_repro.py +++ b/tests/func/test_repro.py @@ -1044,7 +1044,6 @@ class TestReproExternalHDFS(TestReproExternalBase): self.assertEqual(p.returncode, 0) -@pytest.mark.skip @flaky(max_runs=3, min_passes=1) class TestReproExternalSSH(TestReproExternalBase): _dir = None From cfbd381d42e712afbd7e2d8baa17bb988cbc1e97 Mon Sep 17 00:00:00 2001 From: pawel Date: Sat, 8 Jun 2019 15:17:49 +0200 Subject: [PATCH 4/6] remote: create unpacked dir after status check if needed --- dvc/remote/base.py | 10 ++++++---- dvc/remote/local/__init__.py | 31 +++++++++++++++++-------------- tests/func/remote/test_local.py | 25 +++++++++++++++++++++---- 3 files changed, 44 insertions(+), 22 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index ea620eeed..541f6c6ed 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -372,7 +372,6 @@ class RemoteBASE(object): self.state.save_link(path_info) self.state.save(cache_info, checksum) self.state.save(path_info, checksum) - return dir_info def is_empty(self, path_info): return False @@ -543,10 +542,11 @@ class RemoteBASE(object): return False for entry in self.get_dir_cache(checksum): - checksum = entry[self.PARAM_CHECKSUM] - if self.changed_cache_file(checksum): + entry_checksum = entry[self.PARAM_CHECKSUM] + if self.changed_cache_file(entry_checksum): return True + self._update_unpacked_dir(checksum) return False def changed_cache(self, checksum): @@ -639,7 +639,6 @@ class RemoteBASE(object): self.state.save_link(path_info) self.state.save(path_info, checksum) - return dir_info def _remove_redundant_files(self, path_info, dir_info, force): existing_files = set( @@ -714,3 +713,6 @@ class RemoteBASE(object): def _changed_unpacked_dir(self, checksum): return True + + def _update_unpacked_dir(self, checksum): + pass diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index 038863adc..c65a5987c 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -237,7 +237,8 @@ class RemoteLOCAL(RemoteBASE): if path_info.scheme != "local": raise NotImplementedError - remove(path_info.path) + if self.exists(path_info): + remove(path_info.path) def move(self, from_info, to_info): if from_info.scheme != "local" or to_info.scheme != "local": @@ -605,24 +606,28 @@ class RemoteLOCAL(RemoteBASE): return unpacked_dir_info - def _update_unpacked_dir(self, checksum, dir_info): + def _path_info_changed(self, path_info): + if self.exists(path_info) and self.state.get(path_info): + return False + return True + + def _update_unpacked_dir(self, checksum): unpacked_dir_info = self._get_unpacked_dir_path_info(checksum) - if self.exists(unpacked_dir_info): - if self.state.get(unpacked_dir_info): - return + if not self._path_info_changed(unpacked_dir_info): + return - shutil.rmtree(unpacked_dir_info.path) + self.remove(unpacked_dir_info) try: + dir_info = self.get_dir_cache(checksum) self._create_unpacked_dir(checksum, dir_info, unpacked_dir_info) except Exception: logger.warning( "Could not create '{}'".format(unpacked_dir_info.path) ) - if self.exists(unpacked_dir_info): - self.remove(unpacked_dir_info) + self.remove(unpacked_dir_info) def _create_unpacked_dir(self, checksum, dir_info, unpacked_dir_info): self.makedirs(unpacked_dir_info) @@ -645,18 +650,16 @@ class RemoteLOCAL(RemoteBASE): return not self.state.get(status_unpacked_dir_info) def _save_dir(self, path_info, checksum): - dir_info = super(RemoteLOCAL, self)._save_dir(path_info, checksum) - self._update_unpacked_dir(checksum, dir_info) - return dir_info + super(RemoteLOCAL, self)._save_dir(path_info, checksum) + self._update_unpacked_dir(checksum) def _checkout_dir( self, path_info, checksum, force, progress_callback=None ): - dir_info = super(RemoteLOCAL, self)._checkout_dir( + super(RemoteLOCAL, self)._checkout_dir( path_info, checksum, force, progress_callback ) - self._update_unpacked_dir(checksum, dir_info) - return dir_info + self._update_unpacked_dir(checksum) def extract_used_local_checksums(self, cinfos): used = super(RemoteLOCAL, self).extract_used_local_checksums(cinfos) diff --git a/tests/func/remote/test_local.py b/tests/func/remote/test_local.py index 2cbbbadd0..488abcb72 100644 --- a/tests/func/remote/test_local.py +++ b/tests/func/remote/test_local.py @@ -15,8 +15,8 @@ from tests.utils import trees_equal def test_should_create_cache_status_check_dir_on_add(dvc_repo, repo_dir): stages = dvc_repo.add(repo_dir.DATA_DIR) - unpacked_dir = ( - stages[0].outs[0].cache_path + RemoteLOCAL.UNPACKED_DIR_SUFFIX + unpacked_dir = RemoteLOCAL._append_unpacked_suffix( + stages[0].outs[0].cache_path ) assert os.path.exists(unpacked_dir) @@ -42,8 +42,9 @@ def test_should_create_cache_status_check_dir_on_checkout(dvc_repo, repo_dir): dvc_repo = Repo(dvc_repo.dvc_dir) stages = dvc_repo.add(repo_dir.DATA_DIR) - unpacked_dir = ( - stages[0].outs[0].cache_path + RemoteLOCAL.UNPACKED_DIR_SUFFIX + + unpacked_dir = RemoteLOCAL._append_unpacked_suffix( + stages[0].outs[0].cache_path ) assert dvc_repo.push() == 1 @@ -69,3 +70,19 @@ def test_should_not_fail_add_on_unpacked_dir_creation_exception( ): ret = main(["add", repo_dir.DATA_DIR]) assert ret == 0 + + +def test_should_create_unpacked_dir_on_status_check(dvc_repo, repo_dir): + stages = dvc_repo.add(repo_dir.DATA_DIR) + assert len(stages) == 1 + + unpacked_dir = RemoteLOCAL._append_unpacked_suffix( + stages[0].outs[0].cache_path + ) + + assert os.path.exists(unpacked_dir) + shutil.rmtree(unpacked_dir) + assert not os.path.exists(unpacked_dir) + + assert dvc_repo.status("{}.dvc".format(repo_dir.DATA_DIR)) == {} + assert os.path.exists(unpacked_dir) From ac5ca4656a451a6974d1f72853c363eb0f8d1f2f Mon Sep 17 00:00:00 2001 From: pawel Date: Mon, 10 Jun 2019 14:42:41 +0200 Subject: [PATCH 5/6] remote: progress callback for unpacked dir creation --- dvc/remote/base.py | 8 +++-- dvc/remote/local/__init__.py | 52 ++++++++++++++----------------- tests/func/remote/test_local.py | 54 --------------------------------- tests/func/test_checkout.py | 6 +++- 4 files changed, 33 insertions(+), 87 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 541f6c6ed..0074351f8 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -705,14 +705,18 @@ class RemoteBASE(object): def unprotect(path_info): pass + def _get_unpacked_dir_names(self, checksums): + return set() + def extract_used_local_checksums(self, cinfos): from dvc.remote import RemoteLOCAL used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]} - return used + unpacked = self._get_unpacked_dir_names(used) + return used | unpacked def _changed_unpacked_dir(self, checksum): return True - def _update_unpacked_dir(self, checksum): + def _update_unpacked_dir(self, checksum, progress_callback=None): pass diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index c65a5987c..3e50591d2 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -35,7 +35,7 @@ from dvc.utils import ( ) from dvc.config import Config from dvc.exceptions import DvcException -from dvc.progress import progress +from dvc.progress import progress, ProgressCallback from concurrent.futures import ThreadPoolExecutor @@ -144,10 +144,11 @@ class RemoteLOCAL(RemoteBASE): return if not link_type: - self._default_link(from_info, to_info) + link_types = self.cache_types else: - link_method = self._get_link_method(link_type) - self._link(from_info, to_info, link_method) + link_types = [link_type] + + self._try_links(from_info, to_info, link_types) @classmethod def _get_link_method(cls, link_type): @@ -165,6 +166,7 @@ class RemoteLOCAL(RemoteBASE): ) else: link_method(from_info.path, to_info.path) + if self.protected: self.protect(to_info) @@ -178,18 +180,18 @@ class RemoteLOCAL(RemoteBASE): logger.debug(msg) @slow_link_guard - def _default_link(self, from_info, to_info): - i = len(self.cache_types) + def _try_links(self, from_info, to_info, link_types): + i = len(link_types) while i > 0: - link_method = self._get_link_method(self.cache_types[0]) + link_method = self._get_link_method(link_types[0]) try: self._link(from_info, to_info, link_method) return except DvcException as exc: msg = "Cache type '{}' is not supported: {}" - logger.debug(msg.format(self.cache_types[0], str(exc))) - del self.cache_types[0] + logger.debug(msg.format(link_types[0], str(exc))) + del link_types[0] i -= 1 raise DvcException("no possible cache types left to try out.") @@ -622,7 +624,7 @@ class RemoteLOCAL(RemoteBASE): try: dir_info = self.get_dir_cache(checksum) self._create_unpacked_dir(checksum, dir_info, unpacked_dir_info) - except Exception: + except DvcException: logger.warning( "Could not create '{}'".format(unpacked_dir_info.path) ) @@ -630,8 +632,9 @@ class RemoteLOCAL(RemoteBASE): self.remove(unpacked_dir_info) def _create_unpacked_dir(self, checksum, dir_info, unpacked_dir_info): + progress_callback = ProgressCallback(len(dir_info)) self.makedirs(unpacked_dir_info) - for entry in dir_info: + for index, entry in enumerate(dir_info): entry_cache_info = self.checksum_to_path_info( entry[self.PARAM_CHECKSUM] ) @@ -642,6 +645,8 @@ class RemoteLOCAL(RemoteBASE): unpacked_dir_info.path, relpath ) self.link(entry_cache_info, unpacked_entry_info, "hardlink") + + progress_callback.update("Creating unpacked dir.") self.state.save(unpacked_dir_info, checksum) def _changed_unpacked_dir(self, checksum): @@ -649,22 +654,9 @@ class RemoteLOCAL(RemoteBASE): return not self.state.get(status_unpacked_dir_info) - def _save_dir(self, path_info, checksum): - super(RemoteLOCAL, self)._save_dir(path_info, checksum) - self._update_unpacked_dir(checksum) - - def _checkout_dir( - self, path_info, checksum, force, progress_callback=None - ): - super(RemoteLOCAL, self)._checkout_dir( - path_info, checksum, force, progress_callback - ) - self._update_unpacked_dir(checksum) - - def extract_used_local_checksums(self, cinfos): - used = super(RemoteLOCAL, self).extract_used_local_checksums(cinfos) - unpacked_dir = set() - for checksum in used: - if self.is_dir_checksum(checksum): - unpacked_dir.add(self._append_unpacked_suffix(checksum)) - return used | unpacked_dir + def _get_unpacked_dir_names(self, checksums): + unpacked = set() + for c in checksums: + if self.is_dir_checksum(c): + unpacked.add(self._append_unpacked_suffix(c)) + return unpacked diff --git a/tests/func/remote/test_local.py b/tests/func/remote/test_local.py index 488abcb72..35703b423 100644 --- a/tests/func/remote/test_local.py +++ b/tests/func/remote/test_local.py @@ -1,62 +1,10 @@ import os -import shutil import mock import dvc from dvc.main import main from dvc.remote.local import RemoteLOCAL -from dvc.repo import Repo -from dvc.utils import file_md5 -from dvc.utils.fs import get_inode -from tests.utils import trees_equal - - -def test_should_create_cache_status_check_dir_on_add(dvc_repo, repo_dir): - - stages = dvc_repo.add(repo_dir.DATA_DIR) - unpacked_dir = RemoteLOCAL._append_unpacked_suffix( - stages[0].outs[0].cache_path - ) - - assert os.path.exists(unpacked_dir) - - trees_equal(repo_dir.DATA_DIR, unpacked_dir) - data_cache_path = dvc_repo.cache.local.get(file_md5(repo_dir.DATA)[0]) - data_sub_cache_path = dvc_repo.cache.local.get( - file_md5(repo_dir.DATA_SUB)[0] - ) - assert get_inode(os.path.join(unpacked_dir, "data")) == get_inode( - data_cache_path - ) - assert get_inode( - os.path.join(unpacked_dir, "data_sub_dir", "data_sub") - ) == get_inode(data_sub_cache_path) - - -def test_should_create_cache_status_check_dir_on_checkout(dvc_repo, repo_dir): - local_storage = repo_dir.mkdtemp() - ret = main(["remote", "add", "-d", "storage", local_storage]) - assert ret == 0 - - dvc_repo = Repo(dvc_repo.dvc_dir) - - stages = dvc_repo.add(repo_dir.DATA_DIR) - - unpacked_dir = RemoteLOCAL._append_unpacked_suffix( - stages[0].outs[0].cache_path - ) - - assert dvc_repo.push() == 1 - - shutil.rmtree(dvc_repo.cache.local.cache_dir) - shutil.rmtree(repo_dir.DATA_DIR) - assert not os.path.exists(unpacked_dir) - - dvc_repo.pull("{}.dvc".format(repo_dir.DATA_DIR)) - - assert os.path.exists(unpacked_dir) - trees_equal(repo_dir.DATA_DIR, unpacked_dir) def test_should_not_fail_add_on_unpacked_dir_creation_exception( @@ -80,8 +28,6 @@ def test_should_create_unpacked_dir_on_status_check(dvc_repo, repo_dir): stages[0].outs[0].cache_path ) - assert os.path.exists(unpacked_dir) - shutil.rmtree(unpacked_dir) assert not os.path.exists(unpacked_dir) assert dvc_repo.status("{}.dvc".format(repo_dir.DATA_DIR)) == {} diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py index b9ed9d264..2ac817a9f 100644 --- a/tests/func/test_checkout.py +++ b/tests/func/test_checkout.py @@ -422,7 +422,11 @@ class TestCheckoutShouldHaveSelfClearingProgressBar(TestDvc): self.write_args = [w_c[1][0] for w_c in write_calls] pattern = re.compile(".*\\[.{30}\\].*%.*") - progress_bars = [arg for arg in self.write_args if pattern.match(arg)] + progress_bars = [ + arg + for arg in self.write_args + if pattern.match(arg) and "unpacked" not in arg + ] update_bars = progress_bars[:-1] finish_bar = progress_bars[-1] From fe3f83ee051d7bb48dce43266116c728a0bc47cf Mon Sep 17 00:00:00 2001 From: pawel Date: Tue, 11 Jun 2019 10:15:07 +0200 Subject: [PATCH 6/6] remote: local: test: refactor unpacked dir tests --- tests/func/remote/test_local.py | 88 ++++++++++++++++++++++++++------- 1 file changed, 69 insertions(+), 19 deletions(-) diff --git a/tests/func/remote/test_local.py b/tests/func/remote/test_local.py index 35703b423..e26b2061f 100644 --- a/tests/func/remote/test_local.py +++ b/tests/func/remote/test_local.py @@ -1,34 +1,84 @@ import os import mock +import pytest import dvc -from dvc.main import main +from dvc.exceptions import DvcException from dvc.remote.local import RemoteLOCAL +from tests.utils import trees_equal -def test_should_not_fail_add_on_unpacked_dir_creation_exception( - dvc_repo, repo_dir -): +class TestUnpackedDirStatusOptimization(object): + @pytest.fixture(autouse=True) + def setUp(self, dvc_repo, repo_dir): + stages = dvc_repo.add(repo_dir.DATA_DIR) + assert len(stages) == 1 + self.dir_out = stages[0].outs[0] + self.unpacked_dir = RemoteLOCAL._append_unpacked_suffix( + self.dir_out.cache_path + ) - with mock.patch.object( - dvc.remote.local.RemoteLOCAL, - "_create_unpacked_dir", - side_effect=Exception, + def test_should_not_fail_add_on_unpacked_dir_creation_exception( + self, dvc_repo ): - ret = main(["add", repo_dir.DATA_DIR]) - assert ret == 0 + with mock.patch.object( + dvc.remote.local.RemoteLOCAL, + "_create_unpacked_dir", + side_effect=self._raise_dvc_exception, + ) as unpacked_create_spy, dvc_repo.state: + assert not dvc_repo.cache.local.changed_cache( + self.dir_out.checksum + ) + assert unpacked_create_spy.call_count == 1 + def test_should_remove_unpacked_dir_on_create_exception(self, dvc_repo): + # artificial unpacked dir for test purpose + os.makedirs(self.unpacked_dir) + assert os.path.exists(self.unpacked_dir) -def test_should_create_unpacked_dir_on_status_check(dvc_repo, repo_dir): - stages = dvc_repo.add(repo_dir.DATA_DIR) - assert len(stages) == 1 + with mock.patch.object( + dvc.remote.local.RemoteLOCAL, + "_create_unpacked_dir", + side_effect=self._raise_dvc_exception, + ), dvc_repo.state: + assert not dvc_repo.cache.local.changed_cache( + self.dir_out.checksum + ) - unpacked_dir = RemoteLOCAL._append_unpacked_suffix( - stages[0].outs[0].cache_path - ) + assert not os.path.exists(self.unpacked_dir) - assert not os.path.exists(unpacked_dir) + def test_should_create_unpacked_dir_on_status_check( + self, dvc_repo, repo_dir + ): + assert not os.path.exists(self.unpacked_dir) - assert dvc_repo.status("{}.dvc".format(repo_dir.DATA_DIR)) == {} - assert os.path.exists(unpacked_dir) + with dvc_repo.state: + assert not dvc_repo.cache.local.changed_cache( + self.dir_out.checksum + ) + + assert os.path.exists(self.unpacked_dir) + + trees_equal(repo_dir.DATA_DIR, self.unpacked_dir) + + def test_should_proceed_with_full_check_on_modified_cache(self, dvc_repo): + + some_file_md5 = self.dir_out.dir_cache[0]["md5"] + assert not os.path.exists(self.unpacked_dir) + + with dvc_repo.state: + assert not dvc_repo.cache.local.changed_cache( + self.dir_out.checksum + ) + assert os.path.exists(self.unpacked_dir) + + with open(dvc_repo.cache.local.get(some_file_md5), "a") as fobj: + fobj.write("string") + + with dvc_repo.state: + assert dvc_repo.cache.local.changed_cache(self.dir_out.checksum) + + @staticmethod + def _raise_dvc_exception(*args): + raise DvcException("msg")