From 927d58cfad91e5b96009342eae2e39735dbd0228 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 7 Dec 2017 21:00:30 +0200 Subject: [PATCH] dvc: introduce stage files instead of state files These describe single stage of pipeline and are allowed to be placed anywhere in current or parent directories(up to project's root) of the data file. By default, 'dvc run' writes stage file to ./stage.dvc, but it can be changed by using '-f' option(see functest/test-repro.py). For now we're traversing project tree to find needed stage files for specific data files. This is very wasteful and not optimal, but will do for now, as this patch is aimed at stabilizing design of our repo files. Signed-off-by: Ruslan Kuprieiev --- dvc/cli.py | 4 ++ dvc/command/add.py | 5 +- dvc/command/checkout.py | 28 +-------- dvc/command/gc.py | 38 ++---------- dvc/command/merge.py | 43 ++++++------- dvc/command/repro.py | 15 ++--- dvc/command/run.py | 56 ++++++++--------- dvc/path/data_item.py | 5 +- dvc/path/factory.py | 30 ++------- dvc/state_file.py | 119 ++++++++++++++++++++++++++++++------ dvc/utils.py | 2 +- functest/test-repro-code.sh | 8 +-- 12 files changed, 177 insertions(+), 176 deletions(-) diff --git a/dvc/cli.py b/dvc/cli.py index ba42d9f5b..9553e8cd3 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -113,6 +113,10 @@ def parse_args(argv=None): action='store_true', default=False, help='Lock data item - disable reproduction.') + run_parser.add_argument('-f', + '--file', + default='stage.dvc', + help='Specify name of the state file') run_parser.add_argument( 'command', nargs=argparse.REMAINDER, diff --git a/dvc/command/add.py b/dvc/command/add.py index 60c1b8951..cbfd1ec96 100644 --- a/dvc/command/add.py +++ b/dvc/command/add.py @@ -3,6 +3,7 @@ import os from dvc.command.common.base import CmdBase from dvc.logger import Logger from dvc.state_file import StateFile +from dvc.path.data_item import DataItem class CmdAdd(CmdBase): @@ -39,9 +40,9 @@ class CmdAdd(CmdBase): for data_item in targets: Logger.debug('Creating state file for {}'.format(data_item.data.relative)) - state_file = StateFile(data_item=data_item, + state_file = StateFile(fname=data_item.data.relative + StateFile.STATE_FILE_SUFFIX, cmd=None, - out=StateFile.parse_deps_state(self.settings, [data_item.data.dvc]), + out=StateFile.parse_deps_state(self.settings, [data_item.data.relative], currdir=data_item.data.dirname), out_git=[], deps=[], locked=True) diff --git a/dvc/command/checkout.py b/dvc/command/checkout.py index 9796c35fa..f163b934e 100644 --- a/dvc/command/checkout.py +++ b/dvc/command/checkout.py @@ -35,30 +35,6 @@ class CmdCheckout(CmdBase): Logger.info('Checkout \'{}\''.format(item.data.relative)) def run(self): - try: - states = self.git.state_files_for_previous_commit() - prev_items = self.settings.path_factory.data_items_from_states(states, existing=False) - except Exception as ex: - Logger.error(u'Unable to get data files from previous commit: {}'.format(ex)) - return 1 - - curr_items = self.settings.path_factory.all_existing_data_items() - self.checkout(curr_items) - - self.remove_files(list(set(prev_items) - set(curr_items))) + items = self.settings.path_factory.all_existing_data_items() + self.checkout(items) return 0 - - @staticmethod - def remove_files(removed_items_set): - for item in removed_items_set: - path = item.data.relative - if not os.path.exists(path): - Logger.debug(u'File \'{}\' does not exist'.format(path)) - else: - Logger.info(u'Remove \'{}\''.format(path)) - os.remove(path) - dir = os.path.dirname(path) - if not os.listdir(dir): - Logger.info(u'Remove directory \'{}\''.format(dir)) - os.removedirs(dir) - pass diff --git a/dvc/command/gc.py b/dvc/command/gc.py index ada0bac8e..7d138a553 100644 --- a/dvc/command/gc.py +++ b/dvc/command/gc.py @@ -1,49 +1,23 @@ import os -from dvc.command.common.traverse import Traverse +from dvc.command.common.base import CmdBase from dvc.logger import Logger -from dvc.path.data_item import DataItemError from dvc.config import ConfigI +from dvc.state_file import StateFile -class CmdGC(Traverse): +class CmdGC(CmdBase): def __init__(self, settings): - super(CmdGC, self).__init__(settings, "garbage collect", do_not_start_from_root=False) - self.clist = [] + super(CmdGC, self).__init__(settings) def run(self): - if not self._traverse(self.git.git_dir_abs): - Logger.error('Failed to collect used cache') - return 1 + clist = [str(x) for x in StateFile.find_all_cache_files(self.git)] for cache in os.listdir(ConfigI.CACHE_DIR): fname = os.path.join(ConfigI.CACHE_DIR, cache) - if fname in self.clist: + if os.path.basename(fname) in clist: continue os.remove(fname) - self._remove_cloud_cache(self.settings.path_factory.data_item('.empty', fname)) Logger.info('Cache \'{}\' was removed'.format(fname)) return 0 - - def process_file(self, target): - Logger.debug(u'[Cmd-GC] GC file {}.'.format(target)) - - try: - data_item = self.settings.path_factory.existing_data_item(target) - except DataItemError: - return - - self.clist.append(data_item.cache.relative) - Logger.debug(u'[Cmd-GC] GC data item {}. Success.'.format(data_item.data.relative)) - - def is_recursive(self): - return self.parsed_args.recursive - - @property - def no_git_actions(self): - return True - - @staticmethod - def not_committed_changes_warning(): - pass diff --git a/dvc/command/merge.py b/dvc/command/merge.py index 2a9566344..1d6e8436d 100644 --- a/dvc/command/merge.py +++ b/dvc/command/merge.py @@ -14,44 +14,35 @@ class CmdMerge(CmdBase): super(CmdMerge, self).__init__(settings) def print_info(self, targets): - for item in targets: + for fname in targets: Logger.info('Restored original data after merge:') - Logger.info(' {}'.format(item.data.relative)) - - def collect_data(self): - dlist = [] - flist = self.git.get_last_merge_changed_files() - for fname in flist: - if fname.endswith(DataItem.STATE_FILE_SUFFIX): - data = fname[:-len(DataItem.STATE_FILE_SUFFIX)] - dlist.append(data) - return dlist + Logger.info(' {}'.format(fname)) def collect_targets(self): targets = [] - flist = self.collect_data() - items = self.settings.path_factory.to_data_items(flist)[0] - for item in items: - command = StateFile.load(item) + for fname in self.git.get_last_merge_changed_files(): + if not StateFile._is_state_file(fname): + continue - if not command.cmd and command.locked: - targets.append(item) + state = StateFile.load(fname) + if not state.cmd and state.locked: + targets.append(fname) return targets def checkout_targets(self, targets): - data = [] - for item in targets: - prev_state = StateFile.loads(self.git.get_file_content_before_last_merge(item.state.relative)) - curr_state = StateFile.load(item) + items = [] + for fname in targets: + self.git.checkout_file_before_last_merge(fname) + state = StateFile.load(fname) + for out in state.out: + item = self.settings.path_factory.data_item(os.path.join(state.cwd, out)) + items.append(item) - curr_state.out[item.data.dvc] = prev_state.out[item.data.dvc] - curr_state.save() + CmdCheckout.checkout(items) - CmdCheckout.checkout([item]) - - msg = 'DVC merge files: {}'.format(' '.join(data)) + msg = 'DVC merge files: {}'.format(' '.join(targets)) self.commit_if_needed(msg) def run(self): diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 2544664e7..0919f530a 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -88,7 +88,7 @@ class ReproChange(object): self._recursive = recursive self._force = force - self.command = StateFile.load(data_item) + self.command = StateFile.find(data_item) if not self.command.cmd and not self.command.locked: msg = 'Error: data item "{}" is not locked, but has no command for reproduction' @@ -105,15 +105,15 @@ class ReproChange(object): return self._cmd_obj def remove_output_files(self): - for output_dvc in self.command.out: - Logger.debug('Removing output file {} before reproduction.'.format(output_dvc)) - + for out in self.command.out: + path = os.path.join(self.command.cwd, out) + Logger.debug('Removing output file {} before reproduction.'.format(path)) try: - data_item = self.cmd_obj.settings.path_factory.data_item_from_dvc_path(output_dvc) + data_item = self.cmd_obj.settings.path_factory.data_item(path) os.remove(data_item.data.relative) except Exception as ex: msg = 'Data item {} cannot be removed before reproduction: {}' - Logger.debug(msg.format(output_dvc, ex)) + Logger.debug(msg.format(path, ex)) def reproduce_run(self): Logger.info('Reproducing run command for data item {}. Args: {}'.format( @@ -169,7 +169,8 @@ class ReproChange(object): def reproduce_deps(self, data_item_dvc, recursive): result = False - for path,md5 in self.command.deps.items(): + for name,md5 in self.command.deps.items(): + path = os.path.join(self.command.cwd, name) if self.reproduce_dep(path, md5, recursive): result = True diff --git a/dvc/command/run.py b/dvc/command/run.py index a1a53b59e..a9d4b9bed 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -19,44 +19,40 @@ class CmdRun(CmdBase): def run(self): cmd = ' '.join(self.parsed_args.command) - state = StateFile(data_item=None, - cmd=cmd, - out=self.parsed_args.out, - out_git=self.parsed_args.out_git, - deps=self.parsed_args.deps, - locked=self.parsed_args.lock) + + if os.path.isfile(self.parsed_args.file): + Logger.error("Stage file {} already exists".format(self.parsed_args.file)) + return 1 + + state = StateFile(fname=self.parsed_args.file, + cmd=cmd, + out=self.parsed_args.out, + out_git=self.parsed_args.out_git, + deps=self.parsed_args.deps, + locked=self.parsed_args.lock) self.run_command(self.settings, state) return self.commit_if_needed('DVC run: {}'.format(state.cmd)) @staticmethod def run_command(settings, state): - Executor.exec_cmd_only_success(state.cmd, shell=True) + Executor.exec_cmd_only_success(state.cmd, cwd=state.cwd, shell=True) - CmdRun.apply_to_files(state.out, state, CmdRun._create_cache_and_state_files, settings) - CmdRun.apply_to_files(state.out_git, state, CmdRun._create_state_file, settings) + CmdRun.move_output_to_cache(settings, state) + CmdRun.update_state_file(settings, state) @staticmethod - def apply_to_files(files, state, func, settings): - items = settings.path_factory.to_data_items(files)[0] - [func(i, state, settings) for i in items] + def update_state_file(settings, state): + Logger.debug('Update state file "{}"'.format(state.path)) + state.out = StateFile.parse_deps_state(settings, state.out) + state.out_git = StateFile.parse_deps_state(settings, state.out_git) + state.deps = StateFile.parse_deps_state(settings, state.deps) + state.save() @staticmethod - def _create_cache_and_state_files(data_item, state, settings): - Logger.debug('Move output file "{}" to cache dir "{}" and create a hardlink'.format( - data_item.data.relative, data_item.cache_dir_abs)) - data_item.move_data_to_cache() - return CmdRun._create_state_file(data_item, state, settings) - - @staticmethod - def _create_state_file(data_item, state, settings): - Logger.debug('Create state file "{}"'.format(data_item.state.relative)) - - state_file = StateFile(data_item=data_item, - cmd=state.cmd, - out=StateFile.parse_deps_state(settings, state.out), - out_git=StateFile.parse_deps_state(settings, state.out_git), - locked=state.locked, - deps=StateFile.parse_deps_state(settings, state.deps)) - state_file.save() - return state_file + def move_output_to_cache(settings, state): + items = settings.path_factory.to_data_items(state.out)[0] + for item in items: + Logger.debug('Move output file "{}" to cache dir "{}" and create a hardlink'.format( + item.data.relative, item.cache_dir_abs)) + item.move_data_to_cache() diff --git a/dvc/path/data_item.py b/dvc/path/data_item.py index 277c7b212..f9592919d 100644 --- a/dvc/path/data_item.py +++ b/dvc/path/data_item.py @@ -33,7 +33,6 @@ class NotInGitDirError(DataDirError): class DataItem(object): - STATE_FILE_SUFFIX = '.dvc' LOCAL_STATE_FILE_PREFIX = '.' LOCAL_STATE_FILE_SUFFIX = '.dvc_local_state' CACHE_FILE_SEP = '_' @@ -78,7 +77,7 @@ class DataItem(object): @cached_property def state(self): - return self._state('', self.STATE_FILE_SUFFIX) + return Path(StateFile.find(self).path, self._git) @cached_property def cache_dir_abs(self): @@ -99,7 +98,7 @@ class DataItem(object): if self._cache_file: file_name = os.path.relpath(os.path.realpath(self._cache_file), cache_dir) else: - file_name = StateFile.load(self).out[self.data.dvc] + file_name = str(StateFile.find_md5(self)) cache_file = os.path.join(cache_dir, file_name) return Path(cache_file, self._git) diff --git a/dvc/path/factory.py b/dvc/path/factory.py index 0793994cc..f16f01bb1 100644 --- a/dvc/path/factory.py +++ b/dvc/path/factory.py @@ -6,6 +6,7 @@ from dvc.path.data_item import DataItem, DataItemError, DataDirError from dvc.path.path import Path from dvc.path.stated_data_item import StatedDataItem from dvc.system import System +from dvc.state_file import StateFile class PathFactory(object): @@ -31,7 +32,8 @@ class PathFactory(object): return self.data_item(path.relative) def is_data_item(self, fname): - return os.path.isfile(fname + DataItem.STATE_FILE_SUFFIX) + item = DataItem(fname, self._git, self._config) + return StateFile.find(item) != None def existing_data_item(self, fname): if not self.is_data_item(fname): @@ -53,27 +55,5 @@ class PathFactory(object): return result, externally_created_files def all_existing_data_items(self, subdir='.', cache_exists=True): - states = [] - - for root, dirs, files in os.walk(os.path.join(self._git.git_dir_abs, subdir)): - for fname in files: - path = os.path.join(root, fname) - - if not fname.endswith(DataItem.STATE_FILE_SUFFIX): - continue - - states.append(path) - - data_items = self.data_items_from_states(states) - if cache_exists: - data_items = filter(lambda i: os.path.exists(i.state.relative), data_items) - - return data_items - - def data_items_from_states(self, states, existing=True): - return [self.data_item_from_dvc_path(self.state_path_to_dvc_path(s), existing) - for s in states] - - @staticmethod - def state_path_to_dvc_path(state): - return re.sub(DataItem.STATE_FILE_SUFFIX + '$', '', state) + files = StateFile.find_all_data_files(self._git, subdir) + return self.to_data_items(files)[0] diff --git a/dvc/state_file.py b/dvc/state_file.py index 395f623af..897ae6159 100644 --- a/dvc/state_file.py +++ b/dvc/state_file.py @@ -3,6 +3,7 @@ import yaml from dvc.exceptions import DvcException from dvc.data_cloud import file_md5 +from dvc.path.path import Path class StateFileError(DvcException): @@ -24,8 +25,7 @@ class StateFileBase(object): class StateFile(StateFileBase): - MAGIC = 'DVC-State' - VERSION = '0.1' + STATE_FILE_SUFFIX = '.dvc' PARAM_CMD = 'cmd' PARAM_DEPS = 'deps' @@ -34,14 +34,13 @@ class StateFile(StateFileBase): PARAM_OUT_GIT = 'out-git' def __init__(self, - data_item=None, + fname=None, cmd=None, out=None, out_git=None, deps=None, locked=None): super(StateFile, self).__init__() - self.data_item = data_item self.cmd = cmd self.out = out @@ -49,24 +48,30 @@ class StateFile(StateFileBase): self.deps = deps self.locked = locked + self.path = os.path.abspath(fname) if fname else None + self.cwd = os.path.dirname(self.path) if self.path else None + @staticmethod - def parse_deps_state(settings, deps): + def parse_deps_state(settings, deps, currdir=None): state = {} for dep in deps: if settings.path_factory.is_data_item(dep): item = settings.path_factory.data_item(dep) - state[item.data.dvc] = StateFile.load(item).out[item.data.dvc] + md5 = StateFile.find_md5(item) else: - state[dep] = file_md5(os.path.join(settings.git.git_dir_abs, dep))[0] + md5 = file_md5(os.path.join(settings.git.git_dir_abs, dep))[0] + + if currdir: + name = os.path.relpath(dep, currdir) + else: + name = dep + + state[name] = md5 return state - @property - def file(self): - return self.data_item.state.relative - @staticmethod - def loadd(data): - return StateFile(data_item=None, + def loadd(data, fname=None): + return StateFile(fname=fname, cmd=data.get(StateFile.PARAM_CMD, None), out=data.get(StateFile.PARAM_OUT, None), out_git=data.get(StateFile.PARAM_OUT_GIT, None), @@ -74,8 +79,86 @@ class StateFile(StateFileBase): locked=data.get(StateFile.PARAM_LOCKED, None)) @staticmethod - def load(data_item): - return StateFile._load(data_item.state.relative, StateFile) + def load(fname): + return StateFile._load(fname, StateFile, fname) + + @staticmethod + def _is_state_file(path): + return path.endswith(StateFile.STATE_FILE_SUFFIX) and os.path.isfile(path) + + @staticmethod + def _find_state(fname, dname): + name = os.path.relpath(fname, dname) + for entry in os.listdir(dname): + state_file = os.path.join(dname, entry) + if not StateFile._is_state_file(state_file): +# if not state_file.endswith(StateFile.STATE_FILE_SUFFIX) or not os.path.isfile(state_file): + continue + state = StateFile.load(os.path.join(dname, state_file)) + if name in state.out: + return state + return None + + @staticmethod + def _find(name, start, finish): + dname = start + fname = name + while True: + state = StateFile._find_state(fname, dname) + if state: + return state + + if dname == finish: + break + + dname = os.path.dirname(dname) + return None + + @staticmethod + def find(data_item): + return StateFile._find(data_item.data.abs, data_item.data.dirname, data_item._git.git_dir_abs) + + @staticmethod + def find_all_state_files(git, subdir='.'): + states = [] + for root, dirs, files in os.walk(os.path.join(git.git_dir_abs, subdir)): + for fname in files: + path = os.path.join(root, fname) + + if not StateFile._is_state_file(path): + continue + + states.append(path) + return states + + @staticmethod + def find_all_states(git, subdir='.'): + state_files = StateFile.find_all_state_files(git, subdir) + return [StateFile.load(state_file) for state_file in state_files] + + @staticmethod + def find_all_data_files(git, subdir='.'): + states = StateFile.find_all_states(git, subdir) + files = [] + for state in states: + for out in state.out: + files.append(os.path.join(state.cwd, out)) + return files + + @staticmethod + def find_all_cache_files(git, subdir='.'): + states = StateFile.find_all_states(git, subdir) + cache_files = [] + for state in states: + for out,md5 in state.out.items(): + cache_files.append(md5) + return cache_files + + @staticmethod + def find_md5(data_item): + state = StateFile.find(data_item) + name = os.path.relpath(data_item.data.abs, os.path.dirname(state.path)) + return state.out[name] @staticmethod def loads(content): @@ -91,11 +174,7 @@ class StateFile(StateFileBase): self.PARAM_LOCKED: self.locked } - file_dir = os.path.dirname(self.file) - if file_dir != '' and not os.path.isdir(file_dir): - os.makedirs(file_dir) - - self._save(self.file, res) + self._save(self.path, res) class LocalStateFile(StateFileBase): diff --git a/dvc/utils.py b/dvc/utils.py index 2a43660ec..0e5cb5294 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -79,7 +79,7 @@ def map_progress(func, targets, n_threads): try: ret = pool.map(func, targets) except Exception as exc: - Logger.error('Unexpected exception while processing targets: {}'.format(exc)) + Logger.error('Unexpected exception while processing targets: {}'.format(exc), exc_info=True) finally: progress.finish() diff --git a/functest/test-repro-code.sh b/functest/test-repro-code.sh index d8e592186..304efbe31 100755 --- a/functest/test-repro-code.sh +++ b/functest/test-repro-code.sh @@ -10,7 +10,7 @@ function test_generic() { dvc_create_repo dvc_info 'Copy foo into foo1' - dvc run -d code/code.sh -d data/foo -o data/foo1 bash code/code.sh data/foo data/foo1 + dvc run -f copy_foo_foo1.dvc -d code/code.sh -d data/foo -o data/foo1 bash code/code.sh data/foo data/foo1 dvc_info 'Modify code' echo " " >> code/code.sh @@ -52,9 +52,9 @@ function test_partial() { git commit -m "copy code" dvc_info "Create repro chain foo -> foo1 -> foo2 -> foo3" - dvc run -d code/code1.sh -d data/foo -o data/foo1 cp data/foo data/foo1 - dvc run -d code/code2.sh -d data/foo1 -o data/foo2 cp data/foo1 data/foo2 - dvc run -d code/code3.sh -d data/foo2 -o data/foo3 cp data/foo2 data/foo3 + dvc run -f copy_foo_foo1.dvc -d code/code1.sh -d data/foo -o data/foo1 cp data/foo data/foo1 + dvc run -f copy_foo1_foo2.dvc -d code/code2.sh -d data/foo1 -o data/foo2 cp data/foo1 data/foo2 + dvc run -f copy_foo2_foo3.dvc -d code/code3.sh -d data/foo2 -o data/foo3 cp data/foo2 data/foo3 dvc_info "Save original timestamps" FOO_TS=$(dvc_timestamp data/foo)