Merge pull request #1400 from vernt/deterministic

Add dvc run --deterministic option.
2026-11-02 07:18:25 +01:00 · 2018-12-04 12:39:11 +02:00
parent 2245ae0772 ee16d2f1d7
commit 9f4dd2db68
6 changed files with 429 additions and 53 deletions
@@ -268,9 +268,28 @@ def parse_args(argv=None):
                        '--yes',
                        action='store_true',
                        default=False,
-                        help="Automatic 'yes' answer to all prompts. E.g. "
+                        help="(OBSOLETED, use --overwrite-dvcfile instead) "
+                             "Automatic 'yes' answer to all prompts. E.g. "
                             "when '.dvc' file exists and dvc asks if you "
                             "want to overwrite it.")
+    run_parser.add_argument(
+                        '--overwrite-dvcfile',
+                        action='store_true',
+                        default=False,
+                        help="Overwrite existing dvc file without asking "
+                             "for confirmation.")
+    run_parser.add_argument(
+                        '--ignore-build-cache',
+                        action='store_true',
+                        default=False,
+                        help="Run this stage even if it has been already "
+                        "ran with the same command/dependencies/outputs/etc "
+                        "before.")
+    run_parser.add_argument(
+                        '--remove-outs',
+                        action='store_true',
+                        default=False,
+                        help="Remove outputs before running the command.")
    run_parser.add_argument(
                        'command',
                        nargs=argparse.REMAINDER,
@@ -20,10 +20,8 @@ class CmdRun(CmdBase):
        return cmd

    def run(self):
+        overwrite = (self.args.yes or self.args.overwrite_dvcfile)
        try:
-            if self.args.yes:
-                self.project.prompt.default = True
-
            self.project.run(cmd=self._joined_cmd(),
                             outs=self.args.outs,
                             outs_no_cache=self.args.outs_no_cache,
@@ -31,7 +29,10 @@ class CmdRun(CmdBase):
                             deps=self.args.deps,
                             fname=self.args.file,
                             cwd=self.args.cwd,
-                             no_exec=self.args.no_exec)
+                             no_exec=self.args.no_exec,
+                             overwrite=overwrite,
+                             ignore_build_cache=self.args.ignore_build_cache,
+                             remove_outs=self.args.remove_outs)
        except DvcException as ex:
            self.project.logger.error('Failed to run command', ex)
            return 1
@@ -325,16 +325,21 @@ class Project(object):
    def _unprotect_file(self, path):
        import stat
        import uuid
+        from dvc.system import System
        from dvc.utils import copyfile, move, remove

-        self.logger.debug("Unprotecting '{}'".format(path))
+        if System.is_symlink(path) or System.is_hardlink(path):
+            self.logger.debug("Unprotecting '{}'".format(path))

-        tmp = os.path.join(os.path.dirname(path), '.' + str(uuid.uuid4()))
-        move(path, tmp)
+            tmp = os.path.join(os.path.dirname(path), '.' + str(uuid.uuid4()))
+            move(path, tmp)

-        copyfile(tmp, path)
+            copyfile(tmp, path)

-        remove(tmp)
+            remove(tmp)
+        else:
+            self.logger.debug("Skipping copying for '{}', since it is not "
+                              "a symlink or a hardlink.".format(path))

        os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE)

@@ -363,18 +368,26 @@ class Project(object):
            fname=None,
            cwd=os.curdir,
            no_exec=False,
-            overwrite=False):
+            overwrite=False,
+            ignore_build_cache=False,
+            remove_outs=False):
        from dvc.stage import Stage

-        stage = Stage.loads(project=self,
-                            fname=fname,
-                            cmd=cmd,
-                            cwd=cwd,
-                            outs=outs,
-                            outs_no_cache=outs_no_cache,
-                            metrics_no_cache=metrics_no_cache,
-                            deps=deps,
-                            overwrite=overwrite)
+        with self.state:
+            stage = Stage.loads(project=self,
+                                fname=fname,
+                                cmd=cmd,
+                                cwd=cwd,
+                                outs=outs,
+                                outs_no_cache=outs_no_cache,
+                                metrics_no_cache=metrics_no_cache,
+                                deps=deps,
+                                overwrite=overwrite,
+                                ignore_build_cache=ignore_build_cache,
+                                remove_outs=remove_outs)
+
+        if stage is None:
+            return None

        all_stages = self.stages()

@@ -29,6 +29,12 @@ class StageFileDoesNotExistError(DvcException):
        super(StageFileDoesNotExistError, self).__init__(msg)


+class StageFileAlreadyExistsError(DvcException):
+    def __init__(self, relpath):
+        msg = "Stage '{}' already exists".format(relpath)
+        super(StageFileAlreadyExistsError, self).__init__(msg)
+
+
 class StageFileIsNotDvcFileError(DvcException):
    def __init__(self, fname):
        msg = "'{}' is not a dvc file".format(fname)
@@ -149,7 +155,10 @@ class Stage(object):
               len(self.deps) == 1 and \
               len(self.outs) == 1

-    def _changed_deps(self, print_info, log):
+    def _changed_deps(self, log):
+        if self.locked:
+            return False
+
        if self.is_callback:
            msg = "Dvc file '{}' is a 'callback' stage (has a command and " \
                  "no dependencies) and thus always considered as changed."
@@ -159,37 +168,35 @@ class Stage(object):
        for dep in self.deps:
            if not dep.changed():
                continue
-            if print_info:
-                msg = "Dependency '{}' of '{}' changed."
-                log(msg.format(dep, self.relpath))
+            log("Dependency '{}' of '{}' changed.".format(dep, self.relpath))
            return True

        return False

-    def changed(self, print_info=False):
-        ret = False
+    def _changed_outs(self, log):
+        for out in self.outs:
+            if not out.changed():
+                continue
+            log("Output '{}' of '{}' changed.".format(out, self.relpath))
+            return True

+        return False
+
+    def _changed_md5(self, log):
+        if self.changed_md5():
+            log("Dvc file '{}' changed.".format(self.relpath))
+            return True
+        return False
+
+    def changed(self, print_info=False):
        if print_info:
            log = self.project.logger.info
        else:
            log = self.project.logger.debug

-        if not self.locked:
-            ret = self._changed_deps(print_info, log)
-
-        for out in self.outs:
-            if not out.changed():
-                continue
-            if print_info:
-                msg = "Output '{}' of '{}' changed."
-                log(msg.format(out, self.relpath))
-            ret = True
-
-        if self.changed_md5():
-            if print_info:
-                msg = "Dvc file '{}' changed."
-                log(msg.format(self.relpath))
-            ret = True
+        ret = any([self._changed_deps(log),
+                   self._changed_outs(log),
+                   self._changed_md5(log)])

        if ret:
            msg = "Stage '{}' changed.".format(self.relpath)
@@ -206,6 +213,12 @@ class Stage(object):
        for out in self.outs:
            out.remove(ignore_remove=ignore_remove)

+    def unprotect_outs(self):
+        for out in self.outs:
+            if out.path_info['scheme'] != 'local' or not out.exists:
+                continue
+            self.project.unprotect(out.path)
+
    def remove(self):
        self.remove_outs(ignore_remove=True)
        os.unlink(self.path)
@@ -290,6 +303,37 @@ class Stage(object):
        if not os.path.realpath(cwd).startswith(proj_dir):
            raise StageBadCwdError(cwd)

+    def is_cached(self):
+        """
+        Checks if this stage has been already ran and saved to the same
+        dvc file.
+        """
+        from dvc.remote.local import RemoteLOCAL
+        from dvc.remote.s3 import RemoteS3
+
+        old = Stage.load(self.project, self.path)
+        if old._changed_outs(log=self.project.logger.debug):
+            return False
+
+        # NOTE: need to save checksums for deps in order to compare them
+        # with what is written in the old stage.
+        for dep in self.deps:
+            dep.save()
+
+        old_d = old.dumpd()
+        new_d = self.dumpd()
+
+        # NOTE: need to remove checksums from old dict in order to compare
+        # it to the new one, since the new one doesn't have checksums yet.
+        old_d.pop(self.PARAM_MD5, None)
+        new_d.pop(self.PARAM_MD5, None)
+        outs = old_d.get(self.PARAM_OUTS, [])
+        for out in outs:
+            out.pop(RemoteLOCAL.PARAM_MD5, None)
+            out.pop(RemoteS3.PARAM_ETAG, None)
+
+        return old_d == new_d
+
    @staticmethod
    def loads(project=None,
              cmd=None,
@@ -301,7 +345,10 @@ class Stage(object):
              cwd=os.curdir,
              locked=False,
              add=False,
-              overwrite=True):
+              overwrite=True,
+              ignore_build_cache=False,
+              remove_outs=False):
+
        stage = Stage(project=project,
                      cwd=cwd,
                      cmd=cmd,
@@ -325,17 +372,28 @@ class Stage(object):
        cwd = os.path.abspath(cwd)
        path = os.path.join(cwd, fname)

-        if os.path.exists(path):
-            relpath = os.path.relpath(path)
-            msg = "'{}' already exists. " \
-                  "Do you wish to run the command and overwrite it?"
-            if not overwrite \
-               and not project.prompt.prompt(msg.format(relpath), False):
-                raise DvcException("'{}' already exists".format(relpath))
-
        stage.cwd = cwd
        stage.path = path

+        # NOTE: remove outs before we check build cache
+        if remove_outs:
+            stage.remove_outs(ignore_remove=False)
+            project.logger.warn("Build cache is ignored when using "
+                                "--remove-outs.")
+            ignore_build_cache = True
+        else:
+            stage.unprotect_outs()
+
+        if os.path.exists(path):
+            if not ignore_build_cache and stage.is_cached():
+                Logger.info('Stage is cached, skipping.')
+                return None
+
+            msg = "'{}' already exists. Do you wish to run the command and " \
+                  "overwrite it?".format(stage.relpath)
+            if not overwrite and not project.prompt.prompt(msg, False):
+                raise StageFileAlreadyExistsError(stage.relpath)
+
        return stage

    @staticmethod
@@ -127,6 +127,11 @@ class System(object):
        from ctypes import c_void_p, c_wchar_p, Structure, WinError, POINTER
        from ctypes.wintypes import DWORD, HANDLE, BOOL

+        # NOTE: use this flag to open symlink itself and not the target
+        # See https://docs.microsoft.com/en-us/windows/desktop/api/
+        # fileapi/nf-fileapi-createfilew#symbolic-link-behavior
+        FILE_FLAG_OPEN_REPARSE_POINT = 0x00200000
+
        FILE_FLAG_BACKUP_SEMANTICS = 0x02000000
        FILE_SHARE_READ = 0x00000001
        OPEN_EXISTING = 3
@@ -147,7 +152,7 @@ class System(object):
                        ("nFileIndexHigh", DWORD),
                        ("nFileIndexLow", DWORD)]

-        flags = FILE_FLAG_BACKUP_SEMANTICS
+        flags = FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT

        func = ctypes.windll.kernel32.CreateFileW
        func.argtypes = [c_wchar_p,
@@ -240,3 +245,22 @@ class System(object):
            return System._wait_for_input_posix(timeout)
        else:
            return System._wait_for_input_windows(timeout)
+
+    @staticmethod
+    def is_symlink(path):
+        if System.is_unix():
+            return os.path.islink(path)
+
+        # https://docs.microsoft.com/en-us/windows/desktop/fileio/
+        # file-attribute-constants
+        FILE_ATTRIBUTE_REPARSE_POINT = 0x400
+        info = System.getdirinfo(path)
+        return info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT
+
+    @staticmethod
+    def is_hardlink(path):
+        if System.is_unix():
+            return os.stat(path).st_nlink > 1
+
+        info = System.getdirinfo(path)
+        return info.nNumberOfLinks > 1
@@ -1,12 +1,16 @@
 import os
 import mock
+import time
+import shutil
 import filecmp
 import subprocess

+from dvc.project import Project
 from dvc.main import main
 from dvc.utils import file_md5
+from dvc.system import System
 from dvc.stage import Stage, StageFileBadNameError, MissingDep
-from dvc.stage import StageBadCwdError
+from dvc.stage import StageBadCwdError, StageFileAlreadyExistsError
 from dvc.command.run import CmdRun
 from dvc.exceptions import (OutputDuplicationError,
                            CircularDependencyError,
@@ -180,3 +184,260 @@ class TestCmdRun(TestDvc):
    def test_keyboard_interrupt(self, mock_popen):
        ret = main(['run', 'mycmd'])
        self.assertEqual(ret, 252)
+
+
+class TestRunRemoveOuts(TestDvc):
+    def test(self):
+        with open(self.CODE, 'w+') as fobj:
+            fobj.write("import sys\n")
+            fobj.write("import os\n")
+            fobj.write("if os.path.exists(sys.argv[1]):\n")
+            fobj.write("    sys.exit(1)\n")
+            fobj.write("open(sys.argv[1], 'w+').close()\n")
+
+        ret = main(['run',
+                    '--remove-outs',
+                    '-d', self.CODE,
+                    '-o', self.FOO,
+                    'python', self.CODE, self.FOO])
+        self.assertEqual(ret, 0)
+
+
+class TestRunUnprotectOutsCopy(TestDvc):
+    def test(self):
+        with open(self.CODE, 'w+') as fobj:
+            fobj.write("import sys\n")
+            fobj.write("with open(sys.argv[1], 'a+') as fobj:\n")
+            fobj.write("    fobj.write('foo')\n")
+
+        ret = main(['config', 'cache.protected', 'true'])
+        self.assertEqual(ret, 0)
+
+        ret = main(['config', 'cache.type', 'copy'])
+        self.assertEqual(ret, 0)
+
+        ret = main(['run',
+                    '-d', self.CODE,
+                    '-o', self.FOO,
+                    'python', self.CODE, self.FOO])
+        self.assertEqual(ret, 0)
+        self.assertFalse(os.access(self.FOO, os.W_OK))
+        self.assertEqual(open(self.FOO, 'r').read(), 'foofoo')
+
+        ret = main(['run',
+                    '--overwrite-dvcfile',
+                    '--ignore-build-cache',
+                    '-d', self.CODE,
+                    '-o', self.FOO,
+                    'python', self.CODE, self.FOO])
+        self.assertEqual(ret, 0)
+        self.assertFalse(os.access(self.FOO, os.W_OK))
+        self.assertEqual(open(self.FOO, 'r').read(), 'foofoofoo')
+
+
+class TestRunUnprotectOutsSymlink(TestDvc):
+    def test(self):
+        with open(self.CODE, 'w+') as fobj:
+            fobj.write("import sys\n")
+            fobj.write("import os\n")
+            fobj.write("assert os.path.exists(sys.argv[1])\n")
+            fobj.write("with open(sys.argv[1], 'a+') as fobj:\n")
+            fobj.write("    fobj.write('foo')\n")
+
+        ret = main(['config', 'cache.protected', 'true'])
+        self.assertEqual(ret, 0)
+
+        ret = main(['config', 'cache.type', 'symlink'])
+        self.assertEqual(ret, 0)
+
+        self.assertEqual(ret, 0)
+        ret = main(['run',
+                    '-d', self.CODE,
+                    '-o', self.FOO,
+                    'python', self.CODE, self.FOO])
+        self.assertEqual(ret, 0)
+        self.assertFalse(os.access(self.FOO, os.W_OK))
+        self.assertTrue(System.is_symlink(self.FOO))
+        self.assertEqual(open(self.FOO, 'r').read(), 'foofoo')
+
+        ret = main(['run',
+                    '--overwrite-dvcfile',
+                    '--ignore-build-cache',
+                    '-d', self.CODE,
+                    '-o', self.FOO,
+                    'python', self.CODE, self.FOO])
+        self.assertEqual(ret, 0)
+        self.assertFalse(os.access(self.FOO, os.W_OK))
+        self.assertTrue(System.is_symlink(self.FOO))
+        self.assertEqual(open(self.FOO, 'r').read(), 'foofoofoo')
+
+
+class TestRunUnprotectOutsHardlink(TestDvc):
+    def test(self):
+        with open(self.CODE, 'w+') as fobj:
+            fobj.write("import sys\n")
+            fobj.write("import os\n")
+            fobj.write("assert os.path.exists(sys.argv[1])\n")
+            fobj.write("with open(sys.argv[1], 'a+') as fobj:\n")
+            fobj.write("    fobj.write('foo')\n")
+
+        ret = main(['config', 'cache.protected', 'true'])
+        self.assertEqual(ret, 0)
+
+        ret = main(['config', 'cache.type', 'hardlink'])
+        self.assertEqual(ret, 0)
+
+        self.assertEqual(ret, 0)
+        ret = main(['run',
+                    '-d', self.CODE,
+                    '-o', self.FOO,
+                    'python', self.CODE, self.FOO])
+        self.assertEqual(ret, 0)
+        self.assertFalse(os.access(self.FOO, os.W_OK))
+        self.assertTrue(System.is_hardlink(self.FOO))
+        self.assertEqual(open(self.FOO, 'r').read(), 'foofoo')
+
+        ret = main(['run',
+                    '--overwrite-dvcfile',
+                    '--ignore-build-cache',
+                    '-d', self.CODE,
+                    '-o', self.FOO,
+                    'python', self.CODE, self.FOO])
+        self.assertEqual(ret, 0)
+        self.assertFalse(os.access(self.FOO, os.W_OK))
+        self.assertTrue(System.is_hardlink(self.FOO))
+        self.assertEqual(open(self.FOO, 'r').read(), 'foofoofoo')
+
+
+class TestCmdRunOverwrite(TestDvc):
+    def test(self):
+        ret = main(['run',
+                    '-d', self.FOO,
+                    '-d', self.CODE,
+                    '-o', 'out',
+                    '-f', 'out.dvc',
+                    'python', self.CODE, self.FOO, 'out'])
+        self.assertEqual(ret, 0)
+
+        stage_mtime = os.path.getmtime('out.dvc')
+
+        ret = main(['run',
+                    '-d', self.FOO,
+                    '-d', self.CODE,
+                    '-o', 'out',
+                    '-f', 'out.dvc',
+                    'python', self.CODE, self.FOO, 'out'])
+        self.assertEqual(ret, 0)
+
+        # NOTE: check that dvcfile was NOT overwritten
+        self.assertEqual(stage_mtime, os.path.getmtime('out.dvc'))
+        stage_mtime = os.path.getmtime('out.dvc')
+
+        ret = main(['run',
+                    '-d', self.FOO,
+                    '-d', self.CODE,
+                    '--overwrite-dvcfile',
+                    '--ignore-build-cache',
+                    '-o', 'out',
+                    '-f', 'out.dvc',
+                    'python', self.CODE, self.FOO, 'out'])
+        self.assertEqual(ret, 0)
+
+        # NOTE: check that dvcfile was overwritten
+        self.assertNotEqual(stage_mtime, os.path.getmtime('out.dvc'))
+        stage_mtime = os.path.getmtime('out.dvc')
+
+        ret = main(['run',
+                    '--overwrite-dvcfile',
+                    '-f', 'out.dvc',
+                    '-d', self.BAR])
+        self.assertEqual(ret, 0)
+
+        # NOTE: check that dvcfile was overwritten
+        self.assertNotEqual(stage_mtime, os.path.getmtime('out.dvc'))
+
+
+class TestRunDeterministicBase(TestDvc):
+    def setUp(self):
+        super(TestRunDeterministicBase, self).setUp()
+        self.out_file = 'out'
+        self.stage_file = self.out_file + '.dvc'
+        self.cmd = 'python {} {} {}'.format(self.CODE, self.FOO, self.out_file)
+        self.deps = [self.FOO, self.CODE]
+        self.outs = [self.out_file]
+        self.overwrite = False
+        self.ignore_build_cache = False
+
+        self._run()
+
+    def _run(self):
+        self.stage = self.dvc.run(cmd=self.cmd,
+                                  fname=self.stage_file,
+                                  overwrite=self.overwrite,
+                                  ignore_build_cache=self.ignore_build_cache,
+                                  deps=self.deps,
+                                  outs=self.outs)
+
+
+class TestRunDeterministic(TestRunDeterministicBase):
+    def test(self):
+        self._run()
+
+
+class TestRunDeterministicOverwrite(TestRunDeterministicBase):
+    def test(self):
+        self.overwrite = True
+        self.ignore_build_cache = True
+        self._run()
+
+
+class TestRunDeterministicCallback(TestRunDeterministicBase):
+    def test(self):
+        self.stage.remove()
+        self.deps = []
+        self._run()
+        self._run()
+
+
+class TestRunDeterministicChangedDep(TestRunDeterministicBase):
+    def test(self):
+        os.unlink(self.FOO)
+        shutil.copy(self.BAR, self.FOO)
+        with self.assertRaises(StageFileAlreadyExistsError):
+           self._run()
+
+
+class TestRunDeterministicChangedDepsList(TestRunDeterministicBase):
+    def test(self):
+        self.deps = [self.BAR, self.CODE]
+        with self.assertRaises(StageFileAlreadyExistsError):
+           self._run()
+
+
+class TestRunDeterministicNewDep(TestRunDeterministicBase):
+    def test(self):
+        self.deps = [self.FOO, self.BAR, self.CODE]
+        with self.assertRaises(StageFileAlreadyExistsError):
+           self._run()
+
+
+class TestRunDeterministicRemoveDep(TestRunDeterministicBase):
+    def test(self):
+        self.deps = [self.CODE]
+        with self.assertRaises(StageFileAlreadyExistsError):
+           self._run()
+
+
+class TestRunDeterministicChangedOut(TestRunDeterministicBase):
+    def test(self):
+        os.unlink(self.out_file)
+        self.out_file_mtime = None
+        with self.assertRaises(StageFileAlreadyExistsError):
+           self._run()
+
+
+class TestRunDeterministicChangedCmd(TestRunDeterministicBase):
+    def test(self):
+        self.cmd += ' arg'
+        with self.assertRaises(StageFileAlreadyExistsError):
+           self._run()