remote: local: add cache status check hardlink optimization

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

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)

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)

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))