dvc: uptimize add/checkout/state

Signed-off-by: Ruslan Kuprieiev <[email protected]>

dvc/output/local.py

@@ -41,13 +41,10 @@ class OutputLOCAL(DependencyLOCAL):
         return ret
 
     def changed(self):
-        if super(OutputLOCAL, self).changed():
-            return True
+        if not self.use_cache:
+            return super(OutputLOCAL, self).changed()
 
-        if self.use_cache and self.info != self.project.cache.local.save_info(self.path_info):
-            return True
-
-        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
 

dvc/remote/local.py

@@ -84,13 +84,8 @@ class RemoteLOCAL(RemoteBase):
 
         return False
 
-    def link(self, md5, path):
-        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):
@@ -141,8 +136,12 @@ class RemoteLOCAL(RemoteBase):
         if not os.path.isdir(dname):
             os.makedirs(dname)
 
-        with open(path, 'w+') as fd:
+        # 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):
@@ -157,6 +156,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)))
@@ -166,7 +171,7 @@ class RemoteLOCAL(RemoteBase):
         Logger.debug(msg.format(os.path.relpath(path), md5))
 
         if not self.is_dir_cache(cache):
-            self.link(md5, path)
+            self.link(cache, path)
             self.link_state.update(path)
             return
 
@@ -177,9 +182,10 @@ class RemoteLOCAL(RemoteBase):
 
         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)
+            self.link(c, p)
         self.link_state.update(path)
 
     def _move(self, inp, outp):
@@ -192,12 +198,17 @@ 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}
 
@@ -207,9 +218,18 @@ class RemoteLOCAL(RemoteBase):
 
         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 self.changed(m):
+                self._move(p, c)
+            else:
+                remove(p)
+
+            self.link(c, p)
+
+        self.link_state.update(path)
 
         return {self.PARAM_MD5: md5}
 
@@ -220,13 +240,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':

dvc/state.py

@@ -72,7 +72,12 @@ 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})
 
         md5 = dict_md5(dir_info) + self.MD5_DIR_SUFFIX

tests/test_cache.py

@@ -41,13 +41,13 @@ class TestCacheLoadBadDirCache(TestDvc):
         self.assertEqual(len(ret), 0)
 
     def test(self):
-        fname = self.dvc.cache.local.get('123')
+        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 = self.dvc.cache.local.get('234')
+        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):