Repository: iterative/dvc
Pull request #849 (2018-07-02 15:56:35)

dvc: misc improvements
Signed-off-by: Ruslan Kuprieiev <kupruser@gmail.com>


Patch:
diff --git a/dvc/config.py b/dvc/config.py
index 88d0751f0..148c75e13 100644
--- a/dvc/config.py
+++ b/dvc/config.py
@@ -76,7 +76,6 @@ class Config(object):
     SECTION_AWS = 'aws'
     SECTION_AWS_STORAGEPATH = 'storagepath'
     SECTION_AWS_CREDENTIALPATH = 'credentialpath'
-    SECTION_AWS_ENDPOINT_URL = 'endpointurl'
     SECTION_AWS_REGION = 'region'
     SECTION_AWS_PROFILE = 'profile'
     SECTION_AWS_SCHEMA = {
@@ -84,7 +83,6 @@ class Config(object):
         schema.Optional(SECTION_AWS_REGION): str,
         schema.Optional(SECTION_AWS_PROFILE, default='default'): str,
         schema.Optional(SECTION_AWS_CREDENTIALPATH, default = ''): str,
-        schema.Optional(SECTION_AWS_ENDPOINT_URL, default=None): str,
     }
 
     # backward compatibility
@@ -113,7 +111,6 @@ class Config(object):
         schema.Optional(SECTION_AWS_REGION): str,
         schema.Optional(SECTION_AWS_PROFILE, default='default'): str,
         schema.Optional(SECTION_AWS_CREDENTIALPATH, default = ''): str,
-        schema.Optional(SECTION_AWS_ENDPOINT_URL, default=None): str,
         schema.Optional(SECTION_GCP_PROJECTNAME): str,
         schema.Optional(SECTION_CACHE_TYPE): SECTION_CACHE_TYPE_SCHEMA,
         schema.Optional(SECTION_REMOTE_USER): str,
diff --git a/dvc/output/local.py b/dvc/output/local.py
index 3c2627071..53fc11aab 100644
--- a/dvc/output/local.py
+++ b/dvc/output/local.py
@@ -41,13 +41,10 @@ class OutputLOCAL(DependencyLOCAL):
         return ret
 
     def changed(self):
-        if super(OutputLOCAL, self).changed():
-            return True
-
-        if self.use_cache and self.info != self.project.cache.local.save_info(self.path_info):
-            return True
+        if not self.use_cache:
+            return super(OutputLOCAL, self).changed()
 
-        return False
+        return self.info != self.project.cache.local.save_info(self.path_info)
 
     def checkout(self):
         if not self.use_cache:
@@ -61,15 +58,24 @@ class OutputLOCAL(DependencyLOCAL):
         self.project.cache.local.checkout(self.path_info, self.info)
 
     def save(self):
-        if not self.changed():
-            msg = 'Output \'{}\' didn\'t change. Skipping saving.'
+        if not self.use_cache:
+            super(OutputLOCAL, self).save()
+            msg = 'Output \'{}\' doesn\'t use cache. Skipping saving.'
             self.project.logger.debug(msg.format(self.rel_path))
             return
 
-        super(OutputLOCAL, self).save()
+        if not os.path.exists(self.path):
+            raise self.DoesNotExistError(self.rel_path)
 
-        if not self.use_cache:
-            msg = 'Output \'{}\' doesn\'t use cache. Skipping saving.'
+        if not os.path.isfile(self.path) and not os.path.isdir(self.path):
+            raise self.NotFileOrDirError(self.rel_path)
+
+        if (os.path.isfile(self.path) and os.path.getsize(self.path) == 0) or \
+           (os.path.isdir(self.path) and len(os.listdir(self.path)) == 0):
+            self.project.logger.warn("File/directory '{}' is empty.".format(self.rel_path))
+
+        if not self.changed():
+            msg = 'Output \'{}\' didn\'t change. Skipping saving.'
             self.project.logger.debug(msg.format(self.rel_path))
             return
 
diff --git a/dvc/project.py b/dvc/project.py
index 09106c2d0..b45ca1aee 100644
--- a/dvc/project.py
+++ b/dvc/project.py
@@ -10,7 +10,7 @@ from dvc.logger import Logger
 from dvc.exceptions import DvcException
 from dvc.stage import Stage
 from dvc.config import Config
-from dvc.state import LinkState
+from dvc.state import LinkState, State
 from dvc.lock import Lock
 from dvc.scm import SCM, Base
 from dvc.cache import Cache
@@ -50,7 +50,10 @@ class Project(object):
         self.config = Config(self.dvc_dir)
         self.scm = SCM(self.root_dir)
         self.lock = Lock(self.dvc_dir)
-        self.link_state = LinkState(self.root_dir, self.dvc_dir)
+        # NOTE: storing state and link_state in the repository itself to avoid
+        # any possible state corruption in 'shared cache dir' scenario.
+        self.state = State(self)
+        self.link_state = LinkState(self)
         self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None))
         self.cache = Cache(self)
         self.cloud = DataCloud(self, config=self.config._config)
@@ -241,14 +244,24 @@ class Project(object):
                 raise ReproductionError(stages[n].relpath, ex)
         return result
 
+    def _cleanup_unused_links(self, all_stages):
+        used = []
+        for stage in all_stages:
+            for out in stage.outs:
+                used.append(out.path)
+        self.link_state.remove_unused(used)
+
     def checkout(self, target=None):
+        all_stages = self.stages()
+
         if target:
             if not Stage.is_stage_file(target):
                 raise NotDvcFileError(target)
             stages = [Stage.load(self, target)]
         else:
-            self.link_state.remove_all()
-            stages = self.stages()
+            stages = all_stages
+
+        self._cleanup_unused_links(all_stages)
 
         for stage in stages:
             stage.checkout()
diff --git a/dvc/remote/local.py b/dvc/remote/local.py
index 7ddb6ab95..987369084 100644
--- a/dvc/remote/local.py
+++ b/dvc/remote/local.py
@@ -31,6 +31,7 @@ class RemoteLOCAL(RemoteBase):
 
     def __init__(self, project, config):
         self.project = project
+        self.state = self.project.state
         self.link_state = project.link_state
         storagepath = config.get(Config.SECTION_AWS_STORAGEPATH, None)
         self.cache_dir = config.get(Config.SECTION_REMOTE_URL, storagepath)
@@ -46,8 +47,6 @@ class RemoteLOCAL(RemoteBase):
         if self.cache_dir != None and not os.path.exists(self.cache_dir):
             os.mkdir(self.cache_dir)
 
-        self.state = State(self.cache_dir)
-
     @property
     def prefix(self):
         return self.cache_dir
@@ -85,36 +84,32 @@ class RemoteLOCAL(RemoteBase):
 
         return False
 
-    def link(self, md5, path, dump=True):
-        cache = self.get(md5)
-        if not cache or not os.path.exists(cache) or self.changed(md5):
-            if cache:
-                msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.'
-                Logger.warn(msg.format(md5, os.path.relpath(path)))
-            return
+    def link(self, cache, path):
+        assert os.path.isfile(cache)
 
         dname = os.path.dirname(path)
         if not os.path.exists(dname):
             os.makedirs(dname)
 
-        if self.cache_types != None:
-            types = self.cache_types
-        else:
-            types = self.CACHE_TYPES
-
-        for typ in types:
+        i = 0
+        N = len(self.cache_types)
+        while i < N:
             try:
-                self.CACHE_TYPE_MAP[typ](cache, path)
-                self.link_state.update(path, dump=dump)
+                self.CACHE_TYPE_MAP[self.cache_types[i]](cache, path)
                 return
             except Exception as exc:
-                msg = 'Cache type \'{}\' is not supported'.format(typ)
+                msg = 'Cache type \'{}\' is not supported'.format(self.cache_types[i])
                 Logger.debug(msg)
-                if typ == types[-1]:
-                    raise DvcException(msg, cause=exc)
+                del self.cache_types[i]
+                i += 1
+
+        raise DvcException('No possible cache types left to try out.')
+
+    def load_dir_cache(self, md5):
+        path = self.get(md5)
+
+        assert self.is_dir_cache(path)
 
-    @staticmethod
-    def load_dir_cache(path):
         try:
             with open(path, 'r') as fd:
                 d = json.load(fd)
@@ -130,6 +125,23 @@ class RemoteLOCAL(RemoteBase):
 
         return d
 
+    def dump_dir_cache(self, md5, dir_info):
+        path = self.get(md5)
+        dname = os.path.dirname(path)
+
+        assert self.is_dir_cache(path)
+        assert isinstance(dir_info, list)
+
+        if not os.path.isdir(dname):
+            os.makedirs(dname)
+
+        # NOTE: Writing first and renaming after that
+        # to make sure that the operation is atomic.
+        tmp = '{}.{}'.format(path, str(uuid.uuid4()))
+        with open(tmp, 'w+') as fd:
+            json.dump(dir_info, fd, sort_keys=True)
+        move(tmp, path)
+
     @staticmethod
     def is_dir_cache(cache):
         return cache.endswith(State.MD5_DIR_SUFFIX)
@@ -143,6 +155,12 @@ class RemoteLOCAL(RemoteBase):
             Logger.warn('No cache info for \'{}\'. Skipping checkout.'.format(os.path.relpath(path)))
             return
 
+        if self.changed(md5):
+            msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.'
+            Logger.warn(msg.format(md5, os.path.relpath(path)))
+            remove(path)
+            return
+
         if os.path.exists(path):
             msg = u'Data \'{}\' exists. Removing before checkout'
             Logger.debug(msg.format(os.path.relpath(path)))
@@ -152,7 +170,8 @@ class RemoteLOCAL(RemoteBase):
         Logger.debug(msg.format(os.path.relpath(path), md5))
 
         if not self.is_dir_cache(cache):
-            self.link(md5, path, dump=True)
+            self.link(cache, path)
+            self.link_state.update(path)
             return
 
         # Create dir separately so that dir is created
@@ -160,12 +179,13 @@ class RemoteLOCAL(RemoteBase):
         if not os.path.exists(path):
             os.makedirs(path)
 
-        for entry in self.load_dir_cache(cache):
+        for entry in self.load_dir_cache(md5):
             md5 = entry[self.PARAM_MD5]
+            c = self.get(md5)
             relpath = entry[self.PARAM_RELPATH]
             p = os.path.join(path, relpath)
-            self.link(md5, p, dump=False)
-        self.link_state.dump()
+            self.link(c, p)
+        self.link_state.update(path)
 
     def _move(self, inp, outp):
         # moving in two stages to make last the move atomic in
@@ -177,36 +197,38 @@ class RemoteLOCAL(RemoteBase):
     def _save_file(self, path_info):
         path = path_info['path']
         md5 = self.state.update(path)
+        assert md5 != None
+
         cache = self.get(md5)
+
         if self.changed(md5):
-            Logger.debug(u'Saving \'{}\' to \'{}\''.format(os.path.relpath(path),
-                                                           os.path.relpath(cache)))
             self._move(path, cache)
-            self.state.update(cache)
+        else:
+            remove(path)
+
+        self.link(cache, path)
+        self.link_state.update(path)
 
         return {self.PARAM_MD5: md5}
 
     def _save_dir(self, path_info):
         path = path_info['path']
-        md5 = self.state.update(path)
-        cache = self.get(md5)
-        dname = os.path.dirname(cache)
-        dir_info = self.state.collect_dir(path)
+        md5, dir_info = self.state.update_info(path)
 
         for entry in dir_info:
             relpath = entry[State.PARAM_RELPATH]
+            m = entry[State.PARAM_MD5]
             p = os.path.join(path, relpath)
+            c = self.get(m)
 
-            self._save_file({'scheme': 'local', 'path': p})
-
-        if not os.path.isdir(dname):
-            os.makedirs(dname)
+            if self.changed(m):
+                self._move(p, c)
+            else:
+                remove(p)
 
-        Logger.debug(u'Saving directory \'{}\' to \'{}\''.format(os.path.relpath(path),
-                                                                 os.path.relpath(cache)))
+            self.link(c, p)
 
-        with open(cache, 'w+') as fd:
-            json.dump(dir_info, fd, sort_keys=True)
+        self.link_state.update(path)
 
         return {self.PARAM_MD5: md5}
 
@@ -217,13 +239,9 @@ class RemoteLOCAL(RemoteBase):
         path = path_info['path']
 
         if os.path.isdir(path):
-            checksum_info = self._save_dir(path_info)
+            return self._save_dir(path_info)
         else:
-            checksum_info = self._save_file(path_info)
-
-        self.checkout(path_info, checksum_info)
-
-        return checksum_info
+            return self._save_file(path_info)
 
     def save_info(self, path_info):
         if path_info['scheme'] != 'local':
@@ -318,7 +336,7 @@ class RemoteLOCAL(RemoteBase):
                 continue
             if not os.path.exists(cache):
                 missing.append(info)
-            collected.extend(self.load_dir_cache(cache))
+            collected.extend(self.load_dir_cache(md5))
         collected.extend(checksum_infos)
         return collected, missing
 
diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py
index c57e34906..b56586e86 100644
--- a/dvc/remote/s3.py
+++ b/dvc/remote/s3.py
@@ -45,7 +45,6 @@ class RemoteS3(RemoteBase):
         self.region = config.get(Config.SECTION_AWS_REGION, None)
         self.profile = config.get(Config.SECTION_AWS_PROFILE, None)
         self.credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)
-        self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL, None)
 
     @property
     def bucket(self):
@@ -57,7 +56,7 @@ class RemoteS3(RemoteBase):
 
     @property
     def s3(self):
-        return boto3.resource('s3', endpoint_url=self.endpoint_url)
+        return boto3.resource('s3')
 
     def get_etag(self, bucket, key):
         try:
diff --git a/dvc/state.py b/dvc/state.py
index 4f172e2c7..33733b126 100644
--- a/dvc/state.py
+++ b/dvc/state.py
@@ -48,22 +48,23 @@ class State(object):
     PARAM_MD5 = 'md5'
     MD5_DIR_SUFFIX = '.dir'
 
-    def __init__(self, dvc_dir):
-        self.dvc_dir = dvc_dir
+    def __init__(self, project):
+        self.project = project
+        self.dvc_dir = project.dvc_dir
         self._lock = threading.Lock()
-        if dvc_dir:
-            self.state_file = os.path.join(dvc_dir, self.STATE_FILE)
-            self._lock_file = Lock(dvc_dir, self.STATE_LOCK_FILE)
+        if self.dvc_dir:
+            self.state_file = os.path.join(self.dvc_dir, self.STATE_FILE)
+            self._lock_file = Lock(self.dvc_dir, self.STATE_LOCK_FILE)
         else:
             self.state_file = None
             self._lock_file = threading.Lock()
         self._db = self.load()
 
     @staticmethod
-    def init(dvc_dir):
-        return State(dvc_dir)
+    def init(project):
+        return State(project)
 
-    def collect_dir(self, dname):
+    def _collect_dir(self, dname):
         dir_info = []
 
         for root, dirs, files in os.walk(dname):
@@ -71,19 +72,25 @@ class State(object):
                 path = os.path.join(root, fname)
                 relpath = os.path.relpath(path, dname)
 
-                md5 = self.update(path, dump=False)
+                # FIXME: we could've used `md5 = self.update(path, dump=False)` here,
+                # but it is around twice as slow(on ssd, don't know about hdd) for a
+                # directory with small files. What we could do here is introduce some
+                # kind of a limit for file size, after which we would actually register
+                # it in our state file.
+                md5 = file_md5(path)[0]
                 dir_info.append({self.PARAM_RELPATH: relpath, self.PARAM_MD5: md5})
 
-        self.dump()
+        md5 = dict_md5(dir_info) + self.MD5_DIR_SUFFIX
+        if self.project.cache.local.changed(md5):
+            self.project.cache.local.dump_dir_cache(md5, dir_info)
 
-        return dir_info
+        return (md5, dir_info)
 
-    def compute_md5(self, path):
+    def _collect(self, path):
         if os.path.isdir(path):
-            dir_info = self.collect_dir(path)
-            return dict_md5(dir_info) + self.MD5_DIR_SUFFIX
+            return self._collect_dir(path)
         else:
-            return file_md5(path)[0]
+            return (file_md5(path)[0], None)
 
     def changed(self, path, md5):
         actual = self.update(path)
@@ -120,18 +127,18 @@ class State(object):
     def inode(path):
         return str(System.inode(path))
 
-    def update(self, path, dump=True):
+    def _do_update(self, path, dump=True):
         if not os.path.exists(path):
-            return None
+            return (None, None)
 
         mtime = self.mtime(path)
         inode = self.inode(path)
 
         md5 = self._get(inode, mtime)
         if md5:
-            return md5
+            return (md5, None)
 
-        md5 = self.compute_md5(path)
+        md5, info = self._collect(path)
         state = StateEntry(md5, mtime)
         d = state.dumpd()
 
@@ -142,7 +149,16 @@ class State(object):
                 if dump:
                     self.dump()
 
-        return md5
+        return (md5, info)
+
+    def update(self, path, dump=True):
+        return self._do_update(path, dump=dump)[0]
+
+    def update_info(self, path, dump=True):
+        md5, info = self._do_update(path, dump=dump)
+        if not info:
+            info = self.project.cache.local.load_dir_cache(md5)
+        return (md5, info)
 
     def _get(self, inode, mtime):
         with self._lock:
@@ -190,9 +206,9 @@ class LinkState(State):
     STATE_FILE = 'link.state'
     STATE_LOCK_FILE = STATE_FILE + '.lock'
 
-    def __init__(self, root_dir, dvc_dir):
-        super(LinkState, self).__init__(dvc_dir)
-        self.root_dir = root_dir
+    def __init__(self, project):
+        super(LinkState, self).__init__(project)
+        self.root_dir = project.root_dir
 
     def update(self, path, dump=True):
         if not os.path.exists(path):
@@ -209,11 +225,15 @@ class LinkState(State):
                 with self._lock_file:
                     self.dump()
 
-    def _do_remove_all(self):
-        for p, s in self._db.items():
+    def _do_remove_unused(self, used):
+        items = self._db.copy().items()
+        for p, s in items:
             path = os.path.join(self.root_dir, p)
             state = LinkStateEntry.loadd(s)
 
+            if path in used:
+                continue
+
             if not os.path.exists(path):
                 continue
 
@@ -221,12 +241,12 @@ class LinkState(State):
             mtime = self.mtime(path)
 
             if inode == state.inode and mtime == state.mtime:
+                Logger.debug('Removing \'{}\' as unused link.'.format(path))
                 remove(path)
+                del self._db[p]
 
-        self._db = {}
-
-    def remove_all(self):
+    def remove_unused(self, used):
         with self._lock:
             with self._lock_file:
-                self._do_remove_all()
+                self._do_remove_unused(used)
                 self.dump()


Test patch:
diff --git a/tests/test_cache.py b/tests/test_cache.py
index 37963f2ce..d148c4d05 100644
--- a/tests/test_cache.py
+++ b/tests/test_cache.py
@@ -41,13 +41,13 @@ class TestCacheLoadBadDirCache(TestDvc):
         self.assertEqual(len(ret), 0)
 
     def test(self):
-        fname = 'not-json'
+        fname = self.dvc.cache.local.get('123.dir')
         self.create(fname, '<clearly>not,json')
-        self._do_test(RemoteLOCAL.load_dir_cache(fname))
+        self._do_test(self.dvc.cache.local.load_dir_cache(fname))
 
-        fname = 'not-list'
+        fname = self.dvc.cache.local.get('234.dir')
         self.create(fname, '{"a": "b"}')
-        self._do_test(RemoteLOCAL.load_dir_cache(fname))
+        self._do_test(self.dvc.cache.local.load_dir_cache(fname))
 
 
 class TestExternalCacheDir(TestDvc):
diff --git a/tests/test_state.py b/tests/test_state.py
index 10ec64005..7ea8af24a 100644
--- a/tests/test_state.py
+++ b/tests/test_state.py
@@ -16,7 +16,7 @@ class TestState(TestDvc):
         mtime = os.path.getmtime(path)
         inode = System.inode(path)
 
-        state = State(self.dvc.dvc_dir)
+        state = State(self.dvc)
 
         entry_md5 = state.update(path)
         self.assertEqual(entry_md5, md5)
@@ -41,6 +41,6 @@ class TestState(TestDvc):
 
         # Reload state db to make sure that it stays the same
         old_db = state._db
-        state = State(self.dvc.dvc_dir)
+        state = State(self.dvc)
         new_db = state._db
         self.assertEqual(old_db, new_db)