Repository: dataversioncontrol/dvc
Pull request #639 (2018-04-09 22:00:19)

cloud: cmp_checksum improvements


Patch:
diff --git a/dvc/cloud/aws.py b/dvc/cloud/aws.py
index e14ed71c4..fa2e81ed7 100644
--- a/dvc/cloud/aws.py
+++ b/dvc/cloud/aws.py
@@ -17,7 +17,6 @@ from dvc.logger import Logger
 from dvc.progress import progress
 from dvc.cloud.credentials_aws import AWSCredentials
 from dvc.cloud.base import DataCloudError, DataCloudBase
-from dvc.utils import file_md5
 
 
 def sizeof_fmt(num, suffix='B'):
@@ -98,27 +97,6 @@ class DataCloudAWS(DataCloudBase):
         if self.bucket is None:
             raise DataCloudError('Storage path {} is not setup correctly'.format(self.storage_bucket))
 
-    @staticmethod
-    def _cmp_checksum(key, fname):
-        """
-        Verify local and remote checksums. Used 'dvc-md5' metadata if supported
-        or falls back to etag.
-        """
-
-        # Calling get_key() once more to avoid empty metadata.
-        # See http://blog.bidiuk.com/2014/02/get-amazon-s3-metadata-in-python-using-boto/
-        key = key.bucket.get_key(key.name)
-        md5_cloud = key.metadata.get('dvc-md5', None)
-        md5_local = file_md5(fname)[0]
-
-        if md5_cloud == None:
-            md5_cloud = key.etag[1:-1]
-
-        if md5_cloud == md5_local:
-            return True
-
-        return False
-
     @staticmethod
     def _upload_tracker(fname):
         """
@@ -221,10 +199,7 @@ class DataCloudAWS(DataCloudBase):
         """
         Create multipart upload and save info to tracker file.
         """
-        # AWS doesn't provide easilly accessible md5 for multipart
-        # objects, so we have to store our own md5 sum to use later.
-        metadata = {'dvc-md5' : str(file_md5(fname)[0])}
-        multipart = key.bucket.initiate_multipart_upload(key.name, metadata=metadata)
+        multipart = key.bucket.initiate_multipart_upload(key.name)
         self._write_upload_tracker(fname, multipart.id)
         return multipart
 
diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py
index 3e41a5a2b..b23e19728 100644
--- a/dvc/cloud/base.py
+++ b/dvc/cloud/base.py
@@ -1,7 +1,6 @@
 import os
 import re
 import tempfile
-from checksumdir import dirhash
 
 from dvc.logger import Logger
 from dvc.exceptions import DvcException
@@ -31,6 +30,14 @@ class DataCloudError(DvcException):
         super(DataCloudError, self).__init__('Data sync error: {}'.format(msg))
 
 
+class CloudSettings(object):
+    def __init__(self, cache=None, state=None, global_storage_path=None, cloud_config=None):
+        self.cache = cache
+        self.state = state
+        self.cloud_config = cloud_config
+        self.global_storage_path = global_storage_path
+
+
 class DataCloudBase(object):
     """ Base class for DataCloud """
 
@@ -146,6 +153,13 @@ class DataCloudBase(object):
 
         return ret
 
+    def _cmp_checksum(self, blob, fname):
+        md5 = self._cloud_settings.cache.path_to_md5(fname)
+        if self._cloud_settings.state.changed(fname, md5=md5):
+            return False
+
+        return True
+
     def push(self, path):
         key = self._get_key(path)
         if key:
diff --git a/dvc/cloud/data_cloud.py b/dvc/cloud/data_cloud.py
index 319890aa4..c88679934 100644
--- a/dvc/cloud/data_cloud.py
+++ b/dvc/cloud/data_cloud.py
@@ -1,7 +1,6 @@
 import re
 from multiprocessing.pool import ThreadPool
 
-from dvc.cloud.instance_manager import CloudSettings
 from dvc.logger import Logger
 from dvc.exceptions import DvcException
 from dvc.config import Config, ConfigError
@@ -11,7 +10,7 @@ from dvc.cloud.aws import DataCloudAWS
 from dvc.cloud.gcp import DataCloudGCP
 from dvc.cloud.ssh import DataCloudSSH
 from dvc.cloud.local import DataCloudLOCAL
-from dvc.cloud.base import DataCloudBase
+from dvc.cloud.base import DataCloudBase, CloudSettings
 
 
 class DataCloud(object):
@@ -24,9 +23,10 @@ class DataCloud(object):
         'local' : DataCloudLOCAL,
     }
 
-    def __init__(self, cache, config):
+    def __init__(self, cache=None, config=None, state=None):
         self._cache = cache
         self._config = config
+        self._state = state
 
         remote = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_REMOTE, '')
         if remote == '':
@@ -58,7 +58,7 @@ class DataCloud(object):
         if not cloud_type:
             raise ConfigError("Unsupported url '{}'".format(url))
 
-        return self._init_cloud(self._cache, cloud_config, cloud_type)
+        return self._init_cloud(cloud_config, cloud_type)
 
     def __init__compat(self):
         cloud_name = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, '').strip().lower()
@@ -74,29 +74,22 @@ class DataCloud(object):
         if not cloud_config:
             raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_name)
 
-        return self._init_cloud(self._cache, cloud_config, cloud_type)
+        return self._init_cloud(cloud_config, cloud_type)
 
-    def _init_cloud(self, cache, cloud_config, cloud_type):
-        cloud_settings = self.get_cloud_settings(cache,
-                                                 self._config,
-                                                 cloud_config)
+    def _init_cloud(self, cloud_config, cloud_type):
+        global_storage_path = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_STORAGEPATH, None)
+        if global_storage_path:
+            Logger.warn('Using obsoleted config format. Consider updating.')
+
+        cloud_settings = CloudSettings(cache=self._cache,
+                                       state=self._state,
+                                       global_storage_path=global_storage_path,
+                                       cloud_config=cloud_config)
 
         cloud = cloud_type(cloud_settings)
         cloud.sanity_check()
         return cloud
 
-    @staticmethod
-    def get_cloud_settings(cache, config, cloud_config):
-        """
-        Obtain cloud settings from config.
-        """
-        global_storage_path = config[Config.SECTION_CORE].get(Config.SECTION_CORE_STORAGEPATH, None)
-        if global_storage_path:
-            Logger.warn('Using obsoleted config format. Consider updating.')
-
-        cloud_settings = CloudSettings(cache, global_storage_path, cloud_config)
-        return cloud_settings
-
     def _collect(self, cloud, targets, jobs, local):
         collected = set()
         pool = ThreadPool(processes=jobs)
diff --git a/dvc/cloud/gcp.py b/dvc/cloud/gcp.py
index 28d345dfe..563e6f720 100644
--- a/dvc/cloud/gcp.py
+++ b/dvc/cloud/gcp.py
@@ -6,7 +6,6 @@ from google.cloud import storage as gc
 from dvc.logger import Logger
 from dvc.config import Config, ConfigError
 from dvc.cloud.base import DataCloudError, DataCloudBase
-from dvc.utils import file_md5
 from dvc.progress import progress
 
 
@@ -28,19 +27,6 @@ class DataCloudGCP(DataCloudBase):
             raise DataCloudError('sync up: google cloud bucket {} '
                                  'doesn\'t exist'.format(self.storage_bucket))
 
-    @staticmethod
-    def _cmp_checksum(blob, fname):
-        """
-        Verify local and remote checksums.
-        """
-        md5 = file_md5(fname)[1]
-        b64_encoded_md5 = base64.b64encode(md5).decode() if md5 else None
-
-        if blob.md5_hash == b64_encoded_md5:
-            return True
-
-        return False
-
     def _pull_key(self, key, path, no_progress_bar=False):
         self._makedirs(path)
 
diff --git a/dvc/cloud/instance_manager.py b/dvc/cloud/instance_manager.py
deleted file mode 100644
index e8c1db151..000000000
--- a/dvc/cloud/instance_manager.py
+++ /dev/null
@@ -1,5 +0,0 @@
-class CloudSettings(object):
-    def __init__(self, cache, global_storage_path, cloud_config):
-        self.cache = cache
-        self.cloud_config = cloud_config
-        self.global_storage_path = global_storage_path
diff --git a/dvc/cloud/local.py b/dvc/cloud/local.py
index 08d3b7135..0a3130596 100644
--- a/dvc/cloud/local.py
+++ b/dvc/cloud/local.py
@@ -25,9 +25,6 @@ class DataCloudLOCAL(DataCloudBase):
     def cache_file_key(self, path):
         return os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.cache_dir)
 
-    def _cmp_checksum(self, key, path):
-        return filecmp.cmp(path, key.path)
-
     def _get_key(self, path):
         key_name = self.cache_file_key(path)
         key = LocalKey(self.storage_path, key_name)
diff --git a/dvc/cloud/ssh.py b/dvc/cloud/ssh.py
index 680317af9..80f7eed89 100644
--- a/dvc/cloud/ssh.py
+++ b/dvc/cloud/ssh.py
@@ -64,16 +64,6 @@ class DataCloudSSH(DataCloudBase):
                 raise
         return False
 
-    def _cmp_checksum(self, key, path):
-        #FIXME actually compare md5
-        path_exists = os.path.exists(path)
-        key_exists = self._isfile_remote(key.path)
-
-        if path_exists and key_exists:
-            return True
-
-        return False
-
     def _get_key(self, path):
         key_name = self.cache_file_key(path)
         key = SSHKey(self.path, key_name)
diff --git a/dvc/output.py b/dvc/output.py
index e81205aa5..c2d032a98 100644
--- a/dvc/output.py
+++ b/dvc/output.py
@@ -5,7 +5,6 @@ import shutil
 import schema
 import posixpath
 import ntpath
-from checksumdir import dirhash
 
 from dvc.system import System
 from dvc.utils import file_md5
@@ -195,8 +194,8 @@ class Output(Dependency):
         return [cls.loads(project, x, use_cache=use_cache, cwd=cwd) for x in s_list]
 
     def _changed_cache(self, cache):
-        md5 = self.project.state.update(cache)
-        if md5 != self.project.cache.path_to_md5(cache):
+        md5 = self.project.cache.path_to_md5(cache)
+        if self.project.state.changed(cache, md5=md5):
             self.project.logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache)))
             os.unlink(cache)
             return True
@@ -213,24 +212,12 @@ class Output(Dependency):
         return True
 
     def _changed_dir(self):
-        if not os.path.isdir(self.path) or not os.path.isfile(self.cache):
-            return True
-
-        dir_info = self._collect_dir() # slow!
-        dir_info_cached = self.load_dir_cache(self.cache) # slow. why?
-
-        if not self.are_dir_info_equal(dir_info, dir_info_cached):
-            return True
-
-        return False
-
-    @staticmethod
-    def are_dir_info_equal(dir_info1, dir_info2):
-        return Output.dir_info_dict(dir_info1) == Output.dir_info_dict(dir_info2)
+        if os.path.isdir(self.path) and \
+           os.path.isfile(self.cache) and \
+           not self._changed_cache(self.cache):
+            return False
 
-    @staticmethod
-    def dir_info_dict(dir_info):
-        return {i['relpath']: i['md5'] for i in dir_info}
+        return True
 
     def changed(self):
         if not self.use_cache:
@@ -330,24 +317,9 @@ class Output(Dependency):
             path = os.path.join(self.path, relpath)
             self.hardlink(cache, path)
 
-    def _collect_dir(self):
-        dir_info = []
-
-        for root, dirs, files in os.walk(self.path):
-            for fname in files:
-                path = os.path.join(root, fname)
-                relpath = os.path.relpath(path, self.path)
-
-                md5 = self.project.state.update(path, dump=False)
-                dir_info.append({self.PARAM_RELPATH: relpath, self.PARAM_MD5: md5})
-
-        self.project.state.dump()
-
-        return dir_info
-
     def _save_dir(self):
         dname = os.path.dirname(self.cache)
-        dir_info = self._collect_dir()
+        dir_info = self.project.state.collect_dir(self.path)
 
         for entry in dir_info:
             md5 = entry[self.PARAM_MD5]
@@ -365,7 +337,7 @@ class Output(Dependency):
             os.makedirs(dname)
 
         with open(self.cache, 'w+') as fd:
-            json.dump(dir_info, fd)
+            json.dump(dir_info, fd, sort_keys=True)
 
     def save(self):
         super(Output, self).save()
diff --git a/dvc/project.py b/dvc/project.py
index 71f943f84..1537a411a 100644
--- a/dvc/project.py
+++ b/dvc/project.py
@@ -41,7 +41,7 @@ class Project(object):
         self.state = State(self.root_dir, self.dvc_dir)
         self.config = Config(self.dvc_dir)
         self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None))
-        self.cloud = DataCloud(self.cache, self.config._config)
+        self.cloud = DataCloud(cache=self.cache, state=self.state, config=self.config._config)
 
     @staticmethod
     def init(root_dir=os.curdir):
diff --git a/dvc/state.py b/dvc/state.py
index 38011873b..a1df5015a 100644
--- a/dvc/state.py
+++ b/dvc/state.py
@@ -1,11 +1,11 @@
 import os
 import json
 import nanotime
-from checksumdir import dirhash
+import threading
 
 from dvc.system import System
 from dvc.output import Output
-from dvc.utils import file_md5
+from dvc.utils import file_md5, bytes_md5
 from dvc.exceptions import DvcException
 
 
@@ -46,19 +46,43 @@ class State(object):
         self.dvc_dir = dvc_dir
         self.state_file = os.path.join(dvc_dir, self.STATE_FILE)
         self._db = self.load()
+        self._lock = threading.Lock()
 
     @staticmethod
     def init(root_dir, dvc_dir):
         return State(root_dir, dvc_dir)
 
+    def collect_dir(self, dname):
+        dir_info = []
+
+        for root, dirs, files in os.walk(dname):
+            for fname in files:
+                path = os.path.join(root, fname)
+                relpath = os.path.relpath(path, dname)
+
+                md5 = self.update(path, dump=False)
+                dir_info.append({Output.PARAM_RELPATH: relpath, Output.PARAM_MD5: md5})
+
+        self.dump()
+
+        return dir_info
+
     def compute_md5(self, path):
         if os.path.isdir(path):
-            return dirhash(path, hashfunc='md5') + Output.MD5_DIR_SUFFIX
+            dir_info = self.collect_dir(path)
+            byts = json.dumps(dir_info, sort_keys=True).encode('utf-8')
+            md5 = bytes_md5(byts)
+            return md5 + Output.MD5_DIR_SUFFIX
         else:
             return file_md5(path)[0]
 
     def changed(self, path, md5):
-        return self.update(path) != md5
+        actual = self.update(path)
+
+        if not md5 or not actual:
+            return True
+
+        return actual.split('.')[0] != md5.split('.')[0]
 
     def load(self):
         if not os.path.isfile(self.state_file):
@@ -90,15 +114,18 @@ class State(object):
         md5 = self.compute_md5(path)
         state = StateEntry(md5, mtime)
         d = state.dumpd()
-        self._db[inode] = d
 
-        if dump:
-            self.dump()
+        with self._lock:
+            self._db[inode] = d
+
+            if dump:
+                self.dump()
 
         return md5
 
     def _get(self, inode, mtime):
-        d = self._db.get(inode, None)
+        with self._lock:
+            d = self._db.get(inode, None)
         if not d:
             return None
 
diff --git a/dvc/utils.py b/dvc/utils.py
index 56270352f..395acc52f 100644
--- a/dvc/utils.py
+++ b/dvc/utils.py
@@ -28,6 +28,12 @@ def file_md5(fname):
         return (None, None)
 
 
+def bytes_md5(byts):
+    hasher = hashlib.md5()
+    hasher.update(byts)
+    return hasher.hexdigest()
+
+
 def cached_property(func):
     '''A decorator for caching properties in classes.'''
     def get(self):
diff --git a/requirements.txt b/requirements.txt
index 7914e1987..286627bef 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -11,7 +11,6 @@ networkx>=1.11
 pyyaml>=3.12
 gitpython>=2.1.8
 ntfsutils>=0.1.4
-checksumdir>=1.1.4
 setuptools>=34.0.0
 nanotime>=0.5.2
 pyasn1>=0.4.1
diff --git a/setup.py b/setup.py
index 2e9289c2f..0a59e02e6 100644
--- a/setup.py
+++ b/setup.py
@@ -17,7 +17,6 @@ install_requires = [
     "pyyaml>=3.12",
     "gitpython>=2.1.8",
     "ntfsutils>=0.1.4",
-    "checksumdir>=1.1.4",
     "setuptools>=34.0.0",
     "nanotime>=0.5.2",
     "pyasn1>=0.4.1",


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
index bc0468a6d..3926fc5ff 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -3,3 +3,6 @@ import os
 if os.name == 'nt':
     import win32file
     win32file._setmaxstdio(2048)
+else:
+    import resource
+    resource.setrlimit(resource.RLIMIT_NOFILE, (2048, 2048))
diff --git a/tests/test_add.py b/tests/test_add.py
index ad025d12e..01b5aa665 100644
--- a/tests/test_add.py
+++ b/tests/test_add.py
@@ -2,7 +2,6 @@ import os
 import stat
 import shutil
 import filecmp
-from checksumdir import dirhash
 
 from dvc.main import main
 from dvc.utils import file_md5
diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py
index accc5c945..f12195fb4 100644
--- a/tests/test_data_cloud.py
+++ b/tests/test_data_cloud.py
@@ -8,8 +8,7 @@ import tempfile
 from dvc.main import main
 from dvc.config import Config, ConfigError
 from dvc.cloud.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudGCP
-from dvc.cloud.base import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED
-from dvc.cloud.instance_manager import CloudSettings
+from dvc.cloud.base import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED, CloudSettings
 
 from tests.basic_env import TestDvc
 
@@ -95,7 +94,7 @@ def get_gcp_url():
 
 class TestDataCloud(TestDvc):
     def _test_cloud(self, config, cl):
-        cloud = DataCloud(self.dvc.cache, config)
+        cloud = DataCloud(cache=self.dvc.cache, state=self.dvc.state, config=config)
         self.assertIsInstance(cloud._cloud, cl)
 
     def test(self):
@@ -109,7 +108,7 @@ class TestDataCloud(TestDvc):
         with self.assertRaises(ConfigError) as cx:
             config = TEST_CONFIG
             config[TEST_SECTION][Config.SECTION_REMOTE_URL] = 'notsupportedscheme://a/b'
-            DataCloud(self.dvc.cache, config)
+            DataCloud(cache=self.dvc.cache, state=self.dvc.state, config=config)
 
 
 class TestDataCloudBase(TestDvc):
@@ -191,7 +190,9 @@ class TestDataCloudAWS(TestDataCloudBase):
         config = TEST_CONFIG
         config[TEST_SECTION][Config.SECTION_REMOTE_URL] = repo
         config[TEST_SECTION][Config.SECTION_AWS_REGION] = TEST_AWS_REPO_REGION
-        cloud_settings = CloudSettings(self.dvc.cache, None, config[TEST_SECTION])
+        cloud_settings = CloudSettings(cache=self.dvc.cache,
+                                       state=self.dvc.state,
+                                       cloud_config=config[TEST_SECTION])
         self.cloud = DataCloudAWS(cloud_settings)
 
     def test(self):
@@ -209,7 +210,9 @@ class TestDataCloudGCP(TestDataCloudBase):
         # Setup cloud
         config = TEST_CONFIG
         config[TEST_SECTION][Config.SECTION_REMOTE_URL] = repo
-        cloud_settings = CloudSettings(self.dvc.cache, None, config[TEST_SECTION])
+        cloud_settings = CloudSettings(cache=self.dvc.cache,
+                                       state=self.dvc.state,
+                                       cloud_config=config[TEST_SECTION])
         self.cloud = DataCloudGCP(cloud_settings)
 
     def test(self):
@@ -223,7 +226,9 @@ class TestDataCloudLOCAL(TestDataCloudBase):
 
         config = TEST_CONFIG
         config[TEST_SECTION][Config.SECTION_REMOTE_URL] = self.dname
-        cloud_settings = CloudSettings(self.dvc.cache, None, config[TEST_SECTION])
+        cloud_settings = CloudSettings(cache=self.dvc.cache,
+                                       state=self.dvc.state,
+                                       cloud_config=config[TEST_SECTION])
         self.cloud = DataCloudLOCAL(cloud_settings)
 
     def test(self):