Merge pull request #382 from dataversioncontrol/dvc_fsck

Dvc fsck

dvc/cache.py

@@ -1,5 +1,7 @@
 import os
 
+from dvc.system import System
+
 
 class Cache(object):
     CACHE_DIR = 'cache'
@@ -23,3 +25,13 @@ class Cache(object):
 
     def get(self, md5):
         return os.path.join(self.cache_dir, md5)
+
+    def find_cache(self, files):
+        file_set = set(files)
+        cached = {}
+        for cache_file in self.all():
+            cached_files = list(filter(lambda f: System.samefile(cache_file, f, True), file_set))
+            cached.update(dict((f, os.path.basename(cache_file)) for f in cached_files))
+            file_set = file_set - set(cached_files)
+        return cached
+

dvc/cli.py

@@ -6,6 +6,7 @@ import argparse
 from multiprocessing import cpu_count
 
 from dvc.command.find import CmdFind
+from dvc.command.fsck import CmdFsck
 from dvc.command.init import CmdInit
 from dvc.command.remove import CmdRemove
 from dvc.command.run import CmdRun
@@ -414,6 +415,22 @@ def parse_args(argv=None):
                         default=False,
                         help='Show metrics value.')
 
+    fsck_parser = subparsers.add_parser(
+                        'fsck',
+                        parents=[parent_parser],
+                        help='Data file consistency check')
+    fsck_parser.add_argument(
+                        'targets',
+                        nargs='*',
+                        help='Data files to check')
+    fsck_parser.add_argument(
+                        '-p',
+                        '--physical',
+                        action='store_true',
+                        default=False,
+                        help='Compute actual md5')
+    fsck_parser.set_defaults(func=CmdFsck)
+
     if isinstance(argv, str):
         argv = argv.split()
 

dvc/command/checkout.py

@@ -2,7 +2,6 @@ import os
 
 from dvc.command.common.base import CmdBase
 from dvc.logger import Logger
-from dvc.system import System
 
 
 class CmdCheckout(CmdBase):
@@ -12,9 +11,9 @@ class CmdCheckout(CmdBase):
         return 0
 
     def remove_untracked_hardlinks(self):
-        untracked_cache_files = self.untracked_hardlinks_files()
+        untracked = self.project.scm.untracked_files()
 
-        for file in untracked_cache_files:
+        for file in self.project.cache.find_cache(untracked).keys():
             Logger.info(u'Remove \'{}\''.format(file))
             os.remove(file)
 
@@ -23,13 +22,3 @@ class CmdCheckout(CmdBase):
                 Logger.info(u'Remove empty directory \'{}\''.format(dir))
                 os.removedirs(dir)
         pass
-
-    def untracked_hardlinks_files(self):
-        untracked_files = set(self.project.scm.untracked_files())
-        cached_files = []
-        for cache_file in self.project.cache.all():
-            hardlinks = list(filter(lambda f: System.samefile(cache_file, f), untracked_files))
-            cached_files.extend(hardlinks)
-            untracked_files = untracked_files - set(hardlinks)
-
-        return cached_files

dvc/command/fsck.py

@@ -0,0 +1,70 @@
+import os
+
+from dvc.command.common.base import CmdBase
+from dvc.data_cloud import file_md5
+
+
+class CmdFsck(CmdBase):
+    def run(self):
+        directions = self.all_directions()
+        files_and_stages = self.directions_by_datafile(directions)
+
+        if self.args.targets:
+            dvc_files = [os.path.relpath(os.path.abspath(f), self.project.root_dir)
+                         for f in self.args.targets]
+        else:
+            dvc_files = files_and_stages.keys()
+
+        caches = self.project.cache.find_cache(dvc_files)
+
+        self.print_fsck(caches, dvc_files, files_and_stages)
+        return 0
+
+    def all_directions(self):
+        result = []
+        for stage in self.project.stages():
+            for dep in stage.deps + stage.outs:
+                result.append((stage, dep))
+        return result
+
+    def print_fsck(self, caches, dvc_files, files_and_stages):
+        for file in dvc_files:
+            print(u'File {}:'.format(file))
+
+            full_path = os.path.join(self.project.root_dir, file)
+            if self.args.physical:
+                md5 = file_md5(full_path)[0]
+                print(u'    Actual checksum:        {}'.format(md5))
+            else:
+                md5 = None
+
+            hardlink_md5 = caches.get(file)
+            hardlink_msg = hardlink_md5 if hardlink_md5 else 'No cache file found'
+            hardlink_error = '!!!' if hardlink_md5 and md5 and md5 != hardlink_md5 else ''
+            print(u'    Hardlink to cache file: {} {}'.format(hardlink_msg, hardlink_error))
+
+            state = self.project.state.get(file)
+            local_state_error = '!!!' if state.md5 and hardlink_md5 and state.md5 != hardlink_md5 else ''
+            print(u'    Local state checksum:   {} {}'.format(state.md5 if state else '', local_state_error))
+            print(u'    Local state mtime:      {}'.format(state.mtime if state else ''))
+            mtime = os.path.getmtime(full_path) if os.path.exists(full_path) else 'None'
+            print(u'    Actual mtime:           {}'.format(mtime))
+
+            for stage, dep in files_and_stages.get(file, []):
+                stage_error = '!!!' if dep.md5 and hardlink_md5 and dep.md5 != hardlink_md5 else ''
+                print(u'    Stage file: {}'.format(stage.dvc_path))
+                print(u'        Type:               {}'.format(type(dep).__name__))
+                print(u'        Checksum:           {} {}'.format(dep.md5, stage_error))
+                print(u'        Use cache:          {}'.format(str(dep.use_cache).lower()))
+        pass
+
+    @staticmethod
+    def directions_by_datafile(directions):
+        result = {}
+
+        for stage, dep in directions:
+            if dep.dvc_path not in result:
+                result[dep.dvc_path] = []
+            result[dep.dvc_path].append((stage, dep))
+
+        return result

dvc/project.py

@@ -233,6 +233,13 @@ class Project(object):
 
         return G
 
+    def all_directions(self):
+        result = []
+        for stage in self.stages():
+            for dep in stage.deps + stage.outs:
+                result.append((stage, dep))
+        return result
+
     def stages(self):
         stages = []
         for root, dirs, files in os.walk(self.root_dir):

dvc/stage.py

@@ -54,7 +54,18 @@ class Output(object):
         self.md5 = md5
         self.use_cache = use_cache
 
+    @property
+    def dvc_path(self):
+        return os.path.relpath(self.path, self.project.root_dir)
+
+    @property
+    def dvc_path(self):
+        return os.path.relpath(self.path, self.project.root_dir)
+
     def _changed_md5(self):
+        if not os.path.exists(self.path):
+            return True
+
         state = self.project.state.get(self.path)
         if state and state.mtime == self.mtime():
             md5 = state.md5
@@ -213,7 +224,11 @@ class Stage(object):
 
     @property
     def relpath(self):
-        return os.path.relpath(self.path, '.')
+        return os.path.relpath(self.path)
+
+    @property
+    def dvc_path(self):
+        return os.path.relpath(self.path, self.project.root_dir)
 
     @staticmethod
     def is_stage_file(path):

dvc/state.py

@@ -54,6 +54,7 @@ class State(object):
         return State(root_dir, dvc_dir)
 
     def update(self, path, md5, mtime):
+        mtime = int(mtime)
         existing = self.get(path)
         if not existing:
             return self.add(path, md5, mtime)
@@ -64,6 +65,7 @@ class State(object):
         return state
 
     def add(self, path, md5, mtime):
+        mtime = int(mtime)
         entry = StateEntry(self.root_dir, path, md5, mtime)
         self._db.insert(entry.dumpd())
         return entry

dvc/system.py

@@ -26,7 +26,11 @@ class System(object):
         return winlink.create(source, link_name)
 
     @staticmethod
-    def samefile(path1, path2):
+    def samefile(path1, path2, check_exist=False):
+        if check_exist:
+            if not os.path.exists(path1) or not os.path.exists(path2):
+                return False
+
         if System.is_unix():
             return os.path.samefile(path1, path2)
 

tests/test_cache.py

@@ -1,6 +1,7 @@
 import os
 
 from dvc.cache import Cache
+from dvc.system import System
 
 from tests.basic_env import TestDvc
 
@@ -22,3 +23,20 @@ class TestCache(TestDvc):
     def test_get(self):
         cache = Cache(self.dvc.dvc_dir).get(os.path.basename(self.cache1))
         self.assertEquals(cache, self.cache1)
+
+    def test_find_cache(self):
+        fname1 = os.path.basename(self.cache1)
+        fname1_md5 = fname1
+        fname2 = os.path.basename(self.cache2)
+        fname2_md5 = fname2
+        fname3 = 'non_existing'
+
+        System.hardlink(self.cache1, fname1)
+        System.hardlink(self.cache2, fname2)
+
+        cache = Cache(self.dvc.dvc_dir).find_cache([fname1, fname2, fname3])
+
+        expected = {fname1: fname1_md5, fname2: fname2_md5}
+
+        self.assertEqual(len(cache), 2)
+        self.assertEqual(cache, expected)